diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 34e83e27a..4637ea15d 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -5106,6 +5106,9 @@ return __qmret(results) // 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, floor advance from pass 1). + // Pass 2 always runs: when the batch is already full it registers the variants + // pass 1 could not see rather than serving them, which is what keeps a backlog + // queued before the flag went on from being unreachable. this.redis.defineCommand("dequeueMessagesFromCkQueueVtimeTracked", { numberOfKeys: 13, lua: ` @@ -5264,24 +5267,49 @@ end -- Pass 1: fair order (lowest virtual start tag first) local vtimeCandidates = redis.call('ZRANGE', ckVtimeKey, 0, window - 1) +-- NEW: the window read doubles as a free membership set for pass 2's discovery +-- step. It is complete whenever ckVtime holds no more than window variants, +-- which is the common case; when it is truncated the discovery ZADD is NX so the +-- variants it cannot rule out cost correctness nothing. +local registered = {} +for _, ckQueueName in ipairs(vtimeCandidates) do + registered[ckQueueName] = true +end for _, ckQueueName in ipairs(vtimeCandidates) do if dequeuedCount >= actualMaxCount then break end tryServe(ckQueueName, true) end -- Pass 2: fill + discovery in age order (work conservation, mixed-deploy safety). --- Never runs when pass 1 filled the batch. -if dequeuedCount < actualMaxCount then - -- Clamp to at least 3x so pass 2 never scans fewer index variants than the old command, preserving work conservation regardless of the configured multiplier - local pass2Window = math.max(window, actualMaxCount * 3) - local ckQueues = redis.call('ZRANGEBYSCORE', ckIndexKey, '-inf', tostring(currentTime), 'LIMIT', 0, pass2Window) - for _, ckQueueName in ipairs(ckQueues) do - if dequeuedCount >= actualMaxCount then break end - if not attempted[ckQueueName] then +-- Clamp to at least 3x so pass 2 never scans fewer index variants than the old command, preserving work conservation regardless of the configured multiplier +local pass2Window = math.max(window, actualMaxCount * 3) +local ckQueues = redis.call('ZRANGEBYSCORE', ckIndexKey, '-inf', tostring(currentTime), 'LIMIT', 0, pass2Window) +-- NEW: pass 2 runs even when pass 1 filled the batch. A variant that reached +-- ckIndex without a ckVtime entry (queued before the flag went on, enqueued by an +-- instance that still has it off, or left behind by an expired ckVtime) is invisible +-- to pass 1, and pass 1 filling the batch off the registered variants alone kept it +-- that way until one of them drained. With no batch slot left we only register it, +-- at the floor, so the next call's pass 1 leads with it. Serving is still capped at +-- actualMaxCount, so this adds no serve the old gate would have refused. +local discovered = nil +for _, ckQueueName in ipairs(ckQueues) do + if not attempted[ckQueueName] then + if dequeuedCount < actualMaxCount then tryServe(ckQueueName, false) + elseif not registered[ckQueueName] then + -- Collected into one variadic ZADD: discovery costs at most a single op per + -- call however many variants it registers. Skipping attempted matters: + -- tryServe GCs a drained variant out of both indexes, and re-adding it here + -- would resurrect a ckVtime entry with no ckIndex member. + if discovered == nil then discovered = {ckVtimeKey, 'NX'} end + table.insert(discovered, tostring(floor)) + table.insert(discovered, ckQueueName) end end end +if discovered ~= nil then + redis.call('ZADD', unpack(discovered)) +end -- NEW: persist floor and refresh TTLs if minServableTag ~= nil and minServableTag > floor then 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 0fd96615e..5f6f317f4 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 @@ -1273,4 +1273,140 @@ describe("CK virtual-time (SFQ) dequeue", () => { } } ); + + // Cold start: the flag is turned on over a backlog that was queued while it was + // off, so every variant is in ckIndex and none is in :ckVtime. Pass 1 can only + // see registered variants, so the cohort pass 2 happens to register on the first + // call is the only cohort pass 1 ever serves; while that cohort keeps the batch + // full, pass 2 never runs again and the rest of the backlog is unreachable until + // the cohort drains. A variant that gets no further enqueues and no nacks has no + // other route into the fair order, so the bound below is the whole guarantee. + // + // The same shape covers a mixed deploy (an instance with the flag still off + // enqueues through the non-vtime command) and a :ckVtime that expired while + // ckIndex survived. + describe("cold start over an unregistered backlog", () => { + type ColdStartShape = { + variants: number; + perVariant: number; + maxCount: number; + scanWindowMultiplier?: number; + // Calls the coldest variant (newest head, so last in the age order pass 2 + // walks) may wait before its first serve. + bound: number; + }; + + // Enqueues the backlog with the flag OFF, then reopens the same keyspace with + // it ON and drains, recording the call each variant was first served on. + async function runColdStart(redisContainer: any, shape: ColdStartShape) { + const t0 = Date.now() - 500_000; + const cks = Array.from( + { length: shape.variants }, + (_, k) => `ck${String(k).padStart(2, "0")}` + ); + + const before = createQueue(redisContainer, null); + try { + for (let i = 0; i < shape.perVariant; i++) { + for (let k = 0; k < cks.length; k++) { + await before.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ + runId: `${cks[k]}-${i}`, + concurrencyKey: cks[k], + // 1s of head-age spacing per variant, so ck00 is oldest and the + // age order never reshuffles as heads advance by 1ms per serve. + timestamp: t0 + k * 1_000 + i, + }), + workerQueue: authenticatedEnvDev.id, + skipDequeueProcessing: true, + }); + } + } + } finally { + await before.quit(); + } + + const after = createQueue(redisContainer, { + ...(shape.scanWindowMultiplier === undefined + ? {} + : { scanWindowMultiplier: shape.scanWindowMultiplier }), + }); + try { + const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(variantName(cks[0]!)); + const ckIndexKey = testOptions.keys.ckIndexKeyFromQueue(variantName(cks[0]!)); + // The premise: the whole backlog is in the age index and nothing is in the + // fair order. + expect(await after.redis.zcard(ckIndexKey)).toBe(shape.variants); + expect(await after.redis.zcard(ckVtimeKey)).toBe(0); + + const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2); + const total = shape.variants * shape.perVariant; + const firstServeCall = new Map(); + let served = 0; + + for (let call = 0; call < total + 10 && served < total; call++) { + const messages = await after.testDequeueFromMasterQueue( + shard, + authenticatedEnvDev.id, + shape.maxCount + ); + for (const m of messages) { + const ck = m.message.concurrencyKey!; + if (!firstServeCall.has(ck)) firstServeCall.set(ck, call); + served++; + // Ack immediately so env concurrency never gates a serve: the only + // thing under test is reachability. + await after.acknowledgeMessage(authenticatedEnvDev.organization.id, m.messageId, { + skipDequeueProcessing: true, + }); + } + } + + return { firstServeCall, served, total, coldest: cks[cks.length - 1]! }; + } finally { + await after.quit(); + } + } + + // Bounds are the measured value, not headroom: the harness has no wall-clock + // wait and no randomness. The pre-fix figure in each comment is what the same + // shape did when pass 2 was gated on dequeuedCount < actualMaxCount. + const shapes: [string, ColdStartShape][] = [ + // Registered cohort (5) smaller than the backlog (8), both inside the + // pass-1 window (15) and the pass-2 scan window (15). Pre-fix: call 12. + ["8 variants, batch 5", { variants: 8, perVariant: 12, maxCount: 5, bound: 1 }], + // Backlog exactly fills the pass-1 window (12), so discovery has to land in + // more than one call. Pre-fix: call 16. + ["12 variants, batch 4", { variants: 12, perVariant: 8, maxCount: 4, bound: 2 }], + // scanWindowMultiplier 1 puts the pass-1 window (5) below the backlog, so + // the window read can no longer tell which variants are already registered + // and discovery falls back to the NX. Pre-fix: call 12. + [ + "15 variants, batch 5, narrow fair window", + { variants: 15, perVariant: 6, maxCount: 5, scanWindowMultiplier: 1, bound: 2 }, + ], + ]; + + for (const [name, shape] of shapes) { + redisTest( + `${name}: the coldest variant is served within ${shape.bound + 1} calls`, + async ({ redisContainer }) => { + const { firstServeCall, served, total, coldest } = await runColdStart( + redisContainer, + shape + ); + + // Work conservation: the backlog still drains completely. + expect(served).toBe(total); + expect(firstServeCall.size).toBe(shape.variants); + + expect( + firstServeCall.get(coldest), + `coldest variant ${coldest} first served on call ${firstServeCall.get(coldest)}` + ).toBeLessThanOrEqual(shape.bound); + } + ); + } + }); }); diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts index 9edbd7728..9dc0d31f7 100644 --- a/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts @@ -346,10 +346,12 @@ describe("CK virtual-time concurrency and op-count budget", () => { expect(off.served).toBe(cks.length * perKey); expect(on.served).toBe(cks.length * perKey); - // Per dequeue call the vtime path adds at worst 7 fixed ops: GET floor, - // ZRANGE min, ZRANGE window, the pass-2 ZRANGEBYSCORE, SET floor, - // EXISTS ckVtime, EXPIRE ckVtime — plus per serve one ZSCORE and one ZADD. - const budget = dequeueCalls * (7 + 2 * maxCount); + // Per dequeue call the vtime path adds at worst 8 fixed ops: GET floor, + // ZRANGE min, ZRANGE window, the pass-2 ZRANGEBYSCORE, the pass-2 + // discovery ZADD, SET floor, EXISTS ckVtime, EXPIRE ckVtime, plus per + // serve one ZSCORE and one ZADD. The discovery ZADD is variadic, so it + // stays a single op however many variants one call registers. + const budget = dequeueCalls * (8 + 2 * maxCount); expect( on.totalCalls, `on_total ${on.totalCalls} exceeds off_total ${off.totalCalls} + budget ${budget}`