fix(metrics-pipeline): drop metric emits while the metrics Redis is not ready

Without a readiness guard, every fire-and-forget emit during a metrics
Redis outage queued a command in ioredis's in-memory offline queue until
rejection. Metrics are loss-tolerant by design, so drop instead;
waitUntilReady() lets embedders await the initial connect.

Author: ericallam

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

@@ -212,6 +212,8 @@ redisTest(
       flag: { enabled: () => true },
     });
 
+    // Emits before the connection is ready are dropped by design (loss-tolerant).
+    await emitter.waitUntilReady();
     emitter.emitGauge("q1", {
       op: "gauge",
       q: "q1",

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

@@ -114,6 +114,9 @@ export class MetricsStreamEmitter {
   // the caller. Shares the counter stream (one stream family on the metrics Redis).
   emitGauge(shardKey: string, fields: MetricFields): void {
     if (!this.flag.enabled()) return;
+    // Drop rather than queue while the metrics Redis is unreachable: ioredis would hold
+    // every command in its offline queue until rejection, and metrics are loss-tolerant.
+    if (this.redis.status !== "ready") return;
     const op = String(fields.op ?? "gauge");
     const stream = streamKey(this.def, shardFor(shardKey, this.def.shardCount));
     const args: string[] = [];
@@ -134,6 +137,7 @@ export class MetricsStreamEmitter {
   // lost XADD self-heals (the next reading restates the total); the INCR is never sampled.
   emit(shardKey: string, fields: MetricFields): void {
     if (!this.flag.enabled()) return;
+    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}`;
@@ -161,6 +165,12 @@ export class MetricsStreamEmitter {
       });
   }
 
+  // Resolves once the metrics Redis connection is ready (emits before that are dropped).
+  waitUntilReady(): Promise<void> {
+    if (this.redis.status === "ready") return Promise.resolve();
+    return new Promise((resolve) => this.redis.once("ready", () => resolve()));
+  }
+
   async close(): Promise<void> {
     await this.redis.quit();
   }