diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 92fa6dd8..659f193a 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -2304,6 +2304,7 @@ Override destination based on content type metadata. | `EXT_BYPASSER_URL` | URL of the external bypasser service (e.g., FlareSolverr). | string | `http://flaresolverr:8191` | | `EXT_BYPASSER_PATH` | API path for the external bypasser. | string | `/v1` | | `EXT_BYPASSER_TIMEOUT` | Timeout for external bypasser requests in milliseconds. | number | `60000` | +| `BYPASS_BROWSER_IDLE_TIMEOUT` | How long the bypass helper process may sit unused before it is shut down. Higher keeps more searches fast, lower frees memory sooner. | number | `180` |
Detailed descriptions @@ -2359,6 +2360,17 @@ Timeout for external bypasser requests in milliseconds. - **Requires restart:** Yes - **Constraints:** min: 10000, max: 300000 +#### `BYPASS_BROWSER_IDLE_TIMEOUT` + +**Bypasser Idle Timeout (seconds)** + +How long the bypass helper process may sit unused before it is shut down. Higher keeps more searches fast, lower frees memory sooner. + +- **Type:** number +- **Default:** `180` +- **Requires restart:** Yes +- **Constraints:** min: 30, max: 3600 +
### Direct Download: Mirrors diff --git a/shelfmark/bypass/internal_bypasser.py b/shelfmark/bypass/internal_bypasser.py index 647231fc..bec856bc 100644 --- a/shelfmark/bypass/internal_bypasser.py +++ b/shelfmark/bypass/internal_bypasser.py @@ -62,6 +62,19 @@ _BYPASS_SUBPROCESS_TIMEOUT_SECONDS = 420.0 # branches of get() are bounded the same way. _IN_PROCESS_BYPASS_TIMEOUT_SECONDS = _BYPASS_SUBPROCESS_TIMEOUT_SECONDS _BYPASS_CHILD_ENV = "SHELFMARK_INTERNAL_BYPASSER_CHILD" +# The helper bounds each bypass below the parent's deadline, so it is the side that gives +# up first: it still gets to report the timeout and close its browser, and stays available +# for the next request. A parent that hit its deadline first could only kill the helper, +# throwing away a process the next request would have to start again. +_CHILD_BYPASS_TIMEOUT_SECONDS = _BYPASS_SUBPROCESS_TIMEOUT_SECONDS - 30.0 +# The helper publishes its answer by writing the result file the request named, so the +# parent waits by watching for that file rather than by reading a stream it would have to +# demultiplex from the helper's log output. +_HELPER_RESULT_POLL_SECONDS = 0.05 +# Closing the helper's stdin asks it to shut down; this is how long it may take to finish +# what it is doing and exit before its session is killed instead. +_HELPER_SHUTDOWN_GRACE_SECONDS = 15.0 +_HELPER_IDLE_TIMEOUT_DEFAULT = 180.0 _PARENT_WATCHDOG_INTERVAL_SECONDS = 5.0 @@ -805,13 +818,21 @@ def _run_bypass_in_current_process(url: str, retry: int, cancel_flag: Event | No if driver: await _close_cdp_driver(driver) - if os.environ.get(_BYPASS_CHILD_ENV) == "1": - return asyncio.run(_run_bypass()) - # Bound the wait: this path runs in-process (non-Docker installs), holds the module-wide - # LOCKED for its whole duration, and neither page.get() nor page.wait() has a timeout of - # its own. Without a deadline here a single wedged CDP session blocks every subsequent - # bypass in the process forever. - return _CDP_WORKER.run(_run_bypass(), timeout=_IN_PROCESS_BYPASS_TIMEOUT_SECONDS) + # Bound the wait: this holds the module-wide LOCKED for its whole duration, and neither + # page.get() nor page.wait() has a timeout of its own. Without a deadline here a single + # wedged CDP session blocks every subsequent bypass in the process forever. + # + # The helper goes through the worker too, rather than asyncio.run: that owns a loop for + # one call and closes it on the way out, so a helper serving many requests would build + # and tear down a loop per bypass and would carry no deadline of its own. The worker's + # loop lives in a thread, outlives any single bypass, and cancels the coroutine when the + # deadline passes. + timeout = ( + _CHILD_BYPASS_TIMEOUT_SECONDS + if os.environ.get(_BYPASS_CHILD_ENV) == "1" + else _IN_PROCESS_BYPASS_TIMEOUT_SECONDS + ) + return _CDP_WORKER.run(_run_bypass(), timeout=timeout) def _store_child_bypass_state(payload: dict[str, Any]) -> None: @@ -853,6 +874,192 @@ def _terminate_helper_session(proc: subprocess.Popen[str]) -> None: proc.wait(timeout=5) +class _BypassHelper: + """The helper subprocess that runs the bypasses, kept alive across them. + + Spawning it costs about 4.5 seconds of interpreter start and imports before any work + begins, paid on every protected request - and a single search issues several. What it + keeps is the process, not the browser: each bypass still starts and closes its own + Chrome, so nothing accumulates between requests. + + Protocol: one JSON request per line on stdin, answered by writing the result file that + request named. stdout and stderr stay attached to the parent's, so helper logs keep + showing up in `docker logs` as before. + + Only one request is ever in flight - get() serializes every bypass behind LOCKED. The + lock here is for the idle reaper, which runs on a timer thread. + """ + + def __init__(self) -> None: + self._lock = threading.RLock() + self._proc: subprocess.Popen[str] | None = None + self._last_used = 0.0 + self._idle_timer: threading.Timer | None = None + + def _idle_timeout(self) -> float: + return _coerce_non_negative_float( + app_config.get("BYPASS_BROWSER_IDLE_TIMEOUT", _HELPER_IDLE_TIMEOUT_DEFAULT), + _HELPER_IDLE_TIMEOUT_DEFAULT, + ) + + def _spawn(self) -> subprocess.Popen[str]: + env_vars = os.environ.copy() + env_vars[_BYPASS_CHILD_ENV] = "1" + env_vars = _prepare_child_browser_env(env_vars) + return subprocess.Popen( + [sys.executable, "-m", "shelfmark.bypass.internal_bypasser"], + stdin=subprocess.PIPE, + text=True, + env=env_vars, + # Give the helper its own session: Chrome, Xvfb and ffmpeg inherit its process + # group, which is what lets the cleanup sweep tell this helper's browsers apart + # from a concurrent worker's (#1231) and lets us kill the whole tree below. + start_new_session=True, + ) + + def _running(self) -> subprocess.Popen[str] | None: + proc = self._proc + if proc is None: + return None + if proc.poll() is not None or proc.stdin is None or proc.stdin.closed: + return None + return proc + + def _ensure_running(self) -> subprocess.Popen[str]: + proc = self._running() + if proc is not None: + return proc + if self._proc is not None: + logger.info("Bypass helper exited (code %s), starting a new one", self._proc.returncode) + self._discard() + self._proc = self._spawn() + return self._proc + + def _discard(self) -> None: + """Stop the helper and forget it.""" + proc = self._proc + self._proc = None + if proc is None: + return + + # Closing stdin ends the helper's request loop, so an idle helper gets to exit on + # its own. One mid-bypass cannot answer, and is killed below once the grace passes. + with suppress(OSError): + if proc.stdin is not None and not proc.stdin.closed: + proc.stdin.close() + try: + proc.wait(timeout=_HELPER_SHUTDOWN_GRACE_SECONDS) + except subprocess.TimeoutExpired: + logger.warning("Bypass helper did not exit on request, killing its session") + + # Tear the session down either way: a helper killed mid-bypass leaves its Chrome + # and Xvfb running, and those leftovers are what made the next worker's browser + # fail to start. Harmless once it has already exited. + _terminate_helper_session(proc) + + def _cancel_idle_timer(self) -> None: + if self._idle_timer is not None: + self._idle_timer.cancel() + self._idle_timer = None + + def _arm_idle_timer(self) -> None: + self._cancel_idle_timer() + timeout = self._idle_timeout() + if self._proc is None or timeout <= 0: + return + timer = threading.Timer(timeout, self._reap_if_idle) + timer.daemon = True + self._idle_timer = timer + timer.start() + + def _reap_if_idle(self) -> None: + with self._lock: + if self._proc is None: + return + idle_for = time.monotonic() - self._last_used + timeout = self._idle_timeout() + if idle_for < timeout: + # A bypass started while this timer was waiting for the lock. + self._arm_idle_timer() + return + logger.info("Closing idle bypass helper after %.0fs without work", idle_for) + self._discard() + + def run( + self, + payload: dict[str, Any], + timeout: float, + cancel_flag: Event | None, + ) -> dict[str, Any]: + with self._lock: + self._cancel_idle_timer() + try: + return self._exchange(payload, timeout, cancel_flag) + finally: + self._last_used = time.monotonic() + self._arm_idle_timer() + + def _exchange( + self, + payload: dict[str, Any], + timeout: float, + cancel_flag: Event | None, + ) -> dict[str, Any]: + request_line = json.dumps(payload) + "\n" + proc = self._ensure_running() + try: + self._write(proc, request_line) + except OSError as exc: + # A live helper can die between the liveness check and the write, so one retry + # on a fresh process. A fresh one failing here is a real failure. + logger.info("Bypass helper closed its pipe (%s), retrying on a new one", exc) + self._discard() + proc = self._ensure_running() + self._write(proc, request_line) + + return self._await_result(proc, Path(str(payload["result_path"])), timeout, cancel_flag) + + def _write(self, proc: subprocess.Popen[str], request_line: str) -> None: + if proc.stdin is None: + msg = "Bypass helper has no stdin pipe" + raise OSError(msg) + proc.stdin.write(request_line) + proc.stdin.flush() + + def _await_result( + self, + proc: subprocess.Popen[str], + result_path: Path, + timeout: float, + cancel_flag: Event | None, + ) -> dict[str, Any]: + deadline = time.monotonic() + timeout + while not result_path.exists(): + if proc.poll() is not None: + returncode = proc.returncode + self._discard() + msg = f"Internal bypasser helper exited without a result (code {returncode})" + raise RuntimeError(msg) + if cancel_flag is not None and cancel_flag.is_set(): + # The helper is mid-bypass and cannot be told to stop, so it goes. + self._discard() + _check_cancellation(cancel_flag, "Bypass cancelled while waiting for helper") + if time.monotonic() >= deadline: + self._discard() + msg = "Internal bypasser helper process timed out" + raise TimeoutError(msg) + time.sleep(_HELPER_RESULT_POLL_SECONDS) + + try: + return json.loads(result_path.read_text(encoding="utf-8")) + finally: + with suppress(OSError): + result_path.unlink() + + +_BYPASS_HELPER = _BypassHelper() + + def _get_via_subprocess(url: str, retry: int, cancel_flag: Event | None = None) -> str: """Run the browser bypass in a helper process isolated from gunicorn/gevent.""" _check_cancellation(cancel_flag, "Bypass cancelled before helper process") @@ -863,50 +1070,15 @@ def _get_via_subprocess(url: str, retry: int, cancel_flag: Event | None = None) # freshly spawned helper would otherwise pre-resolve AA hostnames against the system # resolver - which may be blocked or hijacked by the user's ISP. Pass the parent's # active DNS config so the helper mirrors it (e.g. DoH) when building Chrome's host - # resolver rules. + # resolver rules. Sent with every request, not just at spawn, because a helper outlives + # changes the parent makes to its DNS provider. payload = { "url": url, "retry": retry, "result_path": str(result_path), "dns_config": network.get_dns_config(), } - env_vars = os.environ.copy() - env_vars[_BYPASS_CHILD_ENV] = "1" - env_vars = _prepare_child_browser_env(env_vars) - - proc = subprocess.Popen( - [sys.executable, "-m", "shelfmark.bypass.internal_bypasser"], - stdin=subprocess.PIPE, - text=True, - env=env_vars, - # Give the helper its own session: Chrome, Xvfb and ffmpeg inherit its process - # group, which is what lets the cleanup sweep tell this bypass's browsers apart - # from a concurrent worker's (#1231) and lets us kill the whole tree below. - start_new_session=True, - ) - timed_out = False - try: - proc.communicate(json.dumps(payload), timeout=_BYPASS_SUBPROCESS_TIMEOUT_SECONDS) - except subprocess.TimeoutExpired: - timed_out = True - finally: - # Always tear the session down, not just on timeout: killing the helper alone - # leaves its Chrome and Xvfb running, and those leftovers are what made the next - # worker's browser fail to start in the first place. - _terminate_helper_session(proc) - - if timed_out: - msg = "Internal bypasser helper process timed out" - raise TimeoutError(msg) - - try: - result = json.loads(result_path.read_text()) - except FileNotFoundError as exc: - msg = f"Internal bypasser helper exited without a result (code {proc.returncode})" - raise RuntimeError(msg) from exc - finally: - with suppress(OSError): - result_path.unlink() + result = _BYPASS_HELPER.run(payload, _BYPASS_SUBPROCESS_TIMEOUT_SECONDS, cancel_flag) if not isinstance(result, dict): msg = "Internal bypasser helper returned an invalid result" @@ -1299,9 +1471,20 @@ def _start_parent_watchdog() -> None: ).start() -def _run_child_process() -> int: - """CLI entrypoint used by the Docker helper subprocess.""" - request = json.loads(sys.stdin.read() or "{}") +def _publish_result(result_path: Path, payload: dict[str, Any]) -> None: + """Write the result file atomically. + + The parent decides the request is answered the moment this path exists, so it must + never observe a half-written file. Rename within the same directory is atomic. + """ + tmp_path = result_path.with_name(result_path.name + ".part") + tmp_path.write_text(json.dumps(payload), encoding="utf-8") + tmp_path.replace(result_path) + + +def _handle_child_request(request_line: str) -> int: + """Answer one request from the parent.""" + request = json.loads(request_line or "{}") result_path = Path(str(request["result_path"])) url = str(request["url"]) retry = _coerce_positive_int( @@ -1321,7 +1504,7 @@ def _run_child_process() -> int: "cookies": cookies, "user_agents": user_agents, } - result_path.write_text(json.dumps(payload), encoding="utf-8") + _publish_result(result_path, payload) except Exception as exc: # noqa: BLE001 - helper boundary must serialize failures. payload = { "ok": False, @@ -1329,11 +1512,28 @@ def _run_child_process() -> int: "error": str(exc), "traceback": traceback.format_exc(), } - result_path.write_text(json.dumps(payload), encoding="utf-8") + _publish_result(result_path, payload) return 1 return 0 +def _run_child_process() -> int: + """CLI entrypoint used by the Docker helper subprocess. + + Serves one request per line of stdin until the parent closes the pipe, so a burst of + protected requests - a single search is several - pays the interpreter start and imports + once instead of per request. Each bypass still gets its own browser, closed before the + answer is published. + """ + exit_code = 0 + for line in sys.stdin: + request_line = line.strip() + if not request_line: + continue + exit_code = _handle_child_request(request_line) + return exit_code + + if __name__ == "__main__": # Started here rather than in _run_child_process() so it only ever watches a real # spawned helper, never a test or an embedded call. diff --git a/shelfmark/config/settings.py b/shelfmark/config/settings.py index 202668ae..d0b44d11 100644 --- a/shelfmark/config/settings.py +++ b/shelfmark/config/settings.py @@ -1655,6 +1655,19 @@ def cloudflare_bypass_settings() -> list[SettingsField]: requires_restart=True, show_when={"field": "USING_EXTERNAL_BYPASSER", "value": True}, ), + NumberField( + key="BYPASS_BROWSER_IDLE_TIMEOUT", + label="Bypasser Idle Timeout (seconds)", + description=( + "How long the bypass helper process may sit unused before it is shut down. " + "Higher keeps more searches fast, lower frees memory sooner." + ), + default=180, + min_value=30, + max_value=3600, + requires_restart=True, + show_when={"field": "USING_EXTERNAL_BYPASSER", "value": False}, + ), ] diff --git a/tests/bypass/test_internal_bypasser.py b/tests/bypass/test_internal_bypasser.py index 16f13fd1..4b41fdce 100644 --- a/tests/bypass/test_internal_bypasser.py +++ b/tests/bypass/test_internal_bypasser.py @@ -545,30 +545,59 @@ def test_cleanup_is_skipped_without_proc(monkeypatch, tmp_path): assert internal_bypasser._cleanup_orphan_processes() == 0 +class _FakeHelperStdin: + """The request pipe: a write is how the helper receives one request.""" + + def __init__(self, process): + self._process = process + self.closed = False + + def write(self, data): + self._process.serve(data) + + def flush(self): + return None + + def close(self): + self.closed = True + + class _FakeHelperProcess: - """Stand-in for the bypass helper subprocess.""" + """Stand-in for the bypass helper subprocess. + + The helper serves one request per line of stdin and answers by writing the result file + the request named, so that is what this fakes: a write produces an answer. + """ def __init__(self, *_args, **kwargs): self.kwargs = kwargs self.pid = 4242 - self.returncode = 0 - self.timed_out = False + self.returncode = None + self.answers = True self.killed = False self.waited = False + self.stdin = _FakeHelperStdin(self) + self.requests: list[dict] = [] - def communicate(self, payload, timeout=None): - if self.timed_out: - raise subprocess.TimeoutExpired(cmd="helper", timeout=timeout) + def serve(self, payload): request = json.loads(payload) + self.requests.append(request) + if not self.answers: + return result = {"ok": True, "html": "solved", "cookies": {}, "user_agents": {}} Path(request["result_path"]).write_text(json.dumps(result), encoding="utf-8") - return "", "" + + def poll(self): + return self.returncode def kill(self): self.killed = True + self.returncode = -9 def wait(self, timeout=None): self.waited = True + if self.returncode is None: + self.returncode = 0 return self.returncode @@ -578,13 +607,47 @@ def _patch_helper_subprocess(monkeypatch, internal_bypasser, process, killed_gro monkeypatch.setattr( internal_bypasser.os, "killpg", lambda pgid, _sig: killed_groups.append(pgid) ) + # A fresh helper per test: the module-level one is shared, and a process parked by one + # test would be handed to the next. + helper = internal_bypasser._BypassHelper() + monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 0.0) + monkeypatch.setattr(internal_bypasser, "_BYPASS_HELPER", helper) + return helper def test_helper_runs_in_its_own_session_and_is_torn_down(monkeypatch): """Regression test for issue #1231: the helper's Chrome and Xvfb must belong to the helper's own process group, and the whole group must die with it - otherwise the leftovers break the next worker's browser and can only be cleared by a sweep broad - enough to kill a concurrent worker's browser too.""" + enough to kill a concurrent worker's browser too. + + The helper outlives a single request, so the teardown happens when it is dropped rather + than after every solve. Each bypass still closes its own browser, so what survives in + between is the process, not a Chrome. + """ + import shelfmark.bypass.internal_bypasser as internal_bypasser + + processes: list[_FakeHelperProcess] = [] + killed_groups: list[int] = [] + + def _make_process(*args, **kwargs): + process = _FakeHelperProcess(*args, **kwargs) + processes.append(process) + return process + + helper = _patch_helper_subprocess(monkeypatch, internal_bypasser, _make_process, killed_groups) + + assert internal_bypasser._get_via_subprocess("https://example.com", 1) == "solved" + assert processes[0].kwargs["start_new_session"] is True + assert killed_groups == [], "the helper was torn down after a single request" + + helper._discard() + + assert killed_groups == [processes[0].pid] + + +def test_helper_serves_a_second_request_without_respawning(monkeypatch): + """The interpreter start and imports are paid once, not per protected request.""" import shelfmark.bypass.internal_bypasser as internal_bypasser processes: list[_FakeHelperProcess] = [] @@ -597,9 +660,14 @@ def test_helper_runs_in_its_own_session_and_is_torn_down(monkeypatch): _patch_helper_subprocess(monkeypatch, internal_bypasser, _make_process, killed_groups) - assert internal_bypasser._get_via_subprocess("https://example.com", 1) == "solved" - assert processes[0].kwargs["start_new_session"] is True - assert killed_groups == [processes[0].pid] + internal_bypasser._get_via_subprocess("https://example.com/one", 1) + internal_bypasser._get_via_subprocess("https://example.com/two", 1) + + assert len(processes) == 1 + assert [request["url"] for request in processes[0].requests] == [ + "https://example.com/one", + "https://example.com/two", + ] def test_helper_timeout_kills_the_whole_session(monkeypatch): @@ -611,11 +679,12 @@ def test_helper_timeout_kills_the_whole_session(monkeypatch): def _make_process(*args, **kwargs): process = _FakeHelperProcess(*args, **kwargs) - process.timed_out = True + process.answers = False # accepts the request, never writes a result processes.append(process) return process _patch_helper_subprocess(monkeypatch, internal_bypasser, _make_process, killed_groups) + monkeypatch.setattr(internal_bypasser, "_BYPASS_SUBPROCESS_TIMEOUT_SECONDS", 0.1) with pytest.raises(TimeoutError): internal_bypasser._get_via_subprocess("https://example.com", 1) diff --git a/tests/bypass/test_warm_helper.py b/tests/bypass/test_warm_helper.py new file mode 100644 index 00000000..b2a0848c --- /dev/null +++ b/tests/bypass/test_warm_helper.py @@ -0,0 +1,433 @@ +"""Tests for keeping the bypass helper process alive between requests. + +The browser is deliberately not kept: every bypass starts and closes its own Chrome. What +survives is the helper process, whose interpreter start and imports are pure overhead. +""" + +import asyncio +import json + +import pytest + + +class _FakeStdin: + def __init__(self) -> None: + self.closed = False + self.written: list[str] = [] + + def write(self, data: str) -> None: + if self.closed: + raise BrokenPipeError("stdin is closed") + self.written.append(data) + + def flush(self) -> None: + return None + + def close(self) -> None: + self.closed = True + + +class _FakeProc: + """Enough of subprocess.Popen for the helper's process bookkeeping.""" + + _next_pid = 90001 + + def __init__(self) -> None: + self.stdin = _FakeStdin() + self.returncode: int | None = None + # A pid nothing may actually be signalled by: _terminate_helper_session is patched + # out in these tests, and a stray killpg on a live pid would take out the test run. + type(self)._next_pid += 1 + self.pid = type(self)._next_pid + + def poll(self) -> int | None: + return self.returncode + + def wait(self, timeout: float | None = None) -> int: + if self.returncode is None: + self.returncode = 0 + return self.returncode + + def kill(self) -> None: + self.returncode = -9 + + +def _helper_with_fake_spawn(monkeypatch, procs: list[_FakeProc], terminated=None): + """Build a helper that hands out fake processes and never arms a real timer.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + def _spawn(_self) -> _FakeProc: + proc = _FakeProc() + procs.append(proc) + return proc + + def _terminate(proc) -> None: + if terminated is not None: + terminated.append(proc) + + monkeypatch.setattr(internal_bypasser._BypassHelper, "_spawn", _spawn) + monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 0.0) + monkeypatch.setattr(internal_bypasser, "_terminate_helper_session", _terminate) + return internal_bypasser._BypassHelper() + + +def _answered_payload(tmp_path, name: str = "result.json") -> dict: + """A request whose result file already exists, so the helper resolves immediately.""" + result_path = tmp_path / name + result_path.write_text(json.dumps({"ok": True, "html": ""}), encoding="utf-8") + return {"url": "https://example.com", "retry": 1, "result_path": str(result_path)} + + +def test_helper_serves_consecutive_requests_from_one_process(monkeypatch, tmp_path): + """The point of the whole thing: request two and three must not re-pay the spawn.""" + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + + for i in range(3): + result = helper.run(_answered_payload(tmp_path, f"r{i}.json"), timeout=5, cancel_flag=None) + assert result["ok"] is True + + assert len(procs) == 1, "each request spawned its own helper" + assert len(procs[0].stdin.written) == 3 + assert all(line.endswith("\n") for line in procs[0].stdin.written), ( + "requests must be newline-delimited or the helper's loop cannot split them" + ) + + +def test_helper_respawns_after_the_previous_one_died(monkeypatch, tmp_path): + """A helper can be reaped while idle; the next request must not fail on it.""" + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + + helper.run(_answered_payload(tmp_path, "a.json"), timeout=5, cancel_flag=None) + procs[0].returncode = 1 # died between requests + + result = helper.run(_answered_payload(tmp_path, "b.json"), timeout=5, cancel_flag=None) + + assert result["ok"] is True + assert len(procs) == 2 + + +def test_helper_retries_once_when_the_pipe_breaks_on_write(monkeypatch, tmp_path): + """poll() can still say alive when the far end is already gone.""" + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + + helper.run(_answered_payload(tmp_path, "a.json"), timeout=5, cancel_flag=None) + procs[0].stdin.closed = True # pipe gone, but poll() still reports running + + result = helper.run(_answered_payload(tmp_path, "b.json"), timeout=5, cancel_flag=None) + + assert result["ok"] is True + assert len(procs) == 2 + + +def test_helper_reports_a_helper_that_exits_without_answering(monkeypatch, tmp_path): + import shelfmark.bypass.internal_bypasser as internal_bypasser + + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + + payload = { + "url": "https://example.com", + "retry": 1, + "result_path": str(tmp_path / "never-written.json"), + } + + def _die_on_write(_self, proc, _line) -> None: + proc.returncode = 3 + + monkeypatch.setattr(internal_bypasser._BypassHelper, "_write", _die_on_write) + + with pytest.raises(RuntimeError, match="exited without a result"): + helper.run(payload, timeout=5, cancel_flag=None) + + +def test_helper_times_out_and_discards_the_wedged_process(monkeypatch, tmp_path): + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + + payload = { + "url": "https://example.com", + "retry": 1, + "result_path": str(tmp_path / "never-written.json"), + } + + with pytest.raises(TimeoutError): + helper.run(payload, timeout=0.05, cancel_flag=None) + + assert helper._proc is None, "a wedged helper must not be handed to the next request" + + +def test_idle_reaper_rearms_when_work_arrived_while_it_waited(monkeypatch, tmp_path): + """The timer fires on its own thread and can lose the race against a new request.""" + import time + + import shelfmark.bypass.internal_bypasser as internal_bypasser + + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None) + + rearmed: list[bool] = [] + monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 3600.0) + monkeypatch.setattr( + internal_bypasser._BypassHelper, "_arm_idle_timer", lambda _self: rearmed.append(True) + ) + helper._last_used = time.monotonic() + + helper._reap_if_idle() + + assert rearmed == [True] + assert helper._proc is not None, "helper was killed despite recent work" + + +def test_idle_reaper_closes_a_genuinely_idle_helper(monkeypatch, tmp_path): + import time + + import shelfmark.bypass.internal_bypasser as internal_bypasser + + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None) + + monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 60.0) + helper._last_used = time.monotonic() - 120 + + helper._reap_if_idle() + + assert helper._proc is None + assert procs[0].stdin.closed + + +def test_discard_tears_down_the_whole_session(monkeypatch, tmp_path): + """Dropping the helper must reach its browser tree, not just the helper itself. + + The helper is a session leader (start_new_session), so a Chrome left behind by one + killed mid-bypass would keep a process group alive that the cleanup sweep is then not + allowed to reclaim - the leak #1231 was about. + """ + procs: list[_FakeProc] = [] + terminated: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs, terminated) + + helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None) + helper._discard() + + assert terminated == [procs[0]] + + +def test_helper_asks_before_it_kills(monkeypatch, tmp_path): + """An idle helper should get to exit on its own; the kill is the fallback.""" + procs: list[_FakeProc] = [] + helper = _helper_with_fake_spawn(monkeypatch, procs) + + helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None) + helper._discard() + + assert procs[0].stdin.closed, "stdin must be closed to end the helper's request loop" + assert procs[0].returncode == 0, "an idle helper should have exited on its own" + + +def _bypass_with_recorded_driver(monkeypatch, get_impl): + """Wire up a bypass whose browser creation and closing are observable.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + driver = object() + closed: list[object] = [] + + async def _create(_url): + return driver + + async def _close(drv): + closed.append(drv) + + monkeypatch.setattr(internal_bypasser, "_create_cdp_browser", _create) + monkeypatch.setattr(internal_bypasser, "_get", get_impl) + monkeypatch.setattr(internal_bypasser, "_close_cdp_driver", _close) + return driver, closed + + +def test_successful_bypass_closes_its_browser(monkeypatch): + """A living helper must not accumulate browsers: each bypass ends with Chrome gone.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + async def _get(_url, _driver, _cancel=None): + return "ok" + + driver, closed = _bypass_with_recorded_driver(monkeypatch, _get) + + result = internal_bypasser._run_bypass_in_current_process("https://example.com", 1) + + assert result == "ok" + assert closed == [driver] + + +def test_failed_bypass_closes_its_browser(monkeypatch): + """The same has to hold when the bypass raises on its way out.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + async def _get(_url, _driver, _cancel=None): + raise internal_bypasser.BypassCancelledError("cancelled") + + driver, closed = _bypass_with_recorded_driver(monkeypatch, _get) + + with pytest.raises(internal_bypasser.BypassCancelledError): + internal_bypasser._run_bypass_in_current_process("https://example.com", 1) + + assert closed == [driver] + + +def test_child_process_serves_every_line_it_is_given(monkeypatch, tmp_path): + """One helper, several requests: the loop is what saves the repeated process start.""" + import io + + import shelfmark.bypass.internal_bypasser as internal_bypasser + + urls: list[str] = [] + + def _fake_get(url, retry=None, cancel_flag=None): + urls.append(url) + return f"{url}" + + requests = [ + {"url": "https://example.com/one", "retry": 1, "result_path": str(tmp_path / "1.json")}, + {"url": "https://example.com/two", "retry": 1, "result_path": str(tmp_path / "2.json")}, + ] + stdin = io.StringIO("\n".join(json.dumps(request) for request in requests) + "\n") + + monkeypatch.setattr(internal_bypasser, "get", _fake_get) + monkeypatch.setattr(internal_bypasser.sys, "stdin", stdin) + + assert internal_bypasser._run_child_process() == 0 + assert urls == ["https://example.com/one", "https://example.com/two"] + + for index, request in enumerate(requests, start=1): + result = json.loads((tmp_path / f"{index}.json").read_text(encoding="utf-8")) + assert result["ok"] is True + assert result["html"] == f"{request['url']}" + + +def test_child_process_keeps_serving_after_a_failed_request(monkeypatch, tmp_path): + """One failing URL must not take the helper - and everything queued - down.""" + import io + + import shelfmark.bypass.internal_bypasser as internal_bypasser + + def _fake_get(url, retry=None, cancel_flag=None): + if url.endswith("boom"): + raise RuntimeError("bypass exploded") + return "ok" + + requests = [ + {"url": "https://example.com/boom", "retry": 1, "result_path": str(tmp_path / "1.json")}, + {"url": "https://example.com/fine", "retry": 1, "result_path": str(tmp_path / "2.json")}, + ] + stdin = io.StringIO("\n".join(json.dumps(request) for request in requests) + "\n") + + monkeypatch.setattr(internal_bypasser, "get", _fake_get) + monkeypatch.setattr(internal_bypasser.sys, "stdin", stdin) + + assert internal_bypasser._run_child_process() == 0 + + failed = json.loads((tmp_path / "1.json").read_text(encoding="utf-8")) + assert failed["ok"] is False + assert failed["error"] == "bypass exploded" + + served = json.loads((tmp_path / "2.json").read_text(encoding="utf-8")) + assert served["ok"] is True + + +def test_result_file_becomes_visible_only_when_complete(tmp_path): + """The parent treats the file's existence as the answer, so no partial writes.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + result_path = tmp_path / "result.json" + internal_bypasser._publish_result(result_path, {"ok": True, "html": ""}) + + assert json.loads(result_path.read_text(encoding="utf-8"))["ok"] is True + assert list(tmp_path.iterdir()) == [result_path], "temporary file was left behind" + + +def test_child_bypass_runs_on_the_long_lived_worker_loop(monkeypatch): + """A helper serving many requests must not build and close a loop per bypass. + + asyncio.run() owns the loop for one call and closes it on the way out, which is why the + child goes through the worker unconditionally: one loop for the process's lifetime. + """ + import shelfmark.bypass.internal_bypasser as internal_bypasser + + monkeypatch.setenv("SHELFMARK_INTERNAL_BYPASSER_CHILD", "1") + + loops: list[asyncio.AbstractEventLoop] = [] + + async def _record_loop(_url, _driver, _cancel=None): + loops.append(asyncio.get_running_loop()) + return "ok" + + _bypass_with_recorded_driver(monkeypatch, _record_loop) + + internal_bypasser._run_bypass_in_current_process("https://example.com", 1) + internal_bypasser._run_bypass_in_current_process("https://example.com", 1) + + assert len(loops) == 2 + assert loops[0] is loops[1], "second bypass ran on a different loop than the first" + assert not loops[0].is_closed() + + +def test_child_bypass_carries_its_own_deadline(monkeypatch): + """The child bounds itself, rather than relying only on the parent's deadline.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + monkeypatch.setenv("SHELFMARK_INTERNAL_BYPASSER_CHILD", "1") + + timeouts: list[float | None] = [] + real_run = internal_bypasser._CDP_WORKER.run + + def _record_timeout(coro, timeout=None): + timeouts.append(timeout) + return real_run(coro, timeout=timeout) + + async def _get(_url, _driver, _cancel=None): + return "ok" + + _bypass_with_recorded_driver(monkeypatch, _get) + monkeypatch.setattr(internal_bypasser._CDP_WORKER, "run", _record_timeout) + + internal_bypasser._run_bypass_in_current_process("https://example.com", 1) + + assert timeouts == [internal_bypasser._CHILD_BYPASS_TIMEOUT_SECONDS] + + +def test_child_deadline_leaves_the_parent_room_to_hear_the_answer(monkeypatch): + """If the parent gave up first it could only kill the helper, losing a warm process.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + assert ( + internal_bypasser._CHILD_BYPASS_TIMEOUT_SECONDS + < internal_bypasser._BYPASS_SUBPROCESS_TIMEOUT_SECONDS + ) + + +def test_in_process_bypass_keeps_the_parents_budget(monkeypatch): + """Non-Docker installs run in-process, where there is no helper to outlive anything.""" + import shelfmark.bypass.internal_bypasser as internal_bypasser + + monkeypatch.delenv("SHELFMARK_INTERNAL_BYPASSER_CHILD", raising=False) + + timeouts: list[float | None] = [] + real_run = internal_bypasser._CDP_WORKER.run + + def _record_timeout(coro, timeout=None): + timeouts.append(timeout) + return real_run(coro, timeout=timeout) + + async def _get(_url, _driver, _cancel=None): + return "ok" + + _bypass_with_recorded_driver(monkeypatch, _get) + monkeypatch.setattr(internal_bypasser._CDP_WORKER, "run", _record_timeout) + + internal_bypasser._run_bypass_in_current_process("https://example.com", 1) + + assert timeouts == [internal_bypasser._IN_PROCESS_BYPASS_TIMEOUT_SECONDS]