fix(metrics-pipeline): use BigInt order keys and namespaced odometer counters

entryOrderKey returns a string built with BigInt math so ordering stays correct at real epoch magnitudes. Odometer keys are namespaced by definition name. The consumer reports null lag for a missing consumer group instead of 0, and empty gauge values parse as NaN rather than 0.

Author: ericallam

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

@@ -100,7 +100,8 @@ export class CachedRedisNumber {
       redis: options.redis,
       key: options.key,
       parse: (raw) => {
-        const n = raw == null ? Number.NaN : Number(raw);
+        // Number("") is 0 (not NaN), so treat blank/whitespace as missing => fallback.
+        const n = raw == null || raw.trim() === "" ? Number.NaN : Number(raw);
         return Number.isFinite(n) ? clamp(n) : fallback;
       },
       defaultValue: fallback,

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

@@ -114,7 +114,7 @@ export class MetricsStreamConsumer<TRow> {
       description: "Failed inserts (batch left pending for retry)",
       valueType: ValueType.INT,
     });
-    this.insertDuration = this.meter.createHistogram("queue_metrics.consumer.insert_duration_ms", {
+    this.insertDuration = this.meter.createHistogram("queue_metrics.consumer.insert_duration", {
       description: "Sink insert latency",
       unit: "ms",
       valueType: ValueType.INT,
@@ -302,9 +302,10 @@ export async function probeShardStates(
   for (let shard = 0; shard < keys.length; shard++) {
     const key = keys[shard]!;
     const depth = Number(await redis.xlen(key)) || 0;
-    // lag defaults to null (unknown) and only becomes a number when Redis reports one:
-    // a nil lag means entries were trimmed past the group's read position (data loss).
-    let lag: number | null = 0;
+    // lag defaults to null (unknown) and only becomes a number when the group is found and
+    // Redis reports one: a nil lag (or a missing group on an existing stream) means we can't
+    // compute it, e.g. entries were trimmed past the group's read position (data loss).
+    let lag: number | null = null;
     let pending = 0;
     try {
       const groups = (await redis.call("XINFO", "GROUPS", key)) as unknown[];

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

@@ -136,7 +136,7 @@ export class MetricsStreamEmitter {
     if (!this.flag.enabled()) return;
     const op = String(fields.op ?? "unknown");
     const q = String(fields.q ?? "");
-    const odometerKey = `queue_metrics_cum:${op}:${q}`;
+    const odometerKey = `${this.def.name}_cum:${op}:${q}`;
     const stream = streamKey(this.def, shardFor(shardKey, this.def.shardCount));
     const extra: string[] = [];
     for (const [field, value] of Object.entries(fields)) {

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

@@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest";
 import { createMetricsGaugeComputeLua } from "./lua.js";
 import { dedupTokenFromEntryIds } from "./idempotency.js";
 import { fnv1a32, shardFor } from "./hash.js";
-import { allStreamKeys, entryTimeMs, streamKey } from "./types.js";
+import { allStreamKeys, entryOrderKey, entryTimeMs, streamKey } from "./types.js";
 
 describe("shardFor", () => {
   it("is deterministic and in range", () => {
@@ -33,6 +33,17 @@ describe("stream keys", () => {
     expect(entryTimeMs("1717000000000-5")).toBe(1717000000000);
     expect(entryTimeMs("nope")).toBeNull();
   });
+
+  it("entryOrderKey stays exact and strictly monotonic at real epoch magnitudes", () => {
+    const ms = 1783000000000; // ~2026: ms*1e5 is past JS safe-integer range, so a number key
+    const k = (seq: number) => BigInt(entryOrderKey(`${ms}-${seq}`));
+    // adjacent seq within one ms must not collapse to the same key (the float bug)
+    expect(k(0)).toBe(BigInt(ms) * 100000n);
+    expect(k(1) - k(0)).toBe(1n);
+    expect(k(2) - k(1)).toBe(1n);
+    // a later ms always outranks any seq of an earlier ms
+    expect(BigInt(entryOrderKey(`${ms + 1}-0`))).toBeGreaterThan(k(99999));
+  });
 });
 
 describe("createMetricsGaugeComputeLua", () => {

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

@@ -32,9 +32,10 @@ export function entryTimeMs(id: string): number | null {
   return Number.isFinite(ms) ? ms : null;
 }
 
-// Strictly-monotonic-per-stream ordering key from a stream id (`<ms>-<seq>`): ms*1e5+seq.
-// Used to order cumulative readings for deltaSumTimestamp so within-ms ties don't misorder.
-export function entryOrderKey(id: string): number {
+// Ordering key from a stream id (`<ms>-<seq>`) = ms*1e5+seq, for deltaSumTimestamp. BigInt +
+// string because ms*1e5 exceeds JS safe-integer range at real epoch magnitudes (a number would
+// collapse nearby seq values); the ClickHouse order_key column is UInt64 and takes the string.
+export function entryOrderKey(id: string): string {
   const [ms, seq] = id.split("-");
-  return (Number(ms) || 0) * 100000 + (Number(seq) || 0);
+  return (BigInt(Number(ms) || 0) * 100000n + BigInt(Number(seq) || 0)).toString();
 }