.

lib/tests/test_zodb.py

@@ -22,10 +22,13 @@ from wendelin.lib.testing import getTestDB
 from wendelin.lib import testing
 from persistent import Persistent, UPTODATE, GHOST, CHANGED
 from ZODB import DB, POSException
+from ZODB.FileStorage import FileStorage
 from BTrees.IOBTree import IOBTree
 import transaction
 from transaction import TransactionManager
 from golang import defer, func
+from golang import context, sync
+from random import randint
 from pytest import raises
 import pytest; xfail = pytest.mark.xfail
 
@@ -307,6 +310,138 @@ def test_zconn_at():
     assert zconn_at(conn_at0) == at0
 
 
+# verify that zconn_at is robust wrt races.
+# see e.g.
+# - https://github.com/zopefoundation/ZODB/issues/290
+# - https://github.com/zopefoundation/ZODB/pull/307#discussion_r434145034
+# - https://github.com/zopefoundation/ZEO/issues/155
+# for what kind of data corruption issues might be there.
+@xfail(zmajor < 4, reason="zconn_at is TODO for ZODB3")
+@func
+def test_zconn_at_vs_races():
+    # XXX doc
+    stor0 = testdb.getZODBStorage()
+    db0   = DB(stor0)
+    defer(db0.close)
+
+    def dbopen():
+        # in ZODB/py FileStorage does not support opening the database from several clients
+        if isinstance(stor0, FileStorage):
+            return db0
+
+        stor = testdb.getZODBStorage()
+        db = DB(stor)
+        return db
+
+    def dbclose(db):
+        # see ^^^ about FileStorage
+        if db is db0:
+            return
+        db.close()
+
+    # init the database with two integer objects - obj1/obj2 that are set to 0.
+    @func
+    def _():
+        db = dbopen()
+        defer(lambda: dbclose(db))
+
+        transaction.begin()
+        conn = db.open()
+
+        root = conn.root()
+        root['obj1'] = XInt(0)
+        root['obj2'] = XInt(0)
+
+        transaction.commit()
+        conn.close()
+    _()
+
+
+    # T is a worker that accesses obj1/obj2 in a loop and verifies
+    # `obj1.i == obj2.i` invariant.
+    #
+    # access to obj1 is organized to go through zconn.
+    # access to obj2 goes through zconn2 that is opened with at=zconn_at(zconn).
+    #
+    # this verifies that at=zconn_at(zconn) is consistent with database state observed by zconn.
+    #
+    # Once in a while T tries to modify obj{1,2}.i maintaining the invariant as
+    # test source of changes for other workers.
+    @func
+    def T(ctx, name, N):
+        db = dbopen()
+        defer(lambda: dbclose(db))
+
+        @func
+        def t1():
+            transaction.begin()
+            zconn = db.open()
+            defer(zconn.close)
+
+            tm2 = TransactionManager()
+            at = zconn_at(zconn)
+            zconn2 = db.open(at=at, transaction_manager=tm2)
+            defer(zconn2.close)
+            defer(tm2.abort)
+
+            root = zconn.root()
+            obj1 = root['obj1']
+
+            root2 = zconn2.root()
+            obj2_ = root['obj2']
+            #obj2_._p_invalidate()
+
+            # both objects must have the same values (zconn vs zconn2)
+            i1  = obj1.i
+            i2_ = obj2_.i
+            if i1 != i2_:
+                raise AssertionError("T%s: obj1.i (%d)  !=  obj2_.i (%d)" % (name, i1, i2_))
+
+            # ----//---- (zconn vs zconn)
+            obj2 = root['obj2']
+            i2 = obj2.i
+            if i1 != i2:
+                raise AssertionError("T%s: obj1.i (%d)  !=  obj2.i (%d)" % (name, i1, i2))
+
+            # change objects once in a while
+            if randint(0,4) == 0:
+                #print("T%s: modify" % name)
+                obj1.i += 1
+                obj2.i += 1
+
+            # verify that zconn_at result stays the same during lifetime of one transaction
+            at_ = zconn_at(zconn)
+            if at_ != at:
+                raise AssertionError("T%s: at_ (%r)  !=  at (%r)" % (name, at_, at))
+            assert at_ == at
+
+            try:
+                transaction.commit()
+            except POSException.ConflictError:
+                #print('conflict -> ignore')
+                transaction.abort()
+
+        for i in range(N):
+            e = ctx.err()
+            if e is not None:
+                raise e
+            #print('T%s.%d' % (name, i))
+            t1()
+
+
+    # run 8 T workers concurrently. As of 20200123, likely due to race conditions   XXX
+    # in ZEO, it triggers the bug where T sees stale obj2 with obj1.i != obj2.i
+
+    N = 1000
+    wg = sync.WorkGroup(context.background())
+    for x in range(8):
+        wg.go(T, x, N)
+    wg.wait()
+
+    print('OK')
+
+
+
 # verify that ZODB.Connection.onResyncCallback works
 @xfail(zmajor < 4, reason="ZODB.Connection.onResyncCallback is TODO for ZODB3")
 @func