diff --git a/archivebox/opencode/views.py b/archivebox/opencode/views.py index c689abee..14876093 100644 --- a/archivebox/opencode/views.py +++ b/archivebox/opencode/views.py @@ -34,9 +34,11 @@ from django.views.decorators.csrf import csrf_exempt _PROCESS: subprocess.Popen | None = None +_PROCESS_READY: subprocess.Popen | None = None _PROCESS_LOCK = threading.Lock() _SESSION_LOCK = threading.Lock() _LOGGER = logging.getLogger(__name__) +_PROCESS_HEALTH_GRACE = 5 _PROXY_PREFIX = "/admin/agent/opencode" _PROXY_PREFIX_REGEX = _PROXY_PREFIX.replace("/", r"\/") _PROXY_PREFIX_NO_SLASH_REGEX = _PROXY_PREFIX.lstrip("/").replace("/", r"\/") @@ -99,11 +101,12 @@ def _signal_owned_process(process: subprocess.Popen, sig: signal.Signals) -> Non def _stop_owned_process(process: subprocess.Popen | None = None) -> None: - global _PROCESS + global _PROCESS, _PROCESS_READY owned_process = process or _PROCESS if owned_process is None: return if owned_process.poll() is None: + _signal_owned_process(owned_process, signal.SIGCONT) _signal_owned_process(owned_process, signal.SIGTERM) try: owned_process.wait(timeout=5) @@ -112,6 +115,8 @@ def _stop_owned_process(process: subprocess.Popen | None = None) -> None: owned_process.wait() if _PROCESS is owned_process: _PROCESS = None + if _PROCESS_READY is owned_process: + _PROCESS_READY = None atexit.register(_stop_owned_process) @@ -368,11 +373,21 @@ def _ensure_default_session(settings: dict) -> str: return session_id -def _health(settings: dict) -> bool: +def _owned_process_running() -> bool: + process = _PROCESS + return process is not None and process.poll() is None + + +def _owned_process_ready() -> bool: + process = _PROCESS_READY + return process is not None and process is _PROCESS and process.poll() is None + + +def _health(settings: dict, timeout: float = 2) -> bool: try: response = requests.get( f"{settings['origin']}/global/health", - timeout=2, + timeout=timeout, ) return response.status_code == 200 except requests.RequestException: @@ -393,15 +408,28 @@ def _opencode_version(settings: dict) -> str: def _ensure_opencode(settings: dict) -> tuple[bool, str]: - global _PROCESS + global _PROCESS, _PROCESS_READY started_process: subprocess.Popen | None = None workdir = settings["workdir"].resolve() with _PROCESS_LOCK: - if _health(settings): - return True, "" - if _PROCESS is not None and _PROCESS.poll() is None: + if _owned_process_running(): + deadline = time.monotonic() + min(_PROCESS_HEALTH_GRACE, settings["timeout"]) + while _owned_process_running(): + remaining = deadline - time.monotonic() + if remaining <= 0: + break + if _health(settings, timeout=min(2, remaining)): + if time.monotonic() <= deadline: + _PROCESS_READY = _PROCESS + return True, "" + break + remaining = deadline - time.monotonic() + if remaining > 0: + time.sleep(min(0.25, remaining)) _stop_owned_process(_PROCESS) + elif _health(settings): + return True, "" try: binary, git_binary, binary_env = _resolve_binary( @@ -477,6 +505,7 @@ def _ensure_opencode(settings: dict) -> tuple[bool, str]: stdout=subprocess.DEVNULL, start_new_session=True, ) + _PROCESS_READY = None started_process = _PROCESS except FileNotFoundError: return False, f"OpenCode binary not found: {settings['binary']}" @@ -484,10 +513,13 @@ def _ensure_opencode(settings: dict) -> tuple[bool, str]: deadline = time.monotonic() + settings["timeout"] while time.monotonic() < deadline: if _health(settings): + _PROCESS_READY = started_process return True, "" if started_process and started_process.poll() is not None: if _PROCESS is started_process: _PROCESS = None + if _PROCESS_READY is started_process: + _PROCESS_READY = None return False, "OpenCode exited before the web server became ready." time.sleep(0.25) @@ -603,11 +635,12 @@ async def _event_chunks( params: tuple[tuple[str, str], ...], headers: dict[str, str], ): - ok, error = await sync_to_async(_ensure_opencode, thread_sensitive=False)(settings) - if not ok: - _LOGGER.warning("OpenCode event stream unavailable: %s", error) - yield b'event: error\ndata: {"error":"OpenCode upstream unavailable"}\n\n' - return + if not _owned_process_ready(): + ok, error = await sync_to_async(_ensure_opencode, thread_sensitive=False)(settings) + if not ok: + _LOGGER.warning("OpenCode event stream unavailable: %s", error) + yield b'event: error\ndata: {"error":"OpenCode upstream unavailable"}\n\n' + return timeout = httpx.Timeout(settings["timeout"], read=None) url = _proxy_url(settings, path) @@ -729,9 +762,10 @@ def opencode_proxy_view(request: HttpRequest, path: str | None = None): response.headers["X-Accel-Buffering"] = "no" return response - ok, error = _ensure_opencode(settings) - if not ok: - return _proxy_error_response(error) + if path == "global/health" or not _owned_process_ready(): + ok, error = _ensure_opencode(settings) + if not ok: + return _proxy_error_response(error) try: method = request.method or "GET" diff --git a/archivebox/tests/test_opencode_agent.py b/archivebox/tests/test_opencode_agent.py index 4bd89e2f..65c21b27 100644 --- a/archivebox/tests/test_opencode_agent.py +++ b/archivebox/tests/test_opencode_agent.py @@ -1,7 +1,9 @@ import asyncio import os +import signal import socket import subprocess +import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path from types import SimpleNamespace @@ -301,6 +303,75 @@ def test_opencode_restarts_an_unhealthy_owned_process(live_opencode): assert views._health(settings) +def test_opencode_preserves_a_transiently_unhealthy_owned_process(live_opencode): + from archivebox.opencode import views + + process = views._PROCESS + assert process is not None + views._signal_owned_process(process, signal.SIGSTOP) + + def resume_process(): + time.sleep(1.5) + views._signal_owned_process(process, signal.SIGCONT) + + try: + with ThreadPoolExecutor(max_workers=1) as executor: + resumed = executor.submit(resume_process) + started_at = time.monotonic() + ok, error = views._ensure_opencode(live_opencode.settings) + elapsed = time.monotonic() - started_at + resumed.result() + finally: + views._signal_owned_process(process, signal.SIGCONT) + + assert ok, error + assert views._PROCESS is process + assert process.poll() is None + assert elapsed < views._PROCESS_HEALTH_GRACE + + +def test_opencode_proxy_does_not_wait_for_recovery_lock(admin_client, live_opencode): + from archivebox.opencode import views + + workdir = quote(str(live_opencode.config.data_dir.resolve())) + assert views._owned_process_ready() + executor = ThreadPoolExecutor(max_workers=1) + views._PROCESS_LOCK.acquire() + try: + request = executor.submit( + admin_client.get, + f"/admin/agent/opencode/path?directory={workdir}", + HTTP_HOST=ADMIN_TEST_HOST, + HTTP_SEC_FETCH_SITE="same-origin", + ) + response = request.result(timeout=5) + finally: + views._PROCESS_LOCK.release() + executor.shutdown(wait=True) + + assert response.status_code == 200 + assert str(live_opencode.config.data_dir.resolve()).encode() in response.content + + +def test_opencode_proxy_waits_for_owned_process_readiness(admin_client, live_opencode): + from archivebox.opencode import views + + process = views._PROCESS + assert process is not None + views._PROCESS_READY = None + workdir = quote(str(live_opencode.config.data_dir.resolve())) + + response = admin_client.get( + f"/admin/agent/opencode/path?directory={workdir}", + HTTP_HOST=ADMIN_TEST_HOST, + HTTP_SEC_FETCH_SITE="same-origin", + ) + + assert response.status_code == 200 + assert views._PROCESS is process + assert views._PROCESS_READY is process + + def test_opencode_proxy_sse_response_is_unbuffered(admin_client, live_opencode): response = admin_client.get( "/admin/agent/opencode/global/event",