fix(run-engine): advance the ck virtual-time floor from servable variants

The floor only tracked the lowest tag on record, so any registered variant that
could not be served held it there: one sitting at its own concurrency ceiling, or
one whose head is not ready yet. The keys actually being served advanced past it,
and since new keys register at the floor, a later arrival started underneath the
incumbents and took the fair pass until it caught up.

The floor now also rises to the lowest tag that was servable on the call. Pass 1
walks candidates in ascending tag order, so that is a safe lower bound. The repair
from the lowest tag on record stays, because both routes only ever raise it and it
still recovers a floor that was lost while ckVtime survived.

Also adds the fairness scenario the suite was missing. None of the six scenarios
nacked or used future-scored messages, so a stalled variant never existed and this
class of bug could not show up. In the window after it lands the latecomer now
takes 5 of 20 serves, against 12 of 20 without the fix.

Author: 1stvamp

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

@@ -5117,7 +5117,10 @@ end
 
 local window = actualMaxCount * windowMultiplier
 
--- Monotonic floor, advanced to the minimum stored virtual-time tag
+-- 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
+-- this call (minServableTag). The second route matters because an unservable variant
+-- keeps a stale low tag, which left the first route unable to advance at all.
 local floor = tonumber(redis.call('GET', ckVtimeFloorKey) or '0')
 local minEntry = redis.call('ZRANGE', ckVtimeKey, 0, 0, 'WITHSCORES')
 if #minEntry > 0 then
@@ -5126,6 +5129,7 @@ if #minEntry > 0 then
     floor = minTag
   end
 end
+local minServableTag = nil
 
 local results = {}
 local dequeuedCount = 0
@@ -5180,6 +5184,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
+            minServableTag = tag
+          end
           redis.call('ZADD', ckVtimeKey, tostring(tag + (quantum / weight)), ckQueueName)
         end
       else
@@ -5202,12 +5210,8 @@ 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.
+        -- 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
@@ -5236,6 +5240,9 @@ if dequeuedCount < actualMaxCount then
 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)

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

@@ -709,6 +709,75 @@ describe("CK virtual-time (SFQ) dequeue", () => {
     }
   );
 
