A ZEO threading bug could cause transactions to read inconsistent

data.  (This sometimes caused an AssertionError in
Connection._setstate_noncurrent.)

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,9 +1158,10 @@ class ClientStorage(object):
                 r = self._check_serials()
                 assert r is None or len(r) == 0, "unhandled serialnos: %s" % r
             except:
-                # The server successfully committed.  If we get a failure
-                # here, our own state will be in question, so reconnect.
-                self._connection.close()
+                if finished:
+                    # The server successfully committed.  If we get a failure
+                    # here, our own state will be in question, so reconnect.
+                    self._connection.close()
                 raise
 
             self.end_transaction()

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()