diff --git a/changelog.d/fixes/13821-resource-pressure-gate-never-resamples.md b/changelog.d/fixes/13821-resource-pressure-gate-never-resamples.md new file mode 100644 index 0000000000..81222520cc --- /dev/null +++ b/changelog.d/fixes/13821-resource-pressure-gate-never-resamples.md @@ -0,0 +1 @@ +- fix(resilience): the chat admission gate's pressure check now actively re-samples instead of reading a passive cache, so the `resource_pressure` guard can observe recovery and stop shedding once real pressure clears, instead of requiring a full process restart diff --git a/src/shared/middleware/chatBodyAdmission.ts b/src/shared/middleware/chatBodyAdmission.ts index f838e92224..ef38f4086a 100644 --- a/src/shared/middleware/chatBodyAdmission.ts +++ b/src/shared/middleware/chatBodyAdmission.ts @@ -38,6 +38,7 @@ import { type IngestBudgetAcquireResult, } from "./ingestByteAdmission"; import { + checkResourcePressureGuard, getResourcePressureObservation, type PressureSeverity, } from "@omniroute/open-sse/utils/resourcePressure.ts"; @@ -217,10 +218,45 @@ export type ChatAdmissionShedReason = | "inflight_bytes_budget" | "resource_pressure"; -/** Read cached pressure severity; sampling failures must not cause false sheds. */ +/** + * Read pressure severity for admission decisions. + * + * This MUST drive an active re-sample (`checkResourcePressureGuard`), not a + * passive cache read of `getResourcePressureObservation`. The resource-pressure + * runtime only refreshes its sample and re-evaluates recovery from *inside* + * `check()` (via `scheduleRefresh`) — nothing else in the singleton mutates + * `state` or schedules a refresh. The structural admission gate that calls + * this function runs *before* every other code path that would otherwise call + * `check()` (`handleChatCore`, `checkResourcePressureBeforeProviderWork`, + * `AdaptiveAdmissionRuntimeImpl.acquire`) — so once `state.severity` flips to + * "critical", a passive read here sheds every subsequent request before any + * of those downstream paths can run, which means `check()` never gets called + * again and the guard can never observe recovery. See + * https://github.com/diegosouzapw/OmniRoute/issues/13821. + * + * `checkResourcePressureGuard()` is cheap on the hot path: it only does a + * synchronous `process.memoryUsage()` read plus a timestamp comparison per + * call; the actual signal sampling (`/proc/pressure/memory`, cgroup reads) + * happens asynchronously via `scheduleRefresh()` and is throttled by + * `staleAfterMs`, so calling this on every admitted request does not add + * per-request I/O. + * + * A non-null guard is this request's authoritative "shed now" answer and maps + * to "critical". A null guard means this request is not shed, but the + * observation's cached label can still read "critical" for a few more + * milliseconds until the async refresh settles (or if the last real sample + * merely went stale — `check()`'s own `maxStaleMs` fallback) — reporting that + * stale "critical" label to callers that branch on severity (e.g. the queue + * wait sizing at admitChatRequest's `reserve()`) would just re-introduce the + * same "never downgrades" problem for the "high" queueing bucket, so it is + * downgraded to "high" here instead. + */ export function defaultPressureSeverity(): PressureSeverity { try { - return getResourcePressureObservation().state.severity; + const guard = checkResourcePressureGuard(); + if (guard) return "critical"; + const severity = getResourcePressureObservation().state.severity; + return severity === "critical" ? "high" : severity; } catch { return "normal"; } diff --git a/tests/unit/resource-pressure-gate-recovery.test.ts b/tests/unit/resource-pressure-gate-recovery.test.ts new file mode 100644 index 0000000000..564903f5df --- /dev/null +++ b/tests/unit/resource-pressure-gate-recovery.test.ts @@ -0,0 +1,132 @@ +// Regression for https://github.com/diegosouzapw/OmniRoute/issues/13821. +// +// The structural admission gate (chatBodyAdmission.ts, admitChatRequest) is +// the FIRST caller in the request path to consult pressure severity, ahead of +// every other code path that would otherwise call checkResourcePressureGuard() +// (handleChatCore, checkResourcePressureBeforeProviderWork, +// AdaptiveAdmissionRuntimeImpl.acquire). In production, one of those other +// paths is what first observes a real critical condition (a request that +// slips past the gate before it starts shedding, or the AdaptiveAdmission +// runtime for a different combo route) and flips the singleton's cached +// `state.severity` to "critical" via the sustained-sample tracker. From that +// point on, the structural gate sheds every subsequent request before any of +// those downstream paths can run again — so `check()` never gets called +// again and the guard can never observe recovery, even after the real +// condition clears. Only a full process restart clears it. +// +// This test drives the singleton to "critical" through the SAME sustained +// sample-and-recover path production uses (classifyAdaptiveResourcePressure's +// v8_heap_ratio + a two-sample streak, not the synchronous immediate-heap +// escape hatch), using an injected mock clock so no scheduled refresh from +// the setup phase can resolve on its own and contaminate the assertion. Only +// the exact calls under test (`defaultPressureSeverity`, twice) are allowed +// to drive anything after the singleton is latched critical. +import test from "node:test"; +import assert from "node:assert/strict"; + +const { defaultPressureSeverity } = + await import("../../src/shared/middleware/chatBodyAdmission.ts"); +const { reloadResourcePressureRuntime, checkResourcePressureGuard } = + await import("../../open-sse/utils/resourcePressure.ts"); + +const MiB = 1024 * 1024; + +function signals(observedAtMs: number, heapUsedMb: number) { + return { + observedAtMs, + v8: { heapUsedBytes: heapUsedMb * MiB, heapLimitBytes: 1000 * MiB }, + process: { + rssBytes: 0, + externalBytes: 0, + arrayBuffersBytes: 0, + availableBytes: null, + constrainedBytes: null, + }, + cgroup: { currentBytes: null, maxBytes: null, highBytes: null, fileBytes: null, events: null }, + psi: null, + }; +} + +/** Advances the mock clock and drives one refresh cycle to completion. */ +async function tick(runtime: { whenRefreshSettled: () => Promise }): Promise { + checkResourcePressureGuard(); + await runtime.whenRefreshSettled(); +} + +test("defaultPressureSeverity recovers to normal once the underlying pressure clears, driven only by repeated calls to itself", async () => { + let underPressure = true; + let clockMs = 0; + const staleAfterMs = 1_000; + const runtime = reloadResourcePressureRuntime({ + heapThresholdMb: null, + immediateHeapUsedMb: () => 0, // never trip the synchronous escape hatch — force the sample path + nowMs: () => clockMs, + sample: async () => signals(clockMs, underPressure ? 950 : 100), // 950/1000 = 0.95 >= criticalRatio(0.92) + staleAfterMs, + }); + + try { + // Drive two sustained critical samples (default sustainedSamplesCritical + // is 2) via checkResourcePressureGuard directly, NOT defaultPressureSeverity — + // this mirrors some other request's handleChatCore call being the thing + // that first observes the real condition in production, not the gate + // itself. Each call is followed by advancing the clock past staleAfterMs + // so the NEXT call is the one that schedules and awaits the next sample. + await tick(runtime); + clockMs += staleAfterMs + 1; + await tick(runtime); + + // The seed is fully settled now; nextRefreshAtMs is in the past relative + // to the current clock only once we advance it again below — right now + // there is nothing scheduled, so nothing can resolve on its own. + assert.equal(defaultPressureSeverity(), "critical"); + + // The underlying condition clears and enough time passes for the next + // sample to be due. From here on nothing but defaultPressureSeverity's + // own two calls touches the singleton. + underPressure = false; + clockMs += staleAfterMs + 1; + + // This call's synchronous return still reflects the pre-refresh cached + // decision (matches production: check() answers instantly, the resample + // it schedules resolves in the background) — both the old passive read + // and the fixed active read must still say "critical" here. + assert.equal(defaultPressureSeverity(), "critical"); + + // Let whatever got scheduled by the call above resolve. Before the fix, + // defaultPressureSeverity never called check() at all, so nothing was + // scheduled here and this is a no-op — the singleton stays latched at + // "critical" forever. The fix must have scheduled and awaited a real + // resample from its own call above. + await runtime.whenRefreshSettled(); + assert.equal(defaultPressureSeverity(), "normal"); + } finally { + runtime.dispose(); + } +}); + +test("defaultPressureSeverity still sheds while genuinely critical, across repeated calls", async () => { + let clockMs = 0; + const staleAfterMs = 1_000; + const runtime = reloadResourcePressureRuntime({ + heapThresholdMb: null, + immediateHeapUsedMb: () => 0, + nowMs: () => clockMs, + sample: async () => signals(clockMs, 950), + staleAfterMs, + }); + + try { + await tick(runtime); + clockMs += staleAfterMs + 1; + await tick(runtime); + + assert.equal(defaultPressureSeverity(), "critical"); + clockMs += staleAfterMs + 1; + assert.equal(defaultPressureSeverity(), "critical"); + await runtime.whenRefreshSettled(); + assert.equal(defaultPressureSeverity(), "critical"); + } finally { + runtime.dispose(); + } +});