fix(run-engine): stop unready ck variants blocking the fair pass, and idle polls writing

Pass 1 now steps over a variant whose head is scheduled in the future without
spending one of its window slots, so a retry storm across enough keys can no
longer fill the window with variants that cannot be served and freeze the
virtual-time floor. The variant stays registered and stays scanned, which is
what keeps it reachable; only the budget is spared. The scan is capped at twice
the window, so a wider block still degrades to pass 2's age order.

A dequeue that serves nothing now persists nothing. Both things that block would
write are re-derivable: minServableTag is only set inside a successful serve, and
discovery only runs once the batch is full, so the floor read-repair is recomputed
from ckVtime on the next call anyway.

Refits the two tests whose premise these change: the freeze test now pins the
residual beyond the scan cap, and the floor test pins that a zero-serve call
persists nothing while the repair still lands on the next serving call.

internal-packages/run-engine/src/run-queue/index.ts

@@ -5105,7 +5105,9 @@ return __qmret(results)
     // :ckVtime / :ckVtimeFloor keys hold virtual times; ckIndex and the master
     // queue keep their timestamp score domain. The per-candidate serve body is a
     // verbatim copy of dequeueMessagesFromCkQueueTracked's, with the marked NEW
-    // lines added (tag advance on serve, ZREM ckVtime on GC, floor advance from pass 1).
+    // lines added (tag advance on serve, ZREM ckVtime on GC, floor advance from pass 1,
+    // and the notReady report that lets pass 1 step over a future-headed variant without
+    // spending a window slot on it).
     // Pass 2 always runs: when the batch is already full it registers the variants
     // pass 1 could not see rather than serving them, which is what keeps a backlog
     // queued before the flag went on from being unreachable.
@@ -5164,6 +5166,10 @@ if actualMaxCount <= 0 then
 end
 
 local window = actualMaxCount * windowMultiplier
+-- Pass 1 reads further than it will spend, so a variant whose head is scheduled in the
+-- future can be passed over without costing a window slot. Capped rather than unbounded:
+-- a block wider than this still degrades to pass 2's age order, which is safe.
+local scanLimit = window * 2
 
 -- Floor only ever rises, by two independent routes: to the lowest tag on record (repairs
 -- a floor that was lost while ckVtime survived), and to the lowest tag actually servable
@@ -5235,6 +5241,10 @@ local function tryServe(ckQueueName, mayRaiseFloor)
           -- Pass 1 only: it walks in ascending tag order, so anything it has not visited
           -- sits above this. Pass 2 goes by message age, so its tag says nothing about the
           -- entries it skipped and must not move the floor over them.
+          -- Pass 1 now steps over future-headed variants below this tag, so the floor can
+          -- rise past one of them. That is the same forfeiture a variant at its concurrency
+          -- ceiling already takes: it keeps its entry, loses the sub-floor credit, and is
+          -- clamped up to the floor when it next becomes servable.
           if mayRaiseFloor and (minServableTag == nil or tag < minServableTag) then
             minServableTag = tag
           end
@@ -5260,24 +5270,37 @@ local function tryServe(ckQueueName, mayRaiseFloor)
         redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW
       else
         redis.call('ZADD', ckIndexKey, any[2], ckQueueName)
+        -- NEW: backlog, but the head is scheduled later, so nothing here is servable this
+        -- call. The readiness is already known from the ZRANGEBYSCORE above, so reporting
+        -- it costs nothing and lets pass 1 decline to spend a window slot on it.
+        return 'notReady'
       end
     end
   end
 end
 
 -- Pass 1: fair order (lowest virtual start tag first)
-local vtimeCandidates = redis.call('ZRANGE', ckVtimeKey, 0, window - 1)
--- NEW: the window read doubles as a free membership set for pass 2's discovery
--- step. It is complete whenever ckVtime holds no more than window variants,
+local vtimeCandidates = redis.call('ZRANGE', ckVtimeKey, 0, scanLimit - 1)
+-- NEW: the scan read doubles as a free membership set for pass 2's discovery
+-- step. It is complete whenever ckVtime holds no more than scanLimit variants,
 -- which is the common case; when it is truncated the discovery ZADD is NX so the
 -- variants it cannot rule out cost correctness nothing.
 local registered = {}
 for _, ckQueueName in ipairs(vtimeCandidates) do
   registered[ckQueueName] = true
 end
+-- NEW: a variant whose head is scheduled in the future stays registered and stays
+-- scanned, it just does not spend one of the window's slots. Without this a retry storm
+-- across enough keys fills the window with variants that cannot be served, pass 1 serves
+-- nothing, and because minServableTag is the only route that can lift the floor over a
+-- stale low tag, the floor freezes for as long as the storm lasts. Every other outcome
+-- (served, gated on concurrency, drained, reaped) still spends a slot, as before.
+local windowBudget = window
 for _, ckQueueName in ipairs(vtimeCandidates) do
-  if dequeuedCount >= actualMaxCount then break end
-  tryServe(ckQueueName, true)
+  if dequeuedCount >= actualMaxCount or windowBudget <= 0 then break end
+  if tryServe(ckQueueName, true) ~= 'notReady' then
+    windowBudget = windowBudget - 1
+  end
 end
 
 -- Pass 2: fill + discovery in age order (work conservation, mixed-deploy safety).
@@ -5311,13 +5334,21 @@ if discovered ~= nil then
   redis.call('ZADD', unpack(discovered))
 end
 
