test(run-engine,metrics-pipeline): bound emitter-readiness waits and cover the consumer round-trip

An unreachable Redis leaves waitUntilReady pending forever, so the
bounded wait fails fast with a descriptive error instead of burning the
test timeout. The consumer round-trip test gains the same readiness
wait its gauge sibling already had, closing the remaining first-emission
drop flake.

Author: matt-aitken

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

@@ -43,6 +43,7 @@ redisTest(
     });
 
     await consumer.start();
+    await emitter.waitUntilReady();
     emitter.emit("queueA", { op: "enqueue", q: "queueA" });
     emitter.emit("queueB", { op: "started", q: "queueB", wait: 42 });
 

internal-packages/run-engine/src/run-queue/metrics.test.ts

@@ -24,6 +24,17 @@ const authenticatedEnvDev = {
   organization: { id: "o1234" },
 };
 
+// A dead Redis leaves waitUntilReady() pending forever (the client retries
+// indefinitely), which would burn the whole test timeout with no diagnostic.
+async function emitterReady(emitter: MetricsStreamEmitter) {
+  await Promise.race([
+    emitter.waitUntilReady(),
+    setTimeout(15_000).then(() => {
+      throw new Error("metrics emitter Redis connection never became ready");
+    }),
+  ]);
+}
+
 async function readAllEntries(
   redisOptions: {
     host: string;
@@ -81,7 +92,7 @@ describe("RunQueue queue-metrics emission", () => {
       definition,
       flag: { enabled: () => true },
     });
-    await emitter.waitUntilReady();
+    await emitterReady(emitter);
 
     const queue = new RunQueue({
       name: "rq",
@@ -184,7 +195,7 @@ describe("RunQueue queue-metrics emission", () => {
         definition,
         flag: { enabled: () => true },
       });
-      await emitter.waitUntilReady();
+      await emitterReady(emitter);
       const queue = new RunQueue({
         name: "rq",
         tracer: trace.getTracer("rq"),
@@ -257,7 +268,7 @@ describe("RunQueue queue-metrics emission", () => {
       maxLen: 1000,
     };
     const emitter = new MetricsStreamEmitter({ redis, definition, flag: { enabled: () => true } });
-    await emitter.waitUntilReady();
+    await emitterReady(emitter);
     const queue = new RunQueue({
       name: "rq",
       tracer: trace.getTracer("rq"),
@@ -357,7 +368,7 @@ describe("RunQueue queue-metrics emission", () => {
       flag: { enabled: () => true },
       gaugeSampleRate: 0,
     });
-    await emitter.waitUntilReady();
+    await emitterReady(emitter);
     const queue = new RunQueue({
       name: "rq",
       tracer: trace.getTracer("rq"),