fix(admission): resolve adaptive latency-collapse self-lock with solo-progress and idle recovery (#10111)

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)

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,

open-sse/services/admission/controller.ts

@@ -283,6 +283,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 +342,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);

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);
   });
 });
 

tests/unit/adaptive-admission-latency-collapse.test.ts

@@ -0,0 +1,201 @@
+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<number, { due: number; fn: () => 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> = {}): 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<void> {
+    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();
+    }
+  });
+});