diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 4dcc88603..d3575ab27 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -4705,12 +4705,17 @@ if totalConcurrencyEnabled then -- so a member with no message key is provably dead. Members of re-queued -- runs keep their message key and clear through the mirrored ack when the -- run completes. The short lock bounds a saturated queue to one pass per - -- interval instead of one per dequeue attempt. + -- interval, and SSCAN with a persisted cursor bounds each pass to one + -- batch so a large set never blocks Redis for a full traversal; successive + -- passes cover the whole set. if groupCurrentConcurrency >= totalConcurrencyLimit then local reconcileLockKey = groupConcurrencyKey .. ':reconcileLock' if redis.call('SET', reconcileLockKey, '1', 'NX', 'EX', '10') then - local groupMembers = redis.call('SMEMBERS', groupConcurrencyKey) - for _, groupMemberId in ipairs(groupMembers) do + local reconcileCursorKey = groupConcurrencyKey .. ':reconcileCursor' + local reconcileCursor = redis.call('GET', reconcileCursorKey) or '0' + local scanResult = redis.call('SSCAN', groupConcurrencyKey, reconcileCursor, 'COUNT', '500') + redis.call('SET', reconcileCursorKey, scanResult[1], 'EX', '3600') + for _, groupMemberId in ipairs(scanResult[2]) do if redis.call('EXISTS', messageKeyPrefix .. groupMemberId) == 0 then redis.call('SREM', groupConcurrencyKey, groupMemberId) end diff --git a/internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts b/internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts index 7ad89c8c5..cd6d299f1 100644 --- a/internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts @@ -364,4 +364,49 @@ describe("RunQueue total concurrency limit", () => { } } ); + + redisTest( + "reconciles a large leaked backlog across bounded passes", + async ({ redisContainer }) => { + const queue = createQueue(redisContainer, true); + try { + const keys = testOptions.keys; + const groupKey = keys.queueGroupConcurrencyKey(authenticatedEnvDev, "task/my-task"); + await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5); + await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1); + + /** 1,200 dead members: more than one SSCAN batch, none with a message key. */ + const dead = Array.from({ length: 1200 }, (_, i) => `dead-${i}`); + await queue.redis.sadd(groupKey, ...dead); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1200); + + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ + runId: "r0", + concurrencyKey: "ck-a", + timestamp: Date.now() - 1000, + }), + workerQueue: "main", + }); + + /** + * Each dequeue attempt reconciles at most one SSCAN batch behind a 10s + * lock; dropping the lock between polls lets the passes run back to + * back instead of waiting out the interval. + */ + const r0Admitted = await waitFor(async () => { + await queue.redis.del(`${groupKey}:reconcileLock`); + const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main", { + blockingPop: false, + }); + return next?.messageId === "r0"; + }, 30_000); + expect(r0Admitted).toBe(true); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + } finally { + await queue.quit(); + } + } + ); });