From d25ccccc5ed0adea1f44529332e1bc4b31643a5d Mon Sep 17 00:00:00 2001 From: Wes Mason Date: Tue, 18 Aug 2026 13:01:37 +0100 Subject: [PATCH] fix(run-engine): remember a drained ck variant's virtual time A concurrency-key variant whose message zset drained was removed from ckVtime, throwing away the virtual-time tag it had accumulated. Its next enqueue re-registered it at the floor, so it came back with full credit. A variant holding a persistent backlog keeps advancing its tag instead, so it lost every pass-1 slot to variants that drain and reset each call, and pass 2 could not help once the batch was already full. Measured on a backlogged variant against five trickle variants: 1 serve out of 600 with the flag on, against 120 with it off, inverting the fairness the feature exists to provide. A parked tag now survives the drain in a sibling :ckVtimeIdle zset, and every registration path (enqueue, enqueue-with-ttl, nack, and the gated-variant branch) starts the variant at max(floor, idleTag) rather than at the floor. The out-of-band drains (ack, dead-letter, TTL expiry) park the tag too. Entries at or below the floor confer nothing, so a single ZREMRANGEBYSCORE per serving call reaps them and bounds the set. Deriving the floor differently was tried first and rejected by measurement: any variant with a tag that never advances, which includes any concurrency-gated key, pins the minimum and defeats it. Backlogged variant now lands on its round-robin share in every shape, including with a pinned-low gated or future-headed variant present, and with more trickle variants than batch slots. Op-count overhead goes from 587 to 641 against a budget of 900. The flag-off Lua is still byte-identical: 7 vtime command variants changed, the other 31 commands hash the same as HEAD. --- .../run-engine/src/run-queue/index.ts | 192 +++++++-- .../run-engine/src/run-queue/keyProducer.ts | 5 + .../src/run-queue/tests/ckVtime.test.ts | 7 +- .../run-queue/tests/ckVtimeStarvation.test.ts | 389 ++++++++++++++++++ .../src/run-queue/tests/keyProducer.test.ts | 6 + .../run-engine/src/run-queue/types.ts | 1 + 6 files changed, 562 insertions(+), 38 deletions(-) create mode 100644 internal-packages/run-engine/src/run-queue/tests/ckVtimeStarvation.test.ts diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 7d0429a70..9d66f087e 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -1701,20 +1701,28 @@ export class RunQueue { const visibilityTimeoutMs = (ttlSystem.visibilityTimeoutMs ?? 30_000).toString(); // Atomically get and remove expired runs from TTL set, ack them from normal queues, and enqueue to TTL worker - const expireTtlRuns = this.#ckVtimeEnabled - ? this.redis.expireTtlRunsVtimeTracked.bind(this.redis) - : this.redis.expireTtlRunsTracked.bind(this.redis); - - const results = await expireTtlRuns( - ttlQueueKey, - keyPrefix, - now.toString(), - batchSize.toString(), - shardCount.toString(), - workerQueueKey, - workerItemsKey, - visibilityTimeoutMs - ); + const results = this.#ckVtimeEnabled + ? await this.redis.expireTtlRunsVtimeTracked( + ttlQueueKey, + keyPrefix, + now.toString(), + batchSize.toString(), + shardCount.toString(), + workerQueueKey, + workerItemsKey, + visibilityTimeoutMs, + String(this.#ckVtimeStateTtl) + ) + : await this.redis.expireTtlRunsTracked( + ttlQueueKey, + keyPrefix, + now.toString(), + batchSize.toString(), + shardCount.toString(), + workerQueueKey, + workerItemsKey, + visibilityTimeoutMs + ); if (!results || results.length === 0) { return []; @@ -2240,6 +2248,7 @@ export class RunQueue { baseQueueKey, this.keys.ckVtimeKeyFromQueue(message.queue), this.keys.ckVtimeFloorKeyFromQueue(message.queue), + this.keys.ckVtimeIdleKeyFromQueue(message.queue), // args queueName, messageId, @@ -2315,6 +2324,7 @@ export class RunQueue { baseQueueKey, this.keys.ckVtimeKeyFromQueue(message.queue), this.keys.ckVtimeFloorKeyFromQueue(message.queue), + this.keys.ckVtimeIdleKeyFromQueue(message.queue), // args queueName, messageId, @@ -2634,6 +2644,7 @@ export class RunQueue { runningCounterKey, this.keys.ckVtimeKeyFromQueue(ckWildcardQueue), this.keys.ckVtimeFloorKeyFromQueue(ckWildcardQueue), + this.keys.ckVtimeIdleKeyFromQueue(ckWildcardQueue), //args ckWildcardQueue, String(Date.now()), @@ -2918,11 +2929,13 @@ export class RunQueue { lengthCounterKey, runningCounterKey, this.keys.ckVtimeKeyFromQueue(message.queue), + this.keys.ckVtimeIdleKeyFromQueue(message.queue), messageId, messageQueue, messageKeyValue, removeFromWorkerQueue ? "1" : "0", - ckWildcardName + ckWildcardName, + String(this.#ckVtimeStateTtl) ); } @@ -3071,6 +3084,7 @@ export class RunQueue { runningCounterKey, this.keys.ckVtimeKeyFromQueue(message.queue), this.keys.ckVtimeFloorKeyFromQueue(message.queue), + this.keys.ckVtimeIdleKeyFromQueue(message.queue), //args messageId, messageQueue, @@ -3161,9 +3175,11 @@ export class RunQueue { lengthCounterKey, runningCounterKey, this.keys.ckVtimeKeyFromQueue(message.queue), + this.keys.ckVtimeIdleKeyFromQueue(message.queue), messageId, messageQueue, - ckWildcardName + ckWildcardName, + String(this.#ckVtimeStateTtl) ); } else { await this.redis.moveToDeadLetterQueueCkTracked( @@ -4223,7 +4239,7 @@ return __qmret(0) // brand-new key is present in the fair order from its first enqueue. The // fast path (direct-to-worker-queue) neither registers nor advances. this.redis.defineCommand("enqueueMessageCkVtimeTracked", { - numberOfKeys: 17, + numberOfKeys: 18, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -4242,9 +4258,10 @@ local envConcurrencyLimitBurstFactorKey = KEYS[13] -- Counter keys (KEYS 14-15) local lengthCounterKey = KEYS[14] local baseQueueKey = KEYS[15] --- Virtual-time keys (KEYS 16-17) +-- Virtual-time keys (KEYS 16-18) local ckVtimeKey = KEYS[16] local ckVtimeFloorKey = KEYS[17] +local ckVtimeIdleKey = KEYS[18] local queueName = ARGV[1] local messageId = ARGV[2] @@ -4329,12 +4346,20 @@ if #earliest > 0 then redis.call('ZADD', ckIndexKey, earliest[2], queueName) end --- Register this variant in the virtual-time index at the floor. NX means an --- already-advanced tag is never rewound. +-- Register this variant in the virtual-time index. NX means an already-advanced tag is +-- never rewound. The start is max(floor, remembered idle tag): a variant that drained and +-- came back would otherwise be handed full credit at the floor on every re-enqueue, which +-- starves any variant carrying a persistent backlog. local vfloor = redis.call('GET', ckVtimeFloorKey) or '0' -redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName) +local vstart = tonumber(vfloor) +local vidle = redis.call('ZSCORE', ckVtimeIdleKey, queueName) +if vidle and tonumber(vidle) > vstart then + vstart = tonumber(vidle) +end +redis.call('ZADD', ckVtimeKey, 'NX', tostring(vstart), queueName) redis.call('EXPIRE', ckVtimeKey, stateTtl) redis.call('EXPIRE', ckVtimeFloorKey, stateTtl) +redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) -- Rebalance master queue with ck:* member local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES') @@ -4365,7 +4390,7 @@ return __qmret(0) // Vtime variant of enqueueMessageWithTtlCkTracked. Same slow-path-only // registration as enqueueMessageCkVtimeTracked above. this.redis.defineCommand("enqueueMessageWithTtlCkVtimeTracked", { - numberOfKeys: 18, + numberOfKeys: 19, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -4385,9 +4410,10 @@ local envConcurrencyLimitBurstFactorKey = KEYS[14] -- Counter keys (KEYS 15-16) local lengthCounterKey = KEYS[15] local baseQueueKey = KEYS[16] --- Virtual-time keys (KEYS 17-18) +-- Virtual-time keys (KEYS 17-19) local ckVtimeKey = KEYS[17] local ckVtimeFloorKey = KEYS[18] +local ckVtimeIdleKey = KEYS[19] local queueName = ARGV[1] local messageId = ARGV[2] @@ -4467,12 +4493,20 @@ if #earliest > 0 then redis.call('ZADD', ckIndexKey, earliest[2], queueName) end --- Register this variant in the virtual-time index at the floor. NX means an --- already-advanced tag is never rewound. +-- Register this variant in the virtual-time index. NX means an already-advanced tag is +-- never rewound. The start is max(floor, remembered idle tag): a variant that drained and +-- came back would otherwise be handed full credit at the floor on every re-enqueue, which +-- starves any variant carrying a persistent backlog. local vfloor = redis.call('GET', ckVtimeFloorKey) or '0' -redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName) +local vstart = tonumber(vfloor) +local vidle = redis.call('ZSCORE', ckVtimeIdleKey, queueName) +if vidle and tonumber(vidle) > vstart then + vstart = tonumber(vidle) +end +redis.call('ZADD', ckVtimeKey, 'NX', tostring(vstart), queueName) redis.call('EXPIRE', ckVtimeKey, stateTtl) redis.call('EXPIRE', ckVtimeFloorKey, stateTtl) +redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) -- Rebalance master queue with ck:* member local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES') @@ -4732,6 +4766,7 @@ local shardCount = tonumber(ARGV[4]) local workerQueueKey = ARGV[5] local workerItemsKey = ARGV[6] local visibilityTimeoutMs = tonumber(ARGV[7]) +local stateTtl = tonumber(ARGV[8] or '86400') local function decrFloored(key) if tonumber(redis.call('GET', key) or '0') > 0 then @@ -4809,7 +4844,16 @@ for i, member in ipairs(expiredMembers) do redis.call('ZREM', ckIndexKey, rawQueueKey) -- NEW: derived rather than passed in, because this sweep discovers the queues it -- touches inside the script, exactly as ckIndexKey above is derived. - redis.call('ZREM', keyPrefix .. ckMatch .. ":ckVtime", rawQueueKey) + -- NEW: park the tag first so the variant's next enqueue re-registers with the + -- credit it earned rather than at the floor. + local ckVtimeKey = keyPrefix .. ckMatch .. ":ckVtime" + local ckVtimeIdleKey = keyPrefix .. ckMatch .. ":ckVtimeIdle" + local idleTag = redis.call('ZSCORE', ckVtimeKey, rawQueueKey) + if idleTag then + redis.call('ZADD', ckVtimeIdleKey, idleTag, rawQueueKey) + redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) + end + redis.call('ZREM', ckVtimeKey, rawQueueKey) else redis.call('ZADD', ckIndexKey, earliest[2], rawQueueKey) end @@ -5282,7 +5326,7 @@ return __qmret(results) // 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, + numberOfKeys: 14, lua: ` local ckIndexKey = KEYS[1] local queueConcurrencyLimitKey = KEYS[2] @@ -5297,6 +5341,7 @@ local lengthCounterKey = KEYS[10] local runningCounterKey = KEYS[11] local ckVtimeKey = KEYS[12] local ckVtimeFloorKey = KEYS[13] +local ckVtimeIdleKey = KEYS[14] local ckWildcardName = ARGV[1] local currentTime = tonumber(ARGV[2]) @@ -5364,6 +5409,9 @@ local attempted = {} local function tryServe(ckQueueName, mayRaiseFloor) attempted[ckQueueName] = true local fullQueueKey = keyPrefix .. ckQueueName + -- NEW: the tag this call wrote back, if it served. Site A below reuses it rather than + -- re-reading the score it just wrote. + local servedTag = nil local ckConcurrencyKey = fullQueueKey .. ':currentConcurrency' local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0') @@ -5418,7 +5466,8 @@ local function tryServe(ckQueueName, mayRaiseFloor) if mayRaiseFloor and (minServableTag == nil or tag < minServableTag) then minServableTag = tag end - redis.call('ZADD', ckVtimeKey, tostring(tag + (quantum / weight)), ckQueueName) + servedTag = tag + (quantum / weight) + redis.call('ZADD', ckVtimeKey, tostring(servedTag), ckQueueName) end else redis.call('ZREM', fullQueueKey, messageId) @@ -5429,6 +5478,24 @@ local function tryServe(ckQueueName, mayRaiseFloor) local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES') if #earliest == 0 then redis.call('ZREM', ckIndexKey, ckQueueName) + -- NEW: park the tag in the idle set so the next enqueue re-registers with the + -- credit this variant earned instead of full credit at the floor. Only above the + -- floor is worth keeping: registration takes max(floor, idleTag), so an entry at + -- or below the floor confers nothing and just grows the set. + -- servedTag is nil when the branch above dropped an expired or payload-less message + -- rather than serving one; the tag is still on record, so read it back before the + -- ZREM discards it. Only that rare path pays the extra read. + local parkTag = servedTag + if parkTag == nil then + local storedTag = redis.call('ZSCORE', ckVtimeKey, ckQueueName) + if storedTag then + parkTag = tonumber(storedTag) + end + end + if parkTag ~= nil and parkTag > floor then + redis.call('ZADD', ckVtimeIdleKey, tostring(parkTag), ckQueueName) + redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) + end redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW else redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName) @@ -5437,6 +5504,13 @@ local function tryServe(ckQueueName, mayRaiseFloor) local any = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES') if #any == 0 then redis.call('ZREM', ckIndexKey, ckQueueName) + -- NEW: nothing was served, so no tag is in hand. Read it before the ZREM discards + -- it, and keep it only if it is above the floor (see Site A). + local idleTag = redis.call('ZSCORE', ckVtimeKey, ckQueueName) + if idleTag and tonumber(idleTag) > floor then + redis.call('ZADD', ckVtimeIdleKey, idleTag, ckQueueName) + redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) + end redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW else redis.call('ZADD', ckIndexKey, any[2], ckQueueName) @@ -5458,7 +5532,14 @@ local function tryServe(ckQueueName, mayRaiseFloor) -- already registered keeps the tag it earned, which is the usual case and costs one op. -- The TTL only needs setting when this actually registered something, since that is the -- path that can recreate a ckVtime key which expired out from under a live ckIndex. - if redis.call('ZADD', ckVtimeKey, 'NX', tostring(floor), ckQueueName) == 1 then + -- NEW: registers at max(floor, remembered idle tag), same rule as the enqueue path, so + -- a variant that drained under the gate does not come back with full credit. + local gateStart = floor + local gateIdle = redis.call('ZSCORE', ckVtimeIdleKey, ckQueueName) + if gateIdle and tonumber(gateIdle) > gateStart then + gateStart = tonumber(gateIdle) + end + if redis.call('ZADD', ckVtimeKey, 'NX', tostring(gateStart), ckQueueName) == 1 then redis.call('EXPIRE', ckVtimeKey, stateTtl) end end @@ -5531,6 +5612,9 @@ if dequeuedCount > 0 then floor = minServableTag end redis.call('SET', ckVtimeFloorKey, tostring(floor), 'EX', stateTtl) + -- NEW: an idle entry at or below the floor confers no credit, because registration takes + -- max(floor, idleTag). Dropping it bounds the idle set at one op per serving call. + redis.call('ZREMRANGEBYSCORE', ckVtimeIdleKey, '-inf', tostring(floor)) if redis.call('EXISTS', ckVtimeKey) == 1 then redis.call('EXPIRE', ckVtimeKey, stateTtl) end @@ -6074,7 +6158,7 @@ end }); this.redis.defineCommand("acknowledgeMessageCkVtimeTracked", { - numberOfKeys: 13, + numberOfKeys: 14, lua: ` -- Keys: local masterQueueKey = KEYS[1] @@ -6090,6 +6174,7 @@ local ckIndexKey = KEYS[10] local lengthCounterKey = KEYS[11] local runningCounterKey = KEYS[12] local ckVtimeKey = KEYS[13] +local ckVtimeIdleKey = KEYS[14] -- Args: local messageId = ARGV[1] @@ -6097,6 +6182,7 @@ local messageQueueName = ARGV[2] local messageKeyValue = ARGV[3] local removeFromWorkerQueue = ARGV[4] local ckWildcardName = ARGV[5] +local stateTtl = tonumber(ARGV[6] or '86400') local function decrFloored(key) if tonumber(redis.call('GET', key) or '0') > 0 then @@ -6123,6 +6209,14 @@ if #earliestInCkQueue == 0 then -- NEW: the variant has drained, so it leaves the fair order too. Previously only the -- vtime dequeue removed a ckVtime entry, so an ack left one behind whose tag had -- stopped advancing until some later scan happened to visit and collect it. + -- NEW: park the tag first, so the variant's next enqueue re-registers with the credit it + -- earned rather than at the floor. No floor key here, so this saves unconditionally; the + -- dequeue command reaps everything at or below the floor on its next serving call. + local idleTag = redis.call('ZSCORE', ckVtimeKey, messageQueueName) + if idleTag then + redis.call('ZADD', ckVtimeIdleKey, idleTag, messageQueueName) + redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) + end redis.call('ZREM', ckVtimeKey, messageQueueName) else redis.call('ZADD', ckIndexKey, earliestInCkQueue[2], messageQueueName) @@ -6262,7 +6356,7 @@ end // of the variant into the :ckVtime ZSET at the floor (NX), so a GC'd variant // that a nack revives rejoins the fair order. this.redis.defineCommand("nackMessageCkVtimeTracked", { - numberOfKeys: 13, + numberOfKeys: 14, lua: ` -- Keys: local masterQueueKey = KEYS[1] @@ -6279,6 +6373,7 @@ local runningCounterKey = KEYS[11] -- Virtual-time keys (KEYS 12-13) local ckVtimeKey = KEYS[12] local ckVtimeFloorKey = KEYS[13] +local ckVtimeIdleKey = KEYS[14] -- Args: local messageId = ARGV[1] @@ -6341,12 +6436,20 @@ if #earliest > 0 then redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName) end --- Register this variant in the virtual-time index at the floor. NX means an --- already-advanced tag is never rewound. +-- Register this variant in the virtual-time index. NX means an already-advanced tag is +-- never rewound. The start is max(floor, remembered idle tag): a nack after the variant +-- drained would otherwise hand back full credit at the floor, which is the same starvation +-- the enqueue path guards against. local vfloor = redis.call('GET', ckVtimeFloorKey) or '0' -redis.call('ZADD', ckVtimeKey, 'NX', vfloor, messageQueueName) +local vstart = tonumber(vfloor) +local vidle = redis.call('ZSCORE', ckVtimeIdleKey, messageQueueName) +if vidle and tonumber(vidle) > vstart then + vstart = tonumber(vidle) +end +redis.call('ZADD', ckVtimeKey, 'NX', tostring(vstart), messageQueueName) redis.call('EXPIRE', ckVtimeKey, stateTtl) redis.call('EXPIRE', ckVtimeFloorKey, stateTtl) +redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) -- Rebalance master queue with ck:* member local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES') @@ -6445,7 +6548,7 @@ end }); this.redis.defineCommand("moveToDeadLetterQueueCkVtimeTracked", { - numberOfKeys: 13, + numberOfKeys: 14, lua: ` -- Keys: local masterQueueKey = KEYS[1] @@ -6461,11 +6564,13 @@ local ckIndexKey = KEYS[10] local lengthCounterKey = KEYS[11] local runningCounterKey = KEYS[12] local ckVtimeKey = KEYS[13] +local ckVtimeIdleKey = KEYS[14] -- Args: local messageId = ARGV[1] local messageQueueName = ARGV[2] local ckWildcardName = ARGV[3] +local stateTtl = tonumber(ARGV[4] or '86400') local function decrFloored(key) if tonumber(redis.call('GET', key) or '0') > 0 then @@ -6489,6 +6594,12 @@ if #earliest == 0 then -- NEW: the variant has drained, so it leaves the fair order too. Previously only the -- vtime dequeue removed a ckVtime entry, so an ack left one behind whose tag had -- stopped advancing until some later scan happened to visit and collect it. + -- NEW: park the tag first, same rule as the ack path. + local idleTag = redis.call('ZSCORE', ckVtimeKey, messageQueueName) + if idleTag then + redis.call('ZADD', ckVtimeIdleKey, idleTag, messageQueueName) + redis.call('EXPIRE', ckVtimeIdleKey, stateTtl) + end redis.call('ZREM', ckVtimeKey, messageQueueName) else redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName) @@ -7184,6 +7295,7 @@ declare module "@internal/redis" { baseQueueKey: string, ckVtimeKey: string, ckVtimeFloorKey: string, + ckVtimeIdleKey: string, queueName: string, messageId: string, messageData: string, @@ -7220,6 +7332,7 @@ declare module "@internal/redis" { baseQueueKey: string, ckVtimeKey: string, ckVtimeFloorKey: string, + ckVtimeIdleKey: string, queueName: string, messageId: string, messageData: string, @@ -7275,6 +7388,7 @@ declare module "@internal/redis" { runningCounterKey: string, ckVtimeKey: string, ckVtimeFloorKey: string, + ckVtimeIdleKey: string, ckWildcardName: string, currentTime: string, defaultEnvConcurrencyLimit: string, @@ -7330,11 +7444,13 @@ declare module "@internal/redis" { lengthCounterKey: string, runningCounterKey: string, ckVtimeKey: string, + ckVtimeIdleKey: string, messageId: string, messageQueueName: string, messageKeyValue: string, removeFromWorkerQueue: string, ckWildcardName: string, + stateTtl: string, callback?: Callback ): Result; @@ -7374,6 +7490,7 @@ declare module "@internal/redis" { runningCounterKey: string, ckVtimeKey: string, ckVtimeFloorKey: string, + ckVtimeIdleKey: string, messageId: string, messageQueueName: string, messageData: string, @@ -7418,9 +7535,11 @@ declare module "@internal/redis" { lengthCounterKey: string, runningCounterKey: string, ckVtimeKey: string, + ckVtimeIdleKey: string, messageId: string, messageQueueName: string, ckWildcardName: string, + stateTtl: string, callback?: Callback ): Result; @@ -7445,6 +7564,7 @@ declare module "@internal/redis" { workerQueueKey: string, workerItemsKey: string, visibilityTimeoutMs: string, + stateTtl: string, callback?: Callback ): Result; diff --git a/internal-packages/run-engine/src/run-queue/keyProducer.ts b/internal-packages/run-engine/src/run-queue/keyProducer.ts index c00ee47ab..b21409aeb 100644 --- a/internal-packages/run-engine/src/run-queue/keyProducer.ts +++ b/internal-packages/run-engine/src/run-queue/keyProducer.ts @@ -24,6 +24,7 @@ const constants = { CK_INDEX_PART: "ckIndex", CK_VTIME_PART: "ckVtime", CK_VTIME_FLOOR_PART: "ckVtimeFloor", + CK_VTIME_IDLE_PART: "ckVtimeIdle", LENGTH_COUNTER_PART: "lengthCounter", RUNNING_COUNTER_PART: "runningCounter", } as const; @@ -325,6 +326,10 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer { return `${this.baseQueueKeyFromQueue(queue)}:${constants.CK_VTIME_FLOOR_PART}`; } + ckVtimeIdleKeyFromQueue(queue: string): string { + return `${this.baseQueueKeyFromQueue(queue)}:${constants.CK_VTIME_IDLE_PART}`; + } + // indexOf instead of /:ck:.+$/ (queue names are user-controlled; polynomial regex). // Only strips when at least one character follows ":ck:", matching the old semantics. baseQueueKeyFromQueue(queue: string): string { 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 ef7c8059a..36cce5d88 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 @@ -1575,9 +1575,11 @@ describe("CK virtual-time (SFQ) dequeue", () => { testOptions.keys.queueLengthCounterKeyFromQueue(v), testOptions.keys.queueRunningCounterKeyFromQueue(v), ckVtimeKey, + testOptions.keys.ckVtimeIdleKeyFromQueue(v), "r-dlq", v, - testOptions.keys.toCkWildcard(v) + testOptions.keys.toCkWildcard(v), + "86400" ); expect(await queue.redis.zscore(ckIndexKey, v)).toBeNull(); @@ -1629,7 +1631,8 @@ describe("CK virtual-time (SFQ) dequeue", () => { "2", "ttlworker", "ttlworkeritems", - "30000" + "30000", + "86400" ); expect(await queue.redis.zscore(ckIndexKey, v)).toBeNull(); diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtimeStarvation.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtimeStarvation.test.ts new file mode 100644 index 000000000..418f93317 --- /dev/null +++ b/internal-packages/run-engine/src/run-queue/tests/ckVtimeStarvation.test.ts @@ -0,0 +1,389 @@ +import { redisTest } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { Logger } from "@trigger.dev/core/logger"; +import { Decimal } from "@trigger.dev/database"; +import { describe } from "node:test"; +import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js"; +import { RunQueue } from "../index.js"; +import { RunQueueFullKeyProducer } from "../keyProducer.js"; +import type { InputPayload } from "../types.js"; + +// Starvation of a persistently-backlogged CK variant by variants that drain on every call. +// +// A variant that empties is GC'd out of ckVtime, and before the idle zset its next enqueue +// re-registered it at the floor: full credit, every call. A variant carrying a backlog +// keeps advancing its tag and never gets it back, so it loses every comparison. Measured +// on the plain shape below: 1 serve out of 600 with the flag on, against 120 with it off. +// +// ckVtimeIdle remembers the tag across a drain, so re-registration takes +// max(floor, idleTag) and a drain buys nothing. Each shape here asserts the backlogged +// variant lands near its round-robin share of 1/(1 + competitors). +// +// Shapes covered: the plain trickle shape, more trickles than batch slots, a +// concurrency-gated variant pinned at a low tag, a future-headed variant pinned at a low +// tag, and a deep-queue control whose competitors never drain (so they were never affected +// by the bug, and it shows what the fair share is). + +const testOptions = { + name: "rq", + tracer: trace.getTracer("rq"), + workers: 1, + defaultEnvConcurrency: 25, + logger: new Logger("RunQueue", "warn"), + retryOptions: { + maxAttempts: 5, + factor: 1.1, + minTimeoutInMs: 100, + maxTimeoutInMs: 1_000, + randomize: true, + }, + keys: new RunQueueFullKeyProducer(), +}; + +const baseEnv = { + id: "e1234", + type: "DEVELOPMENT" as const, + maximumConcurrencyLimit: 20, + concurrencyLimitBurstFactor: new Decimal(1), + project: { id: "p1234" }, + organization: { id: "o1234" }, +}; + +const QUEUE = "task/my-task"; + +function createQueue(redisContainer: any, keyPrefix: string, vtimeEnabled: boolean) { + return new RunQueue({ + ...testOptions, + masterQueueConsumersDisabled: true, + workerOptions: { disabled: true }, + ...(vtimeEnabled ? { ckVirtualTimeScheduling: { enabled: true } } : {}), + queueSelectionStrategy: new FairQueueSelectionStrategy({ + redis: { keyPrefix, host: redisContainer.getHost(), port: redisContainer.getPort() }, + keys: testOptions.keys, + }), + redis: { keyPrefix, host: redisContainer.getHost(), port: redisContainer.getPort() }, + }); +} + +function makeMessage(overrides: Partial = {}): InputPayload { + return { + runId: "r1", + taskIdentifier: "task/my-task", + orgId: "o1234", + projectId: "p1234", + environmentId: "e1234", + environmentType: "DEVELOPMENT", + queue: QUEUE, + timestamp: Date.now(), + attempt: 0, + ...overrides, + }; +} + +function variantName(ck: string): string { + return testOptions.keys.queueKey(baseEnv, QUEUE, ck); +} + +type RunOpts = { + trickleCount: number; + maxCount: number; + calls: number; + backlogSize: number; + envLimit: number; + // Control: give each competitor a deep backlog instead of one message, so it never + // drains and is never GC'd out of ckVtime / re-registered. + steadyCompetitors?: boolean; + // A variant parked at the per-key concurrency ceiling for the whole run. It stays in + // ckVtime holding a permanently low tag, which is what defeated the earlier floor-only + // prototype of this fix. + gatedVariant?: boolean; + // A variant whose whole backlog is scheduled an hour out. Also stays registered at a + // permanently low tag, by the 'notReady' route rather than the concurrency gate. + futureVariant?: boolean; +}; + +type RunResult = { + backlogServed: number; + trickleServed: number; + totalServed: number; + lastBacklogServeCall: number; + finalFloor: string | null; + finalTags: Record; + idleSize: number; +}; + +async function runShape( + redisContainer: any, + label: string, + vtimeEnabled: boolean, + opts: RunOpts +): Promise { + const keyPrefix = `runqueue:test:${label}:`; + const queue = createQueue(redisContainer, keyPrefix, vtimeEnabled); + + try { + const env = { ...baseEnv, maximumConcurrencyLimit: opts.envLimit }; + await queue.updateEnvConcurrencyLimits(env); + + const t0 = Date.now() - 10_000_000; + + const enqueue = async (runId: string, ck: string, timestamp: number) => { + await queue.enqueueMessage({ + env, + message: makeMessage({ runId, concurrencyKey: ck, timestamp }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + }; + + // Backlogged variant FIRST so its heads are the oldest in ckIndex. + for (let i = 0; i < opts.backlogSize; i++) { + await enqueue(`b${i}`, "backlog", t0 + i); + } + + if (opts.gatedVariant) { + // Ready, old work so it is a live ckIndex/ckVtime member every call... + for (let i = 0; i < 50; i++) { + await enqueue(`g${i}`, "gated", t0 + 100 + i); + } + // ...but parked at the per-key ceiling, so tryServe never serves it. The ceiling is + // min(queue limit, env limit) and the queue limit is unset here, so it is envLimit. + const members = Array.from({ length: opts.envLimit + 10 }, (_, i) => `held-${i}`); + await queue.redis.sadd(`${variantName("gated")}:currentConcurrency`, ...members); + } + + if (opts.futureVariant) { + const future = Date.now() + 60 * 60 * 1000; + for (let i = 0; i < 50; i++) { + await enqueue(`f${i}`, "future", future + i); + } + } + + // One ready message per trickle variant, newer than the whole backlog, so age order + // alone would never prefer them. Under steadyCompetitors each gets a deep queue. + const seedPerCompetitor = opts.steadyCompetitors ? opts.calls + 10 : 1; + for (let t = 0; t < opts.trickleCount; t++) { + for (let i = 0; i < seedPerCompetitor; i++) { + await enqueue(`t${t}-seed-${i}`, `trickle-${t}`, t0 + 5_000_000 + i); + } + } + + const shard = testOptions.keys.masterQueueShardForEnvironment(env.id, 2); + + let backlogServed = 0; + let trickleServed = 0; + let totalServed = 0; + let lastBacklogServeCall = -1; + let refeed = 0; + + for (let call = 0; call < opts.calls; call++) { + const messages = await queue.testDequeueFromMasterQueue(shard, env.id, opts.maxCount); + + const drainedTrickles = new Set(); + for (const m of messages) { + const ck = m.message.concurrencyKey ?? ""; + totalServed++; + if (ck === "backlog") { + backlogServed++; + lastBacklogServeCall = call; + } else if (ck.startsWith("trickle-")) { + trickleServed++; + drainedTrickles.add(ck); + } + // Ack immediately: concurrency is never the limiting factor here. + await queue.acknowledgeMessage(env.organization.id, m.messageId, { + skipDequeueProcessing: true, + }); + } + + if (opts.steadyCompetitors) continue; + + // Re-feed every trickle variant that drained, so it is ready again next call with a + // strictly newer head than the backlog. This is the re-registration under test. + for (const ck of drainedTrickles) { + refeed++; + await enqueue(`t-${ck}-${refeed}`, ck, t0 + 5_000_000 + refeed); + } + } + + const backlogVariant = variantName("backlog"); + const finalFloor = await queue.redis.get( + testOptions.keys.ckVtimeFloorKeyFromQueue(backlogVariant) + ); + const raw = await queue.redis.zrange( + testOptions.keys.ckVtimeKeyFromQueue(backlogVariant), + 0, + -1, + "WITHSCORES" + ); + const finalTags: Record = {}; + for (let i = 0; i < raw.length; i += 2) { + const member = raw[i]!; + const short = member.includes(":ck:") ? member.slice(member.indexOf(":ck:") + 4) : member; + finalTags[short] = Number(raw[i + 1]); + } + const idleSize = await queue.redis.zcard( + testOptions.keys.ckVtimeIdleKeyFromQueue(backlogVariant) + ); + + return { + backlogServed, + trickleServed, + totalServed, + lastBacklogServeCall, + finalFloor, + finalTags, + idleSize, + }; + } finally { + await queue.quit(); + } +} + +// Round-robin share of the served batch for one variant among 1 + trickleCount claimants, +// which is what the deep-queue control measures out at. +function fairShare(opts: RunOpts, served: number): number { + return served / (1 + opts.trickleCount); +} + +// 0.7 rather than 1.0: pass 1 walks in tag order and pass 2 fills by age, so a variant can +// lose a slot to rounding at the batch boundary. The gap being defended against is two +// orders of magnitude (1 vs 100), so this has plenty of room and is not tuned to a number. +function expectFairish(label: string, opts: RunOpts, r: RunResult) { + const target = fairShare(opts, r.totalServed); + expect( + r.backlogServed, + `${label}: backlog served ${r.backlogServed} of ${r.totalServed}, fair share ${target.toFixed( + 1 + )}` + ).toBeGreaterThanOrEqual(target * 0.7); +} + +vi.setConfig({ testTimeout: 300_000 }); + +describe("CK vtime starvation by drain-and-re-register", () => { + redisTest( + "backlogged variant keeps its share against trickle variants (flag on vs off)", + async ({ redisContainer }) => { + const opts: RunOpts = { + trickleCount: 5, + maxCount: 5, + calls: 120, + backlogSize: 400, + envLimit: 20, + }; + + const on = await runShape(redisContainer, "starve-on", true, opts); + const off = await runShape(redisContainer, "starve-off", false, opts); + + expect(on.totalServed).toBe(600); + expect(off.totalServed).toBe(600); + + // The comparison arm: flag off is pure age order, so the always-oldest backlog wins + // a slot on every call. That is the number the flag-on path regressed against. + expect(off.backlogServed).toBeGreaterThanOrEqual(100); + + expectFairish("flag on", opts, on); + + // It was served throughout, not just drained early and then starved. + expect(on.lastBacklogServeCall).toBeGreaterThanOrEqual(opts.calls - 10); + + // The idle zset is reaped at or below the floor on every serving call, so it holds + // at most the variants that drained since the last one. + expect(on.idleSize).toBeLessThanOrEqual(opts.trickleCount + 2); + } + ); + + redisTest( + "holds when there are more trickle variants than batch slots", + async ({ redisContainer }) => { + for (const trickleCount of [6, 8]) { + const opts: RunOpts = { + trickleCount, + maxCount: 5, + calls: 60, + backlogSize: 400, + envLimit: 20, + }; + const on = await runShape(redisContainer, `over-on-${trickleCount}`, true, opts); + expect(on.totalServed).toBe(300); + expectFairish(`trickle=${trickleCount}`, opts, on); + } + } + ); + + redisTest( + "a concurrency-gated variant pinned at a low tag does not defeat it", + async ({ redisContainer }) => { + const opts: RunOpts = { + trickleCount: 5, + maxCount: 5, + calls: 120, + backlogSize: 400, + envLimit: 20, + gatedVariant: true, + }; + const on = await runShape(redisContainer, "pin-gated-on", true, opts); + + // The gated variant is still registered and still holding a tag below the floor, which + // is the state that defeated the earlier floor-only prototype. + expect(on.finalTags["gated"]).toBeLessThan(Number(on.finalFloor)); + expectFairish("gated", opts, on); + } + ); + + redisTest( + "a future-headed variant pinned at a low tag does not defeat it", + async ({ redisContainer }) => { + const opts: RunOpts = { + trickleCount: 5, + maxCount: 5, + calls: 120, + backlogSize: 400, + envLimit: 20, + futureVariant: true, + }; + const on = await runShape(redisContainer, "pin-future-on", true, opts); + + expect(on.finalTags["future"]).toBeLessThan(Number(on.finalFloor)); + expectFairish("future", opts, on); + } + ); + + redisTest("both pinned-low variants at once", async ({ redisContainer }) => { + const opts: RunOpts = { + trickleCount: 5, + maxCount: 5, + calls: 120, + backlogSize: 400, + envLimit: 20, + gatedVariant: true, + futureVariant: true, + }; + const on = await runShape(redisContainer, "pin-both-on", true, opts); + expectFairish("gated+future", opts, on); + }); + + redisTest( + "control: competitors with deep queues never drain, so they never re-registered", + async ({ redisContainer }) => { + const opts: RunOpts = { + trickleCount: 5, + maxCount: 5, + calls: 120, + backlogSize: 400, + envLimit: 20, + steadyCompetitors: true, + }; + const on = await runShape(redisContainer, "steady-on", true, opts); + + // This shape never triggered the bug, so it measures what fair looks like: the + // trickle shapes above are held to the same standard. + expect(on.totalServed).toBe(600); + expectFairish("steady", opts, on); + + // Nothing drained, so nothing was ever parked. + expect(on.idleSize).toBe(0); + } + ); +}); diff --git a/internal-packages/run-engine/src/run-queue/tests/keyProducer.test.ts b/internal-packages/run-engine/src/run-queue/tests/keyProducer.test.ts index 1f61ed1cf..15ed27623 100644 --- a/internal-packages/run-engine/src/run-queue/tests/keyProducer.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/keyProducer.test.ts @@ -442,9 +442,15 @@ describe("KeyProducer", () => { expect(keyProducer.ckVtimeFloorKeyFromQueue(q)).toBe( "{org:o1}:proj:p1:env:e1:queue:task/my-task:ckVtimeFloor" ); + expect(keyProducer.ckVtimeIdleKeyFromQueue(q)).toBe( + "{org:o1}:proj:p1:env:e1:queue:task/my-task:ckVtimeIdle" + ); // ck wildcard and base-queue inputs normalise the same way expect(keyProducer.ckVtimeKeyFromQueue(q.replace(":ck:tenant-a", ":ck:*"))).toBe( keyProducer.ckVtimeKeyFromQueue(q) ); + expect(keyProducer.ckVtimeIdleKeyFromQueue(q.replace(":ck:tenant-a", ":ck:*"))).toBe( + keyProducer.ckVtimeIdleKeyFromQueue(q) + ); }); }); diff --git a/internal-packages/run-engine/src/run-queue/types.ts b/internal-packages/run-engine/src/run-queue/types.ts index b82153bc9..3ce859989 100644 --- a/internal-packages/run-engine/src/run-queue/types.ts +++ b/internal-packages/run-engine/src/run-queue/types.ts @@ -134,6 +134,7 @@ export interface RunQueueKeyProducer { ckIndexKeyFromQueue(queue: string): string; ckVtimeKeyFromQueue(queue: string): string; ckVtimeFloorKeyFromQueue(queue: string): string; + ckVtimeIdleKeyFromQueue(queue: string): string; baseQueueKeyFromQueue(queue: string): string; isCkWildcard(queue: string): boolean; toCkWildcard(queue: string): string;