Race in Connection.open() vs invalidations -> corrupt data

[navytux]

Hello up there,

For wendelin.core v2 I need a way to know at which particular database state application-level ZODB connection is viewing the database. I was looking for that in Connection.open / MVCCAdapter code and noticed concurrency bug that leads to wrong data returned by ZODB to application:

---- 8< ---- (zopenrace.py)

#!/usr/bin/env python
"""Program zopenrace.py demonstrates concurrency bug in ZODB Connection.open()
that leads to stale live cache and wrong data provided by database to users.

The bug is that when a connection is opened, it syncs to storage and processes
invalidations received from the storage in two _separate_ steps, potentially
leading to situation where invalidations for transactions _past_ opened
connection's view of the database are included into opened connection's cache
invalidation. This leads to stale connection cache and old data provided by
ZODB.Connection when it is reopened next time.

That in turn can lead to loose of Consistency of the database if mix of current
and old data is used to process a transaction. A classic example would be bank
accounts A, B and C with A<-B and A<-C transfer transactions. If transaction
that handles A<-C sees stale data for A when starting its processing, it
results in A loosing what it should have received from B.

Below is timing diagram on how the bug happens on ZODB5:

    Client1 or Thread1                                  Client2 or Thread2

    # T1 begins transaction and opens zodb connection
    newTransaction():
        # implementation in Connection.py[1]
        ._storage.sync()
        invalidated = ._storage.poll_invalidations():
            # implementation in MVCCAdapterInstance [2]

            # T1 settles on as of which particular database state it will be
            # viewing the database.
            ._storage._start = ._storage._storage.lastTrasaction() + 1:
                s = ._storage._storage
                s._lock.acquire()
                head = s._ltid
                s._lock.release()
                return head
                                                        # T2 commits here.
                                                        # Time goes by and storage server sends
                                                        # corresponding invalidation message to T1,
                                                        # which T1 queues in its _storage._invalidations

            # T1 retrieves queued invalidations which _includes_
            # invalidation for transaction that T2 just has committed past @head.
            ._storage._lock.acquire()
                r = _storage._invalidations
            ._storage._lock.release()
            return r

        # T1 processes invalidations for [... head] _and_ invalidations for past-@head transaction.
        # T1 thus will _not_ process invalidations for that next transaction when
        # opening zconn _next_ time. The next opened zconn will thus see _stale_ data.
        ._cache.invalidate(invalidated)


The program simulates two clients: one (T2) constantly modifies two integer
objects preserving invariant that their values stay equal. The other client
(T1) constantly opens the database and verifies the invariant. T1 forces access
to one of the object to always go through loading from the database, and this
way if live cache becomes stale the bug is observed as invariant breakage.

Here is example failure:

    $ taskset -c 1,2 ./zopenrace.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 "./zopenrace.py", line 136, in T1
        t1()
      File "./zopenrace.py", line 130, in t1
        raise AssertionError("t1: obj1.i (%d)  !=  obj2.i (%d)" % (i1, i2))
    AssertionError: t1: obj1.i (147)  !=  obj2.i (146)

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

NOTE ZODB4 and ZODB3 do not have this particular open vs invalidation race.

[1] https://github.com/zopefoundation/ZODB/blob/5.5.1-29-g0b3db5aee/src/ZODB/Connection.py#L734-L742
[2] https://github.com/zopefoundation/ZODB/blob/5.5.1-29-g0b3db5aee/src/ZODB/mvccadapter.py#L130-L139
"""

from __future__ import print_function
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)


    # init initializes the database with two integer objects - obj1/obj2 that are set to 0.
    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.abort() # we did not changed anything; also fails with commit
            zconn.close()

        for i in range(N):
            #print('T1.%d' % i)
            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):
            #print('T2.%d' % i)
            t2()
        okv[1] = True


    # run T1 and T2 concurrently. As of 20191210, due to race condition in
    # Connection.open, it triggers the bug where T1 sees stale obj2 with obj1.i != obj2.i
    init()

    N = 1000
    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()

Thanks beforehand,
Kirill

/cc @jimfulton

