X race bug in ZODB Connection.open that leads to stale/corrupt data

(neo) (z-dev) (g.env) kirr@deco:~/src/wendelin/wendelin.core$ ./zrace.py
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.7/threading.py", line 754, in run
    self.__target(*self.__args, **self.__kwargs)
  File "./zrace.py", line 95, in T1
    t1()
  File "./zrace.py", line 89, in t1
    raise AssertionError("t1: obj1.i (%d)  !=  obj2.i (%d)" % (i1, i2))
AssertionError: t1: obj1.i (107)  !=  obj2.i (106)

Traceback (most recent call last):
  File "./zrace.py", line 136, in <module>
    main()
  File "./zrace.py", line 131, in main
    raise AssertionError('FAIL')
AssertionError: FAIL

zrace.py

+#!/usr/bin/env python
+# XXX doc
+#
+#newTransaction
+#   ._storage.sync()
+#   invalidated = ._storage.poll_invalidations():
+#
+#       ._storage._start = ._storage._storage.lastTrasaction() + 1:
+#           s = ._storage._storage
+#           s._lock.acquire()
+#           head = s._ltid
+#           s._lock.release()
+#           return head
+#                                               XXX T2 commits here: objX
+#
+#       ._storage._lock.acquire()
+#           r = _storage._invalidations        ; T1 receives invalidations for some transactions after head
+#       ._storage._lock.release()
+#       return r
+#
+#   # T1 processes invalidates for [... head] _and_ some next transactions.
+#   # T1 thus will _not_ process invalidations for those next transactions when
+#   # opening zconn _next_ time. The next opened zconn will thus see _stale_ data.
+#   ._cache.invalidate(invalidated)
+
+from ZODB import DB
+from ZODB.MappingStorage import MappingStorage
+import transaction
+from persistent import Persistent
+
+# don't depend on pygolang
+# ( but it is more easy and structured with sync.WorkGroup
+#   https://pypi.org/project/pygolang/#concurrency )
+#from golang import sync, context
+import threading
+def go(f, *argv, **kw):
+    t = threading.Thread(target=f, args=argv, kwargs=kw)
+    t.start()
+    return t
+
+
+
+# PInt is persistent integer:
+class PInt(Persistent):
+    def __init__(self, i):
+        self.i = i
+
+
+def main():
+    zstor = MappingStorage()
+    db    = DB(zstor)
+
+
+    def init():
+        transaction.begin()
+        zconn = db.open()
+
+        root = zconn.root()
+        root['obj1'] = PInt(0)
+        root['obj2'] = PInt(0)
+
+        transaction.commit()
+        zconn.close()
+
+
+    okv = [False, False]
+
+    # T1 accesses obj1/obj2 in a loop and verifies that obj1.i == obj2.i
+    #
+    # access to obj1 is organized to always trigger loading from zstor.
+    # access to obj2 goes through zconn cache and so verifies whether the cache is not stale.
+    def T1(N):
+        def t1():
+            transaction.begin()
+            zconn = db.open()
+
+            root = zconn.root()
+            obj1 = root['obj1']
+            obj2 = root['obj2']
+
+            # obj1 - reload it from zstor
+            # obj2 - get it from zconn cache
+            obj1._p_invalidate()
+
+            # both objects must have the same values
+            i1 = obj1.i
+            i2 = obj2.i
+            if i1 != i2:
+                raise AssertionError("t1: obj1.i (%d)  !=  obj2.i (%d)" % (i1, i2))
+
+            transaction.commit()
+            zconn.close()
+
+        for i in range(N):
+            t1()
+        okv[0] = True
+
+
+    # T2 changes obj1/obj2 in a loop by doing `objX.i += 1`.
+    #
+    # Since both objects start from 0, the invariant that `obj1.i == obj2.i` is always preserved.
+    def T2(N):
+        def t2():
+            transaction.begin()
+            zconn = db.open()
+
+            root = zconn.root()
+            obj1 = root['obj1']
+            obj2 = root['obj2']
+            obj1.i += 1
+            obj2.i += 1
+            assert obj1.i == obj2.i
+
+            transaction.commit()
+            zconn.close()
+
+        for i in range(N):
+            t2()
+        okv[1] = True
+
+
+    N = 1000
+    init()
+
+    t1 = go(T1, N)
+    t2 = go(T2, N)
+    t1.join()
+    t2.join()
+
+    if not all(okv):
+        raise AssertionError('FAIL')
+    print('OK')
+
+
+if __name__ == '__main__':
+    main()