diff --git a/docs/concurrency.mdx b/docs/concurrency.mdx
new file mode 100644
index 00000000000..02f17a8c055
--- /dev/null
+++ b/docs/concurrency.mdx
@@ -0,0 +1,370 @@
+---
+title: "Concurrency"
+description: "Limit how many runs execute at once: per task, per tenant, or shared across tasks."
+---
+
+When you trigger a task, the [run](/runs) waits in a [queue](/queues) and runs start in trigger order as capacity allows. Concurrency is what decides how much capacity there is.
+
+By default, concurrency is only limited by your environment concurrency limit. If you need more control (for example, to limit concurrency or share limits across multiple tasks), you can set a concurrency limit on the task, or declare a named limit and share it, as described below.
+
+Controlling concurrency is useful when you have a task that can't be run concurrently, or when you want to limit the number of runs to avoid overloading a resource.
+
+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 a limit between tasks](#sharing-a-limit-between-tasks)
+- **Give each tenant its own separate concurrency**: [Concurrency keys and per-tenant limits](#concurrency-keys-and-per-tenant-limits)
+- **Per-tenant limits with a ceiling on the total**: [Per-key and total limits together](#per-key-and-total-limits-together)
+- **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)
+- **Switch a run's limits when you trigger it**: [Setting limits when you trigger a run](#setting-limits-when-you-trigger-a-run)
+
+## Default concurrency
+
+By default, all tasks have an unbounded concurrency limit, limited only by the overall concurrency limits of your environment.
+
+
+ Your environment has a base concurrency limit and a burstable limit (default burst factor of 2.0x
+ the base limit). Individual tasks and limits are capped by the base concurrency limit, not the
+ burstable limit. For example, if your base limit is 10, your environment can burst up to 20
+ concurrent runs, but any single limit can allow at most 10 concurrent runs. If you're a paying
+ customer you can request higher burst limits by [contacting us](https://www.trigger.dev/contact).
+
+
+## Setting task concurrency
+
+Set the `concurrency` option on a task to limit how many of its runs execute at once. `{ total: n }` caps the task outright:
+
+```ts /trigger/one-at-a-time.ts
+import { task } from "@trigger.dev/sdk";
+
+// This task will only run one at a time
+export const oneAtATime = task({
+ id: "one-at-a-time",
+ concurrency: { total: 1 },
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+This is useful if you need to control access to a shared resource, like a database or an API that has rate limits.
+
+There are two ways to bound a task, and you can combine them:
+
+- `total` caps every run of the task together, whether or not runs use a `concurrencyKey`.
+- `perKey` caps each `concurrencyKey` pool separately; runs triggered without a key share one pool.
+
+```ts /trigger/per-user.ts
+import { task } from "@trigger.dev/sdk";
+
+export const generateReport = task({
+ id: "generate-report",
+ // each user runs at most 1 at a time, and at most 10 run in total
+ concurrency: { perKey: 1, total: 10 },
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+
+ The `queue: {"{ concurrencyLimit: n }"}` option keeps working but is deprecated in favor of
+ `concurrency`. Its single number means "per key when runs pass a `concurrencyKey`, whole queue
+ when they don't" — the `concurrency` shape says which you mean explicitly.
+
+
+## Sharing a limit between tasks
+
+Declare a named limit with `concurrencyLimit()` and put it in each task's `concurrency`. Every task holding the limit draws from the same pools:
+
+```ts /trigger/limits.ts
+import { concurrencyLimit, task } from "@trigger.dev/sdk";
+
+// at most 25 concurrent runs across every task that holds this limit
+export const openaiLimit = concurrencyLimit({ name: "openai", total: 25 });
+
+export const generateSummary = task({
+ id: "generate-summary",
+ concurrency: openaiLimit,
+ run: async (payload) => {
+ // ...
+ },
+});
+
+export const generateTitle = task({
+ id: "generate-title",
+ concurrency: openaiLimit,
+ run: async (payload) => {
+ // ...
+ },
+});
+```
+
+A task's `concurrency` takes a single item or an array: at most one inline shape (which caps that task alone) plus up to two named limits. A run starts only when every limit it holds has capacity, and it occupies a slot in each while it executes:
+
+```ts /trigger/summarize.ts
+import { concurrencyLimit, task } from "@trigger.dev/sdk";
+
+export const openaiLimit = concurrencyLimit({ name: "openai", total: 25 });
+
+export const summarizeThread = task({
+ id: "summarize-thread",
+ // this task runs at most 5 at once, and also counts towards the shared openai limit
+ concurrency: [{ total: 5 }, openaiLimit],
+ run: async (payload) => {
+ // ...
+ },
+});
+```
+
+
+ Names you declare with `concurrencyLimit()` are 1-122 characters using only letters, numbers,
+ underscores and hyphens.
+
+
+## Setting limits when you trigger a run
+
+The trigger-time `concurrency` option takes limit names and replaces the task's declared **named** limits for that run. The task's inline limit always applies:
+
+```ts app/api/report/route.ts
+import { generateReport } from "~/trigger/reports";
+
+// this run counts towards "priority" instead of the task's declared named limits
+await generateReport.trigger(data, { concurrency: ["priority"] });
+```
+
+Pass an empty array to run with only the task's inline limit.
+
+## Concurrency keys and per-tenant limits
+
+If you're building an application where you want to run tasks for your users, you might want a separate limit for each of your users (or orgs, projects, etc.).
+
+You can do this by passing a `concurrencyKey` when you trigger. Each unique key value gets its own pool under every `perKey` bound the run holds:
+
+```ts app/api/report/route.ts
+import { generateReport } from "~/trigger/per-user";
+
+export async function POST(request: Request) {
+ const data = await request.json();
+
+ const handle = await generateReport.trigger(data, {
+ // every user gets their own concurrency pool
+ concurrencyKey: data.userId,
+ });
+
+ return Response.json(handle);
+}
+```
+
+## Per-key and total limits together
+
+`perKey` on its own lets total concurrency grow with the number of active keys: ten active users under `perKey: 5` can run 50 at once. Add `total` to bound everything as a group. Each key still gets at most `perKey`, and all runs together — keyed or not — never exceed `total`:
+
+```ts /trigger/per-user-capped.ts
+import { task } from "@trigger.dev/sdk";
+
+export const processUpload = task({
+ id: "process-upload",
+ // each user runs at most 1 at a time, and at most 10 run in total
+ concurrency: { perKey: 1, total: 10 },
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+## A per-tenant cap across multiple tasks
+
+A named limit's `perKey` bound follows each run's own `concurrencyKey`, so one declaration caps each tenant across every task holding the limit:
+
+```ts /trigger/webhooks.ts
+import { concurrencyLimit, task } from "@trigger.dev/sdk";
+
+// each tenant runs at most 10 at once across every task that holds this limit
+export const tenantLimit = concurrencyLimit({ name: "tenant", perKey: 10 });
+
+export const processWebhook = task({
+ id: "process-webhook",
+ // webhooks themselves are capped at 2 per tenant, within the tenant's overall 10
+ concurrency: [{ perKey: 2 }, tenantLimit],
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+```ts app/api/webhook/route.ts
+// the run counts towards this tenant's pool in both limits
+await processWebhook.trigger(payload, { concurrencyKey: tenantId });
+```
+
+## 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 limit with only a `total` and share it. It counts every run holding it, whether or not the run has a `concurrencyKey`:
+
+```ts /trigger/sync.ts
+import { concurrencyLimit, task } from "@trigger.dev/sdk";
+
+// at most 10 concurrent provider calls across every task and every tenant
+export const providerApiLimit = concurrencyLimit({ name: "provider-api", total: 10 });
+
+export const syncToProvider = task({
+ id: "sync-to-provider",
+ concurrency: [{ total: 20 }, providerApiLimit],
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+Runs with and without a `concurrencyKey` share the same `total`, so this works even when only some of your triggers have a natural key.
+
+## Concurrency and subtasks
+
+When you trigger a task that has subtasks, the subtasks will not inherit the parent's limits. Unless otherwise specified, subtasks run under their own task's configuration:
+
+```ts /trigger/subtasks.ts
+export const parentTask = task({
+ id: "parent-task",
+ run: async (payload) => {
+ //trigger a subtask
+ await subtask.triggerAndWait(payload);
+ },
+});
+
+// This subtask runs under its own limits
+export const subtask = task({
+ id: "subtask",
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+## Waits and concurrency
+
+With our [task checkpoint system](/how-it-works#the-checkpoint-resume-system), tasks can wait at various waitpoints (like waiting for subtasks to complete, delays, or external events). The way this system interacts with the concurrency system is important to understand.
+
+Concurrency is only released when a run reaches a waitpoint and is checkpointed. When a run is checkpointed, it transitions to the `WAITING` state and releases its concurrency slots back to every limit it holds and the environment, allowing other runs to execute or resume.
+
+This means that:
+
+- Only actively executing runs count towards concurrency limits
+- Runs in the `WAITING` state (checkpointed at waitpoints) do not consume concurrency slots
+- You can have more runs in the `WAITING` state than a limit allows to execute
+- When a waiting run resumes (e.g., when a subtask completes), it must re-acquire its slots
+
+For example, if a task has `concurrency: { total: 1 }`:
+
+- You can only have exactly 1 run executing at a time
+- You may have multiple runs in the `WAITING` state for that task
+- When the executing run reaches a waitpoint and checkpoints, it releases its slot
+- The next queued run can then begin execution
+
+### Short time-based waits keep their slot
+
+Checkpointing takes time, so a run doesn't checkpoint the moment it reaches a waitpoint. For [`wait.for()`](/wait-for) and [`wait.until()`](/wait-until) it happens 60 seconds into the wait, so anything shorter stays `EXECUTING` and holds its slots for the whole wait. If you're polling in a loop, use an interval comfortably above 60 seconds so the slots are actually released between polls.
+
+### Waiting for a subtask
+
+When a parent task triggers and waits for a subtask, the parent task will checkpoint and release its concurrency slots once it reaches the wait point. This prevents environment deadlocks where all concurrency slots would be occupied by waiting tasks.
+
+```ts /trigger/waiting.ts
+export const parentTask = task({
+ id: "parent-task",
+ concurrency: { total: 1 },
+ run: async (payload) => {
+ //trigger a subtask and wait for it to complete
+ await subtask.triggerAndWait(payload);
+ // The parent task checkpoints here and releases its concurrency slots
+ // allowing other tasks to execute while waiting
+ },
+});
+
+export const subtask = task({
+ id: "subtask",
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+When the parent task reaches the `triggerAndWait` call, it checkpoints and transitions to the `WAITING` state, releasing its slots. Once the subtask completes, the parent task will resume and re-acquire them.
+
+## Managing concurrency limits with the SDK
+
+The `concurrencyLimits` namespace manages your named limits at runtime (anonymous inline limits appear under derived `task/` names):
+
+```ts
+import { concurrencyLimits } from "@trigger.dev/sdk";
+```
+
+### Listing limits
+
+```ts
+import { concurrencyLimits } from "@trigger.dev/sdk";
+
+// List all limits (returns paginated results)
+const allLimits = await concurrencyLimits.list();
+
+// With pagination options
+const pagedLimits = await concurrencyLimits.list({
+ page: 1,
+ perPage: 20,
+});
+```
+
+### Retrieving a limit
+
+Retrieve a limit by its name to see its bounds and live counts:
+
+```ts
+import { concurrencyLimits } from "@trigger.dev/sdk";
+
+const limit = await concurrencyLimits.retrieve("openai");
+```
+
+The limit object contains each bound plus live counts:
+
+```ts
+{
+ id: "climit_1234",
+ name: "openai",
+ perKey: {
+ current: null, // Enforced right now (null = no per-key bound)
+ base: null, // Declared in your code
+ override: null, // Override value (if set)
+ overriddenAt: null, // When the override was applied
+ },
+ total: {
+ current: 25,
+ base: 25,
+ override: null,
+ overriddenAt: null,
+ },
+ running: 14, // Runs executing that hold this limit
+ queued: 100, // Runs queued that must clear this limit to execute
+}
+```
+
+### Overriding a limit
+
+Overrides change only the fields you pass; the declared values are kept and restored by `reset`. Overriding `total` to `0` blocks every run holding the limit, which is how you pause a limit:
+
+```ts
+import { concurrencyLimits } from "@trigger.dev/sdk";
+
+// Raise the shared cap
+await concurrencyLimits.override("openai", { total: 50 });
+
+// Pause the limit: nothing holding it can start
+await concurrencyLimits.override("openai", { total: 0 });
+
+// Back to the values declared in your code
+await concurrencyLimits.reset("openai");
+```
+
+Overrides survive deploys: redeploying your code keeps an active override until you reset it.
diff --git a/docs/database-connections.mdx b/docs/database-connections.mdx
index df808e66088..caf6b0927fa 100644
--- a/docs/database-connections.mdx
+++ b/docs/database-connections.mdx
@@ -68,7 +68,7 @@ Set the pool small. A task usually runs its queries in sequence, so one connecti
| Drizzle (node-postgres) | 10 (the underlying `pg` pool) |
| [MongoDB driver](https://www.mongodb.com/docs/drivers/node/current/connect/connection-options/connection-pools/) | 100 (`maxPoolSize`) |
-Keep `concurrent runs × pool size` under your provider's connection limit, and cap how many runs execute at once with [concurrency limits](/queue-concurrency) so runs queue instead of overrunning the database. Direct connection limits for common Postgres providers:
+Keep `concurrent runs × pool size` under your provider's connection limit, and cap how many runs execute at once with [concurrency limits](/concurrency) so runs queue instead of overrunning the database. Direct connection limits for common Postgres providers:
| Provider | Direct connection limit |
| --- | --- |
@@ -197,7 +197,7 @@ export const myChat = chat.agent({
## Troubleshooting
-`too many connections` or connection refused: `concurrent runs × pool size` is over your provider's limit. Lower the pool size, cap [concurrency](/queue-concurrency), or connect through a pooler.
+`too many connections` or connection refused: `concurrent runs × pool size` is over your provider's limit. Lower the pool size, cap [concurrency](/concurrency), or connect through a pooler.
The worker crashes right after resuming from a wait: an idle connection that closed during the suspend emitted an unhandled `error` event. Attach `pool.on("error", ...)` on a `pg` pool (node-postgres or Drizzle); Prisma and the MongoDB driver handle this internally.
@@ -208,6 +208,6 @@ When a task waits, the runtime can [checkpoint](/how-it-works#the-checkpoint-res
## See also
- [Wait](/wait) for the primitives that trigger a checkpoint.
-- [Concurrency and queues](/queue-concurrency) to cap how many runs execute at once.
+- [Concurrency](/concurrency) to cap how many runs execute at once.
- [Lifecycle functions](/tasks/overview#onwait-and-onresume-functions) for global `tasks.onWait` and `tasks.onResume`.
- [Chat agent lifecycle hooks](/ai-chat/lifecycle-hooks) for `onChatSuspend` and `onChatResume`.
diff --git a/docs/docs.json b/docs/docs.json
index 55771592538..860233fa51a 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -65,7 +65,8 @@
"group": "Wait",
"pages": ["wait", "wait-for", "wait-until", "wait-for-token"]
},
- "queue-concurrency",
+ "queues",
+ "concurrency",
"versioning",
"machines",
"idempotency",
@@ -383,6 +384,15 @@
"management/queues/concurrency-reset"
]
},
+ {
+ "group": "Concurrency limits API",
+ "pages": [
+ "management/concurrency-limits/list",
+ "management/concurrency-limits/retrieve",
+ "management/concurrency-limits/override",
+ "management/concurrency-limits/reset"
+ ]
+ },
{
"group": "Schedules API",
"pages": [
@@ -729,6 +739,11 @@
]
},
"redirects": [
+ {
+ "source": "/queue-concurrency",
+ "destination": "/concurrency",
+ "permanent": true
+ },
{
"source": "/config/extensions",
"destination": "/config/extensions/overview",
diff --git a/docs/introduction.mdx b/docs/introduction.mdx
index 155965b2a73..e7fec1884b1 100644
--- a/docs/introduction.mdx
+++ b/docs/introduction.mdx
@@ -71,8 +71,11 @@ We provide everything you need to build and manage background tasks: a CLI and S
>
Learn how to handle errors and retries.
-
- Configure what you want to happen when there is more than one run at a time.
+
+ Limit how many runs execute at once: per task, per tenant, or shared across tasks.
+
+
+ Control the order your runs execute in.
- Your environment has a base concurrency limit and a burstable limit (default burst factor of 2.0x
- the base limit). Individual queues are limited by the base concurrency limit, not the burstable
- limit. For example, if your base limit is 10, your environment can burst up to 20 concurrent runs,
- but any single queue can have at most 10 concurrent runs. If you're a paying customer you can
- request higher burst limits by [contacting us](https://www.trigger.dev/contact).
-
-
-## Setting task concurrency
-
-You can set the concurrency limit for a task by setting the `concurrencyLimit` property on the task's queue. This limits the number of runs that can be executing at any one time:
-
-```ts /trigger/one-at-a-time.ts
-// This task will only run one at a time
-export const oneAtATime = task({
- id: "one-at-a-time",
- queue: {
- concurrencyLimit: 1,
- },
- run: async (payload) => {
- //...
- },
-});
-```
-
-This is useful if you need to control access to a shared resource, like a database or an API that has rate limits.
-
-## Sharing concurrency between tasks
-
-As well as putting queue settings directly on a task, you can define a queue and reuse it across multiple tasks. This allows you to share the same concurrency limit:
-
-```ts /trigger/queue.ts
-export const myQueue = queue({
- name: "my-queue",
- concurrencyLimit: 1,
-});
-
-export const task1 = task({
- id: "task-1",
- queue: myQueue,
- run: async (payload: { message: string }) => {
- // ...
- },
-});
-
-export const task2 = task({
- id: "task-2",
- queue: myQueue,
- run: async (payload: { message: string }) => {
- // ...
- },
-});
-```
-
-In this example, `task1` and `task2` share the same queue, so only one of them can run at a time.
-
-## Setting the queue when you trigger a run
-
-When you trigger a task you can override the default queue. This is really useful if you sometimes have high priority runs.
-
-The task and queue definition:
-
-```ts /trigger/override-concurrency.ts
-const paidQueue = queue({
- name: "paid-users",
- concurrencyLimit: 10,
-});
-
-export const generatePullRequest = task({
- id: "generate-pull-request",
- queue: {
- //normally when triggering this task it will be limited to 1 run at a time
- concurrencyLimit: 1,
- },
- run: async (payload) => {
- //todo generate a PR using OpenAI
- },
-});
-```
-
-Triggering from your backend and overriding the queue:
-
-```ts app/api/push/route.ts
-import { generatePullRequest } from "~/trigger/override-concurrency";
-
-export async function POST(request: Request) {
- const data = await request.json();
-
- if (data.branch === "main") {
- //trigger the task, with the paid users queue
- const handle = await generatePullRequest.trigger(data, {
- // Set the paid users queue
- queue: "paid-users",
- });
-
- return Response.json(handle);
- } else {
- //triggered with the default queue (concurrency of 1)
- const handle = await generatePullRequest.trigger(data);
- return Response.json(handle);
- }
-}
-```
-
-## Concurrency keys and per-tenant queuing
-
-If you're building an application where you want to run tasks for your users, you might want a separate queue for each of your users (or orgs, projects, etc.).
-
-You can do this by using `concurrencyKey`. It creates a copy of the queue for each unique value of the key.
-
-Your backend code:
-
-```ts app/api/pr/route.ts
-import { generatePullRequest } from "~/trigger/override-concurrency";
-
-export async function POST(request: Request) {
- const data = await request.json();
-
- if (data.isFreeUser) {
- //the "free-users" queue has a concurrency limit of 1
- const handle = await generatePullRequest.trigger(data, {
- queue: "free-users",
- //this creates a free-users queue for each user
- concurrencyKey: data.userId,
- });
-
- //return a success response with the handle
- return Response.json(handle);
- } else {
- //the "paid-users" queue has a concurrency limit of 10
- const handle = await generatePullRequest.trigger(data, {
- queue: "paid-users",
- //this creates a paid-users queue for each user
- concurrencyKey: data.userId,
- });
-
- //return a success response with the handle
- return Response.json(handle);
- }
-}
-```
-
-## 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
-
-```ts /trigger/subtasks.ts
-export const parentTask = task({
- id: "parent-task",
- run: async (payload) => {
- //trigger a subtask
- await subtask.triggerAndWait(payload);
- },
-});
-
-// This subtask will run on its own queue
-export const subtask = task({
- id: "subtask",
- run: async (payload) => {
- //...
- },
-});
-```
-
-## Waits and concurrency
-
-With our [task checkpoint system](/how-it-works#the-checkpoint-resume-system), tasks can wait at various waitpoints (like waiting for subtasks to complete, delays, or external events). The way this system interacts with the concurrency system is important to understand.
-
-Concurrency is only released when a run reaches a waitpoint and is checkpointed. When a run is checkpointed, it transitions to the `WAITING` state and releases its concurrency slot back to both the queue and the environment, allowing other runs to execute or resume.
-
-This means that:
-
-- Only actively executing runs count towards concurrency limits
-- Runs in the `WAITING` state (checkpointed at waitpoints) do not consume concurrency slots
-- You can have more runs in the `WAITING` state than your queue's concurrency limit
-- When a waiting run resumes (e.g., when a subtask completes), it must re-acquire a concurrency slot
-
-For example, if you have a queue with a `concurrencyLimit` of 1:
-
-- You can only have exactly 1 run executing at a time
-- You may have multiple runs in the `WAITING` state that belong to that queue
-- When the executing run reaches a waitpoint and checkpoints, it releases its slot
-- The next queued run can then begin execution
-
-### Short time-based waits keep their slot
-
-Checkpointing takes time, so a run doesn't checkpoint the moment it reaches a waitpoint. For [`wait.for()`](/wait-for) and [`wait.until()`](/wait-until) it happens 60 seconds into the wait, so anything shorter stays `EXECUTING` and holds its slot for the whole wait. If you're polling in a loop, use an interval comfortably above 60 seconds so the slot is actually released between polls.
-
-### Waiting for a subtask on a different queue
-
-When a parent task triggers and waits for a subtask on a different queue, the parent task will checkpoint and release its concurrency slot once it reaches the wait point. This prevents environment deadlocks where all concurrency slots would be occupied by waiting tasks.
-
-```ts /trigger/waiting.ts
-export const parentTask = task({
- id: "parent-task",
- queue: {
- concurrencyLimit: 1,
- },
- run: async (payload) => {
- //trigger a subtask and wait for it to complete
- await subtask.triggerAndWait(payload);
- // The parent task checkpoints here and releases its concurrency slot
- // allowing other tasks to execute while waiting
- },
-});
-
-export const subtask = task({
- id: "subtask",
- run: async (payload) => {
- //...
- },
-});
-```
-
-When the parent task reaches the `triggerAndWait` call, it checkpoints and transitions to the `WAITING` state, releasing its concurrency slot back to both its queue and the environment. Once the subtask completes, the parent task will resume and re-acquire a concurrency slot.
-
-## Managing queues with the SDK
-
-The SDK provides a `queues` namespace that allows you to manage queues programmatically. You can list, retrieve, pause, resume, and modify concurrency limits for queues.
-
-
- Import from `@trigger.dev/sdk`:
- ```ts
- import { queues } from "@trigger.dev/sdk";
- ```
-
-
-### Listing queues
-
-You can list all queues in your environment with pagination support:
-
-```ts
-import { queues } from "@trigger.dev/sdk";
-
-// List all queues (returns paginated results)
-const allQueues = await queues.list();
-
-// With pagination options
-const pagedQueues = await queues.list({
- page: 1,
- perPage: 20,
-});
-```
-
-### Retrieving a queue
-
-You can retrieve a specific queue by its ID, or by its type and name:
-
-```ts
-import { queues } from "@trigger.dev/sdk";
-
-// Using queue ID (starts with "queue_")
-const queueById = await queues.retrieve("queue_1234");
-
-// Using type and name for a task's default queue
-const taskQueue = await queues.retrieve({
- type: "task",
- name: "my-task-id",
-});
-
-// Using type and name for a custom queue
-const customQueue = await queues.retrieve({
- type: "custom",
- name: "my-custom-queue",
-});
-```
-
-The queue object contains useful information about the queue state:
-
-```ts
-{
- id: "queue_1234", // Queue ID
- name: "my-task-id", // Queue name
- type: "task", // "task" or "custom"
- running: 5, // Currently executing runs
- queued: 10, // Runs waiting to execute
- paused: false, // Whether the queue is paused
- concurrencyLimit: 10, // Current concurrency limit
- concurrency: {
- current: 10, // Effective limit
- base: 10, // Default limit from code
- override: null, // Override value (if set)
- overriddenAt: null, // When override was applied
- overriddenBy: null, // Who applied the override
- }
-}
-```
-
-### Pausing and resuming queues
-
-You can pause a queue to prevent new runs from starting. Runs that are currently executing will continue to completion.
-
-```ts
-import { queues } from "@trigger.dev/sdk";
-
-// Pause a queue using its ID
-await queues.pause("queue_1234");
-
-// Or using type and name
-await queues.pause({ type: "task", name: "my-task-id" });
-await queues.pause({ type: "custom", name: "my-custom-queue" });
-```
-
-To resume a paused queue and allow new runs to start:
-
-```ts
-import { queues } from "@trigger.dev/sdk";
-
-// Resume a queue using its ID
-await queues.resume("queue_1234");
-
-// Or using type and name
-await queues.resume({ type: "task", name: "my-task-id" });
-await queues.resume({ type: "custom", name: "my-custom-queue" });
-```
-
-### Overriding concurrency limits
-
-You can temporarily override a queue's concurrency limit. This is useful for scaling up or down based on demand:
-
-```ts
-import { queues } from "@trigger.dev/sdk";
-
-// Set concurrency limit to 5
-await queues.overrideConcurrencyLimit("queue_1234", 5);
-
-// Or using type and name
-await queues.overrideConcurrencyLimit({ type: "task", name: "my-task-id" }, 20);
-```
-
-To reset the concurrency limit back to the base value defined in your code:
-
-```ts
-import { queues } from "@trigger.dev/sdk";
-
-// Reset concurrency limit to the base value
-await queues.resetConcurrencyLimit("queue_1234");
-
-// Or using type and name
-await queues.resetConcurrencyLimit({ type: "task", name: "my-task-id" });
-```
diff --git a/docs/queues.mdx b/docs/queues.mdx
new file mode 100644
index 00000000000..10de0ea659e
--- /dev/null
+++ b/docs/queues.mdx
@@ -0,0 +1,235 @@
+---
+title: "Queues"
+description: "Control the order your runs execute in."
+---
+
+import Priority from "/snippets/priority.mdx";
+
+When you trigger a task, it isn't executed immediately. Instead, the task [run](/runs) is placed into a queue for execution. Runs in a queue are started in the order they were triggered, first in, first out, unless you give a run a [priority](#priority) at trigger time.
+
+By default, each task gets its own queue, so triggering the same task ten times executes those runs in trigger order. How many of them execute *at once* is a separate question, governed by [concurrency](/concurrency) — your environment's concurrency limit, plus any limits you set on the task.
+
+## Sharing a queue between tasks
+
+Declare a queue with `queue()` and set it on several tasks to interleave their runs in one queue. Runs from every task on the queue start in the order they were triggered, whichever task they belong to:
+
+```ts /trigger/emails.ts
+import { queue, task } from "@trigger.dev/sdk";
+
+const emailQueue = queue({ name: "emails" });
+
+export const sendWelcomeEmail = task({
+ id: "send-welcome-email",
+ queue: emailQueue,
+ run: async (payload) => {
+ //...
+ },
+});
+
+export const sendDigestEmail = task({
+ id: "send-digest-email",
+ queue: emailQueue,
+ run: async (payload) => {
+ //...
+ },
+});
+```
+
+
+ If what you want to share is a concurrency cap rather than ordering, you don't need a shared
+ queue: declare a named limit with `concurrencyLimit()` and each task keeps its own queue while
+ drawing from the shared limit. See [sharing a limit between
+ tasks](/concurrency#sharing-a-limit-between-tasks).
+
+
+## Setting the queue when you trigger a run
+
+When you trigger a task you can override its queue by name. This is really useful if you sometimes have high priority runs:
+
+```ts /trigger/override-queue.ts
+import { queue, task } from "@trigger.dev/sdk";
+
+const paidQueue = queue({ name: "paid-users" });
+
+export const generatePullRequest = task({
+ id: "generate-pull-request",
+ // normally this task is limited to 1 run at a time
+ concurrency: { total: 1 },
+ run: async (payload) => {
+ //todo generate a PR using OpenAI
+ },
+});
+```
+
+Triggering from your backend and overriding the queue:
+
+```ts app/api/push/route.ts
+import { generatePullRequest } from "~/trigger/override-queue";
+
+export async function POST(request: Request) {
+ const data = await request.json();
+
+ if (data.branch === "main") {
+ //trigger the task, with the paid users queue
+ const handle = await generatePullRequest.trigger(data, {
+ // Set the paid users queue
+ queue: "paid-users",
+ });
+
+ return Response.json(handle);
+ } else {
+ //triggered with the default queue
+ const handle = await generatePullRequest.trigger(data);
+ return Response.json(handle);
+ }
+}
+```
+
+## Priority
+
+You can re-order runs within a queue at trigger time by giving them a priority:
+
+
+
+## Managing queues with the SDK
+
+The SDK provides a `queues` namespace that allows you to manage queues programmatically. You can list, retrieve, pause, resume, and modify concurrency limits for queues.
+
+
+ Import from `@trigger.dev/sdk`:
+ ```ts
+ import { queues } from "@trigger.dev/sdk";
+ ```
+
+
+### Listing queues
+
+You can list all queues in your environment with pagination support:
+
+```ts
+import { queues } from "@trigger.dev/sdk";
+
+// List all queues (returns paginated results)
+const allQueues = await queues.list();
+
+// With pagination options
+const pagedQueues = await queues.list({
+ page: 1,
+ perPage: 20,
+});
+```
+
+### Retrieving a queue
+
+You can retrieve a specific queue by its ID, or by its type and name:
+
+```ts
+import { queues } from "@trigger.dev/sdk";
+
+// Using queue ID (starts with "queue_")
+const queueById = await queues.retrieve("queue_1234");
+
+// Using type and name for a task's default queue
+const taskQueue = await queues.retrieve({
+ type: "task",
+ name: "my-task-id",
+});
+
+// Using type and name for a custom queue
+const customQueue = await queues.retrieve({
+ type: "custom",
+ name: "my-custom-queue",
+});
+```
+
+The queue object contains useful information about the queue state, and its `version` discriminates the shape:
+
+```ts
+// V1: the queue carries its own concurrency limit and override state
+{
+ id: "queue_1234", // Queue ID
+ name: "my-task-id", // Queue name
+ type: "task", // "task" or "custom"
+ version: "V1",
+ running: 5, // Currently executing runs
+ queued: 10, // Runs waiting to execute
+ paused: false, // Whether the queue is paused
+ concurrencyLimit: 10, // The queue's own limit
+ concurrency: {
+ current: 10, // Effective limit
+ base: 10, // Default limit from code
+ override: null, // Override value (if set)
+ overriddenAt: null, // When override was applied
+ overriddenBy: null, // Who applied the override
+ }
+}
+
+// V2: the queue is only the line runs wait in. Concurrency is declared with
+// the task `concurrency` option and managed through `concurrencyLimits`.
+{
+ id: "queue_5678",
+ name: "my-v2-task-id",
+ type: "task",
+ version: "V2",
+ running: 3,
+ queued: 12,
+ paused: false,
+ concurrencyLimit: null,
+}
+```
+
+### Pausing and resuming queues
+
+You can pause a queue to prevent new runs from starting. Runs that are currently executing will continue to completion.
+
+```ts
+import { queues } from "@trigger.dev/sdk";
+
+// Pause a queue using its ID
+await queues.pause("queue_1234");
+
+// Or using type and name
+await queues.pause({ type: "task", name: "my-task-id" });
+await queues.pause({ type: "custom", name: "my-custom-queue" });
+```
+
+To resume a paused queue and allow new runs to start:
+
+```ts
+import { queues } from "@trigger.dev/sdk";
+
+// Resume a queue using its ID
+await queues.resume("queue_1234");
+
+// Or using type and name
+await queues.resume({ type: "task", name: "my-task-id" });
+await queues.resume({ type: "custom", name: "my-custom-queue" });
+```
+
+### Overriding concurrency limits
+
+You can temporarily override a queue's concurrency limit. This is useful for scaling up or down based on demand:
+
+```ts
+import { queues } from "@trigger.dev/sdk";
+
+// Set concurrency limit to 5
+await queues.overrideConcurrencyLimit("queue_1234", 5);
+
+// Or using type and name
+await queues.overrideConcurrencyLimit({ type: "task", name: "my-task-id" }, 20);
+```
+
+To reset the concurrency limit back to the base value defined in your code:
+
+```ts
+import { queues } from "@trigger.dev/sdk";
+
+// Reset concurrency limit to the base value
+await queues.resetConcurrencyLimit("queue_1234");
+
+// Or using type and name
+await queues.resetConcurrencyLimit({ type: "task", name: "my-task-id" });
+```
+
+Overrides survive deploys: redeploying your code keeps an active override until you reset it.
diff --git a/docs/reports.mdx b/docs/reports.mdx
index 07d517bc3b0..c4a330a569c 100644
--- a/docs/reports.mdx
+++ b/docs/reports.mdx
@@ -148,7 +148,7 @@ An unknown report key returns `404` with the list of available keys.
Every tool the MCP server exposes, including `get_report`.
-
+
Configure the concurrency limits the Flow verdict checks against.
diff --git a/docs/runs/priority.mdx b/docs/runs/priority.mdx
index 20da1894629..374b248078f 100644
--- a/docs/runs/priority.mdx
+++ b/docs/runs/priority.mdx
@@ -3,29 +3,6 @@ title: "Priority"
description: "Specify a priority when triggering a run."
---
-You can set a priority when you trigger a run. This allows you to prioritize some of your runs over others, so they are started sooner. This is very useful when:
+import Priority from "/snippets/priority.mdx";
-- You have critical work that needs to start more quickly (and you have long queues).
-- You want runs for your premium users to take priority over free users.
-
-The value for priority is a time offset in seconds that determines the order of dequeuing.
-
-
-
-If you specify a priority of `10` the run will dequeue before runs that were triggered with no priority 8 seconds ago, like in this example:
-
-```ts
-// no priority = 0
-await myTask.trigger({ foo: "bar" });
-
-//... imagine 8s pass by
-
-// this run will start before the run above that was triggered 8s ago (with no priority)
-await myTask.trigger({ foo: "bar" }, { priority: 10 });
-```
-
-If you passed a value of `3600` the run would dequeue before runs that were triggered an hour ago (with no priority).
-
-
- Setting a high priority will not allow you to beat runs from other organizations. It will only affect the order of your own runs.
-
+
diff --git a/docs/snippets/priority.mdx b/docs/snippets/priority.mdx
new file mode 100644
index 00000000000..91d99a6de6e
--- /dev/null
+++ b/docs/snippets/priority.mdx
@@ -0,0 +1,27 @@
+You can set a priority when you trigger a run. This allows you to prioritize some of your runs over others, so they are started sooner. This is very useful when:
+
+- You have critical work that needs to start more quickly (and you have long queues).
+- You want runs for your premium users to take priority over free users.
+
+The value for priority is a time offset in seconds that determines the order of dequeuing.
+
+
+
+If you specify a priority of `10` the run will dequeue before runs that were triggered with no priority 8 seconds ago, like in this example:
+
+```ts
+// no priority = 0
+await myTask.trigger({ foo: "bar" });
+
+//... imagine 8s pass by
+
+// this run will start before the run above that was triggered 8s ago (with no priority)
+await myTask.trigger({ foo: "bar" }, { priority: 10 });
+```
+
+If you passed a value of `3600` the run would dequeue before runs that were triggered an hour ago (with no priority).
+
+
+ Setting a high priority will not allow you to beat runs from other organizations. It will only
+ affect the order of your own runs.
+
diff --git a/docs/tasks/overview.mdx b/docs/tasks/overview.mdx
index d7cb4d8f4d9..9d287a93334 100644
--- a/docs/tasks/overview.mdx
+++ b/docs/tasks/overview.mdx
@@ -98,16 +98,14 @@ For more information read [the retrying guide](/errors-retrying).
It's also worth mentioning that you can [retry a block of code](/errors-retrying) inside your tasks as well.
-### `queue` options
+### `queue` and `concurrency` options
-Queues allow you to control the concurrency of your tasks. This allows you to have one-at-a-time execution and parallel executions. There are also more advanced techniques like having different concurrencies for different sets of your users. For more information read [the concurrency & queues guide](/queue-concurrency).
+Queues control the order your runs execute in, and concurrency limits control how many execute at once: one-at-a-time execution, parallel executions, and more advanced techniques like separate concurrency for different sets of your users. For more information read the [Queues](/queues) and [Concurrency](/concurrency) guides.
```ts /trigger/one-at-a-time.ts
export const oneAtATime = task({
id: "one-at-a-time",
- queue: {
- concurrencyLimit: 1,
- },
+ concurrency: { total: 1 },
run: async (payload: any, { ctx }) => {
//...
},
diff --git a/docs/v3-openapi.yaml b/docs/v3-openapi.yaml
index a97ae70307e..f7a225b0131 100644
--- a/docs/v3-openapi.yaml
+++ b/docs/v3-openapi.yaml
@@ -3062,7 +3062,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:
@@ -3108,6 +3109,203 @@ paths:
name: "my-task-id",
});
+ "/api/v1/concurrency-limits":
+ get:
+ operationId: list_concurrency_limits_v1
+ summary: List concurrency limits
+ description: |
+ List the environment's declared concurrency limits (anonymous inline limits
+ appear under their derived `task/` names), with each limit's bounds
+ and its live running and queued counts. Results are ordered by the underlying
+ row name, so named limits sort before `task/`-derived inline limits.
+ parameters:
+ - name: page
+ in: query
+ required: false
+ schema:
+ type: integer
+ minimum: 1
+ default: 1
+ - name: perPage
+ in: query
+ required: false
+ schema:
+ type: integer
+ minimum: 1
+ default: 25
+ maximum: 100
+ responses:
+ "200":
+ description: Concurrency limits retrieved successfully
+ content:
+ application/json:
+ schema:
+ type: object
+ properties:
+ data:
+ type: array
+ items:
+ "$ref": "#/components/schemas/ConcurrencyLimitObject"
+ pagination:
+ type: object
+ properties:
+ currentPage:
+ type: integer
+ totalPages:
+ type: integer
+ count:
+ type: integer
+ "401":
+ description: Unauthorized request
+ tags:
+ - concurrency-limits
+ security:
+ - secretKey: []
+ x-codeSamples:
+ - lang: typescript
+ source: |-
+ import { concurrencyLimits } from "@trigger.dev/sdk";
+
+ const limits = await concurrencyLimits.list({ page: 1, perPage: 20 });
+
+ "/api/v1/concurrency-limits/{name}":
+ get:
+ operationId: retrieve_concurrency_limit_v1
+ summary: Retrieve concurrency limit
+ description: Retrieve a concurrency limit by name, with its bounds and live running and queued counts.
+ parameters:
+ - name: name
+ in: path
+ required: true
+ schema:
+ type: string
+ description: The limit's name, as declared with `concurrencyLimit()`
+ responses:
+ "200":
+ description: Concurrency limit retrieved successfully
+ content:
+ application/json:
+ schema:
+ "$ref": "#/components/schemas/ConcurrencyLimitObject"
+ "401":
+ description: Unauthorized request
+ "404":
+ description: Concurrency limit not found
+ tags:
+ - concurrency-limits
+ security:
+ - secretKey: []
+ x-codeSamples:
+ - lang: typescript
+ source: |-
+ import { concurrencyLimits } from "@trigger.dev/sdk";
+
+ const limit = await concurrencyLimits.retrieve("openai");
+
+ "/api/v1/concurrency-limits/{name}/override":
+ post:
+ operationId: override_concurrency_limit_v1
+ summary: Override concurrency limit
+ description: |
+ Override a concurrency limit's bounds. Only the given fields change; the declared
+ values are kept and restored by reset. Overriding `total` to `0` blocks every run
+ holding the limit, which is how a limit is paused.
+ parameters:
+ - name: name
+ in: path
+ required: true
+ schema:
+ type: string
+ description: The limit's name
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ type: object
+ minProperties: 1
+ additionalProperties: false
+ description: At least one of `perKey` or `total` must be provided.
+ properties:
+ perKey:
+ type: integer
+ minimum: 0
+ maximum: 100000
+ description: Caps each concurrencyKey pool; runs without a key share one pool. May not exceed the environment concurrency limit.
+ total:
+ type: integer
+ minimum: 0
+ maximum: 100000
+ description: Caps every run holding this limit, keys or not. May not exceed the environment concurrency limit.
+ responses:
+ "200":
+ description: Concurrency limit overridden successfully
+ content:
+ application/json:
+ schema:
+ "$ref": "#/components/schemas/ConcurrencyLimitObject"
+ "400":
+ description: Invalid request parameters, or a bound exceeds the environment concurrency limit
+ "401":
+ description: Unauthorized request
+ "404":
+ description: Concurrency limit not found
+ "409":
+ description: The limit changed concurrently; retry the request
+ tags:
+ - concurrency-limits
+ security:
+ - secretKey: []
+ x-codeSamples:
+ - lang: typescript
+ source: |-
+ import { concurrencyLimits } from "@trigger.dev/sdk";
+
+ // Raise the shared cap
+ await concurrencyLimits.override("openai", { total: 50 });
+
+ // Pause the limit
+ await concurrencyLimits.override("openai", { total: 0 });
+
+ "/api/v1/concurrency-limits/{name}/reset":
+ post:
+ operationId: reset_concurrency_limit_v1
+ summary: Reset concurrency limit
+ description: Reset a concurrency limit back to the values declared in your code, clearing any override.
+ parameters:
+ - name: name
+ in: path
+ required: true
+ schema:
+ type: string
+ description: The limit's name
+ responses:
+ "200":
+ description: Concurrency limit reset successfully
+ content:
+ application/json:
+ schema:
+ "$ref": "#/components/schemas/ConcurrencyLimitObject"
+ "400":
+ description: The limit has no override to reset
+ "401":
+ description: Unauthorized request
+ "404":
+ description: Concurrency limit not found
+ "409":
+ description: The limit changed concurrently; retry the request
+ tags:
+ - concurrency-limits
+ security:
+ - secretKey: []
+ x-codeSamples:
+ - lang: typescript
+ source: |-
+ import { concurrencyLimits } from "@trigger.dev/sdk";
+
+ await concurrencyLimits.reset("openai");
+
+
"/api/v1/waitpoints/tokens":
post:
operationId: create_waitpoint_token_v1
@@ -4260,9 +4458,85 @@ components:
minimum: 0
maximum: 1000
description: An optional property that specifies the maximum number of concurrent run executions. If this property is omitted, the task can potentially use up the full concurrency of an environment.
+ ConcurrencyLimitObject:
+ type: object
+ required:
+ - id
+ - name
+ - perKey
+ - total
+ - running
+ - queued
+ properties:
+ id:
+ type: string
+ description: The limit's id, starting with `climit_`
+ example: "climit_abcdef123456"
+ name:
+ type: string
+ description: The limit's name, as declared with `concurrencyLimit()`
+ example: "openai"
+ perKey:
+ type: object
+ description: Caps each concurrencyKey pool; runs without a key share one pool
+ properties:
+ current:
+ type: integer
+ nullable: true
+ description: Enforced right now (null = no per-key bound). Enforcement clamps it to the environment concurrency limit at admit time.
+ example: null
+ base:
+ type: integer
+ nullable: true
+ description: The declared value an override reverts to on reset
+ example: null
+ override:
+ type: integer
+ nullable: true
+ description: The overridden value, when an override is active
+ example: null
+ overriddenAt:
+ type: string
+ format: date-time
+ nullable: true
+ description: When the override was applied
+ example: null
+ total:
+ type: object
+ description: Caps every run holding this limit, keys or not
+ properties:
+ current:
+ type: integer
+ nullable: true
+ description: Enforced right now (null = no total bound). Enforcement clamps it to the environment concurrency limit at admit time.
+ example: 25
+ base:
+ type: integer
+ nullable: true
+ description: The declared value an override reverts to on reset
+ example: 25
+ override:
+ type: integer
+ nullable: true
+ description: The overridden value, when an override is active
+ example: null
+ overriddenAt:
+ type: string
+ format: date-time
+ nullable: true
+ description: When the override was applied
+ example: null
+ running:
+ type: integer
+ description: Runs executing that hold this limit
+ example: 14
+ queued:
+ type: integer
+ description: Runs that are queued and must clear this limit to execute
+ example: 100
QueueObject:
type: object
- required: ["id", "name", "type", "running", "queued", "paused"]
+ required: ["id", "name", "type", "version", "running", "queued", "paused"]
properties:
id:
type: string
@@ -4280,6 +4554,14 @@ components:
- `task`: Created automatically for each task
- `custom`: Created explicitly in your code using `queue()`
example: task
+ version:
+ type: string
+ enum: [V1, V2]
+ description: |
+ Discriminates the shape:
+ - `V1`: the queue carries its own `concurrencyLimit` (applied per key when runs pass a `concurrencyKey`, to the whole queue when they don't) and its `concurrency` override state
+ - `V2`: the queue is only the line runs wait in; concurrency is declared with the task `concurrency` option and managed through the concurrency-limits endpoints. `concurrencyLimit` is always null and `concurrency` is absent.
+ example: V1
running:
type: integer
description: The number of runs currently executing
@@ -4295,11 +4577,11 @@ components:
concurrencyLimit:
type: integer
nullable: true
- description: The current concurrency limit of the queue
+ description: The queue's own concurrency limit. Meaningful on V1 queues; always null on V2 queues.
example: 10
concurrency:
type: object
- description: Detailed concurrency information
+ description: Detailed concurrency information. V1 queues only.
properties:
current:
type: integer
diff --git a/docs/writing-tasks-introduction.mdx b/docs/writing-tasks-introduction.mdx
index 5f0bc330912..fe738d4db23 100644
--- a/docs/writing-tasks-introduction.mdx
+++ b/docs/writing-tasks-introduction.mdx
@@ -15,7 +15,8 @@ Before digging deeper into the details of writing tasks, you should read the [fu
| [Logging](/logging) | View and send logs and traces from your tasks. |
| [Errors & retrying](/errors-retrying) | How to deal with errors and write reliable tasks. |
| [Wait](/wait) | Wait for periods of time or for external events to occur before continuing. |
-| [Concurrency & Queues](/queue-concurrency) | Configure what you want to happen when there is more than one run at a time. |
+| [Queues](/queues) | Control the order your runs execute in. |
+| [Concurrency](/concurrency) | Limit how many runs execute at once: per task, per tenant, or shared across tasks. |
| [Realtime notifications](/realtime/overview) | Send realtime notifications from your task that you can subscribe to from your backend or frontend. |
| [Versioning](/versioning) | How versioning works. |
| [Machines](/machines) | Configure the CPU and RAM of the machine your task runs on |