P.S. It would be nice to provide ZODB.Connection.at() explicitly similarly to https://godoc.org/lab.nexedi.com/kirr/neo/go/zodb#Connection

[jmuchemb]

I didn't read completely but my rework on #185 starts with 3d5ad16: would it fix this issue ?

[navytux]

It is easy to try: I've cherry-picked 3d5ad16 onto current master (5.5.1-29-g0b3db5aee) and reran the test:

(neo) (z-dev) (g.env) kirr@deco:~/src/wendelin/wendelin.core$ taskset -c 1,2 ./zopenrace.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 "./zopenrace.py", line 160, in T1
    t1()
  File "./zopenrace.py", line 153, in t1
    raise AssertionError("T1: obj1.i (%d)  !=  obj2.i (%d)" % (i1, i2))
AssertionError: T1: obj1.i (17)  !=  obj2.i (16)

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

[jamadden]

Yup, that's an issue. In effect, an invalidation is being "dropped" because it's processed too soon. Yes, the object gets invalidated, but its state is re-loaded as-of the current view of the database, which is pre-invalidation. And then the invalidation is gone.

At first glance, I suspect the fix is to make use of the currently-ignored tid parameter supplied to the callback function given to the underlying storage's tpc_finish method, passing that up to the MVCCAdapter._invalidate_finish method, which in turn passes it on to MVCCAdapterInstance._invalidate. The instances would need to change self._invalidations from a set into a datastructure that tracks the TIDs at which something was invalidated, and make poll_invalidations only return invalidations that are from TIDs <= self._start. A BTree {tid: set(oid)} could probably answer all those questions efficiently. A more advanced solution would collapse runs of TIDs together and discard TIDs older than the current transaction (I think that's a related problem; it seems possible to get invalidations from older transactions given the right sequence of locking. The consequence would only be an extra database load, not a consistency error.)

With a bit of mocking it should be possible to produce these race conditions deterministically in a single thread.

[jmuchemb]

In MVCCAdapterInstance.poll_invalidations, I'd first try to move the self._start = ... inside the with self._lock:.

[jamadden]

That's a recipe for deadlock; we (MVCCAdapterInstance) can't call into the underlying storage (which has its own lock) while holding our lock, because the underlying storage calls back into us (the callback from tpc_finish) while holding its lock, and that callback needs to take our lock:

********************************************************************************
* Threads
********************************************************************************
Thread 0x7000084c6000 (Thread-2)

  File "/lib/python3.7/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/lib/python3.7/threading.py", line 926, in _bootstrap_inner
    self.run()
  File "/lib/python3.7/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/tmp/race.py", line 97, in T2
    t2()
  File "/tmp/race.py", line 92, in t2
    transaction.commit()
  File "/transaction/transaction/_manager.py", line 257, in commit
    return self.manager.commit()
  File "/transaction/transaction/_manager.py", line 135, in commit
    return self.get().commit()
  File "/transaction/transaction/_transaction.py", line 273, in commit
    self._commitResources()
  File "/transaction/transaction/_transaction.py", line 447, in _commitResources
    rm.tpc_finish(self)
  File "/ZODB/src/ZODB/Connection.py", line 709, in tpc_finish
    serial = self._storage.tpc_finish(transaction)
  File "/ZODB/src/ZODB/mvccadapter.py", line 189, in tpc_finish
    return self._storage.tpc_finish(transaction, invalidate_finish)
  File "/ZODB/src/ZODB/utils.py", line 288, in __call__
    return func(*args, **kw)
  File "/ZODB/src/ZODB/MappingStorage.py", line 307, in tpc_finish
    func(tid)
  File "/ZODB/src/ZODB/mvccadapter.py", line 186, in invalidate_finish
    self._base._invalidate_finish(modified, self)
  File "/ZODB/src/ZODB/mvccadapter.py", line 86, in _invalidate_finish
    instance._invalidate(oids)
  File "/ZODB/src/ZODB/mvccadapter.py", line 125, in _invalidate
    with self._lock:

