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

The dequeue brake released the moment its signal became unreadable.
`refresh()` caught any error from `source.read()` and set the verdict to
`null`, which `computeEngaged()` treats as not-engaged — so a few failed
reads dropped an engaged brake, silently, with no log and no metric.

That handling was symmetric while the risk is not. A source that has
stopped answering correlates with the pressure the brake exists for, so
releasing on read failure gives up protection at exactly the wrong
moment; holding too long only costs throughput.

Now a failed read keeps the last verdict instead of discarding it. The
verdict then ages normally, so the existing `maxVerdictAgeMs` check
becomes the grace window and still bounds how long a dead source can
hold the brake — a permanently unreachable source releases it rather
than pinning dequeuing forever. Because `computeEngaged()` only consults
staleness for an *engaged* verdict, a released one is unaffected and
stays released.

The default grace moves from 15s to 120s, comparable to how long the
brake normally stays engaged.

One guard worth calling out: holding is only safe when something bounds
it, so when `maxVerdictAgeMs` is unset the previous discard behaviour is
kept. Otherwise an unbounded hold could pin the brake indefinitely.

Read failures were previously invisible — the catch block neither logged
nor counted. Adds a `read_failures_total` counter, plus an error log on
the transition into failure rather than once per tick, since the refresh
loop runs every second.

The post-release ramp needs no change: it anchors off the
engaged-to-released transition, so a grace-window release still ramps
back up instead of snapping to full rate, which is what you want after a
blind period.

Tests cover holding while reads fail, releasing past the max age, and
the existing unbounded-config paths are unchanged.
This commit is contained in:
nicktrn
2026-08-03 17:06:23 +01:00
committed by GitHub
parent 8f9db53350
commit 3fba04573d
5 changed files with 103 additions and 11 deletions
@@ -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.
@@ -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 source reads that failed (threw). */
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 threw",
registers: [opts.register],
});
}
}
@@ -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<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,
@@ -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 = {
@@ -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<BackpressureVerdict | null>;
@@ -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<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 (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
+1 -1
View File
@@ -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(),