From c5a08d22e05a7dd2b929f3cca76ee3fb42f122db Mon Sep 17 00:00:00 2001 From: Abhay Singh Date: Wed, 12 Aug 2026 10:35:56 +0530 Subject: [PATCH] fix(proxy): time-cap the compression timeout-debt quarantine (#2360) (#2412) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Fixes #2360. The proxy runs compression on a bounded thread-pool executor with a per-request deadline. Because Python cannot preempt a worker after its `asyncio.wait_for` times out, the code quarantines new compression while a timed-out worker is still running (`_compression_timed_out_in_flight > 0`), to avoid piling more work onto a saturating executor. The gap: that counter only decrements when the worker finally exits. A worker that **never returns** — a hung or pathological compression of a large frame — keeps the counter above zero forever, so the quarantine stays open permanently and every subsequent compression raises `CompressionQuarantinedError`. On Codex WS this is exactly what #2360 reports: one 5s timeout, then Token Savings pinned at ~0% with no recovery, even though the machine is fine. The "parity with direct upstream" nature of the accounting was correct; the only missing piece is an upper bound on how long a single stuck worker may hold the quarantine. ## Fix Add a time cap on the quarantine: - A deadline (`_compression_quarantine_deadline`) is (re)armed on every fresh timeout, to `now + HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS` (default **60s**). - The gate quarantines only while `timed_out_in_flight > 0` **and** `now < deadline`. Once the deadline lapses with no new timeouts, the worker is presumed leaked/abandoned and compression resumes. The release is counted once (a `"released"` quarantine metric + a warning), and the deadline is cleared so it is not re-counted on every later request. - The bounded executor still caps thread growth, and any new timeout re-arms the quarantine, so ongoing genuine slowness keeps quarantining while a single hung worker cannot pin it forever. This preserves the original protection (a burst of slow compressions still quarantines) while guaranteeing recovery. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `headroom/proxy/server.py`: add `_compression_quarantine_deadline` / `_compression_quarantine_max_seconds` (from `HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS`, default 60s) and `_compression_quarantine_releases`; arm the deadline when timeout debt is recorded; release the quarantine (once) in the gate when the deadline lapses. - `tests/test_platform_stabilization_functional.py`: add a test that a standing timed-out worker quarantines within the cap and releases (running compression again, counted once) past it. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [ ] Manual testing performed ### Test Output ```text $ uvx ruff@0.15.17 check headroom/proxy/server.py tests/test_platform_stabilization_functional.py All checks passed! $ uvx mypy@1.20.2 --ignore-missing-imports headroom/proxy/server.py Success: no issues found in 1 source file ``` ## Real Behavior Proof - Environment: Windows 11, Python 3.12, `uvx ruff@0.15.17` / `uvx mypy@1.20.2`. A full `pytest` here imports the ML stack and OOMs this box, so I modeled the gate/deadline state machine with a dependency-free script and left the added `create_app` test to CI. - Exact command / steps: simulated a standing timed-out worker, then exercised the gate at times within the cap, past the cap, and after a fresh timeout, plus a normal worker exit. - Observed result: within the cap the gate quarantines (raises); past the cap it releases exactly once and then lets compression run; a new timeout re-arms the quarantine; a normal worker exit clears the debt. Matching the added handler test (`_run_compression_in_executor` raises `CompressionQuarantinedError` within the cap and returns the callable's result past it, with `_compression_quarantine_releases == 1`). - Not tested: a live Codex WS session hanging a real worker; the added test drives `_run_compression_in_executor` directly with the quarantine state set. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [ ] New and existing unit tests pass locally with my changes - [x] I have updated the CHANGELOG.md if applicable ## Additional Notes The default cap (60s) is deliberately well above a normal slow-but-completing compression so the original saturation protection is unchanged in practice; it only ever fires for a worker that has run far past its deadline. Tunable via `HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS`. The `"released"` quarantine metric and a one-time warning make the recovery observable. The "unit tests pass locally" box is unchecked because the added `create_app` test imports the ML stack (OOM on this box); it runs under the normal CI job, and the state machine is verified by the standalone proof above. --- headroom/proxy/server.py | 47 ++++++++++++++++++- .../test_platform_stabilization_functional.py | 32 +++++++++++++ 2 files changed, 77 insertions(+), 2 deletions(-) diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index 10e5260c7..e47332eb9 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -1156,6 +1156,19 @@ class HeadroomProxy( self._compression_timed_out_in_flight_max: int = 0 self._compression_quarantine_activations: int = 0 self._compression_quarantine_skips: int = 0 + # Time cap on the timeout-debt quarantine. Python cannot preempt a + # worker, so one that never returns (a hung/pathological compression) + # would keep ``_compression_timed_out_in_flight > 0`` forever and pin the + # quarantine open — zeroing out ALL further compression (#2360). The + # deadline is (re)set on every fresh timeout; once it lapses with no new + # timeouts the leaked worker is presumed abandoned and compression + # resumes. The bounded executor still caps thread growth, and any new + # timeout re-arms the quarantine. + self._compression_quarantine_deadline: float = 0.0 + self._compression_quarantine_max_seconds: float = _get_env_float( + "HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS", 60.0 + ) + self._compression_quarantine_releases: int = 0 self._compression_metrics_lock = threading.Lock() # Backend for Anthropic API (direct, LiteLLM, or any-llm) @@ -1413,12 +1426,36 @@ class HeadroomProxy( ``asyncio.TimeoutError`` subclass) if a prior timed-out worker is still running. """ + now = time.monotonic() with self._compression_metrics_lock: timed_out_in_flight = self._compression_timed_out_in_flight - if timed_out_in_flight > 0: + quarantined = timed_out_in_flight > 0 and now < self._compression_quarantine_deadline + # Debt outlived the cap: presume the worker leaked/hung and stop + # blocking on it. Count the release once per lapse (while debt stands + # and the deadline has passed) so operators can see it happened. + released = ( + timed_out_in_flight > 0 + and not quarantined + and self._compression_quarantine_deadline > 0.0 + ) + if quarantined: self._compression_quarantine_skips += 1 + if released: + self._compression_quarantine_releases += 1 + # Clear the deadline so the release is recorded once, not on + # every subsequent request until the worker (maybe never) exits. + self._compression_quarantine_deadline = 0.0 - if timed_out_in_flight > 0: + if released: + self.metrics.record_compression_quarantine("released") + logger.warning( + "Compression quarantine released after %.0fs cap; %d timed-out " + "worker(s) presumed leaked. Compression resumes.", + self._compression_quarantine_max_seconds, + timed_out_in_flight, + ) + + if quarantined: self.metrics.record_compression_quarantine("skipped") raise CompressionQuarantinedError( f"compression quarantined: {timed_out_in_flight} timed-out worker(s) still running" @@ -1457,6 +1494,12 @@ class HeadroomProxy( self._compression_timed_out_in_flight_max, self._compression_timed_out_in_flight, ) + # (Re)arm the quarantine time cap on every fresh timeout, so ongoing + # slowness keeps quarantining while a single leaked worker cannot + # hold it past the cap (#2360). + self._compression_quarantine_deadline = ( + time.monotonic() + self._compression_quarantine_max_seconds + ) state["timeout_debt_recorded"] = True if was_clear: self._compression_quarantine_activations += 1 diff --git a/tests/test_platform_stabilization_functional.py b/tests/test_platform_stabilization_functional.py index 7be75b7c2..946367d05 100644 --- a/tests/test_platform_stabilization_functional.py +++ b/tests/test_platform_stabilization_functional.py @@ -168,3 +168,35 @@ def test_v1_compress_real_json_tool_payload_reduces_tokens(monkeypatch) -> None: assert body["tokens_saved"] > 0 assert body["compression_ratio"] < 1.0 assert body["transforms_applied"], body + + +def test_compression_quarantine_releases_after_time_cap(monkeypatch) -> None: + """A leaked/hung timed-out worker must not pin the quarantine open forever: + once the time cap lapses, compression resumes and the release is counted + once (#2360).""" + import asyncio + + from headroom.proxy.server import CompressionQuarantinedError + + monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1") + app = create_app(_proxy_config()) + proxy = app.state.proxy + + # Simulate a timed-out worker that is still running (debt standing). + proxy._compression_timed_out_in_flight = 1 + + # Within the cap: new compression is quarantined. + proxy._compression_quarantine_deadline = time.monotonic() + 1000.0 + with pytest.raises(CompressionQuarantinedError): + asyncio.run(proxy._run_compression_in_executor(lambda: "unused", timeout=5.0)) + assert proxy._compression_quarantine_releases == 0 + + # Past the cap: the worker is presumed leaked and compression runs again; + # the release is recorded once. + proxy._compression_quarantine_deadline = time.monotonic() - 1.0 + assert asyncio.run(proxy._run_compression_in_executor(lambda: "ran", timeout=5.0)) == "ran" + assert proxy._compression_quarantine_releases == 1 + + # A subsequent request is not re-counted and still runs. + assert asyncio.run(proxy._run_compression_in_executor(lambda: "ok", timeout=5.0)) == "ok" + assert proxy._compression_quarantine_releases == 1