From fb6a8f34b0075ba7688dc359d25c7b7e94bc9d66 Mon Sep 17 00:00:00 2001
From: bot-backend <bot-backend@gendsgn.ru>
Date: Wed, 12 Aug 2026 17:12:25 +0500
Subject: [PATCH] =?UTF-8?q?fix(tradein/scraper):=20=D0=BF=D1=83=D1=81?=
 =?UTF-8?q?=D1=82=D0=BE=D0=B9=20=D0=BF=D1=83=D0=BB=20=D0=BF=D1=80=D0=BE?=
 =?UTF-8?q?=D0=BA=D1=81=D0=B8=20=D0=BF=D0=B5=D1=80=D0=B5=D1=81=D1=82=D0=B0?=
 =?UTF-8?q?=D1=91=D1=82=20=D1=81=D1=82=D0=B8=D1=80=D0=B0=D1=82=D1=8C=20?=
 =?UTF-8?q?=D1=87=D0=B5=D0=BA=D0=BF=D0=BE=D0=B8=D0=BD=D1=82=20=D0=BF=D0=BE?=
 =?UTF-8?q?=D0=BB=D0=BD=D0=BE=D0=B3=D0=BE=20=D0=BE=D0=B1=D1=85=D0=BE=D0=B4?=
 =?UTF-8?q?=D0=B0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

NoProxyAvailableError — наследник RuntimeError, поэтому в run_avito_full_load /
run_cian_full_load / run_yandex_full_load он попадал в общую ветку `except
RuntimeError` → mark_failed. mark_failed, в отличие от mark_banned, НЕ пишет
done_buckets: чекпоинт всех пройденных бакетов терялся ровно на НАШЕМ отказе —
том исходе, для которого #2686 явно требовал его сохранять наравне с блокировкой
площадкой. Плюс ban_kind_of_exception отдавал для него 'unknown', хотя причина
здесь не «не установлена», а названа самим типом (proxy_errors.py: «это НАША
инфраструктура (нет живого прокси), не внешний блок»).

Достижимость на проде (замер 12.08, read-only): под avito доступны три узла
(id 9/10/11; id 1 забанен парой до 13.08). BrowserFetcher меняет lease после
_LEASE_ROTATE_AFTER_FAILS=3 подряд неудачных /fetch, а proxy_pool.acquire не
выдаёт узел с consecutive_fails >= MAX_CONSECUTIVE_FAILS=3 — девять неудачных
POST'ов опустошают пул под источник. Лестница одного полного обхода при
отказывающем сайдкаре делает до 24 POST'ов (4 бакета × 3 попытки × 2 внутренних
httpx-ретрая), то есть проходит через это состояние заведомо. Цена — прогон 3547
(09.08): 35 бакетов, 5496 объявлений за 2ч58м.

Refs #2687, #2686
---
 ...t_2687_pool_exhaustion_keeps_checkpoint.py | 166 ++++++++++++++++++
 .../src/scraper_kit/orchestration/pipeline.py |  53 +++++-
 2 files changed, 218 insertions(+), 1 deletion(-)
 create mode 100644 tradein-mvp/backend/tests/test_2687_pool_exhaustion_keeps_checkpoint.py

diff --git a/tradein-mvp/backend/tests/test_2687_pool_exhaustion_keeps_checkpoint.py b/tradein-mvp/backend/tests/test_2687_pool_exhaustion_keeps_checkpoint.py
new file mode 100644
index 00000000..6dd54899
--- /dev/null
+++ b/tradein-mvp/backend/tests/test_2687_pool_exhaustion_keeps_checkpoint.py
@@ -0,0 +1,166 @@
+"""#2687: опустевший пул прокси mid-run стирал чекпоинт полного обхода.
+
+Найдено при разборе шести `ban_kind='infra'` у `avito_full_load` (29.07-03.08). Сами
+шесть объяснены и починены раньше (#2637 подключил браузерный путь Авито к пулу,
+#2634/#2640 добили запасной путь), но рядом с ними на main живёт соседний дефект того
+же класса, что и #2686: наш собственный отказ финализируется веткой, которая теряет
+чекпоинт.
+
+Механизм. `NoProxyAvailableError` (proxy_errors.py) — наследник `RuntimeError`, и его
+собственный докстринг говорит прямо: «это НАША инфраструктура (нет живого прокси), не
+внешний блок». В `run_*_full_load` он попадал в общую ветку `except RuntimeError` →
+`mark_failed`, а `mark_failed`, в отличие от `mark_banned`, НЕ пишет `done_buckets`
+(pipeline.py, ветка блока пишет его явно). Плюс `ban_kind_of_exception` возвращал для
+него `unknown` — диагноз «наша инфраструктура» терялся дважды: и как метка, и как
+сохранённый прогресс.
+
+Достижимость на проде (замер 2026-08-12, read-only):
+  - под `avito` доступны три узла (id 9/10/11; id 1 забанен парой до 13.08);
+  - `BrowserFetcher` меняет lease после `_LEASE_ROTATE_AFTER_FAILS=3` подряд неудачных
+    `/fetch`, а `proxy_pool.acquire` не выдаёт узел с
+    `consecutive_fails >= MAX_CONSECUTIVE_FAILS=3` → девять неудачных POST'ов
+    опустошают пул под источник;
+  - лестница ОДНОГО полного обхода при отказывающем сайдкаре делает до 24 POST'ов
+    (4 бакета `_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED` × 3 попытки
+    `_AVITO_SIDECAR_TRANSIENT_RETRIES`+1 × 2 внутренних httpx-ретрая `fetch`),
+    то есть заведомо проходит через это состояние.
+Цена потери чекпоинта — прогон 3547 (09.08): 35 бакетов, 5496 объявлений за 2ч58м.
+
+Фальсификация (красный прогон на коде до правки):
+  - `ban_kind_of_exception` → `'unknown'` вместо `'infra'`;
+  - `run_*_full_load` → `mark_failed` вместо `mark_banned`, исключение улетает наружу
+    (тест падает на непойманном `NoProxyAvailableError`), `done_buckets` нигде нет.
+"""
+
+from __future__ import annotations
+
+import os
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.orchestration import runs as kit_runs
+from scraper_kit.orchestration.pipeline import (
+    ban_kind_of_exception,
+    run_avito_full_load,
+    run_cian_full_load,
+    run_yandex_full_load,
+)
+from scraper_kit.proxy_errors import NoProxyAvailableError
+
+PFX = "scraper_kit.orchestration.pipeline"
+
+_FULL_LOAD = {
+    "avito": (run_avito_full_load, f"{PFX}.AvitoScraper"),
+    "cian": (run_cian_full_load, f"{PFX}.CianScraper"),
+    "yandex": (run_yandex_full_load, f"{PFX}.YandexRealtyScraper"),
+}
+
+
+class _Recorder:
+    """Двойник scrape_runs: интересны имя финализатора, ban_kind и counters."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str, dict[str, Any]]] = []
+
+    def is_cancelled(self, db: Any, run_id: int) -> bool:
+        return False
+
+    def update_heartbeat(self, db: Any, run_id: int, counters: dict[str, Any]) -> None:
+        pass
+
+    def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None:
+        self.calls.append(("mark_done", "", dict(counters)))
+
+    def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None:
+        self.calls.append(("mark_failed", "", dict(counters)))
+
+    def mark_banned(
+        self,
+        db: Any,
+        run_id: int,
+        error: str,
+        counters: dict[str, Any],
+        *,
+        ban_kind: str = kit_runs.BAN_KIND_UNKNOWN,
+    ) -> None:
+        self.calls.append(("mark_banned", ban_kind, dict(counters)))
+
+
+def _config() -> SimpleNamespace:
+    return SimpleNamespace(
+        scraper_fetch_mode="curl_cffi",
+        browser_http_endpoint="http://browser.test/fetch",
+        scraper_proxy_url=None,
+        avito_proxy_max_rotations=0,
+        avito_serp_ok_not_banned=True,
+        avito_proxy_rotate_settle_s=0.0,
+        proxy_rotate_attempts=1,
+        proxy_rotate_attempt_timeout_s=1.0,
+        cian_proxy_max_rotations=0,
+        yandex_proxy_max_rotations=0,
+        cian_full_load_per_fetch_timeout_s=0.0,
+        scraper_skip_seen_today=False,
+    )
+
+
+def _scraper_that_saves_one_bucket_then_runs_out_of_proxies() -> MagicMock:
+    """Двойник скрапера: один бакет доехал до on_bucket, затем пул опустел."""
+
+    async def _fetch(*_a: Any, on_bucket: Any = None, **_k: Any) -> None:
+        on_bucket("2к:0-5m", [MagicMock(source_id="a1")])
+        raise NoProxyAvailableError("avito")
+
+    m = MagicMock()
+    m.__aenter__ = AsyncMock(return_value=m)
+    m.__aexit__ = AsyncMock(return_value=None)
+    m.fetch_all_secondary = _fetch
+    m.state_extraction_attempts = 0
+    m.state_extraction_failures = 0
+    m.gate_fetch_attempts = 0
+    m.gate_fetch_failures = 0
+    m._browser = None
+    return m
+
+
+def test_ban_kind_of_no_proxy_is_infra() -> None:
+    """Пустой пул — наша инфраструктура по определению самого исключения.
+
+    Раньше классификатор отдавал 'unknown' (дефолт #2764): для НЕизвестной причины
+    это верно, но здесь причина известна и названа в типе.
+    """
+    assert ban_kind_of_exception(NoProxyAvailableError("avito")) == kit_runs.BAN_KIND_INFRA
+
+
+@pytest.mark.parametrize("source", ["avito", "cian", "yandex"])
+async def test_full_load_pool_exhaustion_keeps_checkpoint(source: str) -> None:
+    """Пул опустел mid-run → 'banned'/'infra' и done_buckets целы, а не mark_failed.
+
+    Требование #2686 («чекпоинт сохраняется и при нашем сбое, и при блокировке
+    площадкой») до этой правки держалось только для отказа сайдкара; отказ пула шёл
+    мимо него и обнулял прогресс следующего прогона.
+    """
+    fn, scraper_target = _FULL_LOAD[source]
+    recorder = _Recorder()
+    extra: dict[str, Any] = {}
+    if source == "yandex":
+        enrichment = MagicMock()
+        enrichment.record_yandex_price_history = MagicMock(return_value=0)
+        extra["enrichment"] = enrichment
+
+    scraper = _scraper_that_saves_one_bucket_then_runs_out_of_proxies()
+    with (
+        patch(scraper_target, return_value=scraper),
+        patch(f"{PFX}.save_listings", MagicMock(return_value=(1, 0))),
+        patch(f"{PFX}.runs", recorder),
+    ):
+        await fn(MagicMock(), run_id=1, config=_config(), matcher=MagicMock(), **extra)
+
+    assert [c[0] for c in recorder.calls] == ["mark_banned"]
+    _method, ban_kind, counters = recorder.calls[0]
+    assert ban_kind == kit_runs.BAN_KIND_INFRA
+    assert counters["done_buckets"] == ["2к:0-5m"]
diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py
index eb6852b3..5c090d75 100644
--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py
@@ -54,6 +54,7 @@ from scraper_kit.avito_exceptions import (
 from scraper_kit.base import ScrapedLot, save_listings
 from scraper_kit.browser_fetcher import BrowserFetcher
 from scraper_kit.orchestration import runs
+
 # Константы диагноза берём напрямую, а не через `runs.` — helper ниже обязан
 # работать и когда тесты подменяют весь модуль runs двойником (#2686).
 from scraper_kit.orchestration.runs import BAN_KIND_INFRA, BAN_KIND_PLATFORM, BAN_KIND_UNKNOWN
@@ -76,6 +77,7 @@ from scraper_kit.providers.yandex.serp import (
     ROOM_PATH,
     YandexRealtyScraper,
 )
+from scraper_kit.proxy_errors import NoProxyAvailableError
 
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
@@ -208,7 +210,11 @@ def ban_kind_of_exception(exc: BaseException) -> str:
     Один узел на все avito-сайты mark_banned: разводить исход в каждом из трёх
     было бы тремя копиями одного условия.
     """
