diff --git a/changelog.d/fixes/10111-adaptive-admission-latency-collapse.md b/changelog.d/fixes/10111-adaptive-admission-latency-collapse.md new file mode 100644 index 0000000000..1b806d53e6 --- /dev/null +++ b/changelog.d/fixes/10111-adaptive-admission-latency-collapse.md @@ -0,0 +1 @@ +- **fix(admission):** stop the adaptive latency-gradient collapse from permanently locking out ordinary requests — individually valid requests now make solo progress when the system is idle and normal pressure, and the collapsed limit actively recovers on sustained idle windows instead of being stuck; the critical-pressure fuse still wins over solo progress (#10111) \ No newline at end of file diff --git a/open-sse/services/admission/adaptation.ts b/open-sse/services/admission/adaptation.ts index c266c4b8f6..992c6f4b06 100644 --- a/open-sse/services/admission/adaptation.ts +++ b/open-sse/services/admission/adaptation.ts @@ -17,6 +17,13 @@ export interface AdaptationParams { export interface AdaptationState { currentLimit: number; + /** + * Idle-recovery target: the healthy starting aggregate budget (initialLimit). + * Used to climb the limit back up when a latency-gradient decrease has collapsed it + * below serviceable requests but the system is otherwise idle (#10111). Never grows + * beyond the configured maxLimit. + */ + recoveryCeiling: number; shortLatencyEwma: number; longLatencyEwma: number; pressure: AdmissionPressure; @@ -47,6 +54,7 @@ export function createAdaptationState( ): AdaptationState { return { currentLimit: clampLimit(initialLimit, minLimit, maxLimit), + recoveryCeiling: clampLimit(initialLimit, minLimit, maxLimit), shortLatencyEwma: 0, longLatencyEwma: 0, pressure: "normal", @@ -138,6 +146,14 @@ export function closeAdaptationWindow( // A genuinely low-utilization window recovers the latency baseline so stale gradients expire. if (state.utilization <= params.lowUtilizationThreshold) { state.shortLatencyEwma = state.longLatencyEwma; + // #10111 idle recovery (extracted helper): a latency-gradient decrease must not + // permanently lock the aggregate budget below serviceable requests. On a window with no + // completed work and low utilization (system idle), actively raise the limit back toward + // the recovery ceiling so ordinary requests can re-enter. The high-utilization/completed + // work increase branch above handles growth under load; this covers the no-progress + // starvation case. A window that completed a request (windowCompleted > 0) is the one + // whose latency samples triggered a decrease, so the two branches never fight. + next = applyIdleRecovery(state, params, next); } state.currentLimit = clampLimit(next, params.minLimit, params.maxLimit); @@ -150,6 +166,25 @@ export function closeAdaptationWindow( state.pressure = "normal"; } +/** + * #10111 idle-recovery helper. When a latency-gradient decrease has collapsed the aggregate + * limit below serviceable requests and the system is idle (no completed work, low + * utilization, normal non-critical pressure), raise the limit back toward the recovery + * ceiling by one bounded step so ordinary requests can re-enter. + */ +function applyIdleRecovery(state: AdaptationState, params: AdaptationParams, next: number): number { + if ( + state.pressure !== "critical" && + !state.freezeGrowth && + state.windowCompleted === 0 && + state.currentLimit < state.recoveryCeiling + ) { + const step = Math.min(params.increaseStep, params.maxIncreasePerWindow); + return Math.min(state.recoveryCeiling, next + step); + } + return next; +} + export function sampleActiveIntegral( state: AdaptationState, activeCost: number, diff --git a/open-sse/services/admission/controller.ts b/open-sse/services/admission/controller.ts index 8a6db9be74..28808f1252 100644 --- a/open-sse/services/admission/controller.ts +++ b/open-sse/services/admission/controller.ts @@ -1,4 +1,5 @@ import { + clampLimit, closeAdaptationWindow, createAdaptationState, noteLatency, @@ -151,6 +152,11 @@ export class AdaptiveAdmissionController { next.maxLimit, Math.max(next.minLimit, this.adaptation.currentLimit) ); + // #10111: the idle-recovery ceiling must track a new initialLimit (and the + // possibly-also-new min/maxLimit) instead of staying pinned to the value computed + // at construction time — otherwise a raised initialLimit can never recover past the + // stale ceiling, and a lowered one leaves the ceiling above the new maxLimit. + this.adaptation.recoveryCeiling = clampLimit(next.initialLimit, next.minLimit, next.maxLimit); this.adaptation.windowStartMs = this.clock.now(); this.adaptation.windowActiveCostIntegral = 0; this.adaptation.windowCompleted = 0; @@ -283,6 +289,17 @@ export class AdaptiveAdmissionController { // enforce if (cost > limit) { + // #10111 solo-progress: the adaptive aggregate limit can collapse below an + // individually-valid request (a slow-provider turn shrinks currentLimit via the + // latency gradient, and no increase can fire because every path to "completed" + // requires an admission). A request within the healthy aggregate ceiling must never + // be terminally rejected as oversized while the system is otherwise idle — admit a + // single bounded solo request so the pipeline keeps making progress and the limit can + // recover. The hard per-request ceiling (maxLimit), the critical/high pressure fuse, + // and a busy system (active/queued work present) all take precedence over solo. + if (this.shouldAdmitSolo(cost)) { + return this.admit(cost); + } return this.reject("ADMISSION_OVERSIZED", "request cost exceeds max budget"); } @@ -331,6 +348,24 @@ export class AdaptiveAdmissionController { this.clearLaneEviction(); } + /** + * #10111: whether a request that currently exceeds the temporary aggregate limit may run + * solo. True only when the request fits the healthy aggregate ceiling (maxLimit), the + * system is otherwise idle (no active/queued/lane work) and pressure is normal — so an + * individually-valid request is not terminally rejected as oversized just because a + * latency-gradient decrease collapsed the temporary limit. Under genuine load, critical + * pressure, or an over-ceiling request the caller falls through to the terminal reject. + */ + private shouldAdmitSolo(cost: number): boolean { + return ( + cost <= this.config.maxLimit && + this.active.size === 0 && + this.queue.size === 0 && + this.laneTotalQueuedCount() === 0 && + this.adaptation.pressure === "normal" + ); + } + private resolveCost(request: AdmissionRequest): number { if (request.cost !== undefined) { return normalizeRequestCost(request.cost, this.config.maxRequestCost); diff --git a/tests/unit/adaptive-admission-controller.test.ts b/tests/unit/adaptive-admission-controller.test.ts index bd0785c0a3..8dd1d61f87 100644 --- a/tests/unit/adaptive-admission-controller.test.ts +++ b/tests/unit/adaptive-admission-controller.test.ts @@ -862,17 +862,25 @@ describe("adaptive algorithm", () => { // Immediate fast decrease: 80 * 0.5 = 40. assert.equal(c.snapshot().currentLimit, 40); - // Closing the same window must not multiply again (would become 20). + // Closing the same window must not multiply again (would become 20). #10111 idle + // recovery may climb the collapsed limit upward on the subsequent idle window, so it + // can exceed 40 — the invariant is that closing the window never RE-decreases toward + // the multiplied 20, and recovery stays below the 80 ceiling. clock.advance(100); c.tick(); - assert.equal(c.snapshot().currentLimit, 40); + assert.ok(c.snapshot().currentLimit >= 40); + assert.ok(c.snapshot().currentLimit < 80); - // A fresh critical observation in a later window still decreases once. + // A fresh critical observation in a later window still decreases once — the immediate + // path applies an exact halving regardless of how far idle recovery had climbed first. + const beforeSecond = c.snapshot().currentLimit; c.observePressure("critical"); - assert.equal(c.snapshot().currentLimit, 20); + assert.equal(c.snapshot().currentLimit, Math.floor(beforeSecond / 2)); + const secondFloor = c.snapshot().currentLimit; clock.advance(100); c.tick(); - assert.equal(c.snapshot().currentLimit, 20); + assert.ok(c.snapshot().currentLimit >= secondFloor); + assert.ok(c.snapshot().currentLimit < 80); }); it("decreases on high pressure or sustained latency gradient", async () => { @@ -971,7 +979,13 @@ describe("adaptive algorithm", () => { assert.ok(afterObservedWindow < 80); clock.advance(500); - assert.equal(c.snapshot().currentLimit, afterObservedWindow); + // Stale latency/pressure evidence is still consumed only in its observed window and + // never re-applied as a further decrease (the limit does not drop below + // afterObservedWindow). #10111 idle-recovery instead CLIMBS the collapsed limit back + // toward the recovery ceiling on sustained idle windows, so it recovers upward while + // staying below the initial 80 ceiling. + assert.ok(c.snapshot().currentLimit >= afterObservedWindow); + assert.ok(c.snapshot().currentLimit < 80); }); }); diff --git a/tests/unit/adaptive-admission-latency-collapse.test.ts b/tests/unit/adaptive-admission-latency-collapse.test.ts new file mode 100644 index 0000000000..7b4c7ecc4c --- /dev/null +++ b/tests/unit/adaptive-admission-latency-collapse.test.ts @@ -0,0 +1,288 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { + AdaptiveAdmissionController, + type AdaptiveAdmissionConfig, + type AdmissionRequest, +} from "../../open-sse/services/admission/index.ts"; + +/** + * Regression guard for #10111 — adaptive admission latency collapse. + * + * A slow-provider turn shrinks the adaptive aggregate limit below an ordinary request's + * cost; in enforce mode that request was rejected ADMISSION_OVERSIZED forever, and idle + * recovery could never fire because the only limit-increase path requires a completed + * admission (which can never happen once nothing can be admitted) — a self-lock. + * + * Fix: (1) solo-progress — an individually-valid request within the healthy aggregate + * ceiling admitted while the system is idle & normal pressure, and (2) idle recovery — + * sustained idle windows actively raise the collapsed limit back toward the recovery + * ceiling. Reproduced deterministically with the shipping defaults via a fake clock. + * + * Uses the exact repro harness from the triage plan (same FakeClock, same defaults). + */ +class FakeClock { + nowMs = 0; + private nextId = 1; + private timers = new Map void }>(); + now = () => this.nowMs; + setTimer = (fn: () => void, delayMs: number): number => { + const id = this.nextId++; + this.timers.set(id, { due: this.nowMs + Math.max(0, delayMs), fn }); + return id; + }; + clearTimer = (id: number): void => { this.timers.delete(id); }; + advance(ms: number): void { + const target = this.nowMs + ms; + while (true) { + let nextId: number | undefined; + let nextDue = Number.POSITIVE_INFINITY; + for (const [id, t] of this.timers) { + if (t.due <= target && t.due < nextDue) { nextDue = t.due; nextId = id; } + } + if (nextId === undefined) { this.nowMs = target; return; } + const timer = this.timers.get(nextId)!; + this.timers.delete(nextId); + this.nowMs = timer.due; + timer.fn(); + } + } +} + +function shippingDefaults(overrides: Partial = {}): AdaptiveAdmissionConfig { + return { + mode: "enforce", minLimit: 8, initialLimit: 64, maxLimit: 1000, + maxQueueCount: 128, maxQueueCost: 2000, defaultMaxWaitMs: 5_000, windowMs: 1_000, + decreaseFactor: 0.8, criticalDecreaseFactor: 0.5, increaseStep: 1, maxIncreasePerWindow: 1, + shortLatencyAlpha: 0.5, longLatencyAlpha: 0.1, + highUtilizationThreshold: 0.7, lowUtilizationThreshold: 0.3, latencyGradientThreshold: 0.25, + ...overrides, + }; +} +function req(cost: number): AdmissionRequest { return { tenantKey: "t-default", cost }; } + +const REQUEST_COST = 63; + +function newController(clock: FakeClock): AdaptiveAdmissionController { + return new AdaptiveAdmissionController(shippingDefaults(), { + now: clock.now, setTimer: clock.setTimer, clearTimer: clock.clearTimer, + }); +} + +class Orchestrator { + controller: AdaptiveAdmissionController; + clock: FakeClock; + constructor() { + this.clock = new FakeClock(); + this.controller = newController(this.clock); + } + shutdown(): void { this.controller.shutdown(); } + + /** Admit+release one cost-63 request with the given end-to-end latency sample. */ + async ordinaryTurn(latencyMs: number): Promise { + const r = await this.controller.acquire(req(REQUEST_COST)); + assert.equal(r.status, "admitted"); + if (r.status === "admitted") { + this.clock.advance(1_000); + r.lease.release("success", { latencyMs }); + this.clock.advance(1_000); + } + } +} + +describe("#10111 — adaptive admission latency collapse", () => { + it("a slow-provider gradient collapse does not terminally reject an idle, individually-valid request (solo-progress)", async () => { + const env = new Orchestrator(); + try { + await env.ordinaryTurn(1_000); + await env.ordinaryTurn(10_000); + + const collapsed = env.controller.snapshot().currentLimit; + assert.ok(collapsed < REQUEST_COST, `expected limit collapsed below ${REQUEST_COST}, got ${collapsed}`); + assert.equal(env.controller.snapshot().activeCount, 0); + assert.equal(env.controller.snapshot().queuedCount, 0); + + // System idle & normal pressure: must make progress, not be terminally rejected. + const r3 = await env.controller.acquire(req(REQUEST_COST)); + assert.equal(r3.status, "admitted"); + if (r3.status === "admitted") r3.lease.release("success", { latencyMs: 100 }); + } finally { + env.shutdown(); + } + }); + + it("sustained idle windows actively recover the collapsed limit so normal requests re-enter", async () => { + const env = new Orchestrator(); + try { + await env.ordinaryTurn(1_000); + await env.ordinaryTurn(10_000); + + assert.ok(env.controller.snapshot().currentLimit < REQUEST_COST); + + // No work in flight; run 20 idle windows. + for (let i = 0; i < 20; i++) env.clock.advance(1_000); + + const recovered = env.controller.snapshot().currentLimit; + assert.ok( + recovered >= REQUEST_COST, + `expected idle recovery to restore the limit >= ${REQUEST_COST}, got ${recovered}` + ); + assert.equal(env.controller.snapshot().activeCount, 0); + assert.equal(env.controller.snapshot().queuedCount, 0); + + const r4 = await env.controller.acquire(req(REQUEST_COST)); + assert.equal(r4.status, "admitted"); + if (r4.status === "admitted") r4.lease.release("success", { latencyMs: 100 }); + } finally { + env.shutdown(); + } + }); + + it("critical pressure fuse still wins over solo-progress", async () => { + const env = new Orchestrator(); + try { + await env.ordinaryTurn(1_000); + await env.ordinaryTurn(10_000); + + assert.ok(env.controller.snapshot().currentLimit < REQUEST_COST); + env.controller.observePressure("critical"); + + const r = await env.controller.acquire(req(REQUEST_COST)); + assert.equal(r.status, "rejected"); + if (r.status === "rejected") assert.equal(r.code, "ADMISSION_OVERSIZED"); + } finally { + env.shutdown(); + } + }); + + it("solo-progress does not bypass the healthy aggregate ceiling (maxLimit) or run under load", async () => { + const env = new Orchestrator(); + try { + // A request beyond the healthy aggregate ceiling is still rejected even when idle. + // Distinct maxLimit (20) vs maxRequestCost (100): raw cost 50 is within the hard + // per-request ceiling but exceeds the aggregate ceiling → solo-progress must not + // admit it. + const ceilingGuard = new AdaptiveAdmissionController( + shippingDefaults({ minLimit: 8, initialLimit: 20, maxLimit: 20, cost: { maxRequestCost: 100 } }), + { now: env.clock.now, setTimer: env.clock.setTimer, clearTimer: env.clock.clearTimer } + ); + try { + const heavy = await ceilingGuard.acquire(req(50)); + assert.equal(heavy.status, "rejected"); + if (heavy.status === "rejected") assert.equal(heavy.code, "ADMISSION_OVERSIZED"); + } finally { + ceilingGuard.shutdown(); + } + + // A busy controller (active + queued work present): solo-progress must NOT admit a + // request over the limit — genuine load still sheds oversized-for-limit work. + const busy = new AdaptiveAdmissionController( + shippingDefaults({ minLimit: 8, initialLimit: 20, maxLimit: 20 }), + { now: env.clock.now, setTimer: env.clock.setTimer, clearTimer: env.clock.clearTimer } + ); + try { + const first = await busy.acquire(req(15)); + assert.equal(first.status, "admitted"); + const second = await busy.acquire(req(15)); + assert.equal(second.status, "queued"); + // Active + queued present → not idle → solo must not bypass; oversized rejected. + const over = await busy.acquire(req(25)); + assert.equal(over.status, "rejected"); + if (over.status === "rejected") assert.equal(over.code, "ADMISSION_OVERSIZED"); + if (first.status === "admitted") first.lease.release("success", { latencyMs: 1_000 }); + if (second.status === "queued") { (await second.promise).lease.release("success"); } + } finally { + busy.shutdown(); + } + } finally { + env.shutdown(); + } + }); +}); + +describe("#10111 — updateConfig refreshes the idle-recovery ceiling", () => { + it("a larger initialLimit raises the recovery ceiling, clamped to the (possibly new) maxLimit", () => { + const clock = new FakeClock(); + const controller = new AdaptiveAdmissionController( + shippingDefaults({ minLimit: 5, initialLimit: 20, maxLimit: 50, increaseStep: 1, maxIncreasePerWindow: 1 }), + { now: clock.now, setTimer: clock.setTimer, clearTimer: clock.clearTimer } + ); + try { + // Collapse the limit well below the original ceiling (20) via one critical hit. + controller.observePressure("critical"); // 20 -> floor(20 * 0.5) = 10 + assert.equal(controller.snapshot().currentLimit, 10); + // Close the window the critical hit landed in — idle recovery is suppressed for a + // window where pressure is still critical, so the limit stays put. + clock.advance(1_000); + assert.equal(controller.snapshot().currentLimit, 10); + + // Raise initialLimit far past the (unchanged) maxLimit, and widen the recovery step + // so a single idle window jumps straight to the new ceiling. + controller.updateConfig( + shippingDefaults({ + minLimit: 5, + initialLimit: 1_000, + maxLimit: 50, + increaseStep: 1_000, + maxIncreasePerWindow: 1_000, + }) + ); + + // Idle window: no active/queued work, pressure normal — idle recovery climbs + // straight to the recovery ceiling. + clock.advance(1_000); + assert.equal( + controller.snapshot().currentLimit, + 50, + "recoveryCeiling must track the raised initialLimit, clamped to maxLimit (50)" + ); + } finally { + controller.shutdown(); + } + }); + + it("a smaller initialLimit lowers the recovery ceiling, clamped to the (possibly new) minLimit", () => { + const clock = new FakeClock(); + const controller = new AdaptiveAdmissionController( + shippingDefaults({ minLimit: 5, initialLimit: 100, maxLimit: 200, increaseStep: 1, maxIncreasePerWindow: 1 }), + { now: clock.now, setTimer: clock.setTimer, clearTimer: clock.clearTimer } + ); + try { + // Three separate critical-pressure windows collapse the limit well below the + // original ceiling (100): 100 -> 50 -> 25 -> 12. Each hit lands in its own window + // (advance closes it) so criticalDecreaseConsumed resets and the next hit re-fires. + controller.observePressure("critical"); // 100 -> 50 + clock.advance(1_000); + controller.observePressure("critical"); // 50 -> 25 + clock.advance(1_000); + controller.observePressure("critical"); // 25 -> 12 + clock.advance(1_000); + assert.equal(controller.snapshot().currentLimit, 12); + + // Lower initialLimit below the new minLimit, and widen the recovery step so a stale + // (unrefreshed) ceiling would be unmistakable: it would let idle recovery jump the + // limit straight back up to the old ceiling (100). + controller.updateConfig( + shippingDefaults({ + minLimit: 5, + initialLimit: 1, + maxLimit: 200, + increaseStep: 1_000, + maxIncreasePerWindow: 1_000, + }) + ); + + // Idle window: with the ceiling correctly refreshed to clampLimit(1, 5, 200) = 5, + // currentLimit (12) is already above the ceiling, so idle recovery must not grow it + // at all — in particular it must not climb back to the stale 100 ceiling. + clock.advance(1_000); + assert.equal( + controller.snapshot().currentLimit, + 12, + "recoveryCeiling must track the lowered initialLimit (clamped to minLimit), not the stale higher ceiling" + ); + } finally { + controller.shutdown(); + } + }); +}); \ No newline at end of file