From f6bafe4b88bf0cb385998bce47350e6ecd56fde4 Mon Sep 17 00:00:00 2001 From: Pandu dwi Putra Date: Sat, 19 Sep 2026 10:04:24 +0700 Subject: [PATCH] fix(resilience): resource_pressure admission gate actively re-samples instead of reading a stale cache (#13823) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `admitChatRequest`'s structural admission gate (chatBodyAdmission.ts) 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). `defaultPressureSeverity()` read the resourcePressure singleton's cached `state` directly (getResourcePressureObservation) instead of driving `check()` — so once `state.severity` flipped to "critical" (from any pressure trigger: PSI, v8 heap ratio, a worker-leak spike, etc.), every subsequent request was shed at this cheap cached-read gate before it could ever reach the one function capable of drawing a fresh sample and observing recovery. The gate and the only means of clearing it were mutually exclusive once tripped: the gate's own rejection starved the sampler that would clear the gate. Only a full process restart cleared it. `defaultPressureSeverity()` now calls `checkResourcePressureGuard()` first. That call is cheap on the hot path — a synchronous `process.memoryUsage()` read plus a timestamp comparison; the actual signal sampling (`/proc/pressure/memory`, cgroup reads) stays asynchronous via `scheduleRefresh()` and throttled by `staleAfterMs`, so this adds no 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 raw cached label can still briefly read "critical" until the async refresh settles (or if the last real sample simply went stale), so that case is downgraded to "high" rather than re-introducing the same problem for the queue-wait-sizing branch that also reads this value. Fixes #13821 Test: new tests/unit/resource-pressure-gate-recovery.test.ts drives the resourcePressure singleton to critical through the same sustained-sample path production uses (not the synchronous immediate-heap escape hatch), using an injected mock clock so no scheduled refresh from setup can resolve on its own. Confirmed red on the base commit — the recovery assertion fails with `actual: 'critical', expected: 'normal'`, i.e. the singleton never recovers on its own — and green with the fix. --- ...-resource-pressure-gate-never-resamples.md | 1 + src/shared/middleware/chatBodyAdmission.ts | 40 +++++- .../resource-pressure-gate-recovery.test.ts | 132 ++++++++++++++++++ 3 files changed, 171 insertions(+), 2 deletions(-) create mode 100644 changelog.d/fixes/13821-resource-pressure-gate-never-resamples.md create mode 100644 tests/unit/resource-pressure-gate-recovery.test.ts 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(); + } +});