feat(supervisor): add BackpressureMonitor for dequeue gating

Cached, fail-open monitor that decides whether to skip dequeues based on a
pluggable signal source. Disabled is a total no-op (no refresh, no reads).
The hot-path read is synchronous and never performs I/O; every failure mode
(source throws, returns null, or verdict goes stale) fails open.

Author: nicktrn

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

@@ -0,0 +1,154 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { BackpressureMonitor, type BackpressureSignalSource } from "./backpressureMonitor.js";
+
+function countingSource(verdict: { engaged: boolean } | null): {
+  source: BackpressureSignalSource;
+  reads: () => number;
+} {
+  let reads = 0;
+  return {
+    source: {
+      read: async () => {
+        reads++;
+        return verdict;
+      },
+    },
+    reads: () => reads,
+  };
+}
+
+describe("BackpressureMonitor", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("when disabled, never skips dequeue and never reads the signal source", () => {
+    // Even though the source would report "engaged", a disabled monitor must be
+    // a complete no-op: this is the backwards-compatibility guarantee.
+    const { source, reads } = countingSource({ engaged: true });
+    const monitor = new BackpressureMonitor({ enabled: false, source });
+
+    monitor.start();
+
+    expect(monitor.shouldSkipDequeue()).toBe(false);
+    expect(reads()).toBe(0);
+
+    monitor.stop();
+  });
+
+  it("when enabled and the source reports engaged, skips dequeue after a refresh", async () => {
+    const { source } = countingSource({ engaged: true });
+    const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0); // flush the initial async read
+
+    expect(monitor.shouldSkipDequeue()).toBe(true);
+
+    monitor.stop();
+  });
+
+  it("when enabled and the source reports clear, does not skip dequeue", async () => {
+    const { source } = countingSource({ engaged: false });
+    const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+
+    expect(monitor.shouldSkipDequeue()).toBe(false);
+
+    monitor.stop();
+  });
+
+  it("fails open (stops skipping) when the source throws", async () => {
+    let call = 0;
+    const source: BackpressureSignalSource = {
+      read: async () => {
+        call++;
+        if (call === 1) {
+          return { engaged: true };
+        }
+        throw new Error("signal source unreachable");
+      },
+    };
+    const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(monitor.shouldSkipDequeue()).toBe(true); // engaged from the first read
+
+    await vi.advanceTimersByTimeAsync(1000); // next refresh throws
+    expect(monitor.shouldSkipDequeue()).toBe(false); // fail-open: a dead source must not pin the brake
+
+    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 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+
+    expect(monitor.shouldSkipDequeue()).toBe(false);
+
+    monitor.stop();
+  });
+
+  it("fails open when the cached verdict goes stale (older than max age)", async () => {
+    // Source stops updating (e.g. hangs) after the first read; the verdict ages out.
+    const source: BackpressureSignalSource = {
+      read: async () => ({ engaged: true, ts: Date.now() }),
+    };
+    const monitor = new BackpressureMonitor({
+      enabled: true,
+      source,
+      refreshIntervalMs: 1_000_000, // effectively only the initial read fires
+      maxVerdictAgeMs: 15_000,
+    });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(monitor.shouldSkipDequeue()).toBe(true);
+
+    await vi.advanceTimersByTimeAsync(15_001); // verdict now older than max age
+    expect(monitor.shouldSkipDequeue()).toBe(false);
+
+    monitor.stop();
+  });
+
+  it("does not read the source on the hot path (reads are driven by the refresh tick)", async () => {
+    const { source, reads } = countingSource({ engaged: true });
+    const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(reads()).toBe(1); // just the initial refresh
+
+    for (let i = 0; i < 1000; i++) {
+      monitor.shouldSkipDequeue();
+    }
+
+    expect(reads()).toBe(1); // hot-path calls performed zero I/O
+
+    monitor.stop();
+  });
+
+  it("stops refreshing after stop()", async () => {
+    const { source, reads } = countingSource({ engaged: true });
+    const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    const readsAtStop = reads();
+
+    monitor.stop();
+    await vi.advanceTimersByTimeAsync(5000);
+
+    expect(reads()).toBe(readsAtStop);
+  });
+});

apps/supervisor/src/backpressure/backpressureMonitor.ts

@@ -0,0 +1,78 @@
+export type BackpressureVerdict = {
+  engaged: boolean;
+  /** Epoch ms the verdict was produced. Used for consumer-side staleness fail-open. */
+  ts?: number;
+};
+
+/**
+ * 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).
+ */
+export interface BackpressureSignalSource {
+  read(): Promise<BackpressureVerdict | null>;
+}
+
+export type BackpressureMonitorOptions = {
+  enabled: boolean;
+  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).
+   */
+  maxVerdictAgeMs?: number;
+};
+
+const DEFAULT_REFRESH_INTERVAL_MS = 1000;
+
+export class BackpressureMonitor {
+  private verdict: BackpressureVerdict | null = null;
+  private timer?: ReturnType<typeof setInterval>;
+
+  constructor(private readonly opts: BackpressureMonitorOptions) {}
+
+  start(): void {
+    if (!this.opts.enabled) {
+      return;
+    }
+
+    void this.refresh();
+    this.timer = setInterval(
+      () => void this.refresh(),
+      this.opts.refreshIntervalMs ?? DEFAULT_REFRESH_INTERVAL_MS
+    );
+  }
+
+  stop(): void {
+    if (this.timer) {
+      clearInterval(this.timer);
+      this.timer = undefined;
+    }
+  }
+
+  /** Hot-path read: synchronous, never performs I/O. */
+  shouldSkipDequeue(): boolean {
+    const verdict = this.verdict;
+    if (verdict?.engaged !== true) {
+      return false;
+    }
+
+    const maxAge = this.opts.maxVerdictAgeMs;
+    if (maxAge !== undefined && verdict.ts !== undefined && Date.now() - verdict.ts > maxAge) {
+      return false;
+    }
+
+    return true;
+  }
+
+  private async refresh(): Promise<void> {
+    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;
+    }
+  }
+}