diff --git a/tradein-mvp/backend/app/services/backfill_block_breaker.py b/tradein-mvp/backend/app/services/backfill_block_breaker.py index 75352846..9c55100d 100644 --- a/tradein-mvp/backend/app/services/backfill_block_breaker.py +++ b/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} в окне " diff --git a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py index 10041b4f..94053726 100644 --- a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py +++ b/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, diff --git a/tradein-mvp/backend/tests/test_3288b_block_kind_ratio.py b/tradein-mvp/backend/tests/test_3288b_block_kind_ratio.py index 7d3ec9c9..a43dcefd 100644 --- a/tradein-mvp/backend/tests/test_3288b_block_kind_ratio.py +++ b/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 — рвать не по чему" + )