refactor: consolidate browser loop management across Google Search, Indeed Jobs, and Reddit platforms

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

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."""

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

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())
+    )

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 "<html>home</html>", 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"])