fix(supervisor): make dequeue latency buckets survive retry backoff; review fixes

The HTTP client retries internally (5 attempts, >=7.5s of backoff before
a retryable error surfaces), so the 5s bucket ceiling would have pushed
nearly every retried error into +Inf. Extend buckets to 30s and state
in the help text that one observation spans the whole logical call
including retries. Also: stop clobbering a caller-supplied consumer
metrics instance, correct the catch-branch comment (defensive only -
wrapZodFetch never throws), and cover the pool-to-consumer metrics
injection with tests.

Author: myftija

packages/core/src/v3/runEngineWorker/supervisor/consumerPool.test.ts

@@ -7,6 +7,8 @@ import {
 import { SupervisorHttpClient } from "./http.js";
 import type { WorkerApiDequeueResponseBody } from "./schemas.js";
 import type { QueueConsumer } from "./queueConsumer.js";
+import { ConsumerPoolMetrics } from "./consumerPoolMetrics.js";
+import { Registry } from "prom-client";
 
 // Mock only the logger
 vi.mock("../../utils/structuredLogger.js");
@@ -16,9 +18,11 @@ class TestQueueConsumer implements QueueConsumer {
   public started = false;
   public stopped = false;
   public onDequeue?: (messages: WorkerApiDequeueResponseBody) => Promise<void>;
+  public metrics?: ConsumerPoolMetrics;
 
   constructor(opts: any) {
     this.onDequeue = opts.onDequeue;
+    this.metrics = opts.metrics;
   }
 
   start(): void {
@@ -719,6 +723,38 @@ describe("RunQueueConsumerPool", () => {
     });
   });
 
+  describe("Metrics wiring", () => {
+    it("injects the pool's shared ConsumerPoolMetrics into every consumer when a registry is provided", async () => {
+      pool = new RunQueueConsumerPool({
+        ...defaultOptions,
+        metricsRegistry: new Registry(),
+        scaling: { strategy: "none", maxConsumerCount: 3 },
+      });
+
+      await pool.start();
+
+      expect(testConsumers.length).toBe(3);
+      const poolMetrics = pool["promMetrics"];
+      expect(poolMetrics).toBeInstanceOf(ConsumerPoolMetrics);
+      testConsumers.forEach((consumer) => {
+        expect(consumer.metrics).toBe(poolMetrics);
+      });
+    });
+
+    it("preserves a caller-supplied consumer metrics instance when no registry is provided", async () => {
+      const callerMetrics = new ConsumerPoolMetrics({ register: new Registry() });
+      pool = new RunQueueConsumerPool({
+        ...defaultOptions,
+        consumer: { ...defaultOptions.consumer, metrics: callerMetrics },
+        scaling: { strategy: "none", maxConsumerCount: 1 },
+      });
+
+      await pool.start();
+
+      expect(testConsumers[0]?.metrics).toBe(callerMetrics);
+    });
+  });
+
   describe("Backpressure scale-up freeze", () => {
     it("freezes scale-up while shouldPauseScaling returns true, then resumes", async () => {
       let paused = true;

packages/core/src/v3/runEngineWorker/supervisor/consumerPool.ts

@@ -370,8 +370,9 @@ export class RunQueueConsumerPool {
       const consumer = this.consumerFactory({
         ...this.consumerOptions,
         // Share the pool's single metrics instance so every consumer records onto
-        // the same histogram (re-registering the metric name would throw).
-        metrics: this.promMetrics,
+        // the same histogram (re-registering the metric name would throw). Fall
+        // back to a caller-supplied instance rather than clobbering it.
+        metrics: this.promMetrics ?? this.consumerOptions.metrics,
         onDequeue: async (messages, timing) => {
           // Always update queue length, default to 0 for empty dequeues or missing value
           this.updateQueueLength(messages[0]?.workerQueueLength ?? 0);

packages/core/src/v3/runEngineWorker/supervisor/consumerPoolMetrics.ts

@@ -117,10 +117,13 @@ export class ConsumerPoolMetrics {
 
     this.dequeueDurationSeconds = new Histogram({
       name: `${this.prefix}_dequeue_duration_seconds`,
-      help: "Client-side latency of the dequeue API call (POST /engine/v1/worker-actions/dequeue), measured around the HTTP request",
+      help: "Client-side duration of the dequeue API call (POST /engine/v1/worker-actions/dequeue), including the HTTP client's internal retries and backoff",
       labelNames: ["outcome"],
-      // Tuned for the dequeue endpoint: sub-10ms hits through a multi-second tail.
-      buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5],
+      // The HTTP client retries internally (up to 5 attempts with 0.5-5s backoff),
+      // so one observation can span multiple requests plus sleeps. A retryable
+      // failure surfaces as `error` only after >=7.5s of backoff - the 10/30s
+      // buckets exist so that mode doesn't collapse into +Inf.
+      buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 30],
       registers: [this.register],
     });
   }

packages/core/src/v3/runEngineWorker/supervisor/queueConsumer.test.ts

@@ -79,7 +79,9 @@ describe("RunQueueConsumer dequeue latency metric", () => {
     );
   });
 
-  it('records outcome="error" when the dequeue call throws (network error / timeout)', async () => {
+  // Defensive path: wrapZodFetch traps all errors today, so the real client
+  // never throws - this guards against a future client that does.
+  it('records outcome="error" when the dequeue call throws', async () => {
     await runOneDequeue({
       dequeueImpl: async () => {
         throw new Error("network down");

packages/core/src/v3/runEngineWorker/supervisor/queueConsumer.ts

@@ -152,7 +152,9 @@ export class RunQueueConsumer implements QueueConsumer {
         }
       }
     } catch (clientError) {
-      // Captures network errors and timeouts - exactly the tail latencies we care about.
+      // wrapZodFetch traps all errors into { success: false }, so this branch is
+      // unreachable with the real client today. Record defensively so a future
+      // client that throws can't silently lose error samples.
       this.metrics?.observeDequeueLatency((performance.now() - dequeueStart) / 1000, "error");
       this.logger.error("client.dequeue error", { error: clientError });
     }