feat(tradein/avito): бан площадки не менял IP — прогон добивался в тот же забаненный адрес (#3283)

Сайдкар с #3304 честно опознаёт бан-страницу и поднимает AvitoBlockedError, но
реакция на неё оставалась прежней: один bare-сброс контекста за прогон и дальше
попытки с ТОГО ЖЕ exit-IP, который площадка уже отвергла. Ротация была только
плановой, по счётчику попыток, и до конца прогона могла не наступить ни разу.

Теперь бан — самостоятельный триггер смены адреса, но с потолком, потому что
безоглядный сброс контекста на каждый блок даёт самоподдерживающийся каскад
(#3251/#3212): он выбрасывает пройденный QRATOR PoW, а адрес остаётся тем же.
Здесь этого не происходит — новый IP физически не несёт сожжённый PoW старого.

- avito_detail_backfill_rotate_on_ban_max=2, ..._min_gap=10 (карточек между
  ротациями). max=0 возвращает прежнее поведение байт-в-байт.
- Бюджет тратится и на ОТКАЗАВШЕЙ ротации — иначе отказавший rotate_proxy
  дёргался бы на каждом следующем бане до конца прогона.
- context_reset_used выставляется только по факту успеха: при тихом отказе
  rotate_proxy (исчерпан лимит, нет rotate_url, сеть) сброса внутри не было, и
  съесть им одноразовый bare-reset значило бы остаться и без адреса, и без сброса.
- Логи _rotate_current_proxy получили ярлык триггера: rotate-on-ban против
  rotate-by-attempts, иначе разбор прод-логов по факту бана вводит в заблуждение.

Тесты: 5 на ban-ветку, включая падение ротации → откат на bare-reset.

tradein-mvp/backend/app/core/config.py

@@ -1177,6 +1177,26 @@ class Settings(BaseSettings):
         default=15, ge=1, validation_alias="AVITO_DETAIL_BACKFILL_ROTATE_AFTER_ATTEMPTS"
     )
 
+    # #3283g: ротация exit-IP НА САМ БАН площадки, а не только по счётчику попыток.
+    # Бан привязан к IP (замерено вживую: rotate_proxy() лечит забаненный узел за
+    # секунды, clear_source_bans снимает запись из scrape_proxy_source_bans), но
+    # #3251/#3212 запрещают сбрасывать browser-context на КАЖДЫЙ блок -- сброс без
+    # смены IP выбрасывает пройденный QRATOR-PoW и запускает самоподдерживающийся
+    # каскад блоков на том же адресе. rotate_on_ban МЕНЯЕТ IP вместе со сбросом,
+    # поэтому не подвержена этому каскаду и может срабатывать больше одного раза за
+    # прогон -- но всё равно ограничена бюджетом, чтобы не выжигать прокси-пул на
+    # длинной серии блоков. 0 = выключено, полный no-op (никаких доп. ротаций/логов).
+    # ENV: AVITO_DETAIL_BACKFILL_ROTATE_ON_BAN_MAX.
+    avito_detail_backfill_rotate_on_ban_max: int = Field(
+        default=2, ge=0, validation_alias="AVITO_DETAIL_BACKFILL_ROTATE_ON_BAN_MAX"
+    )
+    # Минимум попыток между ЛЮБЫМИ двумя ротациями (по счётчику ИЛИ по бану) --
+    # не даёт двум сбросам контекста идти подряд, даже если оба бюджета формально
+    # ещё не исчерпаны. ENV: AVITO_DETAIL_BACKFILL_ROTATE_ON_BAN_MIN_GAP.
+    avito_detail_backfill_rotate_on_ban_min_gap: int = Field(
+        default=10, ge=0, validation_alias="AVITO_DETAIL_BACKFILL_ROTATE_ON_BAN_MIN_GAP"
+    )
+
     # #3184: доля блоков в скользящем окне последних N попыток -- критерий обрыва
     # avito_detail_backfill (app.services.backfill_block_breaker.BlockRatioBreaker),
     # взамен голого "N блоков подряд". ТОЛЬКО avito -- изначальный план распространить

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

@@ -181,8 +181,17 @@ def _top_failure(census: Counter[str]) -> str | None:
 _DEFAULT_ROTATE_RECONNECT_DELAY_S = 10.0
 
 
-async def _rotate_current_proxy(db: Session, run_id: int, browser_fetcher: BrowserFetcher) -> None:
-    """Ротация exit-IP арендованного прокси по счётчику попыток + сброс browser-контекста.
+async def _rotate_current_proxy(
+    db: Session,
+    run_id: int,
+    browser_fetcher: BrowserFetcher,
+    trigger: str = "rotate-by-attempts",
+) -> bool:
+    """Ротация exit-IP арендованного прокси + сброс browser-контекста.
+
+    Два триггера, оба пишут свой ярлык в логи через ``trigger``: плановый по счётчику
+    попыток (``rotate-by-attempts``) и реактивный по бану площадки (``rotate-on-ban``,
+    #3283). Разделение нужно только для читаемости логов — механика одна и та же.
 
     Свежий адрес со старыми куками бесполезен — личность браузера должна меняться
     ВМЕСТЕ с адресом, иначе следующий запрос уходит с нового IP, но с cookie-следом
@@ -210,20 +219,22 @@ async def _rotate_current_proxy(db: Session, run_id: int, browser_fetcher: Brows
     proxy_id = browser_fetcher.lease_id
     if proxy_id is None:
         logger.info(
-            "avito_detail_backfill: run_id=%d rotate-by-attempts skipped -- no leased proxy",
+            "avito_detail_backfill: run_id=%d %s skipped -- no leased proxy",
             run_id,
+            trigger,
         )
-        return
+        return False
 
     result = await rotate_proxy(db, proxy_id)
     if not result.ok:
         logger.warning(
-            "avito_detail_backfill: run_id=%d rotate-by-attempts FAILED proxy_id=%d: %s",
+            "avito_detail_backfill: run_id=%d %s FAILED proxy_id=%d: %s",
             run_id,
+            trigger,
             proxy_id,
             result.reason,
         )
-        return
+        return False
 
     delay = (
         result.reconnect_delay_s
@@ -231,15 +242,17 @@ async def _rotate_current_proxy(db: Session, run_id: int, browser_fetcher: Brows
         else _DEFAULT_ROTATE_RECONNECT_DELAY_S
     )
     logger.info(
-        "avito_detail_backfill: run_id=%d rotate-by-attempts OK proxy_id=%d new_ip=%s -- "
+        "avito_detail_backfill: run_id=%d %s OK proxy_id=%d new_ip=%s -- "
         "waiting %.1fs for channel reconnect",
         run_id,
+        trigger,
         proxy_id,
         result.new_ip,
         delay,
     )
     await asyncio.sleep(delay)
     browser_fetcher.request_context_reset()
+    return True
 
 
 @dataclass
@@ -530,6 +543,12 @@ async def run_avito_detail_backfill(
         # после принятой снова блокируется — одна осечка превращается в необратимый
         # каскад блоков (см. except-ветку ниже).
         context_reset_used = False
+        # #3283g: сколько раз ЗА ПРОГОН уже сработала ротация-на-бан (в отличие от
+        # context_reset_used эта ротация меняет IP вместе со сбросом, поэтому не
+        # подвержена каскаду #3251 и может срабатывать несколько раз, но бюджетно).
+        # Gap между любыми ротациями (по счётчику ИЛИ по бану) считается через
+        # attempts_since_rotation — он и так обнуляется на каждой ротации.
+        rotate_on_ban_used = 0
         # Перепись причин (блоки + отказы) — переживает пересоздание контейнера,
         # в отличие от логов; см. _failure_signature.
         failure_census: Counter[str] = Counter()
@@ -729,22 +748,77 @@ async def run_avito_detail_backfill(
                 failure_census[_failure_signature(e)] += 1
                 block_ban_kinds[ban_kind_of_exception(e)] += 1
                 do_sleep = False
-                # #3251: сброс тёплого browser-context'а — ровно один раз за прогон,
-                # и только на настоящий бан ПЛОЩАДКОЙ (AvitoBlockedError и подтипы:
-                # AvitoContentBlockedError, AvitoWarmupCookiesMissingError). НЕ на
-                # AvitoSidecarUnavailableError (подтип AvitoRateLimitedError) — это
-                # отказ НАШЕГО тракта, площадка тут ни при чём, context цел и сбрасывать
-                # его незачем. См. domclick_detail_backfill.py (#3212) — тот же приём:
-                # сброс на каждый блок сам себя поддерживает (пройденный QRATOR-PoW
-                # живёт в context'е, сброс его выбрасывает → следующая проверка с того
-                # же IP снова блокируется).
-                if (
-                    browser_fetcher is not None
-                    and isinstance(e, AvitoBlockedError)
-                    and not context_reset_used
-                ):
-                    context_reset_used = True
-                    browser_fetcher.request_context_reset()
+                # #3251/#3283g: на настоящий бан ПЛОЩАДКОЙ (AvitoBlockedError и подтипы:
+                # AvitoContentBlockedError, AvitoWarmupCookiesMissingError; НЕ на
+                # AvitoSidecarUnavailableError — подтип AvitoRateLimitedError, отказ
+                # НАШЕГО тракта, площадка ни при чём) есть два инструмента:
+                #   1) rotate-on-ban (#3283g) — меняет exit-IP И сбрасывает context
+                #      (через _rotate_current_proxy), бюджетно (rotate_on_ban_max за
+                #      прогон) и с gap'ом (rotate_on_ban_min_gap попыток от последней
+                #      ротации любого происхождения, gap считаем по
+                #      attempts_since_rotation — она и так обнуляется на ротациях).
+                #      Смена IP лечит бан вживую за секунды (см. module docstring
+                #      задачи) и НЕ подвержена каскаду ниже — новый адрес не хранит
+                #      сожжённый PoW старого.
+                #   2) bare context reset (#3251) — БЕЗ смены IP, ровно один раз за
+                #      прогон (domclick_detail_backfill #3212 — тот же приём). Сброс
+                #      на каждый блок сам себя поддерживает: пройденный QRATOR-PoW
+                #      живёт в context'е, сброс его выбрасывает → следующая проверка
+                #      с ТОГО ЖЕ IP снова блокируется — необратимый каскад.
+                # Приоритет — rotate-on-ban: если она сработала, context уже сброшен
+                # вместе со сменой IP, поэтому bare reset для этого события и всех
+                # последующих (пока context_reset_used не сброшен — а он не сбрасывается
+                # в рамках прогона) больше не нужен. Оба сброса в ОДНОМ событии не
+                # допускаем: разные механизмы, но request_context_reset() внутри один,
+                # и два подряд — то самое "два сброса подряд", которого просит избежать
+                # задача, даже если один из них идёт с новым IP.
+                if browser_fetcher is not None and isinstance(e, AvitoBlockedError):
+                    ban_budget = settings.avito_detail_backfill_rotate_on_ban_max
+                    if ban_budget > 0 and rotate_on_ban_used >= ban_budget:
+                        logger.info(
+                            "avito_detail_backfill: run_id=%d rotate-on-ban budget "
+                            "exhausted (%d/%d) -- staying on current proxy",
+                            run_id,
+                            rotate_on_ban_used,
+                            ban_budget,
+                        )
+                    elif (
+                        ban_budget > 0
+                        and attempts_since_rotation
+                        < settings.avito_detail_backfill_rotate_on_ban_min_gap
+                    ):
+                        logger.info(
+                            "avito_detail_backfill: run_id=%d rotate-on-ban skipped -- "
+                            "only %d attempts since last rotation (need >=%d)",
+                            run_id,
+                            attempts_since_rotation,
+                            settings.avito_detail_backfill_rotate_on_ban_min_gap,
+                        )
+                    elif ban_budget > 0:
+                        # Бюджет тратим и на неудачу — иначе отказавший rotate_proxy
+                        # дёргался бы на КАЖДОМ следующем бане до конца прогона.
+                        rotate_on_ban_used += 1
+                        attempts_since_rotation = 0
+                        logger.warning(
+                            "avito_detail_backfill: run_id=%d rotate-on-ban #%d/%d -- "
+                            "platform ban, rotating exit IP instead of bare reset",
+                            run_id,
+                            rotate_on_ban_used,
+                            ban_budget,
+                        )
+                        # context_reset_used выставляем ТОЛЬКО по факту успеха: при
+                        # тихом отказе rotate_proxy (лимит, нет rotate_url, сеть)
+                        # сброса контекста внутри НЕ произошло, и съесть им
+                        # одноразовый bare-reset #3251 значило бы остаться и без
+                        # нового IP, и без сброса вообще.
+                        if await _rotate_current_proxy(
+                            db, run_id, browser_fetcher, trigger="rotate-on-ban"
+                        ):
+                            context_reset_used = True
+
+                    if not context_reset_used:
+                        context_reset_used = True
+                        browser_fetcher.request_context_reset()
                 logger.warning(
                     "avito_detail_backfill: run_id=%d BLOCKED #%d/%d (consecutive=%d): %s",
                     run_id,

tradein-mvp/backend/tests/test_3251_avito_warm_context.py

@@ -73,6 +73,13 @@ def _fake_settings(**overrides: object) -> MagicMock:
         # TypeError: '>=' not supported between 'int' and 'MagicMock'. Та же
         # ловушка, что уже описана здесь для detail_backfill_block_ratio_window.
         "avito_detail_backfill_rotate_after_attempts": 15,
+        # #3283g: та же ловушка -- bare MagicMock отдаёт child-MagicMock на
+        # `ban_budget > 0` / `attempts_since_rotation < settings.avito_detail_backfill_
+        # rotate_on_ban_min_gap` -> TypeError. Значения = реальные дефолты config.py,
+        # чтобы существующие 3251-сценарии (<10 попыток в прогоне) не задевали
+        # rotate-on-ban веткой -- она молчаливо остаётся в "gap не выдержан".
+        "avito_detail_backfill_rotate_on_ban_max": 2,
+        "avito_detail_backfill_rotate_on_ban_min_gap": 10,
     }
     defaults.update(overrides)
     return MagicMock(**defaults)

tradein-mvp/backend/tests/test_3283g_rotate_on_platform_ban.py

@@ -0,0 +1,305 @@
+"""Ротация exit-IP НА САМ БАН площадки в avito_detail_backfill (#3283g).
+
+Бан привязан к IP (замерено вживую на проде: rotate_proxy() лечит забаненный
+узел за секунды, снимает запись из scrape_proxy_source_bans), но #3251/#3212
+запрещают сбрасывать browser-context на КАЖДЫЙ блок -- голый сброс (без смены
+IP) выбрасывает пройденный QRATOR-PoW и запускает самоподдерживающийся каскад
+блоков на ТОМ ЖЕ адресе. rotate-on-ban обходит этот запрет: она меняет IP
+ВМЕСТЕ со сбросом (через существующий _rotate_current_proxy, #3298), поэтому
+безопасна повторно -- но бюджетно (rotate_on_ban_max за прогон) и с gap'ом
+(rotate_on_ban_min_gap попыток от последней ротации любого происхождения).
+
+Зеркалит helpers tests/test_3251_avito_warm_context.py.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+
+import pytest  # noqa: E402
+from scraper_kit.avito_exceptions import AvitoBlockedError  # noqa: E402
+
+from app.core import shutdown as _sd  # noqa: E402
+from app.tasks.avito_detail_backfill import run_avito_detail_backfill  # noqa: E402
+
+_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
+_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
+_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
+_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
+_SETTINGS = "app.tasks.avito_detail_backfill.settings"
+_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
+_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
+_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"
+_ROTATE_PROXY = "app.tasks.avito_detail_backfill.rotate_proxy"
+
+
+@pytest.fixture(autouse=True)
+def _reset_shutdown() -> None:
+    _sd.reset_shutdown()
+    yield
+    _sd.reset_shutdown()
+
+
+def _fake_settings(**overrides: object) -> MagicMock:
+    defaults: dict[str, object] = {
+        "scraper_fetch_mode": "browser",
+        "avito_detail_backfill_use_curl": False,
+        "detail_backfill_block_ratio_window": 20,
+        "detail_backfill_block_ratio_threshold": 0.7,
+        "browser_http_endpoint": "http://browser:9000",
+        "avito_detail_backfill_rotate_after_attempts": 15,
+        "avito_detail_backfill_rotate_on_ban_max": 2,
+        "avito_detail_backfill_rotate_on_ban_min_gap": 10,
+    }
+    defaults.update(overrides)
+    return MagicMock(**defaults)
+
+
+def _make_snapshot(n: int) -> list[dict]:
+    return [
+        {
+            "id": i + 1,
+            "source_url": f"https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_{i + 1}",
+        }
+        for i in range(n)
+    ]
+
+
+def _mock_db(snapshot: list[dict]) -> MagicMock:
+    db = MagicMock()
+    sel = MagicMock()
+    sel.mappings.return_value.all.return_value = snapshot
+    db.execute.return_value = sel
+    return db
+
+
+def _mock_browser_fetcher_cls(lease_id: int = 42) -> tuple[MagicMock, AsyncMock]:
+    instance = AsyncMock()
+    instance.__aenter__ = AsyncMock(return_value=instance)
+    instance.__aexit__ = AsyncMock(return_value=False)
+    instance.request_context_reset = MagicMock()
+    instance.lease_id = lease_id
+    return MagicMock(return_value=instance), instance
+
+
+def _mock_rotate_proxy_ok() -> AsyncMock:
+    result = MagicMock(ok=True, reason=None, new_ip="178.176.79.156", reconnect_delay_s=0.0)
+    return AsyncMock(return_value=result)
+
+
+def _mock_rotate_proxy_fail() -> AsyncMock:
+    """Тихий отказ rotate_proxy: исчерпан суточный лимит / нет rotate_url / сеть.
+
+    Именно тихий (ok=False + reason), а не исключение — так эта функция и устроена.
+    """
+    result = MagicMock(ok=False, reason="daily rotation limit reached", new_ip=None,
+                       reconnect_delay_s=None)
+    return AsyncMock(return_value=result)
+
+
+@pytest.mark.asyncio
+async def test_rotate_on_ban_max_zero_is_full_noop() -> None:
+    """rotate_on_ban_max=0 -> rotate_proxy никогда не вызывается, поведение как до #3283g."""
+    blocked_exc = AvitoBlockedError("firewall/soft-block")
+    snapshot = _make_snapshot(3)
+    db = _mock_db(snapshot)
+    mock_fetch = AsyncMock(side_effect=blocked_exc)
+    fake_settings = _fake_settings(avito_detail_backfill_rotate_on_ban_max=0)
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    mock_rotate_proxy = _mock_rotate_proxy_ok()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, MagicMock()),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_ROTATE_PROXY, mock_rotate_proxy),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=301,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
+        )
+
+    assert result.blocked == 3
+    mock_rotate_proxy.assert_not_called()
+    # 0 = no-op -> старый bare-reset механизм (#3251) работает как раньше, ровно 1 раз.
+    assert instance.request_context_reset.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_rotate_on_ban_budget_exhausted_logs_and_continues() -> None:
+    """Бюджет исчерпан после N ротаций -> дальше блоки идут БЕЗ ротации, но с логом."""
+    blocked_exc = AvitoBlockedError("firewall/soft-block")
+    snapshot = _make_snapshot(5)
+    db = _mock_db(snapshot)
+    mock_fetch = AsyncMock(side_effect=blocked_exc)
+    fake_settings = _fake_settings(
+        avito_detail_backfill_rotate_on_ban_max=2,
+        avito_detail_backfill_rotate_on_ban_min_gap=0,
+    )
+    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
+    mock_rotate_proxy = _mock_rotate_proxy_ok()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, MagicMock()),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_ROTATE_PROXY, mock_rotate_proxy),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=302,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 10},
+        )
+
+    assert result.blocked == 5
+    # Бюджет = 2 ротации за прогон -- дальше 3 блока идут молча (в смысле IP), но
+    # исчерпание квоты должно быть видно в логе, а не тихим отказом.
+    assert mock_rotate_proxy.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_rotate_on_ban_min_gap_skips_when_too_early() -> None:
+    """Большой min_gap при коротком прогоне -> ротация ни разу не успевает выстрелить.
+
+    Бюджет ЕСТЬ (max=5), но gap (10 попыток) не набирается за 3 блока подряд --
+    поведение сводится к старому bare context reset (#3251), ровно 1 раз.
+    """
+    blocked_exc = AvitoBlockedError("firewall/soft-block")
+    snapshot = _make_snapshot(10)
+    db = _mock_db(snapshot)
+    mock_fetch = AsyncMock(side_effect=blocked_exc)
+    fake_settings = _fake_settings(
+        avito_detail_backfill_rotate_on_ban_max=5,
+        avito_detail_backfill_rotate_on_ban_min_gap=10,
+    )
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    mock_rotate_proxy = _mock_rotate_proxy_ok()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, MagicMock()),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_ROTATE_PROXY, mock_rotate_proxy),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=303,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
+        )
+
+    assert result.blocked == 3
+    mock_rotate_proxy.assert_not_called()
+    assert instance.request_context_reset.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_successful_ban_rotation_resets_attempts_since_rotation() -> None:
+    """Успешная ротация по бану обнуляет attempts_since_rotation -- проверка через
+
+    взаимодействие с #3298 (rotate_after_attempts). Первая карточка -- блок ->
+    ban-rotation (budget=1, gap=0), обнуляет счётчик попыток. Следующие 5 карточек
+    успешны; rotate_after_attempts=3 должен сработать ПОСЛЕ 3 успешных попыток
+    ПОСЛЕ сброса (т.е. на 4-й карточке по счёту в снапшоте, 3-й успешной), а НЕ
+    раньше -- если бы reset не произошёл, порог набрался бы на карточке раньше.
+    """
+    blocked_exc = AvitoBlockedError("firewall/soft-block")
+    snapshot = _make_snapshot(6)
+    db = _mock_db(snapshot)
+    mock_fetch = AsyncMock(
+        side_effect=[blocked_exc, MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()]
+    )
+    fake_settings = _fake_settings(
+        avito_detail_backfill_rotate_after_attempts=3,
+        avito_detail_backfill_rotate_on_ban_max=1,
+        avito_detail_backfill_rotate_on_ban_min_gap=0,
+    )
+    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
+    mock_rotate_proxy = _mock_rotate_proxy_ok()
+    rotate_calls_at_fetch_count: list[int] = []
+
+    async def _record_rotate(*_args: object, **_kwargs: object) -> MagicMock:
+        rotate_calls_at_fetch_count.append(mock_fetch.call_count)
+        return MagicMock(ok=True, reason=None, new_ip="1.2.3.4", reconnect_delay_s=0.0)
+
+    mock_rotate_proxy.side_effect = _record_rotate
+
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, MagicMock()),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, return_value=True),
+        patch(_ROTATE_PROXY, mock_rotate_proxy),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=304,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 10},
+        )
+
+    assert result.blocked == 1
+    # ban-rotation срабатывает сразу на 1-й (заблокированной) карточке; следующая
+    # (attempts-based) -- только через 3 успешные попытки ПОСЛЕ сброса, т.е. на
+    # 4-й карточке снапшота. Если бы attempts_since_rotation не обнулялся, вторая
+    # ротация выстрелила бы на карточке 3, а не 4.
+    assert rotate_calls_at_fetch_count == [1, 4]
+
+
+@pytest.mark.asyncio
+async def test_failed_ban_rotation_falls_back_to_bare_context_reset() -> None:
+    """Отказавшая ротация не должна съедать одноразовый bare-reset (#3251).
+
+    rotate_proxy отдаёт ok=False ТИХО и сброса контекста внутри при этом НЕ делает.
+    Если бы мы всё равно считали context_reset_used израсходованным, прогон остался
+    бы и без нового IP, и без сброса вообще — то есть строго хуже, чем до #3283g.
+    """
+    blocked_exc = AvitoBlockedError("firewall/soft-block")
+    snapshot = _make_snapshot(3)
+    db = _mock_db(snapshot)
+    mock_fetch = AsyncMock(side_effect=blocked_exc)
+    fake_settings = _fake_settings(
+        avito_detail_backfill_rotate_on_ban_max=2,
+        avito_detail_backfill_rotate_on_ban_min_gap=0,
+    )
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    mock_rotate_proxy = _mock_rotate_proxy_fail()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, MagicMock()),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_ROTATE_PROXY, mock_rotate_proxy),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=305,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
+        )
+
+    assert result.blocked == 3
+    assert mock_rotate_proxy.called, "ротацию пробовали"
+    # Ротация провалилась -> запасной механизм отработал ровно один раз за прогон.
+    assert instance.request_context_reset.call_count == 1