-    if isinstance(exc, AvitoSidecarUnavailableError):
+    if isinstance(exc, AvitoSidecarUnavailableError | NoProxyAvailableError):
+        # NoProxyAvailableError (#2687): пул прокси пуст в проде. Причина здесь не
+        # «не установлена» (тогда был бы 'unknown', #2764), а названа самим типом —
+        # см. proxy_errors.py: «это НАША инфраструктура (нет живого прокси), не
+        # внешний блок». Тот же диагноз, что и у отказа сайдкара.
         return BAN_KIND_INFRA
     if isinstance(exc, AvitoBlockedError | AvitoRateLimitedError):
         return BAN_KIND_PLATFORM
@@ -3217,6 +3223,20 @@ async def run_cian_full_load(
         )
         return counters
 
+    except NoProxyAvailableError as exc:
+        # #2687, см. ту же ветку в run_avito_full_load: наш отказ (пул пуст) обязан
+        # сохранять done_buckets, а generic-RuntimeError → mark_failed его теряет.
+        logger.error("cian-full-load run_id=%d: no proxy available — %s", run_id, exc)
+        counters.errors_count += 1
+        runs.mark_banned(
+            db,
+            run_id,
+            f"cian full load aborted: {exc}",
+            {**counters.to_dict(), "done_buckets": sorted(done)},
+            ban_kind=ban_kind_of_exception(exc),
+        )
+        return counters
+
     except RuntimeError as exc:
         # on_bucket кидает RuntimeError("cancelled") при cooperative cancel,
         # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a).
