fix: use fresh context cache for each transaction (#409)

In order to enforce transactional integrity, the context cache can't be shared
across transactions.

Fixes #394 (again)

Author: Chris Rossi

packages/google-cloud-ndb/google/cloud/ndb/_transaction.py

@@ -128,7 +128,9 @@ def _transaction_async(context, callback, read_only=False):
 
     on_commit_callbacks = []
     tx_context = context.new(
-        transaction=transaction_id, on_commit_callbacks=on_commit_callbacks
+        transaction=transaction_id,
+        on_commit_callbacks=on_commit_callbacks,
+        cache=None,  # Use new, empty cache for transaction
     )
     with tx_context.use():
         try:

packages/google-cloud-ndb/tests/system/test_crud.py

@@ -1173,6 +1173,7 @@ class SomeKind(ndb.Model):
     assert retreived.foo == ["", ""]
 
 
+@pytest.mark.skipif(not USE_REDIS_CACHE, reason="Redis is not configured")
 @pytest.mark.usefixtures("redis_context")
 def test_multi_get_weirdness_with_redis(dispose_of):
     """Regression test for issue #294.

packages/google-cloud-ndb/tests/system/test_misc.py

@@ -15,12 +15,15 @@
 """
 Difficult to classify regression tests.
 """
+import os
 import pickle
 
 import pytest
 
 from google.cloud import ndb
 
+USE_REDIS_CACHE = bool(os.environ.get("REDIS_CACHE_URL"))
+
 
 # Pickle can only pickle/unpickle global classes
 class PickleOtherKind(ndb.Model):
@@ -150,3 +153,81 @@ def concurrent_tasks(id):
 
     entity = SomeKind.get_by_id(id)
     assert entity.foo == 242
+
+
+def test_parallel_transactions_w_context_cache(client_context, dispose_of):
+    """Regression test for Issue #394
+
+    https://github.com/googleapis/python-ndb/issues/394
+    """
+
+    class SomeKind(ndb.Model):
+        foo = ndb.IntegerProperty()
+
+    @ndb.transactional_tasklet()
+    def update(id, add, delay=0):
+        entity = yield SomeKind.get_by_id_async(id)
+        foo = entity.foo
+        foo += add
+
+        yield ndb.sleep(delay)
+        entity.foo = foo
+
+        yield entity.put_async()
+
+    @ndb.tasklet
+    def concurrent_tasks(id):
+        yield [
+            update(id, 100),
+            update(id, 100, 0.01),
+        ]
+
+    with client_context.new(cache_policy=None).use():
+        key = SomeKind(foo=42).put()
+        dispose_of(key._key)
+        id = key.id()
+
+        concurrent_tasks(id).get_result()
+
+        entity = SomeKind.get_by_id(id)
+        assert entity.foo == 242
+
+
+@pytest.mark.skipif(not USE_REDIS_CACHE, reason="Redis is not configured")
+@pytest.mark.usefixtures("redis_context")
+def test_parallel_transactions_w_redis_cache(dispose_of):
+    """Regression test for Issue #394
+
+    https://github.com/googleapis/python-ndb/issues/394
+    """
+
+    class SomeKind(ndb.Model):
+        foo = ndb.IntegerProperty()
+
+    @ndb.transactional_tasklet()
+    def update(id, add, delay=0):
+        entity = yield SomeKind.get_by_id_async(id)
+        foo = entity.foo
+        foo += add
+
+        yield ndb.sleep(delay)
+        entity.foo = foo
+
+        yield entity.put_async()
+
+    @ndb.tasklet
+    def concurrent_tasks(id):
+        yield [
+            update(id, 100),
+            update(id, 100, 0.01),
+        ]
+
+    key = SomeKind(foo=42).put()
+    dispose_of(key._key)
+    id = key.id()
+
+    SomeKind.get_by_id(id)
+    concurrent_tasks(id).get_result()
+
+    entity = SomeKind.get_by_id(id)
+    assert entity.foo == 242

packages/google-cloud-ndb/tests/unit/test__transaction.py

@@ -84,10 +84,13 @@ class Test_transaction_async:
     @pytest.mark.usefixtures("in_context")
     @mock.patch("google.cloud.ndb._datastore_api")
     def test_success(_datastore_api):
+        context_module.get_context().cache["foo"] = "bar"
         on_commit_callback = mock.Mock()
 
         def callback():
-            context_module.get_context().call_on_commit(on_commit_callback)
+            context = context_module.get_context()
+            assert not context.cache
+            context.call_on_commit(on_commit_callback)
             return "I tried, momma."
 
         begin_future = tasklets.Future("begin transaction")