fix(tradein/proxy): бан площадки снимает узел адресно, а не гасит его здоровье для всех источников

Сайдкар на бан-странице Авито отвечает HTTP 500 с ban_page-маркером, клиент
поднимает SidecarBanPageError — но она подкласс httpx.HTTPStatusError, и общий
except Exception в _post_fetch звал mark_health(ok=False). Это ГЛОБАЛЬНОЕ решение
по узлу: три бан-страницы Авито выбивали его из выдачи и Яндексу, и Циану, и
Домклику (прод-замер 31.08-01.09: узлы 9/13/14 на потолке MAX_CONSECUTIVE_FAILS
при banned_for_source=0, живая проба тех же узлов проходила).

Теперь бан-страница ловится отдельной веткой ДО общего except и уходит в
mark_banned(source=...) — приговор паре «узел×источник», которую фильтрует
acquire(source). Здоровье узла не трогаем; транспортный сбой (таймаут, плоская
500) как и раньше идёт в mark_health(ok=False). report_ban дедуплицирован по
lease: одно событие доезжало до него трижды (POST, ретрай fetch(), провайдер), а
каждый вызов растит ban_count и кратно удлиняет отдых пары.

Refs #3288

tradein-mvp/backend/tests/test_3288_avito_ban_per_source.py

