Skip to content

Commit ccf919e

Browse files
committed
feat(webapp): named concurrency limits appear on the queues page
The list interleaves LIMIT rows alongside queues (name order, shared pagination; the type filter stays queues-only): running is the limit's group cardinality, queued is its exact per-gate counter, the name strips the limit/ prefix with a Limit badge and its own icon, and V2 rows' Limit cell self-labels as per key or total. Limit rows open the detail page (their gauges already chart) and hide the queue-only pause and override actions.
1 parent c462da5 commit ccf919e

5 files changed

Lines changed: 319 additions & 153 deletions

File tree

‎apps/webapp/app/components/runs/v3/QueueName.tsx‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,49 @@
11
import { TasksIcon } from "~/assets/icons/TasksIcon";
2+
import { ConcurrencyIcon } from "~/assets/icons/ConcurrencyIcon";
23
import { SimpleTooltip } from "~/components/primitives/Tooltip";
34
import { cn } from "~/utils/cn";
45
import { RectangleStackIcon } from "@heroicons/react/20/solid";
56

7+
const LIMIT_PREFIX = "limit/";
8+
const TASK_PREFIX = "task/";
9+
610
export function QueueName({
711
name,
812
type,
13+
kind,
914
paused,
1015
className,
1116
}: {
1217
name: string;
1318
type: "task" | "custom";
19+
/** "limit" rows are named concurrency limits rather than queues. */
20+
kind?: "queue" | "limit";
1421
paused?: boolean;
1522
className?: string;
1623
}) {
24+
if (kind === "limit") {
25+
const displayName = name.startsWith(LIMIT_PREFIX) ? name.slice(LIMIT_PREFIX.length) : name;
26+
return (
27+
<span className={cn("flex items-center gap-1", className)}>
28+
<SimpleTooltip
29+
button={
30+
<ConcurrencyIcon
31+
className={cn("size-[1.125rem] text-amber-500", paused && "opacity-50")}
32+
/>
33+
}
34+
content={
35+
displayName.startsWith(TASK_PREFIX)
36+
? `This is the inline concurrency limit of your "${displayName.slice(
37+
TASK_PREFIX.length
38+
)}" task`
39+
: "This is a named concurrency limit declared in your code."
40+
}
41+
/>
42+
<span className={paused ? "opacity-50" : undefined}>{displayName}</span>
43+
</span>
44+
);
45+
}
46+
1747
return (
1848
<span className={cn("flex items-center gap-1", className)}>
1949
{type === "task" ? (

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

Lines changed: 59 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,10 @@ import { toQueueItem } from "./QueueRetrievePresenter.server";
1111

1212
type QueueListEngine = Pick<
1313
RunEngine,
14-
"lengthOfQueues" | "currentConcurrencyOfQueues" | "totalConcurrencyOfQueues"
14+
| "lengthOfQueues"
15+
| "currentConcurrencyOfQueues"
16+
| "totalConcurrencyOfQueues"
17+
| "gateQueuedCountOfQueues"
1518
>;
1619

1720
export const QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE = 25;
@@ -42,6 +45,8 @@ const queueListSelect = {
4245
totalConcurrencyLimitOverriddenAt: true,
4346
type: true,
4447
paused: true,
48+
role: true,
49+
concurrencyVersion: true,
4550
} satisfies Prisma.TaskQueueSelect;
4651

4752
type QueueListRow = Prisma.TaskQueueGetPayload<{ select: typeof queueListSelect }>;
@@ -50,6 +55,10 @@ type QueueListRow = Prisma.TaskQueueGetPayload<{ select: typeof queueListSelect
5055
// schema (that's a public contract), so we surface it as an extra field on the list item.
5156
type QueueListItem = ReturnType<typeof toQueueItem> & {
5257
concurrencyLimitOverridePercent: number | null;
58+
/** "queue" rows wait and order runs; "limit" rows are named concurrency limits. */
59+
kind: "queue" | "limit";
60+
/** V2 rows hold the new perKey/total vocabulary in their limit columns. */
61+
concurrencyVersion: "V1" | "V2";
5362
};
5463

5564
type QueueListPagination =
@@ -76,7 +85,9 @@ function buildQueueListWhere(
7685

7786
return {
7887
runtimeEnvironmentId: environmentId,
79-
role: "QUEUE" as const,
88+
/** The type filter names queue shapes, so applying it scopes the list to queue rows;
89+
* without it the list interleaves named limits alongside queues. */
90+
role: type ? ("QUEUE" as const) : { in: ["QUEUE" as const, "LIMIT" as const] },
8091
version: "V2",
8192
name: trimmedQuery
8293
? {
@@ -345,25 +356,44 @@ export class QueueListPresenter extends BasePresenter {
345356
totalConcurrencyLimitOverriddenAt: Date | null;
346357
type: TaskQueueType;
347358
paused: boolean;
359+
role: "QUEUE" | "LIMIT";
360+
concurrencyVersion: "V1" | "V2";
348361
}[]
349362
): Promise<QueueListItem[]> {
350-
const queuesWithTotalCap = queues.filter((q) => q.totalConcurrencyLimit !== null);
351-
const [queuedByQueue, runningByQueue, totalRunningByQueue] = await Promise.all([
352-
this.engineClient.lengthOfQueues(
353-
environment,
354-
queues.map((q) => q.name)
355-
),
356-
this.engineClient.currentConcurrencyOfQueues(
357-
environment,
358-
queues.map((q) => q.name)
359-
),
360-
queuesWithTotalCap.length > 0
361-
? this.engineClient.totalConcurrencyOfQueues(
362-
environment,
363-
queuesWithTotalCap.map((q) => q.name)
364-
)
365-
: Promise.resolve({} as Record<string, number>),
366-
]);
363+
const queueRows = queues.filter((q) => q.role === "QUEUE");
364+
const limitRows = queues.filter((q) => q.role === "LIMIT");
365+
/**
366+
* Queue rows read their zset length and home concurrency; limit rows read the
367+
* group set (every holder, keyed or keyless) as running and the per-gate queued
368+
* counter as queued. The group read also serves queue rows with a total cap.
369+
*/
370+
const rowsWithGroupRead = [
371+
...queueRows.filter((q) => q.totalConcurrencyLimit !== null),
372+
...limitRows,
373+
];
374+
const [queuedByQueue, runningByQueue, totalRunningByQueue, gateQueuedByQueue] =
375+
await Promise.all([
376+
this.engineClient.lengthOfQueues(
377+
environment,
378+
queueRows.map((q) => q.name)
379+
),
380+
this.engineClient.currentConcurrencyOfQueues(
381+
environment,
382+
queueRows.map((q) => q.name)
383+
),
384+
rowsWithGroupRead.length > 0
385+
? this.engineClient.totalConcurrencyOfQueues(
386+
environment,
387+
rowsWithGroupRead.map((q) => q.name)
388+
)
389+
: Promise.resolve({} as Record<string, number>),
390+
limitRows.length > 0
391+
? this.engineClient.gateQueuedCountOfQueues(
392+
environment,
393+
limitRows.map((q) => q.name)
394+
)
395+
: Promise.resolve({} as Record<string, number>),
396+
]);
367397

368398
// Manually "join" the overridden users because there is no way to implement the relationship
369399
// in prisma without adding a foreign key constraint
@@ -381,8 +411,14 @@ export class QueueListPresenter extends BasePresenter {
381411
friendlyId: queue.friendlyId,
382412
name: queue.name,
383413
type: queue.type,
384-
running: runningByQueue[queue.name] ?? 0,
385-
queued: queuedByQueue[queue.name] ?? 0,
414+
running:
415+
queue.role === "LIMIT"
416+
? (totalRunningByQueue[queue.name] ?? 0)
417+
: (runningByQueue[queue.name] ?? 0),
418+
queued:
419+
queue.role === "LIMIT"
420+
? (gateQueuedByQueue[queue.name] ?? 0)
421+
: (queuedByQueue[queue.name] ?? 0),
386422
concurrencyLimit: queue.concurrencyLimit ?? null,
387423
concurrencyLimitBase: queue.concurrencyLimitBase ?? null,
388424
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null,
@@ -401,6 +437,8 @@ export class QueueListPresenter extends BasePresenter {
401437
queue.concurrencyLimitOverridePercent !== null
402438
? Number(queue.concurrencyLimitOverridePercent)
403439
: null,
440+
kind: queue.role === "LIMIT" ? ("limit" as const) : ("queue" as const),
441+
concurrencyVersion: queue.concurrencyVersion,
404442
}));
405443
}
406444
}

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

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,27 @@ export type FoundQueue = Prettify<
2222
export async function getQueue(
2323
prismaClient: PrismaClientOrTransaction,
2424
environment: AuthenticatedEnvironment,
25-
queue: RetrieveQueueParam
25+
queue: RetrieveQueueParam,
26+
options?: {
27+
/**
28+
* The dashboard's detail page shows limit rows too; the public API and pause
29+
* flows stay scoped to queue rows.
30+
*/
31+
includeLimits?: boolean;
32+
}
2633
) {
34+
const role = options?.includeLimits
35+
? { in: ["QUEUE" as const, "LIMIT" as const] }
36+
: ("QUEUE" as const);
37+
2738
if (typeof queue === "string") {
2839
return joinQueueWithUser(
2940
prismaClient,
3041
await prismaClient.taskQueue.findFirst({
3142
where: {
3243
friendlyId: queue,
3344
runtimeEnvironmentId: environment.id,
34-
role: "QUEUE",
45+
role,
3546
},
3647
})
3748
);
@@ -45,7 +56,7 @@ export async function getQueue(
4556
where: {
4657
name: queueName,
4758
runtimeEnvironmentId: environment.id,
48-
role: "QUEUE",
59+
role,
4960
},
5061
})
5162
);
@@ -77,21 +88,28 @@ export class QueueRetrievePresenter extends BasePresenter {
7788
public async call({
7889
environment,
7990
queueInput,
91+
includeLimits,
8092
}: {
8193
environment: AuthenticatedEnvironment;
8294
queueInput: RetrieveQueueParam;
95+
includeLimits?: boolean;
8396
}) {
84-
const queue = await getQueue(this._replica, environment, queueInput);
97+
const queue = await getQueue(this._replica, environment, queueInput, { includeLimits });
8598
if (!queue) {
8699
return {
87100
success: false as const,
88101
code: "queue-not-found",
89102
};
90103
}
91104

105+
const isLimitRow = queue.role === "LIMIT";
92106
const results = await Promise.all([
93-
engine.lengthOfQueues(environment, [queue.name]),
94-
engine.currentConcurrencyOfQueues(environment, [queue.name]),
107+
isLimitRow
108+
? engine.gateQueuedCountOfQueues(environment, [queue.name])
109+
: engine.lengthOfQueues(environment, [queue.name]),
110+
isLimitRow
111+
? engine.totalConcurrencyOfQueues(environment, [queue.name])
112+
: engine.currentConcurrencyOfQueues(environment, [queue.name]),
95113
queue.totalConcurrencyLimit != null
96114
? engine.totalConcurrencyOfQueues(environment, [queue.name])
97115
: undefined,
@@ -126,6 +144,8 @@ export class QueueRetrievePresenter extends BasePresenter {
126144
queue.concurrencyLimitOverridePercent !== null
127145
? Number(queue.concurrencyLimitOverridePercent)
128146
: null,
147+
kind: isLimitRow ? ("limit" as const) : ("queue" as const),
148+
concurrencyVersion: queue.concurrencyVersion,
129149
},
130150
};
131151
}

0 commit comments

Comments
 (0)