feat(webapp,run-engine): queue metrics and health dashboard

.server-changes/queue-metrics-dashboard.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Queue metrics and health on the Queues page: per-queue depth, throughput, concurrency, throttling, and scheduling-delay charts, plus a per-queue detail view. Off by default; enabled per organization.

apps/webapp/app/components/primitives/UsageSparkline.tsx

@@ -27,6 +27,8 @@ export type UsageSparklineProps = {
   color?: string;
   /** Unit shown in the tooltip (e.g. calls, tokens). */
   unitLabel?: UnitLabel;
+  /** Trailing scalar shown after the chart. Defaults to the sum of buckets (override for gauges, e.g. peak). */
+  total?: number;
   /** Format the trailing total. Defaults to `toLocaleString`. */
   formatTotal?: (total: number) => string;
   /** Class for the trailing total label. */
@@ -44,14 +46,16 @@ export function UsageSparkline({
   bucketIntervalMs,
   color = "#3B82F6",
   unitLabel = { singular: "call", plural: "calls" },
+  total: totalOverride,
   formatTotal,
   totalClassName = "text-blue-400",
 }: UsageSparklineProps) {
-  if (!data || data.every((v) => v === 0)) {
+  const hasTotalOverride = totalOverride !== undefined;
+  if (!data || data.length === 0 || (data.every((v) => v === 0) && !hasTotalOverride)) {
     return <span className="text-text-dimmed">–</span>;
   }
 
-  const total = data.reduce((a, b) => a + b, 0);
+  const total = totalOverride ?? data.reduce((a, b) => a + b, 0);
   const max = Math.max(...data);
 
   // Map each bucket to a dated point so the tooltip can show the window it

apps/webapp/app/components/primitives/charts/Chart.tsx

@@ -216,7 +216,7 @@ const ChartTooltipContent = React.forwardRef<
                       )}
                       <div
                         className={cn(
-                          "flex flex-1 justify-between leading-none",
+                          "flex flex-1 justify-between gap-3 leading-none",
                           nestLabel ? "items-end" : "items-center"
                         )}
                       >

apps/webapp/app/components/primitives/charts/ChartLine.tsx

@@ -4,6 +4,7 @@ import {
   CartesianGrid,
   Line,
   LineChart,
+  ReferenceLine,
   XAxis,
   YAxis,
   type XAxisProps,
@@ -48,12 +49,38 @@ export type ChartLineRendererProps = {
   tooltipLabelFormatter?: (label: string, payload: any[]) => string;
   /** Optional formatter for numeric tooltip values (e.g. bytes, duration) */
   tooltipValueFormatter?: (value: number) => string;
+  /** Draw a dot at each data point. Defaults to true; turn off for dense/compact charts. */
+  showDots?: boolean;
+  /** Horizontal reference lines (e.g. limits); the y-domain extends to include them. */
+  referenceLines?: Array<{ y: number; label?: string; color?: string }>;
   /** Width injected by ResponsiveContainer */
   width?: number;
   /** Height injected by ResponsiveContainer */
   height?: number;
 };
 
+/** Reference-line label: right-aligned just below the line (recharts injects viewBox). */
+function ReferenceLineLabel({
+  viewBox,
+  value,
+}: {
+  viewBox?: { x: number; y: number; width: number };
+  value: string;
+}) {
+  if (!viewBox) return null;
+  return (
+    <text
+      x={viewBox.x + viewBox.width - 4}
+      y={viewBox.y + 12}
+      textAnchor="end"
+      fill="#878C99"
+      fontSize={10}
+    >
+      {value}
+    </text>
+  );
+}
+
 /**
  * Line chart renderer for the compound component system.
  * Must be used within a Chart.Root.
@@ -73,6 +100,8 @@ export function ChartLineRenderer({
   stacked = false,
   tooltipLabelFormatter,
   tooltipValueFormatter,
+  showDots = true,
+  referenceLines,
   width,
   height,
 }: ChartLineRendererProps) {
@@ -176,6 +205,17 @@ export function ChartLineRenderer({
           labelFormatter={tooltipLabelFormatter}
         />
         {/* Note: Legend is now rendered by ChartRoot outside the chart container */}
+        {referenceLines?.map((line) => (
+          <ReferenceLine
+            key={`ref-${line.y}-${line.label ?? ""}`}
+            y={line.y}
+            stroke={line.color ?? "#4D525B"}
+            strokeDasharray="4 4"
+            strokeWidth={1}
+            ifOverflow="extendDomain"
+            label={line.label ? <ReferenceLineLabel value={line.label} /> : undefined}
+          />
+        ))}
         {visibleSeries.map((key) => (
           <Area
             key={key}
@@ -222,14 +262,25 @@ export function ChartLineRenderer({
         labelFormatter={tooltipLabelFormatter}
       />
       {/* Note: Legend is now rendered by ChartRoot outside the chart container */}
+      {referenceLines?.map((line) => (
+        <ReferenceLine
+          key={`ref-${line.y}-${line.label ?? ""}`}
+          y={line.y}
+          stroke={line.color ?? "#4D525B"}
+          strokeDasharray="4 4"
+          strokeWidth={1}
+          ifOverflow="extendDomain"
+          label={line.label ? <ReferenceLineLabel value={line.label} /> : undefined}
+        />
+      ))}
       {visibleSeries.map((key) => (
         <Line
           key={key}
           dataKey={key}
           type={lineType}
           stroke={config[key]?.color}
           strokeWidth={1}
-          dot={{ r: 1.5, fill: config[key]?.color, strokeWidth: 0 }}
+          dot={showDots ? { r: 1.5, fill: config[key]?.color, strokeWidth: 0 } : false}
           activeDot={{ r: 4 }}
           isAnimationActive={false}
         />

apps/webapp/app/entry.server.tsx

@@ -11,6 +11,7 @@ import * as Worker from "~/services/worker.server";
 import { initMollifierDrainerWorker } from "~/v3/mollifierDrainerWorker.server";
 import { initMollifierStaleSweepWorker } from "~/v3/mollifierStaleSweepWorker.server";
 import { initBillingLimitWorker } from "~/v3/billingLimitWorker.server";
+import { initQueueMetricsConsumer, initQueueMetricsEmitter } from "~/v3/queueMetrics.server";
 import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import type { OperatingSystemPlatform } from "./components/primitives/OperatingSystemProvider";
@@ -234,6 +235,8 @@ Worker.init().catch((error) => {
 initMollifierDrainerWorker();
 initMollifierStaleSweepWorker();
 initBillingLimitWorker();
+initQueueMetricsEmitter();
+initQueueMetricsConsumer();
 
 bootstrap().catch((error) => {
   logError(error);

apps/webapp/app/env.server.ts

@@ -883,6 +883,28 @@ const EnvironmentSchema = z
     RUN_ENGINE_REUSE_SNAPSHOT_COUNT: z.coerce.number().int().default(0),
     RUN_ENGINE_MAXIMUM_ENV_COUNT: z.coerce.number().int().optional(),
     RUN_ENGINE_RUN_QUEUE_SHARD_COUNT: z.coerce.number().int().default(4),
+    // Queue metrics ingestion (Redis Stream -> ClickHouse). The runtime on/off is the
+    // `queue_metrics:enabled` Redis key; these gate emitter construction + consumer boot.
+    QUEUE_METRICS_EMIT_ENABLED: z.string().default("0"),
+    QUEUE_METRICS_CONSUMER_ENABLED: z.string().default("0"),
+    QUEUE_METRICS_STREAM_SHARD_COUNT: z.coerce.number().int().default(4),
+    QUEUE_METRICS_CONSUMER_BATCH_SIZE: z.coerce.number().int().default(1000),
+    // Counter stream (exact counts, loss-intolerant). Unset host => the run-queue Redis;
+    // set it to a dedicated instance so counter backlog never competes with the run queue.
+    QUEUE_METRICS_REDIS_HOST: z.string().optional(),
+    QUEUE_METRICS_REDIS_PORT: z.coerce.number().optional(),
+    QUEUE_METRICS_REDIS_USERNAME: z.string().optional(),
+    QUEUE_METRICS_REDIS_PASSWORD: z.string().optional(),
+    QUEUE_METRICS_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+    QUEUE_METRICS_COUNTER_STREAM_MAXLEN: z.coerce.number().int().default(8_000_000),
+    // TTL (seconds) on the per-(queue,op) cumulative odometer key, refreshed on every write.
+    // Idle-past-TTL queues purge and self-heal (restart from 1) on return; default 7 days.
+    QUEUE_METRICS_COUNTER_ODOMETER_TTL_SECONDS: z.coerce.number().int().default(604_800),
+    // Per-env distinct queue_name cap (0 = unlimited); overflow maps to "__overflow__".
+    QUEUE_METRICS_MAX_QUEUE_NAMES_PER_ENV: z.coerce.number().int().default(1000),
+    // Fraction (0..1) of ops that emit a gauge; counters are never sampled. Dial below 1
+    // only if EngineCPU is too high in slow-path-heavy regions (hurts low-traffic queues).
+    QUEUE_METRICS_GAUGE_SAMPLE_RATE: z.coerce.number().min(0).max(1).default(1),
     RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
     RUN_ENGINE_RETRY_WARM_START_THRESHOLD_MS: z.coerce.number().int().default(30_000),
     RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS: z.coerce.number().int().default(200),

apps/webapp/app/hooks/useMetricResourceQuery.ts

@@ -0,0 +1,109 @@
+import { useCallback, useEffect, useRef, useState } from "react";
+import { useInterval } from "./useInterval";
+
+export type MetricResourceRow = Record<string, number | string | null>;
+
+type MetricResourceResponse =
+  | { success: true; data: { rows: MetricResourceRow[] } }
+  | { success: false; error: string };
+
+export type MetricResourceTimeRange = {
+  period: string | null;
+  from: string | null;
+  to: string | null;
+};
+
+export type MetricResourceQueryOptions = {
+  organizationId: string;
+  projectId: string;
+  environmentId: string;
+  timeRange: MetricResourceTimeRange;
+  defaultPeriod: string;
+  queues?: string[];
+  fillGaps?: boolean;
+  refreshIntervalMs?: number;
+};
+
+/**
+ * Client-fetch a TRQL query from the metric resource route (like the dashboard
+ * widgets): own loading state, interval + on-focus refresh, abort on change/unmount.
+ */
+export function useMetricResourceQuery(query: string, opts: MetricResourceQueryOptions) {
+  const [rows, setRows] = useState<MetricResourceRow[] | null>(null);
+  const [isLoading, setIsLoading] = useState(true);
+  const [failed, setFailed] = useState(false);
+  const abortRef = useRef<AbortController | null>(null);
+
+  const {
+    organizationId,
+    projectId,
+    environmentId,
+    defaultPeriod,
+    fillGaps,
+    refreshIntervalMs = 60_000,
+  } = opts;
+  const { period, from, to } = opts.timeRange;
+  const queuesKey = opts.queues && opts.queues.length > 0 ? opts.queues.join(",") : undefined;
+
+  const load = useCallback(() => {
+    abortRef.current?.abort();
+    const controller = new AbortController();
+    abortRef.current = controller;
+    setIsLoading(true);
+    fetch("/resources/metric", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        query,
+        scope: "environment",
+        period: period ?? (from || to ? null : defaultPeriod),
+        from,
+        to,
+        fillGaps: !!fillGaps,
+        organizationId,
+        projectId,
+        environmentId,
+        ...(queuesKey !== undefined ? { queues: queuesKey.split(",") } : {}),
+      }),
+      signal: controller.signal,
+    })
+      .then((res) => res.json() as Promise<MetricResourceResponse>)
+      .then((data) => {
+        if (controller.signal.aborted) return;
+        if (data.success) {
+          setRows(data.data.rows);
+          setFailed(false);
+        } else {
+          setFailed(true);
+        }
+        setIsLoading(false);
+      })
+      .catch((error) => {
+        if (error instanceof DOMException && error.name === "AbortError") return;
+        if (!controller.signal.aborted) {
+          setFailed(true);
+          setIsLoading(false);
+        }
+      });
+  }, [
+    query,
+    period,
+    from,
+    to,
+    defaultPeriod,
+    fillGaps,
+    organizationId,
+    projectId,
+    environmentId,
+    queuesKey,
+  ]);
+
+  useEffect(() => {
+    load();
+    return () => abortRef.current?.abort();
+  }, [load]);
+
+  useInterval({ interval: refreshIntervalMs, onLoad: false, onFocus: true, callback: load });
+
+  return { rows: rows ?? [], isLoading, showLoading: isLoading && !rows, failed };
+}