Skip to content

Commit f493d0a

Browse files
committed
feat(core,webapp): version-discriminated queue reads, combined never ships
Queue reads now report a version that discriminates the shape. V1 queues keep today's fields exactly (their own concurrencyLimit and its override state) so nothing existing breaks; V2 queues are only the line runs wait in and carry no queue-level concurrency, since their limits are read and overridden through concurrencyLimits (concurrencyLimit stays as null so older clients keep parsing). The combined block is gone from the public schema entirely. The public queue routes now serialize through the schema, so dashboard-only extras can never leak onto the wire; the dashboard reads configured bounds from a presenter-attached limits object instead of the public fields.
1 parent 177b8ff commit f493d0a

13 files changed

Lines changed: 279 additions & 176 deletions

File tree

‎.changeset/task-concurrency.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,3 +21,5 @@ export const generateSummary = task({
2121
`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.
2222

2323
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`.

‎apps/webapp/app/components/queues/QueueControls.tsx‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
import { AdjustmentsHorizontalIcon, PauseIcon, PlayIcon } from "@heroicons/react/20/solid";
22
import { DialogClose } from "@radix-ui/react-dialog";
33
import { Form, useNavigation } from "@remix-run/react";
4-
import type { QueueItem } from "@trigger.dev/core/v3/schemas";
54
import { useEffect, useState } from "react";
65
import { cn } from "~/utils/cn";
6+
import type { QueueLimits } from "~/components/queues/queue-limits";
77
import { Button, type ButtonVariant } from "~/components/primitives/Buttons";
88
import { Dialog, DialogContent, DialogHeader, DialogTrigger } from "~/components/primitives/Dialog";
99
import { FormButtons } from "~/components/primitives/FormButtons";
@@ -159,7 +159,12 @@ export function QueueOverrideConcurrencyButton({
159159
environmentConcurrencyLimit,
160160
trigger,
161161
}: {
162-
queue: QueueItem & { concurrencyLimitOverridePercent: number | null };
162+
queue: {
163+
id: string;
164+
name: string;
165+
limits: QueueLimits;
166+
concurrencyLimitOverridePercent: number | null;
167+
};
163168
environmentConcurrencyLimit: number;
164169
/** How to render the dialog trigger. "menu-item" (default) is a PopoverMenuItem for row menus;
165170
* "button" is a standalone labeled button; "icon" is an icon-only button with the label in a
@@ -172,14 +177,14 @@ export function QueueOverrideConcurrencyButton({
172177
queue.concurrencyLimitOverridePercent !== null ? "percent" : "absolute"
173178
);
174179
const [concurrencyLimit, setConcurrencyLimit] = useState<string>(
175-
queue.concurrencyLimit?.toString() ?? environmentConcurrencyLimit.toString()
180+
queue.limits.perKey.current?.toString() ?? environmentConcurrencyLimit.toString()
176181
);
177182
const [percent, setPercent] = useState<string>(
178183
queue.concurrencyLimitOverridePercent?.toString() ?? "100"
179184
);
180185

181-
const isOverridden = !!queue.concurrency?.overriddenAt;
182-
const currentLimit = queue.concurrencyLimit ?? environmentConcurrencyLimit;
186+
const isOverridden = !!queue.limits.perKey.overriddenAt;
187+
const currentLimit = queue.limits.perKey.current ?? environmentConcurrencyLimit;
183188

184189
useEffect(() => {
185190
if (navigation.state === "loading" || navigation.state === "idle") {
@@ -277,18 +282,18 @@ export function QueueOverrideConcurrencyButton({
277282
{isOverridden ? (
278283
<Paragraph variant="small">
279284
This queue's concurrency limit is currently overridden to {currentLimit}.
280-
{typeof queue.concurrency?.base === "number" &&
281-
` The original limit set in code was ${queue.concurrency.base}.`}{" "}
285+
{typeof queue.limits.perKey.base === "number" &&
286+
` The original limit set in code was ${queue.limits.perKey.base}.`}{" "}
282287
You can update the override or remove it to restore the{" "}
283-
{typeof queue.concurrency?.base === "number"
288+
{typeof queue.limits.perKey.base === "number"
284289
? "limit set in code"
285290
: "environment concurrency limit"}
286291
.
287292
</Paragraph>
288293
) : (
289294
<Paragraph variant="small">
290295
Override this queue's concurrency limit. The current limit is {currentLimit}, which is
291-
set {queue.concurrencyLimit !== null ? "in code" : "by the environment"}.
296+
set {queue.limits.perKey.current !== null ? "in code" : "by the environment"}.
292297
</Paragraph>
293298
)}
294299
<Form method="post" onSubmit={() => setIsOpen(false)} className="space-y-3">
Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/**
2+
* Dashboard-only view of a queue row's configured bounds. The public QueueItem
3+
* shape is version-discriminated (V2 queues carry no queue-level concurrency),
4+
* but the dashboard shows configured limits for every row, so presenters attach
5+
* this alongside the public fields.
6+
*/
7+
export type QueueLimitBound = {
8+
/** The enforced value right now (declared, or the override when one is active) */
9+
current: number | null;
10+
/** The declared value an override reverts to */
11+
base: number | null;
12+
/** The overridden value, when an override is active */
13+
override: number | null;
14+
overriddenAt: Date | null;
15+
/** Display name of who applied the override (null when via the API) */
16+
overriddenBy: string | null;
17+
};
18+
19+
export type QueueTotalBound = {
20+
current: number;
21+
base: number | null;
22+
override: number | null;
23+
overriddenAt: Date | null;
24+
/** Runs in flight across every pool of the row (keyed and keyless) */
25+
running: number | null;
26+
};
27+
28+
export type QueueLimits = {
29+
perKey: QueueLimitBound;
30+
/** Null when the row declares no total bound */
31+
total: QueueTotalBound | null;
32+
};

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

Lines changed: 44 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,8 @@ import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstan
77
import { logger } from "~/services/logger.server";
88
import { engine } from "~/v3/runEngine.server";
99
import { BasePresenter } from "./basePresenter.server";
10-
import { toQueueItem } from "./QueueRetrievePresenter.server";
10+
import { toQueueItem, toQueueLimits } from "./QueueRetrievePresenter.server";
11+
import type { QueueLimits } from "~/components/queues/queue-limits";
1112

1213
type QueueListEngine = Pick<
1314
RunEngine,
@@ -59,6 +60,8 @@ type QueueListItem = ReturnType<typeof toQueueItem> & {
5960
kind: "queue" | "limit";
6061
/** V2 rows hold the new perKey/total vocabulary in their limit columns. */
6162
concurrencyVersion: "V1" | "V2";
63+
/** The row's configured bounds, dashboard-only (the public shape hides them on V2). */
64+
limits: QueueLimits;
6265
};
6366

6467
type QueueListPagination =
@@ -427,39 +430,45 @@ export class QueueListPresenter extends BasePresenter {
427430

428431
const overriddenByMap = new Map(overriddenByUsers.map((u) => [u.id, u]));
429432

430-
return queues.map((queue) => ({
431-
...toQueueItem({
432-
friendlyId: queue.friendlyId,
433-
name: queue.name,
434-
type: queue.type,
435-
running:
436-
queue.role === "LIMIT"
437-
? (totalRunningByQueue[queue.name] ?? 0)
438-
: (runningByQueue[queue.name] ?? 0),
439-
queued:
440-
queue.role === "LIMIT"
441-
? (gateQueuedByQueue[queue.name] ?? 0)
442-
: (queuedByQueue[queue.name] ?? 0),
443-
concurrencyLimit: queue.concurrencyLimit ?? null,
444-
concurrencyLimitBase: queue.concurrencyLimitBase ?? null,
445-
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null,
446-
concurrencyLimitOverriddenBy: queue.concurrencyLimitOverriddenBy
447-
? (overriddenByMap.get(queue.concurrencyLimitOverriddenBy) ?? null)
448-
: null,
449-
paused: queue.paused,
450-
totalConcurrencyLimit: queue.totalConcurrencyLimit,
451-
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
452-
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
453-
totalRunning:
454-
queue.totalConcurrencyLimit !== null ? (totalRunningByQueue[queue.name] ?? 0) : null,
455-
}),
456-
// Prisma returns Decimal; the client only needs a plain number (null for absolute overrides).
457-
concurrencyLimitOverridePercent:
458-
queue.concurrencyLimitOverridePercent !== null
459-
? Number(queue.concurrencyLimitOverridePercent)
460-
: null,
461-
kind: queue.role === "LIMIT" ? ("limit" as const) : ("queue" as const),
462-
concurrencyVersion: queue.concurrencyVersion,
463-
}));
433+
return queues.map((queue) => {
434+
const overriddenByUser = queue.concurrencyLimitOverriddenBy
435+
? (overriddenByMap.get(queue.concurrencyLimitOverriddenBy) ?? null)
436+
: null;
437+
return {
438+
...toQueueItem({
439+
friendlyId: queue.friendlyId,
440+
name: queue.name,
441+
type: queue.type,
442+
version: queue.concurrencyVersion,
443+
running:
444+
queue.role === "LIMIT"
445+
? (totalRunningByQueue[queue.name] ?? 0)
446+
: (runningByQueue[queue.name] ?? 0),
447+
queued:
448+
queue.role === "LIMIT"
449+
? (gateQueuedByQueue[queue.name] ?? 0)
450+
: (queuedByQueue[queue.name] ?? 0),
451+
concurrencyLimit: queue.concurrencyLimit ?? null,
452+
concurrencyLimitBase: queue.concurrencyLimitBase ?? null,
453+
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null,
454+
concurrencyLimitOverriddenBy: overriddenByUser,
455+
paused: queue.paused,
456+
}),
457+
// Prisma returns Decimal; the client only needs a plain number (null for absolute overrides).
458+
concurrencyLimitOverridePercent:
459+
queue.concurrencyLimitOverridePercent !== null
460+
? Number(queue.concurrencyLimitOverridePercent)
461+
: null,
462+
kind: queue.role === "LIMIT" ? ("limit" as const) : ("queue" as const),
463+
concurrencyVersion: queue.concurrencyVersion,
464+
limits: toQueueLimits(queue, {
465+
totalRunning:
466+
queue.totalConcurrencyLimit !== null ? (totalRunningByQueue[queue.name] ?? 0) : null,
467+
overriddenByName: overriddenByUser
468+
? (overriddenByUser.displayName ?? overriddenByUser.name ?? null)
469+
: null,
470+
}),
471+
};
472+
});
464473
}
465474
}

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

Lines changed: 90 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,11 @@
11
import { assertExhaustive } from "@trigger.dev/core";
2-
import { type Prettify, type QueueItem, type RetrieveQueueParam } from "@trigger.dev/core/v3";
2+
import {
3+
QueueItem as QueueItemSchema,
4+
type Prettify,
5+
type QueueItem,
6+
type RetrieveQueueParam,
7+
} from "@trigger.dev/core/v3";
8+
import type { QueueLimits } from "~/components/queues/queue-limits";
39
import {
410
type PrismaClientOrTransaction,
511
type TaskQueue,
@@ -99,40 +105,91 @@ export class QueueRetrievePresenter extends BasePresenter {
99105
: undefined,
100106
]);
101107

102-
// Transform queues to include running and queued counts
108+
/** The returned queue = the public QueueItem fields plus dashboard extras
109+
* (percent override source, configured bounds); the public API routes strip
110+
* the extras via `toPublicQueueItem`. Prisma returns Decimal for the
111+
* percent; the client only needs a plain number (null for absolute). */
103112
return {
104113
success: true as const,
105114
queue: {
106115
...toQueueItem({
107116
friendlyId: queue.friendlyId,
108117
name: queue.name,
109118
type: queue.type,
119+
version: queue.concurrencyVersion,
110120
running: results[1]?.[queue.name] ?? 0,
111121
queued: results[0]?.[queue.name] ?? 0,
112122
concurrencyLimit: queue.concurrencyLimit ?? null,
113123
concurrencyLimitBase: queue.concurrencyLimitBase ?? null,
114124
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null,
115125
concurrencyLimitOverriddenBy: queue.concurrencyLimitOverriddenBy ?? null,
116126
paused: queue.paused,
117-
totalConcurrencyLimit: queue.totalConcurrencyLimit ?? null,
118-
totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase ?? null,
119-
totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt ?? null,
120-
totalRunning:
121-
queue.totalConcurrencyLimit != null ? (results[2]?.[queue.name] ?? 0) : null,
122127
}),
123-
// The percent source-of-truth for percent-based overrides isn't part of the shared
124-
// `QueueItem` schema (that's a public contract), so we surface it as an extra field on
125-
// the returned queue — mirroring QueueListPresenter. Prisma returns Decimal; the client
126-
// only needs a plain number (null for absolute overrides).
127128
concurrencyLimitOverridePercent:
128129
queue.concurrencyLimitOverridePercent !== null
129130
? Number(queue.concurrencyLimitOverridePercent)
130131
: null,
132+
limits: toQueueLimits(queue, {
133+
totalRunning:
134+
queue.totalConcurrencyLimit != null ? (results[2]?.[queue.name] ?? 0) : null,
135+
overriddenByName: toQueueConcurrencyOverriddenBy(
136+
queue.concurrencyLimitOverriddenBy ?? null
137+
),
138+
}),
131139
},
132140
};
133141
}
134142
}
135143

144+
/**
145+
* The dashboard-only configured bounds of a row, independent of the public
146+
* shape's version discrimination (V2 queues expose no queue-level concurrency
147+
* publicly, but the dashboard still shows what's configured).
148+
*/
149+
export function toQueueLimits(
150+
row: {
151+
concurrencyLimit: number | null;
152+
concurrencyLimitBase: number | null;
153+
concurrencyLimitOverriddenAt: Date | null;
154+
totalConcurrencyLimit: number | null;
155+
totalConcurrencyLimitBase: number | null;
156+
totalConcurrencyLimitOverriddenAt: Date | null;
157+
},
158+
extras: { totalRunning: number | null; overriddenByName: string | null }
159+
): QueueLimits {
160+
return {
161+
perKey: {
162+
current: row.concurrencyLimit,
163+
base: row.concurrencyLimitOverriddenAt ? row.concurrencyLimitBase : row.concurrencyLimit,
164+
override: row.concurrencyLimitOverriddenAt ? row.concurrencyLimit : null,
165+
overriddenAt: row.concurrencyLimitOverriddenAt,
166+
overriddenBy: extras.overriddenByName,
167+
},
168+
total:
169+
row.totalConcurrencyLimit !== null
170+
? {
171+
current: row.totalConcurrencyLimit,
172+
base: row.totalConcurrencyLimitOverriddenAt
173+
? row.totalConcurrencyLimitBase
174+
: row.totalConcurrencyLimit,
175+
override: row.totalConcurrencyLimitOverriddenAt ? row.totalConcurrencyLimit : null,
176+
overriddenAt: row.totalConcurrencyLimitOverriddenAt,
177+
running: extras.totalRunning,
178+
}
179+
: null,
180+
};
181+
}
182+
183+
/**
184+
* The public API wire shape: the `QueueItem` contract exactly (dashboard extras
185+
* stripped by the schema parse) plus the legacy field older clients require.
186+
*/
187+
export function toPublicQueueItem(item: QueueItem): QueueItem & {
188+
releaseConcurrencyOnWaitpoint: boolean;
189+
} {
190+
return { ...QueueItemSchema.parse(item), releaseConcurrencyOnWaitpoint: true };
191+
}
192+
136193
export function queueTypeFromType(type: TaskQueueType) {
137194
switch (type) {
138195
case "NAMED":
@@ -145,54 +202,56 @@ export function queueTypeFromType(type: TaskQueueType) {
145202
}
146203

147204
/**
148-
* Converts raw queue data into a standardized QueueItem format
149-
* @param data Raw queue data containing required queue properties
150-
* @returns A validated QueueItem object
205+
* Converts raw queue data into the public QueueItem shape. The queue's version
206+
* discriminates it: V1 carries the queue's own limit and override state; V2
207+
* carries neither (a V2 queue is only the line — concurrency lives on the task
208+
* `concurrency` option and the `concurrencyLimits` surface), with
209+
* `concurrencyLimit` kept as null so older clients keep parsing.
151210
*/
152211
export function toQueueItem(data: {
153212
friendlyId: string;
154213
name: string;
155214
type: TaskQueueType;
215+
version: "V1" | "V2";
156216
running: number;
157217
queued: number;
158218
concurrencyLimit: number | null;
159219
concurrencyLimitBase: number | null;
160220
concurrencyLimitOverriddenAt: Date | null;
161221
concurrencyLimitOverriddenBy: User | null;
162222
paused: boolean;
163-
totalConcurrencyLimit?: number | null;
164-
totalConcurrencyLimitBase?: number | null;
165-
totalConcurrencyLimitOverriddenAt?: Date | null;
166-
totalRunning?: number | null;
167223
}): QueueItem & { releaseConcurrencyOnWaitpoint: boolean } {
168-
return {
224+
const common = {
169225
id: data.friendlyId,
170226
//remove the task/ prefix if it exists
171227
name: data.name.replace(/^task\//, ""),
172228
type: queueTypeFromType(data.type),
173229
running: data.running,
174230
queued: data.queued,
175231
paused: data.paused,
232+
// TODO: This needs to be removed but keeping this here for now to avoid breaking existing clients
233+
releaseConcurrencyOnWaitpoint: true,
234+
};
235+
236+
if (data.version === "V2") {
237+
return {
238+
...common,
239+
version: "V2" as const,
240+
concurrencyLimit: null,
241+
};
242+
}
243+
244+
return {
245+
...common,
246+
version: "V1" as const,
176247
concurrencyLimit: data.concurrencyLimit,
177248
concurrency: {
178249
current: data.concurrencyLimit,
179250
base: data.concurrencyLimitBase,
180251
override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null,
181252
overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy),
182253
overriddenAt: data.concurrencyLimitOverriddenAt,
183-
combined:
184-
data.totalConcurrencyLimit !== undefined
185-
? {
186-
current: data.totalConcurrencyLimit,
187-
base: data.totalConcurrencyLimitBase ?? null,
188-
override: data.totalConcurrencyLimitOverriddenAt ? data.totalConcurrencyLimit : null,
189-
overriddenAt: data.totalConcurrencyLimitOverriddenAt ?? null,
190-
running: data.totalRunning ?? null,
191-
}
192-
: undefined,
193254
},
194-
// TODO: This needs to be removed but keeping this here for now to avoid breaking existing clients
195-
releaseConcurrencyOnWaitpoint: true,
196255
};
197256
}
198257

0 commit comments

Comments
 (0)