client: Fix cache corruption on loadBefore and prefetch (#169)

Currently loadBefore and prefetch spawn async protocol.load_before task,
and, after waking up on its completion, populate the cache with received
data. But in between the time when protocol.load_before handler is run
and the time when protocol.load_before caller wakes up, there is a
window in which event loop might be running some other code, including
the code that handles invalidateTransaction messages from the server.

This means that cache updates and cache invalidations can be processed on
the client not in the order that server sent it. And such difference in
the order can lead to data corruption if e.g server sent

	<- loadBefore oid serial=tid1 next_serial=None
	<- invalidateTransaction tid2 oid

and client processed it as

	invalidateTransaction tid2 oid
	cache.store(oid, tid1, next_serial=None)

because here the end effect is that invalidation for oid@tid2 is not
applied to the cache.

The fix is simple: perform cache updates right after loadBefore reply is
received.

Fixes: https://github.com/zopefoundation/ZEO/issues/155

The fix is based on analysis and initial patch by @jmuchemb:

https://github.com/zopefoundation/ZEO/issues/155#issuecomment-581046248

A tests corresponding to the fix is coming coming through

https://github.com/zopefoundation/ZEO/pull/170  and
https://github.com/zopefoundation/ZODB/pull/345

For the reference, my original ZEO-specific data corruption reproducer
is here:

https://github.com/zopefoundation/ZEO/issues/155#issuecomment-577602842
https://lab.nexedi.com/kirr/wendelin.core/blob/ecd0e7f0/zloadrace5.py

/cc @jamadden, @dataflake, @jimfulton
/reviewed-by @jmuchemb, @d-maurer
/reviewed-on https://github.com/zopefoundation/ZEO/pull/169

CHANGES.rst

@@ -4,6 +4,8 @@ Changelog
 5.2.3 (unreleased)
 ------------------
 
+- Fix data corruption due to race between load and external invalidations.
+  See `issue 155 <https://github.com/zopefoundation/ZEO/issues/155>`_.
 
 5.2.2 (2020-08-11)
 ------------------

src/ZEO/asyncio/client.py

@@ -252,10 +252,19 @@ class Protocol(base.Protocol):
         message_id = (oid, tid)
         future = self.futures.get(message_id)
         if future is None:
-            future = asyncio.Future(loop=self.loop)
+            future = Fut()
             self.futures[message_id] = future
             self._write(
                 self.encode(message_id, False, 'loadBefore', (oid, tid)))
+            @future.add_done_callback
+            def _(future):
+                try:
+                    data = future.result()
+                except Exception:
+                    return
+                if data:
+                    data, start, end = data
+                    self.client.cache.store(oid, start, end, data)
         return future
 
     # Methods called by the server.
@@ -608,9 +617,6 @@ class Client(object):
                 future.set_exception(exc)
             else:
                 future.set_result(data)
-                if data:
-                    data, start, end = data
-                    self.cache.store(oid, start, end, data)
         elif wait_ready:
             self._when_ready(
                 self.load_before_threadsafe, future, wait_ready, oid, tid)
@@ -620,10 +626,7 @@ class Client(object):
     @future_generator
     def _prefetch(self, oid, tid):
         try:
-            data = yield self.protocol.load_before(oid, tid)
-            if data:
-                data, start, end = data
-                self.cache.store(oid, start, end, data)
+            yield self.protocol.load_before(oid, tid)
         except Exception:
             logger.exception("prefetch %r %r" % (oid, tid))
 
@@ -888,20 +891,25 @@ class ClientThread(ClientRunner):
                 raise self.exception
 
 class Fut(object):
-    """Lightweight future that calls it's callback immediately rather than soon
+    """Lightweight future that calls it's callbacks immediately rather than soon
     """
 
+    def __init__(self):
+        self.cbv = []
+
     def add_done_callback(self, cb):
-        self.cb = cb
+        self.cbv.append(cb)
 
     exc = None
     def set_exception(self, exc):
         self.exc = exc
-        self.cb(self)
+        for cb in self.cbv:
+            cb(self)
 
     def set_result(self, result):
         self._result = result
-        self.cb(self)
+        for cb in self.cbv:
+            cb(self)
 
     def result(self):
         if self.exc: