From b5f81b31571c8d9e0fb4b2d36bedb76fcc4eea40 Mon Sep 17 00:00:00 2001 From: Wes Mason Date: Fri, 31 Jul 2026 10:56:41 +0100 Subject: [PATCH] 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. --- .../run-engine/src/run-queue/index.ts | 17 ++- .../src/run-queue/tests/ckVtime.test.ts | 112 ++++++++++++++---- 2 files changed, 95 insertions(+), 34 deletions(-) diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index e145d31f9..34e83e27a 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -5105,7 +5105,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: ` @@ -5182,7 +5182,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 @@ -5229,8 +5229,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) @@ -5255,9 +5257,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 @@ -5267,7 +5266,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). @@ -5279,7 +5278,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 diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts index 13d2b865d..0fd96615e 100644 --- a/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts +++ b/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(); }