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

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,9 @@ export class QueueRetrievePresenter extends BasePresenter {
     const results = await Promise.all([
       engine.lengthOfQueues(environment, [queue.name]),
       engine.currentConcurrencyOfQueues(environment, [queue.name]),
+      queue.totalConcurrencyLimit != null
+        ? engine.totalConcurrencyOfQueues(environment, [queue.name])
+        : undefined,
     ]);
 
     // Transform queues to include running and queued counts
@@ -107,6 +110,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 +156,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 +176,16 @@ export function toQueueItem(data: {
       override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null,
       overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy),
       overriddenAt: data.concurrencyLimitOverriddenAt,
+      combined:
+        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

@@ -115,6 +115,7 @@ import {
 import { queueMetricsMaxPeriodDays } from "~/components/queues/queueMetricsPeriod.server";
 import { isQueueAtCapacity } from "~/components/queues/queue-thresholds";
 import { pageMeta } from "~/utils/pageTitle";
+import { InlineCode } from "~/components/code/InlineCode";
 
 const SearchParamsSchema = z.object({
   query: z.string().optional(),
@@ -218,8 +219,13 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
         const timeRange = clipQueueMetricsWindow(
           timeFilterFromTo({
             period:
-              resolveQueueMetricsPeriod({ period, from, to, defaultPeriod, maxPeriodDays }) ??
-              undefined,
+              resolveQueueMetricsPeriod({
+                period,
+                from,
+                to,
+                defaultPeriod,
+                maxPeriodDays,
+              }) ?? undefined,
             from: parseFiniteInt(from),
             to: parseFiniteInt(to),
             defaultPeriod,
@@ -243,7 +249,9 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
           };
         }
       } catch (error) {
-        logger.warn("Queue list metrics unavailable, rendering without them", { error });
+        logger.warn("Queue list metrics unavailable, rendering without them", {
+          error,
+        });
       }
     }
 
@@ -706,7 +714,14 @@ function QueuesWithMetricsView() {
                 <TableHeaderCell>Name</TableHeaderCell>
                 <TableHeaderCell alignment="right">Queued</TableHeaderCell>
                 <TableHeaderCell alignment="right">Running</TableHeaderCell>
-                <TableHeaderCell alignment="right">Limit</TableHeaderCell>
+                <TableHeaderCell
+                  alignment="right"
+                  disableTooltipHoverableContent
+                  tooltip={limitTooltip}
+                  tooltipContentClassName="max-w-xs"
+                >
+                  Limit
+                </TableHeaderCell>
                 <TableHeaderCell
                   alignment="right"
                   tooltipContentClassName="max-w-max"
@@ -849,7 +864,14 @@ function QueuesWithMetricsView() {
                         className={cn(
                           "w-[1%]",
                           queue.paused ? "opacity-50" : undefined,
-                          queue.running > 0 && "text-text-bright"
+                          queue.concurrency?.combined?.current != null &&
+                            (queue.concurrency.combined.running ?? 0) >=
+                              Math.min(
+                                queue.concurrency.combined.current,
+                                environment.concurrencyLimit
+                              )
+                            ? "text-warning"
+                            : queue.running > 0 && "text-text-bright"
                         )}
                       >
                         {queue.running}
@@ -863,12 +885,46 @@ function QueuesWithMetricsView() {
                           queue.paused ? "opacity-50" : undefined,
                           queue.concurrency?.overriddenAt && "font-medium text-text-bright"
                         )}
+                        // The combined-limit hint is a tooltip button, so it renders beside the
+                        // link (trailing) rather than nested inside the <a>; the number stays the
+                        // link.
+                        trailingContent={
+                          queue.concurrency?.combined?.current != null ? (
+                            <SimpleTooltip
+                              disableHoverableContent
+                              buttonClassName="-ml-1 cursor-default"
+                              button={
+                                <span className="text-text-dimmed bg-repeat-x pb-[3px] [background-image:linear-gradient(to_right,currentColor_2px,transparent_2px)] [background-position:bottom] [background-size:4px_1px] group-hover/table-row:text-text-bright">
+                                  (
+                                  {Math.min(
+                                    queue.concurrency.combined.current,
+                                    environment.concurrencyLimit
+                                  )}
+                                  )
+                                </span>
+                              }
+                              content={
+                                <>
+                                  Combined limit: at most{" "}
+                                  {Math.min(
+                                    queue.concurrency.combined.current,
+                                    environment.concurrencyLimit
+                                  )}{" "}
+                                  runs across all concurrency keys of this queue. The main limit
+                                  applies to each key separately.
+                                </>
+                              }
+                              className="max-w-[260px]"
+                            />
+                          ) : undefined
+                        }
                       >
                         {queue.concurrencyLimitOverridePercent !== null ? (
                           <>
                             {limit}
                             <span className="ml-1 text-text-dimmed group-hover/table-row:text-text-bright">
-                              ({formatOverridePercent(queue.concurrencyLimitOverridePercent)}%)
+                              ({formatOverridePercent(queue.concurrencyLimitOverridePercent)}
+                              %)
                             </span>
                           </>
                         ) : (
@@ -1265,7 +1321,11 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
           value: limit > 0 ? Math.round((tileNumber(r.running) / limit) * 100) : 0,
         };
       });
-      return { points, total: peakOf(points), formatTotal: (v) => `${v}% peak` };
+      return {
+        points,
+        total: peakOf(points),
+        formatTotal: (v) => `${v}% peak`,
+      };
     },
   },
   {
@@ -1279,7 +1339,11 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
         bucket: tileTimeToMs(r.t),
         value: tileNumber(r.queued),
       }));
