fix(webapp): carry a queue's depth forward across empty buckets

The per-queue metrics route mapped ClickHouse rows straight to an array, so a
bucket with no sample shortened the trend and shifted every later point in time.
Fill a fixed-width grid the way the two sibling callers do.

Author: kathiekiwi

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

@@ -3,6 +3,7 @@ import { z } from "zod";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { logger } from "~/services/logger.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { queueDepthSeries } from "~/v3/queueDepthSeries";
 
 /**
  * Per-queue metrics over a window. `queueParam` is the queue name; `?type=task` (the default)
@@ -62,8 +63,12 @@ export const loader = createLoaderApiRoute(
     const windowMinutes = windowMs / 60_000;
     const bucketSeconds = Math.max(60, Math.round(windowMs / 1000 / TREND_POINTS));
     // Snap both bounds to the bucket grid so repeated calls share ClickHouse cache entries.
-    const endMs = Math.ceil(Date.now() / (bucketSeconds * 1000)) * bucketSeconds * 1000;
+    const bucketIntervalMs = bucketSeconds * 1000;
+    const endMs = Math.ceil(Date.now() / bucketIntervalMs) * bucketIntervalMs;
     const startMs = endMs - windowMs;
+    // The trend grid covers whole buckets, so a period that isn't a bucket multiple still lines up.
+    const gridStartMs = Math.floor(startMs / bucketIntervalMs) * bucketIntervalMs;
+    const numBuckets = Math.round((endMs - gridStartMs) / bucketIntervalMs);
 
     try {
       const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
@@ -115,12 +120,13 @@ export const loader = createLoaderApiRoute(
         startedCount,
         startedPerMin: Number((startedCount / windowMinutes).toFixed(2)),
         throttledCount: summary?.throttled_count ?? 0,
-        bucketIntervalMs: bucketSeconds * 1000,
-        // Oldest first; buckets with no sample are omitted, so gaps carry the previous depth.
-        depthTrend: (trendRows ?? [])
-          .slice()
-          .sort((a, b) => a.bucket.localeCompare(b.bucket))
-          .map((row) => row.depth),
+        bucketIntervalMs,
+        // Oldest first, one point per bucket: a bucket with no sample carries the previous depth.
+        depthTrend: queueDepthSeries(trendRows ?? [], {
+          startMs: gridStartMs,
+          bucketIntervalMs,
+          numBuckets,
+        }).depth,
       });
     } catch (error) {
       // Rethrow Responses: swallowing one would turn it into a 500.

apps/webapp/app/v3/queueDepthSeries.ts

@@ -0,0 +1,49 @@
+/**
+ * `getQueueDepthSparklines` emits a row only for buckets that reported, so a caller has to place
+ * every row on the bucket grid itself. Depth is carry-forward filled: no emission means unchanged,
+ * not zero. Throttled is not filled — only real per-bucket counts tint a bar.
+ */
+
+export type QueueDepthBucketRow = { bucket: string; depth: number; throttled: number };
+
+export type QueueDepthGrid = { startMs: number; bucketIntervalMs: number; numBuckets: number };
+
+/** Rows placed on the grid by bucket index. Rows outside the window are dropped. */
+export function indexQueueDepthRows(
+  rows: QueueDepthBucketRow[],
+  grid: QueueDepthGrid
+): Map<number, { depth: number; throttled: number }> {
+  const byIndex = new Map<number, { depth: number; throttled: number }>();
+  for (const row of rows) {
+    const bucketMs = Date.parse(row.bucket.replace(" ", "T") + "Z");
+    if (Number.isNaN(bucketMs)) continue;
+    const index = Math.round((bucketMs - grid.startMs) / grid.bucketIntervalMs);
+    if (index < 0 || index >= grid.numBuckets) continue;
+    byIndex.set(index, { depth: row.depth, throttled: row.throttled });
+  }
+  return byIndex;
+}
+
+/** A fixed-width series per grid bucket, so a gap can never shift later points in time. */
+export function fillQueueDepthSeries(
+  byIndex: Map<number, { depth: number; throttled: number }>,
+  numBuckets: number
+): { depth: number[]; throttled: number[] } {
+  const depth: number[] = new Array(numBuckets);
+  const throttled: number[] = new Array(numBuckets);
+  let last = 0;
+  for (let i = 0; i < numBuckets; i++) {
+    const bucket = byIndex.get(i);
+    if (bucket !== undefined) last = bucket.depth;
+    depth[i] = last;
+    throttled[i] = bucket?.throttled ?? 0;
+  }
+  return { depth, throttled };
+}
+
+export function queueDepthSeries(
+  rows: QueueDepthBucketRow[],
+  grid: QueueDepthGrid
+): { depth: number[]; throttled: number[] } {
+  return fillQueueDepthSeries(indexQueueDepthRows(rows, grid), grid.numBuckets);
+}

apps/webapp/test/queueDepthSeries.test.ts

@@ -0,0 +1,58 @@
+import { describe, expect, it } from "vitest";
+import { queueDepthSeries } from "~/v3/queueDepthSeries";
+
+const BUCKET_MS = 300_000;
+const START_MS = Date.parse("2026-01-01T00:00:00Z");
+
+function bucketAt(index: number): string {
+  return new Date(START_MS + index * BUCKET_MS).toISOString().slice(0, 19).replace("T", " ");
+}
+
+function row(index: number, depth: number, throttled = 0) {
+  return { bucket: bucketAt(index), depth, throttled };
+}
+
+const grid = { startMs: START_MS, bucketIntervalMs: BUCKET_MS, numBuckets: 6 };
+
+describe("queueDepthSeries", () => {
+  it("keeps later points in place when a bucket in the middle has no sample", () => {
+    // Buckets 2 and 3 never reported; 4 and 5 must stay at index 4 and 5.
+    const series = queueDepthSeries([row(0, 10), row(1, 20), row(4, 90), row(5, 95)], grid);
+
+    expect(series.depth).toEqual([10, 20, 20, 20, 90, 95]);
+  });
+
+  it("emits one point per bucket regardless of how many rows came back", () => {
+    expect(queueDepthSeries([row(3, 7)], grid).depth).toHaveLength(6);
+    expect(queueDepthSeries([], grid).depth).toHaveLength(6);
+  });
+
+  it("carries the previous depth across a gap rather than dropping to zero", () => {
+    expect(queueDepthSeries([row(0, 42)], grid).depth).toEqual([42, 42, 42, 42, 42, 42]);
+  });
+
+  it("starts at zero until the first sample", () => {
+    expect(queueDepthSeries([row(2, 5)], grid).depth).toEqual([0, 0, 5, 5, 5, 5]);
+  });
+
+  it("orders points oldest first whatever order the rows arrive in", () => {
+    expect(queueDepthSeries([row(5, 95), row(0, 10), row(2, 30)], grid).depth).toEqual([
+      10, 10, 30, 30, 30, 95,
+    ]);
+  });
+
+  it("does not carry throttled counts across a gap", () => {
+    expect(queueDepthSeries([row(0, 10, 4), row(3, 20, 1)], grid).throttled).toEqual([
+      4, 0, 0, 1, 0, 0,
+    ]);
+  });
+
+  it("drops rows outside the grid and unparseable buckets", () => {
+    const series = queueDepthSeries(
+      [row(-1, 999), row(6, 999), { bucket: "not-a-date", depth: 999, throttled: 0 }, row(1, 8)],
+      grid
+    );
+
+    expect(series.depth).toEqual([0, 8, 8, 8, 8, 8]);
+  });
+});