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

"""Критерий обрыва avito_detail_backfill по доле блоков (#3184).

Модуль написан источник-агностично (см. BlockRatioBreaker ниже), но единственный
текущий вызывающий -- avito_detail_backfill. Исходная задача планировала тот же
критерий и для domclick_detail_backfill ("тот же критерий, что у avito"), но это
снято ревью (#3184 review MAJOR 1): у Домклика ОДИН выделенный residential-прокси
БЕЗ ротации (data/sql/175_scrape_schedules_seed_domclick_detail_backfill.sql,
max_consecutive_blocks=3 против avito=5 -- намеренно ниже, ранний обрыв бережёт
репутацию единственного узла), а калибровка окна/порога ниже сделана на avito-пуле
С ротацией. При устойчивой доле блоков около порога (например 0.65) Домклик по
этому критерию не обрывался бы никогда -- десятки запросов по QRATOR с одного
IP, ровно то, что max_consecutive_blocks=3 был призван предотвратить. Домклик
получит свою калибровку отдельной задачей.

Раньше avito_detail_backfill абортил по голой длине "N блоков подряд"
(max_consecutive_blocks). Замер 40 прогонов avito за 14 суток (2026-08-14..28)
показал: ВСЕ 40 закончились статусом 'banned', при этом доля блоков в них
колебалась 25-100% (47/190=25%, 39/100=39%, 33/71=46%, 20/63=32%, 5/5=100%) --
"N подряд" не отличал выгоревший прокси-пул от прогона, который честно
обогатил 125 карточек из 190. Причина: блоки автокоррелированы и идут
пачками -- при базовой доле 25-46% пачка из 5-6 подряд у независимой модели
была бы редкостью, но встречается в каждом прогоне.

Критерий заменён на долю блоков в скользящем окне последних N попыток: пачка
сама по себе больше не абортит, абортит устойчиво высокая доля (>= порога,
не строго "выше" -- окно 20 при пороге 0.7 абортит уже на 14/20, 13/20 ещё
нет). Окно короче типичной пачки бесполезно, поэтому дефолт (settings.
detail_backfill_block_ratio_window) заметно больше 5-6.

Safety-net для прогонов КОРОЧЕ окна (снапшот меньше window_size -- ratio-
критерий физически недостижим, окно никогда не заполнится) сохранён: если ВСЕ
попытки с начала прогона были блоками (ни одного успеха) и накопилось не
меньше safety_min -- абортим не дожидаясь окна. Раньше это было единственным
критерием под именем max_consecutive_blocks; имя и дефолт (5) сохранены как
safety_min, чтобы не менять поведение коротких прогонов вида 0/5 (#3184).

Условие "снапшот короче окна" обязательно, иначе safety-net гасит сам фикс:
холодный старт сессии/прокси -- самое вероятное место пачки блоков, и на
длинном прогоне (снапшот >= окна) пачка из safety_min блоков В НАЧАЛЕ
абортила бы точно так же, как до правки, до того как ratio-критерий вообще
успел бы включиться (#3184 review MAJOR 2).

Отказы-не-блоки (сеть/таймаут/парсер уровня "ответ есть, но наш") НЕ рвут
"чистоту" серии для safety-net -- это исторический инвариант
avito_detail_backfill (только успех сбрасывал consecutive_blocks,
TimeoutError/Exception его не трогали), сохранён через record_failure():
она двигает окно (значит, доля в ratio-критерии всё равно снижается), но НЕ
трогает _pure_block_run и НЕ сбрасывает серию. Из-за этого 4 таймаута + 5
блоков подряд (ни одного успеха) всё ещё дают safety-net abort -- это
осознанное поведение, а не баг: таймаут не доказывает, что тракт живой ответ
получает, только успех доказывает.
"""

from __future__ import annotations

from collections import Counter, deque
from dataclasses import dataclass, field


@dataclass
class BlockRatioBreaker:
    """Стейт одного прогона detail-backfill'а: окно, safety-net, гистограмма пачек."""

    window_size: int
    ratio_threshold: float
    safety_min: int
    snapshot_size: int
    _window: deque[bool] = field(init=False, repr=False)
    _consecutive_blocks: 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)

    def __post_init__(self) -> None:
        self._window = deque(maxlen=max(self.window_size, 1))

    @property
    def consecutive_blocks(self) -> int:
        """Для логов ABORT/BLOCKED -- та же цифра, что раньше выводилась в consecutive=%d."""
        return self._consecutive_blocks

    def record_block(self) -> None:
        self._consecutive_blocks += 1
        self._window.append(True)

    def record_success(self) -> None:
        """Единственный исход, снимающий safety-net (#3184: пачка блоков ПОСЛЕ хотя
        бы одного успеха -- уже не "чистый с рождения прогона" burst)."""
        self._flush_streak()
        self._pure_block_run = False
        self._window.append(False)

    def record_failure(self) -> None:
        """Отказ-не-блок: в знаменатель окна идёт (тракт получил ответ, не блок),
        но серию/purity НЕ трогает -- зеркалит исторический avito-инвариант."""
        self._window.append(False)

    def record_neutral(self) -> None:
        """404/gone у avito -- не блок и не проверка тракта, не двигает ни окно, ни
        серию (тот же смысл, что "neutral to the breaker" в комментарии у
        AvitoListingGoneError). Класс написан источник-агностично на будущее (не
        только avito), но сейчас единственный вызывающий -- avito_detail_backfill
        (#3184 review MAJOR 1: применение к domclick_detail_backfill снято из этой
        задачи -- своя калибровка, свои ограничения прокси-пула)."""
        return

    def _flush_streak(self) -> None:
        if self._consecutive_blocks:
            self.streak_histogram[self._consecutive_blocks] += 1
        self._consecutive_blocks = 0

    def should_abort(self) -> bool:
        # Safety-net -- ТОЛЬКО когда ratio-критерий физически недостижим (снапшот
        # короче окна), иначе пачка safety_min блоков в начале длинного прогона
        # абортила бы его так же, как до правки (#3184 review MAJOR 2).
        if (
            self.snapshot_size < self.window_size
            and self._pure_block_run
            and self._consecutive_blocks >= self.safety_min
        ):
            return True
        if len(self._window) == self.window_size and self.window_size > 0:
            ratio = sum(self._window) / self.window_size
            if ratio >= self.ratio_threshold:
                return True
        return False

    def finalize(self) -> dict[str, int]:
        """Досчитать хвостовую пачку (прогон оборвался посреди серии блоков, без
        финального успеха) и отдать гистограмму для counters (jsonb) -- ключи СТРОКОЙ:
        jsonb/JSON всё равно хранит только строковые ключи, отдаём их такими сразу,
        чтобы то, что пишем ({6: 1}), совпадало с тем, что потом читаем ({"6": 1})."""
        self._flush_streak()
        return {str(streak): count for streak, count in self.streak_histogram.items()}