docs: combined concurrency limits and queue gates

Author: matt-aitken

docs/docs.json

@@ -379,7 +379,9 @@
               "management/queues/retrieve",
               "management/queues/pause",
               "management/queues/concurrency-override",
-              "management/queues/concurrency-reset"
+              "management/queues/concurrency-reset",
+              "management/queues/combined-concurrency-override",
+              "management/queues/combined-concurrency-reset"
             ]
           },
           {

docs/management/queues/combined-concurrency-override.mdx

@@ -0,0 +1,4 @@
+---
+title: "Override Combined Concurrency Limit"
+openapi: "v3-openapi POST /api/v1/queues/{queueParam}/concurrency/combined/override"
+---

docs/management/queues/combined-concurrency-reset.mdx

@@ -0,0 +1,4 @@
+---
+title: "Reset Combined Concurrency Limit"
+openapi: "v3-openapi POST /api/v1/queues/{queueParam}/concurrency/combined/reset"
+---

docs/queue-concurrency.mdx

@@ -157,6 +157,86 @@ export async function POST(request: Request) {
 }
 ```
 
+## Combined concurrency across keys
+
+`concurrencyKey` gives every key value its own copy of the queue, each with the queue's full `concurrencyLimit`. That means the queue's total concurrency grows with the number of active keys: ten active users on a queue with `concurrencyLimit: 5` can run 50 at once.
+
+To bound the whole queue, set `combinedConcurrencyLimit`. Each key still gets at most `concurrencyLimit`, and the queue as a whole never exceeds the combined limit across all keys:
+
+```ts /trigger/per-user.ts
+export const perUserQueue = queue({
+  name: "per-user-queue",
+  //each user runs at most 1 at a time...
+  concurrencyLimit: 1,
+  //...and at most 10 users can be running at once
+  combinedConcurrencyLimit: 10,
+});
+```
+
+The combined limit only applies to runs triggered with a `concurrencyKey`; runs without a key are governed by `concurrencyLimit` alone. On the Queues page in the dashboard, a queue with a combined limit shows it in brackets next to the per-key limit, e.g. `1 (10)`.
+
+<Note>
+  If you self-host, combined limits are enforced by default and can be disabled with
+  `RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED=0`. When enforcement is disabled the limit is
+  still accepted, stored, and shown, but runs are not held back by it.
+</Note>
+
+## Holding slots in more than one queue (queue gates)
+
+Sometimes one limit isn't enough: a webhook processor should be capped as a task, but each tenant should also have a global cap across every task they run. Queue gates let a run hold a concurrency slot in more than one queue at once.
+
+Pass an array as `queue`: the first entry is the run's home queue (where it waits), and up to two more entries name gates — other queues the run must also have capacity in before it starts, and occupies while it executes:
+
+```ts /trigger/webhooks.ts
+export const tenantQueue = queue({ name: "tenant", concurrencyLimit: 10 });
+
+export const processWebhook = task({
+  id: "process-webhook",
+  queue: [{ name: "webhooks", concurrencyLimit: 2 }, "tenant"],
+  run: async (payload) => {
+    //...
+  },
+});
+```
+
+```ts app/api/webhook/route.ts
+//the run waits in "webhooks" and also counts towards this tenant's cap
+await processWebhook.trigger(payload, { concurrencyKey: tenantId });
+```
+
+A gate without a `concurrencyKey` uses the run's own key, so the shared `tenant` queue above caps each tenant across every task that names it as a gate. Give the gate a literal key to pin it to a single slot pool instead, for example capping all traffic to one external provider across your whole environment:
+
+```ts /trigger/sync.ts
+export const syncToProvider = task({
+  id: "sync-to-provider",
+  queue: [
+    { name: "sync-home", concurrencyLimit: 20 },
+    //every run shares one "provider-api" pool regardless of its own key
+    { name: "provider-api", concurrencyKey: "shared" },
+  ],
+  run: async (payload) => {
+    //...
+  },
+});
+```
+
+The same array form works when you trigger, replacing the task's gates for that run:
+
+```ts
+await processWebhook.trigger(payload, {
+  queue: ["webhooks", "tenant"],
+  concurrencyKey: tenantId,
+});
+```
+
+A run starts only when its home queue and every gate all have capacity, and it releases all of its slots together when it finishes or suspends.
+
+<Note>
+  Queue gates are enforced when the server has them enabled. If you self-host, set
+  `RUN_ENGINE_QUEUE_GATES_ENABLED=1`; servers without gates enabled accept the option but run
+  without it.
+</Note>
+
 ## Concurrency and subtasks
 
 When you trigger a task that has subtasks, the subtasks will not inherit the queue from the parent task. Unless otherwise specified, subtasks will run on their own queue
@@ -356,3 +436,19 @@ await queues.resetConcurrencyLimit("queue_1234");
 // Or using type and name
 await queues.resetConcurrencyLimit({ type: "task", name: "my-task-id" });
 ```
+
+### Overriding the combined concurrency limit
+
+Queues with a `combinedConcurrencyLimit` can have that cap overridden and reset in the same way:
+
+```ts
+import { queues } from "@trigger.dev/sdk";
+
+// Allow up to 100 runs across all concurrency keys
+await queues.overrideCombinedConcurrencyLimit("queue_1234", 100);
+
+// Revert to the combinedConcurrencyLimit declared in your code
+await queues.resetCombinedConcurrencyLimit("queue_1234");
+```
+
+Overrides survive deploys: redeploying your code keeps an active override until you reset it.

docs/v3-openapi.yaml

@@ -3048,6 +3048,132 @@ paths:
               20
             );
 
