fix(webapp,metrics-pipeline): review hardening for overflow counters, cluster slots, and stream caps

Counter readings for names past the per-env cardinality cap are dropped
instead of merging unrelated odometers under the overflow label (gauges
still flow). The odometer key now shares the stream's shard hash tag so
the INCR plus XADD script stays in one Cluster slot. The counter stream
cap defaults lower when the stream shares the run-queue Redis. The
per-bucket counter boundary undercount is documented on the delta
columns.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -896,7 +896,9 @@ const EnvironmentSchema = z
     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),
+    // Default depends on where the stream lives: see metricsDefinition() in
+    // queueMetrics.server.ts (2M on the shared run-queue Redis, 8M on a dedicated one).
+    QUEUE_METRICS_COUNTER_STREAM_MAXLEN: z.coerce.number().int().optional(),
     // 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),

apps/webapp/app/v3/querySchemas.ts

@@ -671,7 +671,7 @@ export const queueMetricsSchema: TableSchema = {
       mergeGroupKey: "queue",
       ...column("String", {
         description:
-          "Runs enqueued (cumulative-counter delta). Read with deltaSumTimestampMerge(enqueue_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues.",
+          "Runs enqueued (cumulative-counter delta). Read with deltaSumTimestampMerge(enqueue_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues. Per-bucket values can undercount by one inter-reading delta at bucket boundaries (the bridge lives in the prior bucket's state); totals over the whole range are exact.",
       }),
       groupable: false,
       sortable: false,
@@ -682,7 +682,7 @@ export const queueMetricsSchema: TableSchema = {
       mergeGroupKey: "queue",
       ...column("String", {
         description:
-          "Runs dequeued/started (throughput). Read with deltaSumTimestampMerge(started_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues.",
+          "Runs dequeued/started (throughput). Read with deltaSumTimestampMerge(started_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues. Per-bucket values can undercount by one inter-reading delta at bucket boundaries (the bridge lives in the prior bucket's state); totals over the whole range are exact.",
         coreColumn: true,
       }),
       groupable: false,

apps/webapp/app/v3/queueMetrics.server.ts

@@ -56,11 +56,14 @@ function metricsRedisOptions(): RedisOptions {
 // One stream family on the metrics Redis carrying both gauge snapshots and cumulative
 // counter readings; one consumer group reads it.
 function metricsDefinition(): MetricDefinition {
+  // A stalled consumer holds up to maxLen entries per shard in Redis memory: cap lower
+  // by default when the stream shares the queue-critical run-queue Redis.
+  const defaultMaxLen = env.QUEUE_METRICS_REDIS_HOST ? 8_000_000 : 2_000_000;
   return {
     name: "queue_metrics",
     shardCount: env.QUEUE_METRICS_STREAM_SHARD_COUNT,
     consumerGroup: "queue_metrics_cg",
-    maxLen: env.QUEUE_METRICS_COUNTER_STREAM_MAXLEN,
+    maxLen: env.QUEUE_METRICS_COUNTER_STREAM_MAXLEN ?? defaultMaxLen,
   };
 }
 

apps/webapp/app/v3/queueMetricsMapping.ts

@@ -74,6 +74,10 @@ export function mapEntryToRow(
   if (!descriptor || !descriptor.queue_name) return null;
   if (limiter) {
     descriptor.queue_name = limiter.limit(descriptor.environment_id, descriptor.queue_name);
+    // Overflowed names share one row, but counter readings are per-queue odometers and
+    // merging different odometers under one key produces garbage deltas: drop counters
+    // for overflow, keep gauges (max across the overflow set is still meaningful).
+    if (descriptor.queue_name === OVERFLOW_QUEUE_NAME && op !== "gauge") return null;
   }
 
   const eventMs = entryTimeMs(entry.id) ?? Date.now();

apps/webapp/test/queueMetricsMapping.test.ts

@@ -122,14 +122,27 @@ describe("mapEntryToRow", () => {
     expect(mapEntryToRow({ id: "1-0", fields: { op: "ack" } })).toBeNull();
   });
 
-  it("applies the queue-name limiter", () => {
+  it("applies the queue-name limiter: gauges overflow, counters drop", () => {
     const limiter = new QueueNameLimiter(1);
     const first = mapEntryToRow({ id: "1-0", fields: { op: "ack", q } }, limiter);
     expect(first!.queue_name).toBe("task/t");
-    const second = mapEntryToRow(
-      { id: "1-1", fields: { op: "ack", q: "{org:o1}:proj:p1:env:e1:queue:task/other" } },
+
+    // Overflowed gauges keep flowing under the shared name (max stays meaningful).
+    const overflowGauge = mapEntryToRow(
+      {
+        id: "1-1",
+        fields: { op: "gauge", q: "{org:o1}:proj:p1:env:e1:queue:task/other", ql: "3" },
+      },
+      limiter
+    );
+    expect(overflowGauge!.queue_name).toBe(OVERFLOW_QUEUE_NAME);
+
+    // Overflowed counters are dropped: merging distinct odometers under one key
+    // produces garbage deltas.
+    const overflowCounter = mapEntryToRow(
+      { id: "1-2", fields: { op: "ack", q: "{org:o1}:proj:p1:env:e1:queue:task/other", cum: "4" } },
       limiter
     );
-    expect(second!.queue_name).toBe(OVERFLOW_QUEUE_NAME);
+    expect(overflowCounter).toBeNull();
   });
 });

internal-packages/metrics-pipeline/src/emitter.ts

@@ -140,8 +140,13 @@ export class MetricsStreamEmitter {
     if (this.redis.status !== "ready") return;
     const op = String(fields.op ?? "unknown");
     const q = String(fields.q ?? "");
-    const odometerKey = `${this.def.name}_cum:${op}:${q}`;
-    const stream = streamKey(this.def, shardFor(shardKey, this.def.shardCount));
+    const shard = shardFor(shardKey, this.def.shardCount);
+    const stream = streamKey(this.def, shard);
+    // The odometer carries the stream's {shard} hash tag so INCR + XADD stay in one
+    // Cluster slot (the shard is derived from the queue, so the mapping is stable).
+    // The key format is part of the rolling-deploy data shape: concurrent old/new
+    // emitters with different formats split an odometer and corrupt its deltas.
+    const odometerKey = `${this.def.name}_cum:{${shard}}:${op}:${q}`;
     const extra: string[] = [];
     for (const [field, value] of Object.entries(fields)) {
       if (field === "op" || field === "q") continue;