fix(run-engine): keep unservable ck variants in the fair order

Reverts the de-registration added earlier on this branch. Dropping a variant from
ckVtime when it had work but nothing ready stranded it: pass 1 is the only reader
of ckVtime, and pass 2 is skipped whenever pass 1 fills the batch, so on a queue
busy enough to keep filling it the variant was never looked at again. A blind
review measured one sitting unserved for over two thousand calls after its head
became ready, and every nack backoff produces exactly that shape, so a single
steady key could hold up another key's retries indefinitely. That is worse than
the floor pinning it was meant to address.

The floor advance from servable variants handles both cases on its own, so the
de-registration bought nothing. It now also only takes its bound from pass 1.
Pass 2 picks candidates by message age, so its tag implies nothing about the
entries it skipped, and letting it move the floor stepped over registered
variants that were servable and simply never visited, confiscating their credit
on the next serve.

Adds the regression test the earlier tests were missing. They proved the variant
was evicted but never that it came back, which is the half that was broken.

Author: 1stvamp

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

@@ -5163,7 +5163,7 @@ 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).
+    // lines added (tag advance on serve, ZREM ckVtime on GC, floor advance from pass 1).
     this.redis.defineCommand("dequeueMessagesFromCkQueueVtimeTracked", {
       numberOfKeys: 13,
       lua: `
@@ -5240,7 +5240,7 @@ local attempted = {}
 
 -- Per-candidate serve. Body is dequeueMessagesFromCkQueueTracked's per-candidate
 -- block, verbatim, with the marked NEW lines added.
-local function tryServe(ckQueueName)
+local function tryServe(ckQueueName, mayRaiseFloor)
   attempted[ckQueueName] = true
   local fullQueueKey = keyPrefix .. ckQueueName
 
@@ -5287,8 +5287,10 @@ local function tryServe(ckQueueName)
           local weight = 1
           local tag = tonumber(redis.call('ZSCORE', ckVtimeKey, ckQueueName) or floor)
           if tag < floor then tag = floor end
-          -- Pass 1 walks in ascending tag order, so anything unvisited is above this.
-          if minServableTag == nil or tag < minServableTag then
+          -- 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.
+          if mayRaiseFloor and (minServableTag == nil or tag < minServableTag) then
             minServableTag = tag
           end
           redis.call('ZADD', ckVtimeKey, tostring(tag + (quantum / weight)), ckQueueName)
@@ -5313,9 +5315,6 @@ local function tryServe(ckQueueName)
         redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW
       else
         redis.call('ZADD', ckIndexKey, any[2], ckQueueName)
-        -- Work but nothing ready (a nack backoff): not competing, so drop it from the fair
-        -- order rather than let it hoard credit. Rejoins at the floor when next served.
-        redis.call('ZREM', ckVtimeKey, ckQueueName)
       end
     end
   end
@@ -5325,7 +5324,7 @@ end
 local vtimeCandidates = redis.call('ZRANGE', ckVtimeKey, 0, window - 1)
 for _, ckQueueName in ipairs(vtimeCandidates) do
   if dequeuedCount >= actualMaxCount then break end
-  tryServe(ckQueueName)
+  tryServe(ckQueueName, true)
 end
 
 -- Pass 2: fill + discovery in age order (work conservation, mixed-deploy safety).
@@ -5337,7 +5336,7 @@ if dequeuedCount < actualMaxCount then
   for _, ckQueueName in ipairs(ckQueues) do
     if dequeuedCount >= actualMaxCount then break end
     if not attempted[ckQueueName] then
-      tryServe(ckQueueName)
+      tryServe(ckQueueName, false)
     end
   end
 end

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

@@ -661,48 +661,110 @@ describe("CK virtual-time (SFQ) dequeue", () => {
     }
   );
 
+  redisTest("future-scheduled variants are skipped without advance", async ({ redisContainer }) => {
+    const queue = createQueue(redisContainer);
+    try {
+      const t0 = Date.now() - 100_000;
+
+      // a normal ready variant so the :ck:* wildcard is selected from the master queue
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r-now", concurrencyKey: "now", timestamp: t0 }),
+        workerQueue: authenticatedEnvDev.id,
+        skipDequeueProcessing: true,
+      });
+      // a future-scheduled variant
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({
+          runId: "r-future",
+          concurrencyKey: "future",
+          timestamp: Date.now() + 60_000,
+        }),
+        workerQueue: authenticatedEnvDev.id,
+        skipDequeueProcessing: true,
+      });
+
+      const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(variantName("now"));
+      const futureVariant = variantName("future");
+      await queue.redis.zadd(ckVtimeKey, 0, variantName("now"), 5, futureVariant);
+
+      const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+      const messages = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 10);
+
+      expect(messages.some((m) => m.message.concurrencyKey === "future")).toBe(false);
+
+      // Not served, so not charged a quantum. It stays registered: pass 1 is the only
+      // path that can reach it, so de-registering it would strand it while pass 1 is
+      // busy. It no longer holds the floor down, which the floor tests cover.
+      const futureTag = Number(await queue.redis.zscore(ckVtimeKey, futureVariant));
+      expect(futureTag).toBe(5);
+    } finally {
+      await queue.quit();
+    }
+  });
+
   redisTest(
-    "future-scheduled variants are skipped, not advanced, and de-registered",
+    "a variant whose backoff elapses is served even while pass 1 stays full",
+    { timeout: 120_000 },
     async ({ redisContainer }) => {
+      // Pass 1 is the only path that reads ckVtime, and pass 2 is skipped whenever pass 1
+      // fills the batch. Dropping a not-ready variant from ckVtime therefore stranded it
+      // for as long as any other key kept the batch full: measured at over 2000 calls.
       const queue = createQueue(redisContainer);
       try {
         const t0 = Date.now() - 100_000;
 
-        // a normal ready variant so the :ck:* wildcard is selected from the master queue
-        await queue.enqueueMessage({
-          env: authenticatedEnvDev,
-          message: makeMessage({ runId: "r-now", concurrencyKey: "now", timestamp: t0 }),
-          workerQueue: authenticatedEnvDev.id,
-          skipDequeueProcessing: true,
-        });
-        // a future-scheduled variant
+        for (let k = 0; k < 3; k++) {
+          for (let i = 0; i < 40; i++) {
+            await queue.enqueueMessage({
+              env: authenticatedEnvDev,
+              message: makeMessage({
+                runId: `b${k}-${i}`,
+                concurrencyKey: `b${k}`,
+                timestamp: t0 + i,
+              }),
+              workerQueue: authenticatedEnvDev.id,
+              skipDequeueProcessing: true,
+            });
+          }
+        }
         await queue.enqueueMessage({
           env: authenticatedEnvDev,
           message: makeMessage({
-            runId: "r-future",
-            concurrencyKey: "future",
-            timestamp: Date.now() + 60_000,
+            runId: "stalled-1",
+            concurrencyKey: "stalled",
+            timestamp: Date.now() + 400,
           }),
           workerQueue: authenticatedEnvDev.id,
           skipDequeueProcessing: true,
         });
 
-        const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(variantName("now"));
-        const futureVariant = variantName("future");
-        await queue.redis.zadd(ckVtimeKey, 0, variantName("now"), 5, futureVariant);
-
         const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
-        const messages = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 10);
+        const drain = async (calls: number) => {
+          let servedStalled = false;
+          for (let call = 0; call < calls; call++) {
+            const messages = await queue.testDequeueFromMasterQueue(
+              shard,
+              authenticatedEnvDev.id,
+              1
+            );
+            for (const m of messages) {
+              if (m.message.concurrencyKey === "stalled") servedStalled = true;
+              await queue.acknowledgeMessage(authenticatedEnvDev.organization.id, m.messageId, {
+                skipDequeueProcessing: true,
+              });
+            }
+          }
+          return servedStalled;
+        };
 
-        expect(messages.some((m) => m.message.concurrencyKey === "future")).toBe(false);
+        // Let the incumbents advance so the stalled variant holds the lowest tag, which is
+        // what pulls it into the pass-1 window and onto the skip path.
+        await drain(6);
+        await new Promise((resolve) => setTimeout(resolve, 700));
 
-        // Not served, so never charged a quantum: its tag is not advanced past the 5 it
-        // was seeded with. It is de-registered instead, because a variant with no ready
-        // work is not competing and must not hold the floor down (a pinned floor is what
-        // let a later arrival register underneath the established keys and take every
-        // pass-1 slot). It rejoins at the floor of the day once it has ready work.
-        const futureTag = await queue.redis.zscore(ckVtimeKey, futureVariant);
-        expect(futureTag).toBeNull();
+        expect(await drain(60)).toBe(true);
       } finally {
         await queue.quit();
       }