+  "/api/v1/queues/{queueParam}/concurrency/combined/override":
+    post:
+      operationId: override_queue_combined_concurrency_v1
+      summary: Override combined concurrency limit
+      description: |
+        Override the combined concurrency limit of a queue: the cap on concurrent runs across
+        all of the queue's `concurrencyKey` values. Useful for temporarily scaling a whole
+        keyed queue up or down without changing each key's own limit.
+      parameters:
+        - in: path
+          name: queueParam
+          required: true
+          schema:
+            type: string
+          description: The queue ID (e.g., `queue_1234`), or the name of the queue when using the `type` body parameter.
+          example: queue_1234
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: ["combinedConcurrencyLimit"]
+              properties:
+                type:
+                  type: string
+                  enum: [id, task, custom]
+                  default: id
+                  description: |
+                    How to interpret the `queueParam` path parameter:
+                    - `id`: Treat as a queue ID (default)
+                    - `task`: Treat as a task ID to get the task's default queue
+                    - `custom`: Treat as a custom queue name
+                combinedConcurrencyLimit:
+                  type: integer
+                  minimum: 0
+                  maximum: 100000
+                  description: |
+                    The new combined concurrency limit to set for the queue. It may not exceed
+                    your environment's maximum concurrency limit: a higher value is rejected
+                    with a 400, not capped to the maximum.
+      responses:
+        "200":
+          description: Combined concurrency limit overridden successfully
+          content:
+            application/json:
+              schema:
+                "$ref": "#/components/schemas/QueueObject"
+        "400":
+          description: |
+            Invalid request parameters, or the requested combined concurrency limit exceeds
+            the environment's maximum concurrency limit.
+        "401":
+          description: Unauthorized request
+        "404":
+          description: Queue not found
+      tags:
+        - queues
+      security:
+        - secretKey: []
+      x-codeSamples:
+        - lang: typescript
+          source: |-
+            import { queues } from "@trigger.dev/sdk";
+
+            // Allow up to 100 runs across all concurrency keys
+            await queues.overrideCombinedConcurrencyLimit("queue_1234", 100);
+
+            // Using type and name
+            await queues.overrideCombinedConcurrencyLimit(
+              { type: "custom", name: "per-user-queue" },
+              100
+            );
+
+  "/api/v1/queues/{queueParam}/concurrency/combined/reset":
+    post:
+      operationId: reset_queue_combined_concurrency_v1
+      summary: Reset combined concurrency limit
+      description: Reset the combined concurrency limit of a queue back to the `combinedConcurrencyLimit` declared in your code.
+      parameters:
+        - in: path
+          name: queueParam
+          required: true
+          schema:
+            type: string
+          description: The queue ID (e.g., `queue_1234`), or the name of the queue when using the `type` body parameter.
+          example: queue_1234
+      requestBody:
+        required: false
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                type:
+                  type: string
+                  enum: [id, task, custom]
+                  default: id
+                  description: |
+                    How to interpret the `queueParam` path parameter:
+                    - `id`: Treat as a queue ID (default)
+                    - `task`: Treat as a task ID to get the task's default queue
+                    - `custom`: Treat as a custom queue name
+      responses:
+        "200":
+          description: Combined concurrency limit reset successfully
+          content:
+            application/json:
+              schema:
+                "$ref": "#/components/schemas/QueueObject"
+        "401":
+          description: Unauthorized request
+        "404":
+          description: Queue not found
+      tags:
+        - queues
+      security:
+        - secretKey: []
+      x-codeSamples:
+        - lang: typescript
+          source: |-
+            import { queues } from "@trigger.dev/sdk";
+
+            // Revert to the combinedConcurrencyLimit declared in code
+            await queues.resetCombinedConcurrencyLimit("queue_1234");
+
   "/api/v1/queues/{queueParam}/concurrency/reset":
     post:
       operationId: reset_queue_concurrency_v1
@@ -4321,6 +4447,37 @@ components:
               format: date-time
               nullable: true
               description: When the concurrency limit was overridden
+            combined:
+              type: object
+              description: |
+                The combined concurrency cap across all `concurrencyKey` values of the queue.
+                Present when the queue has a `combinedConcurrencyLimit`.
+              properties:
+                current:
+                  type: integer
+                  nullable: true
+                  description: The current combined concurrency limit as declared or overridden (null = no cap). Enforcement clamps it to the environment concurrency limit at admit time.
+                  example: 10
+                base:
+                  type: integer
+                  nullable: true
+                  description: The declared combined limit an override reverts to on reset
+                  example: 10
+                override:
+                  type: integer
+                  nullable: true
+                  description: The overridden combined limit, when an override is active
+                  example: null
+                overriddenAt:
+                  type: string
+                  format: date-time
+                  nullable: true
+                  description: When the combined override was applied
+                running:
+                  type: integer
+                  nullable: true
+                  description: Runs currently in flight across all concurrencyKey values
+                  example: 4
               example: null
             overriddenBy:
               type: string