fix(supervisor): hold the last backpressure verdict when a read fails

Author: nicktrn

.server-changes/backpressure-hold-last-verdict.md

@@ -0,0 +1,6 @@
+---
+area: supervisor
+type: fix
+---
+
+Self-hosted Kubernetes deployments no longer resume pulling work the moment the safety check can't be read. It now holds its last decision for a grace period instead of releasing after a few seconds.

apps/supervisor/src/backpressure/backpressureMetrics.ts

@@ -8,6 +8,8 @@ export class BackpressureMetrics {
   readonly dryRun: Gauge<string>;
   /** Dequeue attempts the gate skipped - or would have, in dry-run (labelled). */
   readonly skipsTotal: Counter<string>;
+  /** Verdict reads that failed; the previous verdict is held until it ages out. */
+  readonly readFailuresTotal: Counter<string>;
 
   constructor(opts: { register: Registry; prefix?: string }) {
     const prefix = opts.prefix ?? "supervisor_backpressure";
@@ -30,5 +32,11 @@ export class BackpressureMetrics {
       labelNames: ["dry_run"],
       registers: [opts.register],
     });
+
+    this.readFailuresTotal = new Counter({
+      name: `${prefix}_read_failures_total`,
+      help: "Verdict source reads that failed; the previous verdict is held until it ages out",
+      registers: [opts.register],
+    });
   }
 }

apps/supervisor/src/backpressure/backpressureMonitor.test.ts

@@ -89,6 +89,37 @@ describe("BackpressureMonitor", () => {
     monitor.stop();
   });
 
+  it("holds an engaged verdict while reads fail, then releases past the max age", async () => {
+    let call = 0;
+    const source: BackpressureSignalSource = {
+      read: async () => {
+        call++;
+        if (call === 1) {
+          return { engaged: true, ts: Date.now() };
+        }
+        throw new Error("signal source unreachable");
+      },
+    };
+    const monitor = new BackpressureMonitor({
+      enabled: true,
+      source,
+      refreshIntervalMs: 1000,
+      maxVerdictAgeMs: 15_000,
+    });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(monitor.shouldSkipDequeue()).toBe(true);
+
+    await vi.advanceTimersByTimeAsync(5000);
+    expect(monitor.shouldSkipDequeue()).toBe(true); // read failing, verdict held
+
+    await vi.advanceTimersByTimeAsync(11_000);
+    expect(monitor.shouldSkipDequeue()).toBe(false); // past max age, released
+
+    monitor.stop();
+  });
+
   it("fails open when the source reports unknown (null)", async () => {
     const { source } = countingSource(null);
     const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
@@ -292,6 +323,7 @@ describe("BackpressureMonitor", () => {
     const logs: Array<{ message: string; meta?: Record<string, unknown> }> = [];
     const logger = {
       info: (message: string, meta?: Record<string, unknown>) => logs.push({ message, meta }),
+      error: (message: string, meta?: Record<string, unknown>) => logs.push({ message, meta }),
     };
     const monitor = new BackpressureMonitor({
       enabled: true,

apps/supervisor/src/backpressure/backpressureMonitor.ts

@@ -2,6 +2,7 @@ import type { BackpressureMetrics } from "./backpressureMetrics.js";
 
 export interface BackpressureLogger {
   info(message: string, meta?: Record<string, unknown>): void;
+  error(message: string, meta?: Record<string, unknown>): void;
 }
 
 export type BackpressureVerdict = {
@@ -24,8 +25,9 @@ export type BackpressureMonitorOptions = {
   source: BackpressureSignalSource;
   refreshIntervalMs?: number;
   /**
-   * If set, a cached verdict older than this is treated as unknown (fail-open).
-   * Guards against the source silently going stale (e.g. hanging reads).
+   * If set, an engaged verdict older than this is released (fail-open), bounding how
+   * long a dead source can hold the brake. Reads that fail keep the last verdict, so
+   * this doubles as the grace window for riding out a transient source outage.
    */
   maxVerdictAgeMs?: number;
   /**
@@ -54,6 +56,7 @@ export class BackpressureMonitor {
   private refreshInFlight = false;
   private wasEngaged = false;
   private releasedAt?: number;
+  private readFailing = false;
 
   constructor(private readonly opts: BackpressureMonitorOptions) {
     this.opts.metrics?.dryRun.set(this.opts.dryRun ? 1 : 0);
@@ -152,12 +155,29 @@ export class BackpressureMonitor {
   }
 
   private async refresh(): Promise<void> {
+    let next: BackpressureVerdict | null = null;
+    let readError: unknown;
     try {
-      this.verdict = await this.opts.source.read();
-    } catch {
-      // Fail-open: a dead/unreachable source must never pin the brake. Treat as
-      // unknown (no verdict) so dequeue resumes as if backpressure were off.
-      this.verdict = null;
+      next = await this.opts.source.read();
+    } catch (error) {
+      readError = error;
+    }
+
+    if (next) {
+      this.verdict = next;
+      this.readFailing = false;
+    } else {
+      if (this.opts.maxVerdictAgeMs === undefined) {
+        this.verdict = null; // unbounded hold could pin the brake forever
+      }
+      this.opts.metrics?.readFailuresTotal.inc();
+      if (!this.readFailing) {
+        this.readFailing = true; // log once per outage, not once per tick
+        this.opts.logger?.error("backpressure read failed, holding last verdict", {
+          reason: readError ? String(readError) : "no verdict",
+          engaged: this.computeEngaged(),
+        });
+      }
     }
 
     // Track the engaged→released transition to anchor the resume ramp. Use the

apps/supervisor/src/env.ts

@@ -79,7 +79,7 @@ export const Env = z
       .number()
       .int()
       .positive()
-      .default(15_000), // Stale verdict → fail-open (treat as not engaged)
+      .default(120_000), // Grace window: held verdict older than this → fail-open
     TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST: z.string().optional(),
     TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PORT: z.coerce.number().int().optional(),
     TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_USERNAME: z.string().optional(),