-      return { points, total: peakOf(points), formatTotal: (v) => `${v.toLocaleString()} peak` };
+      return {
+        points,
+        total: peakOf(points),
+        formatTotal: (v) => `${v.toLocaleString()} peak`,
+      };
     },
   },
   {
@@ -1784,7 +1848,13 @@ function ClassicQueuesView() {
                   <TableHeaderCell>Name</TableHeaderCell>
                   <TableHeaderCell alignment="right">Queued</TableHeaderCell>
                   <TableHeaderCell alignment="right">Running</TableHeaderCell>
-                  <TableHeaderCell alignment="right">Limit</TableHeaderCell>
+                  <TableHeaderCell
+                    alignment="right"
+                    tooltip={limitTooltip}
+                    tooltipContentClassName="max-w-xs"
+                  >
+                    Limit
+                  </TableHeaderCell>
                   <TableHeaderCell
                     alignment="right"
                     tooltip={
@@ -1891,7 +1961,14 @@ function ClassicQueuesView() {
                           className={cn(
                             "w-[1%] pl-16 tabular-nums",
                             queue.paused ? "opacity-50" : undefined,
-                            queue.running > 0 && "text-text-bright",
+                            queue.concurrency?.combined?.current != null &&
+                              (queue.concurrency.combined.running ?? 0) >=
+                                Math.min(
+                                  queue.concurrency.combined.current,
+                                  environment.concurrencyLimit
+                                )
+                              ? "text-warning"
+                              : queue.running > 0 && "text-text-bright",
                             isAtConcurrencyLimit && "text-warning"
                           )}
                         >
@@ -1906,6 +1983,34 @@ function ClassicQueuesView() {
                           )}
                         >
                           {limit}
+                          {queue.concurrency?.combined?.current != null ? (
+                            <SimpleTooltip
+                              disableHoverableContent
+                              buttonClassName="ml-1 cursor-default"
+                              button={
+                                <span className="text-text-dimmed bg-repeat-x pb-[3px] [background-image:linear-gradient(to_right,currentColor_2px,transparent_2px)] [background-position:bottom] [background-size:4px_1px]">
+                                  (
+                                  {Math.min(
+                                    queue.concurrency.combined.current,
+                                    environment.concurrencyLimit
+                                  )}
+                                  )
+                                </span>
+                              }
+                              content={
+                                <>
+                                  Combined limit: at most{" "}
+                                  {Math.min(
+                                    queue.concurrency.combined.current,
+                                    environment.concurrencyLimit
+                                  )}{" "}
+                                  runs across all concurrency keys of this queue. The main limit
+                                  applies to each key separately.
+                                </>
+                              }
+                              className="max-w-[260px]"
+                            />
+                          ) : null}
                         </TableCell>
                         <TableCell
                           alignment="right"
@@ -2024,3 +2129,16 @@ function BurstFactorTooltip({
     />
   );
 }
+
+const limitTooltip = (
+  <>
+    <Paragraph variant="extra-small" spacing>
+      How many runs can execute at once.{" "}
+    </Paragraph>
+    <Paragraph variant="extra-small" spacing>
+      <InlineCode variant="extra-extra-small">1 (20)</InlineCode> means 1 run per concurrency key,
+      but at most 20 runs across all keys. Set using{" "}
+      <InlineCode variant="extra-extra-small">combinedConcurrencyLimit</InlineCode> in your code.
+    </Paragraph>
+  </>
+);