--- NEW: persist floor and refresh TTLs
-if minServableTag ~= nil and minServableTag > floor then
-  floor = minServableTag
-end
-redis.call('SET', ckVtimeFloorKey, tostring(floor), 'EX', stateTtl)
-if redis.call('EXISTS', ckVtimeKey) == 1 then
-  redis.call('EXPIRE', ckVtimeKey, stateTtl)
+-- NEW: persist floor and refresh TTLs. A call that served nothing writes nothing: the two
+-- things this block would persist are both re-derivable, since minServableTag is only set
+-- inside a successful serve and pass 2's discovery only runs once the batch is full, so
+-- the only floor movement on a zero-serve call is the min-tag read-repair, which is
+-- recomputed from ckVtime at the top of every call anyway. Idle polling a queue whose work
+-- is all future-scheduled or concurrency-gated therefore costs no writes, matching the old
+-- command's early return.
+if dequeuedCount > 0 then
+  if minServableTag ~= nil and minServableTag > floor then
+    floor = minServableTag
+  end
+  redis.call('SET', ckVtimeFloorKey, tostring(floor), 'EX', stateTtl)
+  if redis.call('EXISTS', ckVtimeKey) == 1 then
+    redis.call('EXPIRE', ckVtimeKey, stateTtl)
+  end
 end
 
 -- Rebalance master queue (ckIndex keeps its timestamp domain)

internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts

@@ -301,8 +301,7 @@ describe("CK virtual-time (SFQ) dequeue", () => {
         prevFloor = floor;
       }
 
-      // Final settle: gate both variants (limit 1 + an occupied slot) so nothing
-      // is served (no advance), and the floor read-repairs up to the current min tag.
+      // Final settle: gate both variants (limit 1 + an occupied slot) so nothing is served.
       await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, QUEUE, 1);
       for (const ck of cks) {
         await queue.redis.sadd(
@@ -312,12 +311,28 @@ describe("CK virtual-time (SFQ) dequeue", () => {
       }
       await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 2);
 
-      const floorAfter = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0");
-      expect(floorAfter).toBeGreaterThanOrEqual(prevFloor);
-
       const minEntry = await queue.redis.zrange(ckVtimeKey, 0, 0, "WITHSCORES");
       const minTag = Number(minEntry[1]);
-      expect(floorAfter).toBe(minTag);
+      expect(minTag).toBeGreaterThan(prevFloor);
+
+      // A call that serves nothing persists nothing. The read-repair to the min tag is
+      // recomputed from ckVtime at the top of every call, so leaving it unwritten here
+      // costs nothing and keeps an idle poll free of writes.
+      const floorWhileGated = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0");
+      expect(floorWhileGated).toBe(prevFloor);
+
+      // Free a slot: the next serving call persists the repaired floor, so the repair
+      // itself is intact, it is only the write that waits for a serve.
+      for (const ck of cks) {
+        await queue.redis.srem(
+          testOptions.keys.queueCurrentConcurrencyKeyFromQueue(variantName(ck)),
+          "occupant"
+        );
+      }
+      await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 2);
+
+      const floorAfter = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0");
+      expect(floorAfter).toBeGreaterThanOrEqual(minTag);
       expect(floorAfter).toBeGreaterThan(10);
     } finally {
       await queue.quit();
@@ -917,14 +932,14 @@ describe("CK virtual-time (SFQ) dequeue", () => {
   );
 
   redisTest(
-    "a pass-1 window filled entirely with unservable variants degrades to age order, and recovers",
+    "a scan filled entirely with unservable variants degrades to age order, and recovers",
     async ({ redisContainer }) => {
-      // The single-stalled case above is handled by the minServableTag route: a servable
-      // variant inside the window raises the floor over the stalled tag. That route needs
-      // pass 1 to serve something. With every window slot (actualMaxCount * multiplier,
-      // 3 here) held by unservable variants, pass 1 serves nothing, minServableTag stays
-      // nil, and the min-tag route is pinned by those same stalled tags, so the floor
-      // cannot move for as long as the block lasts.
+      // Pass 1 steps over a future-headed variant without spending a window slot, so the
+      // window alone can no longer be blocked. The scan behind it is capped though, at
+      // scanLimit = 2 * window (6 here), and this is the residual: with every scanned
+      // position held by an unservable variant, pass 1 still serves nothing, minServableTag
+      // stays nil, and the min-tag route is pinned by those same stalled tags, so the floor
+      // cannot move until the block thins out.
       //
       // Two properties of that state are worth pinning down. It stays work-conserving:
       // pass 2 keeps serving in age order, which is the flag-off behaviour, so a full
@@ -937,9 +952,9 @@ describe("CK virtual-time (SFQ) dequeue", () => {
         const t0 = Date.now() - 100_000;
 
         // Names decide tie order at equal tags, and the point of the fixture is that the
-        // blockers hold every window slot: a0/a1/a2 sort below zbusy, so the window read
-        // returns only them.
-        const blockers = ["a0", "a1", "a2"];
+        // blockers hold every scanned position: a0..a5 sort below zbusy, so the scan read
+        // returns only them and zbusy is never reached.
+        const blockers = ["a0", "a1", "a2", "a3", "a4", "a5"];
         for (const ck of blockers) {
           await queue.enqueueMessage({
             env: authenticatedEnvDev,
@@ -1035,8 +1050,12 @@ describe("CK virtual-time (SFQ) dequeue", () => {
           }
         }
 
+        // The unblocked cohort plus the arrival is 7 keys sitting at the floor against an
+        // incumbent on FREEZE_CALLS, so levelling up costs 7 * FREEZE_CALLS serves before
+        // the incumbent competes again. Drain comfortably past that rather than right on
+        // the boundary, or the bound below is measuring the cutoff instead of the debt.
         const servedAfter: Record<string, number> = {};
-        for (let call = 0; call < 60; call++) {
+        for (let call = 0; call < (blockers.length + 1) * FREEZE_CALLS + 60; call++) {
           for (const ck of await drainOne()) {
             servedAfter[ck] = (servedAfter[ck] ?? 0) + 1;
           }