********************************************************************************
Thread 0x700007fc3000 (Thread-1)

  File "/lib/python3.7/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/lib/python3.7/threading.py", line 926, in _bootstrap_inner
    self.run()
  File "/lib/python3.7/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/tmp/race.py", line 73, in T1
    t1()
  File "/tmp/race.py", line 68, in t1
    transaction.abort() # we did not changed anything; also fails with commit
  File "/transaction/transaction/_manager.py", line 260, in abort
    return self.manager.abort()
  File "/transaction/transaction/_manager.py", line 140, in abort
    return self.get().abort()
  File "/transaction/transaction/_transaction.py", line 574, in abort
    self._synchronizers.map(lambda s: s.afterCompletion(self))
  File "/transaction/transaction/weakset.py", line 61, in map
    f(elt)
  File "/transaction/transaction/_transaction.py", line 574, in <lambda>
    self._synchronizers.map(lambda s: s.afterCompletion(self))
  File "/ZODB/src/ZODB/Connection.py", line 757, in afterCompletion
    self.newTransaction(transaction, False)
  File "/ZODB/src/ZODB/Connection.py", line 737, in newTransaction
    invalidated = self._storage.poll_invalidations()
  File "/ZODB/src/ZODB/mvccadapter.py", line 140, in poll_invalidations
    self._start = p64(u64(self._storage.lastTransaction()) + 1)
  File "/ZODB/src/ZODB/utils.py", line 281, in __call__
    with inst._lock:

