.

wcfs/wcfs_test.py

@@ -36,7 +36,7 @@ from ZODB.utils import z64, u64, p64
 import sys, os, os.path, subprocess, threading, inspect, traceback, re
 from time import gmtime
 from errno import EINVAL
-from golang import go, chan, select, func, defer
+from golang import go, chan, select, func, defer, default
 from golang import context, sync, time
 from golang.gcompat import qq
 from zodbtools.util import ashex as h, fromhex
@@ -1447,6 +1447,57 @@ def test_wcfs_pintimeout_kill():
     wg.wait()
 
 
+# watch with @at > head - must wait for head to become >= at
+# XXX too far ahead - error?
+@func
+def test_wcfs_watch_setup_ahead():
+    t = tDB(); zf = t.zfile
+    defer(t.close)
+
+    f = t.open(zf)
+    at1 = t.commit(zf, {2:'c1'})
+    f.assertData(['','','c1'])
+
+    wg = sync.WorkGroup(timeout())
+    dt = 100*time.millisecond
+    committing = chan() # becomes ready when T2 starts to commit
+
+    # T1: watch @(at1+1·dt)
+    @func
+    def _(ctx):
+        wl = t.openwatch()
+        defer(wl.close)
+
+        wat = tidfromtime(tidtime(at1) + 1*dt)  # > at1, but < at2
+        rxq = wl._sendReq(ctx, b"watch %s @%s" % (h(zf._p_oid), h(wat)))
+        _, _rx = select(
+            ctx.done().recv,    # 0
+            rxq.recv,           # 1
+        )
+        if _ == 0:
+            raise ctx.err()
+        assert ready(committing)
+        assert _rx == b"ok"
+    wg.go(_)
+
+    # T2: sleep(10·dt); commit
+    @func
+    def _(ctx):
+        # reopen connection to database as we are committing from another thread
+        conn = t.root._p_jar.db().open()
+        defer(conn.close)
+        root = conn.root()
+        zf = root['zfile']
+
+        time.sleep(10*dt)
+        committing.close()
+        at2 = t.commit(zf, {2:'c2'})
+        assert tidtime(at2) - tidtime(at1) >= 10*dt
+    wg.go(_)
+
+    wg.wait()
+
+
 # verify that watch setup/update sends correct pins.
 @func
 def test_wcfs_watch_setup():
@@ -1725,9 +1776,6 @@ def test_wcfs_watch_2files():
 
 
 
-    # XXX watch with @at > head - must wait for head to become >= at
-
-
     # XXX new watch request while previous watch request is in progress (over the same /head/watch handle)
 
 
@@ -1886,3 +1934,12 @@ def _xdefer(f):
 # replace defer with xdefer
 defer = xdefer
 del xdefer
+
+
+# ready reports whether chan ch is ready
+def ready(ch):
+    _, _rx = select(
+        default,    # 0
+        ch.recv,    # 1
+    )
+    return bool(_)