You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Commit b81e175
Browse filesBrowse the repository at this point in the historyBrowse files
feat(run-engine,sdk,webapp): cap a queue's total concurrency across keyed and keyless runs (#4823)
## Summary
Adds a `totalConcurrencyLimit` option to queues. On a queue used with
`concurrencyKey`, `concurrencyLimit` applies to each key value
independently, so ten active keys with a limit of 5 can run 50 at once
and nothing bounds the queue as a whole short of the environment limit.
`totalConcurrencyLimit` caps in-flight runs across all keys while each
key still gets at most `concurrencyLimit`:
```ts
export const perUserQueue = queue({
name: "per-user-queue",
concurrencyLimit: 1,
totalConcurrencyLimit: 10,
});
```
Enforcement is gated behind
`RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED` (default off) and applies
to runs triggered with a `concurrencyKey`. With the gate off, admit
paths are unchanged.
## Design
The engine keeps a per-base-queue `groupConcurrency` set shared by every
concurrency-key variant; its cardinality is the queue's total in-flight
count. The concurrency-key dequeue script bounds each batch by
`min(totalLimit, envLimit) - SCARD(group)` and adds admitted runs to the
set. The enqueue fast path checks the same gate and falls back to a
normal enqueue when the queue is at its total limit.
Every release path (ack, nack, dead-letter, concurrency release, TTL
expiry, sweeper clear) removes a run from the group set whenever it
removes it from the per-key `currentConcurrency` set. Those removals run
regardless of the gate, so the set stays correct if the gate is later
turned off, and the existing reconciliation sweep self-heals the group
set because it is always a subset of the per-key sets it acks against.
The stored limit is the raw declared value; readers clamp to the
environment concurrency limit. The option flows through the queue
manifest into a new nullable `TaskQueue.totalConcurrencyLimit` column
(additive migration, internal name) and syncs to the engine on deploy.
The group set self-heals against release paths that miss the removal (an
instance on an older build during a rollout, or a future release script
such as the one
[#4398](#4398) adds).
Every terminal release deletes the run's message key, so when a queue
sits at its total the dequeue gate prunes members whose message key no
longer exists, throttled to one pass per interval per queue. Runs in
flight before the gate is enabled are the opposite case: they are absent
from the set, so a queue can transiently exceed its total by at most
that count, converging as each one completes.
Not included here, planned as follow-ups: a runtime override API for the
total limit, dashboard and metrics surfacing, and skipping queues at
their total during fair-queue selection.
A follow-up in this stack renames the public option to
`combinedConcurrencyLimit` before release; engine internals and storage
keep these names.
Cap a queue's total concurrency across all of its `concurrencyKey` values with the new `totalConcurrencyLimit` queue option. On a keyed queue, `concurrencyLimit` applies to each key value independently, so ten active keys with a limit of 5 can run 50 at once. `totalConcurrencyLimit` bounds the whole queue while each key still gets at most `concurrencyLimit`.
7
+
8
+
```ts
9
+
import { queue } from"@trigger.dev/sdk";
10
+
11
+
exportconst perUserQueue =queue({
12
+
name: "per-user-queue",
13
+
concurrencyLimit: 1,
14
+
totalConcurrencyLimit: 10,
15
+
});
16
+
```
17
+
18
+
Enforcement happens server-side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled total concurrency limits accept the option but do not enforce it yet.
0 commit comments