feat(sdk,core,webapp,run-engine): task concurrency option and shareable named limits

.changeset/queue-combined-concurrency-stats.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Queue retrieve and list API responses now report combined concurrency usage. When a queue has a `combinedConcurrencyLimit`, `concurrency.combined` includes the effective cap, the declared base, any active override, and how many runs are in flight across all concurrency keys.

.changeset/task-concurrency.md

@@ -0,0 +1,23 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+"@trigger.dev/react-hooks": patch
+---
+
+Control a task's concurrency with the new `concurrency` option, and share limits across tasks with named concurrency limits. An inline shape caps the task itself; `concurrencyLimit()` declares a limit any task can hold (up to two named limits per task), and a trigger call can switch a run's named limits with its own `concurrency` option.
+
+```ts
+import { concurrencyLimit, task } from "@trigger.dev/sdk";
+
+export const openaiLimit = concurrencyLimit({ name: "openai", total: 25 });
+
+export const generateSummary = task({
+  id: "generate-summary",
+  concurrency: [{ perKey: 1, total: 5 }, openaiLimit],
+  run: async (payload) => {},
+});
+```
+
+`perKey` caps each `concurrencyKey` pool and `total` caps across everything, keys or not. The queue-level `concurrencyLimit` option keeps working unchanged and is deprecated in favor of `concurrency`. Enforcement happens server-side; servers without support accept the option but do not enforce it yet.
+
+Manage limits at runtime with the new `concurrencyLimits` namespace: `list()` and `retrieve(name)` report each limit's bounds plus its live `running` and `queued` counts, `override(name, { perKey, total })` changes only the given bounds (overriding `total` to `0` pauses the limit), and `reset(name)` restores the declared values.

apps/webapp/app/runEngine/concerns/queues.server.ts

@@ -23,7 +23,12 @@ import {
   Namespace,
 } from "@internal/cache";
 import { singleton } from "~/utils/singleton";
