diff --git a/docs/docs.json b/docs/docs.json index d7524a9664b..60aedbe6a62 100644 --- a/docs/docs.json +++ b/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" ] }, { diff --git a/docs/management/queues/combined-concurrency-override.mdx b/docs/management/queues/combined-concurrency-override.mdx new file mode 100644 index 00000000000..40aba8ae062 --- /dev/null +++ b/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" +--- diff --git a/docs/management/queues/combined-concurrency-reset.mdx b/docs/management/queues/combined-concurrency-reset.mdx new file mode 100644 index 00000000000..4d7031d354c --- /dev/null +++ b/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" +--- diff --git a/docs/queue-concurrency.mdx b/docs/queue-concurrency.mdx index b832ffc26da..64c1f575589 100644 --- a/docs/queue-concurrency.mdx +++ b/docs/queue-concurrency.mdx @@ -11,6 +11,16 @@ Controlling concurrency is useful when you have a task that can't be run concurr It's important to note that only actively executing runs count towards concurrency limits. Runs that are delayed or waiting in a queue do not consume concurrency slots until they begin execution. +## Use cases + +- **Limit how many runs of a task execute at once**: [Setting task concurrency](#setting-task-concurrency) +- **Share one limit across several tasks**: [Sharing concurrency between tasks](#sharing-concurrency-between-tasks) +- **Give each tenant its own separate concurrency**: [Concurrency keys and per-tenant queuing](#concurrency-keys-and-per-tenant-queuing) +- **Per-tenant limits with a ceiling on the whole queue**: [Combined concurrency across keys](#combined-concurrency-across-keys) +- **Cap a tenant across every task they run**: [A per-tenant cap across multiple tasks](#a-per-tenant-cap-across-multiple-tasks) +- **Cap a shared resource, like an external API, across tasks and tenants**: [A global cap for a shared resource](#a-global-cap-for-a-shared-resource) +- **Funnel every run into one shared pool**: [One shared pool ignoring keys](#one-shared-pool-ignoring-keys) + ## Default concurrency By default, all tasks have an unbounded concurrency limit, limited only by the overall concurrency limits of your environment. @@ -157,6 +167,117 @@ 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 total runs across all users + 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)`. + + + On a queue used with `concurrencyKey`, every limit applies per key value except + `combinedConcurrencyLimit`, which is the only cap that spans the whole queue. + + +## Using multiple queues at once + +Sometimes one limit isn't enough. A run always waits in one queue, its **home queue**, but it can also hold a concurrency slot in up to two more queues, called **gates**. A run starts only when its home queue and every gate all have capacity, occupies a slot in each while it executes, and releases them together when it finishes or suspends. + +Pass an array as `queue`: the first entry is the home queue, the rest name gates. 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 per-tenant cap across multiple tasks + +A gate without a `concurrencyKey` uses the run's own key. Declare a shared queue and gate every relevant task on it, and each tenant gets one cap spanning all of those tasks: + +```ts /trigger/webhooks.ts +//each tenant runs at most 10 at once across every task that gates on this queue +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 }); +``` + +Because the trigger passes a `concurrencyKey`, the home queue splits per key as usual: `concurrencyLimit: 2` caps each tenant's webhook runs, not the task overall. + +### A global cap for a shared resource + +To cap something global, like total traffic to an external API, across many tasks and all tenants: declare a queue with only a `combinedConcurrencyLimit` and gate on it. Tenant keys still split the gate into per-key pools, but with no per-key limit the combined cap is the only constraint: + +```ts /trigger/sync.ts +//at most 10 concurrent provider calls across every task and every tenant +export const providerApiQueue = queue({ + name: "provider-api", + combinedConcurrencyLimit: 10, +}); + +export const syncToProvider = task({ + id: "sync-to-provider", + queue: [{ name: "sync-home", concurrencyLimit: 20 }, "provider-api"], + run: async (payload) => { + //... + }, +}); +``` + +```ts app/api/sync/route.ts +//the combined cap only counts keyed runs, so every trigger passes a key +await syncToProvider.trigger(payload, { concurrencyKey: tenantId }); +``` + + + This pattern requires every trigger to pass a `concurrencyKey`. A run triggered without one + bypasses the combined limit entirely. If some runs have no natural key, use the pinned-key + pool below instead. + + +### One shared pool ignoring keys + +Give a gate a literal `concurrencyKey` to pin every run into a single first-come-first-served pool, regardless of each run's own key: + +```ts /trigger/print.ts +export const printerQueue = queue({ name: "printer", concurrencyLimit: 1 }); + +export const printLabel = task({ + id: "print-label", + queue: [ + { name: "print-home", concurrencyLimit: 5 }, + //every run shares the single "printer" slot no matter its own key + { name: "printer", concurrencyKey: "shared" }, + ], + run: async (payload) => { + //... + }, +}); +``` + ## 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 +477,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. diff --git a/docs/v3-openapi.yaml b/docs/v3-openapi.yaml index a97ae70307e..b764b59de3a 100644 --- a/docs/v3-openapi.yaml +++ b/docs/v3-openapi.yaml @@ -3048,6 +3048,135 @@ 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: ["concurrencyLimit"] + 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 + concurrencyLimit: + 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: true + description: At least an empty JSON object `{}` must be sent; a zero-length body is rejected with a 400. + 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" + "400": + description: The queue's combined concurrency limit is not overridden, or invalid request parameters + "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 @@ -3062,7 +3191,8 @@ paths: 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 + required: true + description: At least an empty JSON object `{}` must be sent; a zero-length body is rejected with a 400. content: application/json: schema: @@ -4322,6 +4452,39 @@ components: nullable: true description: When the concurrency limit was overridden example: null + combined: + type: object + description: | + The combined concurrency cap across all `concurrencyKey` values of the queue. + Servers on this version always emit it, with `current` null when the queue + has no combined limit; older servers may omit the field entirely, so check + `combined?.current != null` rather than relying on its presence. + 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 overriddenBy: type: string nullable: true