diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 4637ea15d..810013833 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -5105,7 +5105,9 @@ 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, floor advance from pass 1). + // lines added (tag advance on serve, ZREM ckVtime on GC, floor advance from pass 1, + // and the notReady report that lets pass 1 step over a future-headed variant without + // spending a window slot on it). // 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. @@ -5164,6 +5166,10 @@ if actualMaxCount <= 0 then end local window = actualMaxCount * windowMultiplier +-- Pass 1 reads further than it will spend, so a variant whose head is scheduled in the +-- future can be passed over without costing a window slot. Capped rather than unbounded: +-- a block wider than this still degrades to pass 2's age order, which is safe. +local scanLimit = window * 2 -- 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 @@ -5235,6 +5241,10 @@ local function tryServe(ckQueueName, mayRaiseFloor) -- 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. + -- Pass 1 now steps over future-headed variants below this tag, so the floor can + -- rise past one of them. That is the same forfeiture a variant at its concurrency + -- ceiling already takes: it keeps its entry, loses the sub-floor credit, and is + -- clamped up to the floor when it next becomes servable. if mayRaiseFloor and (minServableTag == nil or tag < minServableTag) then minServableTag = tag end @@ -5260,24 +5270,37 @@ local function tryServe(ckQueueName, mayRaiseFloor) redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW else redis.call('ZADD', ckIndexKey, any[2], ckQueueName) + -- NEW: backlog, but the head is scheduled later, so nothing here is servable this + -- call. The readiness is already known from the ZRANGEBYSCORE above, so reporting + -- it costs nothing and lets pass 1 decline to spend a window slot on it. + return 'notReady' end end end 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, +local vtimeCandidates = redis.call('ZRANGE', ckVtimeKey, 0, scanLimit - 1) +-- NEW: the scan read doubles as a free membership set for pass 2's discovery +-- step. It is complete whenever ckVtime holds no more than scanLimit 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 +-- NEW: a variant whose head is scheduled in the future stays registered and stays +-- scanned, it just does not spend one of the window's slots. Without this a retry storm +-- across enough keys fills the window with variants that cannot be served, pass 1 serves +-- nothing, and because minServableTag is the only route that can lift the floor over a +-- stale low tag, the floor freezes for as long as the storm lasts. Every other outcome +-- (served, gated on concurrency, drained, reaped) still spends a slot, as before. +local windowBudget = window for _, ckQueueName in ipairs(vtimeCandidates) do - if dequeuedCount >= actualMaxCount then break end - tryServe(ckQueueName, true) + if dequeuedCount >= actualMaxCount or windowBudget <= 0 then break end + if tryServe(ckQueueName, true) ~= 'notReady' then + windowBudget = windowBudget - 1 + end end -- Pass 2: fill + discovery in age order (work conservation, mixed-deploy safety). @@ -5311,13 +5334,21 @@ if discovered ~= nil then redis.call('ZADD', unpack(discovered)) 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) +-- NEW: persist floor and refresh TTLs. A call that served nothing writes nothing: the two +-- things this block would persist are both re-derivable, since minServableTag is only set +-- inside a successful serve and pass 2's discovery only runs once the batch is full, so +-- the only floor movement on a zero-serve call is the min-tag read-repair, which is +-- recomputed from ckVtime at the top of every call anyway. Idle polling a queue whose work +-- is all future-scheduled or concurrency-gated therefore costs no writes, matching the old +-- command's early return. +if dequeuedCount > 0 then + 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) + end end -- Rebalance master queue (ckIndex keeps its timestamp domain) 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 9b17a7fb1..8ac6f0a5b 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 @@ -301,8 +301,7 @@ describe("CK virtual-time (SFQ) dequeue", () => { prevFloor = floor; } - // Final settle: gate both variants (limit 1 + an occupied slot) so nothing - // is served (no advance), and the floor read-repairs up to the current min tag. + // Final settle: gate both variants (limit 1 + an occupied slot) so nothing is served. await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, QUEUE, 1); for (const ck of cks) { await queue.redis.sadd( @@ -312,12 +311,28 @@ describe("CK virtual-time (SFQ) dequeue", () => { } await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 2); - const floorAfter = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0"); - expect(floorAfter).toBeGreaterThanOrEqual(prevFloor); - const minEntry = await queue.redis.zrange(ckVtimeKey, 0, 0, "WITHSCORES"); const minTag = Number(minEntry[1]); - expect(floorAfter).toBe(minTag); + expect(minTag).toBeGreaterThan(prevFloor); + + // A call that serves nothing persists nothing. The read-repair to the min tag is + // recomputed from ckVtime at the top of every call, so leaving it unwritten here + // costs nothing and keeps an idle poll free of writes. + const floorWhileGated = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0"); + expect(floorWhileGated).toBe(prevFloor); + + // Free a slot: the next serving call persists the repaired floor, so the repair + // itself is intact, it is only the write that waits for a serve. + for (const ck of cks) { + await queue.redis.srem( + testOptions.keys.queueCurrentConcurrencyKeyFromQueue(variantName(ck)), + "occupant" + ); + } + await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 2); + + const floorAfter = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0"); + expect(floorAfter).toBeGreaterThanOrEqual(minTag); expect(floorAfter).toBeGreaterThan(10); } finally { await queue.quit(); @@ -917,14 +932,14 @@ describe("CK virtual-time (SFQ) dequeue", () => { ); redisTest( - "a pass-1 window filled entirely with unservable variants degrades to age order, and recovers", + "a scan filled entirely with unservable variants degrades to age order, and recovers", async ({ redisContainer }) => { - // The single-stalled case above is handled by the minServableTag route: a servable - // variant inside the window raises the floor over the stalled tag. That route needs - // pass 1 to serve something. With every window slot (actualMaxCount * multiplier, - // 3 here) held by unservable variants, pass 1 serves nothing, minServableTag stays - // nil, and the min-tag route is pinned by those same stalled tags, so the floor - // cannot move for as long as the block lasts. + // Pass 1 steps over a future-headed variant without spending a window slot, so the + // window alone can no longer be blocked. The scan behind it is capped though, at + // scanLimit = 2 * window (6 here), and this is the residual: with every scanned + // position held by an unservable variant, pass 1 still serves nothing, minServableTag + // stays nil, and the min-tag route is pinned by those same stalled tags, so the floor + // cannot move until the block thins out. // // Two properties of that state are worth pinning down. It stays work-conserving: // pass 2 keeps serving in age order, which is the flag-off behaviour, so a full @@ -937,9 +952,9 @@ describe("CK virtual-time (SFQ) dequeue", () => { const t0 = Date.now() - 100_000; // Names decide tie order at equal tags, and the point of the fixture is that the - // blockers hold every window slot: a0/a1/a2 sort below zbusy, so the window read - // returns only them. - const blockers = ["a0", "a1", "a2"]; + // blockers hold every scanned position: a0..a5 sort below zbusy, so the scan read + // returns only them and zbusy is never reached. + const blockers = ["a0", "a1", "a2", "a3", "a4", "a5"]; for (const ck of blockers) { await queue.enqueueMessage({ env: authenticatedEnvDev, @@ -1035,8 +1050,12 @@ describe("CK virtual-time (SFQ) dequeue", () => { } } + // The unblocked cohort plus the arrival is 7 keys sitting at the floor against an + // incumbent on FREEZE_CALLS, so levelling up costs 7 * FREEZE_CALLS serves before + // the incumbent competes again. Drain comfortably past that rather than right on + // the boundary, or the bound below is measuring the cutoff instead of the debt. const servedAfter: Record = {}; - for (let call = 0; call < 60; call++) { + for (let call = 0; call < (blockers.length + 1) * FREEZE_CALLS + 60; call++) { for (const ck of await drainOne()) { servedAfter[ck] = (servedAfter[ck] ?? 0) + 1; }