fix(tradein/scrapers): ABORT-лог называл серию блоков, хотя рвал прогон по доле

Прод-прогон 5210 оборвался по ratio-критерию — 14 блоков из 20, ровно порог 0.7 —
и отчитался строкой «ABORT -- 1 consecutive blocks». Число верное: последняя серия
в тот момент действительно равнялась единице (19-я попытка успех, 20-я блок).
Величина не та. Читатель лога видит цифру, по которой обрыва быть не могло, и идёт
искать несуществующий баг в брейкере.

Причина: #3184 заменил критерий обрыва на долю в скользящем окне, а текст лога
остался от прежнего критерия «N подряд» — то есть ровно та же болезнь, которую
#3178 лечил у соседней строки (литерал «IP rate-limited» вместо измеренной причины).

- BlockRatioBreaker.abort_reason() возвращает "ratio" / "safety_net" / None;
  should_abort() выражен через него, поведение не меняется.
- abort_explanation() даёт текст с той величиной, по которой обрыв и произошёл:
  доля печатает «доля блоков 14/20 в окне (порог 70%)», safety-net — «5 блоков
  подряд без единого успеха (снапшот 5 короче окна 20)».
- counters["abort_reason"] — чтобы причина обрыва читалась SQL-запросом по
  scrape_runs, а не грепом контейнера. Ключа нет, если прогон не обрывался.

Тесты (проверено мутацией источника — на прежнем сообщении оба падают):
- ratio-обрыв на раскладке прогона 5210 (серия на обрыве = 1) требует «14/20»
  в логе и отсутствия слова consecutive;
- safety-net требует «5 блоков подряд» и отсутствия «доля блоков» — без этого
  зеркала первый тест проходил бы и у сообщения, всегда печатающего долю;
- прогон без обрыва (13/20) не пишет abort_reason в counters.

Refs #3184, #3178

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

@@ -78,6 +78,16 @@ class BlockRatioBreaker:
         """Для логов ABORT/BLOCKED -- та же цифра, что раньше выводилась в consecutive=%d."""
         return self._consecutive_blocks
 
+    @property
+    def window_blocks(self) -> int:
+        """Числитель ratio-критерия: сколько блоков в текущем окне."""
+        return sum(self._window)
+
+    @property
+    def window_len(self) -> int:
+        """Знаменатель ratio-критерия: сколько попыток уже влезло в окно."""
+        return len(self._window)
+
     def record_block(self) -> None:
         self._consecutive_blocks += 1
         self._window.append(True)
@@ -108,7 +118,12 @@ class BlockRatioBreaker:
             self.streak_histogram[self._consecutive_blocks] += 1
         self._consecutive_blocks = 0
 
-    def should_abort(self) -> bool:
+    def abort_reason(self) -> str | None:
+        """Какой критерий требует обрыва прямо сейчас, или None.
+
+        Возвращает "safety_net" / "ratio" / None. Состояние не меняет, поэтому
+        вызывать можно сколько угодно раз -- в том числе повторно, ради текста лога.
+        """
         # Safety-net -- ТОЛЬКО когда ratio-критерий физически недостижим (снапшот
         # короче окна), иначе пачка safety_min блоков в начале длинного прогона
         # абортила бы его так же, как до правки (#3184 review MAJOR 2).
@@ -117,12 +132,40 @@ class BlockRatioBreaker:
             and self._pure_block_run
             and self._consecutive_blocks >= self.safety_min
         ):
-            return True
+            return "safety_net"
         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
