feat(run-engine,clickhouse,webapp): concurrency-key health metrics and live per-key breakdown

CK queues now emit two extra gauge fields from the CK-path Lua scripts:
the number of concurrency keys with queued runs (ZCARD of the ckIndex)
and the head-of-line wait of the most-starved key (now minus the oldest
ckIndex score). Both flow through the existing stream into new
max-aggregated columns on the 10s and 5m tiers, and non-CK scripts keep
the 7-field gauge shape.

The queue detail page grows a concurrency-keys section for queues with
CK activity: charts for backlogged keys and most-starved wait, plus a
live per-key table (queued, running, oldest wait) read from the ckIndex
zset, most starved first. Per-key history is intentionally not stored:
key values are user-controlled, so the per-key dimension stays in Redis
where it is bounded by the live backlog.

The queue simulator gains a tenant-hotspot scenario that stages the CK
gauge columns and a live ckIndex so the per-key table and charts render
with data.

Author: ericallam

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

@@ -19,6 +19,15 @@ import {
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { QueueRetrievePresenter } from "~/presenters/v3/QueueRetrievePresenter.server";
+import {
+  Table,
+  TableBody,
+  TableCell,
+  TableHeader,
+  TableHeaderCell,
+  TableRow,
+} from "~/components/primitives/Table";
+import { engine } from "~/v3/runEngine.server";
 import { TimeFilter } from "~/components/runs/v3/SharedFilters";
 import { useSearchParams } from "~/hooks/useSearchParam";
 import { useCurrentPlan } from "../_app.orgs.$organizationSlug/route";
@@ -58,12 +67,18 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const queue = retrieve.queue;
   const fullName = queue.type === "task" ? `task/${queue.name}` : queue.name;
 
+  const ckBreakdown = await engine.concurrencyKeyBreakdown(environment, fullName, {
+    limit: CK_LIVE_LIMIT,
+  });
+
   // Charts + CH-derived stats are fetched client-side per card (see QueueDetailChartCard /
   // useQueueMetric) so the drill-down renders instantly. The loader only returns the live
   // "now" counts + identifiers the client fetches need.
   return typedjson({
     queue,
     fullName,
+    ckBreakdown,
+    loadedAt: Date.now(),
     backPath: url.pathname.replace(/\/[^/]+$/, ""),
     ids: {
       organizationId: environment.organizationId,
@@ -81,16 +96,21 @@ const COLORS = {
   p95: "#F59E0B",
   p99: "#EF4444",
   throttled: "#F59E0B",
+  ckKeys: "#34D399",
+  ckWait: "#F59E0B",
 };
 
+const CK_LIVE_LIMIT = 50;
+
 type Ids = { organizationId: string; projectId: string; environmentId: string };
 
 type TimeRangeParams = MetricResourceTimeRange;
 
 const QUEUE_METRICS_DEFAULT_PERIOD = "1d";
 
 export default function Page() {
-  const { queue, fullName, backPath, ids } = useTypedLoaderData<typeof loader>();
+  const { queue, fullName, ckBreakdown, loadedAt, backPath, ids } =
+    useTypedLoaderData<typeof loader>();
   const plan = useCurrentPlan();
   const maxPeriodDays = plan?.v3Subscription?.plan?.limits?.queryPeriodDays?.number;
 
@@ -169,12 +189,115 @@ export default function Page() {
             queueName={fullName}
             series={[{ key: "throttled", label: "Throttled", color: COLORS.throttled }]}
           />
+          <ConcurrencyKeysSection
+            breakdown={ckBreakdown}
+            loadedAt={loadedAt}
+            ids={ids}
+            timeRange={timeRange}
+            queueName={fullName}
+          />
         </div>
       </PageBody>
     </PageContainer>
   );
 }
 
+type CkBreakdown = {
+  totalBackloggedKeys: number;
+  keys: Array<{
+    concurrencyKey: string;
+    queued: number;
+    running: number;
+    oldestEnqueuedAt: number;
+  }>;
+};
+
+// Rendered only for queues with concurrency-key activity: live keys in the ckIndex, or
+// nonzero CK history in the selected range (one cached scalar query decides).
+function ConcurrencyKeysSection({
+  breakdown,
+  loadedAt,
+  ids,
+  timeRange,
+  queueName,
+}: {
+  breakdown: CkBreakdown;
+  loadedAt: number;
+  ids: Ids;
+  timeRange: TimeRangeParams;
+  queueName: string;
+}) {
+  const { rows, showLoading } = useQueueMetric(
+    `SELECT max(max_ck_backlogged) AS peak_keys, max(max_ck_wait_ms) AS peak_wait\nFROM queue_metrics`,
+    { ids, timeRange, queueName }
+  );
+  const row = rows[0];
+  const hasHistory = row ? toNumber(row.peak_keys) > 0 || toNumber(row.peak_wait) > 0 : false;
+  const hasLive = breakdown.keys.length > 0;
+
+  if (!hasLive && (showLoading || !hasHistory)) return null;
+
+  return (
+    <>
+      <QueueDetailChartCard
+        title="Concurrency keys with backlog"
+        query={`SELECT timeBucket() AS t, max(max_ck_backlogged) AS keys\nFROM queue_metrics\nGROUP BY t\nORDER BY t`}
+        fillGaps
+        ids={ids}
+        timeRange={timeRange}
+        queueName={queueName}
+        series={[{ key: "keys", label: "Keys", color: COLORS.ckKeys }]}
+      />
+      <QueueDetailChartCard
+        title="Most-starved key wait"
+        query={`SELECT timeBucket() AS t, max(max_ck_wait_ms) AS wait\nFROM queue_metrics\nGROUP BY t\nORDER BY t`}
+        fillGaps
+        ids={ids}
+        timeRange={timeRange}
+        queueName={queueName}
+        valueFormat={formatWaitMs}
+        series={[{ key: "wait", label: "Max wait", color: COLORS.ckWait }]}
+      />
+      {hasLive && (
+        <div className="rounded-sm border border-grid-dimmed bg-background-bright">
+          <div className="flex items-baseline justify-between px-3 pt-2">
+            <div className="text-sm text-text-bright">Concurrency keys with queued runs</div>
+            <div className="text-xs text-text-dimmed">
+              {breakdown.totalBackloggedKeys > breakdown.keys.length
+                ? `Showing the ${breakdown.keys.length} most starved of ${breakdown.totalBackloggedKeys.toLocaleString()} keys`
+                : `${breakdown.totalBackloggedKeys.toLocaleString()} ${
+                    breakdown.totalBackloggedKeys === 1 ? "key" : "keys"
+                  }`}
+            </div>
+          </div>
+          <Table>
+            <TableHeader>
+              <TableRow>
+                <TableHeaderCell>Key</TableHeaderCell>
+                <TableHeaderCell alignment="right">Queued</TableHeaderCell>
+                <TableHeaderCell alignment="right">Running</TableHeaderCell>
+                <TableHeaderCell alignment="right">Oldest wait</TableHeaderCell>
+              </TableRow>
+            </TableHeader>
+            <TableBody>
+              {breakdown.keys.map((key) => (
+                <TableRow key={key.concurrencyKey}>
+                  <TableCell>{key.concurrencyKey}</TableCell>
+                  <TableCell alignment="right">{key.queued.toLocaleString()}</TableCell>
+                  <TableCell alignment="right">{key.running.toLocaleString()}</TableCell>
+                  <TableCell alignment="right">
+                    {formatWaitMs(Math.max(0, loadedAt - key.oldestEnqueuedAt))}
+                  </TableCell>
+                </TableRow>
+              ))}
+            </TableBody>
+          </Table>
+        </div>
+      )}
+    </>
+  );
+}
+
 function useQueueMetric(
   query: string,
   opts: { ids: Ids; timeRange: TimeRangeParams; queueName: string; fillGaps?: boolean }

apps/webapp/app/v3/querySchemas.ts

@@ -774,6 +774,22 @@ export const queueMetricsSchema: TableSchema = {
         fillMode: "carry",
       }),
     },
+    max_ck_backlogged: {
+      name: "max_ck_backlogged",
+      ...column("UInt32", {
+        description:
+          "Peak number of distinct concurrency keys with queued runs in the bucket. Aggregate with max(). Zero for queues that do not use concurrency keys.",
+        fillMode: "carry",
+      }),
+    },
+    max_ck_wait_ms: {
+      name: "max_ck_wait_ms",
+      ...column("UInt32", {
+        description:
+          "Worst head-of-line wait (ms) across concurrency keys in the bucket: how long the most-starved key's oldest queued run has been waiting. Aggregate with max(). Zero for queues that do not use concurrency keys.",
+        fillMode: "carry",
+      }),
+    },
     wait_ms_sum: {
       name: "wait_ms_sum",
       ...column("UInt64", {

apps/webapp/app/v3/queueMetricsMapping.ts

@@ -95,6 +95,8 @@ export function mapEntryToRow(
     row.env_running = num(f.ec);
     row.env_limit = num(f.elim);
     row.throttled = num(f.thr);
+    row.ck_backlogged = num(f.ckq);
+    row.ck_max_wait_ms = num(f.ckw);
   } else {
     // Counter op: the monotonic odometer reading + its ordering key (and wait on started).
     row.cumulative = num(f.cum);

apps/webapp/seed-queue-metrics.mts

@@ -19,6 +19,11 @@ type QueueProfile = {
   arrivals: (bucket: number, rng: Rng) => number; // expected new runs enqueued this bucket
   waitBaseMs: number;
   sparse?: boolean; // emit no rows when the queue is fully idle (tests carry-forward gaps)
+  // Concurrency-key queue: adds CK-health gauge fields + live ckIndex staging (--usage)
+  ck?: {
+    backlogged: (bucket: number, rng: Rng) => number;
+    maxWaitMs: (bucket: number, rng: Rng) => number;
+  };
 };
 type Scenario = {
   description: string;
@@ -218,6 +223,31 @@ const scenarios: Record<string, (totalBuckets: number, bucketSec: number) => Sce
     ],
   }),
 
+  // Per-tenant concurrency keys: a hog tenant periodically floods the queue and starves
+  // the others, so the CK charts (keys with backlog, most-starved wait) and the live
+  // per-key table on the queue detail page have something to show. Use with --usage.
+  "tenant-hotspot": () => ({
+    description:
+      "CK queue where a hog tenant starves others: CK charts + live key table (use --usage)",
+    envLimit: () => 40,
+    queues: [
+      {
+        name: "per-tenant",
+        limit: () => 10,
+        arrivals: (b, r) => poisson(b % 60 < 20 ? 25 : 8, r),
+        waitBaseMs: 60,
+        ck: {
+          backlogged: (b, r) => (b % 60 < 20 ? 6 + Math.round(r() * 6) : Math.round(r() * 3)),
+          maxWaitMs: (b, r) =>
+            b % 60 < 20
+              ? Math.round(lognormal(90_000, 0.5, r))
+              : Math.round(lognormal(3_000, 0.6, r)),
+        },
+      },
+      { name: "background", limit: () => 10, arrivals: (_b, r) => poisson(5, r), waitBaseMs: 40 },
+    ],
+  }),
+
   // Default: one env with a variety of queue behaviours + occasional env saturation.
   mixed: (totalBuckets) => ({
     description: "variety of queue profiles in one env, with occasional env saturation",
@@ -355,6 +385,15 @@ function simulateBucket(
       continue; // fully idle: leave a gap so carry-forward is exercised
     }
 
+    // CK-health fields stay coherent with the depth: no queued runs means no backlogged keys.
+    const ckBacklogged = profile.ck
+      ? queued[q] > 0
+        ? Math.max(1, Math.min(profile.ck.backlogged(bucket, rng), queued[q]))
+        : 0
+      : undefined;
+    const ckMaxWaitMs =
+      profile.ck && ckBacklogged ? Math.round(profile.ck.maxWaitMs(bucket, rng)) : undefined;
+
     const gauge: QueueMetricsRawV1Input = {
       ...ids,
       queue_name: profile.name,
@@ -367,6 +406,9 @@ function simulateBucket(
       env_queued: envQueued,
       env_limit: envLimit,
       throttled: queued[q] > 0 && (running[q] >= limit[q] || scale < 1) ? 1 : 0,
+      ...(ckBacklogged !== undefined
+        ? { ck_backlogged: ckBacklogged, ck_max_wait_ms: ckMaxWaitMs ?? 0 }
+        : {}),
     };
     rows.push(gauge);
 
@@ -464,10 +506,23 @@ async function stageRedisUsage(scenario: Scenario, ids: Ids, seed: number, clear
     const { createRedisClient } = await import("@internal/redis");
     const redis = createRedisClient({ host, port });
     const rng = mulberry32(seed + 1);
-    const base = `engine:runqueue:{org:${ids.organization_id}}:proj:${ids.project_id}:env:${ids.environment_id}:queue:`;
+    const prefix = "engine:runqueue:";
+    const logicalBase = `{org:${ids.organization_id}}:proj:${ids.project_id}:env:${ids.environment_id}:queue:`;
+    const base = `${prefix}${logicalBase}`;
     for (const [q, profile] of scenario.queues.entries()) {
       const key = `${base}${profile.name}:currentDequeued`;
       await redis.del(key);
+
+      // CK staging (ckIndex + per-key subqueues) feeds the live per-key table on the queue
+      // detail page. Members are stored unprefixed, exactly like the run-queue Lua does.
+      const ckIndexKey = `${base}${profile.name}:ckIndex`;
+      const lengthCounterKey = `${base}${profile.name}:lengthCounter`;
+      const staleMembers = await redis.zrange(ckIndexKey, 0, -1);
+      for (const member of staleMembers) {
+        await redis.del(`${prefix}${member}`, `${prefix}${member}:currentConcurrency`);
+      }
+      await redis.del(ckIndexKey, lengthCounterKey);
+
       if (clear) continue;
       const limit = profile.limit(0);
       // First queue rides at/over its limit, the rest at 30-90%, sparse mostly idle.
@@ -481,6 +536,35 @@ async function stageRedisUsage(scenario: Scenario, ids: Ids, seed: number, clear
       if (count > 0) {
         await redis.sadd(key, ...Array.from({ length: count }, (_v, i) => `sim_run_${i}`));
       }
+
+      if (profile.ck) {
+        const now = Date.now();
+        const tenants = 12;
+        let totalCkQueued = 0;
+        for (let t = 1; t <= tenants; t++) {
+          const tenant = `tenant-${String(t).padStart(2, "0")}`;
+          const member = `${logicalBase}${profile.name}:ck:${tenant}`;
+          const hog = t === 1;
+          const queuedCount = hog ? 40 : 1 + Math.round(rng() * 5);
+          const runningCount = hog ? limit : Math.round(rng() * 2);
+          const oldestAgeMs = hog ? 15 * 60_000 : 5_000 + Math.round(rng() * 55_000);
+          const zargs: Array<string | number> = [];
+          for (let i = 0; i < queuedCount; i++) {
+            zargs.push(now - oldestAgeMs + i * 250, `sim_${tenant}_run_${i}`);
+          }
+          await redis.zadd(`${prefix}${member}`, ...zargs);
+          if (runningCount > 0) {
+            await redis.sadd(
+              `${prefix}${member}:currentConcurrency`,
+              ...Array.from({ length: runningCount }, (_v, i) => `sim_${tenant}_running_${i}`)
+            );
+          }
+          await redis.zadd(ckIndexKey, now - oldestAgeMs, member);
+          totalCkQueued += queuedCount;
+        }
+        // The aggregate "Queued now" reads ZCARD(base) + this counter; keep them coherent.
+        await redis.set(lengthCounterKey, totalCkQueued, "EX", 24 * 3600);
+      }
     }
     await redis.quit();
     console.log(
@@ -588,7 +672,8 @@ ${lines.join("\n")}
 Example designer setup (one project per scenario):
   pnpm --filter webapp run db:seed:queue-metrics -- --scenario mixed --reset
   pnpm --filter webapp run db:seed:queue-metrics -- --scenario many-queues --project qm-many-queues --reset
-  pnpm --filter webapp run db:seed:queue-metrics -- --scenario throttled-backlog --project qm-throttled --reset`);
+  pnpm --filter webapp run db:seed:queue-metrics -- --scenario throttled-backlog --project qm-throttled --reset
+  pnpm --filter webapp run db:seed:queue-metrics -- --scenario tenant-hotspot --project qm-tenants --usage --reset`);
 }
 
 async function main() {

apps/webapp/test/queueMetricsMapping.test.ts

@@ -100,6 +100,25 @@ describe("mapEntryToRow", () => {
       })
     );
     expect(row!.event_time).toBe("2023-11-14 22:13:20");
+    // Non-CK gauges carry no CK-health tail.
+    expect(row!.ck_backlogged).toBeUndefined();
+    expect(row!.ck_max_wait_ms).toBeUndefined();
+  });
+
+  it("maps the CK-health tail on gauges from CK paths", () => {
+    const row = mapEntryToRow({
+      id: "1700000000000-0",
+      fields: { op: "gauge", q: `${q}:ck:tenant-1`, ql: "4", ckq: "3", ckw: "2500" },
+    });
+    expect(row).toEqual(
+      expect.objectContaining({
+        op: "gauge",
+        queue_name: "task/t",
+        queued: 4,
+        ck_backlogged: 3,
+        ck_max_wait_ms: 2500,
+      })
+    );
   });
 
   it("maps started with wait_ms + cumulative and drops unknown ops", () => {