From 820fd819bb6921e390940d11a75da8e30bb78397 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 17 Jun 2026 23:55:11 +0300 Subject: [PATCH] feat(browser): concurrent page pool (BROWSER_CONCURRENCY, default 4) + raise mem_limit to 5g --- tradein-mvp/browser/server.py | 163 ++++++++++++++++---- tradein-mvp/browser/test_server_smoke.py | 181 ++++++++++++++++++++++- tradein-mvp/docker-compose.prod.yml | 6 +- 3 files changed, 311 insertions(+), 39 deletions(-) diff --git a/tradein-mvp/browser/server.py b/tradein-mvp/browser/server.py index bd692f82..834ca272 100644 --- a/tradein-mvp/browser/server.py +++ b/tradein-mvp/browser/server.py @@ -16,6 +16,8 @@ WS-сервер) несовместим с современными версия Переменные окружения: BROWSER_PORT — TCP-порт HTTP-сервера (default: 3000) + BROWSER_CONCURRENCY — сколько /fetch обрабатывается параллельно (default: 4, clamp >=1). + Один браузер, N параллельных страниц (по странице на запрос). BROWSER_RECYCLE_PAGES — страниц в одном сеансе браузера до перезапуска (default: 15) BROWSER_NAV_TIMEOUT_MS — таймаут page.goto в мс (default: 60000) BROWSER_WAIT_MS — ожидание гидрации listings после DOMContentLoaded, мс (default: 6000) @@ -44,6 +46,30 @@ logger = logging.getLogger(__name__) # ── конфигурация из env ──────────────────────────────────────────────────────── BROWSER_PORT: int = int(os.environ.get("BROWSER_PORT", "3000")) + + +def _parse_concurrency(raw: str | None) -> int: + """Парсит BROWSER_CONCURRENCY → int, clamp >=1, fallback 4 при мусоре. + + Ops поднимает значение в .env.runtime после смоука. Невалидный ввод + (нечисловой / пустой) → дефолт 4, чтобы кривой конфиг не валил сервис. + """ + if raw is None: + return 4 + try: + value = int(raw) + except (TypeError, ValueError): + logger.warning( + "tradein-browser: невалидный BROWSER_CONCURRENCY=%r — дефолт 4", raw + ) + return 4 + return max(1, value) + + +# Сколько /fetch обрабатывается параллельно. Один браузер, N параллельных +# страниц (по странице на запрос). Дефолт 4 — консервативно; ops поднимет до 8 +# после смоука (firefox-страницы прожорливы по RAM, mem_limit поднят до 5g). +BROWSER_CONCURRENCY: int = _parse_concurrency(os.environ.get("BROWSER_CONCURRENCY")) BROWSER_RECYCLE_PAGES: int = int(os.environ.get("BROWSER_RECYCLE_PAGES", "15")) BROWSER_NAV_TIMEOUT_MS: int = int(os.environ.get("BROWSER_NAV_TIMEOUT_MS", "60000")) # 6000 (не 2500): avito гидрирует listings client-side ПОСЛЕ domcontentloaded; @@ -90,7 +116,17 @@ def _parse_proxy(proxy_url: str | None) -> dict[str, str] | None: _browser: object | None = None # текущий экземпляр Browser _browser_cm: object | None = None # AsyncCamoufox context manager (для __aexit__) _page_counter: int = 0 # страниц с момента последнего (пере)запуска -_fetch_lock: asyncio.Lock | None = None # сериализация fetch-запросов +# _launch_lock сериализует ТОЛЬКО launch/relaunch/recycle браузера (один __aenter__ +# за раз на общий глобал _browser — иначе два launch'а затрут друг друга). НЕ +# сериализует сами /fetch — это делает семафор ниже. +_launch_lock: asyncio.Lock | None = None +# _fetch_sem ограничивает число ПАРАЛЛЕЛЬНЫХ /fetch (и /login) до BROWSER_CONCURRENCY. +# Один браузер, N параллельных страниц. Заменяет старый _fetch_lock (сериализация 1-за-раз). +_fetch_sem: asyncio.Semaphore | None = None +# Число fetch/login в полёте прямо сейчас. Нужно чтобы recycle/crash-relaunch НЕ +# рвал общий браузер из-под живых параллельных страниц (teardown под сиблингами → +# каскад крашей). Relaunch выполняется только когда in-flight == 1 (мы единственные). +_inflight: int = 0 # Resilience (#crash-storm): если прокси (mproxy.site) лежит, camoufox делает # eager public_ip(proxy) для geoip-fingerprint и бросает InvalidProxy. Раньше это @@ -182,6 +218,51 @@ async def _try_launch_browser() -> bool: return False +async def _ensure_browser() -> bool: + """Гарантирует поднятый браузер; одна lazy-попытка launch'а под launch-локом. + + Идемпотентна и конкурентно-безопасна: несколько параллельных /fetch могут + вызвать её одновременно — лок сериализует launch, а double-check внутри лока + не даёт второму launch'у затереть уже поднятый первым браузер. + + Returns: + True если браузер доступен (был или поднялся), False если launch не удался + (прокси лежит) — caller отдаёт 503. + """ + if _browser is not None: + return True + assert _launch_lock is not None, "_launch_lock not initialised" + async with _launch_lock: + if _browser is not None: + return True + await _try_launch_browser() + return _browser is not None + + +async def _safe_relaunch(reason: str) -> None: + """Перезапуск браузера под launch-локом ТОЛЬКО когда мы единственный in-flight. + + Под конкуренцией recycle/crash-relaunch не должен рвать общий браузер из-под + живых параллельных страниц (teardown под сиблингами → каскад крашей). Поэтому: + relaunch выполняется лишь если ``_inflight <= 1`` (мы — единственный активный + запрос). Иначе relaunch ПРОПУСКАЕТСЯ — браузер переиспользуется, а recycle + случится позже, когда параллельная нагрузка спадёт. Crash-recovery при этом не + теряется: per-page краш изолирован (своя страница), а реально мёртвый браузер + поднимет _ensure_browser на следующем запросе. + """ + assert _launch_lock is not None, "_launch_lock not initialised" + async with _launch_lock: + if _inflight > 1: + logger.info( + "tradein-browser: relaunch (%s) отложен — %d параллельных запросов в полёте", + reason, + _inflight, + ) + return + logger.info("tradein-browser: relaunch (%s)", reason) + await _relaunch_browser() + + async def _browser_retry_loop() -> None: """Фоновый retry launch'а браузера с backoff, пока прокси не вернётся. @@ -196,10 +277,10 @@ async def _browser_retry_loop() -> None: delay = _BROWSER_RETRY_MIN_S while _browser is None: await asyncio.sleep(delay) - # Лок сериализует launch с in-flight /fetch lazy-launch и recycle — - # иначе два __aenter__ на один глобал _browser затрут друг друга. - assert _fetch_lock is not None, "_fetch_lock not initialised" - async with _fetch_lock: + # Launch-лок сериализует launch с in-flight lazy-launch и recycle — иначе + # два __aenter__ на один глобал _browser затрут друг друга. + assert _launch_lock is not None, "_launch_lock not initialised" + async with _launch_lock: if _browser is not None: return # handler уже поднял браузер — задача отработала logger.info("tradein-browser: фоновый retry launch'а браузера") @@ -227,8 +308,12 @@ def _start_retry_task(app: web.Application) -> None: async def _on_startup(app: web.Application) -> None: - global _fetch_lock - _fetch_lock = asyncio.Lock() + global _launch_lock, _fetch_sem + _launch_lock = asyncio.Lock() + _fetch_sem = asyncio.Semaphore(BROWSER_CONCURRENCY) + logger.info( + "tradein-browser: BROWSER_CONCURRENCY=%d (параллельных /fetch)", BROWSER_CONCURRENCY + ) # Launch НЕ фатален: если прокси лежит, camoufox бросит InvalidProxy. Раньше # это валило startup → exit(1) → restart-loop. Теперь ловим, остаёмся Up и @@ -274,9 +359,10 @@ async def fetch_handler(request: web.Request) -> web.Response: """POST /fetch {"url": "..."} → {"html": "..."} Выполняет навигацию в браузере и возвращает полный HTML страницы. - Запросы сериализованы через lock (один за раз). + До BROWSER_CONCURRENCY запросов обрабатываются параллельно (семафор), у + каждого — своя страница (изоляция крашей). """ - global _page_counter + global _inflight try: body = await request.json() @@ -287,21 +373,20 @@ async def fetch_handler(request: web.Request) -> web.Response: if not url: return web.json_response({"error": "missing 'url' field"}, status=400) - assert _fetch_lock is not None, "_fetch_lock not initialised" + assert _fetch_sem is not None, "_fetch_sem not initialised" - async with _fetch_lock: + async with _fetch_sem: # Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка - # под локом — вдруг прокси уже вернулся. НИКОГДА не фетчим без прокси/ - # браузера: если всё ещё None → 503 (caller словит как HTTPStatusError - # и обработает как failed fetch, а не как crash). - if _browser is None: - await _try_launch_browser() - if _browser is None: + # под launch-локом — вдруг прокси уже вернулся. НИКОГДА не фетчим без + # прокси/браузера: если всё ещё None → 503 (caller словит как + # HTTPStatusError и обработает как failed fetch, а не как crash). + if not await _ensure_browser(): logger.warning("tradein-browser: /fetch 503 — браузер недоступен (proxy may be down)") return web.json_response( {"error": "browser unavailable (proxy may be down)"}, status=503 ) + _inflight += 1 try: html = await _do_fetch(url) except Exception as exc: @@ -309,14 +394,19 @@ async def fetch_handler(request: web.Request) -> web.Response: "tradein-browser: fetch error url=%r: %s: %s", url, type(exc).__name__, exc ) return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=500) + finally: + _inflight -= 1 return web.json_response({"html": html}) async def _do_fetch(url: str) -> str: - """Одна попытка навигации; при краше браузера — перезапускает и повторяет один раз.""" - global _page_counter + """Одна попытка навигации; при краше браузера — перезапускает и повторяет один раз. + Per-page краш изолирован (своя страница на запрос); _safe_relaunch рвёт общий + браузер лишь когда нет параллельных in-flight запросов, иначе пропускает (мёртвый + браузер поднимет _ensure_browser на следующем запросе). + """ try: return await _fetch_once(url) except Exception as exc: @@ -326,13 +416,19 @@ async def _do_fetch(url: str) -> str: type(exc).__name__, url, ) - await _relaunch_browser() + await _safe_relaunch("crash-recovery") + if not await _ensure_browser(): + raise return await _fetch_once(url) raise async def _fetch_once(url: str) -> str: - """Открывает страницу, переходит по URL, ждёт JS, возвращает HTML.""" + """Открывает СОБСТВЕННУЮ страницу, переходит по URL, ждёт JS, возвращает HTML. + + Каждый параллельный /fetch держит свою page (никакого шеринга — общая page под + конкуренцией ломает навигацию), закрывает её в finally. + """ global _page_counter browser = _browser @@ -355,7 +451,7 @@ async def _fetch_once(url: str) -> str: "tradein-browser: recycle threshold (%d) достигнут, перезапуск браузера", BROWSER_RECYCLE_PAGES, ) - await _relaunch_browser() + await _safe_relaunch("recycle") return html @@ -390,8 +486,10 @@ async def login_handler(request: web.Request) -> web.Response: Выполняет вход на указанную страницу (email + пароль), ждёт появления success_cookie, возвращает полный список cookies браузерного контекста. - Запросы сериализованы через lock (один за раз — один браузер). + Делит тот же семафор/браузер что и /fetch (своя страница на запрос). """ + global _inflight + try: body = await request.json() except Exception: @@ -402,19 +500,18 @@ async def login_handler(request: web.Request) -> web.Response: if missing: return web.json_response({"error": f"missing required fields: {missing}"}, status=400) - assert _fetch_lock is not None, "_fetch_lock not initialised" + assert _fetch_sem is not None, "_fetch_sem not initialised" - async with _fetch_lock: + async with _fetch_sem: # Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал). - # Lazy-попытка под локом, иначе 503 — без прокси не логинимся. - if _browser is None: - await _try_launch_browser() - if _browser is None: + # Lazy-попытка под launch-локом, иначе 503 — без прокси не логинимся. + if not await _ensure_browser(): logger.warning("tradein-browser: /login 503 — браузер недоступен (proxy may be down)") return web.json_response( {"error": "browser unavailable (proxy may be down)"}, status=503 ) + _inflight += 1 try: cookies = await _do_login(body) except LoginError as exc: @@ -427,6 +524,8 @@ async def login_handler(request: web.Request) -> web.Response: exc, ) return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=502) + finally: + _inflight -= 1 return web.json_response({"cookies": cookies}) @@ -443,7 +542,9 @@ async def _do_login(params: dict) -> list[dict]: "tradein-browser: краш браузера (%s) при логине, перезапуск + retry", type(exc).__name__, ) - await _relaunch_browser() + await _safe_relaunch("login-crash-recovery") + if not await _ensure_browser(): + raise return await _login_once(params) raise @@ -565,7 +666,7 @@ async def _login_once(params: dict) -> list[dict]: "tradein-browser: recycle threshold (%d) достигнут после login, перезапуск браузера", BROWSER_RECYCLE_PAGES, ) - await _relaunch_browser() + await _safe_relaunch("recycle-after-login") return cookies diff --git a/tradein-mvp/browser/test_server_smoke.py b/tradein-mvp/browser/test_server_smoke.py index 2725bf6c..6addea83 100644 --- a/tradein-mvp/browser/test_server_smoke.py +++ b/tradein-mvp/browser/test_server_smoke.py @@ -61,16 +61,16 @@ def test_fetch_returns_503_when_browser_unavailable( ) -> None: """/fetch отдаёт 503 (не 500/краш) когда браузер не поднят и launch не удался. - Имитирует «прокси лежит»: _browser=None + _try_launch_browser no-op, + Имитирует «прокси лежит»: _browser=None + _ensure_browser no-op, оставляющий None. Никакого реального camoufox-launch'а / direct-fetch. """ monkeypatch.setattr(server, "_browser", None) - monkeypatch.setattr(server, "_fetch_lock", asyncio.Lock()) + monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4)) - async def _no_op_launch() -> bool: + async def _no_ensure() -> bool: return False # прокси всё ещё недоступен → браузер остаётся None - monkeypatch.setattr(server, "_try_launch_browser", _no_op_launch) + monkeypatch.setattr(server, "_ensure_browser", _no_ensure) request = make_mocked_request( "POST", "/fetch", payload=b'{"url": "https://example.com"}' @@ -92,12 +92,12 @@ def test_login_returns_503_when_browser_unavailable( ) -> None: """/login тоже отдаёт 503 когда браузер не поднят (без credential-логирования).""" monkeypatch.setattr(server, "_browser", None) - monkeypatch.setattr(server, "_fetch_lock", asyncio.Lock()) + monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4)) - async def _no_op_launch() -> bool: + async def _no_ensure() -> bool: return False - monkeypatch.setattr(server, "_try_launch_browser", _no_op_launch) + monkeypatch.setattr(server, "_ensure_browser", _no_ensure) login_body = { "url": "https://cian.example/login", @@ -119,6 +119,173 @@ def test_login_returns_503_when_browser_unavailable( assert "browser unavailable" in body["error"] +# ── BROWSER_CONCURRENCY config parse ──────────────────────────────────────────── + + +def test_concurrency_default_is_4() -> None: + """BROWSER_CONCURRENCY не задан → дефолт 4.""" + assert server._parse_concurrency(None) == 4 + + +def test_concurrency_env_override() -> None: + """Валидное число из env применяется как есть.""" + assert server._parse_concurrency("8") == 8 + assert server._parse_concurrency("1") == 1 + + +def test_concurrency_clamps_to_min_1() -> None: + """0 / отрицательное зажимается до >=1 (семафор с 0 завесил бы сервис).""" + assert server._parse_concurrency("0") == 1 + assert server._parse_concurrency("-5") == 1 + + +def test_concurrency_garbage_falls_back_to_4() -> None: + """Нечисловой мусор → дефолт 4 (кривой конфиг не валит сервис).""" + assert server._parse_concurrency("abc") == 4 + assert server._parse_concurrency("") == 4 + + +def test_startup_creates_semaphore_with_concurrency( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """_on_startup создаёт _fetch_sem с ёмкостью BROWSER_CONCURRENCY + launch-лок.""" + monkeypatch.setattr(server, "BROWSER_CONCURRENCY", 3) + monkeypatch.setattr(server, "_fetch_sem", None) + monkeypatch.setattr(server, "_launch_lock", None) + + async def _no_launch() -> bool: + return True # не запускаем реальный camoufox в тесте + + monkeypatch.setattr(server, "_try_launch_browser", _no_launch) + + app = server.web.Application() + asyncio.run(server._on_startup(app)) + + assert isinstance(server._fetch_sem, asyncio.Semaphore) + assert server._fetch_sem._value == 3 # стартовая ёмкость == concurrency + assert isinstance(server._launch_lock, asyncio.Lock) + + +# ── semaphore limits concurrency + per-page isolation ──────────────────────────── + + +class _FakePage: + """Поддельная page: считает open/close и отмечает пик параллелизма.""" + + def __init__(self, tracker: "_Tracker") -> None: + self._tracker = tracker + + async def goto(self, url: str, **kwargs: Any) -> None: + # Внутри goto держим «активную» страницу — здесь меряем параллелизм. + self._tracker.active += 1 + self._tracker.peak = max(self._tracker.peak, self._tracker.active) + await asyncio.sleep(0.02) + self._tracker.active -= 1 + + async def wait_for_timeout(self, ms: int) -> None: + return None + + async def content(self) -> str: + return "ok" + + async def close(self) -> None: + self._tracker.closed += 1 + + +class _FakeBrowser: + def __init__(self, tracker: "_Tracker") -> None: + self._tracker = tracker + + async def new_page(self) -> _FakePage: + self._tracker.opened += 1 + return _FakePage(self._tracker) + + +class _Tracker: + def __init__(self) -> None: + self.opened = 0 + self.closed = 0 + self.active = 0 + self.peak = 0 + + +def _make_fetch_request() -> Any: + request = make_mocked_request("POST", "/fetch") + request.json = lambda: _coro({"url": "https://example.com"}) # type: ignore[method-assign] + return request + + +def test_semaphore_limits_concurrency_to_n(monkeypatch: pytest.MonkeyPatch) -> None: + """Семафор(N) не даёт больше N одновременных goto, даже при M>N запросах.""" + tracker = _Tracker() + browser = _FakeBrowser(tracker) + monkeypatch.setattr(server, "_browser", browser) + monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(2)) + monkeypatch.setattr(server, "_inflight", 0) + # recycle отключаем — высокий порог, чтобы relaunch не вмешивался. + monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) + + async def _ensure() -> bool: + return True + + monkeypatch.setattr(server, "_ensure_browser", _ensure) + + async def _run() -> None: + await asyncio.gather(*[server.fetch_handler(_make_fetch_request()) for _ in range(6)]) + + asyncio.run(_run()) + + assert tracker.peak <= 2, f"параллелизм {tracker.peak} превысил лимит семафора 2" + assert tracker.opened == 6 + assert tracker.closed == 6 # каждая страница своя, закрыта в finally + + +def test_fetch_opens_and_closes_own_page(monkeypatch: pytest.MonkeyPatch) -> None: + """Один /fetch открывает ровно одну собственную страницу и закрывает её.""" + tracker = _Tracker() + browser = _FakeBrowser(tracker) + monkeypatch.setattr(server, "_browser", browser) + monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4)) + monkeypatch.setattr(server, "_inflight", 0) + monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) + + async def _ensure() -> bool: + return True + + monkeypatch.setattr(server, "_ensure_browser", _ensure) + + response = asyncio.run(server.fetch_handler(_make_fetch_request())) + assert response.status == 200 + assert _json_body(response)["html"] == "ok" + assert tracker.opened == 1 + assert tracker.closed == 1 + + +def test_ensure_browser_serialized_by_launch_lock( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """_ensure_browser сериализует launch: при N параллельных вызовах launch один раз.""" + monkeypatch.setattr(server, "_browser", None) + monkeypatch.setattr(server, "_launch_lock", asyncio.Lock()) + calls = {"n": 0} + + async def _launch_once() -> bool: + calls["n"] += 1 + await asyncio.sleep(0.01) + # имитируем успешный launch — выставляем браузер + monkeypatch.setattr(server, "_browser", object()) + return True + + monkeypatch.setattr(server, "_try_launch_browser", _launch_once) + + async def _run() -> list[bool]: + return await asyncio.gather(*[server._ensure_browser() for _ in range(5)]) + + results = asyncio.run(_run()) + assert all(results) + assert calls["n"] == 1, "launch должен произойти ровно один раз (double-check под локом)" + + async def _coro(value: Any) -> Any: """Хелпер: оборачивает значение в awaitable для подмены request.json().""" return value diff --git a/tradein-mvp/docker-compose.prod.yml b/tradein-mvp/docker-compose.prod.yml index 4642d90f..a349595b 100644 --- a/tradein-mvp/docker-compose.prod.yml +++ b/tradein-mvp/docker-compose.prod.yml @@ -16,12 +16,16 @@ services: browser: image: ghcr.io/lekss361/gendesign-tradein-browser:${IMAGE_TAG:-latest} container_name: tradein-browser - mem_limit: 2.5g # headless Firefox (camoufox) потребляет до ~1.5GB (#884/#905) + # 5g: N параллельных headless-Firefox страниц (BROWSER_CONCURRENCY, default 4) + # — каждая ~0.8-1.5GB. VPS имеет ~6.9G свободно. Было 2.5g под 1-за-раз (#884/#905). + mem_limit: 5g env_file: - path: ./backend/.env.runtime required: false environment: BROWSER_PORT: "3000" + # Сколько /fetch параллельно. Дефолт 4; ops поднимет до 8 в .env.runtime после смоука. + BROWSER_CONCURRENCY: ${BROWSER_CONCURRENCY:-4} # SCRAPER_PROXY_URL читается из .env.runtime (см. env_file выше) expose: - "3000"