Skip to content

Commit 1a0d753

Browse files
matt-aitkenmatt-aitken
committed
feat(webapp,run-engine,core,clickhouse): surface total concurrency in metrics and dashboard
Queues with a totalConcurrencyLimit now report how they use it. The gauge pipeline emits total running and the stored cap, ClickHouse aggregates them into the queue metrics tiers, the queues list gets a Total column, the queue detail page charts total running against the cap, and the per-key table shows each key's effective limit including per-key overrides. Queue retrieve and list API responses include the same totals.
1 parent 5110870 commit 1a0d753

20 files changed

Lines changed: 400 additions & 12 deletions

File tree

‎.changeset/queue-total-concurrency-stats.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@trigger.dev/core": patch
3+
---
4+
5+
Queue retrieve and list API responses now report total concurrency usage. When a queue has a `totalConcurrencyLimit`, `concurrency.total` includes the effective cap, the declared base, any active override, and how many runs are in flight across all concurrency keys.

‎apps/webapp/app/presenters/v3/QueueListPresenter.server.ts‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,10 @@ import { engine } from "~/v3/runEngine.server";
99
import { BasePresenter } from "./basePresenter.server";
1010
import { toQueueItem } from "./QueueRetrievePresenter.server";
1111

12-
type QueueListEngine = Pick<RunEngine, "lengthOfQueues" | "currentConcurrencyOfQueues">;
12+
type QueueListEngine = Pick<
13+
RunEngine,
14+
"lengthOfQueues" | "currentConcurrencyOfQueues" | "totalConcurrencyOfQueues"
15+
>;
1316

