fix: increase cache lock expiration time (#740)

Also converted assert to a warning in case a lock expires from the cache
before it can be removed.

Fixes #728

Author: Chris Rossi

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

@@ -27,7 +27,7 @@
 
 _LOCKED_FOR_READ = b"0-"
 _LOCKED_FOR_WRITE = b"00"
-_LOCK_TIME = 32
+_LOCK_TIME = 64
 _PREFIX = b"NDB30"
 
 warnings.filterwarnings("always", module=__name__)
@@ -659,11 +659,27 @@ def global_unlock_for_write(key, lock):
     utils.logging_debug(log, "unlock for write: {}", lock)
 
     def new_value(old_value):
-        assert lock in old_value, "attempt to remove lock that isn't present"
-        value = old_value.replace(lock, b"")
+        value = old_value
+        if value and lock in value:
+            value = value.replace(lock, b"")
+
+        else:
+            warnings.warn(
+                "Attempt to remove a lock that doesn't exist. This is mostly likely "
+                "caused by a long running operation and the lock timing out.",
+                RuntimeWarning,
+            )
+
         if value == _LOCKED_FOR_WRITE:
             value = b""
 
+        if value and not value.startswith(_LOCKED_FOR_WRITE):
+            # If this happens, it means the lock expired and something else got written
+            # to the cache in the meantime. Whatever value that is, since there was a
+            # write operation that is concluding now, we should consider it stale and
+            # write a blank value.
+            value = b""
+
         return value
 
     cache = _global_cache()
@@ -685,6 +701,10 @@ def _update_key(key, new_value):
         value = new_value(old_value)
         utils.logging_debug(log, "new value: {}", value)  # pragma: SYNCPOINT update key
 
+        if old_value == value:
+            utils.logging_debug(log, "nothing to do")
+            return
+
         if old_value is not None:
             utils.logging_debug(log, "compare and swap")
             yield _global_watch(key, old_value)

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

@@ -880,7 +880,7 @@ def test_first_time(_global_cache, _global_get, global_set_if_not_exists, uuid):
 
         assert _cache.global_lock_for_write(b"key").result() == b".arandomuuid"
         _global_get.assert_called_once_with(b"key")
-        global_set_if_not_exists.assert_called_once_with(b"key", lock_value, expires=32)
+        global_set_if_not_exists.assert_called_once_with(b"key", lock_value, expires=64)
 
     @staticmethod
     @mock.patch("google.cloud.ndb._cache.uuid")
@@ -922,8 +922,8 @@ def test_not_first_time_fail_once(
         )
         _global_compare_and_swap.assert_has_calls(
             [
-                mock.call(b"key", new_lock_value, expires=32),
-                mock.call(b"key", new_lock_value, expires=32),
+                mock.call(b"key", new_lock_value, expires=64),
+                mock.call(b"key", new_lock_value, expires=64),
             ]
         )
 
@@ -955,7 +955,102 @@ def test_last_time(
         assert _cache.global_unlock_for_write(b"key", lock).result() is None
         _global_get.assert_called_once_with(b"key")
         _global_watch.assert_called_once_with(b"key", lock_value)
-        _global_compare_and_swap.assert_called_once_with(b"key", b"", expires=32)
+        _global_compare_and_swap.assert_called_once_with(b"key", b"", expires=64)
+
+    @staticmethod
+    @mock.patch("google.cloud.ndb._cache.uuid")
+    @mock.patch("google.cloud.ndb._cache._global_compare_and_swap")
+    @mock.patch("google.cloud.ndb._cache._global_watch")
+    @mock.patch("google.cloud.ndb._cache._global_get")
+    @mock.patch("google.cloud.ndb._cache._global_cache")
+    def test_lock_missing(
+        _global_cache, _global_get, _global_watch, _global_compare_and_swap, uuid
+    ):
+        lock = b".arandomuuid"
+
+        _global_cache.return_value = mock.Mock(
+            transient_errors=(),
+            strict_write=False,
+            spec=("transient_errors", "strict_write"),
+        )
+
+        lock_value = _cache._LOCKED_FOR_WRITE + b".adifferentlock"
+        _global_get.return_value = _future_result(lock_value)
+        _global_watch.return_value = _future_result(None)
+        _global_compare_and_swap.return_value = _future_result(True)
+
+        with warnings.catch_warnings(record=True) as logged:
+            assert _cache.global_unlock_for_write(b"key", lock).result() is None
+            logged = [
+                warning for warning in logged if warning.category is RuntimeWarning
+            ]
+            assert len(logged) == 1
+
+        _global_get.assert_called_once_with(b"key")
+        _global_watch.assert_not_called()
+        _global_compare_and_swap.assert_not_called()
+
+    @staticmethod
+    @mock.patch("google.cloud.ndb._cache.uuid")
+    @mock.patch("google.cloud.ndb._cache.global_set_if_not_exists")
+    @mock.patch("google.cloud.ndb._cache._global_get")
+    @mock.patch("google.cloud.ndb._cache._global_cache")
+    def test_no_value_in_cache(
+        _global_cache, _global_get, global_set_if_not_exists, uuid
+    ):
+        lock = b".arandomuuid"
+
+        _global_cache.return_value = mock.Mock(
+            transient_errors=(),
+            strict_write=False,
+            spec=("transient_errors", "strict_write"),
+        )
+
+        _global_get.return_value = _future_result(None)
+        global_set_if_not_exists.return_value = _future_result(True)
+
+        with warnings.catch_warnings(record=True) as logged:
+            assert _cache.global_unlock_for_write(b"key", lock).result() is None
+            logged = [
+                warning for warning in logged if warning.category is RuntimeWarning
+            ]
+            assert len(logged) == 1
+
+        _global_get.assert_called_once_with(b"key")
+        global_set_if_not_exists.assert_not_called()
+
+    @staticmethod
+    @mock.patch("google.cloud.ndb._cache.uuid")
+    @mock.patch("google.cloud.ndb._cache._global_compare_and_swap")
+    @mock.patch("google.cloud.ndb._cache._global_watch")
+    @mock.patch("google.cloud.ndb._cache._global_get")
+    @mock.patch("google.cloud.ndb._cache._global_cache")
+    def test_lock_overwritten(
+        _global_cache, _global_get, _global_watch, _global_compare_and_swap, uuid
+    ):
+        lock = b".arandomuuid"
+
+        _global_cache.return_value = mock.Mock(
+            transient_errors=(),
+            strict_write=False,
+            spec=("transient_errors", "strict_write"),
+        )
+
+        lock_value = b"SOMERANDOMVALUE"
+        _global_get.return_value = _future_result(lock_value)
+        _global_watch.return_value = _future_result(None)
+        _global_compare_and_swap.return_value = _future_result(True)
+
+        with warnings.catch_warnings(record=True) as logged:
+            assert _cache.global_unlock_for_write(b"key", lock).result() is None
+            logged = [
+                warning for warning in logged if warning.category is RuntimeWarning
+            ]
+            assert len(logged) == 1
+
+        _global_get.assert_called_once_with(b"key")
+        _global_watch.assert_called_once_with(b"key", lock_value)
+        _global_compare_and_swap.assert_called_once_with(b"key", b"", expires=64)
 
     @staticmethod
     @mock.patch("google.cloud.ndb._cache.uuid")
@@ -1023,8 +1118,8 @@ def test_not_last_time_fail_once(
         )
         _global_compare_and_swap.assert_has_calls(
             [
-                mock.call(b"key", new_lock_value, expires=32),
-                mock.call(b"key", new_lock_value, expires=32),
+                mock.call(b"key", new_lock_value, expires=64),
+                mock.call(b"key", new_lock_value, expires=64),
             ]
         )