diff --git a/.server-changes/backpressure-hold-last-verdict.md b/.server-changes/backpressure-hold-last-verdict.md new file mode 100644 index 00000000000..18f3ca7fc50 --- /dev/null +++ b/.server-changes/backpressure-hold-last-verdict.md @@ -0,0 +1,6 @@ +--- +area: supervisor +type: fix +--- + +When the capacity signal drops out, the last decision is held for a grace period rather than released. diff --git a/apps/supervisor/src/backpressure/backpressureMetrics.ts b/apps/supervisor/src/backpressure/backpressureMetrics.ts index ffe57628548..9b622357a38 100644 --- a/apps/supervisor/src/backpressure/backpressureMetrics.ts +++ b/apps/supervisor/src/backpressure/backpressureMetrics.ts @@ -8,6 +8,8 @@ export class BackpressureMetrics { readonly dryRun: Gauge; /** Dequeue attempts the gate skipped - or would have, in dry-run (labelled). */ readonly skipsTotal: Counter; + /** Verdict source reads that failed (threw). */ + readonly readFailuresTotal: Counter; 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 threw", + registers: [opts.register], + }); } } diff --git a/apps/supervisor/src/backpressure/backpressureMonitor.test.ts b/apps/supervisor/src/backpressure/backpressureMonitor.test.ts index 7af28ffc9f5..e6c00394bbb 100644 --- a/apps/supervisor/src/backpressure/backpressureMonitor.test.ts +++ b/apps/supervisor/src/backpressure/backpressureMonitor.test.ts @@ -89,6 +89,60 @@ 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("releases immediately on an explicit null even when a grace window is configured", async () => { + let engaged: boolean | null = true; + const source: BackpressureSignalSource = { + read: async () => (engaged === null ? null : { engaged, ts: Date.now() }), + }; + const monitor = new BackpressureMonitor({ + enabled: true, + source, + refreshIntervalMs: 1000, + maxVerdictAgeMs: 15_000, + }); + + monitor.start(); + await vi.advanceTimersByTimeAsync(0); + expect(monitor.shouldSkipDequeue()).toBe(true); + + engaged = null; + await vi.advanceTimersByTimeAsync(1000); + expect(monitor.shouldSkipDequeue()).toBe(false); // null is an answer, not a failure + + 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 +346,7 @@ describe("BackpressureMonitor", () => { const logs: Array<{ message: string; meta?: Record }> = []; const logger = { info: (message: string, meta?: Record) => logs.push({ message, meta }), + error: (message: string, meta?: Record) => logs.push({ message, meta }), }; const monitor = new BackpressureMonitor({ enabled: true, diff --git a/apps/supervisor/src/backpressure/backpressureMonitor.ts b/apps/supervisor/src/backpressure/backpressureMonitor.ts index 6b4170697e5..aa16fdeaa60 100644 --- a/apps/supervisor/src/backpressure/backpressureMonitor.ts +++ b/apps/supervisor/src/backpressure/backpressureMonitor.ts @@ -2,6 +2,7 @@ import type { BackpressureMetrics } from "./backpressureMetrics.js"; export interface BackpressureLogger { info(message: string, meta?: Record): void; + error(message: string, meta?: Record): void; } export type BackpressureVerdict = { @@ -11,9 +12,10 @@ export type BackpressureVerdict = { }; /** - * Source of the current backpressure verdict. `read()` returns `null` when the - * verdict is unknown (missing/unreadable) - the monitor treats unknown as - * "not engaged" (fail-open). + * Source of the current backpressure verdict. `read()` returns `null` when the source + * answered but there is no verdict - the monitor treats that as "not engaged" + * (fail-open). A thrown error is different: the read itself failed, so the monitor + * keeps the previous verdict until it ages past `maxVerdictAgeMs`. */ export interface BackpressureSignalSource { read(): Promise; @@ -24,8 +26,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 +57,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 +156,31 @@ export class BackpressureMonitor { } private async refresh(): Promise { + 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 (readError === undefined) { + this.verdict = next; // an explicit null means "no pressure", so honour it + this.readFailing = false; + } else { + const held = this.opts.maxVerdictAgeMs !== undefined; + if (!held) { + 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", { + reason: String(readError), + heldPreviousVerdict: held, + engaged: this.computeEngaged(), + }); + } } // Track the engaged→released transition to anchor the resume ramp. Use the diff --git a/apps/supervisor/src/env.ts b/apps/supervisor/src/env.ts index 2d96cb1e407..9642d7bf659 100644 --- a/apps/supervisor/src/env.ts +++ b/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(),