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

.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

@@ -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 {

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

@@ -142,6 +142,9 @@ export class ReplayTaskRunService extends BaseService {
               : undefined,
             concurrencyKey:
               overrideOptions.concurrencyKey ?? existingTaskRun.concurrencyKey ?? undefined,
+            gates: Array.isArray(existingTaskRun.gates)
+              ? (existingTaskRun.gates as Array<{ queue: string; concurrencyKey?: string }>)
+              : undefined,
             maxAttempts: overrideOptions.maxAttempts,
             maxDuration: overrideOptions.maxDurationSeconds,
             machine:

internal-packages/database/prisma/migrations/20260829120000_add_queue_gates/migration.sql

@@ -0,0 +1,5 @@
+-- AlterTable
+ALTER TABLE "BackgroundWorkerTask" ADD COLUMN "gates" JSONB;
+
+-- AlterTable
+ALTER TABLE "TaskRun" ADD COLUMN "gates" JSONB;

internal-packages/database/prisma/schema.prisma

@@ -741,6 +741,9 @@ model BackgroundWorkerTask {
   queueConfig   Json?
   retryConfig   Json?
   machineConfig Json?
+  /// Gates declared on the task: other queues its runs must also hold a concurrency
+  /// slot in while executing. Shape: [{ queue: string, concurrencyKey?: string }]
+  gates         Json?
 
   queueId String?
   queue   TaskQueue? @relation(fields: [queueId], references: [id], onDelete: SetNull, onUpdate: Cascade)
@@ -1107,6 +1110,10 @@ model TaskRun {
 
   concurrencyKey String?
 
+  /// Gates for this run: other queues it must also hold a concurrency slot in while
+  /// executing. Shape: [{ queue: string, concurrencyKey?: string }]
+  gates Json?
+
   delayUntil        DateTime?
   queuedAt          DateTime?
   ttl               String?

internal-packages/run-engine/src/engine/index.ts

@@ -833,6 +833,7 @@ export class RunEngine {
       sdkVersion,
       cliVersion,
       concurrencyKey,
+      gates,
       workerQueue,
       region,
       enableFastPath,
@@ -1011,6 +1012,7 @@ export class RunEngine {
                 sdkVersion,
                 cliVersion,
                 concurrencyKey,
+                gates,
                 queue,
                 lockedQueueId,
                 workerQueue,