Merge pull request #174 from zopefoundation/ZODB5.6_compatible_lastTransaction#166

let `lastTransaction` change only after invalidation processing

CHANGES.rst

@@ -4,6 +4,12 @@ Changelog
 5.2.3 (unreleased)
 ------------------
 
+- Ensure ``ZEO`` satisfies the ``ZODB >= 5.6`` requirement that
+  ``lastTransaction()`` changes only after invalidation processing.
+  Violating this requirement can lead to race conditions and
+  associated data corruption
+  `#166 <https://github.com/zopefoundation/ZEO/issues/166>`_.
+
 - Add automated tests against the ZODB ``master`` branch
   see `issue 177 <https://github.com/zopefoundation/ZEO/issues/177>`_.
 

src/ZEO/asyncio/client.py

@@ -649,11 +649,31 @@ class Client(object):
             try:
                 tid = yield self.protocol.fut('tpc_finish', tid)
                 cache = self.cache
+                # The cache invalidation here and that in
+                # ``invalidateTransaction`` are both performed
+                # in the IO thread. Thus there is no interference.
+                # Other threads might observe a partially invalidated
+                # cache. However, regular loads will access
+                # object state before ``tid``; therefore,
+                # partial invalidation for ``tid`` should not harm.
                 for oid, data, resolved in updates:
                     cache.invalidate(oid, tid)
                     if data and not resolved:
                         cache.store(oid, tid, None, data)
-                cache.setLastTid(tid)
+                # ZODB >= 5.6 requires that ``lastTransaction`` changes
+                # only after invalidation processing (performed in
+                # the ``f`` call below) (for ``ZEO``, ``lastTransaction``
+                # is implemented as ``cache.getLastTid()``).
+                # Some tests involve ``f`` in the verification that
+                # ``tpc_finish`` modifies ``lastTransaction`` and require
+                # that ``cache.setLastTid`` is called before ``f``.
+                # We use locking below to ensure that the
+                # effect of ``setLastTid`` is observable by other
+                # threads only after ``f`` has been called.
+                with cache._lock:
+                    cache.setLastTid(tid)
+                    f(tid)
+                future.set_result(tid)
             except Exception as exc:
                 future.set_exception(exc)
 
@@ -662,9 +682,6 @@ class Client(object):
                 # recovering to a consistent state.
                 self.protocol.close()
                 self.disconnected(self.protocol)
-            else:
-                f(tid)
-                future.set_result(tid)
         else:
             future.set_exception(ClientDisconnected())
 
@@ -674,6 +691,8 @@ class Client(object):
 
     def invalidateTransaction(self, tid, oids):
         if self.ready:
+            # see the cache related comment in ``tpc_finish_threadsafe``
+            # why we think that locking is not necessary at this place
             for oid in oids:
                 self.cache.invalidate(oid, tid)
             self.client.invalidateTransaction(tid, oids)

src/ZEO/asyncio/tests.py

@@ -9,7 +9,7 @@ from zope.testing import setupstack
 from concurrent.futures import Future
 import mock
 from ZODB.POSException import ReadOnlyError
-from ZODB.utils import maxtid
+from ZODB.utils import maxtid, RLock
 
 import collections
 import logging
@@ -699,6 +699,7 @@ class MemoryCache(object):
         # { oid -> [(start, end, data)] }
         self.data = collections.defaultdict(list)
         self.last_tid = None
+        self._lock = RLock()
 
     clear = __init__