bigfile/zodb: ZBlk._p_invalidate() can be called more than once, in particular in case of conflicts

When there is a conflict (on any object, but on ZBlk in particular) ZODB
machinery calls its ._p_invalidate() twice:

  File ".../wendelin.core/bigfile/tests/test_filezodb.py", line 661, in test_bigfile_filezodb_vs_conflicts
    tm2.commit()    # this should raise ConflictError and stay at 11 state
  File ".../transaction/_manager.py", line 111, in commit
    return self.get().commit()
  File ".../transaction/_transaction.py", line 271, in commit
    self._commitResources()
  File ".../transaction/_transaction.py", line 414, in _commitResources
    self._cleanup(L)
  File ".../transaction/_transaction.py", line 426, in _cleanup
    rm.abort(self)
  File ".../ZODB/Connection.py", line 436, in abort
    self._abort()
  File ".../ZODB/Connection.py", line 479, in _abort
    self._cache.invalidate(oid)
  File ".../wendelin.core/bigfile/file_zodb.py", line 148, in _p_invalidate
    traceback.print_stack()

and

  File ".../wendelin.core/bigfile/tests/test_filezodb.py", line 661, in test_bigfile_filezodb_vs_conflicts
    tm2.commit()    # this should raise ConflictError and stay at 11 state
  File ".../transaction/_manager.py", line 111, in commit
    return self.get().commit()
  File ".../transaction/_transaction.py", line 271, in commit
    self._commitResources()
  File ".../transaction/_transaction.py", line 416, in _commitResources
    self._synchronizers.map(lambda s: s.afterCompletion(self))
  File ".../transaction/weakset.py", line 59, in map
    f(elt)
  File ".../transaction/_transaction.py", line 416, in <lambda>
    self._synchronizers.map(lambda s: s.afterCompletion(self))
  File ".../ZODB/Connection.py", line 831, in _storage_sync
    self._flush_invalidations()
  File ".../ZODB/Connection.py", line 539, in _flush_invalidations
    self._cache.invalidate(invalidated)
  File ".../wendelin.core/bigfile/file_zodb.py", line 148, in _p_invalidate
    traceback.print_stack()

i.e. first invalidation is done by commit cleanup:

    https://github.com/zopefoundation/transaction/blob/1.4.4/transaction/_transaction.py#L414
    https://github.com/zopefoundation/ZODB/blob/3.10/src/ZODB/Connection.py#L479

and then Connection.afterCompletion() flushes invalidation again:

    https://github.com/zopefoundation/transaction/blob/1.4.4/transaction/_transaction.py#L416
    https://github.com/zopefoundation/ZODB/blob/3.10/src/ZODB/Connection.py#L833
    https://github.com/zopefoundation/ZODB/blob/3.10/src/ZODB/Connection.py#L539

If there was no conflict - there will be no ConflictError raised and
thus no Transaction._cleanup() done in its ._commitResources() ->
invalidation called only once. But with ConflictError - it is twice.

Adjust ZBlk._p_invalidate() not to delve into real invalidation more
than once - else we will fail, as ZBlk._v_zfile becomes unbound after
invalidation done the first time.

bigarray/tests/test_arrayzodb.py

@@ -22,10 +22,13 @@ from wendelin.lib.testing import getTestDB
 from persistent import UPTODATE
 import transaction
 from transaction import TransactionManager
+from ZODB.POSException import ConflictError
 from numpy import dtype, uint8, all, array_equal, arange
 from threading import Thread
 from six.moves import _thread
 
+from pytest import raises
+
 testdb = None
 def setup_module():
     global testdb
@@ -408,6 +411,64 @@ def test_zbigarray_vs_cache_invalidation():
     dbclose(root1)
 
 
+# verify that conflicts on array content are handled properly
+# ( NOTE this test is almost dup of test_bigfile_filezodb_vs_conflicts() )
+def test_zbigarray_vs_conflicts():
+    root = testdb.dbopen()
+    conn = root._p_jar
+    db   = conn.db()
+    conn.close()
+    del root, conn
+
+    tm1 = TransactionManager()
+    tm2 = TransactionManager()
+
+    conn1 = db.open(transaction_manager=tm1)
+    root1 = conn1.root()
+
+    # setup zarray
+    root1['zarray3a'] = a1 = ZBigArray((10,), uint8)
+    tm1.commit()
+
+    # set zarray initial data
+    a1[0:1] = [1]           # XXX -> [0] = 1  after BigArray can
+    tm1.commit()
+
+    # read zarray in conn2
+    conn2 = db.open(transaction_manager=tm2)
+    root2 = conn2.root()
+
+    a2 = root2['zarray3a']
+    assert a2[0:1] == [1]   # read data in conn2 + make sure read correctly
+                            # XXX -> [0] == 1  after BigArray can
+
+    # now zarray content is both in ZODB.Connection cache and in _ZBigFileH
+    # cache for each conn1 and conn2. Modify data in both conn1 and conn2 and
+    # see how it goes.
+
+    a1[0:1] = [11]          # XXX -> [0] = 11  after BigArray can
+    a2[0:1] = [12]          # XXX -> [0] = 12  after BigArray can
+
+    # txn1 should commit ok
+    tm1.commit()
+
+    # txn2 should raise ConflictError and stay at 11 state
+    raises(ConflictError, 'tm2.commit()')
+    tm2.abort()
+
+    assert a2[0:1] == [11]  # re-read in conn2  XXX -> [0] == 11 after BigArray can
+    a2[0:1] = [13]          # XXX -> [0] = 13 after BigArray can
+    tm2.commit()
+
+    assert a1[0:1] == [11]  # not yet propagated to conn1   XXX -> [0] == 11
+    tm1.commit()            # transaction boundary
+
+    assert a1[0:1] == [13]  # re-read in conn1  XXX -> [0] == 13
+
+    conn2.close()
+    dbclose(root1)
+
+
 # verify how ZBigArray behaves when plain properties are changed / invalidated
 def test_zbigarray_invalidate_shape():
     root = testdb.dbopen()

