diff --git a/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_pipeline_and_manifest.py b/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_pipeline_and_manifest.py index fa87838d..587d5190 100644 --- a/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_pipeline_and_manifest.py +++ b/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_pipeline_and_manifest.py @@ -108,6 +108,111 @@ def factory(_cfg): assert summary.count("stored") == 1 # retry on the fresh browser succeeded +def test_concurrent_rebuild_survives_poisoned_thread_loop_state(tmp_path, make_fetcher): + """Regression: a SIGKILLed sync-Playwright browser leaves its asyncio loop + registered as *running* on the worker thread (teardown is thread-affine, so + ``_close_quietly``'s helper thread can't clear it). The rebuild must reset + that thread-local state — otherwise ``sync_playwright().start()`` raises + "Sync API inside the asyncio loop" and future.result() kills the whole run. + """ + import asyncio + import threading + + config = ArchiveConfig(s3_prefix="t", concurrency=1) + store = ContentStore(LocalBlobStore(tmp_path), config) + urls = ["https://a.test/x", "https://b.test/y"] + builds = {"n": 0} + + class _SyncPlaywrightAlike: + """Mimics sync Playwright's thread behavior: __enter__ refuses if the + thread already reports a running loop, then registers its own; a killed + browser errors on fetch; teardown from a foreign thread fails the way a + greenlet does, leaving the loop state poisoned.""" + + name = "pwalike" + + def __init__(self, dead: bool, inner): + self._dead = dead + self._inner = inner + self._thread = None + + def __enter__(self): + try: + asyncio.get_running_loop() + except RuntimeError: + pass + else: # what playwright's sync context manager raises + raise RuntimeError( + "It looks like you are using Playwright Sync API inside " + "the asyncio loop." + ) + self._thread = threading.current_thread() + asyncio.events._set_running_loop(asyncio.new_event_loop()) + return self + + def fetch(self, url): + if self._dead: + raise RuntimeError("Target page, context or browser has been closed") + return self._inner.fetch(url) + + def __exit__(self, *exc): + if threading.current_thread() is not self._thread: + raise RuntimeError("cannot switch to a different thread") + asyncio.events._set_running_loop(None) + + def factory(_cfg): + builds["n"] += 1 + inner = make_fetcher() + for u in urls: + inner.add(u) + return _SyncPlaywrightAlike(dead=builds["n"] == 1, inner=inner) + + try: + summary = capture_urls_concurrent(urls, store, config, factory) + finally: + asyncio.events._set_running_loop(None) # never leak into other tests + + assert builds["n"] == 2 # death detected -> loop state reset -> rebuilt + assert summary.count("stored") == 2 # retried URL and the rest captured + + +def test_concurrent_failed_rebuild_does_not_kill_the_run(tmp_path, make_fetcher): + """If the rebuild itself fails (e.g. a transient launch error), the run must + keep going: the URL keeps its error outcome and the next URL retries the + rebuild.""" + from contextlib import contextmanager + + config = ArchiveConfig(s3_prefix="t", concurrency=1) + store = ContentStore(LocalBlobStore(tmp_path), config) + urls = ["https://a.test/x", "https://b.test/y"] + builds = {"n": 0} + + class _DeadBrowser: + name = "dead" + + def fetch(self, url): + raise RuntimeError("Target page, context or browser has been closed") + + @contextmanager + def factory(_cfg): + builds["n"] += 1 + if builds["n"] == 1: + yield _DeadBrowser() # first browser is dead + elif builds["n"] == 2: + raise RuntimeError("browser launch flaked") # rebuild attempt fails + else: + fetcher = make_fetcher() + for u in urls: + fetcher.add(u) + yield fetcher # second rebuild attempt works + + summary = capture_urls_concurrent(urls, store, config, factory) + + assert builds["n"] == 3 # dead -> failed rebuild -> successful rebuild + assert summary.count("error") == 1 # first URL kept its error outcome + assert summary.count("stored") == 1 # second URL captured after recovery + + class _BoomFetcher: """Raises an unexpected (non-FetchError) exception, like a bad screenshot.""" diff --git a/forecasting_tools/agents_and_tools/source_archive/fetchers/playwright_fetcher.py b/forecasting_tools/agents_and_tools/source_archive/fetchers/playwright_fetcher.py index efba5575..d81ea82e 100644 --- a/forecasting_tools/agents_and_tools/source_archive/fetchers/playwright_fetcher.py +++ b/forecasting_tools/agents_and_tools/source_archive/fetchers/playwright_fetcher.py @@ -149,11 +149,24 @@ def __enter__(self) -> "PlaywrightFetcher": return self def __exit__(self, *exc) -> None: - if self._browser is not None: - self._browser.close() + # close() raises when the browser process is already gone (crashed or + # killed by the pipeline's reaper). Still attempt stop(): it tears down + # the driver and un-registers the sync API's event loop from this + # thread — without that, the next sync_playwright().start() here fails + # with "Sync API inside the asyncio loop". + try: + if self._browser is not None: + self._browser.close() + except Exception as e: + logger.info("browser close failed (already dead?): %s", e) + finally: self._browser = None - if self._playwright is not None: - self._playwright.stop() + try: + if self._playwright is not None: + self._playwright.stop() + except Exception as e: + logger.info("playwright stop failed: %s", e) + finally: self._playwright = None def _settle(self, page) -> None: diff --git a/forecasting_tools/agents_and_tools/source_archive/pipeline.py b/forecasting_tools/agents_and_tools/source_archive/pipeline.py index 67f5817b..e5564669 100644 --- a/forecasting_tools/agents_and_tools/source_archive/pipeline.py +++ b/forecasting_tools/agents_and_tools/source_archive/pipeline.py @@ -134,6 +134,43 @@ def _close() -> None: done.wait(timeout_s) +def _running_loop(): + import asyncio + + try: + return asyncio.get_running_loop() + except RuntimeError: + return None + + +def _restore_thread_loop_state(baseline) -> None: + """Un-poison a worker thread's asyncio state after abandoning a dead + sync-Playwright instance. + + Sync Playwright drives its asyncio loop *on the calling thread* (via a + greenlet), so while an instance is alive the thread is marked as being + inside a running event loop. A clean ``stop()`` clears that mark — but a + SIGKILLed browser can't be closed cleanly: ``close()``/``stop()`` fail or + hang, and both are thread-affine, so :func:`_close_quietly`'s helper thread + can't reach them either. The abandoned loop then stays registered as + running, and the next ``sync_playwright().start()`` on this thread refuses + with "Playwright Sync API inside the asyncio loop", killing the rebuild. + + Resetting the thread-local marker to its pre-fetcher ``baseline`` lets the + rebuilt fetcher start a fresh loop. The old loop object is leaked on + purpose (its browser processes are swept by the reaper); that is the price + of recovering without touching thread-affine Playwright internals. + """ + import asyncio + + if _running_loop() is baseline: + return + try: + asyncio.events._set_running_loop(baseline) + except Exception: + pass + + def _reap_browser_descendants() -> None: """Best-effort: kill automation Chromium descending from this process. Used both to recover a wedged worker (kill its browser so the blocked sync call @@ -219,6 +256,7 @@ def supervisor() -> None: def work(idx: int, shard: list[str]) -> list[CaptureOutcome]: outcomes: list[CaptureOutcome] = [] + baseline_loop = _running_loop() # almost always None; see _restore_... cm = fetcher_factory(config) pipeline = CapturePipeline(cm.__enter__(), store) try: @@ -231,16 +269,31 @@ def work(idx: int, shard: list[str]) -> list[CaptureOutcome]: "browser died; rebuilding worker %d, retrying %s", idx, url ) _close_quietly(cm) - cm = fetcher_factory(config) - pipeline = CapturePipeline(cm.__enter__(), store) - with hb_lock: - heartbeats[idx] = time.monotonic() - outcome = pipeline.capture_url(url) # one retry on a fresh browser + _restore_thread_loop_state(baseline_loop) + try: + cm = fetcher_factory(config) + pipeline = CapturePipeline(cm.__enter__(), store) + except Exception as e: + # A failed rebuild must never kill the run: keep this + # URL's error outcome and move on — the still-dead + # pipeline makes the next URL error with a dead-browser + # reason, which retries the rebuild. + logger.warning( + "worker %d rebuild failed (%s); keeping error outcome", + idx, + e, + ) + else: + with hb_lock: + heartbeats[idx] = time.monotonic() + # one retry on a fresh browser + outcome = pipeline.capture_url(url) outcomes.append(outcome) with hb_lock: heartbeats[idx] = None finally: _close_quietly(cm) + _restore_thread_loop_state(baseline_loop) return outcomes supervisor_thread = threading.Thread(target=supervisor, daemon=True)