浏览器自动恢复逻辑

app/services/browser_pool.py

@@ -38,6 +38,8 @@ class BrowserPool:
         self._contexts_since_start = 0
         self._browser_started_at: float | None = None
         self._active_contexts = 0
+        # 浏览器实例「代次」：每次成功 start 自增，用于并发重建时判断当前实例是否已被他人换新
+        self._browser_generation = 0
 
     @property
     def enabled(self) -> bool:
@@ -105,6 +107,7 @@ class BrowserPool:
             self._started = True
             self._startup_error = None
             self._contexts_since_start = 0
+            self._browser_generation += 1
             self._browser_started_at = asyncio.get_running_loop().time()
             logger.info("浏览器启动完成")
         except Exception as exc:
@@ -159,10 +162,20 @@ class BrowserPool:
         self._playwright_cm = None
         self._started = False
 
-    async def _restart_browser(self, reason: str) -> None:
-        """在浏览器掉线后串行重建 Playwright 运行时。"""
+    async def _restart_browser(self, reason: str, *, observed_generation: int | None = None) -> None:
+        """在浏览器掉线后串行重建 Playwright 运行时。
+
+        observed_generation 为调用方观察到异常时的浏览器代次：
+        - 传入时：若持锁后代次已变化，说明其他协程已重建，直接返回；否则强制重建
+          （此场景已确认 new_context 因 driver 断链失败，不再依赖 is_connected 探测，
+          规避 connection_lost 事件滞后导致的误判）。
+        - 不传时：沿用 is_connected 探测，连通则跳过（用于获取页面前的预检）。
+        """
         async with self._restart_lock:
-            if self._is_browser_connected():
+            if observed_generation is not None:
+                if self._browser_generation != observed_generation:
+                    return
+            elif self._is_browser_connected():
                 return
 
             logger.warning("检测到浏览器连接不可用，准备重建浏览器实例：reason=%s", reason)
@@ -227,14 +240,22 @@ class BrowserPool:
                 raise BrowserUnavailableError(self._startup_error or "Browser recycle failed")
 
     @staticmethod
-    def _should_retry_context_creation(exc: Exception) -> bool:
-        """识别 Playwright driver 断链类错误，允许执行一次重建后重试。"""
+    def is_disconnect_error(exc: Exception) -> bool:
+        """识别 Playwright driver 断链类错误，作为「重建后重试」的统一判据。
+
+        浏览器池内部与上层（CloudflareSessionManager）共用此判据，集中维护标记，
+        避免各处散落的字符串匹配漂移。
+        """
         message = str(exc).lower()
         retry_markers = (
             "connection closed while reading from the driver",
             "browser has been closed",
             "target page, context or browser has been closed",
             "connection closed",
+            # driver 子进程掉线导致 asyncio 传输管道关闭，写入命令时同步报错
+            "the handler is closed",
+            "transport closed",
+            "unable to perform operation",
         )
         return any(marker in message for marker in retry_markers)
 
@@ -262,16 +283,32 @@ class BrowserPool:
             except Exception:
                 logger.exception("关闭旧的保留调试上下文失败")
 
-    async def _build_context_obj(self, storage_state: dict[str, Any] | None) -> Any:
-        """创建浏览器上下文（含断链重试），仅负责创建本身。
+    async def _run_with_reconnect(self, factory: Any, what: str) -> Any:
+        """执行 factory()；若因 driver 断链失败，重建浏览器后重试一次。
+
+        factory 必须是「可重复调用、每次从零创建上下文/页面」的协程工厂——因为
+        重建浏览器会令旧上下文一并失效，恢复的唯一正确粒度是重新创建。
+        借助 _browser_generation 守卫，并发场景下只重建一次，后到者复用新实例。
+        """
+        for attempt in range(2):
+            observed_generation = self._browser_generation
+            try:
+                return await factory()
+            except Exception as exc:
+                if attempt == 0 and self.is_disconnect_error(exc):
+                    logger.warning("%s 因浏览器断链失败，重建后重试一次：err=%s", what, exc)
+                    await self._restart_browser(
+                        f"{what} failed: {exc}", observed_generation=observed_generation
+                    )
+                    continue
+                raise
+
+    async def _new_context_raw(self, storage_state: dict[str, Any] | None) -> Any:
+        """仅创建浏览器上下文本身并计入回收统计，不含断链重试与生命周期管理。
 