+  redisTest(
+    "a variant at its concurrency ceiling does not pin the floor",
+    async ({ redisContainer }) => {
+      // A saturated variant stops advancing but keeps its tag, which used to hold the
+      // floor down for everyone arriving later.
+      const queue = createQueue(redisContainer);
+      try {
+        const t0 = Date.now() - 100_000;
+
+        // Per-key ceiling of 1, well under the env limit, so hog gates on its own account
+        // rather than by exhausting env capacity.
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, QUEUE, 1);
+
+        for (let i = 0; i < 12; i++) {
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({ runId: `r-hog-${i}`, concurrencyKey: "hog", timestamp: t0 + i }),
+            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 + 500 + i,
+            }),
+            workerQueue: authenticatedEnvDev.id,
+            skipDequeueProcessing: true,
+          });
+        }
+
+        const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+        const hogVariant = variantName("hog");
+        const busyVariant = variantName("busy");
+        const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(busyVariant);
+        const floorKey = testOptions.keys.ckVtimeFloorKeyFromQueue(busyVariant);
+
+        // Ack only busy, so hog accumulates in-flight messages until it is gated.
+        for (let call = 0; call < 10; call++) {
+          const messages = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 2);
+          for (const m of messages) {
+            if (m.message.concurrencyKey === "busy") {
+              await queue.acknowledgeMessage(authenticatedEnvDev.organization.id, m.messageId, {
+                skipDequeueProcessing: true,
+              });
+            }
+          }
+        }
+
+        const hogTag = Number(await queue.redis.zscore(ckVtimeKey, hogVariant));
+        const busyTag = Number(await queue.redis.zscore(ckVtimeKey, busyVariant));
+        const floor = Number((await queue.redis.get(floorKey)) ?? "0");
+
+        // hog is still registered (it has ready work and will be served when a slot
+        // frees), it has simply stopped advancing while saturated.
+        expect(hogTag).not.toBeNaN();
+        expect(busyTag).toBeGreaterThan(hogTag);
+
+        // The floor followed the key that was actually being served, not the stalled one.
+        expect(floor).toBeGreaterThan(hogTag);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
   redisTest(
     "an unservable variant does not pin the floor for later arrivals",
     async ({ redisContainer }) => {

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

@@ -93,7 +93,15 @@ function makeMessage(overrides: Partial<InputPayload> = {}): InputPayload {
   };
 }
 
-type ScenarioMessage = { runId: string; ck: string; timestamp: number };
+type ScenarioMessage = {
+  runId: string;
+  ck: string;
+  timestamp: number;
+  // Enqueued at the top of this step instead of before step 0, for late arrivals.
+  enqueueAtStep?: number;
+  // Head never becomes ready during the run, so it is not expected to drain.
+  neverReady?: boolean;
+};
 
 type Scenario = {
   name: string;
@@ -138,7 +146,7 @@ async function runScenario(
     };
     await queue.updateEnvConcurrencyLimits(env);
 
-    for (const msg of scenario.messages) {
+    const enqueue = async (msg: ScenarioMessage) => {
       await queue.enqueueMessage({
         env,
         message: makeMessage({
@@ -149,13 +157,18 @@ async function runScenario(
         workerQueue: env.id,
         skipDequeueProcessing: true,
       });
+    };
+
+    for (const msg of scenario.messages) {
+      if (msg.enqueueAtStep === undefined) await enqueue(msg);
     }
 
     const shard = testOptions.keys.masterQueueShardForEnvironment(env.id, 2);
-    const total = scenario.messages.length;
+    const total = scenario.messages.filter((m) => !m.neverReady).length;
 
     const remaining = new Map<string, number>();
     for (const m of scenario.messages) {
+      if (m.neverReady) continue;
       remaining.set(m.ck, (remaining.get(m.ck) ?? 0) + 1);
     }
 
@@ -165,6 +178,10 @@ async function runScenario(
     let drainStep = -1;
 
     for (let step = 0; step < scenario.maxSteps && serves.length < total; step++) {
+      for (const msg of scenario.messages) {
+        if (msg.enqueueAtStep === step) await enqueue(msg);
+      }
+
       // evaluated before the dequeue: does this step have cross-key contention?
       let keysWithBacklog = 0;
       for (const count of remaining.values()) {
@@ -221,10 +238,11 @@ function firstServeStep(result: ScenarioResult, matches: (ck: string) => boolean
 
 // No loss and no double-serve, in both runs.
 function assertConservation(scenario: Scenario, on: ScenarioResult, off: ScenarioResult) {
-  expect(on.serves.length).toBe(scenario.messages.length);
-  expect(off.serves.length).toBe(scenario.messages.length);
-  expect(new Set(on.serves.map((s) => s.messageId)).size).toBe(scenario.messages.length);
-  expect(new Set(off.serves.map((s) => s.messageId)).size).toBe(scenario.messages.length);
+  const expected = scenario.messages.filter((m) => !m.neverReady).length;
+  expect(on.serves.length).toBe(expected);
+  expect(off.serves.length).toBe(expected);
+  expect(new Set(on.serves.map((s) => s.messageId)).size).toBe(expected);
+  expect(new Set(off.serves.map((s) => s.messageId)).size).toBe(expected);
 }
 
 function debugLog(name: string, data: Record<string, unknown>) {
@@ -528,4 +546,71 @@ describe("CK virtual-time fairness on the real batched dequeue path", () => {
       expect(on.drainStep).toBe(off.drainStep);
     }
   );
+
+  redisTest(
+    "ckStalledNewcomer: a stalled variant does not let a late arrival starve the incumbents",
+    { timeout: 120_000 },
+    async ({ redisContainer }) => {
+      // The case the other five scenarios cannot express: a variant that is registered but
+      // never servable (its head stays in the future, which is what a nack backoff leaves
+      // behind) used to freeze the virtual-time floor, so the late arrival registered far
+      // below the incumbents and took every fair-pass slot until it caught up.
+      const t0 = Date.now() - 100_000;
+      const messages: ScenarioMessage[] = [];
+
+      messages.push({
+        runId: "stalled-0",
+        ck: "stalled",
+        timestamp: Date.now() + 60 * 60 * 1000,
+        neverReady: true,
+      });
+
+      for (let k = 0; k < 3; k++) {
+        for (let i = 0; i < 40; i++) {
+          messages.push({ runId: `inc-${k}-${i}`, ck: `incumbent-${k}`, timestamp: t0 + i });
+        }
+      }
+
+      // Arrives once the incumbents have advanced well past the stalled variant's tag.
+      for (let i = 0; i < 20; i++) {
+        messages.push({
+          runId: `late-${i}`,
+          ck: "latecomer",
+          timestamp: t0 + 5_000 + i,
+          enqueueAtStep: 30,
+        });
+      }
+
+      const scenario: Scenario = {
+        name: "ckStalledNewcomer",
+        messages,
+        envConcurrencyLimit: 1,
+        holdSteps: 0,
+        maxSteps: 600,
+      };
+
+      const on = await runScenario(redisContainer, scenario, true);
+      const off = await runScenario(redisContainer, scenario, false);
+
+      assertConservation(scenario, on, off);
+
+      // Over the 20 steps after it lands, the latecomer must not monopolise service.
+      const windowServes = (r: ScenarioResult) =>
+        r.serves.filter((s) => s.step >= 30 && s.step < 50);
+      const onWindow = windowServes(on);
+      const onLate = onWindow.filter((s) => s.ck === "latecomer").length;
+
+      debugLog("ckStalledNewcomer", {
+        onWindowTotal: onWindow.length,
+        onLate,
+        offLate: windowServes(off).filter((s) => s.ck === "latecomer").length,
+      });
+
+      // Four keys compete in that window, so a fair share is a quarter of it. Before the
+      // floor fix the latecomer took 12 of 20 here; it now takes its 5.
+      const fairShare = Math.ceil(onWindow.length / 4);
+      expect(onWindow.length).toBeGreaterThan(0);
+      expect(onLate).toBeLessThanOrEqual(fairShare + 2);
+    }
+  );
 });