From 515ca6ffde069922d64725cb2a19cc7c232e87e1 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Fri, 28 Aug 2026 17:31:01 +0100 Subject: [PATCH 1/5] feat(run-engine,sdk,webapp): queue total concurrency limit across keys On a queue used with concurrencyKey, concurrencyLimit applies to each key value independently, so nothing bounds the queue as a whole short of the environment limit. The new totalConcurrencyLimit queue option caps in-flight runs across all keys of the queue while each key still gets at most concurrencyLimit. Enforcement lives in the concurrency-key dequeue and enqueue fast-path scripts, gated behind RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED (default off). A per-base-queue groupConcurrency set tracks total in-flight; every release path mirrors its per-key removal into that set unconditionally so the set stays correct across flag toggles. --- .changeset/queue-total-concurrency-limit.md | 18 ++ apps/webapp/app/env.server.ts | 1 + apps/webapp/app/v3/runEngine.server.ts | 1 + apps/webapp/app/v3/runQueue.server.ts | 17 ++ .../services/createBackgroundWorker.server.ts | 23 ++ .../migration.sql | 2 + .../database/prisma/schema.prisma | 3 + .../run-engine/src/engine/index.ts | 1 + .../run-engine/src/engine/types.ts | 2 + .../run-engine/src/run-queue/index.ts | 235 +++++++++++++-- .../run-engine/src/run-queue/keyProducer.ts | 28 ++ .../run-queue/tests/totalConcurrency.test.ts | 284 ++++++++++++++++++ .../run-engine/src/run-queue/types.ts | 4 + packages/core/src/v3/schemas/schemas.ts | 6 + packages/core/src/v3/types/queues.ts | 21 ++ packages/core/src/v3/types/tasks.ts | 1 + packages/trigger-sdk/src/v3/shared.ts | 2 + 17 files changed, 617 insertions(+), 32 deletions(-) create mode 100644 .changeset/queue-total-concurrency-limit.md create mode 100644 internal-packages/database/prisma/migrations/20260827120000_add_task_queue_total_concurrency_limit/migration.sql create mode 100644 internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts diff --git a/.changeset/queue-total-concurrency-limit.md b/.changeset/queue-total-concurrency-limit.md new file mode 100644 index 00000000000..cc6aab6e2af --- /dev/null +++ b/.changeset/queue-total-concurrency-limit.md @@ -0,0 +1,18 @@ +--- +"@trigger.dev/sdk": patch +"@trigger.dev/core": patch +--- + +Cap a queue's total concurrency across all of its `concurrencyKey` values with the new `totalConcurrencyLimit` queue option. On a keyed queue, `concurrencyLimit` applies to each key value independently, so ten active keys with a limit of 5 can run 50 at once. `totalConcurrencyLimit` bounds the whole queue while each key still gets at most `concurrencyLimit`. + +```ts +import { queue } from "@trigger.dev/sdk"; + +export const perUserQueue = queue({ + name: "per-user-queue", + concurrencyLimit: 1, + totalConcurrencyLimit: 10, +}); +``` + +Enforcement happens server side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled total concurrency limits accept the option but do not enforce it yet. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 2b1fba86980..8b660127648 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -1377,6 +1377,7 @@ const EnvironmentSchema = z RUN_ENGINE_RUN_QUEUE_LOG_LEVEL: z .enum(["log", "error", "warn", "info", "debug"]) .default("info"), + RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED: z.string().default("0"), RUN_ENGINE_TREAT_PRODUCTION_EXECUTION_STALLS_AS_OOM: z.string().default("0"), RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED: z.string().default("0"), RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MIN_MS: z.coerce.number().int().default(50), diff --git a/apps/webapp/app/v3/runEngine.server.ts b/apps/webapp/app/v3/runEngine.server.ts index 76ac35f349b..7a6cb0f8d3d 100644 --- a/apps/webapp/app/v3/runEngine.server.ts +++ b/apps/webapp/app/v3/runEngine.server.ts @@ -62,6 +62,7 @@ function createRunEngine() { queue: { defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT, defaultEnvConcurrencyBurstFactor: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_BURST_FACTOR, + totalConcurrencyEnabled: env.RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED === "1", logLevel: env.RUN_ENGINE_RUN_QUEUE_LOG_LEVEL, redis: { keyPrefix: "engine:", diff --git a/apps/webapp/app/v3/runQueue.server.ts b/apps/webapp/app/v3/runQueue.server.ts index 304819f9f36..2005b059be4 100644 --- a/apps/webapp/app/v3/runQueue.server.ts +++ b/apps/webapp/app/v3/runQueue.server.ts @@ -40,6 +40,23 @@ export async function updateQueueConcurrencyLimits( await engine.runQueue.updateQueueConcurrencyLimits(environment, queueName, concurrency); } +/** Updates the RunQueue total concurrency limit for a queue (the cap across all concurrency-key values) */ +export async function updateQueueTotalConcurrencyLimits( + environment: AuthenticatedEnvironment, + queueName: string, + totalConcurrency: number +) { + await engine.runQueue.updateQueueTotalConcurrencyLimits(environment, queueName, totalConcurrency); +} + +/** Removes the RunQueue total concurrency limit for a queue */ +export async function removeQueueTotalConcurrencyLimits( + environment: AuthenticatedEnvironment, + queueName: string +) { + await engine.runQueue.removeQueueTotalConcurrencyLimits(environment, queueName); +} + /** Removes the RunQueue limits for a queue */ export async function removeQueueConcurrencyLimits( environment: AuthenticatedEnvironment, diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index f8a69e83d2b..f1cebe640d5 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -33,8 +33,10 @@ import { generateFriendlyId } from "../friendlyIdentifiers"; import { engine } from "../runEngine.server"; import { removeQueueConcurrencyLimits, + removeQueueTotalConcurrencyLimits, updateEnvConcurrencyLimits, updateQueueConcurrencyLimits, + updateQueueTotalConcurrencyLimits, } from "../runQueue.server"; import { scheduleEngine } from "../scheduleEngine.server"; import { normalizeScheduleWindow } from "../scheduleWindow.server"; @@ -401,6 +403,7 @@ async function createWorkerTask( { name: task.queue?.name ?? `task/${task.id}`, concurrencyLimit: task.queue?.concurrencyLimit, + totalConcurrencyLimit: task.queue?.totalConcurrencyLimit, }, task.id, task.queue?.name ? "NAMED" : "VIRTUAL", @@ -552,6 +555,7 @@ async function createWorkerQueue( const taskQueue = await upsertWorkerQueueRecord( queueName, baseConcurrencyLimit ?? null, + queue.totalConcurrencyLimit ?? null, orderableName, queueType, worker, @@ -560,6 +564,21 @@ async function createWorkerQueue( const newConcurrencyLimit = taskQueue.concurrencyLimit; + /** + * The total limit key is separate from the per-queue limit key that pause zeroes, + * so it is safe to sync it regardless of the paused state. The engine clamps it + * to the environment limit at read time, so the raw declared value is stored. + */ + if (typeof taskQueue.totalConcurrencyLimit === "number") { + await updateQueueTotalConcurrencyLimits( + environment, + taskQueue.name, + taskQueue.totalConcurrencyLimit + ); + } else { + await removeQueueTotalConcurrencyLimits(environment, taskQueue.name); + } + if (!taskQueue.paused) { if (typeof newConcurrencyLimit === "number") { logger.debug("createWorkerQueue: updating concurrency limit", { @@ -598,6 +617,7 @@ async function createWorkerQueue( async function upsertWorkerQueueRecord( queueName: string, concurrencyLimit: number | null, + totalConcurrencyLimit: number | null, orderableName: string, queueType: TaskQueueType, worker: BackgroundWorker, @@ -624,6 +644,7 @@ async function upsertWorkerQueueRecord( name: queueName, orderableName, concurrencyLimit, + totalConcurrencyLimit, runtimeEnvironmentId: worker.runtimeEnvironmentId, projectId: worker.projectId, type: queueType, @@ -648,6 +669,7 @@ async function upsertWorkerQueueRecord( // If overridden, keep current limit and update base; otherwise update limit normally concurrencyLimit: hasOverride ? undefined : concurrencyLimit, concurrencyLimitBase: hasOverride ? concurrencyLimit : undefined, + totalConcurrencyLimit, }, }); } @@ -659,6 +681,7 @@ async function upsertWorkerQueueRecord( return await upsertWorkerQueueRecord( queueName, concurrencyLimit, + totalConcurrencyLimit, orderableName, queueType, worker, diff --git a/internal-packages/database/prisma/migrations/20260827120000_add_task_queue_total_concurrency_limit/migration.sql b/internal-packages/database/prisma/migrations/20260827120000_add_task_queue_total_concurrency_limit/migration.sql new file mode 100644 index 00000000000..21cdb8664d9 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260827120000_add_task_queue_total_concurrency_limit/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimit" INTEGER; diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index a1423f340c0..ac0ac09b2f0 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -1974,6 +1974,9 @@ model TaskQueue { /// percentage (the source of truth). The absolute concurrencyLimit is materialized from it. /// Decimal(5,2) allows fractional percentages like 12.50% (0.01–100.00). concurrencyLimitOverridePercent Decimal? @db.Decimal(5, 2) + /// Caps total concurrent runs across ALL concurrencyKey values of this queue + /// (concurrencyLimit applies per key value). Null = no total cap. + totalConcurrencyLimit Int? rateLimit Json? paused Boolean @default(false) diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 7917ba68303..fc4d76de6db 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -209,6 +209,7 @@ export class RunEngine { queueSelectionStrategy: new FairQueueSelectionStrategy(queueSelectionStrategyOptions), defaultEnvConcurrency: options.queue?.defaultEnvConcurrency ?? 10, defaultEnvConcurrencyBurstFactor: options.queue?.defaultEnvConcurrencyBurstFactor, + totalConcurrencyEnabled: options.queue?.totalConcurrencyEnabled, logger: new Logger("RunQueue", options.queue?.logLevel ?? "info"), redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` }, retryOptions: options.queue?.retryOptions, diff --git a/internal-packages/run-engine/src/engine/types.ts b/internal-packages/run-engine/src/engine/types.ts index 2516776373d..ed1b3a0c2c9 100644 --- a/internal-packages/run-engine/src/engine/types.ts +++ b/internal-packages/run-engine/src/engine/types.ts @@ -91,6 +91,8 @@ export type RunEngineOptions = { defaultEnvConcurrency?: number; defaultEnvConcurrencyBurstFactor?: number; logLevel?: LogLevel; + /** Enforce per-queue total concurrency limits across concurrency-key variants. See RunQueueOptions.totalConcurrencyEnabled. */ + totalConcurrencyEnabled?: boolean; /** Optional queue-metrics emitter; enables gauge + counter emission from the RunQueue. */ queueMetrics?: RunQueueMetricsEmitter; queueSelectionStrategyOptions?: Pick< diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 4afa5b2bab9..2b8c11fe62d 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -179,6 +179,14 @@ export type RunQueueOptions = { * CK operation re-anchors from ckIndex. Default: 86400 (24h). */ counterTtlSeconds?: number; + /** + * When true, concurrency-keyed queues maintain a per-base-queue groupConcurrency SET + * (total in-flight across all key variants) and enforce the queue's total concurrency + * limit at admit time. Default false: admit paths are byte-identical to before, and + * only the release-side SREM mirror runs (a no-op on an absent set), so the flag can + * be flipped on a fleet that has fully rolled onto this build without draining queues. + */ + totalConcurrencyEnabled?: boolean; workerOptions?: { pollIntervalMs?: number; immediatePollIntervalMs?: number; @@ -464,6 +472,36 @@ export class RunQueue { return result ? Number(result) : undefined; } + public async updateQueueTotalConcurrencyLimits( + env: MinimalAuthenticatedEnvironment, + queue: string, + totalConcurrency: number + ) { + return this.redis.set(this.keys.queueTotalConcurrencyLimitKey(env, queue), totalConcurrency); + } + + public async removeQueueTotalConcurrencyLimits( + env: MinimalAuthenticatedEnvironment, + queue: string + ) { + return this.redis.del(this.keys.queueTotalConcurrencyLimitKey(env, queue)); + } + + public async getQueueTotalConcurrencyLimit(env: MinimalAuthenticatedEnvironment, queue: string) { + const result = await this.redis.get(this.keys.queueTotalConcurrencyLimitKey(env, queue)); + + return result ? Number(result) : undefined; + } + + /** + * Total in-flight runs across all concurrency-key variants of a queue (the + * groupConcurrency SET cardinality). Only maintained while totalConcurrencyEnabled + * is on; returns 0 otherwise. + */ + public async totalConcurrencyOfQueue(env: MinimalAuthenticatedEnvironment, queue: string) { + return this.redis.scard(this.keys.queueGroupConcurrencyKey(env, queue)); + } + public async updateEnvConcurrencyLimits(env: MinimalAuthenticatedEnvironment) { await this.#callUpdateEnvironmentConcurrencyLimits({ envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env), @@ -1240,6 +1278,7 @@ export class RunQueue { this.keys.envCurrentDequeuedKeyFromQueue(message.queue), this.keys.queueRunningCounterKeyFromQueue(message.queue), this.keys.ckIndexKeyFromQueue(message.queue), + this.keys.queueGroupConcurrencyKeyFromQueue(message.queue), messageId, this.options.redis.keyPrefix ?? "", String(this.counterTtlSeconds) @@ -2189,6 +2228,11 @@ export class RunQueue { const lengthCounterKey = this.keys.queueLengthCounterKeyFromQueue(message.queue); const baseQueueKey = this.keys.baseQueueKeyFromQueue(message.queue); const ckKeyPrefix = this.options.redis.keyPrefix ?? ""; + const groupConcurrencyKey = this.keys.queueGroupConcurrencyKeyFromQueue(message.queue); + const totalConcurrencyLimitKey = this.keys.queueTotalConcurrencyLimitKeyFromQueue( + message.queue + ); + const totalConcurrencyEnabledArg = this.options.totalConcurrencyEnabled ? "1" : "0"; if (ttlInfo) { result = await this.redis.enqueueMessageWithTtlCkTracked( @@ -2209,6 +2253,8 @@ export class RunQueue { envConcurrencyLimitBurstFactorKey, lengthCounterKey, baseQueueKey, + groupConcurrencyKey, + totalConcurrencyLimitKey, // args queueName, messageId, @@ -2224,6 +2270,7 @@ export class RunQueue { enableFastPathArg, ckKeyPrefix, String(this.counterTtlSeconds), + totalConcurrencyEnabledArg, metricsGaugeArg ); } else { @@ -2244,6 +2291,8 @@ export class RunQueue { envConcurrencyLimitBurstFactorKey, lengthCounterKey, baseQueueKey, + groupConcurrencyKey, + totalConcurrencyLimitKey, // args queueName, messageId, @@ -2257,6 +2306,7 @@ export class RunQueue { enableFastPathArg, ckKeyPrefix, String(this.counterTtlSeconds), + totalConcurrencyEnabledArg, metricsGaugeArg ); } @@ -2522,6 +2572,8 @@ export class RunQueue { ttlQueueKey, lengthCounterKey, runningCounterKey, + this.keys.queueGroupConcurrencyKeyFromQueue(ckWildcardQueue), + this.keys.queueTotalConcurrencyLimitKeyFromQueue(ckWildcardQueue), //args ckWildcardQueue, String(Date.now()), @@ -2529,6 +2581,7 @@ export class RunQueue { String(this.options.defaultEnvConcurrencyBurstFactor ?? 1), this.options.redis.keyPrefix ?? "", String(maxCount), + this.options.totalConcurrencyEnabled ? "1" : "0", metricsGaugeArg ); @@ -2778,6 +2831,7 @@ export class RunQueue { ckIndexKey, lengthCounterKey, runningCounterKey, + this.keys.queueGroupConcurrencyKeyFromQueue(messageQueue), messageId, messageQueue, messageKeyValue, @@ -2841,6 +2895,7 @@ export class RunQueue { envCurrentDequeuedKey, this.keys.queueRunningCounterKeyFromQueue(queue), this.keys.ckIndexKeyFromQueue(queue), + this.keys.queueGroupConcurrencyKeyFromQueue(queue), messageId, this.options.redis.keyPrefix ?? "", String(this.counterTtlSeconds) @@ -2907,6 +2962,7 @@ export class RunQueue { ckIndexKey, lengthCounterKey, runningCounterKey, + this.keys.queueGroupConcurrencyKeyFromQueue(messageQueue), //args messageId, messageQueue, @@ -2970,6 +3026,7 @@ export class RunQueue { ckIndexKey, lengthCounterKey, runningCounterKey, + this.keys.queueGroupConcurrencyKeyFromQueue(messageQueue), messageId, messageQueue, ckWildcardName @@ -3758,7 +3815,7 @@ return __qmret(0) // *Tracked variants of dequeueMessageFromKey and the ack/nack/dlq/release/clear // scripts. this.redis.defineCommand("enqueueMessageCkTracked", { - numberOfKeys: 15, + numberOfKeys: 17, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -3777,6 +3834,9 @@ local envConcurrencyLimitBurstFactorKey = KEYS[13] -- Counter keys (KEYS 14-15) local lengthCounterKey = KEYS[14] local baseQueueKey = KEYS[15] +-- Total-cap keys (KEYS 16-17) +local groupConcurrencyKey = KEYS[16] +local totalConcurrencyLimitKey = KEYS[17] local queueName = ARGV[1] local messageId = ARGV[2] @@ -3793,6 +3853,7 @@ local enableFastPath = ARGV[10] local keyPrefix = ARGV[11] -- TTL (seconds) applied to counter lazy-init SETs local counterTtl = ARGV[12] +local totalConcurrencyEnabled = ARGV[13] == '1' ${QUEUE_METRICS_GAUGE_PRELUDE} @@ -3813,15 +3874,34 @@ if enableFastPath == '1' then ) if queueCurrent < queueLimit then - redis.call('SET', messageKey, messageData) - redis.call('SADD', queueCurrentConcurrencyKey, messageId) - redis.call('SADD', envCurrentConcurrencyKey, messageId) - redis.call('RPUSH', workerQueueKey, messageKeyValue) + -- Total-cap gate: a fast-path admit consumes a group slot, so it must + -- respect the env-clamped total limit. At the cap we fall through to the + -- slow path (the message queues; the dequeue gate holds it). + local totalAllowsFastPath = true + if totalConcurrencyEnabled then + local rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey) + if rawTotalLimit then + local totalLimit = math.min(tonumber(rawTotalLimit), envLimit) + if tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') >= totalLimit then + totalAllowsFastPath = false + end + end + end + + if totalAllowsFastPath then + redis.call('SET', messageKey, messageData) + redis.call('SADD', queueCurrentConcurrencyKey, messageId) + redis.call('SADD', envCurrentConcurrencyKey, messageId) + if totalConcurrencyEnabled then + redis.call('SADD', groupConcurrencyKey, messageId) + end + redis.call('RPUSH', workerQueueKey, messageKeyValue) ${QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA} - -- Fast-path skips the CK variant zset entirely; lengthCounter is unchanged. - -- runningCounter is bumped later by dequeueMessageFromKeyTracked when the - -- worker pulls the message from the worker queue. - return __qmret(1) + -- Fast-path skips the CK variant zset entirely; lengthCounter is unchanged. + -- runningCounter is bumped later by dequeueMessageFromKeyTracked when the + -- worker pulls the message from the worker queue. + return __qmret(1) + end end end end @@ -3873,8 +3953,12 @@ if queueName ~= ckWildcardName then redis.call('ZREM', masterQueueKey, queueName) end --- Update the concurrency keys -redis.call('SREM', queueCurrentConcurrencyKey, messageId) +-- Update the concurrency keys. The groupConcurrency SREM mirrors the per-CK SREM +-- unconditionally (no flag check) so a disabled flag still drains the group set. +local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId) +if removedFromCurrentConcurrency == 1 then + redis.call('SREM', groupConcurrencyKey, messageId) +end redis.call('SREM', envCurrentConcurrencyKey, messageId) redis.call('SREM', queueCurrentDequeuedKey, messageId) redis.call('SREM', envCurrentDequeuedKey, messageId) @@ -3885,7 +3969,7 @@ return __qmret(0) }); this.redis.defineCommand("enqueueMessageWithTtlCkTracked", { - numberOfKeys: 16, + numberOfKeys: 18, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -3905,6 +3989,9 @@ local envConcurrencyLimitBurstFactorKey = KEYS[14] -- Counter keys (KEYS 15-16) local lengthCounterKey = KEYS[15] local baseQueueKey = KEYS[16] +-- Total-cap keys (KEYS 17-18) +local groupConcurrencyKey = KEYS[17] +local totalConcurrencyLimitKey = KEYS[18] local queueName = ARGV[1] local messageId = ARGV[2] @@ -3923,6 +4010,7 @@ local enableFastPath = ARGV[12] local keyPrefix = ARGV[13] -- TTL (seconds) applied to counter lazy-init SETs local counterTtl = ARGV[14] +local totalConcurrencyEnabled = ARGV[15] == '1' ${QUEUE_METRICS_GAUGE_PRELUDE} @@ -3943,12 +4031,29 @@ if enableFastPath == '1' then ) if queueCurrent < queueLimit then - redis.call('SET', messageKey, messageData) - redis.call('SADD', queueCurrentConcurrencyKey, messageId) - redis.call('SADD', envCurrentConcurrencyKey, messageId) - redis.call('RPUSH', workerQueueKey, messageKeyValue) + -- Total-cap gate: see enqueueMessageCkTracked. + local totalAllowsFastPath = true + if totalConcurrencyEnabled then + local rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey) + if rawTotalLimit then + local totalLimit = math.min(tonumber(rawTotalLimit), envLimit) + if tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') >= totalLimit then + totalAllowsFastPath = false + end + end + end + + if totalAllowsFastPath then + redis.call('SET', messageKey, messageData) + redis.call('SADD', queueCurrentConcurrencyKey, messageId) + redis.call('SADD', envCurrentConcurrencyKey, messageId) + if totalConcurrencyEnabled then + redis.call('SADD', groupConcurrencyKey, messageId) + end + redis.call('RPUSH', workerQueueKey, messageKeyValue) ${QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA} - return __qmret(1) + return __qmret(1) + end end end end @@ -3996,8 +4101,12 @@ if queueName ~= ckWildcardName then redis.call('ZREM', masterQueueKey, queueName) end --- Update the concurrency keys -redis.call('SREM', queueCurrentConcurrencyKey, messageId) +-- Update the concurrency keys. The groupConcurrency SREM mirrors the per-CK SREM +-- unconditionally (no flag check) so a disabled flag still drains the group set. +local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId) +if removedFromCurrentConcurrency == 1 then + redis.call('SREM', groupConcurrencyKey, messageId) +end redis.call('SREM', envCurrentConcurrencyKey, messageId) redis.call('SREM', queueCurrentDequeuedKey, messageId) redis.call('SREM', envCurrentDequeuedKey, messageId) @@ -4180,7 +4289,7 @@ for i, member in ipairs(expiredMembers) do local concurrencyKey = queueKey .. ":currentConcurrency" local dequeuedKey = queueKey .. ":currentDequeued" - redis.call('SREM', concurrencyKey, runId) + local removedFromCurrent = redis.call('SREM', concurrencyKey, runId) local removedFromDequeued = redis.call('SREM', dequeuedKey, runId) local projMatch = string.match(rawQueueKey, ":proj:([^:]+):env:") @@ -4200,6 +4309,10 @@ for i, member in ipairs(expiredMembers) do if removedFromDequeued == 1 then decrFloored(runningCounterKey) end + -- Mirror the per-CK currentConcurrency SREM into the base groupConcurrency set + if removedFromCurrent == 1 then + redis.call('SREM', keyPrefix .. ckMatch .. ":groupConcurrency", runId) + end local ckIndexKey = keyPrefix .. ckMatch .. ":ckIndex" local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES') @@ -4514,7 +4627,7 @@ return results // (normal dequeue, TTL-expired, or stale-orphan path — all of which were // counted at enqueue time). this.redis.defineCommand("dequeueMessagesFromCkQueueTracked", { - numberOfKeys: 11, + numberOfKeys: 13, lua: ` local ckIndexKey = KEYS[1] local queueConcurrencyLimitKey = KEYS[2] @@ -4527,6 +4640,8 @@ local masterQueueKey = KEYS[8] local ttlQueueKey = KEYS[9] local lengthCounterKey = KEYS[10] local runningCounterKey = KEYS[11] +local groupConcurrencyKey = KEYS[12] +local totalConcurrencyLimitKey = KEYS[13] local ckWildcardName = ARGV[1] local currentTime = tonumber(ARGV[2]) @@ -4534,6 +4649,7 @@ local defaultEnvConcurrencyLimit = ARGV[3] local defaultEnvConcurrencyBurstFactor = ARGV[4] local keyPrefix = ARGV[5] local maxCount = tonumber(ARGV[6] or '1') +local totalConcurrencyEnabled = ARGV[7] == '1' ${QUEUE_METRICS_GAUGE_PRELUDE} ${QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA} @@ -4558,6 +4674,20 @@ local queueConcurrencyLimit = math.min(tonumber(redis.call('GET', queueConcurren local envAvailableCapacity = envConcurrencyLimitWithBurstFactor - envCurrentConcurrency local actualMaxCount = math.min(maxCount, envAvailableCapacity) +-- Total-cap gate: bound this batch by the remaining headroom across ALL ck +-- variants (groupConcurrency SCARD vs the env-clamped total limit). Each admit +-- below SADDs into the group set and bumps dequeuedCount, and dequeuedCount is +-- bounded by actualMaxCount, so tightening here is sufficient to prevent +-- over-admitting past the cap within a single batch. +if totalConcurrencyEnabled then + local rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey) + if rawTotalLimit then + local totalConcurrencyLimit = math.min(tonumber(rawTotalLimit), envConcurrencyLimit) + local groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') + actualMaxCount = math.min(actualMaxCount, totalConcurrencyLimit - groupCurrentConcurrency) + end +end + if actualMaxCount <= 0 then return __qmret(nil) end @@ -4615,6 +4745,9 @@ for _, ckQueueName in ipairs(ckQueues) do decrLengthCounter() redis.call('SADD', ckConcurrencyKey, messageId) redis.call('SADD', envCurrentConcurrencyKey, messageId) + if totalConcurrencyEnabled then + redis.call('SADD', groupConcurrencyKey, messageId) + end if ttlQueueKey and ttlQueueKey ~= '' and ttlExpiresAt then local ttlMember = ckQueueName .. '|' .. messageId .. '|' .. (messageData.orgId or '') @@ -5104,7 +5237,7 @@ redis.call('SREM', envCurrentDequeuedKey, messageId) // removed something) and runningCounter (when SREM currentDequeued actually // removed something). this.redis.defineCommand("acknowledgeMessageCkTracked", { - numberOfKeys: 12, + numberOfKeys: 13, lua: ` -- Keys: local masterQueueKey = KEYS[1] @@ -5119,6 +5252,7 @@ local workerQueueKey = KEYS[9] local ckIndexKey = KEYS[10] local lengthCounterKey = KEYS[11] local runningCounterKey = KEYS[12] +local groupConcurrencyKey = KEYS[13] -- Args: local messageId = ARGV[1] @@ -5171,7 +5305,12 @@ end -- Update the concurrency keys. DECR runningCounter only when SREM -- currentDequeued actually removed an entry (the message was in flight). -redis.call('SREM', queueCurrentConcurrencyKey, messageId) +-- The groupConcurrency SREM mirrors the per-CK SREM so the group set drains +-- on every release path. +local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId) +if removedFromCurrentConcurrency == 1 then + redis.call('SREM', groupConcurrencyKey, messageId) +end redis.call('SREM', envCurrentConcurrencyKey, messageId) local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId) redis.call('SREM', envCurrentDequeuedKey, messageId) @@ -5190,7 +5329,7 @@ end // runningCounter (floored); ZADD back to the variant zset INCRs // lengthCounter only when ZADD reported a new entry. this.redis.defineCommand("nackMessageCkTracked", { - numberOfKeys: 11, + numberOfKeys: 12, lua: ` -- Keys: local masterQueueKey = KEYS[1] @@ -5204,6 +5343,7 @@ local envQueueKey = KEYS[8] local ckIndexKey = KEYS[9] local lengthCounterKey = KEYS[10] local runningCounterKey = KEYS[11] +local groupConcurrencyKey = KEYS[12] -- Args: local messageId = ARGV[1] @@ -5229,7 +5369,11 @@ redis.call('SET', messageKey, messageData) -- so we skip the eager lazy-init here (unlike releaseConcurrencyTracked, which -- mirrors the same DECR pattern with init). A post-TTL nack's floored DECR -- no-ops; the next dequeueMessageFromKeyTracked reseeds from current state. -redis.call('SREM', queueCurrentConcurrencyKey, messageId) +-- The groupConcurrency SREM mirrors the per-CK SREM. +local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId) +if removedFromCurrentConcurrency == 1 then + redis.call('SREM', groupConcurrencyKey, messageId) +end redis.call('SREM', envCurrentConcurrencyKey, messageId) local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId) redis.call('SREM', envCurrentDequeuedKey, messageId) @@ -5285,7 +5429,7 @@ end // Tracked variant: same as moveToDeadLetterQueueCk. ZREM may DECR // lengthCounter (defensive); SREM currentDequeued may DECR runningCounter. this.redis.defineCommand("moveToDeadLetterQueueCkTracked", { - numberOfKeys: 12, + numberOfKeys: 13, lua: ` -- Keys: local masterQueueKey = KEYS[1] @@ -5300,6 +5444,7 @@ local deadLetterQueueKey = KEYS[9] local ckIndexKey = KEYS[10] local lengthCounterKey = KEYS[11] local runningCounterKey = KEYS[12] +local groupConcurrencyKey = KEYS[13] -- Args: local messageId = ARGV[1] @@ -5349,8 +5494,12 @@ end redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageId) -- Update the concurrency keys. DECR runningCounter only when SREM --- currentDequeued actually removed an entry. -redis.call('SREM', queueCurrentConcurrencyKey, messageId) +-- currentDequeued actually removed an entry. The groupConcurrency SREM mirrors +-- the per-CK SREM. +local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId) +if removedFromCurrentConcurrency == 1 then + redis.call('SREM', groupConcurrencyKey, messageId) +end redis.call('SREM', envCurrentConcurrencyKey, messageId) local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId) redis.call('SREM', envCurrentDequeuedKey, messageId) @@ -5385,7 +5534,7 @@ redis.call('SREM', envCurrentDequeuedKey, messageId) // something. Caller should only invoke this variant for CK queues — non-CK // queues should keep calling releaseConcurrency. this.redis.defineCommand("releaseConcurrencyTracked", { - numberOfKeys: 6, + numberOfKeys: 7, lua: ` -- Keys: local queueCurrentConcurrencyKey = KEYS[1] @@ -5394,6 +5543,7 @@ local queueCurrentDequeuedKey = KEYS[3] local envCurrentDequeuedKey = KEYS[4] local runningCounterKey = KEYS[5] local ckIndexKey = KEYS[6] +local groupConcurrencyKey = KEYS[7] -- Args: local messageId = ARGV[1] @@ -5415,7 +5565,10 @@ if redis.call('EXISTS', runningCounterKey) == 0 then redis.call('SET', runningCounterKey, total, 'EX', counterTtl) end -redis.call('SREM', queueCurrentConcurrencyKey, messageId) +local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId) +if removedFromCurrentConcurrency == 1 then + redis.call('SREM', groupConcurrencyKey, messageId) +end redis.call('SREM', envCurrentConcurrencyKey, messageId) local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId) redis.call('SREM', envCurrentDequeuedKey, messageId) @@ -5530,7 +5683,7 @@ redis.call('SREM', envCurrentDequeuedKey, messageId) // Tracked variant of clearMessageFromConcurrencySets — see releaseConcurrencyTracked // for the contract. Only invoke for CK queues. this.redis.defineCommand("clearMessageFromConcurrencySetsTracked", { - numberOfKeys: 6, + numberOfKeys: 7, lua: ` -- Keys: local queueCurrentConcurrencyKey = KEYS[1] @@ -5539,6 +5692,7 @@ local queueCurrentDequeuedKey = KEYS[3] local envCurrentDequeuedKey = KEYS[4] local runningCounterKey = KEYS[5] local ckIndexKey = KEYS[6] +local groupConcurrencyKey = KEYS[7] -- Args: local messageId = ARGV[1] @@ -5556,7 +5710,10 @@ if redis.call('EXISTS', runningCounterKey) == 0 then redis.call('SET', runningCounterKey, total, 'EX', counterTtl) end -redis.call('SREM', queueCurrentConcurrencyKey, messageId) +local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId) +if removedFromCurrentConcurrency == 1 then + redis.call('SREM', groupConcurrencyKey, messageId) +end redis.call('SREM', envCurrentConcurrencyKey, messageId) local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId) redis.call('SREM', envCurrentDequeuedKey, messageId) @@ -5950,6 +6107,8 @@ declare module "@internal/redis" { envConcurrencyLimitBurstFactorKey: string, lengthCounterKey: string, baseQueueKey: string, + groupConcurrencyKey: string, + totalConcurrencyLimitKey: string, queueName: string, messageId: string, messageData: string, @@ -5962,6 +6121,7 @@ declare module "@internal/redis" { enableFastPath: string, keyPrefix: string, counterTtl: string, + totalConcurrencyEnabled: string, metricsEnabled: string, callback?: Callback<[number, number[] | null]> ): Result<[number, number[] | null], Context>; @@ -5983,6 +6143,8 @@ declare module "@internal/redis" { envConcurrencyLimitBurstFactorKey: string, lengthCounterKey: string, baseQueueKey: string, + groupConcurrencyKey: string, + totalConcurrencyLimitKey: string, queueName: string, messageId: string, messageData: string, @@ -5997,6 +6159,7 @@ declare module "@internal/redis" { enableFastPath: string, keyPrefix: string, counterTtl: string, + totalConcurrencyEnabled: string, metricsEnabled: string, callback?: Callback<[number, number[] | null]> ): Result<[number, number[] | null], Context>; @@ -6013,12 +6176,15 @@ declare module "@internal/redis" { ttlQueueKey: string, lengthCounterKey: string, runningCounterKey: string, + groupConcurrencyKey: string, + totalConcurrencyLimitKey: string, ckWildcardName: string, currentTime: string, defaultEnvConcurrencyLimit: string, defaultEnvConcurrencyBurstFactor: string, keyPrefix: string, maxCount: string, + totalConcurrencyEnabled: string, metricsEnabled: string, callback?: Callback<[string[] | null, number[] | null]> ): Result<[string[] | null, number[] | null], Context>; @@ -6043,6 +6209,7 @@ declare module "@internal/redis" { ckIndexKey: string, lengthCounterKey: string, runningCounterKey: string, + groupConcurrencyKey: string, messageId: string, messageQueueName: string, messageKeyValue: string, @@ -6063,6 +6230,7 @@ declare module "@internal/redis" { ckIndexKey: string, lengthCounterKey: string, runningCounterKey: string, + groupConcurrencyKey: string, messageId: string, messageQueueName: string, messageData: string, @@ -6086,6 +6254,7 @@ declare module "@internal/redis" { ckIndexKey: string, lengthCounterKey: string, runningCounterKey: string, + groupConcurrencyKey: string, messageId: string, messageQueueName: string, ckWildcardName: string, @@ -6111,6 +6280,7 @@ declare module "@internal/redis" { envCurrentDequeuedKey: string, runningCounterKey: string, ckIndexKey: string, + groupConcurrencyKey: string, messageId: string, keyPrefix: string, counterTtl: string, @@ -6124,6 +6294,7 @@ declare module "@internal/redis" { envCurrentDequeuedKey: string, runningCounterKey: string, ckIndexKey: string, + groupConcurrencyKey: string, messageId: string, keyPrefix: string, counterTtl: string, diff --git a/internal-packages/run-engine/src/run-queue/keyProducer.ts b/internal-packages/run-engine/src/run-queue/keyProducer.ts index 0609b3d719b..98028f5af7b 100644 --- a/internal-packages/run-engine/src/run-queue/keyProducer.ts +++ b/internal-packages/run-engine/src/run-queue/keyProducer.ts @@ -24,6 +24,8 @@ const constants = { CK_INDEX_PART: "ckIndex", LENGTH_COUNTER_PART: "lengthCounter", RUNNING_COUNTER_PART: "runningCounter", + GROUP_CONCURRENCY_PART: "groupConcurrency", + TOTAL_CONCURRENCY_LIMIT_PART: "totalConcurrency", } as const; export class RunQueueFullKeyProducer implements RunQueueKeyProducer { @@ -338,6 +340,32 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer { return `${this.baseQueueKeyFromQueue(queue)}:${constants.RUNNING_COUNTER_PART}`; } + /** + * SET of in-flight messageIds across ALL concurrency-key variants of a base queue. + * SCARD of this set is the queue's total running count, gated by the total + * concurrency limit. Lives at the base queue so every ck variant shares it. + */ + queueGroupConcurrencyKey(env: RunQueueKeyProducerEnvironment, queue: string): string { + return `${this.queueKey(env, queue)}:${constants.GROUP_CONCURRENCY_PART}`; + } + + queueGroupConcurrencyKeyFromQueue(queue: string): string { + return `${this.baseQueueKeyFromQueue(queue)}:${constants.GROUP_CONCURRENCY_PART}`; + } + + /** + * String key holding the queue's total concurrency limit (the cap across all + * concurrency-key variants). Absent = no total cap. Readers clamp to the + * environment limit; the raw requested value is what's stored. + */ + queueTotalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string { + return `${this.queueKey(env, queue)}:${constants.TOTAL_CONCURRENCY_LIMIT_PART}`; + } + + queueTotalConcurrencyLimitKeyFromQueue(queue: string): string { + return `${this.baseQueueKeyFromQueue(queue)}:${constants.TOTAL_CONCURRENCY_LIMIT_PART}`; + } + isCkWildcard(queue: string): boolean { return queue.endsWith(":ck:*"); } 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 new file mode 100644 index 00000000000..41c9ce73228 --- /dev/null +++ b/internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts @@ -0,0 +1,284 @@ +import { assertNonNullable, redisTest } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { setTimeout } from "node:timers/promises"; +import { describe } from "vitest"; +import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js"; +import { RunQueue } from "../index.js"; +import { RunQueueFullKeyProducer } from "../keyProducer.js"; +import type { InputPayload } from "../types.js"; +import { Decimal } from "@trigger.dev/database"; + +const testOptions = { + name: "rq", + tracer: trace.getTracer("rq"), + workers: 1, + defaultEnvConcurrency: 25, + retryOptions: { + maxAttempts: 5, + factor: 1.1, + minTimeoutInMs: 100, + maxTimeoutInMs: 1_000, + randomize: true, + }, + keys: new RunQueueFullKeyProducer(), +}; + +const authenticatedEnvDev = { + id: "e1234", + type: "DEVELOPMENT" as const, + maximumConcurrencyLimit: 10, + concurrencyLimitBurstFactor: new Decimal(2.0), + project: { id: "p1234" }, + organization: { id: "o1234" }, +}; + +function createQueue(redisContainer: any, totalConcurrencyEnabled: boolean) { + return new RunQueue({ + ...testOptions, + totalConcurrencyEnabled, + queueSelectionStrategy: new FairQueueSelectionStrategy({ + redis: { + keyPrefix: "runqueue:test:", + host: redisContainer.getHost(), + port: redisContainer.getPort(), + }, + keys: testOptions.keys, + }), + redis: { + keyPrefix: "runqueue:test:", + 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: "task/my-task", + timestamp: Date.now(), + attempt: 0, + ...overrides, + }; +} + +async function waitFor(condition: () => Promise, timeoutMs = 20_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await condition()) { + return true; + } + await setTimeout(250); + } + return condition(); +} + +vi.setConfig({ testTimeout: 60_000 }); + +describe("RunQueue total concurrency limit", () => { + redisTest( + "caps in-flight runs across concurrency keys at the total limit", + async ({ redisContainer }) => { + const queue = createQueue(redisContainer, true); + try { + await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5); + await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 2); + + const now = Date.now(); + for (const [i, ck] of ["ck-a", "ck-a", "ck-b", "ck-b"].entries()) { + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ + runId: `r${i}`, + concurrencyKey: ck, + timestamp: now - 1000 + i, + }), + workerQueue: "main", + }); + } + + const admittedTwo = await waitFor( + async () => + (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 2 + ); + expect(admittedTwo).toBe(true); + + /** + * The remaining two messages must stay queued: give the master consumers a + * couple of extra polling cycles to prove the gate holds, not just that it + * hadn't caught up yet. + */ + await setTimeout(2000); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(2); + expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(2); + + const dequeued1 = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + assertNonNullable(dequeued1); + const dequeued2 = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + assertNonNullable(dequeued2); + + await queue.acknowledgeMessage(authenticatedEnvDev.organization.id, dequeued1.messageId); + + const thirdAdmitted = await waitFor(async () => { + const total = await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task"); + const queued = await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task"); + return total === 2 && queued === 1; + }); + expect(thirdAdmitted).toBe(true); + } finally { + await queue.quit(); + } + } + ); + + redisTest( + "still enforces the per-key limit under the total limit", + async ({ redisContainer }) => { + const queue = createQueue(redisContainer, true); + try { + await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1); + await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 10); + + const now = Date.now(); + for (const i of [0, 1]) { + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ + runId: `r${i}`, + concurrencyKey: "ck-a", + timestamp: now - 1000 + i, + }), + workerQueue: "main", + }); + } + + const oneAdmitted = await waitFor( + async () => + (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 1 + ); + expect(oneAdmitted).toBe(true); + + await setTimeout(2000); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + expect( + await queue.currentConcurrencyOfQueue(authenticatedEnvDev, "task/my-task", "ck-a") + ).toBe(1); + } finally { + await queue.quit(); + } + } + ); + + redisTest( + "ignores the stored total limit and maintains no group set when disabled", + async ({ redisContainer }) => { + const queue = createQueue(redisContainer, false); + try { + await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5); + await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1); + + const now = Date.now(); + for (const [i, ck] of ["ck-a", "ck-b", "ck-c"].entries()) { + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ + runId: `r${i}`, + concurrencyKey: ck, + timestamp: now - 1000 + i, + }), + workerQueue: "main", + }); + } + + const allAdmitted = await waitFor( + async () => (await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")) === 0 + ); + expect(allAdmitted).toBe(true); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(0); + } finally { + await queue.quit(); + } + } + ); + + redisTest("enqueue fast path respects the total limit", async ({ redisContainer }) => { + const queue = createQueue(redisContainer, true); + try { + await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5); + await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1); + + const now = Date.now(); + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ runId: "r0", concurrencyKey: "ck-a", timestamp: now - 1000 }), + workerQueue: "main", + enableFastPath: true, + skipDequeueProcessing: true, + }); + + /** The fast path admits synchronously, so the group slot is taken immediately. */ + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(0); + + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ runId: "r1", concurrencyKey: "ck-b", timestamp: now - 999 }), + workerQueue: "main", + enableFastPath: true, + skipDequeueProcessing: true, + }); + + /** At the total limit the fast path must fall through to a normal enqueue. */ + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + } finally { + await queue.quit(); + } + }); + + redisTest("nacking releases the total slot", async ({ redisContainer }) => { + const queue = createQueue(redisContainer, true); + try { + await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5); + await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1); + + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ runId: "r0", concurrencyKey: "ck-a", timestamp: Date.now() - 1000 }), + workerQueue: "main", + }); + + const admitted = await waitFor( + async () => (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 1 + ); + expect(admitted).toBe(true); + + const dequeued = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + assertNonNullable(dequeued); + + await queue.nackMessage({ + orgId: authenticatedEnvDev.organization.id, + messageId: dequeued.messageId, + }); + + /** + * The nack returns the run to the queue and must release its group slot; the + * master consumer then re-admits it, taking the slot again. Assert the + * intermediate release by checking the run is never counted twice. + */ + const settled = await waitFor(async () => { + const total = await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task"); + return total <= 1; + }); + expect(settled).toBe(true); + } finally { + await queue.quit(); + } + }); +}); diff --git a/internal-packages/run-engine/src/run-queue/types.ts b/internal-packages/run-engine/src/run-queue/types.ts index 8a7d3c93ec5..10575d8b5d3 100644 --- a/internal-packages/run-engine/src/run-queue/types.ts +++ b/internal-packages/run-engine/src/run-queue/types.ts @@ -91,6 +91,10 @@ export interface RunQueueKeyProducer { queue: string, concurrencyKey?: string ): string; + queueGroupConcurrencyKey(env: RunQueueKeyProducerEnvironment, queue: string): string; + queueGroupConcurrencyKeyFromQueue(queue: string): string; + queueTotalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string; + queueTotalConcurrencyLimitKeyFromQueue(queue: string): string; //env oncurrency envCurrentConcurrencyKey(env: EnvDescriptor): string; diff --git a/packages/core/src/v3/schemas/schemas.ts b/packages/core/src/v3/schemas/schemas.ts index 505622d1ef0..ff669193c42 100644 --- a/packages/core/src/v3/schemas/schemas.ts +++ b/packages/core/src/v3/schemas/schemas.ts @@ -178,6 +178,12 @@ export const QueueManifest = z.object({ * * If this property is omitted, the task can potentially use up the full concurrency of an environment */ concurrencyLimit: z.number().int().min(0).max(100000).optional().nullable(), + /** An optional property that caps the total number of concurrent run executions across ALL + * `concurrencyKey` values of this queue. On a queue with a `concurrencyKey`, `concurrencyLimit` + * applies per key value; this is the ceiling for the whole queue. + * + * Only enforced for runs triggered with a `concurrencyKey`, and requires server-side support. */ + totalConcurrencyLimit: z.number().int().min(0).max(100000).optional().nullable(), }); export type QueueManifest = z.infer; diff --git a/packages/core/src/v3/types/queues.ts b/packages/core/src/v3/types/queues.ts index 9e87f136e2e..01d28fc072f 100644 --- a/packages/core/src/v3/types/queues.ts +++ b/packages/core/src/v3/types/queues.ts @@ -35,4 +35,25 @@ export type QueueOptions = { * * If this property is omitted, the task can potentially use up the full concurrency of an environment */ concurrencyLimit?: number; + /** An optional property that caps the total number of concurrent run executions across ALL + * `concurrencyKey` values of this queue. + * + * On a queue used with a `concurrencyKey`, `concurrencyLimit` applies to each key value + * independently — ten active keys with `concurrencyLimit: 5` can run 50 at once. Setting + * `totalConcurrencyLimit: 20` bounds the whole queue to 20 while each key still gets at + * most `concurrencyLimit`. + * + * @example + * + * ```ts + * const perUserQueue = queue({ + name: "per-user-queue", + concurrencyLimit: 1, + totalConcurrencyLimit: 10, + }); + * ``` + * + * Only enforced for runs triggered with a `concurrencyKey`, and requires server-side support. + */ + totalConcurrencyLimit?: number; }; diff --git a/packages/core/src/v3/types/tasks.ts b/packages/core/src/v3/types/tasks.ts index 224b7604db3..070cc96f3df 100644 --- a/packages/core/src/v3/types/tasks.ts +++ b/packages/core/src/v3/types/tasks.ts @@ -225,6 +225,7 @@ type CommonTaskOptions< queue?: { name?: string; concurrencyLimit?: number; + totalConcurrencyLimit?: number; }; /** Configure the spec of the [machine](https://trigger.dev/docs/machines) you want your task to run on. * diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts index 69b51f0ed3c..b962697d559 100644 --- a/packages/trigger-sdk/src/v3/shared.ts +++ b/packages/trigger-sdk/src/v3/shared.ts @@ -277,6 +277,7 @@ export function createTask< resourceCatalog.registerQueueMetadata({ name: queue.name, concurrencyLimit: queue.concurrencyLimit, + totalConcurrencyLimit: queue.totalConcurrencyLimit, }); } @@ -432,6 +433,7 @@ export function createSchemaTask< resourceCatalog.registerQueueMetadata({ name: queue.name, concurrencyLimit: queue.concurrencyLimit, + totalConcurrencyLimit: queue.totalConcurrencyLimit, }); } From 023c28232eef14931c6cef0ab218c6d1e6ea5a39 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Fri, 28 Aug 2026 18:53:48 +0100 Subject: [PATCH 2/5] fix(run-engine): address review on total concurrency limit Strengthen the nack test so it proves the group slot is released (a second key's run must be admitted after the nack), document the enable-time convergence window on the flag, and correct the totalConcurrencyOfQueue doc to describe the drain-on-disable behavior. --- .changeset/queue-total-concurrency-limit.md | 2 +- .../run-engine/src/run-queue/index.ts | 13 +++++-- .../run-queue/tests/totalConcurrency.test.ts | 34 +++++++++++++------ 3 files changed, 36 insertions(+), 13 deletions(-) diff --git a/.changeset/queue-total-concurrency-limit.md b/.changeset/queue-total-concurrency-limit.md index cc6aab6e2af..742f4925744 100644 --- a/.changeset/queue-total-concurrency-limit.md +++ b/.changeset/queue-total-concurrency-limit.md @@ -15,4 +15,4 @@ export const perUserQueue = queue({ }); ``` -Enforcement happens server side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled total concurrency limits accept the option but do not enforce it yet. +Enforcement happens server-side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled total concurrency limits accept the option but do not enforce it yet. diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 2b8c11fe62d..7ce0e754877 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -185,6 +185,13 @@ export type RunQueueOptions = { * limit at admit time. Default false: admit paths are byte-identical to before, and * only the release-side SREM mirror runs (a no-op on an absent set), so the flag can * be flipped on a fleet that has fully rolled onto this build without draining queues. + * + * Runs already in flight when the flag turns on are not in the set, so a queue can + * transiently exceed its total limit by the number of those runs. The excess is + * one-time and self-corrects as each pre-flag run completes (its release mirror + * no-ops), after which the limit is enforced exactly. Every pod must run this build + * before enabling: an older pod releases without the mirror and would leak members + * the gate then counts forever. */ totalConcurrencyEnabled?: boolean; workerOptions?: { @@ -495,8 +502,10 @@ export class RunQueue { /** * Total in-flight runs across all concurrency-key variants of a queue (the - * groupConcurrency SET cardinality). Only maintained while totalConcurrencyEnabled - * is on; returns 0 otherwise. + * groupConcurrency SET cardinality). Admits only populate the set while + * totalConcurrencyEnabled is on. After the flag is turned off the set drains + * to zero through the release-side mirrors, so a nonzero read reflects real + * runs admitted while it was on, never stale state. */ public async totalConcurrencyOfQueue(env: MinimalAuthenticatedEnvironment, queue: string) { return this.redis.scard(this.keys.queueGroupConcurrencyKey(env, queue)); 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 41c9ce73228..4f4b358cf09 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 @@ -259,24 +259,38 @@ describe("RunQueue total concurrency limit", () => { ); expect(admitted).toBe(true); + /** A second run on another key waits behind the total limit of 1. */ + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ runId: "r1", concurrencyKey: "ck-b", timestamp: Date.now() - 500 }), + workerQueue: "main", + }); + + await setTimeout(2000); + expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + const dequeued = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); assertNonNullable(dequeued); + expect(dequeued.messageId).toBe("r0"); + /** + * Nack r0 with a far-future retryAt so it cannot immediately reclaim the + * slot. If the nack released r0's group slot, r1 is the only eligible run + * and must be admitted; if the slot leaked, the queue stays blocked and r1 + * never surfaces. + */ await queue.nackMessage({ orgId: authenticatedEnvDev.organization.id, - messageId: dequeued.messageId, + messageId: "r0", + retryAt: Date.now() + 120_000, }); - /** - * The nack returns the run to the queue and must release its group slot; the - * master consumer then re-admits it, taking the slot again. Assert the - * intermediate release by checking the run is never counted twice. - */ - const settled = await waitFor(async () => { - const total = await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task"); - return total <= 1; + const r1Admitted = await waitFor(async () => { + const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + return next?.messageId === "r1"; }); - expect(settled).toBe(true); + expect(r1Admitted).toBe(true); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); } finally { await queue.quit(); } From 588ac67d7d4f2c81265c57c91663dd0fa67d994b Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Fri, 28 Aug 2026 19:10:48 +0100 Subject: [PATCH 3/5] feat(run-engine): self-heal leaked total-concurrency members at the gate A release path that misses the group-set mirror (an instance on an older build during rollout, or a future release script) would otherwise leave a member the gate counts forever. Every terminal release deletes the run's message key, so when a queue sits at its total the dequeue gate prunes members whose message key no longer exists, throttled to one pass per interval per queue. Members of re-queued runs keep their message key and clear through the mirrored ack when the run completes. --- .../run-engine/src/run-queue/index.ts | 32 ++++++++- .../run-queue/tests/totalConcurrency.test.ts | 65 +++++++++++++++++++ 2 files changed, 94 insertions(+), 3 deletions(-) diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 7ce0e754877..4dcc8860384 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -189,9 +189,14 @@ export type RunQueueOptions = { * Runs already in flight when the flag turns on are not in the set, so a queue can * transiently exceed its total limit by the number of those runs. The excess is * one-time and self-corrects as each pre-flag run completes (its release mirror - * no-ops), after which the limit is enforced exactly. Every pod must run this build - * before enabling: an older pod releases without the mirror and would leak members - * the gate then counts forever. + * no-ops), after which the limit is enforced exactly. + * + * A release path that misses the group mirror (an instance on an older build during + * rollout) leaves the member behind, briefly under-admitting. The dequeue gate + * reconciles: when a queue sits at its total, members whose message key no longer + * exists are pruned, so such leaks clear within seconds instead of blocking the + * queue. Enabling only after every instance runs this build avoids the noise but is + * no longer load-bearing for correctness. */ totalConcurrencyEnabled?: boolean; workerOptions?: { @@ -4693,6 +4698,27 @@ if totalConcurrencyEnabled then if rawTotalLimit then local totalConcurrencyLimit = math.min(tonumber(rawTotalLimit), envConcurrencyLimit) local groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') + + -- Self-heal before holding the queue at its limit. A terminal release path + -- that misses the group mirror (an older build, or a future script) leaves + -- a member behind, but every terminal path deletes the run's message key, + -- 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. + 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 + if redis.call('EXISTS', messageKeyPrefix .. groupMemberId) == 0 then + redis.call('SREM', groupConcurrencyKey, groupMemberId) + end + end + groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') + end + end + actualMaxCount = math.min(actualMaxCount, totalConcurrencyLimit - groupCurrentConcurrency) end 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 4f4b358cf09..cc52b1acad9 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 @@ -295,4 +295,69 @@ describe("RunQueue total concurrency limit", () => { await queue.quit(); } }); + + redisTest( + "reconciles a leaked group member instead of blocking the queue", + async ({ redisContainer }) => { + const queue = createQueue(redisContainer, true); + try { + const keys = testOptions.keys; + await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5); + await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1); + + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ + runId: "r0", + concurrencyKey: "ck-a", + timestamp: Date.now() - 1000, + }), + workerQueue: "main", + }); + + const admitted = await waitFor( + async () => + (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 1 + ); + expect(admitted).toBe(true); + + const dequeued = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + assertNonNullable(dequeued); + expect(dequeued.messageId).toBe("r0"); + + /** + * Simulate a terminal release from a build without the group mirror: the + * message key is deleted and the per-key and env sets are cleared, but the + * group member is left behind. + */ + await queue.redis.del(keys.messageKey(authenticatedEnvDev.organization.id, "r0")); + await queue.redis.srem( + keys.queueCurrentConcurrencyKey(authenticatedEnvDev, "task/my-task", "ck-a"), + "r0" + ); + await queue.redis.srem(keys.envCurrentConcurrencyKey(authenticatedEnvDev), "r0"); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + + /** The next run must still be admitted: the gate prunes the dead member. */ + await queue.enqueueMessage({ + env: authenticatedEnvDev, + message: makeMessage({ + runId: "r1", + concurrencyKey: "ck-b", + timestamp: Date.now() - 500, + }), + workerQueue: "main", + }); + + const r1Admitted = await waitFor(async () => { + const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + return next?.messageId === "r1"; + }); + expect(r1Admitted).toBe(true); + expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1); + } finally { + await queue.quit(); + } + } + ); }); From 3efb50ba0c5e00d681f41130a22f93e90370c2f8 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Fri, 28 Aug 2026 19:14:53 +0100 Subject: [PATCH 4/5] fix(run-engine,core): non-blocking test polls and document the zero total limit The waitFor polls dequeued with the default 10s blocking pop, which could blow past the helper deadline on a slow runner; poll non-blocking instead. Document that totalConcurrencyLimit: 0 holds every keyed run, matching concurrencyLimit's zero semantics. --- .../src/run-queue/tests/totalConcurrency.test.ts | 8 ++++++-- packages/core/src/v3/types/queues.ts | 3 +++ 2 files changed, 9 insertions(+), 2 deletions(-) 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 cc52b1acad9..7ad89c8c529 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 @@ -286,7 +286,9 @@ describe("RunQueue total concurrency limit", () => { }); const r1Admitted = await waitFor(async () => { - const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main", { + blockingPop: false, + }); return next?.messageId === "r1"; }); expect(r1Admitted).toBe(true); @@ -350,7 +352,9 @@ describe("RunQueue total concurrency limit", () => { }); const r1Admitted = await waitFor(async () => { - const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main"); + const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main", { + blockingPop: false, + }); return next?.messageId === "r1"; }); expect(r1Admitted).toBe(true); diff --git a/packages/core/src/v3/types/queues.ts b/packages/core/src/v3/types/queues.ts index 01d28fc072f..1fba786c868 100644 --- a/packages/core/src/v3/types/queues.ts +++ b/packages/core/src/v3/types/queues.ts @@ -54,6 +54,9 @@ export type QueueOptions = { * ``` * * Only enforced for runs triggered with a `concurrencyKey`, and requires server-side support. + * + * Omit for no total cap. Like `concurrencyLimit`, a value of `0` holds every keyed run in + * the queue rather than removing the cap. */ totalConcurrencyLimit?: number; }; From 6290352d0e36c0d3728b86f6a17d61a04888cf6f Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Fri, 28 Aug 2026 19:24:28 +0100 Subject: [PATCH 5/5] perf(run-engine): bound group-set reconciliation to one SSCAN batch per pass Reconciling a saturated queue's group set with SMEMBERS plus one EXISTS per member runs the whole traversal inside a single Lua call, which blocks Redis for the duration on a large set. Scan one bounded batch per pass instead, persisting the SSCAN cursor between passes so successive intervals cover the whole set. Covered by a test that drains a 1,200-member leaked backlog. --- .../run-engine/src/run-queue/index.ts | 11 +++-- .../run-queue/tests/totalConcurrency.test.ts | 45 +++++++++++++++++++ 2 files changed, 53 insertions(+), 3 deletions(-) diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 4dcc8860384..d3575ab273f 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 7ad89c8c529..cd6d299f151 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(); + } + } + ); });