-import type { TaskMetadataCache, TaskMetadataEntry } from "~/services/taskMetadataCache.server";
+import {
+  parseTaskGates,
+  type TaskMetadataCache,
+  type TaskMetadataEntry,
+  type TaskMetadataGate,
+} from "~/services/taskMetadataCache.server";
 import { taskMetadataCacheInstance } from "~/services/taskMetadataCacheInstance.server";
 import {
   recordTaskMetaResolve,
@@ -95,6 +100,7 @@ export class DefaultQueueManager implements QueueManager {
     let lockedQueueId: string | undefined;
     let taskTtl: string | null | undefined;
     let taskKind: string | undefined;
+    let taskGates: TaskMetadataGate[] | null | undefined;
 
     // Determine queue name based on lockToVersion and provided options
     if (lockedBackgroundWorker) {
@@ -146,6 +152,7 @@ export class DefaultQueueManager implements QueueManager {
           taskTtl = lockedMeta?.ttl ?? undefined;
         }
         taskKind = lockedMeta?.triggerSource;
+        taskGates = lockedMeta?.gates;
       } else {
         // No queue override - resolve default queue + TTL + triggerSource via cache,
         // falling back to a single BackgroundWorkerTask lookup on miss.
@@ -184,6 +191,7 @@ export class DefaultQueueManager implements QueueManager {
         queueName = lockedMeta.queueName;
         lockedQueueId = lockedMeta.queueId ?? undefined;
         taskKind = lockedMeta.triggerSource;
+        taskGates = lockedMeta.gates;
       }
     } else {
       // Task is not locked to a specific version, use regular logic
@@ -199,6 +207,7 @@ export class DefaultQueueManager implements QueueManager {
       queueName = taskInfo.queueName;
       taskTtl = taskInfo.taskTtl;
       taskKind = taskInfo.taskKind;
+      taskGates = taskInfo.taskGates;
     }
 
     // Sanitize the final determined queue name once
@@ -211,17 +220,29 @@ export class DefaultQueueManager implements QueueManager {
       queueName = sanitizedQueueName;
     }
 
+    const requestedGates = request.body.options?.gates ?? taskGates ?? undefined;
+    const gates = requestedGates
+      ?.flatMap((gate) => {
+        const sanitized = sanitizeQueueName(gate.queue);
+        return sanitized ? [{ queue: sanitized, concurrencyKey: gate.concurrencyKey }] : [];
+      })
+      .slice(0, 2);
+
     return {
       queueName,
       lockedQueueId,
       taskTtl,
       taskKind,
+      gates: gates && gates.length > 0 ? gates : undefined,
     };
   }
 
-  private async getTaskQueueInfo(
-    request: TriggerTaskRequest
-  ): Promise<{ queueName: string; taskTtl?: string | null; taskKind?: string | undefined }> {
+  private async getTaskQueueInfo(request: TriggerTaskRequest): Promise<{
+    queueName: string;
+    taskTtl?: string | null;
+    taskKind?: string | undefined;
+    taskGates?: TaskMetadataGate[] | null;
+  }> {
     const { taskId, environment, body } = request;
     const { queue } = body.options ?? {};
 
@@ -243,6 +264,7 @@ export class DefaultQueueManager implements QueueManager {
         queueName: overriddenQueueName,
         taskTtl: meta?.ttl ?? undefined,
         taskKind: meta?.triggerSource,
+        taskGates: meta?.gates,
       };
     }
 
@@ -259,10 +281,20 @@ export class DefaultQueueManager implements QueueManager {
         taskId,
         environmentId: environment.id,
       });
-      return { queueName: defaultQueueName, taskTtl: meta.ttl, taskKind: meta.triggerSource };
+      return {
+        queueName: defaultQueueName,
+        taskTtl: meta.ttl,
+        taskKind: meta.triggerSource,
+        taskGates: meta.gates,
+      };
     }
 
-    return { queueName: meta.queueName, taskTtl: meta.ttl, taskKind: meta.triggerSource };
+    return {
+      queueName: meta.queueName,
+      taskTtl: meta.ttl,
+      taskKind: meta.triggerSource,
+      taskGates: meta.gates,
+    };
   }
 
   /**
@@ -320,6 +352,7 @@ export class DefaultQueueManager implements QueueManager {
       triggerSource: row.triggerSource,
       queueId: row.queue?.id ?? null,
       queueName: row.queue?.name ?? "",
+      gates: parseTaskGates(row.gates),
     };
 
     // Fire-and-forget back-fill — `setByWorker` upserts the single field and
@@ -340,6 +373,7 @@ export class DefaultQueueManager implements QueueManager {
       select: {
         ttl: true,
         triggerSource: true,
+        gates: true,
         queue: { select: { id: true, name: true } },
       },
     });
@@ -378,6 +412,7 @@ export class DefaultQueueManager implements QueueManager {
       select: {
         ttl: true,
         triggerSource: true,
+        gates: true,
         queue: { select: { id: true, name: true } },
       },
     });
@@ -395,6 +430,7 @@ export class DefaultQueueManager implements QueueManager {
       triggerSource: row.triggerSource,
       queueId: row.queue?.id ?? null,
       queueName: row.queue?.name ?? "",
+      gates: parseTaskGates(row.gates),
     };
 
     // Fire-and-forget back-fill — atomically upserts the slug into both

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -445,7 +445,7 @@ export class RunEngineTriggerTaskService {
           const parkedOnExternalDeploymentId =
             externalDeploymentResolution?.outcome === "park" ? externalDeploymentId : undefined;
 
-          const { queueName, lockedQueueId, taskTtl, taskKind } =
+          const { queueName, lockedQueueId, taskTtl, taskKind, gates } =
             await this.queueConcern.resolveQueueProperties(
               triggerRequest,
               lockedToBackgroundWorker ?? undefined
@@ -663,6 +663,7 @@ export class RunEngineTriggerTaskService {
                       options,
                       queueName,
                       lockedQueueId,
+                      gates,
                       workerQueue,
                       region: migrated.region,
                       enableFastPath: migrated.enableFastPath,
@@ -743,6 +744,7 @@ export class RunEngineTriggerTaskService {
                   options,
                   queueName,
                   lockedQueueId,
+                  gates,
                   workerQueue,
                   region: migrated.region,
                   enableFastPath: migrated.enableFastPath,
@@ -905,6 +907,7 @@ export class RunEngineTriggerTaskService {
     options: TriggerTaskServiceOptions;
     queueName: string;
     lockedQueueId?: string;
+    gates?: Array<{ queue: string; concurrencyKey?: string }>;
     workerQueue?: string;
     region?: string;
     enableFastPath: boolean;
@@ -971,6 +974,7 @@ export class RunEngineTriggerTaskService {
           : args.body.options?.concurrencyKey,
       queue: args.queueName,
       lockedQueueId: args.lockedQueueId,
+      gates: args.gates,
       workerQueue: args.workerQueue,
       region: args.region,
       enableFastPath: args.enableFastPath,

apps/webapp/app/runEngine/types.ts

@@ -46,6 +46,8 @@ export type QueueProperties = {
   lockedQueueId?: string;
   taskTtl?: string | null;
   taskKind?: string;
+  /** Other queues the run must also hold a concurrency slot in while executing. */
+  gates?: Array<{ queue: string; concurrencyKey?: string }>;
 };
 
 export type LockedBackgroundWorker = Pick<

apps/webapp/app/services/taskMetadataCache.server.ts

@@ -1,13 +1,18 @@
 import type { Redis, Result, Callback } from "ioredis";
+import { parseGates } from "@internal/run-engine";
 import type { TaskTriggerSource } from "@trigger.dev/database";
 import { logger } from "./logger.server";
 
