diff --git a/.changeset/queue-concurrency-overrides.md b/.changeset/queue-concurrency-overrides.md new file mode 100644 index 00000000000..22f79c5e72c --- /dev/null +++ b/.changeset/queue-concurrency-overrides.md @@ -0,0 +1,15 @@ +--- +"@trigger.dev/sdk": patch +"@trigger.dev/core": patch +--- + +Adjust a queue's combined concurrency limit at runtime. `queues.overrideCombinedConcurrencyLimit` raises or lowers the cap on concurrent runs across all of a queue's `concurrencyKey` values, and `queues.resetCombinedConcurrencyLimit` reverts to the declared configuration. + +```ts +import { queues } from "@trigger.dev/sdk"; + +await queues.overrideCombinedConcurrencyLimit("my-queue", 100); +await queues.resetCombinedConcurrencyLimit("my-queue"); +``` + +Overrides survive deploys. Enforcement happens server-side on servers with combined concurrency limits enabled. diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts new file mode 100644 index 00000000000..c643b77965a --- /dev/null +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts @@ -0,0 +1,98 @@ +import { json } from "@remix-run/server-runtime"; +import { type RetrieveQueueParam, RetrieveQueueType } from "@trigger.dev/core/v3"; +import { z } from "zod"; +import { toQueueItem } from "~/presenters/v3/QueueRetrievePresenter.server"; +import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server"; +import { concurrencySystem } from "~/v3/services/concurrencySystemInstance.server"; + +const BodySchema = z.object({ + type: RetrieveQueueType.default("id"), + concurrencyLimit: z.number().int().min(0).max(100000), +}); + +const route = createActionApiRoute( + { + body: BodySchema, + params: z.object({ + queueParam: z.string().transform((val) => val.replace(/%2F/g, "/")), + }), + authorization: { + action: "write", + resource: () => ({ type: "queues" }), + }, + }, + async ({ params, body, authentication }) => { + const input: RetrieveQueueParam = + body.type === "id" + ? params.queueParam + : { + type: body.type, + name: decodeURIComponent(params.queueParam).replace(/%2F/g, "/"), + }; + + return concurrencySystem.queues + .overrideTotalConcurrencyLimit(authentication.environment, input, body.concurrencyLimit) + .match( + (queue) => { + return json( + toQueueItem({ + friendlyId: queue.friendlyId, + name: queue.name, + type: queue.type, + running: queue.running, + queued: queue.queued, + concurrencyLimit: queue.concurrencyLimit, + concurrencyLimitBase: queue.concurrencyLimitBase, + concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, + concurrencyLimitOverriddenBy: null, + paused: queue.paused, + }), + { status: 200 } + ); + }, + (error) => { + switch (error.type) { + case "queue_not_found": { + return json({ error: "Queue not found" }, { status: 404 }); + } + case "invalid_override": + case "concurrency_limit_exceeds_maximum": { + return json({ error: error.message }, { status: 400 }); + } + case "queue_update_failed": { + return json( + { error: "Failed to update queue total concurrency limit" }, + { status: 500 } + ); + } + case "sync_queue_concurrency_to_engine_failed": { + return json({ error: "Failed to sync the total concurrency limit" }, { status: 500 }); + } + case "get_queue_stats_failed": { + return json({ error: "Failed to read queue stats" }, { status: 500 }); + } + case "other": { + return json( + { error: "Failed to update queue total concurrency limit" }, + { + status: 500, + } + ); + } + default: { + return json( + { error: "Failed to update queue total concurrency limit" }, + { + status: 500, + } + ); + } + } + } + ); + } +); + +export const action = route.action; +/** The builder's loader answers non-POST methods with a 405. */ +export const loader = route.loader; diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts new file mode 100644 index 00000000000..b2841f1efe6 --- /dev/null +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts @@ -0,0 +1,99 @@ +import { json } from "@remix-run/server-runtime"; +import { type RetrieveQueueParam, RetrieveQueueType } from "@trigger.dev/core/v3"; +import { z } from "zod"; +import { toQueueItem } from "~/presenters/v3/QueueRetrievePresenter.server"; +import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server"; +import { concurrencySystem } from "~/v3/services/concurrencySystemInstance.server"; + +const BodySchema = z.object({ + type: RetrieveQueueType.default("id"), +}); + +const route = createActionApiRoute( + { + body: BodySchema, + params: z.object({ + queueParam: z.string().transform((val) => val.replace(/%2F/g, "/")), + }), + authorization: { + action: "write", + resource: () => ({ type: "queues" }), + }, + }, + async ({ params, body, authentication }) => { + const input: RetrieveQueueParam = + body.type === "id" + ? params.queueParam + : { + type: body.type, + name: decodeURIComponent(params.queueParam).replace(/%2F/g, "/"), + }; + + return concurrencySystem.queues + .resetTotalConcurrencyLimit(authentication.environment, input) + .match( + (queue) => { + return json( + toQueueItem({ + friendlyId: queue.friendlyId, + name: queue.name, + type: queue.type, + running: queue.running, + queued: queue.queued, + concurrencyLimit: queue.concurrencyLimit, + concurrencyLimitBase: queue.concurrencyLimitBase, + concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, + concurrencyLimitOverriddenBy: null, + paused: queue.paused, + }), + { status: 200 } + ); + }, + (error) => { + switch (error.type) { + case "queue_not_found": { + return json({ error: "Queue not found" }, { status: 404 }); + } + case "queue_not_overridden": { + return json( + { error: "The queue total concurrency limit is not overridden" }, + { status: 400 } + ); + } + case "queue_update_failed": { + return json( + { error: "Failed to reset the queue total concurrency limit" }, + { status: 500 } + ); + } + case "sync_queue_concurrency_to_engine_failed": { + return json({ error: "Failed to sync the total concurrency limit" }, { status: 500 }); + } + case "get_queue_stats_failed": { + return json({ error: "Failed to read queue stats" }, { status: 500 }); + } + case "other": { + return json( + { error: "Failed to reset the queue total concurrency limit" }, + { + status: 500, + } + ); + } + default: { + return json( + { error: "Failed to reset the queue total concurrency limit" }, + { + status: 500, + } + ); + } + } + } + ); + } +); + +export const action = route.action; +/** The builder's loader answers non-POST methods with a 405. */ +export const loader = route.loader; diff --git a/apps/webapp/app/v3/services/concurrencySystem.server.ts b/apps/webapp/app/v3/services/concurrencySystem.server.ts index 51c51674234..09cc5f9ce11 100644 --- a/apps/webapp/app/v3/services/concurrencySystem.server.ts +++ b/apps/webapp/app/v3/services/concurrencySystem.server.ts @@ -3,7 +3,12 @@ import { errAsync, fromPromise, okAsync } from "neverthrow"; import type { PrismaClientOrTransaction } from "~/db.server"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; -import { removeQueueConcurrencyLimits, updateQueueConcurrencyLimits } from "../runQueue.server"; +import { + removeQueueConcurrencyLimits, + removeQueueTotalConcurrencyLimits, + updateQueueConcurrencyLimits, + updateQueueTotalConcurrencyLimits, +} from "../runQueue.server"; import { engine } from "../runEngine.server"; export type ConcurrencySystemOptions = { @@ -77,6 +82,32 @@ export class ConcurrencySystem { .andThen((queue) => syncQueueConcurrencyToEngine(environment, queue)) .andThen((queue) => getQueueStats(environment, queue)); }, + overrideTotalConcurrencyLimit: ( + environment: AuthenticatedEnvironment, + queue: QueueInput, + totalConcurrencyLimit: number, + overriddenBy?: User + ) => { + return findQueueFromInput(this.db, environment, queue) + .andThen((queue) => + overrideQueueTotalConcurrencyLimit( + this.db, + environment, + queue, + totalConcurrencyLimit, + overriddenBy + ) + ) + .andThen((queue) => syncQueueTotalConcurrencyToEngine(environment, queue)) + .andThen((queue) => getQueueStats(environment, queue)); + }, + resetTotalConcurrencyLimit: (environment: AuthenticatedEnvironment, queue: QueueInput) => { + return findQueueFromInput(this.db, environment, queue) + .andThen((queue) => syncQueueTotalConcurrencyResetToEngine(environment, queue)) + .andThen((queue) => resetQueueTotalConcurrencyLimit(this.db, queue)) + .andThen((queue) => syncQueueTotalConcurrencyToEngine(environment, queue)) + .andThen((queue) => getQueueStats(environment, queue)); + }, /** * Recalculates the materialized limit of every percent-based override in the environment * against its CURRENT maximumConcurrencyLimit and syncs changed queues to the run engine. @@ -316,6 +347,125 @@ function syncQueueConcurrencyToEngine(environment: AuthenticatedEnvironment, que } } +function overrideQueueTotalConcurrencyLimit( + db: PrismaClientOrTransaction, + environment: AuthenticatedEnvironment, + queue: TaskQueue, + totalConcurrencyLimit: number, + overriddenBy?: User +) { + const maximum = environment.maximumConcurrencyLimit; + + if (!Number.isFinite(totalConcurrencyLimit) || totalConcurrencyLimit < 0) { + return errAsync({ + type: "invalid_override" as const, + message: "Combined concurrency limit must be a non-negative number", + }); + } + + if (totalConcurrencyLimit > maximum) { + return errAsync({ + type: "concurrency_limit_exceeds_maximum" as const, + message: `Combined concurrency limit (${totalConcurrencyLimit}) cannot exceed the environment limit (${maximum})`, + }); + } + + const totalConcurrencyLimitBase = queue.totalConcurrencyLimitOverriddenAt + ? queue.totalConcurrencyLimitBase + : queue.totalConcurrencyLimit; + + return fromPromise( + db.taskQueue.update({ + where: { id: queue.id }, + data: { + totalConcurrencyLimit, + totalConcurrencyLimitBase: totalConcurrencyLimitBase ?? null, + totalConcurrencyLimitOverriddenAt: new Date(), + totalConcurrencyLimitOverriddenBy: overriddenBy?.id ?? null, + }, + }), + (error) => ({ + type: "queue_update_failed" as const, + cause: error, + }) + ); +} + +/** + * Enforce first, then persist: syncs the engine to the declared base BEFORE clearing + * the override marker, so an engine failure leaves the marker set and a retry + * converges instead of being rejected while the overridden limit stays enforced. + */ +function syncQueueTotalConcurrencyResetToEngine( + environment: AuthenticatedEnvironment, + queue: TaskQueue +) { + if (queue.totalConcurrencyLimitOverriddenAt === null) { + return errAsync({ type: "queue_not_overridden" as const }); + } + + if (typeof queue.totalConcurrencyLimitBase === "number") { + return fromPromise( + updateQueueTotalConcurrencyLimits(environment, queue.name, queue.totalConcurrencyLimitBase), + (error) => ({ + type: "sync_queue_concurrency_to_engine_failed" as const, + cause: error, + }) + ).andThen(() => okAsync(queue)); + } + + return fromPromise(removeQueueTotalConcurrencyLimits(environment, queue.name), (error) => ({ + type: "sync_queue_concurrency_to_engine_failed" as const, + cause: error, + })).andThen(() => okAsync(queue)); +} + +function resetQueueTotalConcurrencyLimit(db: PrismaClientOrTransaction, queue: TaskQueue) { + if (queue.totalConcurrencyLimitOverriddenAt === null) { + return errAsync({ type: "queue_not_overridden" as const }); + } + + return fromPromise( + db.taskQueue.update({ + where: { id: queue.id }, + data: { + totalConcurrencyLimit: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitBase: null, + totalConcurrencyLimitOverriddenAt: null, + totalConcurrencyLimitOverriddenBy: null, + }, + }), + (error) => ({ + type: "queue_update_failed" as const, + cause: error, + }) + ); +} + +/** + * The total limit key is separate from the per-queue limit key that pause zeroes, + * so it syncs regardless of the paused state. + */ +function syncQueueTotalConcurrencyToEngine( + environment: AuthenticatedEnvironment, + queue: TaskQueue +) { + if (typeof queue.totalConcurrencyLimit === "number") { + return fromPromise( + updateQueueTotalConcurrencyLimits(environment, queue.name, queue.totalConcurrencyLimit), + (error) => ({ + type: "sync_queue_concurrency_to_engine_failed" as const, + cause: error, + }) + ).andThen(() => okAsync(queue)); + } + + return fromPromise(removeQueueTotalConcurrencyLimits(environment, queue.name), (error) => ({ + type: "sync_queue_concurrency_to_engine_failed" as const, + cause: error, + })).andThen(() => okAsync(queue)); +} + function getQueueStats(environment: AuthenticatedEnvironment, queue: TaskQueue) { return fromPromise( Promise.all([ diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index 386f8f038d6..147248bd376 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -660,6 +660,7 @@ async function upsertWorkerQueueRecord( }); } else { const hasOverride = taskQueue.concurrencyLimitOverriddenAt !== null; + const hasTotalOverride = taskQueue.totalConcurrencyLimitOverriddenAt !== null; taskQueue = await prisma.taskQueue.update({ where: { @@ -672,7 +673,8 @@ async function upsertWorkerQueueRecord( // If overridden, keep current limit and update base; otherwise update limit normally concurrencyLimit: hasOverride ? undefined : concurrencyLimit, concurrencyLimitBase: hasOverride ? concurrencyLimit : undefined, - totalConcurrencyLimit, + totalConcurrencyLimit: hasTotalOverride ? undefined : totalConcurrencyLimit, + totalConcurrencyLimitBase: hasTotalOverride ? totalConcurrencyLimit : undefined, }, }); } diff --git a/internal-packages/database/prisma/migrations/20260829150000_add_concurrency_overrides/migration.sql b/internal-packages/database/prisma/migrations/20260829150000_add_concurrency_overrides/migration.sql new file mode 100644 index 00000000000..4b8b6ec6077 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260829150000_add_concurrency_overrides/migration.sql @@ -0,0 +1,4 @@ +-- AlterTable +ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimitOverriddenAt" TIMESTAMP(3); +ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimitOverriddenBy" TEXT; +ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimitBase" INTEGER; diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index c88ba5b5887..406caec57e5 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -1983,7 +1983,13 @@ model TaskQueue { 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? + totalConcurrencyLimit Int? + /// When the total concurrency limit was overridden + totalConcurrencyLimitOverriddenAt DateTime? + /// Who overrode the total concurrency limit (null when overridden via the API) + totalConcurrencyLimitOverriddenBy String? + /// If totalConcurrencyLimit is overridden, the declared value it reverts to on reset + totalConcurrencyLimitBase Int? rateLimit Json? paused Boolean @default(false) @@ -1995,9 +2001,11 @@ model TaskQueue { tasks BackgroundWorkerTask[] workers BackgroundWorker[] + @@unique([runtimeEnvironmentId, name]) } + enum TaskQueueType { VIRTUAL NAMED diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 6ab8d90aaec..7917f3a48be 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -114,12 +114,18 @@ local function __gateReconcile(setKey, msgKeyPrefix, reconcileKeyPrefix) end end -local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msgKeyPrefix) +local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msgKeyPrefix, ckOverridesEnabled) if not msg.gates then return true end for _, gate in ipairs(msg.gates) do local base, variant, gateKey = __gateKeys(gatesKeyPrefix, msg, gate) local occupancy = tonumber(redis.call('SCARD', variant .. ':currentConcurrency') or '0') local perKeyLimit = math.min(tonumber(redis.call('GET', base .. ':concurrency') or '1000000'), envLimit) + if ckOverridesEnabled and gateKey and gateKey ~= '' then + local gateOverride = redis.call('HGET', base .. ':ckLimits', string.sub(variant, #gatesKeyPrefix + 1)) + if gateOverride then + perKeyLimit = math.min(tonumber(gateOverride), envLimit) + end + end if occupancy >= perKeyLimit and redis.call('SISMEMBER', variant .. ':currentConcurrency', messageId) == 0 then __gateReconcile(variant .. ':currentConcurrency', msgKeyPrefix, gatesKeyPrefix) return false @@ -257,6 +263,13 @@ export interface RunQueueMetricsEmitter { emitGauge(shardKey: string, fields: Record): void; } +export class RunQueueConcurrencyKeyLimitExceededError extends Error { + constructor(message: string) { + super(message); + this.name = "RunQueueConcurrencyKeyLimitExceededError"; + } +} + export type RunQueueOptions = { name: string; tracer: Tracer; @@ -305,6 +318,10 @@ export type RunQueueOptions = { * that dead-lettered or suspended through a mirror-less path. Enabling only after * every instance runs this build avoids the noise but is no longer load-bearing * for correctness. + * + * Per-concurrency-key limit overrides are part of the same concurrency-limits + * feature and are deliberately enforced behind this flag too: writes are always + * accepted and durable, and enforcement of both arrives together. */ totalConcurrencyEnabled?: boolean; /** @@ -316,6 +333,8 @@ export type RunQueueOptions = { * the total cap covering releases from builds without the mirror. */ gatesEnabled?: boolean; + /** Cap on per-concurrency-key limit overrides stored per queue. Default 1000. */ + maxConcurrencyKeyOverridesPerQueue?: number; workerOptions?: { pollIntervalMs?: number; immediatePollIntervalMs?: number; @@ -429,6 +448,7 @@ export class RunQueue { private queueSelectionStrategy: RunQueueSelectionStrategy; private shardCount: number; private counterTtlSeconds: number; + private maxConcurrencyKeyOverridesPerQueue: number; private abortController: AbortController; private worker: Worker; private workerQueueResolver: WorkerQueueResolver; @@ -439,6 +459,7 @@ export class RunQueue { constructor(public readonly options: RunQueueOptions) { this.shardCount = options.shardCount ?? 2; this.counterTtlSeconds = options.counterTtlSeconds ?? 86400; + this.maxConcurrencyKeyOverridesPerQueue = options.maxConcurrencyKeyOverridesPerQueue ?? 1000; this.retryOptions = options.retryOptions ?? defaultRetrySettings; this.redis = createRedisClient(options.redis, { onError: (error) => { @@ -633,6 +654,62 @@ export class RunQueue { return this.redis.scard(this.keys.queueGroupConcurrencyKey(env, queue)); } + /** + * Sets a per-concurrency-key limit override for a queue. The stored value is the + * raw requested limit; admit paths clamp to the environment limit at read time. + * Throws RunQueueConcurrencyKeyLimitExceededError when a NEW key would push the + * queue past maxConcurrencyKeyOverridesPerQueue (updates to existing keys always + * succeed). + */ + public async updateQueueConcurrencyKeyLimit( + env: MinimalAuthenticatedEnvironment, + queue: string, + concurrencyKey: string, + limit: number + ) { + const result = await this.redis.setQueueConcurrencyKeyLimit( + this.keys.queueCkLimitsKey(env, queue), + this.keys.queueKey(env, queue, concurrencyKey), + String(limit), + String(this.maxConcurrencyKeyOverridesPerQueue) + ); + + if (result === 0) { + throw new RunQueueConcurrencyKeyLimitExceededError( + `Cannot add a concurrency key override to queue ${queue}: the queue already has ${this.maxConcurrencyKeyOverridesPerQueue} overrides` + ); + } + } + + public async removeQueueConcurrencyKeyLimit( + env: MinimalAuthenticatedEnvironment, + queue: string, + concurrencyKey: string + ) { + return this.redis.hdel( + this.keys.queueCkLimitsKey(env, queue), + this.keys.queueKey(env, queue, concurrencyKey) + ); + } + + /** Returns the raw per-concurrency-key limit overrides for a queue, keyed by concurrency key value. */ + public async getQueueConcurrencyKeyLimits( + env: MinimalAuthenticatedEnvironment, + queue: string + ): Promise> { + const raw = await this.redis.hgetall(this.keys.queueCkLimitsKey(env, queue)); + + const limits: Record = {}; + for (const [variantName, value] of Object.entries(raw)) { + const ckIndex = variantName.indexOf(":ck:"); + if (ckIndex === -1) { + continue; + } + limits[variantName.slice(ckIndex + 4)] = Number(value); + } + return limits; + } + public async updateEnvConcurrencyLimits(env: MinimalAuthenticatedEnvironment) { await this.#callUpdateEnvironmentConcurrencyLimits({ envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env), @@ -2366,6 +2443,7 @@ export class RunQueue { const totalConcurrencyLimitKey = this.keys.queueTotalConcurrencyLimitKeyFromQueue( message.queue ); + const ckLimitsKey = this.keys.queueCkLimitsKeyFromQueue(message.queue); const totalConcurrencyEnabledArg = this.options.totalConcurrencyEnabled ? "1" : "0"; if (ttlInfo) { @@ -2389,6 +2467,7 @@ export class RunQueue { baseQueueKey, groupConcurrencyKey, totalConcurrencyLimitKey, + ckLimitsKey, // args queueName, messageId, @@ -2428,6 +2507,7 @@ export class RunQueue { baseQueueKey, groupConcurrencyKey, totalConcurrencyLimitKey, + ckLimitsKey, // args queueName, messageId, @@ -2477,6 +2557,7 @@ export class RunQueue { enableFastPathArg, this.options.redis.keyPrefix ?? "", this.options.gatesEnabled ? "1" : "0", + this.options.totalConcurrencyEnabled ? "1" : "0", metricsGaugeArg ); } else { @@ -2506,6 +2587,7 @@ export class RunQueue { enableFastPathArg, this.options.redis.keyPrefix ?? "", this.options.gatesEnabled ? "1" : "0", + this.options.totalConcurrencyEnabled ? "1" : "0", metricsGaugeArg ); } @@ -2587,6 +2669,7 @@ export class RunQueue { this.options.redis.keyPrefix ?? "", String(maxCount), this.options.gatesEnabled ? "1" : "0", + this.options.totalConcurrencyEnabled ? "1" : "0", metricsGaugeArg ); @@ -2715,6 +2798,7 @@ export class RunQueue { runningCounterKey, this.keys.queueGroupConcurrencyKeyFromQueue(ckWildcardQueue), this.keys.queueTotalConcurrencyLimitKeyFromQueue(ckWildcardQueue), + this.keys.queueCkLimitsKeyFromQueue(ckWildcardQueue), //args ckWildcardQueue, String(Date.now()), @@ -3583,6 +3667,7 @@ local currentTime = ARGV[8] local enableFastPath = ARGV[9] local keyPrefix = ARGV[10] local gatesEnabled = ARGV[11] == '1' +local totalConcurrencyEnabled = ARGV[12] == '1' ${QUEUE_METRICS_GAUGE_PRELUDE} ${QUEUE_GATES_LUA_HELPERS} @@ -3610,7 +3695,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) end end @@ -3697,6 +3782,7 @@ local currentTime = ARGV[10] local enableFastPath = ARGV[11] local keyPrefix = ARGV[12] local gatesEnabled = ARGV[13] == '1' +local totalConcurrencyEnabled = ARGV[14] == '1' ${QUEUE_METRICS_GAUGE_PRELUDE} ${QUEUE_GATES_LUA_HELPERS} @@ -3724,7 +3810,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) end end @@ -4003,7 +4089,7 @@ return __qmret(0) // *Tracked variants of dequeueMessageFromKey and the ack/nack/dlq/release/clear // scripts. this.redis.defineCommand("enqueueMessageCkTracked", { - numberOfKeys: 17, + numberOfKeys: 18, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -4025,6 +4111,7 @@ local baseQueueKey = KEYS[15] -- Total-cap keys (KEYS 16-17) local groupConcurrencyKey = KEYS[16] local totalConcurrencyLimitKey = KEYS[17] +local ckLimitsKey = KEYS[18] local queueName = ARGV[1] local messageId = ARGV[2] @@ -4062,6 +4149,12 @@ if enableFastPath == '1' then tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envLimit ) + if totalConcurrencyEnabled then + local perKeyOverride = redis.call('HGET', ckLimitsKey, queueName) + if perKeyOverride then + queueLimit = math.min(tonumber(perKeyOverride), envLimit) + end + end if queueCurrent < queueLimit then -- Total-cap gate: a fast-path admit consumes a group slot, so it must @@ -4084,7 +4177,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) end end @@ -4173,7 +4266,7 @@ return __qmret(0) }); this.redis.defineCommand("enqueueMessageWithTtlCkTracked", { - numberOfKeys: 18, + numberOfKeys: 19, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -4196,6 +4289,7 @@ local baseQueueKey = KEYS[16] -- Total-cap keys (KEYS 17-18) local groupConcurrencyKey = KEYS[17] local totalConcurrencyLimitKey = KEYS[18] +local ckLimitsKey = KEYS[19] local queueName = ARGV[1] local messageId = ARGV[2] @@ -4235,6 +4329,12 @@ if enableFastPath == '1' then tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envLimit ) + if totalConcurrencyEnabled then + local perKeyOverride = redis.call('HGET', ckLimitsKey, queueName) + if perKeyOverride then + queueLimit = math.min(tonumber(perKeyOverride), envLimit) + end + end if queueCurrent < queueLimit then -- Total-cap gate: see enqueueMessageCkTracked. @@ -4255,7 +4355,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) end end @@ -4585,6 +4685,7 @@ local defaultEnvConcurrencyBurstFactor = ARGV[4] local keyPrefix = ARGV[5] local maxCount = tonumber(ARGV[6] or '1') local gatesEnabled = ARGV[7] == '1' +local totalConcurrencyEnabled = ARGV[8] == '1' ${QUEUE_METRICS_GAUGE_PRELUDE} ${QUEUE_GATES_LUA_HELPERS} ${QUEUE_METRICS_GAUGE_LUA} @@ -4657,7 +4758,7 @@ for i = 1, #messages, 2 do else local gatesAllow = true if gatesEnabled then - gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix) + gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix, totalConcurrencyEnabled) end if gatesAllow then @@ -4859,7 +4960,7 @@ return results // (normal dequeue, TTL-expired, or stale-orphan path — all of which were // counted at enqueue time). this.redis.defineCommand("dequeueMessagesFromCkQueueTracked", { - numberOfKeys: 13, + numberOfKeys: 14, lua: ` local ckIndexKey = KEYS[1] local queueConcurrencyLimitKey = KEYS[2] @@ -4874,6 +4975,7 @@ local lengthCounterKey = KEYS[10] local runningCounterKey = KEYS[11] local groupConcurrencyKey = KEYS[12] local totalConcurrencyLimitKey = KEYS[13] +local ckLimitsKey = KEYS[14] local ckWildcardName = ARGV[1] local currentTime = tonumber(ARGV[2]) @@ -4959,11 +5061,28 @@ for _, ckQueueName in ipairs(ckQueues) do end local fullQueueKey = keyPrefix .. ckQueueName + local blockedByGates = false local ckConcurrencyKey = fullQueueKey .. ':currentConcurrency' local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0') - if ckCurrentConcurrency < queueConcurrencyLimit then + local perKeyLimit = queueConcurrencyLimit + if totalConcurrencyEnabled then + local perKeyOverride = redis.call('HGET', ckLimitsKey, ckQueueName) + if perKeyOverride then + perKeyLimit = math.min(tonumber(perKeyOverride), envConcurrencyLimit) + end + end + + if ckCurrentConcurrency >= perKeyLimit then + -- Back a blocked variant off so it cannot pin the bounded candidate window + -- and starve later keys (acute with a zero per-key override, which never + -- self-clears). Acks and nacks rebalance the score back to the oldest + -- message, so the key is eligible again the moment capacity frees. + redis.call('ZADD', ckIndexKey, currentTime + 1000, ckQueueName) + end + + if ckCurrentConcurrency < perKeyLimit then local messages = redis.call('ZRANGEBYSCORE', fullQueueKey, '-inf', tostring(currentTime), 'WITHSCORES', 'LIMIT', 0, 1) if #messages >= 2 then @@ -4988,7 +5107,10 @@ for _, ckQueueName in ipairs(ckQueues) do else local gatesAllow = true if gatesEnabled then - gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix) + gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix, totalConcurrencyEnabled) + end + if not gatesAllow then + blockedByGates = true end local alreadyInGroup = false @@ -5032,11 +5154,15 @@ for _, ckQueueName in ipairs(ckQueues) do decrLengthCounter() end - local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES') - if #earliest == 0 then - redis.call('ZREM', ckIndexKey, ckQueueName) + if blockedByGates then + redis.call('ZADD', ckIndexKey, currentTime + 1000, ckQueueName) else - redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName) + local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES') + if #earliest == 0 then + redis.call('ZREM', ckIndexKey, ckQueueName) + else + redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName) + end end else local any = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES') @@ -5879,6 +6005,26 @@ __gatesRelease(keyPrefix, redis.call('GET', messageKey), messageId) `, }); + this.redis.defineCommand("setQueueConcurrencyKeyLimit", { + numberOfKeys: 1, + lua: ` +local ckLimitsKey = KEYS[1] + +local fieldName = ARGV[1] +local limit = ARGV[2] +local maxFields = tonumber(ARGV[3]) + +if redis.call('HEXISTS', ckLimitsKey, fieldName) == 0 then + if redis.call('HLEN', ckLimitsKey) >= maxFields then + return 0 + end +end + +redis.call('HSET', ckLimitsKey, fieldName, limit) +return 1 +`, + }); + this.redis.defineCommand("updateEnvironmentConcurrencyLimits", { numberOfKeys: 2, lua: ` @@ -6069,6 +6215,7 @@ declare module "@internal/redis" { enableFastPath: string, keyPrefix: string, gatesEnabled: string, + totalConcurrencyEnabled: string, metricsEnabled: string, callback?: Callback<[number, number[] | null]> ): Result<[number, number[] | null], Context>; @@ -6102,6 +6249,7 @@ declare module "@internal/redis" { enableFastPath: string, keyPrefix: string, gatesEnabled: string, + totalConcurrencyEnabled: string, metricsEnabled: string, callback?: Callback<[number, number[] | null]> ): Result<[number, number[] | null], Context>; @@ -6140,6 +6288,7 @@ declare module "@internal/redis" { keyPrefix: string, maxCount: string, gatesEnabled: string, + totalConcurrencyEnabled: string, metricsEnabled: string, callback?: Callback<[string[] | null, number[] | null]> ): Result<[string[] | null, number[] | null], Context>; @@ -6240,6 +6389,14 @@ declare module "@internal/redis" { callback?: Callback ): Result; + setQueueConcurrencyKeyLimit( + ckLimitsKey: string, + fieldName: string, + limit: string, + maxFields: string, + callback?: Callback + ): Result; + updateEnvironmentConcurrencyLimits( // keys envConcurrencyLimitKey: string, @@ -6426,6 +6583,7 @@ declare module "@internal/redis" { baseQueueKey: string, groupConcurrencyKey: string, totalConcurrencyLimitKey: string, + ckLimitsKey: string, queueName: string, messageId: string, messageData: string, @@ -6463,6 +6621,7 @@ declare module "@internal/redis" { baseQueueKey: string, groupConcurrencyKey: string, totalConcurrencyLimitKey: string, + ckLimitsKey: string, queueName: string, messageId: string, messageData: string, @@ -6497,6 +6656,7 @@ declare module "@internal/redis" { runningCounterKey: string, groupConcurrencyKey: string, totalConcurrencyLimitKey: string, + ckLimitsKey: string, ckWildcardName: string, currentTime: string, defaultEnvConcurrencyLimit: string, diff --git a/internal-packages/run-engine/src/run-queue/keyProducer.ts b/internal-packages/run-engine/src/run-queue/keyProducer.ts index 98028f5af7b..120e04f8c38 100644 --- a/internal-packages/run-engine/src/run-queue/keyProducer.ts +++ b/internal-packages/run-engine/src/run-queue/keyProducer.ts @@ -366,6 +366,14 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer { return `${this.baseQueueKeyFromQueue(queue)}:${constants.TOTAL_CONCURRENCY_LIMIT_PART}`; } + queueCkLimitsKey(env: RunQueueKeyProducerEnvironment, queue: string): string { + return `${this.queueKey(env, queue)}:ckLimits`; + } + + queueCkLimitsKeyFromQueue(queue: string): string { + return `${this.baseQueueKeyFromQueue(queue)}:ckLimits`; + } + isCkWildcard(queue: string): boolean { return queue.endsWith(":ck:*"); } diff --git a/internal-packages/run-engine/src/run-queue/types.ts b/internal-packages/run-engine/src/run-queue/types.ts index 2cbfe40c775..2961b642314 100644 --- a/internal-packages/run-engine/src/run-queue/types.ts +++ b/internal-packages/run-engine/src/run-queue/types.ts @@ -111,6 +111,9 @@ export interface RunQueueKeyProducer { queueTotalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string; queueTotalConcurrencyLimitKeyFromQueue(queue: string): string; + queueCkLimitsKey(env: RunQueueKeyProducerEnvironment, queue: string): string; + queueCkLimitsKeyFromQueue(queue: string): string; + //env oncurrency envCurrentConcurrencyKey(env: EnvDescriptor): string; envCurrentConcurrencyKey(env: RunQueueKeyProducerEnvironment): string; diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index 585aeaed6c2..4acb1c40089 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/packages/core/src/v3/apiClient/index.ts @@ -1704,6 +1704,51 @@ export class ApiClient { ); } + overrideQueueCombinedConcurrencyLimit( + queue: RetrieveQueueParam, + concurrencyLimit: number, + requestOptions?: ZodFetchOptions + ) { + const type = typeof queue === "string" ? "id" : queue.type; + const value = typeof queue === "string" ? queue : queue.name; + + const encodedValue = encodeURIComponent(value.replace(/\//g, "%2F")); + + return zodfetch( + QueueItem, + `${this.baseUrl}/api/v1/queues/${encodedValue}/concurrency/combined/override`, + { + method: "POST", + headers: this.#getHeaders(false), + body: JSON.stringify({ + type, + concurrencyLimit, + }), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + + resetQueueCombinedConcurrencyLimit(queue: RetrieveQueueParam, requestOptions?: ZodFetchOptions) { + const type = typeof queue === "string" ? "id" : queue.type; + const value = typeof queue === "string" ? queue : queue.name; + + const encodedValue = encodeURIComponent(value.replace(/\//g, "%2F")); + + return zodfetch( + QueueItem, + `${this.baseUrl}/api/v1/queues/${encodedValue}/concurrency/combined/reset`, + { + method: "POST", + headers: this.#getHeaders(false), + body: JSON.stringify({ + type, + }), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + subscribeToRun( runId: string, options?: { diff --git a/packages/trigger-sdk/src/v3/queues.ts b/packages/trigger-sdk/src/v3/queues.ts index 7e76c5f940b..1292cde1a3c 100644 --- a/packages/trigger-sdk/src/v3/queues.ts +++ b/packages/trigger-sdk/src/v3/queues.ts @@ -172,6 +172,81 @@ export function overrideConcurrencyLimit( return apiClient.overrideQueueConcurrencyLimit(queue, concurrencyLimit, $requestOptions); } +/** + * Overrides the combined concurrency limit of a queue: the cap on concurrent runs across + * all of its `concurrencyKey` values. + * + * @param queue - The ID of the queue, or the type and name + * @param concurrencyLimit - The combined concurrency limit to apply + * @returns The updated queue state + */ +export function overrideCombinedConcurrencyLimit( + queue: RetrieveQueueParam, + concurrencyLimit: number, + requestOptions?: ApiRequestOptions +): ApiPromise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "queues.overrideCombinedConcurrencyLimit()", + icon: "queue", + attributes: { + ...flattenAttributes({ queue }), + ...accessoryAttributes({ + items: [ + { + text: typeof queue === "string" ? queue : queue.name, + variant: "normal", + }, + ], + style: "codepath", + }), + }, + }, + requestOptions + ); + + return apiClient.overrideQueueCombinedConcurrencyLimit(queue, concurrencyLimit, $requestOptions); +} + +/** + * Resets the combined concurrency limit of a queue back to its declared value. + * + * @param queue - The ID of the queue, or the type and name + * @returns The updated queue state + */ +export function resetCombinedConcurrencyLimit( + queue: RetrieveQueueParam, + requestOptions?: ApiRequestOptions +): ApiPromise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "queues.resetCombinedConcurrencyLimit()", + icon: "queue", + attributes: { + ...flattenAttributes({ queue }), + ...accessoryAttributes({ + items: [ + { + text: typeof queue === "string" ? queue : queue.name, + variant: "normal", + }, + ], + style: "codepath", + }), + }, + }, + requestOptions + ); + + return apiClient.resetQueueCombinedConcurrencyLimit(queue, $requestOptions); +} + /** * Resets the concurrency limit of a queue to the base value. *