1417
export const QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE = 25;
1518
const MAX_ITEMS_PER_PAGE = 100;
@@ -34,6 +37,9 @@ const queueListSelect = {
3437
concurrencyLimitOverriddenAt: true,
3538
concurrencyLimitOverriddenBy: true,
3639
concurrencyLimitOverridePercent: true,
40+
totalConcurrencyLimit: true,
41+
totalConcurrencyLimitBase: true,
42+
totalConcurrencyLimitOverriddenAt: true,
3743
type: true,
3844
paused: true,
3945
} satisfies Prisma.TaskQueueSelect;
@@ -333,11 +339,15 @@ export class QueueListPresenter extends BasePresenter {
333339
concurrencyLimitOverriddenAt: Date | null;
334340
concurrencyLimitOverriddenBy: string | null;
335341
concurrencyLimitOverridePercent: Prisma.Decimal | null;
342+
totalConcurrencyLimit: number | null;
343+
totalConcurrencyLimitBase: number | null;
344+
totalConcurrencyLimitOverriddenAt: Date | null;
336345
type: TaskQueueType;
337346
paused: boolean;
338347
}[]
339348
): Promise<QueueListItem[]> {
340-
const [queuedByQueue, runningByQueue] = await Promise.all([
349+
const queuesWithTotalCap = queues.filter((q) => q.totalConcurrencyLimit !== null);
350+
const [queuedByQueue, runningByQueue, totalRunningByQueue] = await Promise.all([
341351
this.engineClient.lengthOfQueues(
342352
environment,
343353
queues.map((q) => q.name)
@@ -346,6 +356,12 @@ export class QueueListPresenter extends BasePresenter {
346356
environment,
347357
queues.map((q) => q.name)
348358
),
359+
queuesWithTotalCap.length > 0
360+
? this.engineClient.totalConcurrencyOfQueues(
361+
environment,
362+
queuesWithTotalCap.map((q) => q.name)
363+
)
364+
: Promise.resolve({} as Record<string, number>),
349365
]);
350366

351367
// Manually "join" the overridden users because there is no way to implement the relationship
@@ -373,6 +389,11 @@ export class QueueListPresenter extends BasePresenter {
373389
? (overriddenByMap.get(queue.concurrencyLimitOverriddenBy) ?? null)
374390
: null,
375391
paused: queue.paused,
392+
totalConcurrencyLimit: queue.totalConcurrencyLimit,
393+
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
394+
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
395+
totalRunning:
396+
queue.totalConcurrencyLimit !== null ? (totalRunningByQueue[queue.name] ?? 0) : null,
376397
}),
377398
// Prisma returns Decimal; the client only needs a plain number (null for absolute overrides).
378399
concurrencyLimitOverridePercent:

‎apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,7 @@ export class QueueRetrievePresenter extends BasePresenter {
9090
const results = await Promise.all([
9191
engine.lengthOfQueues(environment, [queue.name]),
9292
engine.currentConcurrencyOfQueues(environment, [queue.name]),
93+
engine.totalConcurrencyOfQueues(environment, [queue.name]),
9394
]);
9495

9596
// Transform queues to include running and queued counts
@@ -107,6 +108,11 @@ export class QueueRetrievePresenter extends BasePresenter {
107108
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null,
108109
concurrencyLimitOverriddenBy: queue.concurrencyLimitOverriddenBy ?? null,
109110
paused: queue.paused,
111+
totalConcurrencyLimit: queue.totalConcurrencyLimit ?? null,
112+
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase ?? null,
113+
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt ?? null,
114+
totalRunning:
115+
queue.totalConcurrencyLimit != null ? (results[2]?.[queue.name] ?? 0) : null,
110116
}),
111117
// The percent source-of-truth for percent-based overrides isn't part of the shared
112118
// `QueueItem` schema (that's a public contract), so we surface it as an extra field on
@@ -148,6 +154,10 @@ export function toQueueItem(data: {
148154
concurrencyLimitOverriddenAt: Date | null;
149155
concurrencyLimitOverriddenBy: User | null;
150156
paused: boolean;
157+
totalConcurrencyLimit?: number | null;
158+
totalConcurrencyLimitBase?: number | null;
159+
totalConcurrencyLimitOverriddenAt?: Date | null;
160+
totalRunning?: number | null;
151161
}): QueueItem & { releaseConcurrencyOnWaitpoint: boolean } {
152162
return {
153163
id: data.friendlyId,
@@ -164,6 +174,16 @@ export function toQueueItem(data: {
164174
override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null,
165175
overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy),
166176
overriddenAt: data.concurrencyLimitOverriddenAt,
177+
total:
178+
data.totalConcurrencyLimit !== undefined
179+
? {
180+
current: data.totalConcurrencyLimit,
181+
base: data.totalConcurrencyLimitBase ?? null,
182+
override: data.totalConcurrencyLimitOverriddenAt ? data.totalConcurrencyLimit : null,
183+
overriddenAt: data.totalConcurrencyLimitOverriddenAt ?? null,
184+
running: data.totalRunning ?? null,
185+
}
186+
: undefined,
167187
},
168188
// TODO: This needs to be removed but keeping this here for now to avoid breaking existing clients
169189
releaseConcurrencyOnWaitpoint: true,

‎apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -698,6 +698,13 @@ function QueuesWithMetricsView() {
698698
<TableHeaderCell alignment="right">Queued</TableHeaderCell>
699699
<TableHeaderCell alignment="right">Running</TableHeaderCell>
700700
<TableHeaderCell alignment="right">Limit</TableHeaderCell>
701+
<TableHeaderCell
702+
alignment="right"
703+
disableTooltipHoverableContent
704+
tooltip="Runs in flight across all concurrency keys, against the queue's total concurrency limit. Only queues with a totalConcurrencyLimit show a value."
705+
>
706+
Total
707+
</TableHeaderCell>
701708
<TableHeaderCell
702709
alignment="right"
703710
tooltipContentClassName="max-w-max"
@@ -866,6 +873,29 @@ function QueuesWithMetricsView() {
866873
limit
867874
)}
868875
</TableCell>
876+
<TableCell
877+
to={queueDetailPath}
878+
alignment="right"
879+
actionClassName="pl-16 tabular-nums"
880+
className={cn(
881+
"w-[1%]",
882+
queue.paused ? "opacity-50" : undefined,
883+
queue.concurrency?.total?.current != null &&
884+
(queue.concurrency.total.running ?? 0) >=
885+
Math.min(
886+
queue.concurrency.total.current,
887+
environment.concurrencyLimit
888+
) &&
889+
"text-warning"
890+
)}
891+
>
892+
{queue.concurrency?.total?.current != null
893+
? `${queue.concurrency.total.running ?? 0}/${Math.min(
894+
queue.concurrency.total.current,
895+
environment.concurrencyLimit
896+
)}`
897+
: "–"}
898+
</TableCell>
869899
<TableCell
870900
to={queueDetailPath}
871901
alignment="right"
@@ -1012,7 +1042,7 @@ function QueuesWithMetricsView() {
10121042
})
10131043
) : (
10141044
<TableRow>
1015-
<TableCell colSpan={9}>
1045+
<TableCell colSpan={10}>
10161046
<div className="grid place-items-center py-6 text-text-dimmed">
10171047
<Paragraph>
10181048
{hasFilters ? "No queues found matching your filters" : "No queues found"}

‎apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx‎

Lines changed: 65 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -392,7 +392,12 @@ export default function Page() {
392392
<ConcurrencyKeysBlankState />
393393
)
394394
) : (
395-
<OverviewCharts ids={ids} timeRange={timeRange} queueName={fullName} />
395+
<OverviewCharts
396+
ids={ids}
397+
timeRange={timeRange}
398+
queueName={fullName}
399+
hasTotalLimit={queue.concurrency?.total?.current != null}
400+
/>
396401
)}
397402
</MetricsLayout.Content>
398403

@@ -402,7 +407,13 @@ export default function Page() {
402407
{view === "keys" && hasKeys ? (
403408
<>
404409
<MetricsLayout.Content>
405-
<KeyStatsTable ids={ids} timeRange={timeRange} queueName={fullName} />
410+
<KeyStatsTable
411+
ids={ids}
412+
timeRange={timeRange}
413+
queueName={fullName}
414+
defaultKeyLimit={queue.concurrencyLimit ?? environmentConcurrencyLimit}
415+
envLimit={environmentConcurrencyLimit}
416+
/>
406417
</MetricsLayout.Content>
407418
{selectedKey ? (
408419
<MetricsLayout.Content inset>
@@ -436,10 +447,12 @@ function OverviewCharts({
436447
ids,
437448
timeRange,
438449
queueName,
450+
hasTotalLimit,
439451
}: {
440452
ids: Ids;
441453
timeRange: TimeRangeParams;
442454
queueName: string;
455+
hasTotalLimit: boolean;
443456
}) {
444457
const zoomToTimeFilter = useZoomToTimeFilter();
445458
return (
@@ -479,6 +492,37 @@ function OverviewCharts({
479492
// leading zeros so the reference line doesn't start with a false 0→limit step.
480493
carryBackfill={["limit"]}
481494
/>
495+
{hasTotalLimit ? (
496+
<QueueDetailChartCard
497+
title="Total concurrency"
498+
info={
499+
<>
500+
Runs in flight across ALL concurrency keys (
501+
<ColorSwatch color={COLORS.running} />) versus the queue's total limit (
502+
<ColorSwatch color={COLORS.limit} />
503+
).
504+
</>
505+
}
506+
showLegend
507+
className="aspect-[2/1]"
508+
query={`SELECT timeBucket() AS t, max(max_total_running) AS running, least(max(max_total_limit), max(max_env_limit)) AS cap\nFROM queue_metrics\nGROUP BY t\nORDER BY t`}
509+
fillGaps
510+
minBucketSeconds={SYNCED_CHART_MIN_BUCKET_SECONDS}
511+
ids={ids}
512+
timeRange={timeRange}
513+
queueName={queueName}
514+
series={[
515+
{ key: "cap", label: "Total limit", color: COLORS.limit },
516+
{ key: "running", label: "Running", color: COLORS.running },
517+
]}
518+
thresholdStroke={{
519+
series: "running",
520+
valueFromSeries: "cap",
521+
aboveColor: "var(--color-warning)",
522+
}}
523+
carryBackfill={["cap"]}
524+
/>
525+
) : null}
482526
<QueueDetailChartCard
483527
title="Queue depth"
484528
info="How many runs are waiting in this queue over time."
@@ -922,10 +966,15 @@ function KeyStatsTable({
922966
ids,
923967
timeRange,
924968
queueName,
969+
defaultKeyLimit,
970+
envLimit,
925971
}: {
926972
ids: Ids;
927973
timeRange: TimeRangeParams;
928974
queueName: string;
975+
/** The limit a key inherits when it has no override (the queue's limit, else the env limit). */
976+
defaultKeyLimit: number;
977+
envLimit: number;
929978
}) {
930979
const { value, replace, del } = useSearchParams();
931980
const selectedKey = value("key");
@@ -968,6 +1017,12 @@ function KeyStatsTable({
9681017
<TableHeaderCell>Key</TableHeaderCell>
9691018
<TableHeaderCell alignment="right">Queued now</TableHeaderCell>
9701019
<TableHeaderCell alignment="right">Running now</TableHeaderCell>
1020+
<TableHeaderCell
1021+
alignment="right"
1022+
tooltip="The key's concurrency limit. Keys inherit the queue's limit unless a per-key override is set via the API."
1023+
>
1024+
Limit
1025+
</TableHeaderCell>
9711026
<TableHeaderCell alignment="right">Oldest wait</TableHeaderCell>
9721027
<TableHeaderCell alignment="right">Started</TableHeaderCell>
9731028
<TableHeaderCell alignment="right">Peak backlog</TableHeaderCell>
@@ -976,11 +1031,11 @@ function KeyStatsTable({
9761031
</TableHeader>
9771032
<TableBody>
9781033
{showLoading ? (
979-
<TableBlankRow colSpan={7} className="text-text-dimmed">
1034+
<TableBlankRow colSpan={8} className="text-text-dimmed">
9801035
Loading…
9811036
</TableBlankRow>
9821037
) : rows.length === 0 ? (
983-
<TableBlankRow colSpan={7} className="text-text-dimmed">
1038+
<TableBlankRow colSpan={8} className="text-text-dimmed">
9841039
{search ? `No keys match “${search}”` : "No concurrency keys"}
9851040
</TableBlankRow>
9861041
) : (
@@ -994,6 +1049,12 @@ function KeyStatsTable({
9941049
<TableCell>{row.key}</TableCell>
9951050
<TableCell alignment="right">{row.queued.toLocaleString()}</TableCell>
9961051
<TableCell alignment="right">{row.running.toLocaleString()}</TableCell>
1052+
<TableCell
1053+
alignment="right"
1054+
className={row.limitOverride !== null ? undefined : "text-text-dimmed"}
1055+
>
1056+
{Math.min(row.limitOverride ?? defaultKeyLimit, envLimit).toLocaleString()}
1057+
</TableCell>
9971058
<TableCell alignment="right">
9981059
{row.oldestWaitMs === null ? "–" : formatWaitMs(row.oldestWaitMs)}
9991060
</TableCell>

‎apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,9 @@ const route = createActionApiRoute(
5252
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
5353
concurrencyLimitOverriddenBy: null,
5454
paused: queue.paused,
55+
totalConcurrencyLimit: queue.totalConcurrencyLimit,
56+
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
57+
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
5558
}),
5659
{ status: 200 }
5760
);

‎apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,9 @@ const route = createActionApiRoute(
4646
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
4747
concurrencyLimitOverriddenBy: null,
4848
paused: queue.paused,
49+
totalConcurrencyLimit: queue.totalConcurrencyLimit,
50+
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
51+
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
4952
}),
5053
{ status: 200 }
5154
);

‎apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,9 @@ const route = createActionApiRoute(
6161
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
6262
concurrencyLimitOverriddenBy: null,
6363
paused: queue.paused,
64+
totalConcurrencyLimit: queue.totalConcurrencyLimit,
65+
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
66+
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
6467
}),
6568
{ status: 200 }
6669
);

‎apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,9 @@ const route = createActionApiRoute(
4343
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
4444
concurrencyLimitOverriddenBy: null,
4545
paused: queue.paused,
46+
totalConcurrencyLimit: queue.totalConcurrencyLimit,
47+
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
48+
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
4649
}),
4750
{ status: 200 }
4851
);

‎apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.total.override.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,9 @@ const route = createActionApiRoute(
4646
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
4747
concurrencyLimitOverriddenBy: null,
4848
paused: queue.paused,
49+
totalConcurrencyLimit: queue.totalConcurrencyLimit,
50+
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
51+
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
4952
}),
5053
{ status: 200 }
5154
);

0 commit comments

Comments
 (0)