+                return "ratio"
+        return None
+
+    def should_abort(self) -> bool:
+        return self.abort_reason() is not None
+
+    def abort_explanation(self) -> str:
+        """Текст для лога ABORT -- ровно та величина, по которой обрыв и произошёл.
+
+        До этого лог печатал "%d consecutive blocks" всегда, а критерий с #3184
+        стал ratio: прогон 5210 (14 блоков из 20, обрыв ровно по порогу 0.7) выдал
+        "ABORT -- 1 consecutive blocks", потому что в момент срабатывания текущая
+        серия равнялась единице. Число верное, величина не та -- читатель лога видит
+        цифру, по которой обрыва быть не могло, и идёт искать несуществующий баг.
+
+        Пустая строка означает "рвать не по чему" -- вызывается только под
+        should_abort(), так что в логи не попадает.
+        """
+        reason = self.abort_reason()
+        if reason == "ratio":
+            return (
+                f"доля блоков {self.window_blocks}/{self.window_len} в окне "
+                f"(порог {self.ratio_threshold * 100:.0f}%)"
+            )
+        if reason == "safety_net":
+            return (
+                f"{self._consecutive_blocks} блоков подряд без единого успеха "
+                f"(снапшот {self.snapshot_size} короче окна {self.window_size})"
+            )
+        return ""
 
     def finalize(self) -> dict[str, int]:
         """Досчитать хвостовую пачку (прогон оборвался посреди серии блоков, без

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

@@ -37,6 +37,11 @@ block_ratio_window/_threshold) ИЛИ safety-net для коротких про
 consecutive_blocks блоков подряд БЕЗ единого успеха с самого начала — старое
 поведение для случая вида 0/5). Гистограмма длин пачек блоков едет в
 counters["block_streak_histogram"], иначе эффект правки нечем измерить постфактум.
+Какой из двух критериев сработал -- в counters["abort_reason"] ("ratio" /
+"safety_net"); ключа нет, если прогон не обрывался. Он же называется в ABORT-логе
+своей величиной: доля печатает "14/20", safety-net -- длину серии. Раньше лог
+печатал серию всегда, и прогон 5210 (обрыв по доле 14/20) отчитался как
+"ABORT -- 1 consecutive blocks".
 """
 
 from __future__ import annotations
@@ -431,6 +436,7 @@ async def run_avito_detail_backfill(
         )
         consecutive_failures = 0
         aborted_by_blocks = False
+        abort_reason: str | None = None
         do_sleep = False
         items_since_warm = 0
         # Перепись причин (блоки + отказы) — переживает пересоздание контейнера,
