Skip to content

Commit ea97581

Browse files
matt-aitkenTrigger.dev RepoOps
matt-aitken
authored and
Trigger.dev RepoOps
committed
feat(run-engine,webapp,sdk,core,cli): queue concurrency limits, gates, overrides, metrics, and management API
Queues get richer concurrency controls: a total concurrency limit across all concurrency keys of a queue (alongside the existing per-key limit), named concurrency limits declared in code and shared across tasks, and trigger-time selection between them. The dashboard gains a Concurrency page with per-queue metrics charts and an override dialog for per-key and total bounds, plus a management API for listing, retrieving, overriding and resetting concurrency limits. Chat sessions can set concurrency options when triggering. Run admission gates in the engine enforce the new limits and are off by default behind RUN_ENGINE_QUEUE_GATES_ENABLED. Mono-RevId: b8b2c98fae12cf9abb09b3f1349fd828d053ab70
1 parent 34c2d69 commit ea97581

147 files changed

Lines changed: 11297 additions & 3879 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.changeset/chat-session-concurrency.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
---
2+
"@trigger.dev/sdk": minor
3+
"@trigger.dev/core": minor
4+
---
5+
6+
Chat agents can now scope concurrency per session. Pass `concurrencyKey` (for example, your chat ID or tenant ID) and trigger-time named limits via `triggerConfig.concurrency` when starting a chat session, from `chat.createStartSessionAction`, the `AgentChat` client, or a handover. Keys are never defaulted, so a session without one shares the task's keyless pool.
7+
8+
```ts
9+
const start = chat.createStartSessionAction("support-chat", {
10+
triggerConfig: { concurrencyKey: user.id },
11+
});
12+
```

‎.changeset/task-concurrency.md‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
---
2+
"@trigger.dev/sdk": minor
3+
"@trigger.dev/core": minor
4+
"@trigger.dev/react-hooks": minor
5+
---
6+
7+
Control a task's concurrency with the new `concurrency` option, and share limits across tasks with named concurrency limits. An inline shape caps the task itself; `concurrencyLimit()` declares a limit any task can hold (up to two named limits per task), and a trigger call can switch a run's named limits with its own `concurrency` option.
8+
9+
```ts
10+
import { concurrencyLimit, task } from "@trigger.dev/sdk";
11+
12+
export const openaiLimit = concurrencyLimit({ name: "openai", total: 25 });
13+
14+
export const generateSummary = task({
15+
id: "generate-summary",
16+
concurrency: [{ perKey: 1, total: 5 }, openaiLimit],
17+
run: async (payload) => {},
18+
});
19+
```
20+
21+
`perKey` caps each `concurrencyKey` pool and `total` caps across everything, keys or not. The queue-level `concurrencyLimit` option keeps working unchanged and is deprecated in favor of `concurrency`. Enforcement happens server-side; servers without support accept the option but do not enforce it yet.
22+
23+
Manage limits at runtime with the new `concurrencyLimits` namespace: `list()` and `retrieve(name)` report each limit's bounds plus its live `running` and `queued` counts, `override(name, { perKey, total })` changes only the given bounds (overriding `total` to `0` pauses the limit), and `reset(name)` restores the declared values.
24+
25+
Queue reads (`queues.list()` and `queues.retrieve()`) now report a `version` that discriminates the shape: `V1` queues keep today's fields (their own `concurrencyLimit` and its override state), while `V2` queues (tasks declared with `concurrency`) carry no queue-level concurrency, since their limits are read and overridden through `concurrencyLimits` (a task's inline limit under its derived `task/<task-id>` name). Existing reads keep compiling: a `V2` queue reports `concurrencyLimit` as null and `concurrency` as undefined.

‎apps/webapp/app/components/billing/OrgBanner.tsx‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,10 @@ import { useOptionalProject, useProject } from "~/hooks/useProject";
2121
import { useShowSelfServe } from "~/hooks/useShowSelfServe";
2222
import { useCurrentPlan } from "~/routes/_app.orgs.$organizationSlug/route";
2323
import {
24+
concurrencyPath,
2425
organizationProjectsPath,
2526
v3BillingLimitsPath,
2627
v3BillingPath,
27-
v3QueuesPath,
2828
} from "~/utils/pathBuilder";
2929
import { ENVIRONMENT_PAUSE_SOURCE_BILLING_LIMIT } from "~/utils/environmentPauseSource";
3030

@@ -71,7 +71,7 @@ export function OrgBanner() {
7171
showSelfServe,
7272
});
7373

74-
const hideQueuesButton = location.pathname.endsWith("/queues");
74+
const hideConcurrencyButton = location.pathname.endsWith("/concurrency");
7575
const hideBillingLimitBanner = location.pathname.endsWith("/settings/billing-limits");
7676

