mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
## Description On Windows, the first call into the native `headroom._core.detect_content_type` can park forever in an ort/`Once` initialization (`WaitOnAddress`) at 0% CPU. A wedged native call cannot be cancelled from Python, so it deadlocks the caller. In the proxy it is worse: each affected request permanently consumes a compression-executor worker, eventually saturating the pool (`running == max_workers`, `leaked_threads_total == 0` because the worker never finishes) and stalling every subsequent request for the full `COMPRESSION_TIMEOUT_SECONDS` before passthrough. The Rust backend is already off by default on Windows — `_resolve_detect_backend()` returns `"python"` there — but an explicit `HEADROOM_DETECT_BACKEND=rust`, or any future regression of that default, re-exposes the hang with no escape hatch. This implements the issue's third ask: a timeout/watchdog so a hung native init degrades gracefully instead of deadlocking the agent / MCP server / proxy. Closes #575 ## 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 - Added a Windows-only watchdog around the native detect call in `transforms/content_router.py`. `_rust_detect_watchdogged()` runs `detect_content_type` on a daemon thread and bounds the caller's wait; on timeout it raises `TimeoutError`, which the existing `except BaseException` handler degrades to the pure-Python regex detector. Detection therefore always returns instead of deadlocking (and, in the proxy, instead of permanently consuming a compression-executor worker). - Added `_detect_timeout_secs()` reading `HEADROOM_DETECT_TIMEOUT_SECS` (default 5s; blank / non-numeric / non-positive values fall back to the default). - Gated the watchdog to `sys.platform == "win32"` — the only platform where the hang is observed. Other platforms keep the direct native call with no per-call thread overhead (the trusted hot path is unchanged). - Added regression tests for the watchdog, env parsing, error relay, the Windows degrade-on-hang path, and the Windows happy path. ## 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 $ ruff check headroom/transforms/content_router.py tests/test_transforms_content_router.py All checks passed! $ ruff format --check headroom/transforms/content_router.py tests/test_transforms_content_router.py 2 files already formatted $ mypy headroom --ignore-missing-imports (exit 0) $ pytest tests/test_transforms_content_router.py -q 33 passed ``` ## Real Behavior Proof - Environment: Windows 11, Python 3.13, ruff 0.15.17 / mypy 1.20.2 / pytest 9.1.0, `headroom._core` built locally, branch `fix/575-native-detect-watchdog`. - Exact command / steps: ran the four checks above. `test_detect_content_watchdog_degrades_on_windows_hang` forces `HEADROOM_DETECT_BACKEND=rust`, patches `sys.platform` to `"win32"`, sets `HEADROOM_DETECT_TIMEOUT_SECS=0.1`, and injects a native `detect_content_type` that blocks on an `Event` (simulating the `WaitOnAddress` park, GIL released) — then asserts detection still returns. The companion tests cover env parsing, error relay through the watchdog, and the fast-native Windows path. - Observed result: with a hung native detector, `_detect_content('[{"id": 1}]')` returns `ContentType.JSON_ARRAY` (the pure-Python degrade path) within the 0.1s budget instead of hanging; with a fast native detector on Windows it returns the native result unchanged; non-Windows behavior (direct call) is untouched and the existing rust-delegation test still passes. All 33 tests in the file pass; ruff / format / mypy clean. - Not tested: the live `from headroom._core import detect_content_type; detect_content_type("hello world")` deadlock on an affected Windows 11 24H2 machine was not reproduced end to end (it requires the specific System32 ONNX Runtime build). The fix is instead covered by the deterministic hung-detector injection test, which exercises the exact degrade path the watchdog adds. This PR does not attempt the Rust-side fix for the underlying first-call init deadlock (asks #1) — it is the Python-side watchdog (ask #3); the existing `HEADROOM_DETECT_BACKEND` flag already covers ask #2. ## 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 - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Additional Notes - The watchdog cannot cancel a wedged native call (no portable way to kill a thread blocked in C). It frees the *caller* and leaves the stuck daemon thread to die with the process; this is marked with a `ponytail:` comment naming the upgrade path (the Rust-side non-blocking first-call init). For the saturation scenario this is still a strict improvement: callers no longer block indefinitely, so the executor drains instead of wedging permanently. --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Parideboy
GitHub
parent
96e1dfe395
commit
95abca3abd
2 changed files with 207 additions and 4 deletions
|
|
@ -42,6 +42,7 @@ import math
|
|||
import os
|
||||
import re
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import dataclass, field
|
||||
|
|
@ -66,6 +67,7 @@ logger = logging.getLogger(__name__)
|
|||
|
||||
_detect_backend_warned = False
|
||||
_detect_panic_warned = False
|
||||
_detect_native_unhealthy = False # circuit breaker: native detect hung once (#575)
|
||||
|
||||
|
||||
def _router_debug_dumps(value: Any) -> str:
|
||||
|
|
@ -129,6 +131,60 @@ def _resolve_detect_backend() -> str:
|
|||
return "python" if sys.platform == "win32" else "rust"
|
||||
|
||||
|
||||
_DETECT_TIMEOUT_ENV = "HEADROOM_DETECT_TIMEOUT_SECS"
|
||||
_DEFAULT_DETECT_TIMEOUT_SECS = 5.0
|
||||
|
||||
|
||||
def _detect_timeout_secs() -> float:
|
||||
"""Watchdog budget (seconds) for one native detect call.
|
||||
|
||||
Override with ``HEADROOM_DETECT_TIMEOUT_SECS``; blank, non-numeric, or
|
||||
non-positive values fall back to the default.
|
||||
"""
|
||||
raw = os.environ.get(_DETECT_TIMEOUT_ENV, "").strip()
|
||||
if not raw:
|
||||
return _DEFAULT_DETECT_TIMEOUT_SECS
|
||||
try:
|
||||
secs = float(raw)
|
||||
except ValueError:
|
||||
return _DEFAULT_DETECT_TIMEOUT_SECS
|
||||
return secs if secs > 0 else _DEFAULT_DETECT_TIMEOUT_SECS
|
||||
|
||||
|
||||
def _rust_detect_watchdogged(rust_detect: Any, content: str, timeout: float) -> Any:
|
||||
"""Run the native detector under a watchdog thread, bounding the caller's wait.
|
||||
|
||||
On Windows the first native ``detect_content_type`` can park forever in an
|
||||
ort/``Once`` init (``WaitOnAddress``) at 0% CPU, and a wedged native call
|
||||
cannot be cancelled from Python (#575). The native call releases the GIL
|
||||
while parked, so a watchdog thread runs it and the caller waits at most
|
||||
``timeout`` seconds before raising ``TimeoutError`` — letting
|
||||
``_detect_content`` degrade to the pure-Python detector instead of
|
||||
deadlocking (and, in the proxy, instead of permanently consuming a
|
||||
compression-executor worker — see #575's executor-saturation report).
|
||||
|
||||
# ponytail: can't kill a GIL-released native call; the watchdog frees the
|
||||
# caller and the stuck daemon thread is left to die with the process. The
|
||||
# upgrade path is the Rust-side fix that makes first-call init non-blocking.
|
||||
"""
|
||||
box: dict[str, Any] = {}
|
||||
|
||||
def _run() -> None:
|
||||
try:
|
||||
box["result"] = rust_detect(content)
|
||||
except BaseException as exc: # noqa: BLE001 — relayed to the caller's degrade path
|
||||
box["error"] = exc
|
||||
|
||||
worker = threading.Thread(target=_run, name="headroom-detect-watchdog", daemon=True)
|
||||
worker.start()
|
||||
worker.join(timeout)
|
||||
if worker.is_alive():
|
||||
raise TimeoutError(f"native detect_content_type exceeded {timeout:.1f}s watchdog")
|
||||
if "error" in box:
|
||||
raise box["error"]
|
||||
return box["result"]
|
||||
|
||||
|
||||
def _detect_content(content: str) -> DetectionResult:
|
||||
"""Detect content type via the native chain, with a safe Windows default.
|
||||
|
||||
|
|
@ -144,7 +200,7 @@ def _detect_content(content: str) -> DetectionResult:
|
|||
only consumed `.content_type` from it; the strategy mapping in
|
||||
`_strategy_from_detection` keys off that field alone.
|
||||
"""
|
||||
global _detect_backend_warned
|
||||
global _detect_backend_warned, _detect_panic_warned, _detect_native_unhealthy
|
||||
|
||||
backend = _resolve_detect_backend()
|
||||
if backend == "python":
|
||||
|
|
@ -157,11 +213,23 @@ def _detect_content(content: str) -> DetectionResult:
|
|||
)
|
||||
return _regex_detect_content_type(content)
|
||||
|
||||
if _detect_native_unhealthy:
|
||||
# Circuit breaker (#575): the native detector hung once under the
|
||||
# watchdog; every later call would wait the full budget and strand
|
||||
# another stuck daemon thread, so route straight to pure-Python.
|
||||
return _regex_detect_content_type(content)
|
||||
|
||||
from headroom._core import detect_content_type as _rust_detect
|
||||
|
||||
global _detect_panic_warned
|
||||
try:
|
||||
rust_result = _rust_detect(content)
|
||||
if sys.platform == "win32":
|
||||
# Windows is the only platform where the native detector can deadlock
|
||||
# on first use (#575); bound it with a watchdog so a hang degrades to
|
||||
# the pure-Python detector below. Elsewhere it is the trusted default
|
||||
# hot path — call it directly, with no per-call thread overhead.
|
||||
rust_result = _rust_detect_watchdogged(_rust_detect, content, _detect_timeout_secs())
|
||||
else:
|
||||
rust_result = _rust_detect(content)
|
||||
# Rust's `content_type` is the lowercase string tag (e.g.
|
||||
# "json_array"); translate to the Python `ContentType` enum so
|
||||
# downstream mapping keys match.
|
||||
|
|
@ -178,7 +246,17 @@ def _detect_content(content: str) -> DetectionResult:
|
|||
# as asyncio.CancelledError — keep them propagating.
|
||||
if isinstance(exc, asyncio.CancelledError):
|
||||
raise
|
||||
if not _detect_panic_warned:
|
||||
if isinstance(exc, TimeoutError):
|
||||
# Watchdog tripped: the native detector hung (#575). Disable it
|
||||
# process-wide so later calls don't each wait the full budget and
|
||||
# strand another daemon thread in the wedged native call.
|
||||
_detect_native_unhealthy = True
|
||||
logger.warning(
|
||||
"Native content detector hung (%s); disabling it for this process "
|
||||
"and using pure-Python detection.",
|
||||
exc,
|
||||
)
|
||||
elif not _detect_panic_warned:
|
||||
_detect_panic_warned = True
|
||||
logger.warning(
|
||||
"Native content detector failed (%s); falling back to pure-Python detection.",
|
||||
|
|
|
|||
|
|
@ -21,6 +21,19 @@ from headroom.transforms.content_router import (
|
|||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_detect_module_state(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Keep the module-level detect flags from leaking across tests.
|
||||
|
||||
The circuit breaker (#575) is process-wide, so a test that trips it would
|
||||
otherwise force later tests onto the pure-Python path. ``monkeypatch.setattr``
|
||||
zeroes each flag for the test and auto-restores it afterward.
|
||||
"""
|
||||
monkeypatch.setattr(content_router_module, "_detect_native_unhealthy", False)
|
||||
monkeypatch.setattr(content_router_module, "_detect_backend_warned", False)
|
||||
monkeypatch.setattr(content_router_module, "_detect_panic_warned", False)
|
||||
|
||||
|
||||
def test_compression_cache_handles_hits_skips_evictions_and_clear(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
|
|
@ -864,3 +877,115 @@ def test_detect_content_python_backend_skips_native(
|
|||
|
||||
result = _detect_content('[{"id": 1}, {"id": 2}]')
|
||||
assert result.content_type is ContentType.JSON_ARRAY
|
||||
|
||||
|
||||
def test_detect_timeout_secs_env_parsing(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The watchdog budget reads HEADROOM_DETECT_TIMEOUT_SECS; bad values → default."""
|
||||
get = content_router_module._detect_timeout_secs
|
||||
default = content_router_module._DEFAULT_DETECT_TIMEOUT_SECS
|
||||
|
||||
monkeypatch.delenv("HEADROOM_DETECT_TIMEOUT_SECS", raising=False)
|
||||
assert get() == default
|
||||
|
||||
monkeypatch.setenv("HEADROOM_DETECT_TIMEOUT_SECS", "0.25")
|
||||
assert get() == 0.25
|
||||
|
||||
monkeypatch.setenv("HEADROOM_DETECT_TIMEOUT_SECS", "nope")
|
||||
assert get() == default
|
||||
|
||||
monkeypatch.setenv("HEADROOM_DETECT_TIMEOUT_SECS", "0")
|
||||
assert get() == default
|
||||
|
||||
|
||||
def test_rust_detect_watchdog_passes_through_result() -> None:
|
||||
"""A fast native detector returns its result unchanged through the watchdog."""
|
||||
sentinel = SimpleNamespace(content_type="json_array", confidence=1.0, metadata={})
|
||||
out = content_router_module._rust_detect_watchdogged(lambda _content: sentinel, "payload", 5.0)
|
||||
assert out is sentinel
|
||||
|
||||
|
||||
def test_rust_detect_watchdog_relays_native_error() -> None:
|
||||
"""An exception raised inside the native detector propagates to the caller."""
|
||||
|
||||
def boom(_content: str) -> None:
|
||||
raise ValueError("native boom")
|
||||
|
||||
with pytest.raises(ValueError, match="native boom"):
|
||||
content_router_module._rust_detect_watchdogged(boom, "payload", 5.0)
|
||||
|
||||
|
||||
def test_detect_content_watchdog_degrades_on_windows_hang(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A hung native detect on Windows degrades to pure-Python, never deadlocks (#575)."""
|
||||
import threading as _threading
|
||||
|
||||
import headroom._core as _core
|
||||
|
||||
monkeypatch.setenv("HEADROOM_DETECT_BACKEND", "rust")
|
||||
monkeypatch.setattr(content_router_module.sys, "platform", "win32")
|
||||
monkeypatch.setenv("HEADROOM_DETECT_TIMEOUT_SECS", "0.1")
|
||||
|
||||
release = _threading.Event()
|
||||
|
||||
def _hang(_content: str):
|
||||
release.wait() # simulate the WaitOnAddress park (GIL released while waiting)
|
||||
return SimpleNamespace(content_type="plain_text", confidence=1.0, metadata={})
|
||||
|
||||
monkeypatch.setattr(_core, "detect_content_type", _hang)
|
||||
|
||||
try:
|
||||
# JSON content: the pure-Python regex fallback recognizes it as JSON_ARRAY,
|
||||
# proving we took the degrade path rather than the (hung) native result.
|
||||
result = _detect_content('[{"id": 1}]')
|
||||
assert result.content_type is ContentType.JSON_ARRAY
|
||||
finally:
|
||||
release.set() # let the daemon worker finish so it does not linger
|
||||
|
||||
|
||||
def test_detect_content_watchdog_uses_native_result_on_windows(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""On Windows with rust forced, a fast native result still flows through unchanged."""
|
||||
import headroom._core as _core
|
||||
|
||||
monkeypatch.setenv("HEADROOM_DETECT_BACKEND", "rust")
|
||||
monkeypatch.setattr(content_router_module.sys, "platform", "win32")
|
||||
|
||||
fake = SimpleNamespace(content_type="source_code", confidence=1.0, metadata={})
|
||||
monkeypatch.setattr(_core, "detect_content_type", lambda _content: fake)
|
||||
|
||||
result = _detect_content("def main(): pass")
|
||||
assert result.content_type is ContentType.SOURCE_CODE
|
||||
|
||||
|
||||
def test_detect_content_circuit_breaker_skips_native_after_hang(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""After one watchdog timeout, native detection is disabled process-wide (#575)."""
|
||||
import threading as _threading
|
||||
|
||||
import headroom._core as _core
|
||||
|
||||
monkeypatch.setenv("HEADROOM_DETECT_BACKEND", "rust")
|
||||
monkeypatch.setattr(content_router_module.sys, "platform", "win32")
|
||||
monkeypatch.setenv("HEADROOM_DETECT_TIMEOUT_SECS", "0.1")
|
||||
|
||||
release = _threading.Event()
|
||||
calls = 0
|
||||
|
||||
def _hang(_content: str):
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
release.wait() # park with GIL released, like the real WaitOnAddress hang
|
||||
return SimpleNamespace(content_type="plain_text", confidence=1.0, metadata={})
|
||||
|
||||
monkeypatch.setattr(_core, "detect_content_type", _hang)
|
||||
try:
|
||||
first = _detect_content('[{"id": 1}]')
|
||||
second = _detect_content('[{"id": 2}]')
|
||||
assert first.content_type is ContentType.JSON_ARRAY
|
||||
assert second.content_type is ContentType.JSON_ARRAY
|
||||
assert calls == 1 # breaker tripped: native entered once, 2nd call skipped it
|
||||
finally:
|
||||
release.set() # let the lone daemon worker finish
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue