fix(proxy): time-cap the compression timeout-debt quarantine (#2360) (#2412)

## 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.
This commit is contained in:
Abhay Singh 2026-08-12 10:35:56 +05:30 committed by GitHub
parent 74275b7c3e
commit c5a08d22e0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 77 additions and 2 deletions

View file

@ -1156,6 +1156,19 @@ class HeadroomProxy(
self._compression_timed_out_in_flight_max: int = 0 self._compression_timed_out_in_flight_max: int = 0
self._compression_quarantine_activations: int = 0 self._compression_quarantine_activations: int = 0
self._compression_quarantine_skips: 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() self._compression_metrics_lock = threading.Lock()
# Backend for Anthropic API (direct, LiteLLM, or any-llm) # 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 ``asyncio.TimeoutError`` subclass) if a prior timed-out worker is
still running. still running.
""" """
now = time.monotonic()
with self._compression_metrics_lock: with self._compression_metrics_lock:
timed_out_in_flight = self._compression_timed_out_in_flight 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 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") self.metrics.record_compression_quarantine("skipped")
raise CompressionQuarantinedError( raise CompressionQuarantinedError(
f"compression quarantined: {timed_out_in_flight} timed-out worker(s) still running" 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_max,
self._compression_timed_out_in_flight, 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 state["timeout_debt_recorded"] = True
if was_clear: if was_clear:
self._compression_quarantine_activations += 1 self._compression_quarantine_activations += 1

View file

@ -168,3 +168,35 @@ def test_v1_compress_real_json_tool_payload_reduces_tokens(monkeypatch) -> None:
assert body["tokens_saved"] > 0 assert body["tokens_saved"] > 0
assert body["compression_ratio"] < 1.0 assert body["compression_ratio"] < 1.0
assert body["transforms_applied"], body 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