ci-parity round 11: redis WATCH/MULTI atomic admission + project_peer NC in-memory migration + abort_toolonly ci_serial

Author: rangehow

lib/runtime_state_store.py

@@ -288,23 +288,29 @@ def _zcap_k(self, kind: str, count_prefix: str) -> str:
     def acquire_slot(self, kind: str, slot_key: str, limit: int, ttl: float,
                      count_prefix: str) -> bool:
         """Bounded acquire via a Redis SORTED SET (score = expiry deadline) —
-        the single source of truth for the cap. Uses only ``ZADD`` / ``ZRANK``
-        / ``ZREM`` / ``ZREMRANGEBYSCORE`` / ``ZCOUNT`` — all guaranteed by
-        fakeredis AND managed/cluster Redis (NOT Lua EVAL / SCAN-in-Lua).
-
-        Algorithm (lock-free, total-order rank — correct under BOTH sequential
-        admits AND a concurrent burst):
+        the single source of truth for the cap. Uses only WATCH/MULTI/EXEC +
+        ``ZADD`` / ``ZSCORE`` / ``ZCOUNT`` / ``ZREMRANGEBYSCORE`` — guaranteed
+        by fakeredis AND managed/cluster Redis (NO Lua EVAL / SCAN-in-Lua).
+
+        Algorithm (optimistic-transaction admission — WATCH on the cap key,
+        count-check and claim commit ATOMICALLY via MULTI/EXEC; a concurrent
+        writer aborts the txn with WatchError and we retry against the fresh
+        count):
           1. ``ZREMRANGEBYSCORE cap 0 now`` — evict members whose expiry
              deadline passed (crash-reclaim, no per-member TTL needed).
-          2. ``ZADD cap {slot: now+ttl}`` — insert/refresh our member with its
-             deadline as score. Re-acquiring an existing member just updates
-             its score (no double-count).
-          3. ``ZRANK cap slot`` — our position in the score-ordered set. The
-             ZSET gives ALL racers the SAME total order, so they independently
-             agree on the lowest-``limit`` winners. Keep iff ``rank < limit``;
-             otherwise ``ZREM`` our OWN member and refuse. This admits EXACTLY
-             ``limit`` under a burst (no livelock, no overshoot) and is stable
-             sequentially (an over-cap claim is rolled back).
+          2. ``ZSCORE`` hit = a live re-acquire → refresh in one txn (never
+             a second count).
+          3. ``ZCOUNT`` of live members ≥ limit → refuse WITHOUT inserting;
+             else ZADD+EXPIRE in the same txn. Because the check and the
+             insert commit atomically, two racers can never both observe
+             ``count == limit-1`` and both admit.
+
+        Why not ZADD-then-ZRANK (the previous design): the score was captured
+        from the wall clock BEFORE the insert, so a racer descheduled between
+        capture and ZADD could land an EARLIER-scored member after another
+        racer's rank check — both then passed the gate. Measured 11 admits on
+        a 10-cap (CI 3.12 leg, de81786): a starved box widens the
+        capture→insert gap until the overshoot is deterministic.
 
         Consistency: ``count`` is ``ZCOUNT`` over the SAME set, so the
         admission gate and the reported count can never drift. A crash leaves
@@ -318,16 +324,36 @@ def acquire_slot(self, kind: str, slot_key: str, limit: int, ttl: float,
             return True  # fail-open
         zk = self._zcap_k(kind, count_prefix)
         ttl_i = max(1, int(ttl))
-        now = time.time()
         try:
-            r.zremrangebyscore(zk, 0, now)          # 1. evict expired members
-            r.zadd(zk, {slot_key: now + ttl_i})     # 2. claim/refresh
-            r.expire(zk, ttl_i * 2)                 # whole-key idle backstop
-            rank = r.zrank(zk, slot_key)            # 3. total-order position
-            if rank is not None and rank < limit:
-                return True
-            r.zrem(zk, slot_key)                    # over capacity → self-evict
-            return False
+            from redis.exceptions import WatchError
+            for _attempt in range(64):
+                now = time.time()
+                try:
+                    with r.pipeline() as p:
+                        p.watch(zk)
+                        p.zremrangebyscore(zk, 0, now)  # evict expired members
+                        if p.zscore(zk, slot_key) is not None:
+                            # Live re-acquire → refresh only, never a count.
+                            p.multi()
+                            p.zadd(zk, {slot_key: now + ttl_i})
+                            p.expire(zk, ttl_i * 2)
+                            p.execute()
+                            return True
+                        if p.zcount(zk, now, '+inf') >= limit:
+                            p.unwatch()
+                            return False
+                        p.multi()
+                        p.zadd(zk, {slot_key: now + ttl_i})
+                        p.expire(zk, ttl_i * 2)  # whole-key idle backstop
+                        p.execute()
+                        return True
+                except WatchError:
+                    continue  # a concurrent racer touched the set — re-read
+            # Far past any real burst; fail-open like a backend error rather
+            # than refuse a legitimate caller forever.
+            logger.warning('[RuntimeStateStore] acquire_slot watch retries '
+                           'exhausted for %s — fail-open', slot_key)
+            return True
         except Exception as e:
             logger.warning('[RuntimeStateStore] acquire_slot failed (%s) — '
                            'fail-open', e)

tests/test_abort_toolonly_rounds_persist.py

@@ -70,7 +70,11 @@
 
 import pytest
 
-pytestmark = pytest.mark.unit
+# ci_serial: real create_task + persist_task_result write through the shared
+# sqlite pool; under the CI parallel lane's contention the seed write exceeded
+# the 30s busy timeout ('database is locked', de81786 3.12 leg) while passing
+# in seconds uncontended.
+pytestmark = [pytest.mark.unit, pytest.mark.ci_serial]
 
 
 def _seed_conv(conv_id):

tests/test_project_peer.py

@@ -37,6 +37,8 @@
 
 import pytest
 
+from tests._nc_harness import patch_restore as _patch_restore
+
 pytestmark = pytest.mark.unit
 
 HERE = os.path.dirname(os.path.abspath(__file__))
@@ -548,13 +550,12 @@ def test_NC_dispatch_drops_peer_markers(monkeypatch):
     """NC-OBSERVE: no-op the marker-propagation block in dispatch_next_queued →
     the persisted turn loses its peer markers → the observability test FAILS
     (the arrival becomes indistinguishable from user input)."""
-    import importlib
-
     captured = {}
 
     def run():
+        # The harness already swapped the NEUTERED module into sys.modules —
+        # this import resolves to it (no reload; a reload would un-neuter).
         import lib.message_queue as mq
-        importlib.reload(mq)
         monkeypatch.setattr(mq, 'dequeue_next', lambda c: {
             'queueId': 'q1', 'config': {},
             'payload': {'text': 'hi', '_peerMessage': True, '_fromConv': 'cS'}})
@@ -578,41 +579,24 @@ def run():
         "            pass  # NC-OBSERVE (marker propagation disabled)",
         run,
     )
-    import lib.message_queue as mq
-    importlib.reload(mq)
 
 
 # ════════════════════════════════════════════════════════════════════
-#  Source-level NEGATIVE CONTROLS (byte-reverting)
+#  Source-level NEGATIVE CONTROLS (in-memory harness: the neutered module is
+#  compiled into a throwaway sys.modules entry — the shipped file is READ-ONLY)
 # ════════════════════════════════════════════════════════════════════
 
-def _patch_restore(path, old, new, run):
-    with open(path, encoding='utf-8') as f:
-        original = f.read()
-    assert old in original, f'anchor not found in {path}'
-    try:
-        with open(path, 'w', encoding='utf-8') as f:
-            f.write(original.replace(old, new, 1))
-        run()
-    finally:
-        with open(path, 'w', encoding='utf-8') as f:
-            f.write(original)
-    with open(path, encoding='utf-8') as f:
-        assert f.read() == original, 'source not restored byte-identical'
-
-
 def test_NC_storm_guard_noop_breaks_rate_limit(_stub_io):
     """NC-STORM: disable the rate cap in _prune_and_check → the 4th message is
     no longer refused → the storm guard test FAILS."""
-    import importlib
-
     def run():
+        # The harness already swapped the NEUTERED module into sys.modules —
+        # this import resolves to it (no reload; a reload would un-neuter).
         import lib.conversations.project_peer as pp
-        importlib.reload(pp)
-        # Reload re-binds _resolve_target_conv_id to the real (DB-reading) fn;
-        # re-stub to identity so this DB-free NC uses synthetic ids.
+        # The neutered module binds the real (DB-reading) resolver; re-stub to
+        # identity on the swapped module so this DB-free NC uses synthetic ids.
         pp._resolve_target_conv_id = lambda t: ((t or '').strip(), '')
-        # Re-stub via module attrs the reloaded code reads at call time.
+        # Re-stub via module attrs the neutered code reads at call time.
         for i in range(3):
             pp.send_peer_message('/p', 'cA', 'cB', f'm{i}')
         blocked = pp.send_peer_message('/p', 'cA', 'cB', 'm4')
@@ -628,24 +612,20 @@ def run():
         "    return True, kept + [now], 0.0  # NC-STORM (rate cap disabled)",
         run,
     )
-    import lib.conversations.project_peer as pp
-    importlib.reload(pp)
 
 
 def test_NC_audit_gate_noop_allows_unapproved_abort(monkeypatch):
     """NC-GATE: no-op the approval check in _authorize_hard_abort → an
     unapproved hard abort now proceeds → the gate test FAILS."""
-    import importlib
-
     aborted = []
     monkeypatch.setattr('lib.tasks_pkg.manager.abort_running_tasks_for_conv',
                         lambda c, **k: aborted.append(c) or 1)
     monkeypatch.setattr('lib.conversations.project_feed.emit_project_event',
                         lambda *a, **k: None)
 
     def run():
+        # Harness already swapped the NEUTERED module into sys.modules.
         import lib.conversations.project_peer as pp
-        importlib.reload(pp)
         pp._resolve_target_conv_id = lambda t: ((t or '').strip(), '')
         monkeypatch.setattr('lib.conversations.project_peer.audit_log',
                             lambda *a, **k: None)
@@ -662,25 +642,21 @@ def run():
         "    return True, 'approved'  # NC-GATE (approval check disabled)",
         run,
     )
-    import lib.conversations.project_peer as pp
-    importlib.reload(pp)
 
 
 def test_NC_deny_branch_noop_runs_abort_despite_denial(_stub_io, monkeypatch):
     """NC-DENY: no-op the deny branch (treat a falsy approval as approved) → a
     DENIED hard abort now runs the abort anyway → the deny-path test FAILS.
     This proves the deny branch is what actually stops an unapproved kill."""
-    import importlib
-
     aborted = []
     monkeypatch.setattr('lib.tasks_pkg.manager.abort_running_tasks_for_conv',
                         lambda c, **k: aborted.append(c) or 1)
     monkeypatch.setattr('lib.conversations.project_feed.emit_project_event',
                         lambda *a, **k: None)
 
     def run():
+        # Harness already swapped the NEUTERED module into sys.modules.
         import lib.conversations.project_peer as pp
-        importlib.reload(pp)
         pp._resolve_target_conv_id = lambda t: ((t or '').strip(), '')
         monkeypatch.setattr('lib.conversations.project_peer.audit_log',
                             lambda *a, **k: None)
@@ -698,18 +674,14 @@ def run():
         "        approved_by = str(approver).strip() if approver else 'nc-deny-forced'  # NC-DENY",
         run,
     )
-    import lib.conversations.project_peer as pp
-    importlib.reload(pp)
 
 
 def test_NC_join_exclude_noop_leaks_self():
     """NC-JOIN: no-op the exclude_conv filter → a conversation sees ITSELF in
     its own peer list → the self-exclusion test FAILS."""
-    import importlib
-
     def run():
+        # Harness already swapped the NEUTERED module into sys.modules.
         import lib.conversations.project_peer as pp
-        importlib.reload(pp)
         view = pp._join_peers(_peers(), {}, {}, exclude_conv='cA')
         # With the exclusion disabled, cA leaks into its own peer view.
         assert any(v['convId'] == 'cA' for v in view), \
@@ -721,5 +693,3 @@ def run():
         "        if not conv_id:\n            continue  # NC-JOIN (self-exclude disabled)",
         run,
     )
-    import lib.conversations.project_peer as pp
-    importlib.reload(pp)