bigfile/file_zodb.py

@@ -32,7 +32,7 @@ from wendelin.bigfile import BigFile, WRITEOUT_STORE, WRITEOUT_MARKSTORED
 from wendelin.lib.mem import bzero, memcpy
 
 from transaction.interfaces import IDataManager, ISynchronizer
-from persistent import Persistent, PickleCache
+from persistent import Persistent, PickleCache, GHOST
 from BTrees.LOBTree import LOBTree
 from zope.interface import implementer
 from ZODB.Connection import Connection
@@ -135,6 +135,9 @@ class ZBlk(Persistent):
     # DB notifies this object has to be invalidated
     # (DB -> invalidate ._v_blkdata -> invalidate memory-page)
     def _p_invalidate(self):
+        # do real invalidation only once - else we already lost ._v_zfile last time
+        if self._p_state is GHOST:
+            return
         # on invalidation we must be already bound
         # (to know which ZBigFileH to propagate invalidation to)
         assert self._v_zfile    is not None

bigfile/tests/test_filezodb.py

@@ -22,6 +22,7 @@ from wendelin.lib.testing import getTestDB
 from persistent import UPTODATE, GHOST, CHANGED
 import transaction
 from transaction import TransactionManager
+from ZODB.POSException import ConflictError
 from numpy import ndarray, array_equal, uint8, zeros
 from threading import Thread
 from six.moves import _thread
@@ -612,6 +613,71 @@ def test_bigfile_filezodb_vs_cache_invalidation():
     dbclose(root1)
 
 
+# verify that conflicts on ZBlk are handled properly
+# ( NOTE this test is almost dupped at test_zbigarray_vs_conflicts() )
+def test_bigfile_filezodb_vs_conflicts():
+    root = dbopen()
+    conn = root._p_jar
+    db   = conn.db()
+    conn.close()
+    del root, conn
+
+    tm1 = TransactionManager()
+    tm2 = TransactionManager()
+
+    conn1 = db.open(transaction_manager=tm1)
+    root1 = conn1.root()
+
+    # setup zfile with fileh view to it
+    root1['zfile3a'] = f1 = ZBigFile(blksize)
+    tm1.commit()
+
+    fh1 = f1.fileh_open()
+    tm1.commit()
+
+    # set zfile initial data
+    vma1 = fh1.mmap(0, 1)
+    Blk(vma1, 0)[0] = 1
+    tm1.commit()
+
+    # read zfile and setup fileh for it in conn2
+    conn2 = db.open(transaction_manager=tm2)
+    root2 = conn2.root()
+
+    f2 = root2['zfile3a']
+    fh2 = f2.fileh_open()
+    vma2 = fh2.mmap(0, 1)
+
+    assert Blk(vma2, 0)[0] == 1 # read data in conn2 + make sure read correctly
+
+    # now zfile content is both in ZODB.Connection cache and in _ZBigFileH
+    # cache for each conn1 and conn2. Modify data in both conn1 and conn2 and
+    # see how it goes.
+
+    Blk(vma1, 0)[0] = 11
+    Blk(vma2, 0)[0] = 12
+
+    # txn1 should commit ok
+    tm1.commit()
+
+    # txn2 should raise ConflictError and stay at 11 state
+    raises(ConflictError, 'tm2.commit()')
+    tm2.abort()
+
+    assert Blk(vma2, 0)[0] == 11    # re-read in conn2
+    Blk(vma2, 0)[0] = 13
+    tm2.commit()
+
+    assert Blk(vma1, 0)[0] == 11    # not yet propagated to conn1
+    tm1.commit()                    # transaction boundary
+
+    assert Blk(vma1, 0)[0] == 13    # re-read in conn1
+
+    conn2.close()
+    dbclose(root1)
+
+
+
 # verify that fileh are garbage-collected after user free them
 def test_bigfile_filezodb_fileh_gc():
     root1= dbopen()