fix(#3394): стоп по пустому пулу — по цепочке причин на уровне якоря; один errors_count; ban_kind=infra под тестом

tradein-mvp/backend/tests/test_3394_cian_sweep_houses_proxy_pool.py

@@ -14,8 +14,13 @@ False → POST /fetch без "proxy" → сайдкар брал env-узел `S
 
 Второй тест — про исход пустого пула: он поднимается ДО запроса, следующий дом упрётся
 ровно в то же самое, поэтому фаза (и свип) обязаны оборваться на первом доме с
-`no_proxy_stop=1`, а не писать 30 одинаковых отказов и уходить в 'done' (образец —
-#3389, yandex-nb-sweep).
+`no_proxy_stop=1` и `ban_kind='infra'`, а не писать 30 одинаковых отказов и уходить в
+'done' (образец — #3389, yandex-nb-sweep).
+
+Третий — тот же исход, когда провайдер ЗАВЕРНУЛ отказ пула в своё исключение: ловля на
+уровне якоря идёт по цепочке причин (`caused_by_no_proxy`), а не по типу, иначе первая же
+такая обёртка (у `fetch_detail` она уже есть) увела бы стоп в generic-обработчик и
+записала бы нашу же нехватку прокси баном площадки.
 
 Сеть/БД/камуфокс замоканы; в сеть тест не ходит.
 """
@@ -82,11 +87,37 @@ class _EmptyPoolFetcher:
         return None
 
 
+class _WrappingEmptyPoolFetcher:
+    """Пул пуст, но провайдер завернул отказ в своё исключение (как `fetch_detail`)."""
+
+    calls: ClassVar[list[str]] = []
+
+    def __init__(self, **_kwargs: Any) -> None:
+        pass
+
+    async def __aenter__(self) -> _WrappingEmptyPoolFetcher:
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+    async def fetch(self, url: str, **_kwargs: Any) -> str:
+        _WrappingEmptyPoolFetcher.calls.append(url)
+        raise RuntimeError("wrapped") from NoProxyAvailableError("cian")
+
+    def report_ban(self, _reason: str) -> None:
+        return None
+
+
 class _RunsRecorder:
     """Двойник scrape_runs: пишет финализаторы, is_cancelled всегда False."""
 
     def __init__(self) -> None:
         self.calls: list[tuple[str, dict[str, Any]]] = []
+        # ban_kind писался в scrape_runs, но двойник его отбрасывал — поле уходило
+        # из-под теста целиком, а именно оно отличает «наша инфраструктура» ('infra')
+        # от «площадка забанила» ('platform'/'unknown').
+        self.ban_kinds: list[str] = []
 
     def is_cancelled(self, _db: Any, _run_id: int) -> bool:
         return False
@@ -107,6 +138,7 @@ class _RunsRecorder:
         ban_kind: str = "unknown",
     ) -> None:
         self.calls.append(("mark_banned", dict(counters)))
+        self.ban_kinds.append(ban_kind)
 
     def mark_failed(self, _db: Any, _run_id: int, _error: str, counters: dict[str, Any]) -> None:
         self.calls.append(("mark_failed", dict(counters)))
@@ -153,7 +185,7 @@ def _nb_lot() -> MagicMock:
 
 async def _drive(
     *, fetcher: type, house_rows: list[dict[str, Any]]
-) -> tuple[dict[str, int], list[tuple[str, dict[str, Any]]], MagicMock]:
+) -> tuple[dict[str, int], _RunsRecorder, MagicMock]:
     recorder = _RunsRecorder()
     provider = MagicMock(name="proxy_provider")
     scraper = MagicMock()
@@ -182,21 +214,25 @@ async def _drive(
             enrich_houses=True,
             newbuilding_only=True,
         )
-    return counters.to_dict(), recorder.calls, provider
+    return counters.to_dict(), recorder, provider
 
 
 @pytest.mark.asyncio
 async def test_houses_phase_builds_fetcher_with_proxy_provider() -> None:
     """(а) фетчер houses-фазы собран С провайдером пула — тот, что пришёл в свип."""
     _CapturingFetcher.captured.clear()
-    counters, _calls, provider = await _drive(
+    counters, _recorder, provider = await _drive(
         fetcher=_CapturingFetcher,
         house_rows=[{"id": 7, "cian_zhk_url": _ZHK_URL}],
     )
     assert counters["houses_attempted"] == 1
     assert len(_CapturingFetcher.captured) == 1, _CapturingFetcher.captured
     kwargs = _CapturingFetcher.captured[0]
-    # Ровно те три аргумента, без которых сайдкар уходит на env-узел.
+    # Эпохи различает РОВНО первый assert: `use_pool`/`environment`
+    # build_browser_fetcher кладёт из config независимо от провайдера (_base.py:233),
+    # то есть и до правки они были такими же. Оставлены как контракт сайдкара: без
+    # любого из трёх (или при use_pool=False / environment != production) запрос
+    # уходит на env-узел SCRAPER_PROXY_URL.
     assert kwargs["proxy_provider"] is provider
     assert kwargs["use_pool"] is True
     assert kwargs["environment"] == "production"
@@ -206,7 +242,7 @@ async def test_houses_phase_builds_fetcher_with_proxy_provider() -> None:
 async def test_empty_pool_stops_houses_phase_after_first_house() -> None:
     """(б) пустой пул → один дом, no_proxy_stop=1, свип оборван (mark_banned)."""
     _EmptyPoolFetcher.calls.clear()
-    counters, calls, _provider = await _drive(
+    counters, recorder, _provider = await _drive(
         fetcher=_EmptyPoolFetcher,
         house_rows=[
             {"id": 7, "cian_zhk_url": _ZHK_URL},
@@ -218,5 +254,35 @@ async def test_empty_pool_stops_houses_phase_after_first_house() -> None:
     assert counters["houses_attempted"] == 1
     assert counters["houses_failed"] == 1
     assert counters["no_proxy_stop"] == 1
-    assert calls[-1][0] == "mark_banned"
-    assert calls[-1][1]["no_proxy_stop"] == 1
+    assert counters["errors_count"] == 1, "одно событие — один errors_count"
+    assert recorder.calls[-1][0] == "mark_banned"
+    assert recorder.calls[-1][1]["no_proxy_stop"] == 1
+    assert recorder.ban_kinds[-1] == "infra", "наша инфраструктура, не бан площадки"
+
+
+@pytest.mark.asyncio
+async def test_empty_pool_stops_even_when_wrapped_by_provider() -> None:
+    """(в) отказ пула, ЗАВЁРНУТЫЙ провайдером, тоже обрывает свип — по цепочке причин.
+
+    Решение «стоп» принимает raise-site (`caused_by_no_proxy`, тип в цепочке
+    `__cause__`), а ловля стояла по конкретному типу `NoProxyAvailableError`. Пока
+    `fetch_newbuilding` не оборачивает, это совпадало; `fetch_detail` уже оборачивает
+    (см. app/tasks/cian_history_backfill.py:236-238), и с появлением такой обёртки стоп
+    уходил бы в generic-обработчик → consecutive_failures → mark_banned «IP likely
+    blocked», то есть НАШ отказ инфраструктуры записывался бы баном площадки.
+    """
+    _WrappingEmptyPoolFetcher.calls.clear()
+    counters, recorder, _provider = await _drive(
+        fetcher=_WrappingEmptyPoolFetcher,
+        house_rows=[
+            {"id": 7, "cian_zhk_url": _ZHK_URL},
+            {"id": 8, "cian_zhk_url": _ZHK_URL},
+            {"id": 9, "cian_zhk_url": _ZHK_URL},
+        ],
+    )
+    assert _WrappingEmptyPoolFetcher.calls == [_ZHK_URL], "второй дом упрётся в то же самое"
+    assert counters["houses_attempted"] == 1
+    assert counters["no_proxy_stop"] == 1
+    assert counters["errors_count"] == 1
+    assert recorder.calls[-1][0] == "mark_banned"
+    assert recorder.ban_kinds[-1] == "infra"

tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py

@@ -2846,9 +2846,12 @@ class CianCitySweepCounters:
     houses_failed: int = 0
     errors_count: int = 0
     # #3394: 1 — прогон оборван «пул прокси пуст» (к площадке не ходили), 0 — нет.
-    # int, а не bool: счётчики читают SQL'ем `counters->>'no_proxy_stop' = '1'`,
-    # мимо JSON-ного `true` он промахнётся молча (тот же довод — в
-    # app/tasks/yandex_newbuilding_sweep.py::SweepResult.to_dict).
+    # int, а не bool — как у всех соседей (avito_detail_backfill.py:1043,
+    # domclick_detail_backfill.py:610, yandex_newbuilding_sweep.py::SweepResult.to_dict):
+    # весь payload counters числовой, и SQL-разбор вида `counters->>'no_proxy_stop' = '1'`
+    # на нём не промахнётся, тогда как мимо JSON-ного `true` промахнулся бы молча.
+    # Такого запроса в репо пока НЕТ (грепом 06.09 — ни в data/sql, ни в коде) — это
+    # конвенция формата на будущее, а не поддержка существующего потребителя.
     no_proxy_stop: int = 0
 
     def to_dict(self) -> dict[str, int]:
@@ -3127,6 +3130,23 @@ async def run_cian_city_sweep(
                             _cian_detail_consec_failures = 0
                         except Exception as exc:
                             counters.detail_failed += 1
+                            if caused_by_no_proxy(exc):
+                                # #3394: тот же довод, что в houses-фазе ниже — пул пуст,
+                                # запрос не уходил, следующий лот упрётся ровно в то же
+                                # самое. Без этой ветки отказ доедал _cian_detail_abort и
+                                # писал в лог «proxy likely banned/stuck» (диагноз
+                                # площадки на НАШЕЙ инфраструктуре), а перед этим ещё
+                                # дёргал ротацию IP, которой пустому пулу нечего менять.
+                                # errors_count инкрементируется один раз — на уровне
+                                # якоря, где прогон обрывается.
+                                logger.error(
+                                    "cian-sweep run_id=%d: detail СТОП — пул прокси пуст, "
+                                    "к площадке не ходили (listing_id=%d): %s",
+                                    run_id,
+                                    listing_id,
+                                    exc,
+                                )
+                                raise
                             counters.errors_count += 1
                             _cian_detail_consec_failures += 1
                             logger.warning(
@@ -3252,14 +3272,14 @@ async def run_cian_city_sweep(
                             )
                     except Exception as exc:
                         counters.houses_failed += 1
-                        counters.errors_count += 1
                         if caused_by_no_proxy(exc):
                             # #3394 (образец — #3197/#3389): пустой пул это НЕ отказ
                             # площадки — запрос не уходил вовсе, и следующий дом
                             # упрётся ровно в то же самое. Перебирать оставшиеся
                             # значило бы писать 30 одинаковых houses_failed и уходить
-                            # в 'done'. Обрываем фазу и свип (ветка
-                            # `except NoProxyAvailableError` в цикле по anchor'ам).
+                            # в 'done'. Обрываем фазу и свип (ветка caused_by_no_proxy
+                            # в цикле по anchor'ам). errors_count инкрементируется там
+                            # же и только там — иначе одно событие считалось дважды.
                             logger.error(
                                 "cian-sweep run_id=%d: houses СТОП — пул прокси пуст, "
                                 "к площадке не ходили (house_id=%d, дом %d/%d): %s",
@@ -3270,6 +3290,7 @@ async def run_cian_city_sweep(
                                 exc,
                             )
                             raise
+                        counters.errors_count += 1
                         logger.warning(
                             "cian-sweep run_id=%d: houses failed house_id=%d: %s",
                             run_id,
@@ -3331,35 +3352,42 @@ async def run_cian_city_sweep(
                 counters.anchors_done = idx
                 runs.update_heartbeat(db, run_id, counters.to_dict())
                 continue
-            except NoProxyAvailableError as exc:
-                # #3394: пул пуст — свип обрывается на первом же доме/лоте, а не
-                # перебирает остаток города одинаковыми отказами. Ветка стоит ДО
-                # generic-except: там NoProxyAvailableError уходил бы в
-                # consecutive_failures и (через 3 якоря) в mark_banned с диагнозом
-                # «IP likely blocked», то есть наш отказ читался бы как бан площадки.
-                # done_buckets сохраняем — следующий прогон не начнёт с нуля (#2686).
-                counters.no_proxy_stop = 1
-                counters.errors_count += 1
-                logger.error(
-                    "cian-sweep run_id=%d: СТОП на anchor #%d/%d (%s) — пул прокси "
-                    "пуст, к площадке не ходили: %s",
-                    run_id,
-                    idx,
-                    len(_anchors),
-                    name,
-                    exc,
-                )
-                _ckpt = {**counters.to_dict(), "done_buckets": sorted(_done_anchors)}
-                runs.update_heartbeat(db, run_id, _ckpt)
-                runs.mark_banned(
-                    db,
-                    run_id,
-                    f"cian sweep aborted: {exc}",
-                    _ckpt,
-                    ban_kind=ban_kind_of_exception(exc),
-                )
-                return counters
             except Exception as e:
+                if caused_by_no_proxy(e):
+                    # #3394: пул пуст — свип обрывается на первом же доме/лоте, а не
+                    # перебирает остаток города одинаковыми отказами. Ловим по ЦЕПОЧКЕ
+                    # причин и ДО generic-разбора ниже: raise-site внутри фаз решает
+                    # «стоп» тем же caused_by_no_proxy, и стоит провайдеру завернуть
+                    # отказ в своё исключение (так уже делает fetch_detail — см.
+                    # app/tasks/cian_history_backfill.py:236-238), как ветка по ТИПУ
+                    # промахнулась бы: наш отказ инфраструктуры уехал бы в
+                    # consecutive_failures и (через 3 якоря) в mark_banned с диагнозом
+                    # «IP likely blocked», то есть читался бы как бан площадки.
+                    # ban_kind — константой, а не ban_kind_of_exception(e): тот смотрит
+                    # isinstance и на завёрнутом отказе дал бы 'unknown'; здесь диагноз
+                    # установлен самим условием ветки.
+                    # done_buckets сохраняем — следующий прогон не начнёт с нуля (#2686).
+                    counters.no_proxy_stop = 1
+                    counters.errors_count += 1
+                    logger.error(
+                        "cian-sweep run_id=%d: СТОП на anchor #%d/%d (%s) — пул прокси "
+                        "пуст, к площадке не ходили: %s",
+                        run_id,
+                        idx,
+                        len(_anchors),
+                        name,
+                        e,
+                    )
+                    _ckpt = {**counters.to_dict(), "done_buckets": sorted(_done_anchors)}
+                    runs.update_heartbeat(db, run_id, _ckpt)
+                    runs.mark_banned(
+                        db,
+                        run_id,
+                        f"cian sweep aborted: {e}",
+                        _ckpt,
+                        ban_kind=BAN_KIND_INFRA,
+                    )
+                    return counters
                 logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name)
                 counters.errors_count += 1
                 consecutive_failures += 1