@@ -3445,6 +3465,20 @@ async def run_yandex_full_load(
         )
         return counters
 
+    except NoProxyAvailableError as exc:
+        # #2687, см. ту же ветку в run_avito_full_load: наш отказ (пул пуст) обязан
+        # сохранять done_buckets, а generic-RuntimeError → mark_failed его теряет.
+        logger.error("yandex-full-load run_id=%d: no proxy available — %s", run_id, exc)
+        counters.errors_count += 1
+        runs.mark_banned(
+            db,
+            run_id,
+            f"yandex full load aborted: {exc}",
+            {**counters.to_dict(), "done_buckets": sorted(done)},
+            ban_kind=ban_kind_of_exception(exc),
+        )
+        return counters
+
     except RuntimeError as exc:
         # on_bucket кидает RuntimeError("cancelled") при cooperative cancel,
         # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a).
@@ -3662,6 +3696,23 @@ async def run_avito_full_load(
         )
         return counters
 
+    except NoProxyAvailableError as exc:
+        # #2687: пул опустел mid-run. Ветка стоит ДО generic-RuntimeError намеренно —
+        # NoProxyAvailableError его подкласс, и без неё отказ уходил в mark_failed,
+        # который (в отличие от mark_banned) НЕ пишет done_buckets. То есть чекпоинт
+        # терялся ровно на НАШЕМ отказе — том исходе, для которого #2686 требовал его
+        # сохранять наравне с блокировкой площадкой.
+        logger.error("avito-full-load run_id=%d: no proxy available — %s", run_id, exc)
+        counters.errors_count += 1
+        runs.mark_banned(
+            db,
+            run_id,
+            f"avito full load aborted: {exc}",
+            {**counters.to_dict(), "done_buckets": sorted(done)},
+            ban_kind=ban_kind_of_exception(exc),
+        )
+        return counters
+
     except RuntimeError as exc:
         # on_bucket кидает RuntimeError("cancelled") при cooperative cancel,
         # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a).