fix(run-engine): make ck vtime pass 2 discover unregistered variants

At the moment the flag is flipped, :ckVtime is empty and every already-queued
variant is unregistered. The first dequeue has an empty pass 1, so pass 2 serves
and registers up to actualMaxCount variants. From the next call on, pass 1 can
serve one message per registered variant and actualMaxCount is often small, so
pass 1 fills the batch off that cohort alone. Pass 2 was gated on
dequeuedCount < actualMaxCount, so it never ran again, and the rest of the
backlog stayed invisible until a registered variant fully drained or an enqueue
or nack happened to land on it. A key that gets no further work has no other
route into the fair order. Same reachability shape 1a6d1a5 fixed for
registered-but-unservable variants, applied to the unregistered cohort. It also
covers a mixed deploy, where an instance with the flag still off enqueues
through the non-vtime command, and a :ckVtime that expired while ckIndex lived.

Pass 2 now always runs. When the batch is already full it registers the variants
pass 1 could not see, at the floor, instead of serving them, so the next call's
pass 1 leads with them. Serving is still capped at actualMaxCount, so no serve
happens that the old gate would have refused, and the fairness scenarios are
byte-identical: ckSkew, ckTrickle, ckSybil, ckBalanced, ckManyKeys, ckHeavyIdle
and ckStalledNewcomer all report the same numbers as before.

Op cost is one extra fixed op per call. The pass-1 window read doubles as a free
membership set, so nothing is registered twice, and the registrations are
collected into a single variadic ZADD NX rather than one call each. Measured on
the op-count budget test: 11.62 ops per dequeue over the flag-off path, against
10.90 before. The budget comment now counts 8 fixed ops rather than 7.

Devin's suggestion of reserving a batch slot for pass 2 does not fix it. Pass 2
walks ckIndex in age order, so its one reserved slot always lands on a variant
that is already registered and never reaches the cohort that is not. Measured
against the new tests: identical to no fix at all, 12, 16 and 12 calls.

Reported by Devin on #4367.

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

@@ -5106,6 +5106,9 @@ return __qmret(results)
     // 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).
