feat(run-engine,sdk,webapp): queue total concurrency limit across keys

.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.

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("1"),
     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),

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:",

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,

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,

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;

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)

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,

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<