Merge pull request 'feat(browser): per-provider concurrency — параллель между поставщиками, ≤1 коннект на поставщика (#1793)' (#1794) from feat/browser-per-provider-concurrency into main

Reviewed-on: #1794

tradein-mvp/browser/test_server.py

@@ -0,0 +1,294 @@
+"""test_server.py — юниты per-provider concurrency для tradein-browser (#1793).
+
+Проверяет ядро per-provider модели:
+  1. _provider_for_url — корректный маппинг host → provider (avito/cian/yandex/generic);
+  2. _provider_proxy — BROWSER_PROXY_* primary с фолбэком на legacy *_PROXY_URL;
+  3. _resolve_provider — явный body["provider"] перебивает host, иначе host;
+  4. конкурентность: два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи),
+     два к ОДНОМУ — сериализованы (один лок). Через _do_fetch с asyncio.Event и
+     замером перекрытия по времени.
+
+camoufox НЕ запускается: _ensure_browser / _do_fetch мокаются. Хендлер вызывается
+напрямую (make_mocked_request не поднимает сокет).
+
+Запуск (из tradein-mvp/browser/)::
+
+    python -m pytest test_server.py -q
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+from aiohttp.test_utils import make_mocked_request
+
+# server.py — не пакет (отдельный сервис без __init__/pyproject). Грузим по пути.
+_SERVER_PATH = Path(__file__).resolve().parent / "server.py"
+_spec = importlib.util.spec_from_file_location("tradein_browser_server", _SERVER_PATH)
+assert _spec is not None and _spec.loader is not None
+server = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(server)
+
+
+def _json_body(response: Any) -> dict[str, Any]:
+    return json.loads(response.body.decode())
+
+
+@pytest.fixture(autouse=True)
+def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Чистое per-provider состояние + инициализированный _locks_guard на каждый тест."""
+    monkeypatch.setattr(server, "_browsers", {})
+    monkeypatch.setattr(server, "_browser_cms", {})
+    monkeypatch.setattr(server, "_page_counters", {})
+    monkeypatch.setattr(server, "_locks", {})
+    monkeypatch.setattr(server, "_retry_tasks", {})
+    monkeypatch.setattr(server, "_locks_guard", asyncio.Lock())
+
+
+# ── _provider_for_url ────────────────────────────────────────────────────────────
+
+
+def test_provider_for_url_avito() -> None:
+    assert server._provider_for_url("https://www.avito.ru/moskva/kvartiry") == "avito"
+    assert server._provider_for_url("https://m.avito.ru/item/123") == "avito"
+
+
+def test_provider_for_url_cian() -> None:
+    assert server._provider_for_url("https://www.cian.ru/sale/flat/1/") == "cian"
+    assert server._provider_for_url("https://spb.cian.ru/x") == "cian"
+
+
+def test_provider_for_url_yandex() -> None:
+    assert server._provider_for_url("https://realty.yandex.ru/offer/1/") == "yandex"
+
+
+def test_provider_for_url_generic_fallback() -> None:
+    assert server._provider_for_url("https://example.com/x") == "generic"
+    assert server._provider_for_url("https://domclick.ru/x") == "generic"
+    # Битый / пустой URL — без хоста → generic, не падаем.
+    assert server._provider_for_url("not-a-url") == "generic"
+    assert server._provider_for_url("") == "generic"
+
+
+# ── _provider_proxy ──────────────────────────────────────────────────────────────
+
+
+def test_provider_proxy_uses_browser_proxy_primary(monkeypatch: pytest.MonkeyPatch) -> None:
+    """BROWSER_PROXY_* — primary источник для каждого поставщика."""
+    monkeypatch.setenv("BROWSER_PROXY_AVITO", "http://u:p@avito-proxy:8080")
+    monkeypatch.setenv("BROWSER_PROXY_CIAN", "http://u:p@cian-proxy:8080")
+    monkeypatch.setenv("BROWSER_PROXY_YANDEX", "http://u:p@yandex-proxy:8080")
+    assert server._provider_proxy("avito") == "http://u:p@avito-proxy:8080"
+    assert server._provider_proxy("cian") == "http://u:p@cian-proxy:8080"
+    assert server._provider_proxy("yandex") == "http://u:p@yandex-proxy:8080"
+
+
+def test_provider_proxy_falls_back_to_legacy(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Без BROWSER_PROXY_* берётся legacy *_PROXY_URL фолбэк."""
+    monkeypatch.delenv("BROWSER_PROXY_AVITO", raising=False)
+    monkeypatch.delenv("BROWSER_PROXY_CIAN", raising=False)
+    monkeypatch.delenv("BROWSER_PROXY_YANDEX", raising=False)
+    monkeypatch.setenv("AVITO_PROXY_URL", "http://u:p@legacy-avito:8080")
+    monkeypatch.setenv("CIAN_PROXY_URL", "http://u:p@legacy-cian:8080")
+    monkeypatch.setenv("YANDEX_PROXY_URL", "http://u:p@legacy-yandex:8080")
+    assert server._provider_proxy("avito") == "http://u:p@legacy-avito:8080"
+    assert server._provider_proxy("cian") == "http://u:p@legacy-cian:8080"
+    assert server._provider_proxy("yandex") == "http://u:p@legacy-yandex:8080"
+
+
+def test_provider_proxy_generic(monkeypatch: pytest.MonkeyPatch) -> None:
+    """generic: SCRAPER_PROXY_URL primary, AVITO_PROXY_URL фолбэк."""
+    monkeypatch.delenv("SCRAPER_PROXY_URL", raising=False)
+    monkeypatch.setenv("AVITO_PROXY_URL", "http://u:p@avito:8080")
+    assert server._provider_proxy("generic") == "http://u:p@avito:8080"
+    monkeypatch.setenv("SCRAPER_PROXY_URL", "http://u:p@scraper:8080")
+    assert server._provider_proxy("generic") == "http://u:p@scraper:8080"
+
+
+def test_provider_proxy_none_when_unset(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Ни одной env → None (direct, как одиночный браузер без прокси)."""
+    for var in (
+        "BROWSER_PROXY_CIAN",
+        "CIAN_PROXY_URL",
+    ):
+        monkeypatch.delenv(var, raising=False)
+    assert server._provider_proxy("cian") is None
+
+
+def test_provider_proxy_browser_proxy_wins_over_legacy(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """BROWSER_PROXY_* перебивает legacy *_PROXY_URL (http vs возможный socks5)."""
+    monkeypatch.setenv("BROWSER_PROXY_CIAN", "http://u:p@http-cian:8080")
+    monkeypatch.setenv("CIAN_PROXY_URL", "socks5://u:p@socks-cian:1080")
+    assert server._provider_proxy("cian") == "http://u:p@http-cian:8080"
+
+
+# ── _resolve_provider ────────────────────────────────────────────────────────────
+
+
+def test_resolve_provider_by_host() -> None:
+    assert server._resolve_provider({}, "https://www.avito.ru/x") == "avito"
+    assert server._resolve_provider({}, "https://www.cian.ru/x") == "cian"
+
+
+def test_resolve_provider_explicit_overrides_host() -> None:
+    """Явный валидный body["provider"] перебивает host."""
+    assert server._resolve_provider({"provider": "yandex"}, "https://www.avito.ru/x") == "yandex"
+
+
+def test_resolve_provider_invalid_explicit_falls_back_to_host() -> None:
+    """Невалидный body["provider"] игнорируется → берём по host."""
+    assert server._resolve_provider({"provider": "bogus"}, "https://www.cian.ru/x") == "cian"
+    assert server._resolve_provider({"provider": 123}, "https://www.cian.ru/x") == "cian"
+
+
+def test_resolve_provider_uses_source_field() -> None:
+    """browser_fetcher шлёт {"url","source"} — explicit берётся и из "source"."""
+    assert server._resolve_provider({"source": "avito"}, "https://example.com/x") == "avito"
+    assert server._resolve_provider({"source": "yandex"}, "https://www.avito.ru/x") == "yandex"
+
+
+def test_resolve_provider_domclick_source_to_generic() -> None:
+    """domclick scraper шлёт source='domklik' (нет в PROVIDERS) + хост domclick.ru.
+
+    'domklik' не валиден → host-detect domclick.ru → generic. Намеренно: в prod
+    нет выделенного BROWSER_PROXY_DOMCLICK.
+    """
+    assert (
+        server._resolve_provider({"source": "domklik"}, "https://domclick.ru/x") == "generic"
+    )
+
+
+# ── _lock_for: один лок на поставщика ────────────────────────────────────────────
+
+
+def test_lock_for_is_stable_per_provider() -> None:
+    """_lock_for возвращает один и тот же объект-лок для поставщика, разные — для разных."""
+
+    async def _run() -> tuple[Any, Any, Any]:
+        a1 = await server._lock_for("avito")
+        a2 = await server._lock_for("avito")
+        c1 = await server._lock_for("cian")
+        return a1, a2, c1
+
+    a1, a2, c1 = asyncio.run(_run())
+    assert a1 is a2, "лок одного поставщика стабилен"
+    assert a1 is not c1, "локи разных поставщиков различны"
+
+
+# ── concurrency: параллель между поставщиками, сериализация внутри ───────────────
+
+
+def _make_fetch_request(url: str) -> Any:
+    request = make_mocked_request("POST", "/fetch")
+    request.json = lambda: _coro({"url": url})  # type: ignore[method-assign]
+    return request
+
+
+class _OverlapProbe:
+    """Мок _do_fetch: меряет максимальное число одновременно активных fetch per-provider."""
+
+    def __init__(self, hold: float = 0.05) -> None:
+        self.hold = hold
+        self.active: dict[str, int] = {}
+        self.peak: dict[str, int] = {}
+        self.global_active = 0
+        self.global_peak = 0
+
+    async def __call__(self, provider: str, url: str) -> str:
+        self.active[provider] = self.active.get(provider, 0) + 1
+        self.peak[provider] = max(self.peak.get(provider, 0), self.active[provider])
+        self.global_active += 1
+        self.global_peak = max(self.global_peak, self.global_active)
+        try:
+            await asyncio.sleep(self.hold)
+        finally:
+            self.active[provider] -= 1
+            self.global_active -= 1
+        return f"<html>{provider}</html>"
+
+
+def test_same_provider_fetches_are_serialized(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Два /fetch к ОДНОМУ поставщику сериализованы (один лок) — peak==1."""
+    probe = _OverlapProbe()
+
+    async def _ensure(provider: str) -> bool:
+        server._browsers[provider] = object()
+        return True
+
+    monkeypatch.setattr(server, "_ensure_browser", _ensure)
+    monkeypatch.setattr(server, "_do_fetch", probe)
+
+    async def _run() -> None:
+        await asyncio.gather(
+            server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
+            server.fetch_handler(_make_fetch_request("https://www.avito.ru/b")),
+            server.fetch_handler(_make_fetch_request("https://www.avito.ru/c")),
+        )
+
+    asyncio.run(_run())
+    assert probe.peak["avito"] == 1, "внутри поставщика строго ≤1 одновременный fetch"
+
+
+def test_different_providers_fetch_concurrently(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи) — global peak>1."""
+    probe = _OverlapProbe()
+
+    async def _ensure(provider: str) -> bool:
+        server._browsers[provider] = object()
+        return True
+
+    monkeypatch.setattr(server, "_ensure_browser", _ensure)
+    monkeypatch.setattr(server, "_do_fetch", probe)
+
+    async def _run() -> list[Any]:
+        return await asyncio.gather(
+            server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
+            server.fetch_handler(_make_fetch_request("https://www.cian.ru/b")),
+            server.fetch_handler(_make_fetch_request("https://realty.yandex.ru/c")),
+        )
+
+    responses = asyncio.run(_run())
+    assert all(r.status == 200 for r in responses)
+    # >=2 (не ==3): на нагруженном CI планировщик может не вывести все три корутины
+    # в overlap одновременно; перекрытие хотя бы двух уже доказывает параллельность
+    # между поставщиками. Same-provider тест (peak==1) остаётся строгим.
+    assert probe.global_peak >= 2, "разные поставщики должны фетчить конкурентно"
+    # У каждого поставщика по одному запросу — внутренний peak == 1.
+    assert probe.peak == {"avito": 1, "cian": 1, "yandex": 1}
+
+
+def test_mixed_load_serializes_within_parallelizes_across(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Смешанная нагрузка: avito×2 сериализованы, но параллельны с cian×1."""
+    probe = _OverlapProbe()
+
+    async def _ensure(provider: str) -> bool:
+        server._browsers[provider] = object()
+        return True
+
+    monkeypatch.setattr(server, "_ensure_browser", _ensure)
+    monkeypatch.setattr(server, "_do_fetch", probe)
+
+    async def _run() -> None:
+        await asyncio.gather(
+            server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
+            server.fetch_handler(_make_fetch_request("https://www.avito.ru/b")),
+            server.fetch_handler(_make_fetch_request("https://www.cian.ru/c")),
+        )
+
+    asyncio.run(_run())
+    assert probe.peak["avito"] == 1, "avito сериализован сам с собой"
+    assert probe.peak["cian"] == 1
+    assert probe.global_peak >= 2, "avito и cian должны перекрываться во времени"
+
+
+async def _coro(value: Any) -> Any:
+    return value

tradein-mvp/browser/test_server_smoke.py

@@ -2,9 +2,13 @@
 
 Полный браузер (camoufox subprocess) юнит-тестить непрактично, поэтому здесь
 только смоук: (1) модуль импортируется; (2) /fetch и /login отдают 503 когда
-браузер не поднят (прокси лежит) — критичный resilience-инвариант, защищающий
-от crash-storm. camoufox в окружении теста НЕ нужен: launch замокан no-op'ом,
-имитируя «прокси всё ещё недоступен».
+браузер поставщика не поднят (прокси лежит) — критичный resilience-инвариант,
+защищающий от crash-storm. camoufox в окружении теста НЕ нужен: launch замокан
+no-op'ом, имитируя «прокси всё ещё недоступен».
+
+Per-provider модель (#1793): состояние keyed по provider (avito/cian/yandex/
+generic), доступ к инстансу сериализован _locks[provider]; разные поставщики
+работают параллельно. Конкурентность проверяется в test_server.py.
 
 Запуск (из tradein-mvp/browser/)::
 
@@ -38,6 +42,17 @@ def _json_body(response: Any) -> dict[str, Any]:
     return json.loads(response.body.decode())
 
 
+@pytest.fixture(autouse=True)
+def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Чистое per-provider состояние + инициализированный _locks_guard на каждый тест."""
+    monkeypatch.setattr(server, "_browsers", {})
+    monkeypatch.setattr(server, "_browser_cms", {})
+    monkeypatch.setattr(server, "_page_counters", {})
+    monkeypatch.setattr(server, "_locks", {})
+    monkeypatch.setattr(server, "_retry_tasks", {})
+    monkeypatch.setattr(server, "_locks_guard", asyncio.Lock())
+
+
 def test_module_imports() -> None:
     """Смоук: модуль импортируется и экспонирует ключевые хендлеры."""
     assert callable(server.fetch_handler)
@@ -45,40 +60,44 @@ def test_module_imports() -> None:
     assert callable(server.health_handler)
 
 
-def test_health_reports_browser_not_ready(monkeypatch: pytest.MonkeyPatch) -> None:
-    """/health всегда 200, browser_ready=False когда браузер не поднят."""
-    monkeypatch.setattr(server, "_browser", None)
+def test_health_reports_browsers_not_ready() -> None:
+    """/health всегда 200, browsers[<provider>]=False когда инстансы не подняты."""
     request = make_mocked_request("GET", "/health")
     response = asyncio.run(server.health_handler(request))
     assert response.status == 200
     body = _json_body(response)
     assert body["status"] == "ok"
-    assert body["browser_ready"] is False
+    assert body["browsers"] == {"avito": False, "cian": False, "yandex": False, "generic": False}
+
+
+def test_health_reports_partial_readiness() -> None:
+    """/health отражает per-provider readiness: один поднят, остальные нет."""
+    server._browsers["avito"] = object()
+    request = make_mocked_request("GET", "/health")
+    response = asyncio.run(server.health_handler(request))
+    body = _json_body(response)
+    assert body["browsers"]["avito"] is True
+    assert body["browsers"]["cian"] is False
 
 
 def test_fetch_returns_503_when_browser_unavailable(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """/fetch отдаёт 503 (не 500/краш) когда браузер не поднят и launch не удался.
+    """/fetch отдаёт 503 (не 500/краш) когда инстанс поставщика не поднят.
 
-    Имитирует «прокси лежит»: _browser=None + _ensure_browser no-op,
-    оставляющий None. Никакого реального camoufox-launch'а / direct-fetch.
+    Имитирует «прокси лежит»: _ensure_browser no-op, оставляющий инстанс None.
+    Никакого реального camoufox-launch'а / direct-fetch.
     """
-    monkeypatch.setattr(server, "_browser", None)
-    monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4))
 
-    async def _no_ensure() -> bool:
+    async def _no_ensure(provider: str) -> bool:
         return False  # прокси всё ещё недоступен → браузер остаётся None
 
     monkeypatch.setattr(server, "_ensure_browser", _no_ensure)
 
-    request = make_mocked_request(
-        "POST", "/fetch", payload=b'{"url": "https://example.com"}'
-    )
+    request = make_mocked_request("POST", "/fetch")
 
     async def _run() -> Any:
-        # make_mocked_request не парсит тело — подменяем .json() на нужный payload.
-        request.json = lambda: _coro({"url": "https://example.com"})  # type: ignore[method-assign]
+        request.json = lambda: _coro({"url": "https://www.avito.ru/x"})  # type: ignore[method-assign]
         return await server.fetch_handler(request)
 
     response = asyncio.run(_run())
@@ -90,17 +109,15 @@ def test_fetch_returns_503_when_browser_unavailable(
 def test_login_returns_503_when_browser_unavailable(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """/login тоже отдаёт 503 когда браузер не поднят (без credential-логирования)."""
-    monkeypatch.setattr(server, "_browser", None)
-    monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4))
+    """/login тоже отдаёт 503 когда инстанс не поднят (без credential-логирования)."""
 
-    async def _no_ensure() -> bool:
+    async def _no_ensure(provider: str) -> bool:
         return False
 
     monkeypatch.setattr(server, "_ensure_browser", _no_ensure)
 
     login_body = {
-        "url": "https://cian.example/login",
+        "url": "https://www.cian.ru/login",
         "email": "u@example.com",
         "password": "secret",
         "email_selector": "#email",
@@ -119,54 +136,7 @@ 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 ────────────────────────────
+# ── per-provider state isolation ─────────────────────────────────────────────────
 
 
 class _FakePage:
@@ -176,13 +146,9 @@ class _FakePage:
         self._tracker = tracker
 
     async def route(self, pattern: str, handler: Any) -> None:
-        # _apply_resource_block навешивает route-interception перед goto (perf:
-        # блокировка image/media/font). Заглушка no-op — поведение route проверяется
-        # отдельно в test_server_resource_block.py.
         return None
 
     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)
@@ -215,47 +181,19 @@ class _Tracker:
         self.peak = 0
 
 
-def _make_fetch_request() -> Any:
+def _make_fetch_request(url: str = "https://www.avito.ru/x") -> Any:
     request = make_mocked_request("POST", "/fetch")
-    request.json = lambda: _coro({"url": "https://example.com"})  # type: ignore[method-assign]
+    request.json = lambda: _coro({"url": url})  # 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)
+    server._browsers["avito"] = _FakeBrowser(tracker)
     monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000)
 
-    async def _ensure() -> bool:
+    async def _ensure(provider: str) -> bool:
         return True
 
     monkeypatch.setattr(server, "_ensure_browser", _ensure)
@@ -267,82 +205,31 @@ def test_fetch_opens_and_closes_own_page(monkeypatch: pytest.MonkeyPatch) -> Non
     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}
+# ── _relaunch resets per-provider page counter (recycle) ─────────────────────────
 
-    async def _launch_once() -> bool:
-        calls["n"] += 1
-        await asyncio.sleep(0.01)
-        # имитируем успешный launch — выставляем браузер
-        monkeypatch.setattr(server, "_browser", object())
+
+def test_fetch_recycles_after_threshold(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Достижение BROWSER_RECYCLE_PAGES → _relaunch_browser + сброс счётчика."""
+    tracker = _Tracker()
+    server._browsers["avito"] = _FakeBrowser(tracker)
+    server._page_counters["avito"] = 0
+    monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 1)
+
+    relaunched = {"n": 0}
+
+    async def _fake_relaunch(provider: str) -> None:
+        relaunched["n"] += 1
+        server._page_counters[provider] = 0
+
+    async def _ensure(provider: str) -> bool:
         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)
+    monkeypatch.setattr(server, "_ensure_browser", _ensure)
 
-    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"
+    asyncio.run(server.fetch_handler(_make_fetch_request()))
+    assert relaunched["n"] == 1, "recycle должен сработать на пороге"
+    assert server._page_counters["avito"] == 0
 
 
 # ── _login_once closes page on CancelledError (BaseException leak fix) ────────────
@@ -353,7 +240,7 @@ class _CancellingLoginPage:
 
     def __init__(self, tracker: "_Tracker") -> None:
         self._tracker = tracker
-        self.url = "https://cian.example/login"
+        self.url = "https://www.cian.ru/login"
 
     async def goto(self, url: str, **kwargs: Any) -> None:
         return None
@@ -362,7 +249,6 @@ class _CancellingLoginPage:
         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:
@@ -381,17 +267,12 @@ class _CancellingLoginBrowser:
 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).
-    """
+    """_login_once закрывает страницу даже при CancelledError (finally, не except)."""
     tracker = _Tracker()
-    browser = _CancellingLoginBrowser(tracker)
-    monkeypatch.setattr(server, "_browser", browser)
+    server._browsers["cian"] = _CancellingLoginBrowser(tracker)
 
     params = {
-        "url": "https://cian.example/login",
+        "url": "https://www.cian.ru/login",
         "email": "u@example.com",
         "password": "secret",
         "email_selector": "#email",
@@ -400,7 +281,7 @@ def test_login_closes_page_on_cancelled_error(
     }
 
     with pytest.raises(asyncio.CancelledError):
-        asyncio.run(server._login_once(params))
+        asyncio.run(server._login_once("cian", params))
 
     assert tracker.opened == 1
     assert tracker.closed == 1, "страница должна закрыться в finally при CancelledError"