From 745f15bdb1a5e3934b829c990572b604a8c44786 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 28 Aug 2026 21:15:36 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein/scrapers):=20=D0=BE=D0=B1=D1=80?= =?UTF-8?q?=D1=8B=D0=B2=20=D0=BF=D0=BE=20=D1=81=D0=B5=D1=80=D0=B8=D0=B8=20?= =?UTF-8?q?=D0=B1=D0=BB=D0=BE=D0=BA=D0=BE=D0=B2=20=D1=80=D0=B2=D0=B0=D0=BB?= =?UTF-8?q?=20=D0=BA=D0=B0=D0=B6=D0=B4=D1=8B=D0=B9=20=D0=BF=D1=80=D0=BE?= =?UTF-8?q?=D0=B3=D0=BE=D0=BD,=20=D0=B2=D0=BA=D0=BB=D1=8E=D1=87=D0=B0?= =?UTF-8?q?=D1=8F=20=D0=B7=D0=B4=D0=BE=D1=80=D0=BE=D0=B2=D1=8B=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Все 40 прогонов avito_detail_backfill за 14 суток закончились 'banned'. Ни один не завершился штатно. Доля блоков при этом колебалась 25-100%: прогон, обогативший 125 карточек из 190, получал тот же статус, что и прогон 0 из 5. Обрывало правило "N блоков подряд". Оно не отличает выгоревший пул от здорового сбора, потому что блоки автокоррелированы и идут пачками: при базовой доле 25-46% пачка из 5-6 подряд у независимой модели была бы редкостью, а встречается в каждом прогоне. Критерий, срабатывающий всегда, не несёт информации. Теперь абортит доля блоков в скользящем окне последних N попыток (дефолт 20 при пороге 0.7, оба в настройках с валидацией диапазона). Пачка сама по себе обрыв не вызывает. Safety-net старого поведения сохранён, но только для прогонов КОРОЧЕ окна, где доля физически недостижима: 0/5 останавливается как раньше. Без этого условия safety-net гасил бы сам фикс -- холодный старт сессии самое вероятное место пачки, и длинный прогон обрывался бы на пятой попытке точно так же, как до правки. Домклик намеренно НЕ трогаем: у него один выделенный residential-прокси без ротации, и max_consecutive_blocks=3 у него ниже avito осознанно (data/sql/175) -- ранний обрыв бережёт репутацию единственного узла. Наша калибровка сделана на пуле С ротацией и туда не переносится; своя калибровка отдельной задачей. В counters пишется гистограмма длин пачек -- без неё эффект этой и любой следующей правки измерить нечем: сейчас там только итоги. Refs #3184 --- tradein-mvp/backend/app/core/config.py | 26 ++ .../app/services/backfill_block_breaker.py | 133 +++++++ .../app/tasks/avito_detail_backfill.py | 63 +++- .../tests/tasks/test_avito_detail_backfill.py | 326 ++++++++++++++++-- 4 files changed, 516 insertions(+), 32 deletions(-) create mode 100644 tradein-mvp/backend/app/services/backfill_block_breaker.py diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 03835c55..af4081b8 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -1114,6 +1114,32 @@ class Settings(BaseSettings): default=90.0, validation_alias="AVITO_DETAIL_FETCH_TIMEOUT_S" ) + # #3184: доля блоков в скользящем окне последних N попыток -- критерий обрыва + # avito_detail_backfill (app.services.backfill_block_breaker.BlockRatioBreaker), + # взамен голого "N блоков подряд". ТОЛЬКО avito -- изначальный план распространить + # тот же критерий на domclick_detail_backfill снят ревью (#3184 review MAJOR 1): + # у Домклика один выделенный residential-прокси БЕЗ ротации (см. + # data/sql/175_scrape_schedules_seed_domclick_detail_backfill.sql), калибровка + # 20/0.7 сделана на пуле С ротацией (avito) и для него не годится без своего + # замера -- отдельная задача. + # Калибровка по 40 прогонам avito за 14 суток (2026-08-14..28): ВСЕ 40 + # закончились 'banned', доля блоков колебалась 25-100% (25%/190 попыток честно + # обогащённых 125, 39%/100, 46%/71, 32%/63, 100%/5) -- "N подряд" не отличал + # выгоревший прокси-пул от здорового прогона, потому что блоки автокоррелированы + # и пачки 5-6 подряд встречаются в каждом прогоне при базовой доле 25-46%. Окно + # 20 выбрано заметно больше типичной пачки, порог 0.7 -- между здоровыми (25-48%) + # и выгоревшими (100%) прогонами большой запас. ENV: DETAIL_BACKFILL_BLOCK_RATIO_ + # WINDOW / _THRESHOLD -- подкрутка без релиза. + # ge/le (#3184 review MINOR 2): WINDOW=0 или THRESHOLD в процентах (70 вместо + # 0.7, частая опечатка оператора) иначе молча выключают критерий навсегда -- + # 0.7 > 70 никогда не бывает True, а окно 0 схлопывает should_abort() в no-op. + detail_backfill_block_ratio_window: int = Field( + default=20, ge=1, validation_alias="DETAIL_BACKFILL_BLOCK_RATIO_WINDOW" + ) + detail_backfill_block_ratio_threshold: float = Field( + default=0.7, ge=0.0, le=1.0, validation_alias="DETAIL_BACKFILL_BLOCK_RATIO_THRESHOLD" + ) + # ── #884/#905/#1805: BrowserFetcher — HTTP-клиент к tradein-browser ───────── # scraper_fetch_mode: "browser" (дефолт с #1805 — HTTP POST к tradein-browser # /fetch через per-provider camoufox + ротирующий backconnect-прокси) или diff --git a/tradein-mvp/backend/app/services/backfill_block_breaker.py b/tradein-mvp/backend/app/services/backfill_block_breaker.py new file mode 100644 index 00000000..bd19679c --- /dev/null +++ b/tradein-mvp/backend/app/services/backfill_block_breaker.py @@ -0,0 +1,133 @@ +"""Критерий обрыва 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()} diff --git a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py index 3fc18228..6e3adf47 100644 --- a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py +++ b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py @@ -26,6 +26,17 @@ rotate IP on every block, abort after max_consecutive_blocks. Статус об 3306 (blocked=5, 0 обогащено) получил 'platform' по УМОЛЧАНИЮ — финализатор диагноз не передавал, а browser-режим fetch_detail всё равно превращал отказ сайдкара в AvitoBlockedError, так что передавать было бы нечего. + +КРИТЕРИЙ ОБРЫВА (#3184): голое "N блоков подряд" (max_consecutive_blocks) +абортило прогон 125/190 (25% блоков, честно обогащённых 125 карточек) статусом +'banned' наравне с прогоном 0/5 (100% блоков) — блоки идут автокоррелированными +пачками, и пачка 5-6 подряд у здорового прогона с базовой долей 25-46% не +редкость. Теперь абортит app.services.backfill_block_breaker.BlockRatioBreaker: +доля блоков в скользящем окне последних N попыток (settings.detail_backfill_ +block_ratio_window/_threshold) ИЛИ safety-net для коротких прогонов (max_ +consecutive_blocks блоков подряд БЕЗ единого успеха с самого начала — старое +поведение для случая вида 0/5). Гистограмма длин пачек блоков едет в +counters["block_streak_histogram"], иначе эффект правки нечем измерить постфактум. """ from __future__ import annotations @@ -70,6 +81,7 @@ from sqlalchemy.orm import Session from app.core.config import settings from app.core.shutdown import shutdown_requested from app.services import scrape_runs as runs_mod +from app.services.backfill_block_breaker import BlockRatioBreaker from app.services.proxy_egress import resolve_proxy_url from app.services.scraper_adapters import RealProxyProvider, RealScraperConfig @@ -193,7 +205,11 @@ async def run_avito_detail_backfill( всплеску свежих oblast-листингов вытеснить ЕКБ из top-N по scraped_at. budget_sec: float -- wall-clock budget per run, default 3600s. request_delay_sec: float -- delay between listings, default 6.0s. - max_consecutive_blocks: int -- abort threshold, default 5. + max_consecutive_blocks: int -- safety_min для BlockRatioBreaker (#3184): + абортит прогоны короче окна, если ВСЕ попытки с начала были блоками + (ни одного успеха), default 5. Основной критерий -- доля блоков в + скользящем окне, см. settings.detail_backfill_block_ratio_window/ + _threshold (module docstring). max_consecutive_failures: int -- порог обрыва по отказам-не-блокам, default 25 (см. комментарий у чтения параметра ниже). @@ -205,6 +221,8 @@ async def run_avito_detail_backfill( oblast_batch_size = int(params.get("oblast_batch_size", 100)) budget_sec = float(params.get("budget_sec", 3600)) request_delay_sec = float(params.get("request_delay_sec", 6.0)) + # #3184: теперь safety_min BlockRatioBreaker (см. module docstring) -- не общий + # критерий обрыва, а страховка для прогонов короче окна. max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5)) # Брейкер на отказы-НЕ-блоки. Блоки свой брейкер имели с самого начала, отказы — # нет, и это стоило трёх ночей подряд: 3-5 августа прогон делал ~1600 попыток, @@ -399,7 +417,18 @@ async def run_avito_detail_backfill( "curl/backconnect" if use_curl else "browser", ) - consecutive_blocks = 0 + # #3184: доля блоков в скользящем окне вместо голого "N подряд" -- см. module + # docstring и app.services.backfill_block_breaker. + breaker = BlockRatioBreaker( + window_size=int(settings.detail_backfill_block_ratio_window), + ratio_threshold=float(settings.detail_backfill_block_ratio_threshold), + safety_min=max_consecutive_blocks, + # snapshot_size гейтит safety-net (#3184 review MAJOR 2): пачка блоков в + # начале ДЛИННОГО прогона не должна абортить его так же, как раньше -- + # safety-net включён только когда снапшот короче окна и ratio-критерий + # физически недостижим (см. backfill_block_breaker module docstring). + snapshot_size=len(snapshot), + ) consecutive_failures = 0 aborted_by_blocks = False do_sleep = False @@ -530,7 +559,7 @@ async def run_avito_detail_backfill( counters.enriched += 1 if use_curl: items_since_warm += 1 - consecutive_blocks = 0 + breaker.record_success() consecutive_failures = 0 except AvitoListingGoneError as gone_exc: @@ -539,10 +568,11 @@ async def run_avito_detail_backfill( # browser-mode рендерит их «Ошибка 404» без item-view → раньше это # ловилось как soft-block → consecutive-block breaker абортил run до # live-листингов (run 458: attempted=5 enriched=0 blocked=5 → abort). - # 404 нейтрален к breaker'у: consecutive_blocks НЕ трогаем (не растим - # и не сбрасываем). Метим is_active=FALSE → листинг уходит из scope - # (snapshot SELECT фильтрует is_active = TRUE) и не тратит фетчи впредь. + # 404 нейтрален к breaker'у (#3184: record_neutral -- ни окно, ни серия, + # ни safety-net не трогаются). Метим is_active=FALSE → листинг уходит из + # scope (snapshot SELECT фильтрует is_active = TRUE) и не тратит фетчи впредь. counters.gone += 1 + breaker.record_neutral() # 404 — честный ответ площадки, значит тракт цел: серия отказов # прерывается (блочный брейкер 404 не трогает — см. #2034). consecutive_failures = 0 @@ -586,7 +616,7 @@ async def run_avito_detail_backfill( ) except (AvitoBlockedError, AvitoRateLimitedError) as e: - consecutive_blocks += 1 + breaker.record_block() counters.blocked += 1 failure_census[_failure_signature(e)] += 1 block_ban_kinds[ban_kind_of_exception(e)] += 1 @@ -596,17 +626,18 @@ async def run_avito_detail_backfill( run_id, idx + 1, len(snapshot), - consecutive_blocks, + breaker.consecutive_blocks, e, ) - # abort FIRST — на последнем блоке не тратим cooldown+research/rotate - # (consecutive_blocks уже инкрементнут выше). - if consecutive_blocks >= max_consecutive_blocks: + # abort FIRST — на последнем блоке не тратим cooldown+research/rotate. + # #3184: доля в скользящем окне ИЛИ safety-net (все попытки — блоки, ни + # одного успеха, накопилось max_consecutive_blocks) -- см. module docstring. + if breaker.should_abort(): logger.error( "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, " "частая причина: %s. enriched=%d attempted=%d", run_id, - consecutive_blocks, + breaker.consecutive_blocks, _top_failure(failure_census) or "причина не определена", counters.enriched, counters.attempted, @@ -655,6 +686,7 @@ async def run_avito_detail_backfill( # run не zombie #1950). Не считаем soft-блоком: rotate не дёргаем. counters.failed += 1 consecutive_failures += 1 + breaker.record_failure() failure_census[_failure_signature(e)] += 1 logger.warning( "avito_detail_backfill: run_id=%d listing %s TIMEOUT (>%.0fs) -- skip", @@ -670,6 +702,7 @@ async def run_avito_detail_backfill( except Exception as e: counters.failed += 1 consecutive_failures += 1 + breaker.record_failure() failure_census[_failure_signature(e)] += 1 logger.warning( "avito_detail_backfill: run_id=%d listing %s failed: %s", @@ -700,6 +733,12 @@ async def run_avito_detail_backfill( counters.duration_sec = time.monotonic() - start current_counters = counters.to_dict() + # #3184: гистограмма длин пачек блоков (streak -> сколько раз встретилась) -- + # иначе эффект правки на #2674-статистике нечем измерить постфактум. finalize() + # досчитывает хвостовую пачку, если прогон оборвался посреди серии. + streak_histogram = breaker.finalize() + if streak_histogram: + current_counters["block_streak_histogram"] = streak_histogram # type: ignore[assignment] runs_mod.mark_backfill_finished( db, run_id, diff --git a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py index b7d024d7..4d5dc758 100644 --- a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py +++ b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py @@ -62,6 +62,19 @@ def _make_snapshot(n: int) -> list[dict]: return [{"id": i + 1, "source_url": f"/items/{i + 1}"} for i in range(n)] +def _fake_settings(**overrides: object) -> MagicMock: + """settings.* мокается атрибутами; bare MagicMock() без явных + detail_backfill_block_ratio_window/_threshold отдаёт int(MagicMock())==1 -- + окно=1, порог=1.0 абортит на первом же блоке (#3184 review MINOR 4). Единая + точка дефолтов вместо 23 копий тех же двух kwargs на каждом call site.""" + defaults: dict[str, object] = { + "detail_backfill_block_ratio_window": 20, + "detail_backfill_block_ratio_threshold": 0.7, + } + defaults.update(overrides) + return MagicMock(**defaults) + + def _mock_db(snapshot: list[dict]) -> MagicMock: """Fake Session: first execute() returns snapshot via .mappings().all().""" db = MagicMock() @@ -99,7 +112,9 @@ async def test_backfill_empty_snapshot_marks_done() -> None: db = _mock_db([]) runs = MagicMock() fake_session = AsyncMock() - fake_settings = MagicMock(scraper_fetch_mode="cffi") + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=fake_session), @@ -130,7 +145,9 @@ async def test_backfill_processes_snapshot_to_completion() -> None: mock_enrichment = MagicMock() mock_fetch = AsyncMock(return_value=mock_enrichment) mock_save = MagicMock(return_value=True) - fake_settings = MagicMock(scraper_fetch_mode="cffi") + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -178,7 +195,7 @@ async def test_backfill_build_warmed_session_receives_config() -> None: mock_enrichment = MagicMock() mock_fetch = AsyncMock(return_value=mock_enrichment) mock_save = MagicMock(return_value=True) - fake_settings = MagicMock( + fake_settings = _fake_settings( scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=True, ) @@ -225,7 +242,10 @@ async def test_backfill_reports_ban_kind_of_the_blocks_it_saw( runs = MagicMock() mock_scraper = MagicMock() mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=False, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -268,7 +288,10 @@ async def test_backfill_ban_kinds_census_keeps_multiplicities() -> None: ) mock_scraper = MagicMock() mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=False, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -302,7 +325,10 @@ async def test_backfill_abort_log_has_no_ip_rate_limited_literal(caplog: Any) -> mock_fetch = AsyncMock(side_effect=AvitoBlockedError("firewall/soft-block")) mock_scraper = MagicMock() mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=False, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -344,7 +370,10 @@ async def test_backfill_blocked_abort_after_max_consecutive() -> None: mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy. - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=False, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -383,7 +412,9 @@ async def test_backfill_sigterm_drain_breaks_and_marks_done_partial() -> None: mock_enrichment = MagicMock() mock_fetch = AsyncMock(return_value=mock_enrichment) mock_save = MagicMock(return_value=True) - fake_settings = MagicMock(scraper_fetch_mode="cffi") + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -417,7 +448,9 @@ async def test_backfill_budget_guard_stops_loop() -> None: db = _mock_db(snapshot) runs = MagicMock() mono_values = iter([0.0, 999.0, 999.0]) - fake_settings = MagicMock(scraper_fetch_mode="cffi") + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -438,7 +471,9 @@ async def test_backfill_top_level_exception_marks_failed() -> None: db = MagicMock() db.execute.side_effect = RuntimeError("DB connection lost") runs = MagicMock() - fake_settings = MagicMock(scraper_fetch_mode="cffi") + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -469,7 +504,10 @@ async def test_backfill_rotate_ip_called_on_each_block() -> None: mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy. - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=False, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -503,7 +541,9 @@ async def test_backfill_snapshot_filters_ekb_active_only() -> None: """ db = _mock_db([]) runs = MagicMock() - fake_settings = MagicMock(scraper_fetch_mode="cffi") + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -649,7 +689,9 @@ async def test_backfill_fetch_exception_continues() -> None: runs = MagicMock() mock_enrichment = MagicMock() mock_fetch = AsyncMock(side_effect=[RuntimeError("parse error"), mock_enrichment]) - fake_settings = MagicMock(scraper_fetch_mode="cffi") + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -700,7 +742,7 @@ async def test_backfill_fetch_timeout_skips_and_continues() -> None: return mock_enrichment # Короткий timeout (50ms) — тест не ждёт реальные 90s; rotate-settle тоже 0. - fake_settings = MagicMock( + fake_settings = _fake_settings( scraper_fetch_mode="cffi", avito_detail_fetch_timeout_s=0.05, avito_proxy_rotate_settle_s=0.0, @@ -747,7 +789,10 @@ async def test_backfill_listing_gone_marks_inactive_no_breaker() -> None: mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy. - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=False, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -796,7 +841,7 @@ async def test_backfill_use_curl_flag_skips_browser_fetcher() -> None: mock_fetch = AsyncMock(return_value=mock_enrichment) mock_save = MagicMock(return_value=True) # Флаг use_curl=True, fetch_mode=browser (но флаг перекрывает) - fake_settings = MagicMock( + fake_settings = _fake_settings( scraper_fetch_mode="browser", avito_detail_backfill_use_curl=True, ) @@ -835,7 +880,7 @@ async def test_backfill_use_curl_false_creates_browser_fetcher() -> None: mock_enrichment = MagicMock() mock_fetch = AsyncMock(return_value=mock_enrichment) mock_save = MagicMock(return_value=True) - fake_settings = MagicMock( + fake_settings = _fake_settings( scraper_fetch_mode="browser", avito_detail_backfill_use_curl=False, ) @@ -900,7 +945,7 @@ async def test_backfill_use_curl_block_cooldown_research_no_rebuild() -> None: blocked_exc = AvitoBlockedError("rate-limited") mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment]) warmed_session = AsyncMock() - fake_settings = MagicMock( + fake_settings = _fake_settings( scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=True, ) @@ -963,7 +1008,10 @@ async def test_backfill_aborts_on_consecutive_failures_and_names_the_reason() -> "See https://curl.se/libcurl/c/libcurl-errors.html" ) ) - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=True) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -1003,7 +1051,10 @@ async def test_backfill_success_resets_failure_streak() -> None: boom = ValueError("avito detail HTTP 500 for https://www.avito.ru/x") # 2 отказа, успех, 2 отказа — при пороге 3 ни одна серия его не достигает. mock_fetch = AsyncMock(side_effect=[boom, boom, MagicMock(), boom, boom]) - fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=True) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) with ( patch(_SETTINGS, fake_settings), patch(_SESSION, return_value=AsyncMock()), @@ -1022,3 +1073,238 @@ async def test_backfill_success_resets_failure_streak() -> None: assert result.attempted == 5 assert result.enriched == 1 assert result.failed == 4 + + +def _spaced_outcomes(total: int, block_positions: set[int]) -> list: + """1-indexed позиции block_positions -> AvitoBlockedError, остальные -- success MagicMock. + + Ровный интервал между блоками (не пачками) -- моделирует прод-профиль 4786/5200 + (#3184): блоки есть, но локально в любом 20-окне их доля намного ниже порога. + """ + return [ + AvitoBlockedError("ip blocked") if pos in block_positions else MagicMock() + for pos in range(1, total + 1) + ] + + +@pytest.mark.asyncio +async def test_backfill_25pct_blocks_spread_does_not_abort() -> None: + """47 блоков из 190 попыток (25%, прогон 4786) -- НЕ обрывается (#3184). + + Раньше это же соотношение (после единственной пачки >= max_consecutive_blocks) + ловилось общим брейкером и уходило в 'banned', хотя прогон честно обогатил + 125/190 карточек. Блоки расставлены каждый 4-й (изолированно, не пачкой) -- + ни в одном 20-окне доля не приближается к порогу 0.7. + """ + total = 190 + block_positions = {pos for pos in range(1, total + 1) if pos % 4 == 0} + assert len(block_positions) == 47 + snapshot = _make_snapshot(total) + db = _mock_db(snapshot) + runs = MagicMock() + mock_fetch = AsyncMock(side_effect=_spaced_outcomes(total, block_positions)) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=101, params={"batch_size": total, "budget_sec": 3600} + ) + + assert result.attempted == total + assert result.blocked == 47 + assert result.enriched == total - 47 + runs.mark_backfill_finished.assert_called_once() + assert runs.mark_backfill_finished.call_args.kwargs["aborted_by_blocks"] is False + + +@pytest.mark.asyncio +async def test_backfill_32pct_blocks_spread_does_not_abort() -> None: + """20 блоков из 63 попыток (32%, прогон 5200) -- НЕ обрывается (#3184).""" + total = 63 + block_positions = {pos for pos in range(3, 61, 3)} + assert len(block_positions) == 20 + snapshot = _make_snapshot(total) + db = _mock_db(snapshot) + runs = MagicMock() + mock_fetch = AsyncMock(side_effect=_spaced_outcomes(total, block_positions)) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=102, params={"batch_size": total, "budget_sec": 3600} + ) + + assert result.attempted == total + assert result.blocked == 20 + assert result.enriched == total - 20 + runs.mark_backfill_finished.assert_called_once() + assert runs.mark_backfill_finished.call_args.kwargs["aborted_by_blocks"] is False + + +@pytest.mark.asyncio +async def test_backfill_burst_mid_healthy_run_does_not_abort_and_streak_logged() -> None: + """6 блоков подряд посреди здорового прогона -- НЕ обрывает (#3184), а пачка + + едет в counters["block_streak_histogram"]. Раньше max_consecutive_blocks=5 + абортил бы уже на 5-м блоке пачки независимо от контекста; серия сама по себе + больше не критерий -- 20 успехов до пачки держат окно далеко от порога 0.7. + """ + total = 50 + block_positions = set(range(21, 27)) # 6 подряд, позиции 21..26 + assert len(block_positions) == 6 + snapshot = _make_snapshot(total) + db = _mock_db(snapshot) + runs = MagicMock() + mock_fetch = AsyncMock(side_effect=_spaced_outcomes(total, block_positions)) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=103, params={"batch_size": total, "budget_sec": 3600} + ) + + assert result.attempted == total + assert result.blocked == 6 + runs.mark_backfill_finished.assert_called_once() + assert runs.mark_backfill_finished.call_args.kwargs["aborted_by_blocks"] is False + counters_arg = runs.mark_backfill_finished.call_args.args[2] + assert counters_arg["block_streak_histogram"] == {"6": 1} # jsonb -- ключи строкой + + +@pytest.mark.asyncio +async def test_backfill_ratio_criterion_aborts_after_window_fills_with_blocks() -> None: + """Успех на 1-й попытке, дальше сплошные блоки -- обрывает ПО ДОЛЕ в окне, не + safety-net (#3184 review MAJOR 3: раньше ни один тест не проверял срабатывание + ratio-критерия, только его отсутствие). + + Окно (20) заполняется целиком ровно на 20-й попытке (1 успех + 19 блоков) -- + ratio=19/20=0.95 >= порога 0.7, абортит там же, до 21-й попытки дело не доходит. + Safety-net здесь не может сработать: success на 1-й попытке снимает + _pure_block_run сразу, до любого блока. + """ + total = 200 + block_positions = set(range(2, total + 1)) # всё, кроме 1-й попытки + snapshot = _make_snapshot(total) + db = _mock_db(snapshot) + runs = MagicMock() + mock_fetch = AsyncMock(side_effect=_spaced_outcomes(total, block_positions)) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=104, params={"batch_size": total, "budget_sec": 3600} + ) + + assert result.attempted == 20 + assert result.blocked == 19 + assert result.enriched == 1 + runs.mark_backfill_finished.assert_called_once() + assert runs.mark_backfill_finished.call_args.kwargs["aborted_by_blocks"] is True + + +@pytest.mark.asyncio +async def test_backfill_ratio_boundary_14_of_20_aborts() -> None: + """14/20 = 0.7 -- ровно порог, should_abort сравнивает через >=, значит абортит + (#3184 review MAJOR 3 -- граница снизу).""" + total = 20 + block_positions = set(range(7, 21)) # позиции 7..20 включительно -- 14 штук + assert len(block_positions) == 14 + snapshot = _make_snapshot(total) + db = _mock_db(snapshot) + runs = MagicMock() + mock_fetch = AsyncMock(side_effect=_spaced_outcomes(total, block_positions)) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=105, params={"batch_size": total, "budget_sec": 3600} + ) + + assert result.blocked == 14 + runs.mark_backfill_finished.assert_called_once() + assert runs.mark_backfill_finished.call_args.kwargs["aborted_by_blocks"] is True + + +@pytest.mark.asyncio +async def test_backfill_ratio_boundary_13_of_20_does_not_abort() -> None: + """13/20 = 0.65 -- ниже порога 0.7, НЕ абортит (#3184 review MAJOR 3 -- граница + сверху, зеркало предыдущего теста).""" + total = 20 + block_positions = set(range(8, 21)) # позиции 8..20 включительно -- 13 штук + assert len(block_positions) == 13 + snapshot = _make_snapshot(total) + db = _mock_db(snapshot) + runs = MagicMock() + mock_fetch = AsyncMock(side_effect=_spaced_outcomes(total, block_positions)) + fake_settings = _fake_settings( + scraper_fetch_mode="cffi", + avito_detail_backfill_use_curl=True, + ) + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=106, params={"batch_size": total, "budget_sec": 3600} + ) + + assert result.attempted == 20 + assert result.blocked == 13 + runs.mark_backfill_finished.assert_called_once() + assert runs.mark_backfill_finished.call_args.kwargs["aborted_by_blocks"] is False +