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

apps/webapp/app/components/queues/QueueMetricCards.tsx

@@ -102,6 +102,12 @@ type QueueMetricChartProps = {
    * are config values that existed all along, so carry the first value backward instead.
    */
   carryBackfill?: string[];
+  /**
+   * Column that marks a bucket as genuinely sampled. When set, carryBackfill only
+   * overwrites buckets where this column is absent or zero, so history from before
+   * a config value existed keeps its truthful gap instead of inheriting the value.
+   */
+  carryBackfillGuard?: string;
   /** Show the series legend below the chart (use for multi-series charts). */
   showLegend?: boolean;
   /**
@@ -141,6 +147,7 @@ export function QueueMetricChart({
   defaultPeriod,
   warningOverlay,
   carryBackfill,
+  carryBackfillGuard,
   thresholdStroke,
   onHasDataChange,
   minBucketSeconds,
@@ -174,12 +181,15 @@ export function QueueMetricChart({
         const first = points.findIndex((p) => toNumber(p[key]) > 0);
         if (first > 0) {
           const value = points[first]![key]!;
-          for (let i = 0; i < first; i++) points[i]![key] = value;
+          for (let i = 0; i < first; i++) {
+            if (carryBackfillGuard && toNumber(points[i]![carryBackfillGuard]) > 0) continue;
+            points[i]![key] = value;
+          }
         }
       }
     }
     return points;
-  }, [rows, series, carryBackfill, sampleCountColumn]);
+  }, [rows, series, carryBackfill, carryBackfillGuard, sampleCountColumn]);
 
   const chartConfig = useMemo(() => {
     const cfg: ChartConfig = {};
@@ -340,7 +350,7 @@ export function QueueSidebarStats({
   };
 
   const { rows, showLoading } = useQueueMetric(
-    `SELECT max(max_queued) AS peak_queued,\n  round(quantilesMerge(0.5, 0.9, 0.95, 0.99)(wait_quantiles)[3]) AS worst_p95\nFROM queue_metrics`,
+    `SELECT max(max_queued) AS peak_queued,\n  round(quantilesMerge(0.5, 0.9, 0.95, 0.99)(wait_quantiles)[3]) AS worst_p95\nFROM concurrency_metrics`,
     { ids, timeRange, queueName, defaultPeriod }
   );
   const row = rows[0];

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

@@ -634,7 +634,7 @@ const queuesDashboard: BuiltInDashboard = {
       "t-pressure": { title: "Queue pressure", query: "", display: { type: "title" } },
       pressure: {
         title: "Queue pressure",
-        query: `SELECT queue,\n  argMax(max_running, bucket_start) AS running,\n  argMax(max_queued, bucket_start) AS queued,\n  argMax(max_limit, bucket_start) AS limit,\n  running + queued AS demand,\n  max(max_queued) AS peak_queued,\n  sum(throttled_count) AS throttled,\n  multiIf(running >= limit AND queued > 0, 'queue-limited', queued > 0, 'backlogged', 'healthy') AS status\nFROM queue_metrics\nGROUP BY queue\nORDER BY peak_queued DESC`,
+        query: `SELECT queue,\n  argMax(max_running, bucket_start) AS running,\n  argMax(max_queued, bucket_start) AS queued,\n  argMax(max_limit, bucket_start) AS limit,\n  running + queued AS demand,\n  max(max_queued) AS peak_queued,\n  sum(throttled_count) AS throttled,\n  multiIf(running >= limit AND queued > 0, 'queue-limited', queued > 0, 'backlogged', 'healthy') AS status\nFROM concurrency_metrics\nGROUP BY queue\nORDER BY peak_queued DESC`,
         display: {
           type: "table",
           prettyFormatting: true,
@@ -644,7 +644,7 @@ const queuesDashboard: BuiltInDashboard = {
       "t-trends": { title: "Per-queue trends", query: "", display: { type: "title" } },
       "running-q": {
         title: "Running by queue",
-        query: `SELECT timeBucket() AS t, queue, max(max_running) AS running\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`,
+        query: `SELECT timeBucket() AS t, queue, max(max_running) AS running\nFROM concurrency_metrics\nGROUP BY t, queue\nORDER BY t`,
         // Grouped gauge: carry each queue's running across idle buckets (per-group LOCF).
         fillGaps: true,
         display: {
@@ -661,7 +661,7 @@ const queuesDashboard: BuiltInDashboard = {
       },
       "queued-q": {
         title: "Queue depth (backlog) by queue",
-        query: `SELECT timeBucket() AS t, queue, max(max_queued) AS queued\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`,
+        query: `SELECT timeBucket() AS t, queue, max(max_queued) AS queued\nFROM concurrency_metrics\nGROUP BY t, queue\nORDER BY t`,
         // Grouped gauge: carry each queue's backlog across idle buckets (per-group LOCF).
         fillGaps: true,
         display: {
@@ -678,7 +678,7 @@ const queuesDashboard: BuiltInDashboard = {
       },
       "throttled-q": {
         title: "Throttled buckets by queue",
-        query: `SELECT timeBucket() AS t, queue, sum(throttled_count) AS throttled\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`,
+        query: `SELECT timeBucket() AS t, queue, sum(throttled_count) AS throttled\nFROM concurrency_metrics\nGROUP BY t, queue\nORDER BY t`,
         // Grouped counter: per-group zero-fill so idle buckets read 0, not a gap.
         fillGaps: true,
         display: {
@@ -697,7 +697,7 @@ const queuesDashboard: BuiltInDashboard = {
         title: "Enqueued vs started",
         // Counter states merge per queue, then sum outside: a single merge across queues
         // mixes unrelated odometers and returns wrong totals.
-        query: `SELECT t, sum(enq) AS enqueued, sum(st) AS started\nFROM (\n  SELECT timeBucket() AS t, queue,\n    deltaSumTimestampMerge(enqueue_delta) AS enq,\n    deltaSumTimestampMerge(started_delta) AS st\n  FROM queue_metrics\n  GROUP BY t, queue\n)\nGROUP BY t\nORDER BY t`,
+        query: `SELECT t, sum(enq) AS enqueued, sum(st) AS started\nFROM (\n  SELECT timeBucket() AS t, queue,\n    deltaSumTimestampMerge(enqueue_delta) AS enq,\n    deltaSumTimestampMerge(started_delta) AS st\n  FROM concurrency_metrics\n  GROUP BY t, queue\n)\nGROUP BY t\nORDER BY t`,
         display: {
           type: "chart",
           chartType: "line",

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/presenters/v3/reports/health/health-data.ts

@@ -212,7 +212,7 @@ function queueWorstQuery(): string {
   return `SELECT
   queue AS name,
   argMax(max_queued, bucket_start) AS latest_queued
-FROM queue_metrics
+FROM concurrency_metrics
 GROUP BY queue
 ORDER BY latest_queued DESC
 LIMIT 20`;
@@ -228,7 +228,7 @@ FROM (
   SELECT
     deltaSumTimestampMerge(dlq_delta) AS dlq,
     argMax(max_queued, bucket_start) AS latest_queued
-  FROM queue_metrics
+  FROM concurrency_metrics
   GROUP BY queue
 )`;
 }

apps/webapp/app/presenters/v3/reports/report-registry.ts

@@ -4,7 +4,7 @@ import { loadHealthInput } from "./health/health-data";
 import { type ReportViewModel } from "./report-view-model";
 
 /** A query table a report may read. Same table names the query API authorizes against. */
-export type ReportQueryTable = "runs" | "env_metrics" | "queue_metrics";
+export type ReportQueryTable = "runs" | "env_metrics" | "concurrency_metrics";
 
 export type ReportLoader<TInput> = {
   /** Authorization metadata: the route derives its per-table JWT scope check from this. */
@@ -19,7 +19,7 @@ function defineReport<TInput>(loader: ReportLoader<TInput>): ReportLoader<unknow
 
 export const REPORT_REGISTRY: Record<string, ReportLoader<unknown>> = {
   health: defineReport({
-    tables: ["runs", "env_metrics", "queue_metrics"],
+    tables: ["runs", "env_metrics", "concurrency_metrics"],
     load: (env, period) => loadHealthInput(env, period),
     interpret: interpretHealth,
   }),