@@ -0,0 +1,262 @@
+"""#3288: бан площадки снимает узел АДРЕСНО (узел×источник), а не глобально по здоровью.
+
+Прод-замер 31.08–01.09: `resolve_proxy_url` отказывал ВСЕМ четырём источникам
+(`ProxyPoolExhaustedError pool_total=7 banned_for_source=0 unhealthy_or_disabled=7`),
+при том что живая проба тех же узлов проходила — они стояли ровно на потолке
+`MAX_CONSECUTIVE_FAILS`. Механизм: сайдкар на бан-странице Авито («доступ ограничен:
+проблема с IP») отвечает HTTP 500 с `ban_page: true`, клиент поднимает
+`SidecarBanPageError` — но она подкласс `httpx.HTTPStatusError`, и общий
+`except Exception` в `_post_fetch` звал `_report_fetch_result(False)` →
+`mark_health(ok=False)`. `mark_health` — решение ГЛОБАЛЬНОЕ для пула: три бан-страницы
+Авито выбивали узел и у Яндекса, и у Циана, и у Домклика, которые не возражали вовсе.
+
+Что закрепляем по ЗНАЧЕНИЮ:
+  (а) подтверждённая бан-страница → `AvitoBlockedError`, `ban_kind_of_exception` ==
+      'platform', `mark_banned(lease, source='avito')` вызван;
+  (б) на бан-странице `mark_health` НЕ вызван вовсе, а на настоящем транспортном сбое
+      (ReadTimeout) — вызван с ok=False, как и раньше;
+  (в) `NoProxyAvailableError` (пул пуст, к площадке не ходили) — ни health-fail, ни бан;
+      через обёртку `fetch_detail` он остаётся распознаваемым ПО ТИПУ в цепочке причин
+      (приём `_iter_causes`), а не по подстроке «no proxy available» (#3272);
+  (г) порядок веток `except`: ban-ветка стоит ДО общего `except Exception`. Пин по
+      значению — тот же HTTP 500 с маркером и без него разводит судьбу узла.
+
+Зеркалит стиль tests/test_3196_domclick_ban_kind.py и
+tests/test_3283_avito_sidecar_ban_is_platform_ban.py; харнесс пула (фейковый провайдер,
+подмена `_client` после `__aenter__`) — из tests/test_kit_browser_fetcher_proxy_pool.py.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.avito_exceptions import AvitoBlockedError, AvitoSidecarUnavailableError
+from scraper_kit.browser_fetcher import BrowserFetcher, SidecarBanPageError
+from scraper_kit.contracts import ProxyLease
+from scraper_kit.orchestration.pipeline import ban_kind_of_exception
+from scraper_kit.providers.avito.detail import fetch_detail
+from scraper_kit.proxy_errors import NoProxyAvailableError
+
+from app.services.scrape_runs import BAN_KIND_INFRA, BAN_KIND_PLATFORM
+
+_ITEM_URL = "https://www.avito.ru/ekaterinburg/kvartiry/x_1234567890"
+_LEASE = ProxyLease(id=13, url="http://node13:8080", kind="http")
+
+
+class _FakePool:
+    """ProxyProvider-заглушка: пишет, ЧТО именно узнал пул об исходе (зеркало
+    _FakeProxyProvider из test_kit_browser_fetcher_proxy_pool.py)."""
+
+    def __init__(self, lease: ProxyLease | None = _LEASE) -> None:
+        self._lease = lease
+        self.health: list[tuple[int, bool]] = []
+        self.banned: list[tuple[int, str]] = []
+        self.touched: list[int] = []
+        self.released: list[int] = []
+
+    def acquire(self, provider: str) -> ProxyLease | None:
+        return self._lease
+
+    def release(self, lease: ProxyLease) -> None:
+        self.released.append(lease.id)
+
+    def mark_health(self, lease: ProxyLease, ok: bool, **_: Any) -> None:
+        self.health.append((lease.id, ok))
+
+    def touch(self, lease: ProxyLease) -> None:
+        self.touched.append(lease.id)
+
+    def mark_banned(self, lease: ProxyLease, *, source: str) -> None:
+        self.banned.append((lease.id, source))
+
+
+def _ban_page_response() -> httpx.Response:
+    """РОВНО то, что отдаёт сайдкар на распознанной бан-странице (browser/server.py:1356):
+    HTTP 500 + `ban_page: true` + статус целевой навигации. Ответ настоящий, а не мок, —
+    чтобы проба шла тем же трактом, что и прод (_raise_for_sidecar_status разбирает тело)."""
+    request = httpx.Request("POST", "http://browser:3000/fetch")
+    return httpx.Response(
+        500,
+        json={
+            "error": "BanPageDetectedError: бан-страница (проблема с IP)",
+            "ban_page": True,
+            "status": 403,
+        },
+        request=request,
+    )
+
+
+def _plain_500_response() -> httpx.Response:
+    """Настоящая внутренняя ошибка сайдкара: тот же код 500, но БЕЗ маркера бана."""
+    request = httpx.Request("POST", "http://browser:3000/fetch")
+    return httpx.Response(
+        500,
+        json={"error": "Error: Page.goto: NS_ERROR_PROXY_BAD_GATEWAY"},
+        request=request,
+    )
+
+
+async def _fetcher(post_result: Any, pool: _FakePool) -> BrowserFetcher:
+    """Фетчер с живым lease из пула и замоканным httpx-клиентом.
+
+    post_result — готовый httpx.Response (вернётся) либо Exception (будет брошен).
+    """
+    bf = BrowserFetcher(
+        endpoint="http://browser:3000",
+        source="avito",
+        proxy_provider=pool,
+        use_pool=True,
+        environment="production",
+    )
+    await bf.__aenter__()
+    client = MagicMock()
+    if isinstance(post_result, Exception):
+        client.post = AsyncMock(side_effect=post_result)
+    else:
+        client.post = AsyncMock(return_value=post_result)
+    client.aclose = AsyncMock(return_value=None)
+    bf._client = client
+    return bf
+
+
+def _iter_causes(exc: BaseException) -> list[BaseException]:
+    """Цепочка причин без зацикливания — копия приёма из domclick_detail_backfill.py."""
+    seen: set[int] = set()
+    out: list[BaseException] = []
+    cur: BaseException | None = exc
+    while cur is not None and id(cur) not in seen:
+        out.append(cur)
+        seen.add(id(cur))
+        cur = cur.__cause__ or cur.__context__
+    return out
+
+
+# ── (а) бан-страница → отказ ПЛОЩАДКИ + адресный бан пары «узел×источник» ──────
+
+
+async def test_ban_page_raises_platform_block_and_bans_node_for_avito_only() -> None:
+    """Бан-страница: AvitoBlockedError, диагноз 'platform', mark_banned(13, 'avito').
+
+    Фальсификация: до фикса SidecarBanPageError доезжала до общего except в
+    _post_fetch → mark_health(ok=False), а строки в scrape_proxy_source_bans не
+    появлялось вовсе (за 7 суток 0 записей для source=avito при 58 прогонах banned).
+    """
+    pool = _FakePool()
+    bf = await _fetcher(_ban_page_response(), pool)
+
+    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
+        with pytest.raises(AvitoBlockedError) as ei:
+            await fetch_detail(_ITEM_URL, browser_fetcher=bf)
+
+    assert ban_kind_of_exception(ei.value) == BAN_KIND_PLATFORM
+    assert pool.banned == [(13, "avito")], (
+        "узел должен быть забанен РОВНО один раз и РОВНО для avito: одно событие "
+        "доезжает до report_ban трижды (POST, ретрай fetch(), провайдер), а каждый "
+        "mark_banned растит ban_count и удлиняет отдых пары"
+    )
+
+
+# ── (б) здоровье узла: бан-страница молчит, транспортный сбой — говорит ────────
+
+
+async def test_ban_page_does_not_mark_health_failure() -> None:
+    """Ключевое значение issue: на бан-странице mark_health НЕ вызывается вовсе.
+
+    Именно этот вызов гнал узел к MAX_CONSECUTIVE_FAILS и выбивал его из выдачи
+    ВСЕМ источникам (замер 01.09: узлы 9, 13, 14 на потолке при banned_for_source=0).
+    """
+    pool = _FakePool()
+    bf = await _fetcher(_ban_page_response(), pool)
+
+    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
+        with pytest.raises(AvitoBlockedError):
+            await fetch_detail(_ITEM_URL, browser_fetcher=bf)
+
+    assert pool.health == [], f"бан площадки не должен трогать здоровье узла: {pool.health}"
+    assert pool.touched, "heartbeat (touch) при этом обязан идти — аренда живая"
+
+
+async def test_real_transport_failure_still_marks_health_failure() -> None:
+    """Контроль к предыдущему: настоящий транспортный сбой (таймаут) health-fail'ит,
+    как и до фикса. Без этой пары фикс неотличим от «выключили учёт здоровья»."""
+    pool = _FakePool()
+    request = httpx.Request("POST", "http://browser:3000/fetch")
+    bf = await _fetcher(httpx.ReadTimeout("sidecar timeout", request=request), pool)
+
+    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
+        with pytest.raises(AvitoSidecarUnavailableError):
+            await fetch_detail(_ITEM_URL, browser_fetcher=bf)
+
+    assert pool.health == [(13, False), (13, False)], (
+        f"таймаут — сбой НАШЕГО тракта, он обязан остаться в здоровье узла "
+        f"(две записи: fetch() делает один ретрай): {pool.health}"
+    )
+    assert pool.banned == [], "таймаут не бан: площадка себя не показала"
+
+
+# ── (в) пустой пул — ни бан, ни health-fail ───────────────────────────────────
+
+
+async def test_empty_pool_is_neither_ban_nor_health_failure() -> None:
+    """`NoProxyAvailableError` поднимается ДО POST'а: узла нет, площадки не видели."""
+    pool = _FakePool(lease=None)
+    bf = BrowserFetcher(
+        endpoint="http://browser:3000",
+        source="avito",
+        proxy_provider=pool,
+        use_pool=True,
+        environment="production",
+    )
+
+    with pytest.raises(NoProxyAvailableError):
+        await bf.__aenter__()
+
+    assert pool.health == []
+    assert pool.banned == []
+
+
+async def test_empty_pool_through_wrapper_stays_infra_and_recognizable_by_type() -> None:
+    """Обёртка `fetch_detail` даёт AvitoSidecarUnavailableError ('infra'), но исходный
+    тип остаётся в цепочке причин — часть B (стоп «нечем ходить») ловит его через
+    `_iter_causes`, а НЕ подстрокой «no proxy available» (на ней обжёгся #3272)."""
+    bf = MagicMock()
+    bf.fetch = AsyncMock(side_effect=NoProxyAvailableError("avito"))
+
+    with pytest.raises(AvitoSidecarUnavailableError) as ei:
+        await fetch_detail(_ITEM_URL, browser_fetcher=bf)
+
+    assert ban_kind_of_exception(ei.value) == BAN_KIND_INFRA
+    assert any(isinstance(c, NoProxyAvailableError) for c in _iter_causes(ei.value))
+    bf.report_ban.assert_not_called()
+
+
+# ── (г) порядок веток except ──────────────────────────────────────────────────
+
+
+def test_sidecar_ban_page_is_a_subclass_of_httpx_error() -> None:
+    """Причина, по которой порядок веток вообще имеет значение (#3287): широкая ловля
+    забирает бан себе. Если это перестанет быть правдой — тесты выше проверяют не то,
+    что думают."""
+    assert issubclass(SidecarBanPageError, httpx.HTTPStatusError)
+
+
+async def test_same_500_different_marker_gives_different_node_fate() -> None:
+    """Пин порядка веток по значению: ОДИН и тот же HTTP 500 разводит судьбу узла
+    ровно по маркеру `ban_page` в теле. Плоская 500 — health-fail без бана; поставь
+    ban-ветку после `except Exception` — и бан-страница даст тот же исход, что здесь."""
+    pool = _FakePool()
+    bf = await _fetcher(_plain_500_response(), pool)
+
+    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
+        with pytest.raises(AvitoSidecarUnavailableError):
+            await fetch_detail(_ITEM_URL, browser_fetcher=bf)
+
+    assert pool.health == [(13, False), (13, False)]
+    assert pool.banned == []

tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py

@@ -420,6 +420,13 @@ class BrowserFetcher:
         self._environment = environment
         self._lease: ProxyLease | None = None
         self._lease_fail_streak: int = 0
+        # id lease'а, по которому бан этому источнику УЖЕ отрапортован (#3288). Одно
+        # событие бана доезжает до report_ban несколько раз: _post_fetch на самой
+        # SidecarBanPageError, ретрай fetch() вторым POST'ом и провайдер сверху
+        # (providers/avito/detail.py). mark_banned на каждый вызов растит ban_count,
+        # то есть ОДИН факт удлинял бы отдых пары «узел×источник» кратно (docstring
+        # proxy_pool.mark_banned: 6ч→12ч). Сбрасывается при взятии нового lease.
+        self._banned_lease_id: int | None = None
         # HTTP-статус ПОСЛЕДНЕЙ целевой навигации сайдкара (#3196). Публичный
         # атрибут, а не возврат fetch(): поток управления менять нельзя — fetch()
         # по-прежнему отдаёт str и по-прежнему не бросает там, где не бросал.
@@ -667,6 +674,8 @@ class BrowserFetcher:
             )
             raise NoProxyAvailableError(self._source)
 
+        # Новый lease — новый узел, признак «бан уже отрапортован» к нему не относится.
+        self._banned_lease_id = None
         return lease
 
     def _release_lease(self) -> None:
@@ -722,6 +731,18 @@ class BrowserFetcher:
         if self._lease is None or self._proxy_provider is None:
             return
         lease = self._lease
+        if self._banned_lease_id == lease.id:
+            # Тот же lease уже помечен забаненным для этого источника (#3288) —
+            # второй рапорт того же факта только эскалирует ban_count. Новый факт
+            # возможен лишь на НОВОМ lease, а его взятие сбрасывает признак.
+            logger.debug(
+                "BrowserFetcher: lease id=%d (%s) уже помечен баном — пропускаем: %s",
+                lease.id,
+                self._source,
+                reason,
+            )
+            return
+        self._banned_lease_id = lease.id
         logger.warning(
             "BrowserFetcher: lease id=%d (%s) BANNED — reporting to pool: %s",
             lease.id,
@@ -752,16 +773,22 @@ class BrowserFetcher:
         """
         return self._lease.id if self._lease is not None else None
 
-    def _report_fetch_result(self, ok: bool) -> None:
+    def _report_fetch_result(self, ok: bool, *, health: bool = True) -> None:
         """Учесть исход ОДНОГО /fetch в здоровье текущего session-lease.
 
         Вызывать на каждый /fetch (успешный и неуспешный) — best-effort, не бросает:
           - `touch()` heartbeat всегда (см. proxy_pool.touch — продлевает leased_at,
             чтобы reap_stale_leases не отобрал прокси у многочасовой сессии);
-          - `mark_health(ok)` всегда — та же грануляция «на каждый /fetch», что была
+          - `mark_health(ok)` — та же грануляция «на каждый /fetch», что была
             до фикса (mark_health решает про DISABLE_THRESHOLD битого узла глобально
             для пула, это НЕ session-locale решение и не должно огрубляться до
-            «одна оценка на всю сессию»);
+            «одна оценка на всю сессию»). `health=False` (#3288) выключает ТОЛЬКО этот
+            шаг — так отказ ПЛОЩАДКИ (подтверждённая бан-страница) не копит глобальный
+            счётчик здоровья узла: он исправен, его отбил конкретный источник, и его
+            судьбу решает `mark_banned(source=...)` по паре «узел×источник»
+            (см. `_report_platform_ban`). До #3288 узел с тремя бан-страницами Авито
+            уходил из выдачи ВСЕМ источникам — прод-замер 31.08: yandex/cian/domclick
+            получали ProxyPoolExhaustedError при banned_for_source=0 и трёх живых узлах;
           - ok=False копит `_lease_fail_streak`; после `_LEASE_ROTATE_AFTER_FAILS`
             подряд lease считается плохим (бан/сетевая труха) — ОСОЗНАННО меняется
             один раз (release старого + acquire нового), счётчик обнуляется. Следующий
@@ -779,12 +806,15 @@ class BrowserFetcher:
             logger.warning(
                 "BrowserFetcher: proxy_pool touch failed for %s", self._source, exc_info=True
             )
-        try:
-            self._proxy_provider.mark_health(lease, ok)
-        except Exception:
-            logger.warning(
-                "BrowserFetcher: proxy_pool mark_health failed for %s", self._source, exc_info=True
-            )
+        if health:
+            try:
+                self._proxy_provider.mark_health(lease, ok)
+            except Exception:
+                logger.warning(
+                    "BrowserFetcher: proxy_pool mark_health failed for %s",
+                    self._source,
+                    exc_info=True,
+                )
 
         if ok:
             self._lease_fail_streak = 0
@@ -816,6 +846,24 @@ class BrowserFetcher:
             )
         self._lease = self._acquire_lease()
 
