-
-
Notifications
You must be signed in to change notification settings - Fork 1.4k
docs: Queues and Concurrency guides for the concurrency option and named limits #4853
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
matt-aitken
wants to merge
17
commits into
feat/queue-concurrency-metrics
from
docs/queue-concurrency-features
Open
Changes from all commits
Commits
Show all changes
17 commits
Select commit
Hold shift + click to select a range
76f9344
docs: combined concurrency limits and queue gates
matt-aitken 0ba71e7
docs: correct the combined override body field, reset 400, and gate e…
matt-aitken 51124e1
docs: per-key home cap wording, combined field always present, reset …
matt-aitken 226685d
docs: older servers may omit the combined field, so check it defensively
matt-aitken f1eac8a
docs: use-case-first structure for queue concurrency
matt-aitken 75d0271
docs: the global-cap gate pattern requires keyed triggers
matt-aitken 4237c84
docs: imports for the standalone queue examples
matt-aitken fe7464d
docs: import for the trigger-time snippet
matt-aitken 2f57b43
docs: scope the combined-limit claim to keyed runs
matt-aitken 4a165b9
docs: task concurrency, named limits and the concurrency limits API
matt-aitken d9fabf6
docs: required response fields, minimum override body, scoped name-ru…
matt-aitken be48f4d
docs: the override body admits only the supported fields
matt-aitken 83223ad
docs: split the Concurrency & Queues page into Queues and Concurrency
matt-aitken 97d5164
docs: pagination minimums in the spec, priority qualifies the queue o…
matt-aitken 71400d2
docs: queue examples drop the deprecated concurrencyLimit field, intr…
matt-aitken 61a0ffd
docs: queue reads are version-discriminated
matt-aitken d3ab45e
docs: the concurrency-limits list orders by underlying row name
matt-aitken File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
|
matt-aitken marked this conversation as resolved.
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
|
matt-aitken marked this conversation as resolved.
|
||
|
|
||
| ## Default concurrency | ||
|
|
||
| By default, all tasks have an unbounded concurrency limit, limited only by the overall concurrency limits of your environment. | ||
|
|
||
| <Note> | ||
| 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). | ||
| </Note> | ||
|
|
||
| ## 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) => { | ||
| //... | ||
| }, | ||
| }); | ||
| ``` | ||
|
|
||
| <Note> | ||
| 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. | ||
| </Note> | ||
|
|
||
| ## 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) => { | ||
| // ... | ||
| }, | ||
| }); | ||
| ``` | ||
|
|
||
| <Note> | ||
| Names you declare with `concurrencyLimit()` are 1-122 characters using only letters, numbers, | ||
| underscores and hyphens. | ||
| </Note> | ||
|
|
||
| ## 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/<task-id>` 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. | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.