-        计入回收阈值统计；不负责并发信号量与关闭，由调用方管理生命周期。
         启用 stealth 时，playwright 已被 use_async 钩子包装，new_context 会自动
         应用 stealth（含 HTTP sec-ch-ua），此处无需再手动 apply。
         """
-        context = None
-        for attempt in range(2):
-            try:
         context = await self._browser.new_context(
             locale=self._settings.browser_locale,
             timezone_id=self._settings.browser_timezone_id,
@@ -279,18 +316,23 @@ class BrowserPool:
             viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900},
             storage_state=storage_state,
         )
-                break
-            except Exception as exc:
-                if attempt == 0 and self._should_retry_context_creation(exc):
-                    logger.warning("创建浏览器上下文失败，尝试重建浏览器后重试一次：err=%s", exc)
-                    await self._restart_browser(f"new_context failed: {exc}")
-                    continue
-                raise
-
         # 上下文创建成功，计入回收阈值统计
         self._contexts_since_start += 1
         return context
 
+    async def _open_context_and_page(self, storage_state: dict[str, Any] | None) -> tuple[Any, Any]:
+        """创建上下文并在其内开页面，作为「断链重试」的最小恢复单元。
+
+        页面创建失败时关闭半开上下文，避免重建/重试时泄漏。
+        """
+        context = await self._new_context_raw(storage_state)
+        try:
+            page = await self._configure_page(context)
+        except Exception:
+            await self.close_context(context)
+            raise
+        return context, page
+
     @staticmethod
     async def _configure_page(context: Any) -> Any:
         """在上下文中创建页面并设置反检测请求头。"""
@@ -306,7 +348,16 @@ class BrowserPool:
         调用方需在不再使用时调用 close_context 释放。
         """
         await self._ensure_browser_ready()
-        return await self._build_context_obj(storage_state)
+        return await self._run_with_reconnect(
+            lambda: self._new_context_raw(storage_state), "创建长存上下文"
+        )
+
+    async def ensure_ready(self) -> None:
+        """对外暴露的浏览器就绪保证：连接断开时自动重建。
+
+        供上层在捕获断链异常后、重试前主动调用，将重建判定收敛到浏览器池内。
+        """
+        await self._ensure_browser_ready()
 
     @asynccontextmanager
     async def page_in_context(self, context: Any) -> AsyncIterator[Any]:
@@ -396,12 +447,14 @@ class BrowserPool:
         try:
             # 创建上下文前检查是否需要主动回收（仅在空闲时执行，不打断进行中的会话）
             await self._maybe_recycle()
-            context = await self._build_context_obj(storage_state)
-            # 上下文创建成功，标记为在用（回收阈值计数已在 _build_context_obj 内完成）
+            # 上下文与页面创建作为一个整体进行断链重试：context 与 browser 实例绑定，
+            # 重建 browser 后必须连同 context 一起重新创建，二者不可拆开重试。
+            context, page = await self._run_with_reconnect(
+                lambda: self._open_context_and_page(storage_state), "打开页面会话"
+            )
+            # 创建成功后才标记为在用（回收阈值计数已在 _new_context_raw 内完成）
             self._active_contexts += 1
             context_counted = True
-
-            page = await self._configure_page(context)
             yield context, page
         except Exception:
             if self._settings.browser_keep_open_on_error_effective and context is not None:

app/services/cloudflare_session.py

@@ -96,7 +96,10 @@ class CloudflareSessionManager:
                 return self._to_verified_session(existing)
 
         logger.debug("创建新会话：session_name=%s force_refresh=%s", session_name, force_refresh)
-        session = await self._create_verified_session(session_name=session_name, target_url=target_url)
+        session = await self._build_with_browser_retry(
+            f"创建验证会话({session_name})",
+            lambda: self._create_verified_session(session_name=session_name, target_url=target_url),
+        )
         return session
 
     async def fetch_html(self, target_url: str) -> tuple[str, VerifiedSession]:
@@ -135,9 +138,28 @@ class CloudflareSessionManager:
                 return html, self._warm_to_session(warm)
 
             logger.info("创建保活上下文：session_name=%s", session_name)
-            warm, html = await self._build_warm_context(session_name, target_url)
+            warm, html = await self._build_with_browser_retry(
+                f"创建保活上下文({session_name})",
+                lambda: self._build_warm_context(session_name, target_url),
+            )
             return html, self._warm_to_session(warm)
 
