Bug Fixed:

A ZEO threading bug could cause transactions to read inconsistent
  data.  (This sometimes caused an AssertionError in
  Connection._setstate_noncurrent.)

src/CHANGES.txt

@@ -8,6 +8,10 @@
 Bugs Fixed
 ----------
 
+- A ZEO threading bug could cause transactions to read inconsistent
+  data.  (This sometimes caused an AssertionError in
+  Connection._setstate_noncurrent.)
+
 - DemoStorage.loadBefore sometimes returned invalid data which
   would trigger AssertionErrors in ZODB.Connection.
 

src/ZEO/ClientStorage.py

@@ -1143,15 +1143,12 @@ class ClientStorage(object):
                 raise ClientDisconnected(
                        'Calling tpc_finish() on a disconnected transaction')
 
-            # The calls to tpc_finish() and _update_cache() should
-            # never run currently with another thread, because the
-            # tpc_cond condition variable prevents more than one
-            # thread from calling tpc_finish() at a time.
-            tid = self._server.tpc_finish(id(txn))
-
+            finished = 0
             try:
                 self._lock.acquire()  # for atomic processing of invalidations
                 try:
+                    tid = self._server.tpc_finish(id(txn))
+                    finished = 1
                     self._update_cache(tid)
                     if f is not None:
                         f(tid)
@@ -1161,6 +1158,7 @@ class ClientStorage(object):
                 r = self._check_serials()
                 assert r is None or len(r) == 0, "unhandled serialnos: %s" % r
             except:
+                if finished:
                     # The server successfully committed.  If we get a failure
                     # here, our own state will be in question, so reconnect.
                     self._connection.close()

src/ZODB/Connection.py

@@ -341,6 +341,10 @@ class Connection(ExportImport, object):
         try:
             if self._txn_time is None:
                 self._txn_time = tid
+            elif tid < self._txn_time:
+                raise AssertionError("invalidations out of order, %r < %r"
+                                     % (tid, self._txn_time))
+
             self._invalidated.update(oids)
         finally:
             self._inv_lock.release()