From 1c8a15dfaba64323cb229660f48e78215e247401 Mon Sep 17 00:00:00 2001 From: "DESKTOP-RTLN3BA\\$punk" Date: Fri, 24 Jul 2026 15:01:01 -0700 Subject: [PATCH] refactor: consolidate browser loop management across Google Search, Indeed Jobs, and Reddit platforms --- .../platforms/google_search/fetch.py | 42 +++--------------- .../platforms/indeed_jobs/fetch.py | 20 +++++++-- .../app/proprietary/platforms/reddit/fetch.py | 27 +++++++----- surfsense_backend/app/utils/browser_loop.py | 44 +++++++++++++++++++ .../indeed_jobs/test_fetch_resilience.py | 33 ++++++++++++++ 5 files changed, 116 insertions(+), 50 deletions(-) create mode 100644 surfsense_backend/app/utils/browser_loop.py diff --git a/surfsense_backend/app/proprietary/platforms/google_search/fetch.py b/surfsense_backend/app/proprietary/platforms/google_search/fetch.py index 49c6fcabd..bc9c4402f 100644 --- a/surfsense_backend/app/proprietary/platforms/google_search/fetch.py +++ b/surfsense_backend/app/proprietary/platforms/google_search/fetch.py @@ -35,7 +35,6 @@ import contextlib import logging import os import random -import sys import threading import time from urllib.parse import urlsplit, urlunsplit @@ -46,6 +45,7 @@ from app.proprietary.platforms.google_search import ( captcha as _captcha, pool_store as _store, ) +from app.utils.browser_loop import in_browser_loop as _in_browser_loop from app.utils.captcha import captcha_enabled, get_captcha_config from app.utils.proxy import get_proxy_url @@ -393,41 +393,11 @@ _MOBILE_UA = "Mozilla/5.0 (Android 14; Mobile; rv:132.0) Gecko/132.0 Firefox/132 _MOBILE_VIEWPORT = {"width": 412, "height": 915} -# patchright launches Chromium via asyncio.create_subprocess_exec, which the -# server's main loop cannot do on Windows (main.py pins a SelectorEventLoop -# for psycopg; Selector loops raise NotImplementedError on subprocess_exec). -# All browser work therefore runs on ONE dedicated background loop that is -# explicitly subprocess-capable; callers await it across threads. This also -# keeps the persistent AsyncStealthySession (and its async page_action) intact -# — the sync-fetcher-in-a-thread pattern the other scrapers use would tear -# down the browser on every fetch. -_browser_loop: asyncio.AbstractEventLoop | None = None -_browser_loop_guard = threading.Lock() - - -def _get_browser_loop() -> asyncio.AbstractEventLoop: - """The lazily-started, process-wide event loop the browser lives on.""" - global _browser_loop - with _browser_loop_guard: - if _browser_loop is None: - loop = ( - asyncio.ProactorEventLoop() - if sys.platform == "win32" - else asyncio.new_event_loop() - ) - threading.Thread( - target=loop.run_forever, name="google-search-browser", daemon=True - ).start() - _browser_loop = loop - return _browser_loop - - -async def _in_browser_loop(coro): - """Run ``coro`` on the browser loop and await its result from this loop.""" - return await asyncio.wrap_future( - asyncio.run_coroutine_threadsafe(coro, _get_browser_loop()) - ) - +# All browser work runs on the shared subprocess-capable loop (Windows: the +# server's SelectorEventLoop cannot spawn Chromium; see app.utils.browser_loop). +# This also keeps the persistent AsyncStealthySession (and its async +# page_action) intact — the sync-fetcher-in-a-thread pattern would tear down +# the browser on every fetch. # One live browser per layout (desktop / mobile — the UA and viewport are # session-level context options). Launching Chromium costs ~5 s, so it's paid diff --git a/surfsense_backend/app/proprietary/platforms/indeed_jobs/fetch.py b/surfsense_backend/app/proprietary/platforms/indeed_jobs/fetch.py index c3be9c2a8..e29aef0a9 100644 --- a/surfsense_backend/app/proprietary/platforms/indeed_jobs/fetch.py +++ b/surfsense_backend/app/proprietary/platforms/indeed_jobs/fetch.py @@ -20,6 +20,7 @@ from datetime import UTC, datetime from typing import Any, Protocol from urllib.parse import urlparse +from app.utils.browser_loop import in_browser_loop from app.utils.proxy import get_proxy_url logger = logging.getLogger(__name__) @@ -109,14 +110,23 @@ class IndeedSession: self._warmed: set[str] = set() self.rotations = 0 + # The session's whole lifecycle (build, start, fetch, close) is marshalled + # onto the shared browser loop: patchright can't spawn Chromium from the + # server's Windows SelectorEventLoop (see app.utils.browser_loop), and its + # internals are bound to the loop they started on. + async def start(self) -> None: - self._session = self._factory() - await self._session.start() + async def _build_and_start() -> _Session: + session = self._factory() + await session.start() + return session + + self._session = await in_browser_loop(_build_and_start()) async def close(self) -> None: if self._session is not None: with suppress(Exception): - await self._session.close() + await in_browser_loop(self._session.close()) self._session = None self._warmed.clear() @@ -130,7 +140,9 @@ class IndeedSession: async def _timed_fetch(self, url: str, **kwargs: Any) -> Any: assert self._session is not None coro: Awaitable[Any] = self._session.fetch(url, **kwargs) - return await asyncio.wait_for(coro, timeout=_PAGE_TIMEOUT_S) + # wait_for runs on the browser loop too, so its timeout task lives on + # the same loop as the fetch it cancels. + return await in_browser_loop(asyncio.wait_for(coro, timeout=_PAGE_TIMEOUT_S)) async def _ensure_warm(self, domain: str) -> None: """Land on the domain home with a Google referer before scraping it.""" diff --git a/surfsense_backend/app/proprietary/platforms/reddit/fetch.py b/surfsense_backend/app/proprietary/platforms/reddit/fetch.py index 50fad389d..a1ac77fc6 100644 --- a/surfsense_backend/app/proprietary/platforms/reddit/fetch.py +++ b/surfsense_backend/app/proprietary/platforms/reddit/fetch.py @@ -43,6 +43,7 @@ from urllib.parse import urlencode from scrapling.fetchers import AsyncFetcher, AsyncStealthySession, FetcherSession +from app.utils.browser_loop import in_browser_loop from app.utils.proxy import get_proxy_url, get_sticky_proxy_url # Shared cross-country rotation walk (also used by the TikTok sibling). Kept under @@ -331,18 +332,24 @@ async def warm_session(proxy: str | None) -> dict[str, str] | None: """ if proxy is None: return None + + # Runs on the shared browser loop: patchright can't spawn Chromium from the + # server's Windows SelectorEventLoop (see app.utils.browser_loop). + async def _warm() -> dict[str, str] | None: + async with AsyncStealthySession( + headless=True, + google_search=True, + network_idle=False, + proxy=proxy, + timeout=_WARM_TIMEOUT_MS, + ) as sess: + page = await sess.fetch(_WARM_HTML_URL) + jar = _browser_cookie_jar(page) + return jar if _LOID_COOKIE in jar else None + async with _warm_slots: try: - async with AsyncStealthySession( - headless=True, - google_search=True, - network_idle=False, - proxy=proxy, - timeout=_WARM_TIMEOUT_MS, - ) as sess: - page = await sess.fetch(_WARM_HTML_URL) - jar = _browser_cookie_jar(page) - return jar if _LOID_COOKIE in jar else None + return await in_browser_loop(_warm()) except Exception as e: # a browser crash must not abort the flow logger.warning("[reddit] browser warm failed: %s", e) return None diff --git a/surfsense_backend/app/utils/browser_loop.py b/surfsense_backend/app/utils/browser_loop.py new file mode 100644 index 000000000..d424b7c6b --- /dev/null +++ b/surfsense_backend/app/utils/browser_loop.py @@ -0,0 +1,44 @@ +"""Process-wide, subprocess-capable event loop for browser work. + +patchright launches Chromium via ``asyncio.create_subprocess_exec``, which the +server's main loop cannot do on Windows (main.py pins a SelectorEventLoop for +psycopg; Selector loops raise NotImplementedError on subprocess_exec). All +browser work (scrapling stealth sessions across the scrapers) therefore runs +on ONE dedicated background loop that is explicitly subprocess-capable; +callers await it across threads. This also lets persistent sessions survive +across fetches — the sync-fetcher-in-a-thread pattern would tear the browser +down every time. +""" + +from __future__ import annotations + +import asyncio +import sys +import threading + +_browser_loop: asyncio.AbstractEventLoop | None = None +_browser_loop_guard = threading.Lock() + + +def get_browser_loop() -> asyncio.AbstractEventLoop: + """The lazily-started, process-wide event loop all browser work runs on.""" + global _browser_loop + with _browser_loop_guard: + if _browser_loop is None: + loop = ( + asyncio.ProactorEventLoop() + if sys.platform == "win32" + else asyncio.new_event_loop() + ) + threading.Thread( + target=loop.run_forever, name="browser-loop", daemon=True + ).start() + _browser_loop = loop + return _browser_loop + + +async def in_browser_loop(coro): + """Run ``coro`` on the browser loop and await its result from this loop.""" + return await asyncio.wrap_future( + asyncio.run_coroutine_threadsafe(coro, get_browser_loop()) + ) diff --git a/surfsense_backend/tests/unit/platforms/indeed_jobs/test_fetch_resilience.py b/surfsense_backend/tests/unit/platforms/indeed_jobs/test_fetch_resilience.py index 258f5154d..684f7a46c 100644 --- a/surfsense_backend/tests/unit/platforms/indeed_jobs/test_fetch_resilience.py +++ b/surfsense_backend/tests/unit/platforms/indeed_jobs/test_fetch_resilience.py @@ -2,6 +2,9 @@ from __future__ import annotations +import asyncio +import sys + import pytest from app.proprietary.platforms.indeed_jobs.fetch import ( @@ -95,6 +98,36 @@ async def test_max_rotations_zero_fails_fast(): assert session.rotations == 0 +def test_browser_work_marshalled_off_selector_loop(): + """The Windows regression this guards: main.py runs the server on a + SelectorEventLoop (psycopg needs it), and Selector loops cannot spawn + subprocesses — patchright's Chromium launch died with NotImplementedError. + A fake session that really spawns a subprocess proves the whole session + lifecycle now runs on the shared subprocess-capable browser loop. + """ + + class _SpawningSession: + async def start(self) -> None: + # The exact call that used to blow up on the server loop. + proc = await asyncio.create_subprocess_exec( + sys.executable, "-c", "print('ok')", stdout=asyncio.subprocess.PIPE + ) + await proc.communicate() + + async def close(self) -> None: + pass + + async def fetch(self, url: str, **_: object) -> _FakePage: + return _FakePage(_OK_HTML if "/jobs" in url else "home", url) + + async def main() -> None: + session = IndeedSession(_SpawningSession) + assert await session.fetch_html(_URL) == _OK_HTML + await session.close() + + asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) + + @pytest.mark.asyncio async def test_warms_domain_once_without_rotation(): ctrl = _Controller(["OK", "OK"])