fix(run-engine): stop an unservable variant pinning the ck virtual-time floor

A concurrency-key variant that has queued work but nothing ready yet, which is
what every nack with a retry backoff produces, stayed registered in ckVtime
holding its old low tag. The floor is the lowest stored tag, so it froze there
while the keys actually being served advanced. New keys register at the floor,
so a key that arrived later started well below the established ones and won
every pass-1 slot until it caught up, which is the starvation the feature is
meant to remove.

The dequeue path now de-registers a variant when it has work but none of it is
ready, alongside the existing GC for variants with no work at all. It stays in
ckIndex, so pass 2 still serves it in age order once its head is ready, and it
rejoins the fair order at the current floor on its next enqueue, nack or serve.
Idle keys no longer hoard priority credit either.

Measured with a control against a treatment on the real dequeue path: with one
future-headed variant present the floor stayed at 0 while served keys reached
25, and a newcomer took 20 of the next 20 serves. With the fix the same run
matches the control, newcomer 5 of 20.

Reported by Devin on #4367.

Author: 1stvamp

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

@@ -5202,6 +5202,13 @@ local function tryServe(ckQueueName)
         redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW
       else
         redis.call('ZADD', ckIndexKey, any[2], ckQueueName)
+        -- The variant has work but none of it is ready yet (a nack backoff, say), so it
+        -- is not competing for service and must not hold the floor down. While it sat in
+        -- ckVtime its low tag pinned the floor, and new keys register at the floor, so a
+        -- key arriving later started far below the established ones and took every pass-1
+        -- slot until it caught up. It re-registers at the floor of the day on its next
+        -- enqueue/nack, or when pass 2 serves it after its head becomes ready.
+        redis.call('ZREM', ckVtimeKey, ckQueueName)
       end
     end
   end

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

@@ -661,45 +661,129 @@ 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;
+  redisTest(
+    "future-scheduled variants are skipped, not advanced, and de-registered",
+    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,
-      });
+        // 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 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 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);
+        expect(messages.some((m) => m.message.concurrencyKey === "future")).toBe(false);
 
-      const futureTag = Number(await queue.redis.zscore(ckVtimeKey, futureVariant));
-      expect(futureTag).toBe(5);
-    } finally {
-      await queue.quit();
+        // 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();
+      } finally {
+        await queue.quit();
+      }
     }
-  });
+  );
+
+  redisTest(
+    "an unservable variant does not pin the floor for later arrivals",
+    async ({ redisContainer }) => {
+      // Regression: a variant with work but nothing ready (a nack backoff is the common
+      // case) used to sit in ckVtime holding the lowest tag. The floor is the minimum
+      // stored tag, so it froze at that value while served keys advanced, and because new
+      // keys register at the floor, a key arriving later started far below the established
+      // ones and won every pass-1 slot until it caught up. That is the starvation this
+      // feature exists to prevent, inverted.
+      const queue = createQueue(redisContainer);
+      try {
+        const t0 = Date.now() - 100_000;
+
+        // Stalled: registered on enqueue, but its head never becomes ready during the test.
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "r-stalled",
+            concurrencyKey: "stalled",
+            timestamp: Date.now() + 60 * 60 * 1000,
+          }),
+          workerQueue: authenticatedEnvDev.id,
+          skipDequeueProcessing: true,
+        });
+
+        for (let i = 0; i < 12; i++) {
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({
+              runId: `r-busy-${i}`,
+              concurrencyKey: "busy",
+              timestamp: t0 + i,
+            }),
+            workerQueue: authenticatedEnvDev.id,
+            skipDequeueProcessing: true,
+          });
+        }
+
+        const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+        for (let call = 0; call < 6; call++) {
+          const messages = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 1);
+          for (const m of messages) {
+            await queue.acknowledgeMessage(authenticatedEnvDev.organization.id, m.messageId, {
+              skipDequeueProcessing: true,
+            });
+          }
+        }
+
+        const busyVariant = variantName("busy");
+        const floorKey = testOptions.keys.ckVtimeFloorKeyFromQueue(busyVariant);
+        const floor = Number((await queue.redis.get(floorKey)) ?? "0");
+
+        // The floor tracked the key that was actually being served.
+        expect(floor).toBeGreaterThan(0);
+
+        // A key arriving now joins level with the established keys rather than underneath
+        // them, so it gets its turn instead of monopolising the fair pass.
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({ runId: "r-newcomer", concurrencyKey: "newcomer", timestamp: t0 }),
+          workerQueue: authenticatedEnvDev.id,
+          skipDequeueProcessing: true,
+        });
+
+        const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(busyVariant);
+        const newcomerTag = Number(await queue.redis.zscore(ckVtimeKey, variantName("newcomer")));
+        const busyTag = Number(await queue.redis.zscore(ckVtimeKey, busyVariant));
+
+        expect(newcomerTag).toBe(floor);
+        expect(busyTag - newcomerTag).toBeLessThanOrEqual(1);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
 
   redisTest(
     "enqueue registers the variant at the current floor with NX",