+    async def _build_with_browser_retry(self, what: str, factory: Any) -> Any:
+        """执行 factory()；若因浏览器断链失败，重建浏览器后整段重试一次。
+
+        factory 每次都从零创建上下文与页面（_build_warm_context / _create_verified_session
+        均如此），故断链重建后重试是安全的。此层覆盖 new_page、goto、content 等
+        「上下文创建之后」才发生的断链——这些阶段浏览器池内逐操作的重试无法触及。
+        """
+        try:
+            return await factory()
+        except Exception as exc:
+            if not BrowserPool.is_disconnect_error(exc):
+                raise
+            logger.warning("%s 因浏览器断链失败，重建后重试一次：err=%s", what, exc)
+            await self._browser_pool.ensure_ready()
+            return await factory()
+
     async def _reuse_warm(self, session_name: str, target_url: str) -> tuple[str, WarmContext] | None:
         """尝试复用已有保活上下文。
 

tests/test_browser_reconnect.py

@@ -0,0 +1,162 @@
+"""浏览器池 driver 断链自动恢复逻辑的单元测试（不启动真实 Chromium）。
+
+覆盖本次修复的核心：
+- is_disconnect_error 对 WriteUnixTransport "the handler is closed" 等断链错误的识别
+- _run_with_reconnect 在断链后重建一次并重试成功
+- _browser_generation 守卫：并发下他人已重建时不重复重建
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.core.config import Settings
+from app.services.browser_pool import BrowserPool
+
+
+class _FakeBrowser:
+    def __init__(self, connected: bool = True):
+        self._connected = connected
+
+    def is_connected(self) -> bool:
+        return self._connected
+
+
+def _make_pool(**overrides) -> BrowserPool:
+    settings = Settings(**overrides)
+    pool = BrowserPool(settings)
+    pool._browser = _FakeBrowser()
+    pool._started = True
+    pool._browser_generation = 1
+    return pool
+
+
+# 复现线上崩溃日志里的真实报错文本
+_HANDLER_CLOSED = (
+    "Stealth.hooked_new_context_async: unable to perform operation on "
+    "<WriteUnixTransport closed=True reading=False 0x7f9a27185a40>; the handler is closed"
+)
+
+
+def test_is_disconnect_error_recognizes_closed_transport():
+    assert BrowserPool.is_disconnect_error(RuntimeError(_HANDLER_CLOSED)) is True
+    assert BrowserPool.is_disconnect_error(Exception("connection closed")) is True
+    # 业务异常不应被误判为断链
+    assert BrowserPool.is_disconnect_error(ValueError("invalid selector")) is False
+
+
+@pytest.mark.asyncio
+async def test_run_with_reconnect_rebuilds_then_retries(monkeypatch):
+    """首次因断链失败 → 重建一次 → 第二次成功。"""
+    pool = _make_pool()
+    rebuilt: list[str] = []
+
+    async def fake_restart(reason, *, observed_generation=None):
+        rebuilt.append(reason)
+        pool._browser_generation += 1  # 模拟 start() 自增代次
+
+    monkeypatch.setattr(pool, "_restart_browser", fake_restart)
+
+    attempts = {"n": 0}
+
+    async def factory():
+        attempts["n"] += 1
+        if attempts["n"] == 1:
+            raise RuntimeError(_HANDLER_CLOSED)
+        return "ok"
+
+    result = await pool._run_with_reconnect(factory, "测试操作")
+
+    assert result == "ok"
+    assert attempts["n"] == 2
+    assert len(rebuilt) == 1
+
+
+@pytest.mark.asyncio
+async def test_run_with_reconnect_gives_up_after_one_retry(monkeypatch):
+    """连续两次断链：只重试一次，第二次仍失败则抛出。"""
+    pool = _make_pool()
+
+    async def fake_restart(reason, *, observed_generation=None):
+        pool._browser_generation += 1
+
+    monkeypatch.setattr(pool, "_restart_browser", fake_restart)
+
+    attempts = {"n": 0}
+
+    async def factory():
+        attempts["n"] += 1
+        raise RuntimeError(_HANDLER_CLOSED)
+
+    with pytest.raises(RuntimeError):
+        await pool._run_with_reconnect(factory, "测试操作")
+
+    assert attempts["n"] == 2  # 初次 + 一次重试
+
+
+@pytest.mark.asyncio
+async def test_run_with_reconnect_passes_through_non_disconnect(monkeypatch):
+    """非断链异常不触发重建，直接抛出。"""
+    pool = _make_pool()
+    rebuilt: list[str] = []
+
+    async def fake_restart(reason, *, observed_generation=None):
+        rebuilt.append(reason)
+
+    monkeypatch.setattr(pool, "_restart_browser", fake_restart)
+
+    async def factory():
+        raise ValueError("业务错误")
+
+    with pytest.raises(ValueError):
+        await pool._run_with_reconnect(factory, "测试操作")
+
+    assert rebuilt == []
+
+
+@pytest.mark.asyncio
+async def test_restart_browser_skips_when_generation_changed():
+    """持锁后代次已变化（他人已重建）→ 直接返回，不重复重建。"""
+    pool = _make_pool()
+
+    closed: list[str] = []
+
+    async def fake_close():
+        closed.append("close")
+
+    pool.close = fake_close  # type: ignore[method-assign]
+
+    # 调用方观察到的是旧代次，而当前代次已被他人推进
+    await pool._restart_browser("stale", observed_generation=pool._browser_generation - 1)
+
+    assert closed == []  # 未触发关闭/重建
+
+
+@pytest.mark.asyncio
+async def test_open_context_and_page_closes_context_on_page_failure(monkeypatch):
+    """页面创建失败时关闭半开上下文，避免泄漏。"""
+    pool = _make_pool()
+
+    class _Ctx:
+        def __init__(self):
+            self.closed = False
+
+        async def new_page(self):
+            raise RuntimeError("new_page boom")
+
+    ctx = _Ctx()
+
+    async def fake_new_context_raw(storage_state):
+        return ctx
+
+    closed: list[object] = []
+
+    async def fake_close_context(c):
+        closed.append(c)
+
+    monkeypatch.setattr(pool, "_new_context_raw", fake_new_context_raw)
+    monkeypatch.setattr(pool, "close_context", fake_close_context)
+
+    with pytest.raises(RuntimeError, match="new_page boom"):
+        await pool._open_context_and_page(None)
+
+    assert closed == [ctx]