Merge pull request #54 from zopefoundation/zeo4-server-support

Zeo4 server support

.travis.yml

@@ -14,6 +14,15 @@ matrix:
         - os: linux
           python: 3.5
           env: ZEO_MTACCEPTOR=1
+        - os: linux
+          python: 2.7
+          env: ZEO4_SERVER=1
+        - os: linux
+          python: 3.4
+          env: ZEO4_SERVER=1
+        - os: linux
+          python: 3.5
+          env: ZEO4_SERVER=1
 install:
     - pip install -U setuptools
     - python bootstrap.py

CHANGES.rst

@@ -1,6 +1,8 @@
 Changelog
 =========
 
+- Fixed bugs in using the ZEO 5 client with ZEO 4 servers.
+
 5.0.0a2 (2016-07-30)
 --------------------
 

src/ZEO/ClientStorage.py

@@ -949,8 +949,7 @@ def restore(self, oid, serial, data, version, prev_txn, transaction):
     def serialnos(self, args):
         """Server callback to pass a list of changed (oid, serial) pairs.
         """
-        for oid, s in args:
-            self._tbuf.serial(oid, s)
+        self._tbuf.serialnos(args)
 
     def info(self, dict):
         """Server callback to update the info dictionary."""

src/ZEO/TransactionBuffer.py

@@ -92,3 +92,17 @@ def __iter__(self):
         for oid in server_resolved:
             if oid not in seen:
                 yield oid, None, True
+
+
+    # Support ZEO4:
+
+    def serialnos(self, args):
+        for oid in args:
+            if isinstance(oid, bytes):
+                self.server_resolved.add(oid)
+            else:
+                oid, serial = oid
+                if isinstance(serial, Exception):
+                    self.exception = serial
+                elif serial == b'rs':
+                    self.server_resolved.add(oid)

src/ZEO/asyncio/client.py

@@ -313,6 +313,12 @@ def __init__(self, loop,
         self.protocols = ()
         self.disconnected(None)
 
+        # Work around odd behavior of ZEO4 server. It may send
+        # invalidations for transactions later than the result of
+        # getInvalidations.  While we support ZEO 4 servers, we'll
+        # need to keep an invalidation queue. :(
+        self.verify_invalidation_queue = []
+
     def new_addrs(self, addrs):
         self.addrs = addrs
         if self.trying_to_connect():
@@ -409,6 +415,8 @@ def register_failed(self, protocol, exc):
 
     @future_generator
     def verify(self, server_tid):
+        self.verify_invalidation_queue = [] # See comment in init :(
+
         protocol = self.protocol
         if server_tid is None:
             server_tid = yield protocol.fut('lastTransaction')
@@ -465,6 +473,12 @@ def verify(self, server_tid):
             self.cache.setLastTid(server_tid)
             self.ready = True
 
+            # Gaaaa, ZEO 4 work around. See comment in __init__. :(
+            for tid, oids in self.verify_invalidation_queue:
+                if tid > server_tid:
+                    self.invalidateTransaction(tid, oids)
+            self.verify_invalidation_queue = []
+
             try:
                 info = yield protocol.fut('get_info')
             except Exception as exc:
@@ -597,16 +611,24 @@ def invalidateTransaction(self, tid, oids):
                 self.cache.invalidate(oid, tid)
             self.client.invalidateTransaction(tid, oids)
             self.cache.setLastTid(tid)
+        else:
+            self.verify_invalidation_queue.append((tid, oids))
 
     def serialnos(self, serials):
+        # Method called by ZEO4 storage servers.
+
         # Before delegating, check for errors (likely ConflictErrors)
         # and invalidate the oids they're associated with.  In the
         # past, this was done by the client, but now we control the
         # cache and this is our last chance, as the client won't call
         # back into us when there's an error.
-        for oid, serial in serials:
-            if isinstance(serial, Exception):
+        for oid in serials:
+            if isinstance(oid, bytes):
                 self.cache.invalidate(oid, None)
+            else:
+                oid, serial = oid
+                if isinstance(serial, Exception) or serial == b'rs':
+                    self.cache.invalidate(oid, None)
 
         self.client.serialnos(serials)
 

src/ZEO/tests/ZEO4/README.rst

@@ -0,0 +1,10 @@
+======================
+Copy of ZEO 4 server
+======================
+
+This copy was made by first converting the ZEO 4 server code to use
+relative imports.  The code was tested with ZEO 4 before copying.  It
+was unchanged aside from the relative imports.
+
+The ZEO 4 server is used for tests if the ZEO4_SERVER environment
+variable is set to a non-empty value.