I think there's a potential opportunity for optimization here, too, if the path I suggested works out, primarily for memory but also a bit of runtime. The MVCCAdapter (not the instances) could hold the master map of invalidations, so instances don't have to all make a copy (and we don't have to take locks on each instance). When an instance polls, it consults the master, who knows the transaction visibility of all instances, and it can discard invalidations for old TIDs that are no longer needed.

I basically did something like that in RelStorage. But given the much simpler underlying storage here, maybe that's over complex.

[jmuchemb]

If we can't call Storage.lastTransaction with the lock, let's not call it at all. We have another to get the last tid: from tpc_finish callback. Which means we would have an extra MVCCAdapterInstance._ltid attribute (updated in MVCCAdapterInstance._invalidate and used in MVCCAdapterInstance.poll_invalidations, this time with the lock).

[jamadden]

We have another to get the last tid: from tpc_finish callback. Which means we would have an extra MVCCAdapterInstance._ltid attribute (updated in MVCCAdapterInstance._invalidate and used in MVCCAdapterInstance.poll_invalidations, this time with the lock).

But that will guarantee that this MVCCAdapter instance is working with old data. Until (and unless) it commits, its view of the database never moves forward. It'll be stuck sending the same old TID to storage.loadBefore().

Or are you suggesting that every instance shares this TID with every other instance somehow? That still doesn't work when the storage can be changing in a different process (e.g., ZEO) or simply a different tree of MVCCAdapter wrappers.

[jmuchemb]

from tpc_finish callback

I was not clear here. There's also MVCCAdapter.invalidate for external invalidations.

What I mean is that MVCCAdapterInstance._invalidate could take a tid parameter that would be saved as _ltid attribute: _ltid would describe the state of _invalidations and it would be used to update _start, which is at the same moment that _invalidations is read.

I forgot MVCCAdapterInstance.tpc_finish would also have to update its own _ltid, which would be done just after the call to self._base._invalidate_finish.

[d-maurer]

Potentially, there is an easy way to fix the problem (even though it may not be optimal): delay the invalidation processing until after the start time is determined. This may see invalidations issued after the start time; ghostifying the referred to objects is no problem (as the transaction has not yet read object state) but it may become necessary to process those invalidations again after the current transaction to clean up the cache. One way to achieve this it to split invalidation processing into two phases: phase 1, process invalidations as now before the determination of the start time; phase 2, after the determination of the start time, ghostify the objects for newly arrived invalidations but do not delete the invalidations themselves such that they are reprocessed after the transaction.

[navytux]

Thanks for feedback, everyone.

I agree with @jamadden on how to fix this issue. In fact this is exactly the same approach that ZODB/go took regarding on how to process invalidations in zodb.DB:

https://lab.nexedi.com/kirr/neo/commit/d8e9d7a9
https://lab.nexedi.com/kirr/neo/commit/918455e7?expand_all_diffs=1
https://lab.nexedi.com/kirr/neo/blob/eccf827f/go/zodb/db.go#L65-115
https://lab.nexedi.com/kirr/neo/blob/eccf827f/go/zodb/δtail.go#L29-267

The data structure that is supporting such up-to-tid querying is ΔTail.

What @jmuchemb says is also correct: we get current head updates from invalidations messages received from storage server - as new transactions are committed by separate process, our process will eventually get corresponding invalidation messages even without any commit activity in the current process.

However when processes work in group on a common task, just eventually receiving invalidations for transactions committed by other processes is not enough: when starting new transaction a sync has to be made with round-trip to the server to make sure connection's view will be for some database state afterwards sync call was made. Current ZODB/py performs this sync automatically, but only if "explicit transactions" mode is used:

https://github.com/zopefoundation/ZODB/blob/5.5.1-31-g8e97bd7db/src/ZODB/Connection.py#L734-L736
https://github.com/zopefoundation/ZODB/blob/5.5.1-31-g8e97bd7db/src/ZODB/Connection.py#L903-L921

For the reference ZODB/go makes it to be a user choice at DB.Open time with "sync" being on by default:

https://godoc.org/lab.nexedi.com/kirr/neo/go/zodb#ConnOptions
https://godoc.org/lab.nexedi.com/kirr/neo/go/zodb#DB.Open

ZODB/go also makes a clear explicit separation of what .lastTransaction() (= .Head() in ZODB/go speak) and .sync() do:

https://godoc.org/lab.nexedi.com/kirr/neo/go/zodb#IStorage
https://lab.nexedi.com/kirr/neo/commit/151d8b79

@d-maurer, invalidating objects again after transaction completes and connection is returned back to DB pool could indeed work at the cost of trashing live cache.

Kirill

[navytux]

@jmuchemb says:

If we can't call Storage.lastTransaction with the lock, let's not call it at all.
...
What I mean is that MVCCAdapterInstance._invalidate could take a tid parameter that would be saved as _ltid attribute: _ltid would describe the state of _invalidations and it would be used to update _start, which is at the same moment that _invalidations is read.

Maybe I'm missing something, but we cannot skip call to .lastTransaction - if users are relying on sync, it is what .lastTransaction returns that should be used as head, not max tid in current invalidated set, as that set may have not yet received all invalidations from the server (or if it is guaranteed to be completely uptodate after sync, it has to documented). Also if ._invalidated set is not guaranteed to be uptodate after sync, it has to be waited for to become uptodate, or else we won't invalidate all objects that are neccessarry to invalidate to get correct view of the database as of head state.

[jamadden]

@navytux wrote:

Maybe I'm missing something, but we cannot skip call to .lastTransaction …

I agree.

@d-maurer, invalidating objects again after transaction completes and connection is returned back to DB pool could indeed work at the cost of trashing live cache.

Also agreed. That still requires tracking which invalidations still need to be re-processed, which means keeping track of the associated TID.

[jmuchemb]

No. sync is out of topic. Neither NEO nor ZEO performs a sync when lastTransaction() is called.

[d-maurer]

Jason Madden wrote at 2019-12-12 05:14 -0800:

> @d-maurer, invalidating objects again after transaction completes and connection is returned back to DB pool could indeed work at the cost of trashing live cache. Also agreed. That still requires tracking which invalidations still need to be re-processed, which means keeping track of the associated TID.

In my comment, I sketched a simpler approach which avoids TID tracking: Now you have: invalidate start_time = ... I propose: invalidate start_time = ... invalidate_and_record and when the transaction finished: process recorded invalidations. With low probability, this may invalidate some objects for which invalidation would not be necessary -- but this is at most unnecessary overhead.

[jamadden]

I don't see how that solves the problem. Unless the same object gets invalidated again during the "record" phase, you're still using outdated information because it's not going to get flushed from the cache after the transaction.