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(); }