feat(webapp,run-engine,core,clickhouse): surface total concurrency in metrics and dashboard

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

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+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

@@ -9,7 +9,10 @@ import { engine } from "~/v3/runEngine.server";
 import { BasePresenter } from "./basePresenter.server";
 import { toQueueItem } from "./QueueRetrievePresenter.server";
 
-type QueueListEngine = Pick<RunEngine, "lengthOfQueues" | "currentConcurrencyOfQueues">;
+type QueueListEngine = Pick<
+  RunEngine,
+  "lengthOfQueues" | "currentConcurrencyOfQueues" | "totalConcurrencyOfQueues"
+>;
 
 export const QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE = 25;
 const MAX_ITEMS_PER_PAGE = 100;
@@ -34,6 +37,9 @@ const queueListSelect = {
   concurrencyLimitOverriddenAt: true,
   concurrencyLimitOverriddenBy: true,
   concurrencyLimitOverridePercent: true,
+  totalConcurrencyLimit: true,
+  totalConcurrencyLimitBase: true,
+  totalConcurrencyLimitOverriddenAt: true,
   type: true,
   paused: true,
 } satisfies Prisma.TaskQueueSelect;
@@ -333,11 +339,15 @@ export class QueueListPresenter extends BasePresenter {
       concurrencyLimitOverriddenAt: Date | null;
       concurrencyLimitOverriddenBy: string | null;
       concurrencyLimitOverridePercent: Prisma.Decimal | null;
+      totalConcurrencyLimit: number | null;
+      totalConcurrencyLimitBase: number | null;
+      totalConcurrencyLimitOverriddenAt: Date | null;
       type: TaskQueueType;
       paused: boolean;
     }[]
   ): Promise<QueueListItem[]> {
-    const [queuedByQueue, runningByQueue] = await Promise.all([
+    const queuesWithTotalCap = queues.filter((q) => q.totalConcurrencyLimit !== null);
+    const [queuedByQueue, runningByQueue, totalRunningByQueue] = await Promise.all([
       this.engineClient.lengthOfQueues(
         environment,
         queues.map((q) => q.name)
@@ -346,6 +356,12 @@ export class QueueListPresenter extends BasePresenter {
         environment,
         queues.map((q) => q.name)
       ),
+      queuesWithTotalCap.length > 0
+        ? this.engineClient.totalConcurrencyOfQueues(
+            environment,
+            queuesWithTotalCap.map((q) => q.name)
+          )
+        : Promise.resolve({} as Record<string, number>),
     ]);
 
     // Manually "join" the overridden users because there is no way to implement the relationship
@@ -373,6 +389,11 @@ export class QueueListPresenter extends BasePresenter {
           ? (overriddenByMap.get(queue.concurrencyLimitOverriddenBy) ?? null)
           : null,
         paused: queue.paused,
+        totalConcurrencyLimit: queue.totalConcurrencyLimit,
+        totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
+        totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
+        totalRunning:
+          queue.totalConcurrencyLimit !== null ? (totalRunningByQueue[queue.name] ?? 0) : null,
       }),
       // Prisma returns Decimal; the client only needs a plain number (null for absolute overrides).
       concurrencyLimitOverridePercent:

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

@@ -90,6 +90,7 @@ export class QueueRetrievePresenter extends BasePresenter {
     const results = await Promise.all([
       engine.lengthOfQueues(environment, [queue.name]),
       engine.currentConcurrencyOfQueues(environment, [queue.name]),
+      engine.totalConcurrencyOfQueues(environment, [queue.name]),
     ]);
 
     // Transform queues to include running and queued counts
@@ -107,6 +108,11 @@ export class QueueRetrievePresenter extends BasePresenter {
           concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null,
           concurrencyLimitOverriddenBy: queue.concurrencyLimitOverriddenBy ?? null,
           paused: queue.paused,
+          totalConcurrencyLimit: queue.totalConcurrencyLimit ?? null,
+          totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase ?? null,
+          totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt ?? null,
+          totalRunning:
+            queue.totalConcurrencyLimit != null ? (results[2]?.[queue.name] ?? 0) : null,
         }),
         // The percent source-of-truth for percent-based overrides isn't part of the shared
         // `QueueItem` schema (that's a public contract), so we surface it as an extra field on
@@ -148,6 +154,10 @@ export function toQueueItem(data: {
   concurrencyLimitOverriddenAt: Date | null;
   concurrencyLimitOverriddenBy: User | null;
   paused: boolean;
+  totalConcurrencyLimit?: number | null;
+  totalConcurrencyLimitBase?: number | null;
+  totalConcurrencyLimitOverriddenAt?: Date | null;
+  totalRunning?: number | null;
 }): QueueItem & { releaseConcurrencyOnWaitpoint: boolean } {
   return {
     id: data.friendlyId,
@@ -164,6 +174,16 @@ export function toQueueItem(data: {
       override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null,
       overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy),
       overriddenAt: data.concurrencyLimitOverriddenAt,
+      total:
+        data.totalConcurrencyLimit !== undefined
+          ? {
+              current: data.totalConcurrencyLimit,
+              base: data.totalConcurrencyLimitBase ?? null,
+              override: data.totalConcurrencyLimitOverriddenAt ? data.totalConcurrencyLimit : null,
+              overriddenAt: data.totalConcurrencyLimitOverriddenAt ?? null,
+              running: data.totalRunning ?? null,
+            }
+          : undefined,
     },
     // TODO: This needs to be removed but keeping this here for now to avoid breaking existing clients
     releaseConcurrencyOnWaitpoint: true,

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

@@ -698,6 +698,13 @@ function QueuesWithMetricsView() {
                 <TableHeaderCell alignment="right">Queued</TableHeaderCell>
                 <TableHeaderCell alignment="right">Running</TableHeaderCell>
                 <TableHeaderCell alignment="right">Limit</TableHeaderCell>
+                <TableHeaderCell
+                  alignment="right"
+                  disableTooltipHoverableContent
+                  tooltip="Runs in flight across all concurrency keys, against the queue's total concurrency limit. Only queues with a totalConcurrencyLimit show a value."
+                >
+                  Total
+                </TableHeaderCell>
                 <TableHeaderCell
                   alignment="right"
                   tooltipContentClassName="max-w-max"
@@ -866,6 +873,29 @@ function QueuesWithMetricsView() {
                           limit
                         )}
                       </TableCell>
+                      <TableCell
+                        to={queueDetailPath}
+                        alignment="right"
+                        actionClassName="pl-16 tabular-nums"
+                        className={cn(
+                          "w-[1%]",
+                          queue.paused ? "opacity-50" : undefined,
+                          queue.concurrency?.total?.current != null &&
+                            (queue.concurrency.total.running ?? 0) >=
+                              Math.min(
+                                queue.concurrency.total.current,
+                                environment.concurrencyLimit
+                              ) &&
+                            "text-warning"
+                        )}
+                      >
+                        {queue.concurrency?.total?.current != null
+                          ? `${queue.concurrency.total.running ?? 0}/${Math.min(
+                              queue.concurrency.total.current,
+                              environment.concurrencyLimit
+                            )}`
+                          : "–"}
+                      </TableCell>
                       <TableCell
                         to={queueDetailPath}
                         alignment="right"
@@ -1012,7 +1042,7 @@ function QueuesWithMetricsView() {
                 })
               ) : (
                 <TableRow>
-                  <TableCell colSpan={9}>
+                  <TableCell colSpan={10}>
                     <div className="grid place-items-center py-6 text-text-dimmed">
                       <Paragraph>
                         {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

@@ -392,7 +392,12 @@ export default function Page() {
               <ConcurrencyKeysBlankState />
             )
           ) : (
-            <OverviewCharts ids={ids} timeRange={timeRange} queueName={fullName} />
+            <OverviewCharts
+              ids={ids}
+              timeRange={timeRange}
+              queueName={fullName}
+              hasTotalLimit={queue.concurrency?.total?.current != null}
+            />
           )}
         </MetricsLayout.Content>
 
@@ -402,7 +407,13 @@ export default function Page() {
         {view === "keys" && hasKeys ? (
           <>
             <MetricsLayout.Content>
-              <KeyStatsTable ids={ids} timeRange={timeRange} queueName={fullName} />
+              <KeyStatsTable
+                ids={ids}
+                timeRange={timeRange}
+                queueName={fullName}
+                defaultKeyLimit={queue.concurrencyLimit ?? environmentConcurrencyLimit}
+                envLimit={environmentConcurrencyLimit}
+              />
             </MetricsLayout.Content>
             {selectedKey ? (
               <MetricsLayout.Content inset>
@@ -436,10 +447,12 @@ function OverviewCharts({
   ids,
   timeRange,
   queueName,
+  hasTotalLimit,
 }: {
   ids: Ids;
   timeRange: TimeRangeParams;
   queueName: string;
+  hasTotalLimit: boolean;
 }) {
   const zoomToTimeFilter = useZoomToTimeFilter();
   return (
@@ -479,6 +492,37 @@ function OverviewCharts({
           // leading zeros so the reference line doesn't start with a false 0→limit step.
           carryBackfill={["limit"]}
         />
+        {hasTotalLimit ? (
+          <QueueDetailChartCard
+            title="Total concurrency"
+            info={
+              <>
+                Runs in flight across ALL concurrency keys (
+                <ColorSwatch color={COLORS.running} />) versus the queue's total limit (
+                <ColorSwatch color={COLORS.limit} />
+                ).
+              </>
+            }
+            showLegend
+            className="aspect-[2/1]"
+            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`}
+            fillGaps
+            minBucketSeconds={SYNCED_CHART_MIN_BUCKET_SECONDS}
+            ids={ids}
+            timeRange={timeRange}
+            queueName={queueName}
+            series={[
+              { key: "cap", label: "Total limit", color: COLORS.limit },
+              { key: "running", label: "Running", color: COLORS.running },
+            ]}
+            thresholdStroke={{
+              series: "running",
+              valueFromSeries: "cap",
+              aboveColor: "var(--color-warning)",
+            }}
+            carryBackfill={["cap"]}
+          />
+        ) : null}
         <QueueDetailChartCard
           title="Queue depth"
           info="How many runs are waiting in this queue over time."
@@ -922,10 +966,15 @@ function KeyStatsTable({
   ids,
   timeRange,
   queueName,
+  defaultKeyLimit,
+  envLimit,
 }: {
   ids: Ids;
   timeRange: TimeRangeParams;
   queueName: string;
+  /** The limit a key inherits when it has no override (the queue's limit, else the env limit). */
+  defaultKeyLimit: number;
+  envLimit: number;
 }) {
   const { value, replace, del } = useSearchParams();
   const selectedKey = value("key");
@@ -968,6 +1017,12 @@ function KeyStatsTable({
             <TableHeaderCell>Key</TableHeaderCell>
             <TableHeaderCell alignment="right">Queued now</TableHeaderCell>
             <TableHeaderCell alignment="right">Running now</TableHeaderCell>
+            <TableHeaderCell
+              alignment="right"
+              tooltip="The key's concurrency limit. Keys inherit the queue's limit unless a per-key override is set via the API."
+            >
+              Limit
+            </TableHeaderCell>
             <TableHeaderCell alignment="right">Oldest wait</TableHeaderCell>
             <TableHeaderCell alignment="right">Started</TableHeaderCell>
             <TableHeaderCell alignment="right">Peak backlog</TableHeaderCell>
@@ -976,11 +1031,11 @@ function KeyStatsTable({
         </TableHeader>
         <TableBody>
           {showLoading ? (
-            <TableBlankRow colSpan={7} className="text-text-dimmed">
+            <TableBlankRow colSpan={8} className="text-text-dimmed">
               Loading…
             </TableBlankRow>
           ) : rows.length === 0 ? (
-            <TableBlankRow colSpan={7} className="text-text-dimmed">
+            <TableBlankRow colSpan={8} className="text-text-dimmed">
               {search ? `No keys match “${search}”` : "No concurrency keys"}
             </TableBlankRow>
           ) : (
@@ -994,6 +1049,12 @@ function KeyStatsTable({
                 <TableCell>{row.key}</TableCell>
                 <TableCell alignment="right">{row.queued.toLocaleString()}</TableCell>
                 <TableCell alignment="right">{row.running.toLocaleString()}</TableCell>
+                <TableCell
+                  alignment="right"
+                  className={row.limitOverride !== null ? undefined : "text-text-dimmed"}
+                >
+                  {Math.min(row.limitOverride ?? defaultKeyLimit, envLimit).toLocaleString()}
+                </TableCell>
                 <TableCell alignment="right">
                   {row.oldestWaitMs === null ? "–" : formatWaitMs(row.oldestWaitMs)}
                 </TableCell>

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

@@ -52,6 +52,9 @@ const route = createActionApiRoute(
               concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
               concurrencyLimitOverriddenBy: null,
               paused: queue.paused,
+              totalConcurrencyLimit: queue.totalConcurrencyLimit,
+              totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
+              totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
             }),
             { status: 200 }
           );

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

@@ -46,6 +46,9 @@ const route = createActionApiRoute(
               concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
               concurrencyLimitOverriddenBy: null,
               paused: queue.paused,
+              totalConcurrencyLimit: queue.totalConcurrencyLimit,
+              totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
+              totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
             }),
             { status: 200 }
           );

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

@@ -61,6 +61,9 @@ const route = createActionApiRoute(
               concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
               concurrencyLimitOverriddenBy: null,
               paused: queue.paused,
+              totalConcurrencyLimit: queue.totalConcurrencyLimit,
+              totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
+              totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
             }),
             { status: 200 }
           );

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

@@ -43,6 +43,9 @@ const route = createActionApiRoute(
             concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
             concurrencyLimitOverriddenBy: null,
             paused: queue.paused,
+            totalConcurrencyLimit: queue.totalConcurrencyLimit,
+            totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
+            totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
           }),
           { status: 200 }
         );

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

@@ -46,6 +46,9 @@ const route = createActionApiRoute(
               concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
               concurrencyLimitOverriddenBy: null,
               paused: queue.paused,
+              totalConcurrencyLimit: queue.totalConcurrencyLimit,
+              totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
+              totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
             }),
             { status: 200 }
           );

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

@@ -45,6 +45,9 @@ const route = createActionApiRoute(
               concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
               concurrencyLimitOverriddenBy: null,
               paused: queue.paused,
+              totalConcurrencyLimit: queue.totalConcurrencyLimit,
+              totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase,
+              totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt,
             }),
             { status: 200 }
           );

apps/webapp/app/routes/resources.queues.concurrency-keys.ts

@@ -43,6 +43,8 @@ export type ConcurrencyKeyRow = {
   peakBacklog: number;
   peakRunning: number;
   meanWaitMs: number;
+  /** Per-key concurrency limit override, when one is set for this key (null = inherits the queue limit). */
+  limitOverride: number | null;
 };
 
 export type ConcurrencyKeysResponse =
@@ -151,8 +153,11 @@ export const action = async ({ request }: ActionFunctionArgs) => {
     const total = rankingRows?.[0]?.ranked_total ?? 0;
     const keys = (rankingRows ?? []).map((r) => r.concurrency_key);
 
-    // Enrich just this page's keys with live "now" counts from Redis.
-    const live = await engine.concurrencyKeyLiveStats(environment, queueName, keys);
+    // Enrich just this page's keys with live "now" counts and any per-key limit overrides from Redis.
+    const [live, keyLimitOverrides] = await Promise.all([
+      engine.concurrencyKeyLiveStats(environment, queueName, keys),
+      engine.runQueue.getQueueConcurrencyKeyLimitsForKeys(environment, queueName, keys),
+    ]);
     const loadedAt = Date.now();
 
     const rows: ConcurrencyKeyRow[] = (rankingRows ?? []).map((r) => {
@@ -168,6 +173,7 @@ export const action = async ({ request }: ActionFunctionArgs) => {
         peakBacklog: r.peak_backlog,
         peakRunning: r.peak_running,
         meanWaitMs: r.mean_wait_ms,
+        limitOverride: keyLimitOverrides[r.concurrency_key] ?? null,
       };
     });