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