+export type TaskMetadataGate = { queue: string; concurrencyKey?: string };
+
 export type TaskMetadataEntry = {
   slug: string;
   ttl: string | null;
   triggerSource: TaskTriggerSource;
   queueId: string | null;
   queueName: string;
+  /** Task-declared gates, applied to every trigger that does not override them. */
+  gates: TaskMetadataGate[] | null;
 };
 
 export interface TaskMetadataCache {
@@ -52,11 +57,21 @@ export type RedisTaskMetadataCacheOptions = {
   byWorkerTtlSeconds?: number;
 };
 
+/**
+ * BackgroundWorkerTask.gates is an untyped Json column; keep only well-shaped
+ * entries so a malformed value can never fail a trigger.
+ */
+export function parseTaskGates(gates: unknown): TaskMetadataGate[] | null {
+  const parsed = parseGates(gates);
+  return parsed.length > 0 ? parsed : null;
+}
+
 type EncodedEntry = {
   t: string | null;
   k: TaskTriggerSource;
   q: string | null;
   n: string;
+  g?: TaskMetadataGate[] | null;
 };
 
 function encode(entry: TaskMetadataEntry): string {
@@ -65,6 +80,7 @@ function encode(entry: TaskMetadataEntry): string {
     k: entry.triggerSource,
     q: entry.queueId,
     n: entry.queueName,
+    g: entry.gates,
   };
   return JSON.stringify(payload);
 }
@@ -78,6 +94,7 @@ function decode(slug: string, raw: string): TaskMetadataEntry | null {
       triggerSource: parsed.k,
       queueId: parsed.q,
       queueName: parsed.n,
+      gates: parseTaskGates(parsed.g ?? null),
     };
   } catch (error) {
     logger.error("Failed to decode task metadata cache entry", { slug, error });

apps/webapp/app/v3/services/changeCurrentDeployment.server.ts

@@ -6,6 +6,7 @@ import { logger } from "~/services/logger.server";
 import { syncTaskIdentifiers } from "~/services/taskIdentifierRegistry.server";
 import {
   type TaskMetadataCache,
+  parseTaskGates,
   type TaskMetadataEntry,
 } from "~/services/taskMetadataCache.server";
 import { taskMetadataCacheInstance } from "~/services/taskMetadataCacheInstance.server";
@@ -119,6 +120,7 @@ export class ChangeCurrentDeploymentService extends BaseService {
           slug: true,
           triggerSource: true,
           ttl: true,
+          gates: true,
           queue: { select: { id: true, name: true } },
         },
       })
@@ -157,6 +159,7 @@ export class ChangeCurrentDeploymentService extends BaseService {
         triggerSource: t.triggerSource,
         queueId: t.queue?.id ?? null,
         queueName: t.queue?.name ?? "",
+        gates: parseTaskGates(t.gates),
       }));
 
       // Cache calls log+swallow internally.

apps/webapp/app/v3/services/createBackgroundWorker.server.ts

@@ -403,9 +403,9 @@ async function createWorkerTask(
         {
           name: task.queue?.name ?? `task/${task.id}`,
           concurrencyLimit: task.queue?.concurrencyLimit,
-          totalConcurrencyLimit: task.queue?.totalConcurrencyLimit,
+          combinedConcurrencyLimit: task.queue?.combinedConcurrencyLimit,
         },
-        task.id,
+        task.queue?.name ?? task.id,
         task.queue?.name ? "NAMED" : "VIRTUAL",
         worker,
         environment,
@@ -437,6 +437,7 @@ async function createWorkerTask(
         exportName: task.exportName,
         retryConfig: task.retry,
         queueConfig: task.queue,
+        gates: task.gates,
         machineConfig: task.machine,
         triggerSource: resolvedTriggerSource,
         config: task.agentConfig ? (task.agentConfig as any) : undefined,
@@ -454,6 +455,7 @@ async function createWorkerTask(
       triggerSource: resolvedTriggerSource,
       queueId: queue.id,
       queueName: queue.name,
+      gates: task.gates ?? null,
     };
   } catch (error) {
     if (error instanceof Prisma.PrismaClientKnownRequestError) {
@@ -477,6 +479,7 @@ async function createWorkerTask(
             triggerSource: resolvedTriggerSource,
             queueId: queue.id,
             queueName: queue.name,
+            gates: task.gates ?? null,
           };
         }
       } else {
@@ -555,7 +558,7 @@ async function createWorkerQueue(
   const taskQueue = await upsertWorkerQueueRecord(
     queueName,
     baseConcurrencyLimit ?? null,
-    queue.totalConcurrencyLimit ?? null,
+    queue.combinedConcurrencyLimit ?? null,
     orderableName,
     queueType,
     worker,

apps/webapp/app/v3/services/replayTaskRun.server.ts

@@ -142,6 +142,15 @@ export class ReplayTaskRunService extends BaseService {
               : undefined,
             concurrencyKey:
               overrideOptions.concurrencyKey ?? existingTaskRun.concurrencyKey ?? undefined,
+            /**
+             * A run that held gates replays with those same gates. A run with none
+             * stored passes undefined (never a fabricated empty array, which reads
+             * as "clear the named limits") so the replay honors whatever limits the
+             * task declares now, like it honors the current queue and retry config.
+             */
+            gates: Array.isArray(existingTaskRun.gates)
+              ? (existingTaskRun.gates as Array<{ queue: string; concurrencyKey?: string }>)
+              : undefined,
             maxAttempts: overrideOptions.maxAttempts,
             maxDuration: overrideOptions.maxDurationSeconds,
             machine: