feat(sdk,core,webapp,run-engine): declare queue gates on tasks and triggers

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

@@ -0,0 +1,18 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Cap a queue's combined concurrency across all of its `concurrencyKey` values with the new `combinedConcurrencyLimit` 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. `combinedConcurrencyLimit` 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,
+  combinedConcurrencyLimit: 10,
+});
+```
+
+Enforcement happens server-side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled combined concurrency limits accept the option but do not enforce it yet.

.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/queue-gates.md

@@ -0,0 +1,22 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Hold a run's concurrency slot in more than one queue with queue gates. Pass an array as `queue`: the first entry is the queue the run waits in, and up to two more name gates, other queues the run must also have capacity in and occupies while it executes. A gate without a `concurrencyKey` uses the run's own key, so a shared `tenant` queue caps a tenant across every task; a literal key pins the gate to one slot pool, capping, say, all traffic to one external provider.
+
+```ts
+import { queue, task } from "@trigger.dev/sdk";
+
+export const tenant = queue({ name: "tenant", concurrencyLimit: 10 });
+
+export const processWebhook = task({
+  id: "process-webhook",
+  queue: [{ name: "webhooks", concurrencyLimit: 2 }, "tenant"],
+  run: async (payload) => {},
+});
+
+await processWebhook.trigger(payload, { concurrencyKey: tenantId });
+```
+
+The same array form works on `queue` when triggering, replacing the task's gates for that run. Enforcement happens server-side; servers without queue gates enabled accept the option but run without it.

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

@@ -2,12 +2,16 @@ import type { Redis, Result, Callback } from "ioredis";
 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 +56,37 @@ 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 {
+  if (!Array.isArray(gates) || gates.length === 0) {
+    return null;
+  }
+
+  const parsed = gates.flatMap((gate) => {
+    if (!gate || typeof gate !== "object" || typeof (gate as any).queue !== "string") {
+      return [];
+    }
+    const concurrencyKey = (gate as any).concurrencyKey;
+    return [
+      {
+        queue: (gate as any).queue,
+        concurrencyKey: typeof concurrencyKey === "string" ? concurrencyKey : undefined,
+      },
+    ];
+  });
+
+  return parsed.length > 0 ? parsed.slice(0, 2) : null;
+}
+
 type EncodedEntry = {
   t: string | null;
   k: TaskTriggerSource;
   q: string | null;
   n: string;
+  g?: TaskMetadataGate[] | null;
 };
 
 function encode(entry: TaskMetadataEntry): string {
@@ -65,6 +95,7 @@ function encode(entry: TaskMetadataEntry): string {
     k: entry.triggerSource,
     q: entry.queueId,
     n: entry.queueName,
+    g: entry.gates,
   };
   return JSON.stringify(payload);
 }
@@ -78,6 +109,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,7 +403,7 @@ 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 ? "NAMED" : "VIRTUAL",
@@ -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,