feat(browser): concurrent page pool (BROWSER_CONCURRENCY, default 4) + raise mem_limit to 5g #1733
3 changed files with 502 additions and 103 deletions
|
|
@ -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,60 @@ 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 на следующем запросе.
|
||||
|
||||
Сброс ``_page_counter = 0`` происходит АТОМАРНО с реальным relaunch'ем (под
|
||||
launch-локом, только когда guard inflight<=1 пройден). Иначе под конкуренцией
|
||||
N>1 несколько корутин видят counter >= порога, все откладывают relaunch (guard
|
||||
inflight>1), но counter НИКОГДА не сбрасывается → recycle-check срабатывает на
|
||||
каждом fetch вечно, а реальный recycle голодает. При отложенном relaunch'е
|
||||
counter НЕ трогаем — следующий запрос (когда нагрузка спадёт) дожмёт recycle.
|
||||
"""
|
||||
global _page_counter
|
||||
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)
|
||||
_page_counter = 0
|
||||
await _relaunch_browser()
|
||||
|
||||
|
||||
async def _browser_retry_loop() -> None:
|
||||
"""Фоновый retry launch'а браузера с backoff, пока прокси не вернётся.
|
||||
|
||||
|
|
@ -196,10 +286,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 +317,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 +368,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 +382,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 +403,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 +425,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 +460,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 +495,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 +509,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 +533,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 +551,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
|
||||
|
||||
|
|
@ -484,7 +594,11 @@ async def _login_once(params: dict) -> list[dict]:
|
|||
success_cookie: str = params.get("success_cookie", "")
|
||||
wait_ms: int = params.get("wait_ms") or BROWSER_WAIT_MS
|
||||
|
||||
# Страница закрывается в finally: asyncio.CancelledError (BaseException, НЕ
|
||||
# Exception) обходит except-ветку ниже — без finally страница утекала бы при
|
||||
# отмене таска (например, на shutdown). _fetch_once делает то же.
|
||||
page = await browser.new_page() # type: ignore[attr-defined]
|
||||
try:
|
||||
try:
|
||||
await page.goto( # type: ignore[attr-defined]
|
||||
url, timeout=BROWSER_NAV_TIMEOUT_MS, wait_until="domcontentloaded"
|
||||
|
|
@ -551,21 +665,21 @@ async def _login_once(params: dict) -> list[dict]:
|
|||
page_url = page.url # type: ignore[attr-defined]
|
||||
except Exception:
|
||||
pass
|
||||
await page.close() # type: ignore[attr-defined]
|
||||
raise LoginError({
|
||||
"error": f"{type(exc).__name__}: {exc}",
|
||||
"page_url": page_url,
|
||||
"screenshot_b64": screenshot_b64,
|
||||
}) from exc
|
||||
|
||||
finally:
|
||||
await page.close() # type: ignore[attr-defined]
|
||||
|
||||
_page_counter += 1
|
||||
if _page_counter >= BROWSER_RECYCLE_PAGES:
|
||||
logger.info(
|
||||
"tradein-browser: recycle threshold (%d) достигнут после login, перезапуск браузера",
|
||||
BROWSER_RECYCLE_PAGES,
|
||||
)
|
||||
await _relaunch_browser()
|
||||
await _safe_relaunch("recycle-after-login")
|
||||
|
||||
return cookies
|
||||
|
||||
|
|
|
|||
|
|
@ -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,287 @@ 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 "<html>ok</html>"
|
||||
|
||||
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"] == "<html>ok</html>"
|
||||
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 под локом)"
|
||||
|
||||
|
||||
# ── _safe_relaunch resets _page_counter (recycle starvation fix) ─────────────────
|
||||
|
||||
|
||||
def test_safe_relaunch_resets_page_counter_when_proceeds(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""_safe_relaunch сбрасывает _page_counter=0 когда relaunch проходит (inflight<=1).
|
||||
|
||||
Регрессия: без сброса под конкуренцией counter копился вечно (>= порога на
|
||||
каждом fetch) → recycle голодал. Сброс атомарен с relaunch'ем под локом.
|
||||
"""
|
||||
monkeypatch.setattr(server, "_launch_lock", asyncio.Lock())
|
||||
monkeypatch.setattr(server, "_inflight", 1) # мы единственный in-flight
|
||||
monkeypatch.setattr(server, "_page_counter", 999)
|
||||
|
||||
relaunched = {"n": 0}
|
||||
|
||||
async def _fake_relaunch() -> None:
|
||||
relaunched["n"] += 1
|
||||
|
||||
monkeypatch.setattr(server, "_relaunch_browser", _fake_relaunch)
|
||||
|
||||
asyncio.run(server._safe_relaunch("recycle"))
|
||||
|
||||
assert relaunched["n"] == 1, "relaunch должен выполниться при inflight<=1"
|
||||
assert server._page_counter == 0, "counter должен сброситься при реальном relaunch"
|
||||
|
||||
|
||||
def test_safe_relaunch_keeps_page_counter_when_deferred(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""_safe_relaunch НЕ трогает _page_counter когда relaunch отложен (inflight>1).
|
||||
|
||||
Отложенный relaunch оставляет counter, чтобы следующий запрос (когда нагрузка
|
||||
спадёт) дожал recycle — иначе recycle потерялся бы навсегда.
|
||||
"""
|
||||
monkeypatch.setattr(server, "_launch_lock", asyncio.Lock())
|
||||
monkeypatch.setattr(server, "_inflight", 3) # есть параллельные запросы
|
||||
monkeypatch.setattr(server, "_page_counter", 42)
|
||||
|
||||
relaunched = {"n": 0}
|
||||
|
||||
async def _fake_relaunch() -> None:
|
||||
relaunched["n"] += 1
|
||||
|
||||
monkeypatch.setattr(server, "_relaunch_browser", _fake_relaunch)
|
||||
|
||||
asyncio.run(server._safe_relaunch("recycle"))
|
||||
|
||||
assert relaunched["n"] == 0, "relaunch должен быть отложен при inflight>1"
|
||||
assert server._page_counter == 42, "counter НЕ трогаем при отложенном relaunch"
|
||||
|
||||
|
||||
# ── _login_once closes page on CancelledError (BaseException leak fix) ────────────
|
||||
|
||||
|
||||
class _CancellingLoginPage:
|
||||
"""Page для /login, которая бросает CancelledError внутри логина."""
|
||||
|
||||
def __init__(self, tracker: "_Tracker") -> None:
|
||||
self._tracker = tracker
|
||||
self.url = "https://cian.example/login"
|
||||
|
||||
async def goto(self, url: str, **kwargs: Any) -> None:
|
||||
return None
|
||||
|
||||
async def wait_for_timeout(self, ms: int) -> None:
|
||||
return None
|
||||
|
||||
async def fill(self, selector: str, value: str, **kwargs: Any) -> None:
|
||||
# Отмена таска (BaseException, не Exception) — обходит except Exception.
|
||||
raise asyncio.CancelledError()
|
||||
|
||||
async def close(self) -> None:
|
||||
self._tracker.closed += 1
|
||||
|
||||
|
||||
class _CancellingLoginBrowser:
|
||||
def __init__(self, tracker: "_Tracker") -> None:
|
||||
self._tracker = tracker
|
||||
|
||||
async def new_page(self) -> _CancellingLoginPage:
|
||||
self._tracker.opened += 1
|
||||
return _CancellingLoginPage(self._tracker)
|
||||
|
||||
|
||||
def test_login_closes_page_on_cancelled_error(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""_login_once закрывает страницу даже при CancelledError (finally, не except).
|
||||
|
||||
CancelledError — BaseException, обходит except Exception, без finally страница
|
||||
утекала бы при отмене таска (shutdown / dropped client).
|
||||
"""
|
||||
tracker = _Tracker()
|
||||
browser = _CancellingLoginBrowser(tracker)
|
||||
monkeypatch.setattr(server, "_browser", browser)
|
||||
|
||||
params = {
|
||||
"url": "https://cian.example/login",
|
||||
"email": "u@example.com",
|
||||
"password": "secret",
|
||||
"email_selector": "#email",
|
||||
"password_selector": "#password",
|
||||
"submit_selector": "#submit",
|
||||
}
|
||||
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
asyncio.run(server._login_once(params))
|
||||
|
||||
assert tracker.opened == 1
|
||||
assert tracker.closed == 1, "страница должна закрыться в finally при CancelledError"
|
||||
|
||||
|
||||
async def _coro(value: Any) -> Any:
|
||||
"""Хелпер: оборачивает значение в awaitable для подмены request.json()."""
|
||||
return value
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue