Reproducer for tree cache event race

01-test_cache.py

from kazoo.client import KazooClient
from kazoo.recipe.cache import TreeCache

zuul_root = "/path/to/src/opendev/zuul/zuul"

k = KazooClient(
    hosts="localhost:2281",
    keyfile=f"{zuul_root}/tools/ca/keys/clientkey.pem",
    certfile=f"{zuul_root}/tools/ca/certs/client.pem",
    ca=f"{zuul_root}/tools/ca/certs/cacert.pem",
    use_ssl=True
)
k.start()


def _watcher(tree_event):
    print(tree_event)
    # We don't expect to see a node added event with data "end"
    if tree_event.event_type == TreeEvent.NODE_ADDED:
        assert tree_event.event_data.data != b"end"


t = TreeCache(k, "/root")
t.listen(_watcher)
t.start()


b = "/root/branch"
for i in range(10):
    k.create(b, b"create")
    k.set(b, b"start")
    k.set(b, b"middle")
    k.set(b, b"end")
    k.delete(b)

t.close()
k.stop()

02-kazoo.diff

diff --git a/kazoo/recipe/cache.py b/kazoo/recipe/cache.py
index dc3b8bd..bb16694 100644
--- a/kazoo/recipe/cache.py
+++ b/kazoo/recipe/cache.py
@@ -301,6 +301,7 @@ def _call_client(self, method_name, path):
             self._tree._in_background, self._process_result,
             method_name, path)
         method = getattr(self._tree._client, method_name + '_async')
+        print("delay ...")
         method(path, watch=self._process_watch).rawlink(callback)

     def _process_watch(self, watched_event):

03-output.log

...
(0, ('/root/branch', b'start', ZnodeStat(czxid=3382, mzxid=3383, ctime=1623929138978, mtime=1623929138978, version=1, cversion=0, aversion=0, ephemeralOwner=0, dataLength=5, numChildren=0, pzxid=3382)))
delay ...
delay ...
(1, ('/root/branch', b'middle', ZnodeStat(czxid=3382, mzxid=3384, ctime=1623929138978, mtime=1623929138979, version=2, cversion=0, aversion=0, ephemeralOwner=0, dataLength=6, numChildren=0, pzxid=3382)))
(2, ('/root/branch', b'middle', ZnodeStat(czxid=3382, mzxid=3384, ctime=1623929138978, mtime=1623929138979, version=2, cversion=0, aversion=0, ephemeralOwner=0, dataLength=6, numChildren=0, pzxid=3382)))
delay ...
(0, ('/root/branch', b'end', ZnodeStat(czxid=3382, mzxid=3385, ctime=1623929138978, mtime=1623929138979, version=3, cversion=0, aversion=0, ephemeralOwner=0, dataLength=3, numChildren=0, pzxid=3382)))
...

04-output-commented.log

...
# OK: TreeEvent.NODE_ADDED event with correct data/version
(0, ('/root/branch', b'start', ZnodeStat(czxid=3382, mzxid=3383, ctime=1623929138978, mtime=1623929138978, version=1, cversion=0, aversion=0, ephemeralOwner=0, dataLength=5, numChildren=0, pzxid=3382)))
delay ...
delay ...

# OK: TreeEvent.NODE_UPDATED event with correct data/version
(1, ('/root/branch', b'middle', ZnodeStat(czxid=3382, mzxid=3384, ctime=1623929138978, mtime=1623929138979, version=2, cversion=0, aversion=0, ephemeralOwner=0, dataLength=6, numChildren=0, pzxid=3382)))

# NOK: TreeEvent.NODE_DELETED event still with old data/version (version 3 was deleted NOT version 2)
(2, ('/root/branch', b'middle', ZnodeStat(czxid=3382, mzxid=3384, ctime=1623929138978, mtime=1623929138979, version=2, cversion=0, aversion=0, ephemeralOwner=0, dataLength=6, numChildren=0, pzxid=3382)))
delay ...

# NOK: TreeEvent.NODE_ADDED event with now deleted data/version
(0, ('/root/branch', b'end', ZnodeStat(czxid=3382, mzxid=3385, ctime=1623929138978, mtime=1623929138979, version=3, cversion=0, aversion=0, ephemeralOwner=0, dataLength=3, numChildren=0, pzxid=3382)))
...