fix(tradein/avito): серия отказов нашего тракта обрывает detail-добор статусом failed (#3439)

Прод: 7193, 7218, 7264, 7302 (15-16.09) и 6608/6631/6643/6662 (10.09) — по
1600 попыток, 1-10 карточек, ~1599 блоков, status='done', по 70+ минут
каждый. В те же окна в логах tradein-browser шторм «browser launch failed …
InvalidIP» и «/fetch 503». 6885 (13.09) — те же 1600 попыток при
ban_kinds {"infra": 1591}, но 'banned'.

Корень — регресс #3288/#3367: record_block(kind != platform) уходит в
record_failure(), то есть только в знаменатель окна. Числитель доли растёт
лишь на platform, safety-net смотрит на platform-серию, счётчика серии
infra нет вовсе. Серия отказов сайдкара любой длины не рвала прогон ничем,
а финализатор #3367 при одной карточке ставил 'done'.

- BlockRatioBreaker: infra_streak_limit и счётчик серии infra. Серию гасит
  любой ответ площадки — успех, её блок, 404. abort_reason() -> "infra_streak"
  (проверяется первым), abort_explanation() называет эту величину.
- avito_detail_backfill: порог = max_consecutive_failures (дефолт 25) — тот же
  разряд «отказ нашей стороны», что и отказы-не-блоки. На обрыве — mark_failed
  «тракт недоступен: N отказов … подряд без единого ответа площадки», в
  counters abort_reason=infra_streak и ban_kinds. Не 'banned' — площадка
  молчала; не через mark_backfill_finished — там одна карточка даёт 'done'.

Тесты по значению: 1 успех + 1599 отказов сайдкара -> attempted=26,
mark_failed, abort_reason=infra_streak, ban_kinds {"infra": 25}; обрыв ровно
на 25-м (24 — нет); 24 infra + ответ площадки (успех/блок/404) + 24 — не
обрыв; 24 infra + ответ, трижды — снапшот пройден; 25% настоящих блоков —
не обрыв; прежний контроль #3367 (20 infra + 10 успехов -> done) зелёный.

Фальсификация: без проводки порога — «assert (1600, 1600) == (26, 26)»; без
сброса серии на блоке площадки — «abort_reason='infra_streak': 24 + ответ
площадки (platform) + 24»; без ветки финала — «прогон ушёл в
backfill-honest-status: там 1 карточка превращает шторм в 'done'».

Пункт 3 задачи (browser_unfit узлу, на котором сайдкар не поднял браузер)
не сделан: это вердикт пула прокси и протокол сайдкара, отдельная задача.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tradein-mvp/backend/app/services/backfill_block_breaker.py

@@ -67,8 +67,12 @@ class BlockRatioBreaker:
     ratio_threshold: float
     safety_min: int
     snapshot_size: int
+    # #3439: сколько отказов НАШЕЙ стороны (kind != platform) подряд, без единого ответа
+    # площадки, рвут прогон. 0 — критерий выключен.
+    infra_streak_limit: int = 0
     _window: deque[bool] = field(init=False, repr=False)
     _consecutive_blocks: int = field(default=0, init=False)
+    _consecutive_infra: int = field(default=0, init=False)
     _pure_block_run: bool = field(default=True, init=False)
     streak_histogram: Counter[int] = field(default_factory=Counter, init=False)
 
@@ -106,10 +110,16 @@ class BlockRatioBreaker:
         короче окна рвал бы прогон по тем же infra-отказам, только другим
         критерием. Дефолт 'platform' сохраняет поведение вызывающих, которые вид
         не считают.
+
+        Свой предохранитель у такого отказа — серия infra_streak_limit (#3439): без
+        него серия любой длины не рвала прогон ничем (7302: 1598 отказов сайдкара из
+        1600, status='done'). Серию гасит любой ответ площадки — успех, её блок, 404.
         """
         if kind != BAN_KIND_PLATFORM:
+            self._consecutive_infra += 1
             self.record_failure()
             return
+        self._consecutive_infra = 0
         self._consecutive_blocks += 1
         self._window.append(True)
 
@@ -117,6 +127,7 @@ class BlockRatioBreaker:
         """Единственный исход, снимающий safety-net (#3184: пачка блоков ПОСЛЕ хотя
         бы одного успеха -- уже не "чистый с рождения прогона" burst)."""
         self._flush_streak()
+        self._consecutive_infra = 0
         self._pure_block_run = False
         self._window.append(False)
 
@@ -131,8 +142,11 @@ class BlockRatioBreaker:
         AvitoListingGoneError). Класс написан источник-агностично на будущее (не
         только avito), но сейчас единственный вызывающий -- avito_detail_backfill
         (#3184 review MAJOR 1: применение к domclick_detail_backfill снято из этой
-        задачи -- своя калибровка, свои ограничения прокси-пула)."""
-        return
+        задачи -- своя калибровка, свои ограничения прокси-пула).
+
+        Единственное, что 404 двигает, -- серия infra (#3439): площадка ответила,
+        значит наш тракт жив."""
+        self._consecutive_infra = 0
 
     def _flush_streak(self) -> None:
         if self._consecutive_blocks:
@@ -142,9 +156,11 @@ class BlockRatioBreaker:
     def abort_reason(self) -> str | None:
         """Какой критерий требует обрыва прямо сейчас, или None.
 
-        Возвращает "safety_net" / "ratio" / None. Состояние не меняет, поэтому
-        вызывать можно сколько угодно раз -- в том числе повторно, ради текста лога.
+        Возвращает "infra_streak" / "safety_net" / "ratio" / None. Состояние не меняет,
+        поэтому вызывать можно сколько угодно раз -- в том числе повторно, ради текста лога.
         """
+        if self.infra_streak_limit > 0 and self._consecutive_infra >= self.infra_streak_limit:
+            return "infra_streak"
         # Safety-net -- ТОЛЬКО когда ratio-критерий физически недостижим (снапшот
         # короче окна), иначе пачка safety_min блоков в начале длинного прогона
         # абортила бы его так же, как до правки (#3184 review MAJOR 2).
@@ -176,6 +192,11 @@ class BlockRatioBreaker:
         should_abort(), так что в логи не попадает.
         """
         reason = self.abort_reason()
+        if reason == "infra_streak":
+            return (
+                f"{self._consecutive_infra} отказов нашего тракта (сайдкар/пул) подряд "
+                "без единого ответа площадки"
+            )
         if reason == "ratio":
             return (
                 f"доля блоков {self.window_blocks}/{self.window_len} в окне "

tradein-mvp/backend/app/tasks/avito_detail_backfill.py

@@ -42,6 +42,12 @@ counters["block_streak_histogram"], иначе эффект правки неч
 своей величиной: доля печатает "14/20", safety-net -- длину серии. Раньше лог
 печатал серию всегда, и прогон 5210 (обрыв по доле 14/20) отчитался как
 "ABORT -- 1 consecutive blocks".
+
+Третий критерий, "infra_streak" (#3439): max_consecutive_failures отказов НАШЕГО
+тракта (сайдкар/пул, ban_kind='infra') подряд без единого ответа площадки. Такой
+отказ в долю не входит (#3288), и до этого серия любой длины не рвала прогон ничем:
+7193/7264/7302 — по 1600 попыток, 1-2 карточки, status='done'. Прогон завершается
+'failed' с причиной «тракт недоступен», а не 'banned': площадка молчала.
 """
 
 from __future__ import annotations
@@ -327,7 +333,8 @@ async def run_avito_detail_backfill(
             скользящем окне, см. settings.detail_backfill_block_ratio_window/
             _threshold (module docstring).
         max_consecutive_failures: int -- порог обрыва по отказам-не-блокам,
-            default 25 (см. комментарий у чтения параметра ниже).
+            default 25 (см. комментарий у чтения параметра ниже). Тот же порог рвёт
+            серию отказов НАШЕГО тракта без ответа площадки (infra_streak, #3439).
 
     Lifecycle: update_heartbeat -> snapshot -> loop with budget guard ->
     mark_backfill_finished (done / banned при блоках / failed при нуле, #2674);
@@ -547,6 +554,10 @@ async def run_avito_detail_backfill(
             window_size=int(settings.detail_backfill_block_ratio_window),
             ratio_threshold=float(settings.detail_backfill_block_ratio_threshold),
             safety_min=max_consecutive_blocks,
+            # #3439: отказ сайдкара/пула — отказ нашей стороны, тот же разряд, что
+            # отказы-не-блоки, и тот же порог. Без него серия infra не рвала прогон
+            # ничем: 7302 — 1598 отказов сайдкара из 1600 попыток, 72 минуты, 'done'.
+            infra_streak_limit=max_consecutive_failures,
             # snapshot_size гейтит safety-net (#3184 review MAJOR 2): пачка блоков в
             # начале ДЛИННОГО прогона не должна абортить его так же, как раньше --
             # safety-net включён только когда снапшот короче окна и ratio-критерий
@@ -1040,21 +1051,28 @@ async def run_avito_detail_backfill(
         # и без этого ключа "banned" опять не отличить по причине (#3178).
         if abort_reason is not None:
             current_counters["abort_reason"] = abort_reason  # type: ignore[assignment]
+        # Остановки по вине НАШЕЙ стороны — НЕ блок, поэтому и не aborted_by_blocks:
+        # иначе прогон уйдёт в 'banned' (или в 'done' по #3367) и запись будет
+        # утверждать про площадку то, чего не было.
+        stop_reason: str | None = None
         if no_proxy_stop:
-            # #3288 (как #3283 у домклика): остановка из-за пустого пула — НЕ блок,
-            # поэтому и не aborted_by_blocks: иначе прогон уйдёт в 'banned' и запись
-            # будет утверждать про площадку то, чего не было. Это отказ нашей стороны.
+            # #3288 (как #3283 у домклика): пустой пул — к площадке не ходили вовсе.
             current_counters["no_proxy_stop"] = 1
-            runs_mod.mark_failed(
-                db,
-                run_id,
-                "пул прокси пуст — к площадке не ходили (#3288)",
-                current_counters,
+            stop_reason = "пул прокси пуст — к площадке не ходили (#3288)"
+        elif abort_reason == "infra_streak":
+            # #3439: сайдкар/пул отказывали подряд — площадка всё это время молчала.
+            stop_reason = (
+                f"тракт недоступен: {breaker.abort_explanation()}, обогащено "
+                f"{counters.enriched} из {counters.attempted}; причина: "
+                f"{_top_failure(failure_census)} (#3439)"
             )
+        if stop_reason is not None:
+            runs_mod.mark_failed(db, run_id, stop_reason, current_counters)
             logger.info(
-                "avito_detail_backfill: run_id=%d FINISHED (пул пуст) -- attempted=%d "
+                "avito_detail_backfill: run_id=%d FINISHED (%s) -- attempted=%d "
                 "enriched=%d blocked=%d gone=%d failed=%d duration=%.1fs",
                 run_id,
+                stop_reason,
                 counters.attempted,
                 counters.enriched,
                 counters.blocked,

tradein-mvp/backend/tests/test_3288b_block_kind_ratio.py

@@ -390,3 +390,113 @@ def test_platform_dominant_run_is_still_banned() -> None:
     assert banned.call_args.kwargs["ban_kind"] == BAN_KIND_PLATFORM, (
         f"ban_kind={banned.call_args.kwargs['ban_kind']!r}, ожидали 'platform'"
     )
+
+
+# ── #3439: серия отказов НАШЕГО тракта рвёт прогон своим критерием ─────────────
+#
+# Прогоны 7193/7264/7302: по 1600 попыток, 1-2 карточки, ~1599 отказов сайдкара
+# (InvalidIP-шторм на узле), status='done'. Infra ушёл из доли (#3288), а своего
+# предохранителя не получил. Порог — max_consecutive_failures (дефолт 25): тот же
+# разряд «отказ нашей стороны», что и отказы-не-блоки.
+
+_INFRA_LIMIT = 25
+
+
+def _infra_breaker() -> BlockRatioBreaker:
+    return BlockRatioBreaker(
+        window_size=_WINDOW,
+        ratio_threshold=_THRESHOLD,
+        safety_min=5,
+        snapshot_size=1600,
+        infra_streak_limit=_INFRA_LIMIT,
+    )
+
+
+def test_infra_streak_aborts_exactly_at_limit() -> None:
+    breaker = _infra_breaker()
+    breaker.record_success()
+    for _ in range(_INFRA_LIMIT - 1):
+        breaker.record_block(BAN_KIND_INFRA)
+    assert breaker.abort_reason() is None, "24 отказа подряд — ещё не порог 25"
+
+    breaker.record_block(BAN_KIND_INFRA)
+    assert breaker.abort_reason() == "infra_streak", (
+        f"abort_reason={breaker.abort_reason()!r} на 25-м отказе сайдкара подряд"
+    )
+
+
+@pytest.mark.parametrize("answer", ["success", "platform", "gone"])
+def test_any_platform_answer_resets_infra_streak(answer: str) -> None:
+    """Площадка ответила (успех / её блок / 404) — тракт жив, серия с нуля."""
+    breaker = _infra_breaker()
+    for _ in range(_INFRA_LIMIT - 1):
+        breaker.record_block(BAN_KIND_INFRA)
+    {
+        "success": breaker.record_success,
+        "platform": lambda: breaker.record_block(BAN_KIND_PLATFORM),
+        "gone": breaker.record_neutral,
+    }[answer]()
+    for _ in range(_INFRA_LIMIT - 1):
+        breaker.record_block(BAN_KIND_INFRA)
+
+    assert breaker.abort_reason() is None, (
+        f"abort_reason={breaker.abort_reason()!r}: 24 + ответ площадки ({answer}) + 24 — "
+        "серии в 25 подряд не было"
+    )
+
+
+@pytest.mark.asyncio
+async def test_sidecar_storm_stops_run_as_failed_not_done() -> None:
+    """Шторм сайдкара: 1 успех, дальше только отказы — обрыв на 25-м, 'failed'."""
+    fetches: list[Any] = [MagicMock()]
+    fetches += [AvitoSidecarUnavailableError("browser unavailable 503") for _ in range(1599)]
+
+    counters, runs = await _run_avito(fetches, snapshot_size=1600)
+
+    _assert_identity(counters, expected_attempted=1 + _INFRA_LIMIT)
+    assert runs.mark_backfill_finished.call_args is None, (
+        "прогон ушёл в backfill-honest-status: там 1 карточка превращает шторм в 'done'"
+    )
+    failed = runs.mark_failed.call_args
+    assert failed is not None, "прогон не финализирован mark_failed"
+    written = failed.args[3]
+    assert (written["abort_reason"], written["ban_kinds"]) == (
+        "infra_streak",
+        {BAN_KIND_INFRA: _INFRA_LIMIT},
+    ), f"counters={written}"
+    assert "тракт недоступен" in failed.args[2], f"причина={failed.args[2]!r}"
+    assert runs.mark_banned.call_args is None, "площадка не отказывала — не 'banned'"
+
+
+@pytest.mark.asyncio
+async def test_interleaved_platform_answers_do_not_trip_infra_stop() -> None:
+    """Контроль: 24 отказа сайдкара + ответ площадки, трижды — снапшот пройден целиком."""
+    fetches: list[Any] = []
+    for answer in (AvitoBlockedError("firewall"), MagicMock(), MagicMock()):
+        fetches += [AvitoSidecarUnavailableError("503") for _ in range(_INFRA_LIMIT - 1)]
+        fetches.append(answer)
+
+    counters, runs = await _run_avito(fetches, snapshot_size=len(fetches))
+
+    _assert_identity(counters, expected_attempted=len(fetches))
+    finished = runs.mark_backfill_finished.call_args
+    assert finished is not None and finished.args[2].get("abort_reason") is None, (
+        f"прогон оборван: counters={finished and finished.args[2]}, "
+        f"mark_failed={runs.mark_failed.call_args}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_quarter_platform_blocks_run_is_not_aborted() -> None:
+    """Контроль: здоровый прогон с 25% настоящих блоков не рвётся ни одним критерием."""
+    fetches: list[Any] = []
+    for _ in range(10):
+        fetches += [MagicMock(), MagicMock(), MagicMock(), AvitoBlockedError("firewall")]
+
+    counters, runs = await _run_avito(fetches, snapshot_size=40)
+
+    _assert_identity(counters, expected_attempted=40)
+    finished = runs.mark_backfill_finished.call_args
+    assert finished is not None and finished.args[2].get("abort_reason") is None, (
+        f"counters={finished and finished.args[2]}: 10 блоков из 40 — рвать не по чему"
+    )