From e8cfce99556852b978f04401f6855f83267fa8f9 Mon Sep 17 00:00:00 2001 From: nicktrn <55853254+nicktrn@users.noreply.github.com> Date: Thu, 4 Jun 2026 18:35:20 +0100 Subject: [PATCH] feat(supervisor): backpressure dry-run mode and prometheus metrics Dry-run (default on via env) keeps the gates inert while computeEngaged still reflects the real signal and verdict transitions are logged. Adds BackpressureMetrics (engaged/dry_run gauges, skipped-dequeues counter). --- .../src/backpressure/backpressureMetrics.ts | 34 +++++++++ .../backpressure/backpressureMonitor.test.ts | 70 +++++++++++++++++++ .../src/backpressure/backpressureMonitor.ts | 53 ++++++++++++-- apps/supervisor/src/env.ts | 3 + apps/supervisor/src/index.ts | 5 ++ 5 files changed, 158 insertions(+), 7 deletions(-) create mode 100644 apps/supervisor/src/backpressure/backpressureMetrics.ts diff --git a/apps/supervisor/src/backpressure/backpressureMetrics.ts b/apps/supervisor/src/backpressure/backpressureMetrics.ts new file mode 100644 index 000000000..ffe576285 --- /dev/null +++ b/apps/supervisor/src/backpressure/backpressureMetrics.ts @@ -0,0 +1,34 @@ +import { Counter, Gauge, type Registry } from "prom-client"; + +/** Prometheus metrics for dequeue backpressure. */ +export class BackpressureMetrics { + /** 1 while backpressure is engaged (computed signal, set even in dry-run). */ + readonly engaged: Gauge; + /** 1 when running in dry-run (gates inert). */ + readonly dryRun: Gauge; + /** Dequeue attempts the gate skipped - or would have, in dry-run (labelled). */ + readonly skipsTotal: Counter; + + constructor(opts: { register: Registry; prefix?: string }) { + const prefix = opts.prefix ?? "supervisor_backpressure"; + + this.engaged = new Gauge({ + name: `${prefix}_engaged`, + help: "1 while dequeue backpressure is engaged (computed signal, regardless of dry-run)", + registers: [opts.register], + }); + + this.dryRun = new Gauge({ + name: `${prefix}_dry_run`, + help: "1 when dequeue backpressure is in dry-run mode (gates inert)", + registers: [opts.register], + }); + + this.skipsTotal = new Counter({ + name: `${prefix}_skipped_dequeues_total`, + help: "Dequeue attempts skipped by backpressure (or would be, in dry-run)", + labelNames: ["dry_run"], + registers: [opts.register], + }); + } +} diff --git a/apps/supervisor/src/backpressure/backpressureMonitor.test.ts b/apps/supervisor/src/backpressure/backpressureMonitor.test.ts index 0c5c8a214..0b9095798 100644 --- a/apps/supervisor/src/backpressure/backpressureMonitor.test.ts +++ b/apps/supervisor/src/backpressure/backpressureMonitor.test.ts @@ -1,5 +1,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { Registry } from "prom-client"; import { BackpressureMonitor, type BackpressureSignalSource } from "./backpressureMonitor.js"; +import { BackpressureMetrics } from "./backpressureMetrics.js"; function countingSource(verdict: { engaged: boolean } | null): { source: BackpressureSignalSource; @@ -225,6 +227,74 @@ describe("BackpressureMonitor", () => { monitor.stop(); }); + it("in dry-run, the gates are inert but computeEngaged still reflects the real signal", async () => { + const { source } = countingSource({ engaged: true }); + const monitor = new BackpressureMonitor({ + enabled: true, + source, + refreshIntervalMs: 1000, + dryRun: true, + }); + + monitor.start(); + await vi.advanceTimersByTimeAsync(0); + + expect(monitor.computeEngaged()).toBe(true); // real signal, for observability/metrics + expect(monitor.isEngaged()).toBe(false); // inert: no scale-up freeze + expect(monitor.shouldSkipDequeue()).toBe(false); // inert: no dequeue skip + + monitor.stop(); + }); + + it("logs on verdict transitions", async () => { + let engaged = true; + const source: BackpressureSignalSource = { read: async () => ({ engaged }) }; + const logs: Array<{ message: string; meta?: Record }> = []; + const logger = { + info: (message: string, meta?: Record) => logs.push({ message, meta }), + }; + const monitor = new BackpressureMonitor({ + enabled: true, + source, + refreshIntervalMs: 1000, + logger, + }); + + monitor.start(); + await vi.advanceTimersByTimeAsync(0); + expect(logs.some((l) => l.meta?.engaged === true)).toBe(true); + + engaged = false; + await vi.advanceTimersByTimeAsync(1000); + expect(logs.some((l) => l.meta?.engaged === false)).toBe(true); + + monitor.stop(); + }); + + it("records prometheus metrics", async () => { + const { source } = countingSource({ engaged: true }); + const register = new Registry(); + const metrics = new BackpressureMetrics({ register }); + const monitor = new BackpressureMonitor({ + enabled: true, + source, + refreshIntervalMs: 1000, + metrics, + }); + + monitor.start(); + await vi.advanceTimersByTimeAsync(0); + + expect(await register.metrics()).toContain("supervisor_backpressure_engaged 1"); + + monitor.shouldSkipDequeue(); + expect(await register.metrics()).toMatch( + /supervisor_backpressure_skipped_dequeues_total\{dry_run="false"\} [1-9]/ + ); + + monitor.stop(); + }); + it("resumes instantly when no ramp is configured", async () => { let engaged = true; const source: BackpressureSignalSource = { read: async () => ({ engaged }) }; diff --git a/apps/supervisor/src/backpressure/backpressureMonitor.ts b/apps/supervisor/src/backpressure/backpressureMonitor.ts index d1aa85483..381f2dbfb 100644 --- a/apps/supervisor/src/backpressure/backpressureMonitor.ts +++ b/apps/supervisor/src/backpressure/backpressureMonitor.ts @@ -1,3 +1,9 @@ +import type { BackpressureMetrics } from "./backpressureMetrics.js"; + +export interface BackpressureLogger { + info(message: string, meta?: Record): void; +} + export type BackpressureVerdict = { engaged: boolean; /** Epoch ms the verdict was produced. Used for consumer-side staleness fail-open. */ @@ -31,6 +37,13 @@ export type BackpressureMonitorOptions = { rampMs?: number; /** Injectable RNG for the resume ramp; defaults to Math.random. */ random?: () => number; + /** + * When true, the gates are inert (never skip dequeues, never freeze scale-up). + * computeEngaged() still reflects the real signal so it can be observed. + */ + dryRun?: boolean; + logger?: BackpressureLogger; + metrics?: BackpressureMetrics; }; const DEFAULT_REFRESH_INTERVAL_MS = 1000; @@ -41,7 +54,9 @@ export class BackpressureMonitor { private wasEngaged = false; private releasedAt?: number; - constructor(private readonly opts: BackpressureMonitorOptions) {} + constructor(private readonly opts: BackpressureMonitorOptions) { + this.opts.metrics?.dryRun.set(this.opts.dryRun ? 1 : 0); + } start(): void { if (!this.opts.enabled) { @@ -63,11 +78,11 @@ export class BackpressureMonitor { } /** - * Hard backpressure state: true while the (fresh) verdict says engaged. This is - * the signal for freezing consumer-pool scale-up - distinct from the dequeue - * gate, which additionally ramps after release. Hot-path read, no I/O. + * Raw hard backpressure state: true while the (fresh) verdict says engaged, + * ignoring dry-run. Used for observability/metrics so the real signal is + * visible even when the gates are inert. */ - isEngaged(): boolean { + computeEngaged(): boolean { const verdict = this.verdict; if (verdict?.engaged !== true) { return false; @@ -81,9 +96,25 @@ export class BackpressureMonitor { return true; } - /** Hot-path read: synchronous, never performs I/O. */ + /** + * Effective hard state: the signal for freezing consumer-pool scale-up. Inert + * (false) in dry-run. Hot-path read, no I/O. + */ + isEngaged(): boolean { + return this.opts.dryRun ? false : this.computeEngaged(); + } + + /** Hot-path read: synchronous, never performs I/O. Inert (false) in dry-run. */ shouldSkipDequeue(): boolean { - if (this.isEngaged()) { + const wouldSkip = this.computeShouldSkip(); + if (wouldSkip) { + this.opts.metrics?.skipsTotal.inc({ dry_run: this.opts.dryRun ? "true" : "false" }); + } + return this.opts.dryRun ? false : wouldSkip; + } + + private computeShouldSkip(): boolean { + if (this.computeEngaged()) { return true; } @@ -113,6 +144,14 @@ export class BackpressureMonitor { // Track the engaged→released transition to anchor the resume ramp. Based on // the raw refreshed verdict, not the staleness-adjusted read. const nowEngaged = this.verdict?.engaged === true; + this.opts.metrics?.engaged.set(nowEngaged ? 1 : 0); + + if (nowEngaged !== this.wasEngaged) { + this.opts.logger?.info("backpressure verdict changed", { + engaged: nowEngaged, + dryRun: !!this.opts.dryRun, + }); + } if (this.wasEngaged && !nowEngaged) { this.releasedAt = Date.now(); } diff --git a/apps/supervisor/src/env.ts b/apps/supervisor/src/env.ts index 3ff764427..176b98a8b 100644 --- a/apps/supervisor/src/env.ts +++ b/apps/supervisor/src/env.ts @@ -52,6 +52,9 @@ const Env = z // while the worker cluster can't schedule pods. Disabled = total no-op: no Redis // client is created, no reads happen, and the dequeue loop is unaffected. TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: BoolEnv.default(false), + // Safety default: even when enabled, backpressure only logs what it would do. + // Set to false to actually skip dequeues / freeze scale-up. + TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN: BoolEnv.default(true), TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_KEY: z.string().default("engine:dequeue:backpressure"), TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS: z.coerce.number().int().positive().default(1000), TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS: z.coerce.number().int().min(0).default(30_000), // Resume ramp window after release; 0 = instant resume diff --git a/apps/supervisor/src/index.ts b/apps/supervisor/src/index.ts index d3b3eb544..64052cadd 100644 --- a/apps/supervisor/src/index.ts +++ b/apps/supervisor/src/index.ts @@ -31,6 +31,7 @@ import { extractTraceparent, getRestoreRunnerId } from "./util.js"; import { createRedisClient } from "@internal/redis"; import { BackpressureMonitor } from "./backpressure/backpressureMonitor.js"; import { RedisBackpressureSignalSource } from "./backpressure/redisBackpressureSignalSource.js"; +import { BackpressureMetrics } from "./backpressure/backpressureMetrics.js"; import { fromContext, recordPhaseSince, @@ -209,6 +210,9 @@ class ManagedSupervisor { refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS, maxVerdictAgeMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS, rampMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS, + dryRun: env.TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN, + logger: this.logger, + metrics: new BackpressureMetrics({ register }), }); this.logger.log("🛑 Dequeue backpressure enabled", { @@ -216,6 +220,7 @@ class ManagedSupervisor { refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS, maxVerdictAgeMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS, rampMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS, + dryRun: env.TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN, }); }