@@ -632,12 +638,18 @@ async def run_avito_detail_backfill(
                 # abort FIRST — на последнем блоке не тратим cooldown+research/rotate.
                 # #3184: доля в скользящем окне ИЛИ safety-net (все попытки — блоки, ни
                 # одного успеха, накопилось max_consecutive_blocks) -- см. module docstring.
-                if breaker.should_abort():
+                abort_reason = breaker.abort_reason()
+                if abort_reason is not None:
+                    # Печатаем ту величину, по которой обрыв и произошёл. Раньше здесь
+                    # всегда стояло "%d consecutive blocks", хотя критерий с #3184 стал
+                    # ratio: прогон 5210 (14 блоков из 20, ровно порог) отпечатал
+                    # "ABORT -- 1 consecutive blocks" -- текущая серия в тот момент
+                    # действительно равнялась единице, но обрыв был не по ней.
                     logger.error(
-                        "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, "
+                        "avito_detail_backfill: run_id=%d ABORT -- %s, "
                         "частая причина: %s. enriched=%d attempted=%d",
                         run_id,
-                        breaker.consecutive_blocks,
+                        breaker.abort_explanation(),
                         _top_failure(failure_census) or "причина не определена",
                         counters.enriched,
                         counters.attempted,
@@ -739,6 +751,11 @@ async def run_avito_detail_backfill(
         streak_histogram = breaker.finalize()
         if streak_histogram:
             current_counters["block_streak_histogram"] = streak_histogram  # type: ignore[assignment]
+        # Какой из двух критериев оборвал прогон -- в counters, а не только в логах:
+        # разбор простоя идёт SQL-запросом по scrape_runs, а не грепом контейнера,
+        # и без этого ключа "banned" опять не отличить по причине (#3178).
+        if abort_reason is not None:
+            current_counters["abort_reason"] = abort_reason  # type: ignore[assignment]
         runs_mod.mark_backfill_finished(
             db,
             run_id,

tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py

@@ -1308,3 +1308,122 @@ async def test_backfill_ratio_boundary_13_of_20_does_not_abort() -> None:
     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_ratio_abort_log_names_the_ratio_not_the_streak(caplog: Any) -> None:
+    """ABORT по доле называет долю, а не текущую серию (#3184 follow-up).
+
+    Воспроизводит прод-прогон 5210: 14 блоков из 20, обрыв ровно по порогу 0.7, но
+    ПОСЛЕДНЯЯ серия блоков к этому моменту равна единице (позиция 19 -- успех,
+    позиция 20 -- блок). Лог печатал "ABORT -- 1 consecutive blocks": число верное,
+    величина не та. Читатель видит цифру, по которой обрыва быть не могло.
+    """
+    total = 20
+    # Раскладка ровно как в прогоне 5210.
+    block_positions = {2, 3, 5, 6, 7, 8, 9, 11, 12, 13, 14, 16, 17, 20}
+    assert len(block_positions) == 14
+    assert 19 not in block_positions and 20 in block_positions, "серия на обрыве = 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),
+        caplog.at_level("ERROR"),
+    ):
+        result = await run_avito_detail_backfill(
+            db, run_id=110, params={"batch_size": total, "budget_sec": 3600}
+        )
+
+    assert result.blocked == 14
+    abort_records = [r.message for r in caplog.records if "ABORT" in r.message]
+    assert abort_records, "ожидался ABORT-лог"
+    msg = abort_records[0]
+    assert "14/20" in msg, msg
+    assert "consecutive" not in msg, msg
+    assert "1 блоков подряд" not in msg, msg
+    counters = runs.mark_backfill_finished.call_args.args[2]
+    assert counters["abort_reason"] == "ratio"
+
+
+@pytest.mark.asyncio
+async def test_backfill_safety_net_abort_log_names_the_streak(caplog: Any) -> None:
+    """ABORT по safety-net называет серию -- ту величину, по которой он и сработал.
+
+    Зеркало предыдущего теста: снапшот короче окна, ratio недостижим, рвёт серия.
+    Если бы обе ветки печатали один и тот же текст, тест выше проходил бы и у
+    сломанного сообщения.
+    """
+    total = 5  # короче окна 20 -- ratio-критерий физически недостижим
+    snapshot = _make_snapshot(total)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(side_effect=AvitoBlockedError("ip blocked"))
+    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),
+        caplog.at_level("ERROR"),
+    ):
+        result = await run_avito_detail_backfill(
+            db, run_id=111, params={"batch_size": total, "budget_sec": 3600}
+        )
+
+    assert result.blocked == 5
+    abort_records = [r.message for r in caplog.records if "ABORT" in r.message]
+    assert abort_records, "ожидался ABORT-лог"
+    msg = abort_records[0]
+    assert "5 блоков подряд" in msg, msg
+    assert "доля блоков" not in msg, msg
+    counters = runs.mark_backfill_finished.call_args.args[2]
+    assert counters["abort_reason"] == "safety_net"
+
+
+@pytest.mark.asyncio
+async def test_backfill_no_abort_leaves_no_abort_reason_in_counters() -> None:
+    """Прогон без обрыва не пишет abort_reason -- ключ означает 'оборвались', а не
+    'считали критерий'."""
+    total = 20
+    block_positions = {8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20}  # 13/20 = 0.65
+    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),
+    ):
+        await run_avito_detail_backfill(
+            db, run_id=112, params={"batch_size": total, "budget_sec": 3600}
+        )
+
+    counters = runs.mark_backfill_finished.call_args.args[2]
+    assert "abort_reason" not in counters