+    // 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.
     this.redis.defineCommand("dequeueMessagesFromCkQueueVtimeTracked", {
       numberOfKeys: 13,
       lua: `
@@ -5264,24 +5267,49 @@ 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,
+-- 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
 for _, ckQueueName in ipairs(vtimeCandidates) do
   if dequeuedCount >= actualMaxCount then break end
   tryServe(ckQueueName, true)
 end
 
 -- Pass 2: fill + discovery in age order (work conservation, mixed-deploy safety).
--- Never runs when pass 1 filled the batch.
-if dequeuedCount < actualMaxCount then
-  -- Clamp to at least 3x so pass 2 never scans fewer index variants than the old command, preserving work conservation regardless of the configured multiplier
-  local pass2Window = math.max(window, actualMaxCount * 3)
-  local ckQueues = redis.call('ZRANGEBYSCORE', ckIndexKey, '-inf', tostring(currentTime), 'LIMIT', 0, pass2Window)
-  for _, ckQueueName in ipairs(ckQueues) do
-    if dequeuedCount >= actualMaxCount then break end
-    if not attempted[ckQueueName] then
+-- Clamp to at least 3x so pass 2 never scans fewer index variants than the old command, preserving work conservation regardless of the configured multiplier
+local pass2Window = math.max(window, actualMaxCount * 3)
+local ckQueues = redis.call('ZRANGEBYSCORE', ckIndexKey, '-inf', tostring(currentTime), 'LIMIT', 0, pass2Window)
+-- NEW: pass 2 runs even when pass 1 filled the batch. A variant that reached
+-- ckIndex without a ckVtime entry (queued before the flag went on, enqueued by an
+-- instance that still has it off, or left behind by an expired ckVtime) is invisible
+-- to pass 1, and pass 1 filling the batch off the registered variants alone kept it
+-- that way until one of them drained. With no batch slot left we only register it,
+-- at the floor, so the next call's pass 1 leads with it. Serving is still capped at
+-- actualMaxCount, so this adds no serve the old gate would have refused.
+local discovered = nil
+for _, ckQueueName in ipairs(ckQueues) do
+  if not attempted[ckQueueName] then
+    if dequeuedCount < actualMaxCount then
       tryServe(ckQueueName, false)
+    elseif not registered[ckQueueName] then
+      -- Collected into one variadic ZADD: discovery costs at most a single op per
+      -- call however many variants it registers. Skipping attempted matters:
+      -- tryServe GCs a drained variant out of both indexes, and re-adding it here
+      -- would resurrect a ckVtime entry with no ckIndex member.
+      if discovered == nil then discovered = {ckVtimeKey, 'NX'} end
+      table.insert(discovered, tostring(floor))
+      table.insert(discovered, ckQueueName)
     end
   end
 end
+if discovered ~= nil then
+  redis.call('ZADD', unpack(discovered))
+end
 
 -- NEW: persist floor and refresh TTLs
 if minServableTag ~= nil and minServableTag > floor then

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

@@ -1273,4 +1273,140 @@ describe("CK virtual-time (SFQ) dequeue", () => {
       }
     }
   );
+
+  // Cold start: the flag is turned on over a backlog that was queued while it was
+  // off, so every variant is in ckIndex and none is in :ckVtime. Pass 1 can only
+  // see registered variants, so the cohort pass 2 happens to register on the first
+  // call is the only cohort pass 1 ever serves; while that cohort keeps the batch
+  // full, pass 2 never runs again and the rest of the backlog is unreachable until
+  // the cohort drains. A variant that gets no further enqueues and no nacks has no
+  // other route into the fair order, so the bound below is the whole guarantee.
+  //
+  // The same shape covers a mixed deploy (an instance with the flag still off
+  // enqueues through the non-vtime command) and a :ckVtime that expired while
+  // ckIndex survived.
+  describe("cold start over an unregistered backlog", () => {
+    type ColdStartShape = {
+      variants: number;
+      perVariant: number;
+      maxCount: number;
+      scanWindowMultiplier?: number;
+      // Calls the coldest variant (newest head, so last in the age order pass 2
+      // walks) may wait before its first serve.
+      bound: number;
+    };
+
+    // Enqueues the backlog with the flag OFF, then reopens the same keyspace with
+    // it ON and drains, recording the call each variant was first served on.
+    async function runColdStart(redisContainer: any, shape: ColdStartShape) {
+      const t0 = Date.now() - 500_000;
+      const cks = Array.from(
+        { length: shape.variants },
+        (_, k) => `ck${String(k).padStart(2, "0")}`
+      );
+
+      const before = createQueue(redisContainer, null);
+      try {
+        for (let i = 0; i < shape.perVariant; i++) {
+          for (let k = 0; k < cks.length; k++) {
+            await before.enqueueMessage({
+              env: authenticatedEnvDev,
+              message: makeMessage({
+                runId: `${cks[k]}-${i}`,
+                concurrencyKey: cks[k],
+                // 1s of head-age spacing per variant, so ck00 is oldest and the
+                // age order never reshuffles as heads advance by 1ms per serve.
+                timestamp: t0 + k * 1_000 + i,
+              }),
+              workerQueue: authenticatedEnvDev.id,
+              skipDequeueProcessing: true,
+            });
+          }
+        }
+      } finally {
+        await before.quit();
+      }
+
+      const after = createQueue(redisContainer, {
+        ...(shape.scanWindowMultiplier === undefined
+          ? {}
+          : { scanWindowMultiplier: shape.scanWindowMultiplier }),
+      });
+      try {
+        const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(variantName(cks[0]!));
+        const ckIndexKey = testOptions.keys.ckIndexKeyFromQueue(variantName(cks[0]!));
+        // The premise: the whole backlog is in the age index and nothing is in the
+        // fair order.
+        expect(await after.redis.zcard(ckIndexKey)).toBe(shape.variants);
+        expect(await after.redis.zcard(ckVtimeKey)).toBe(0);
+
+        const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+        const total = shape.variants * shape.perVariant;
+        const firstServeCall = new Map<string, number>();
+        let served = 0;
+
+        for (let call = 0; call < total + 10 && served < total; call++) {
+          const messages = await after.testDequeueFromMasterQueue(
+            shard,
+            authenticatedEnvDev.id,
+            shape.maxCount
+          );
+          for (const m of messages) {
+            const ck = m.message.concurrencyKey!;
+            if (!firstServeCall.has(ck)) firstServeCall.set(ck, call);
+            served++;
+            // Ack immediately so env concurrency never gates a serve: the only
+            // thing under test is reachability.
+            await after.acknowledgeMessage(authenticatedEnvDev.organization.id, m.messageId, {
+              skipDequeueProcessing: true,
+            });
+          }
+        }
+
+        return { firstServeCall, served, total, coldest: cks[cks.length - 1]! };
+      } finally {
+        await after.quit();
+      }
+    }
+
+    // Bounds are the measured value, not headroom: the harness has no wall-clock
+    // wait and no randomness. The pre-fix figure in each comment is what the same
+    // shape did when pass 2 was gated on dequeuedCount < actualMaxCount.
+    const shapes: [string, ColdStartShape][] = [
+      // Registered cohort (5) smaller than the backlog (8), both inside the
+      // pass-1 window (15) and the pass-2 scan window (15). Pre-fix: call 12.
+      ["8 variants, batch 5", { variants: 8, perVariant: 12, maxCount: 5, bound: 1 }],
+      // Backlog exactly fills the pass-1 window (12), so discovery has to land in
+      // more than one call. Pre-fix: call 16.
+      ["12 variants, batch 4", { variants: 12, perVariant: 8, maxCount: 4, bound: 2 }],
+      // scanWindowMultiplier 1 puts the pass-1 window (5) below the backlog, so
+      // the window read can no longer tell which variants are already registered
+      // and discovery falls back to the NX. Pre-fix: call 12.
+      [
+        "15 variants, batch 5, narrow fair window",
+        { variants: 15, perVariant: 6, maxCount: 5, scanWindowMultiplier: 1, bound: 2 },
+      ],
+    ];
+
+    for (const [name, shape] of shapes) {
+      redisTest(
+        `${name}: the coldest variant is served within ${shape.bound + 1} calls`,
+        async ({ redisContainer }) => {
+          const { firstServeCall, served, total, coldest } = await runColdStart(
+            redisContainer,
+            shape
+          );
+
+          // Work conservation: the backlog still drains completely.
+          expect(served).toBe(total);
+          expect(firstServeCall.size).toBe(shape.variants);
+
+          expect(
+            firstServeCall.get(coldest),
+            `coldest variant ${coldest} first served on call ${firstServeCall.get(coldest)}`
+          ).toBeLessThanOrEqual(shape.bound);
+        }
+      );
+    }
+  });
 });

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

@@ -346,10 +346,12 @@ describe("CK virtual-time concurrency and op-count budget", () => {
       expect(off.served).toBe(cks.length * perKey);
       expect(on.served).toBe(cks.length * perKey);
 
-      // Per dequeue call the vtime path adds at worst 7 fixed ops: GET floor,
-      // ZRANGE min, ZRANGE window, the pass-2 ZRANGEBYSCORE, SET floor,
-      // EXISTS ckVtime, EXPIRE ckVtime — plus per serve one ZSCORE and one ZADD.
-      const budget = dequeueCalls * (7 + 2 * maxCount);
+      // Per dequeue call the vtime path adds at worst 8 fixed ops: GET floor,
+      // ZRANGE min, ZRANGE window, the pass-2 ZRANGEBYSCORE, the pass-2
+      // discovery ZADD, SET floor, EXISTS ckVtime, EXPIRE ckVtime, plus per
+      // serve one ZSCORE and one ZADD. The discovery ZADD is variadic, so it
+      // stays a single op however many variants one call registers.
+      const budget = dequeueCalls * (8 + 2 * maxCount);
       expect(
         on.totalCalls,
         `on_total ${on.totalCalls} exceeds off_total ${off.totalCalls} + budget ${budget}`