7777
switch (bannerKind) {
@@ -89,7 +89,7 @@ export function OrgBanner() {
8989
return isArchived ? (
9090
<ArchivedEnvironmentBanner />
9191
) : (
92-
<PausedEnvironmentBanner hideButton={hideQueuesButton} />
92+
<PausedEnvironmentBanner hideButton={hideConcurrencyButton} />
9393
);
9494
default:
9595
return null;
@@ -303,7 +303,7 @@ function PausedEnvironmentBanner({ hideButton }: { hideButton: boolean }) {
303303
hideButton ? undefined : (
304304
<LinkButton
305305
variant="tertiary/small"
306-
to={v3QueuesPath(organization, project, environment)}
306+
to={concurrencyPath(organization, project, environment)}
307307
>
308308
Manage
309309
</LinkButton>

‎apps/webapp/app/components/dashboard-agent/DashboardAgentPanel.tsx‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ import {
5252
} from "./unread-counts";
5353
import { AgentPanelColumn, type DashboardAgentMode, type DragHandleProps } from "./panel-layout";
5454
import { markerAfterActiveChat, markerAfterActivity } from "./thinking-marker";
55-
import { concurrencyPath } from "~/utils/pathBuilder";
55+
import { concurrencyLimitsPath } from "~/utils/pathBuilder";
5656
import { scopeMatchesPath, sessionPathFor } from "./agent-scope";
5757

5858
function serializePageContext(pageContext: AgentPageContext): string | undefined {
@@ -149,7 +149,7 @@ export function DashboardAgentPanel({
149149
const entityId = agentPageEntityId(pageContext, location.pathname);
150150

151151
const pagePaths = useMemo<Record<string, string>>(
152-
() => ({ raise_env_limit: concurrencyPath(organization, project, environment) }),
152+
() => ({ raise_env_limit: concurrencyLimitsPath(organization, project, environment) }),
153153
[organization, project, environment]
154154
);
155155

‎apps/webapp/app/components/dashboard-agent/page-label.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ const KIND_LABELS: Record<Exclude<AgentPage["kind"], "other">, string> = {
2424
alerts: "Alerts",
2525
apikeys: "API keys",
2626
envvars: "Environment variables",
27-
concurrency: "Concurrency",
27+
concurrency: "Concurrency limits",
2828
regions: "Regions",
2929
settings: "Settings",
3030
waitpoints: "Waitpoints",
@@ -50,6 +50,7 @@ const SECTION_LABELS: Record<string, string> = {
5050
"bulk-actions": "Bulk actions",
5151
branches: "Branches",
5252
concurrency: "Concurrency",
53+
"concurrency-limits": "Concurrency limits",
5354
dashboards: "Dashboards",
5455
deployments: "Deployments",
5556
"dev-branches": "Branches",

‎apps/webapp/app/components/navigation/favoritePages.tsx‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -223,7 +223,8 @@ const ENV_PAGE_META: Record<string, PageMeta> = {
223223
logs: { icon: "logs", name: "Logs" },
224224
errors: { icon: "errors", name: "Errors", singular: "Error" },
225225
query: { icon: "query", name: "Query" },
226-
queues: { icon: "queues", name: "Queues", singular: "Queue" },
226+
queues: { icon: "queues", name: "Concurrency" },
227+
concurrency: { icon: "queues", name: "Concurrency" },
227228
dashboards: { icon: "dashboards", name: "Dashboards", singular: "Dashboard" },
228229
deployments: { icon: "deployments", name: "Deploys", singular: "Deploy" },
229230
"environment-variables": { icon: "environment-variables", name: "Environment variables" },
@@ -234,7 +235,7 @@ const ENV_PAGE_META: Record<string, PageMeta> = {
234235
"bulk-actions": { icon: "bulk-actions", name: "Bulk actions", singular: "Bulk action" },
235236
apikeys: { icon: "apikeys", name: "API keys" },
236237
alerts: { icon: "alerts", name: "Alerts", singular: "Alert" },
237-
concurrency: { icon: "concurrency", name: "Concurrency" },
238+
"concurrency-limits": { icon: "concurrency", name: "Concurrency limits" },
238239
limits: { icon: "limits", name: "Limits" },
239240
schedules: { icon: "schedules", name: "Schedules", singular: "Schedule" },
240241
test: { icon: "test", name: "Test", singular: "Test" },

‎apps/webapp/app/components/navigation/sideMenuSections.tsx‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ import {
2323
type OrgForPath,
2424
type ProjectForPath,
2525
branchesPath,
26-
concurrencyPath,
26+
concurrencyLimitsPath,
2727
limitsPath,
2828
queryPath,
2929
regionsPath,
@@ -39,7 +39,7 @@ import {
3939
v3ProjectAlertsPath,
4040
v3ProjectSettingsIntegrationsPath,
4141
v3PromptsPath,
42-
v3QueuesPath,
42+
concurrencyPath,
4343
v3WaitpointTokensPath,
4444
} from "~/utils/pathBuilder";
4545
import { AlphaBadge, NewBadge } from "../FeatureBadges";
@@ -160,10 +160,10 @@ export function buildSideMenuSections({
160160
} satisfies SideMenuItemConfig,
161161
{
162162
id: "queues",
163-
name: "Queues",
163+
name: "Concurrency",
164164
icon: QueuesIcon,
165165
activeIconColor: "text-queues",
166-
to: v3QueuesPath(organization, project, environment),
166+
to: concurrencyPath(organization, project, environment),
167167
dataAction: "queues",
168168
} satisfies SideMenuItemConfig,
169169
{
@@ -269,10 +269,10 @@ export function buildSideMenuSections({
269269
? [
270270
{
271271
id: "concurrency",
272-
name: "Concurrency",
272+
name: "Concurrency limits",
273273
icon: ConcurrencyIcon,
274274
activeIconColor: "text-text-bright",
275-
to: concurrencyPath(organization, project, environment),
275+
to: concurrencyLimitsPath(organization, project, environment),
276276
dataAction: "concurrency",
277277
} satisfies SideMenuItemConfig,
278278
]

‎apps/webapp/app/components/primitives/charts/MetricChart.tsx‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,8 @@ type MetricChartProps = {
4141
* are config values that existed all along, so carry the first value backward instead.
4242
*/
4343
carryBackfill?: string[];
44+
/** Column that, when positive on a bucket, exempts it from the carryBackfill overwrite. */
45+
carryBackfillGuard?: string;
4446
/**
4547
* Line only. Recolour a series' stroke above a threshold with a gradient split (colour only
4648
* above the line). `value` sets a constant threshold; `valueFromSeries` reads a (roughly
@@ -78,13 +80,22 @@ export function MetricChart({
7880
timeRange,
7981
warningOverlay,
8082
carryBackfill,
83+
carryBackfillGuard,
8184
thresholdStroke,
8285
onHasDataChange,
8386
sampleCountColumn,
8487
}: MetricChartProps) {
8588
const { points, xKind: resolvedXKind } = useMemo(
86-
() => buildMetricPoints(rows, { series, xColumn, xKind, carryBackfill, sampleCountColumn }),
87-
[rows, series, xColumn, xKind, carryBackfill, sampleCountColumn]
89+
() =>
90+
buildMetricPoints(rows, {
91+
series,
92+
xColumn,
93+
xKind,
94+
carryBackfill,
95+
carryBackfillGuard,
96+
sampleCountColumn,
97+
}),
98+
[rows, series, xColumn, xKind, carryBackfill, carryBackfillGuard, sampleCountColumn]
8899
);
89100

90101
const chartConfig = useMemo(() => {

‎apps/webapp/app/components/primitives/charts/metricPoints.ts‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,12 +58,22 @@ export type BuildMetricPointsOptions = {
5858
/** Defaults to `time` when every x is a datetime or epoch number, `category` otherwise. */
5959
xKind?: MetricXKind;
6060
carryBackfill?: string[];
61+
/** Column that, when positive on a bucket, exempts it from the carryBackfill overwrite:
62+
* the bucket has a real sampled value, so an earlier bucket's carry must not clobber it. */
63+
carryBackfillGuard?: string;
6164
sampleCountColumn?: string;
6265
};
6366

6467
export function buildMetricPoints(
6568
rows: MetricChartRow[],
66-
{ series, xColumn = "t", xKind, carryBackfill, sampleCountColumn }: BuildMetricPointsOptions
69+
{
70+
series,
71+
xColumn = "t",
72+
xKind,
73+
carryBackfill,
74+
carryBackfillGuard,
75+
sampleCountColumn,
76+
}: BuildMetricPointsOptions
6777
): { points: MetricPoint[]; xKind: MetricXKind } {
6878
// Rows built by seriesFromRows already carry the coordinate under the reserved key.
6979
const xOf = (row: MetricChartRow) => (METRIC_X_KEY in row ? row[METRIC_X_KEY] : row[xColumn]);
@@ -80,6 +90,9 @@ export function buildMetricPoints(
8090
// A bucket with no value for this series is a gap, not a zero.
8191
point[s.key] = hasSamples && value != null ? toNumber(value) : null;
8292
}
93+
if (carryBackfillGuard) {
94+
point[carryBackfillGuard] = toNumber(r[carryBackfillGuard]);
95+
}
8396
// Set last, so a series key colliding with the reserved key can't displace the coordinate.
8497
point[METRIC_X_KEY] = kind === "time" ? (timeValueMs(xOf(r)) ?? NaN) : String(xOf(r) ?? "");
8598
return point;
@@ -99,7 +112,10 @@ export function buildMetricPoints(
99112
const first = points.findIndex((p) => toNumber(p[key]) > 0);
100113
if (first > 0) {
101114
const value = points[first]![key]!;
102-
for (let i = 0; i < first; i++) points[i]![key] = value;
115+
for (let i = 0; i < first; i++) {
116+
if (carryBackfillGuard && toNumber(points[i]![carryBackfillGuard]) > 0) continue;
117+
points[i]![key] = value;
118+
}
103119
}
104120
}
105121
}

0 commit comments

Comments
 (0)