diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 9d4d5d04d..e145d31f9 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -5162,7 +5162,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 @@ -5171,6 +5174,7 @@ if #minEntry > 0 then floor = minTag end end +local minServableTag = nil local results = {} local dequeuedCount = 0 @@ -5225,6 +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 + minServableTag = tag + end redis.call('ZADD', ckVtimeKey, tostring(tag + (quantum / weight)), ckQueueName) end else @@ -5247,12 +5255,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 @@ -5281,6 +5285,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) 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 f439041f2..13d2b865d 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 @@ -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 }) => { diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtimeFairness.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtimeFairness.test.ts index 622b2cdd2..c5cf3eba6 100644 --- a/internal-packages/run-engine/src/run-queue/tests/ckVtimeFairness.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/ckVtimeFairness.test.ts @@ -93,7 +93,15 @@ function makeMessage(overrides: Partial = {}): 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(); 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) { @@ -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); + } + ); });