+    def _report_platform_ban(self, reason: str) -> None:
+        """Исход /fetch, который сайдкар опознал как бан-страницу площадки (#3288).
+
+        Отличается от обычного провала РОВНО одним: узел не получает `mark_health(False)`.
+        Бан — приговор паре «узел×источник» (`mark_banned`, строка в
+        `scrape_proxy_source_bans`, которую фильтрует `acquire(source)`), а не оценка
+        исправности узла: прод-замер 31.08 показал три узла на потолке
+        MAX_CONSECUTIVE_FAILS при banned_for_source=0 — Авито выбивало их из выдачи и
+        Яндексу, и Циану, и Домклику, хотя живая проба тех же узлов проходила.
+
+        Порядок вызовов существенен: `report_ban` работает по ТЕКУЩЕМУ lease, а
+        `_report_fetch_result` может его сменить (ротация после N подряд провалов).
+        Fail-streak копим по-прежнему — сменить сожжённый площадкой адрес полезно, и
+        новый lease забаненный узел уже не вернёт (`acquire` фильтрует бан по source.)
+        """
+        self.report_ban(reason)
+        self._report_fetch_result(False, health=False)
+
     async def _post_fetch(
         self,
         url: str,
@@ -878,7 +926,20 @@ class BrowserFetcher:
             _raise_for_sidecar_status(resp)  # #2698: причина отказа из тела, не только код
             data: dict = resp.json()
             html = data["html"]
+        except SidecarBanPageError as exc:
+            # ПОРЯДОК ВЕТОК ВАЖЕН (#3288): SidecarBanPageError — подкласс
+            # httpx.HTTPStatusError, и общий `except Exception` ниже забирал её себе,
+            # отправляя подтверждённый отказ ПЛОЩАДКИ в глобальный счётчик здоровья узла.
+            self.last_response_status = None
+            self._report_platform_ban(
+                f"sidecar ban page (upstream={exc.upstream_status}) for {url}"
+            )
+            raise
         except Exception:
+            # NoProxyAvailableError (пустой пул) сюда НЕ приходит: он поднимается в
+            # _acquire_lease ДО этого POST'а — узла, о здоровье которого можно судить,
+            # в этот момент просто нет. Отдельной ветки на него здесь поэтому не
+            # заводим: она была бы защитой, которая не может сработать (#3288 п.4).
             self.last_response_status = None
             self._report_fetch_result(False)
             raise
@@ -926,6 +987,13 @@ class BrowserFetcher:
             resp = await self._client.post(f"{self._endpoint}/fetch-json", json=payload)
             _raise_for_sidecar_status(resp)  # #2698: причина отказа из тела, не только код
             data: dict = resp.json()
+        except SidecarBanPageError as exc:
+            # Ветка ДО общего except по той же причине, что в _post_fetch (#3288):
+            # бан-страница — отказ площадки, а не отказ узла.
+            self._report_platform_ban(
+                f"sidecar ban page on fetch-json (upstream={exc.upstream_status}) for {url}"
+            )
+            raise
         except Exception:
             self._report_fetch_result(False)
             raise