fix(tradein/avito): окно ретроспективы под недельный такт + диагноз полного обхода (#2674)

Окно и такт разъехались. Каденс avito_full_load задаёт default_params.interval_days,
глубину обхода — incremental_days (since = today - N, ранняя остановка по listing_date).
Это два независимых литерала, обязанных совпадать: 129 поставил окно 2 при ежедневном
такте (перекрытие было), 206 перевёл такт на 7 суток и окно не тронул. Прогон видит
[D-2, D] = 3 суток из 7; дни D+1..D+4 не попадают ни в один прогон.

Числа с прода (tradein, read-only 2026-08-06). 2026-06-21 — единственный день, когда
оба обхода отработали: инкрементальный run 297 — 2804 unique, exhaustive run 295 —
9992 unique, то есть окно в 2 суток достаёт 28.1% инвентаря. Когорта run 295, не
виденная после 23.06 (listing_date заморожен): полоса [D-2, D] — 189 лотов, полоса
[D-7, D-3] — ещё 427, расширение окна берёт в 3.26 раза больше. Гистограмма
(last_seen_at::date - listing_date) за 20 суток: возраст 0-2 — 594, возраст ровно 7 —
1212 (51% датированных наблюдений) — у Avito недельный авто-подъём, sortTimeStamp
сдвигается кратно 7. Структурный минимум бездырочного покрытия — 6, но 6 режет ровно
по этому пику; 7 = такт, полосы соседних прогонов смыкаются с суточным перехлёстом
под дрейф расписания (замер: last_run 03.08 13:37 -> next_run_at 10.08 14:16).

Цена: страниц примерно втрое больше на прогон, но прогон недельный. До 206 система
платила ~150-370 страниц семь раз в неделю; после правки — ~500-1200 в неделю, всё
ещё примерно вдвое дешевле, чем до 206.

Чиню в двух местах: миграция 215 выводит окно из фактического interval_days строки
(GREATEST — не сужает окно шире такта), scheduler расширяет его на лету и пишет
warning, чтобы расхождение не вернулось следующей правкой каденса.

Полный обход: причина не в площадке. Все пять banned-прогонов
avito_full_load_exhaustive (05.07-02.08) несут один текст —
"browser-sidecar error: browser unavailable (proxy may be down)", то есть 503 от
своего же сайдкара, у которого не поднялся камуфокс. В те же дни avito_city_sweep
(20 done), avito_newbuilding_sweep (21) и avito_detail_backfill (63) работали.
Корень — мёртвый BROWSER_PROXY_AVITO (ard.mobileproxy.space) в env сайдкара, куда
полный обход проваливался, потому что строил свой BrowserFetcher без пула; починено
не здесь, а #2637 (02.08, пул для браузерного пути Авито) и #2616 шаг 2 (05.08,
снос мёртвых env). Прод подтверждает: 0 лотов 05/12/19/26.07, 2334 и 362 в двух
прогонах после 02.08.

Остаток, который чинится кодом, здесь: 503 сайдкара классифицировался как soft-ban и
уходил в бюджет IP-ротации, а ротация снята (#2616 шаг 2, max_rot=0) — условие
rot_done < max_rot ложно всегда, а бюджет коротких backoff-retry стоял в else и был
для soft-ban недостижим. Ни одного ретрая на самую частую ошибку: один блип сайдкара
стоил бакета, четыре подряд — всего прогона. Бюджет backoff теперь общий для обеих
причин; сайдкар на таком 503 сам поднимает фоновый retry launch'а, поэтому повтор
через пару секунд обычно проходит.

Статус banned на такой ошибке остаётся ложью (площадка не банила) — это же чтение
легло в основание 206. Здесь не трогаю: набор status ограничен CHECK-констрейнтом,
а mark_banned в отличие от mark_failed сохраняет чекпоинт done_buckets.

tradein-mvp/backend/data/sql/215_avito_full_load_window_matches_cadence.sql

@@ -0,0 +1,74 @@
+-- 215_avito_full_load_window_matches_cadence.sql
+-- Окно ретроспективы avito_full_load приводится в согласие с тактом (#2674).
+--
+-- КОНТЕКСТ. Каденс источника задаётся ключом default_params."interval_days"
+-- (scheduler.compute_next_run_at), глубина обхода — ключом "incremental_days"
+-- (run_avito_full_load: since = today - incremental_days, дальше пагинация
+-- newest-first с ранней остановкой на первой странице без карточек >= since).
+-- Это ДВА независимых литерала, которые обязаны совпадать — и они разъехались:
+--   129_avito_full_load_incremental_split.sql поставил incremental_days=2 при
+--   тогдашнем ежедневном такте (2 >= 1 — окно перекрывало такт, дыр не было);
+--   206_scrape_schedules_cut_wasteful_load.sql перевёл источник на interval_days=7,
+--   но incremental_days не тронул.
+-- Итог на проде (scrape_schedules id=138, проверено read-only 2026-08-06):
+--   {"interval_days": 7, "incremental_days": 2, ...}
+-- Прогон видит listing_date в [D-2, D] = 3 календарных дня из 7; следующий прогон
+-- начинает с [D+5, D+7]. Дни D+1..D+4 не попадают НИ в один прогон — 4 суток из 7
+-- (57%) структурно вне поля зрения источника.
+--
+-- ЧИСЛА, обосновывающие новое значение (прод, tradein):
+--   * 2026-06-21 — единственный день, когда оба Avito-обхода отработали успешно:
+--     инкрементальный (окно 2) run 297 — 2804 unique, exhaustive (без отсечки)
+--     run 295 — 9992 unique. Окно в 2 суток достаёт 28.1% того, что достаёт
+--     полный обход; 7188 лотов (71.9%) лежат ниже отсечки.
+--   * Когорта run 295, не виденная после 2026-06-23 (listing_date заморожен):
+--     в полосе [D-2, D] — 189 лотов, в полосе [D-7, D-3] — ещё 427. Расширение
+--     окна 2 -> 7 берёт в 3.26 раза больше лотов.
+--   * Гистограмма (last_seen_at::date - listing_date) по avito за последние 20
+--     суток: возраст 0-2 — 594 лота, возраст ровно 7 — 1212 лотов (51% всех
+--     датированных наблюдений). У Avito недельный авто-подъём: sortTimeStamp
+--     сдвигается кратно 7 суткам, поэтому на возрасте ровно 7 стоит пик.
+--     Структурный минимум бездырочного покрытия — 6 (при 6 полосы соседних
+--     прогонов смыкаются), но 6 режет ровно по этому пику. 7 = такт: полосы
+--     [D-7, D] и [D, D+7] смыкаются с однодневным перехлёстом, который
+--     покрывает дрейф расписания (замер: last_run 2026-08-03 13:37 ->
+--     next_run_at 2026-08-10 14:16 = +7 суток 39 минут за цикл).
+--
+-- ЦЕНА ПО ЗАПРОСАМ. Пагинация останавливается по глубине окна, поэтому число
+-- страниц растёт примерно как число лотов в окне: ~3.3x к прогону. Прогоны с
+-- окном 2 при delay=1.0: 2804 lots / 40 мин (run 297) .. 3989 lots / 99 мин
+-- (run 620) ≈ 150-370 страниц. Окно 7 -> ≈ 500-1200 страниц НА ПРОГОН, но
+-- прогон теперь недельный, а не ежедневный: до 206 система платила те же
+-- 150-370 страниц СЕМЬ раз в неделю (~1050-2600). После этой правки —
+-- ~500-1200 в неделю, то есть по-прежнему примерно вдвое дешевле, чем до 206.
+--
+-- Значение НЕ хардкодим числом 7, а выводим из фактического interval_days строки:
+-- если такт когда-нибудь поменяют снова, повторный прогон файла (или ручной
+-- вызов) снова сведёт окно с тактом. Ту же инвариантность страхует код —
+-- scheduler._job_avito_full_load расширяет окно до такта и пишет warning.
+-- GREATEST с текущим значением: миграция никогда не СУЖАЕТ окно, если оператор
+-- выставил его шире такта осознанно.
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)),
+--              129_avito_full_load_incremental_split.sql (incremental_days),
+--              206_scrape_schedules_cut_wasteful_load.sql (interval_days=7).
+-- ТОЛЬКО данные (UPDATE default_params), DDL нет — idempotent by construction.
+-- source сравнивается на ТОЧНОЕ равенство: 'avito_full_load_exhaustive' — соседняя
+-- джоба, у неё incremental_days нет вовсе (полный обход) и трогать её нельзя.
+
+BEGIN;
+
+UPDATE scrape_schedules
+SET default_params = default_params
+        || jsonb_build_object(
+            'incremental_days',
+            GREATEST(
+                COALESCE(CAST(default_params ->> 'interval_days' AS integer), 1),
+                COALESCE(CAST(default_params ->> 'incremental_days' AS integer), 1)
+            )
+        ),
+    updated_at = NOW()
+WHERE source = 'avito_full_load'
+  AND default_params ? 'incremental_days';
+
+COMMIT;

tradein-mvp/backend/tests/test_2674_avito_full_load_coverage.py

@@ -0,0 +1,220 @@
+"""#2674, Avito full load: окно ретроспективы под недельный такт + 503 сайдкара.
+
+Две находки эпика, обе — продолжающаяся потеря данных, а не история.
+
+1. Окно разошлось с тактом. `interval_days` (каденс, миграция 206) и
+   `incremental_days` (глубина обхода, миграция 129) — два независимых литерала,
+   которые обязаны совпадать. На проде было {interval_days: 7, incremental_days: 2}:
+   прогон видит listing_date в [D-2, D] = 3 суток из 7, дни D+1..D+4 не попадают
+   ни в один прогон. Чиним в двух местах — данными (миграция 215) и кодом
+   (scheduler расширяет окно до такта, чтобы расхождение не вернулось).
+
+2. 503 сайдкара не ретраился ни разу. `_fetch_serp_html_browser` классифицировал
+   "browser unavailable (proxy may be down)" как soft-ban и отправлял в бюджет
+   IP-ротации, а ротация снята в #2616 (max_rot=0, `_rotate_ip()` всегда False) →
+   условие `rot_done < max_rot` ложно всегда. Бюджет коротких backoff-retry стоял
+   в `else:` и для soft-ban был структурно недостижим. Итог на проде: один блип
+   сайдкара съедал бакет, четыре подряд — весь прогон
+   (`_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED = 4`).
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+from scraper_kit.avito_exceptions import AvitoRateLimitedError
+from scraper_kit.orchestration import scheduler as kit_sched
+from scraper_kit.orchestration.scheduler import (
+    SchedulerContext,
+    _job_avito_full_load,
+    _job_avito_full_load_exhaustive,
+)
+from scraper_kit.providers.avito import serp as serp_module
+from scraper_kit.providers.avito.serp import _AVITO_SIDECAR_TRANSIENT_RETRIES, AvitoScraper
+
+from app.services.scraper_adapters import RealScraperConfig
+
+_MIGRATION_215 = (
+    Path(__file__).resolve().parents[1]
+    / "data"
+    / "sql"
+    / "215_avito_full_load_window_matches_cadence.sql"
+)
+
+
+def _ctx() -> SchedulerContext:
+    return SchedulerContext(
+        config=MagicMock(),
+        matcher=MagicMock(),
+        enrichment=MagicMock(),
+        session_factory=MagicMock(),
+        runs=MagicMock(),
+    )
+
+
+async def _incremental_days_passed(params: dict) -> object:
+    """Прогнать _job_avito_full_load с params и вернуть переданный incremental_days."""
+    with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
+        await _job_avito_full_load(MagicMock(), 1, params, _ctx())
+    mock_run.assert_awaited_once()
+    _args, kwargs = mock_run.call_args
+    return kwargs.get("incremental_days")
+
+
+# ── 1. Окно ретроспективы не уже такта ────────────────────────────────────────
+
+
+async def test_window_widened_to_cadence() -> None:
+    """Прод-конфиг на момент находки: такт 7 суток, окно 2 → окно расширяется до 7.
+
+    Фальсификация: без правки в scheduler._job_avito_full_load сюда приезжает 2 —
+    ровно те «двое суток из семи», о которых говорит эпик.
+    """
+    assert await _incremental_days_passed({"interval_days": 7, "incremental_days": 2}) == 7
+
+
+async def test_window_wider_than_cadence_left_alone() -> None:
+    """Окно ШИРЕ такта — осознанный запас оператора, не сужаем."""
+    assert await _incremental_days_passed({"interval_days": 3, "incremental_days": 10}) == 10
+
+
+async def test_daily_cadence_keeps_window() -> None:
+    """Back-compat: без interval_days такт = 1 сутки, окно 2 уже перекрывает его."""
+    assert await _incremental_days_passed({"incremental_days": 2}) == 2
+
+
+async def test_no_window_stays_exhaustive() -> None:
+    """Строка без incremental_days = полный обход; такт не должен её «инкрементализировать»."""
+    assert await _incremental_days_passed({"interval_days": 7}) is None
+
+
+async def test_exhaustive_job_ignores_window_params() -> None:
+    """Соседняя джоба всегда идёт полным обходом, что бы ни лежало в её params."""
+    with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
+        await _job_avito_full_load_exhaustive(
+            MagicMock(), 1, {"interval_days": 7, "incremental_days": 2}, _ctx()
+        )
+    _args, kwargs = mock_run.call_args
+    assert kwargs.get("incremental_days") is None
+
+
+# ── 2. 503 сайдкара получает backoff-retry, а не мгновенный отказ ─────────────
+
+
+def _sidecar_503() -> httpx.HTTPStatusError:
+    request = httpx.Request("POST", "http://tradein-browser:3000/fetch")
+    response = httpx.Response(
+        503,
+        json={"error": "browser unavailable (proxy may be down)"},
+        request=request,
+    )
+    return httpx.HTTPStatusError("503", request=request, response=response)
+
+
+@pytest.mark.asyncio
+async def test_sidecar_503_retried_before_giving_up() -> None:
+    """503 «browser unavailable» — это упавший launch камуфокса, а не бан площадки.
+
+    Сайдкар на таком отказе сам поднимает фоновый retry launch'а, поэтому повтор
+    через пару секунд обычно проходит. Ждём 1 попытку + весь бюджет backoff-retry.
+
+    Фальсификация: без правки soft-ban уходит в ветку ротации (max_rot=0 → условие
+    ложно всегда), бюджет backoff недостижим → ровно 1 вызов fetch и немедленный
+    AvitoRateLimitedError.
+    """
+    scraper = AvitoScraper(RealScraperConfig())
+    assert scraper._cffi is None
+    scraper._browser = AsyncMock()
+    scraper._browser.fetch = AsyncMock(side_effect=_sidecar_503())
+
+    with patch.object(serp_module.asyncio, "sleep", AsyncMock()):
+        with pytest.raises(AvitoRateLimitedError):
+            await scraper._fetch_serp_html(
+                "https://www.avito.ru/ekaterinburg/kvartiry/prodam", page=1
+            )
+
+    assert scraper._browser.fetch.await_count == 1 + _AVITO_SIDECAR_TRANSIENT_RETRIES
+
+
+@pytest.mark.asyncio
+async def test_sidecar_503_recovers_without_aborting_bucket() -> None:
+    """Один блип сайдкара больше не стоит бакета: вторая попытка отдаёт HTML."""
+    scraper = AvitoScraper(RealScraperConfig())
+    scraper._browser = AsyncMock()
+    scraper._browser.fetch = AsyncMock(
+        side_effect=[_sidecar_503(), "<html><body>serp</body></html>"]
+    )
+
+    with patch.object(serp_module.asyncio, "sleep", AsyncMock()):
+        html = await scraper._fetch_serp_html(
+            "https://www.avito.ru/ekaterinburg/kvartiry/prodam", page=1
+        )
+
+    assert html == "<html><body>serp</body></html>"
+    assert scraper._browser.fetch.await_count == 2
+
+
+# ── 3. Миграция 215: статические инварианты ──────────────────────────────────
+# Живой БД в тестах нет (см. conftest.py) — эффект UPDATE'а проверяется на проде,
+# здесь фиксируем форму файла: транзакционность, отсутствие DDL, точное совпадение
+# source (соседняя джоба с похожим именем ловится подстрокой), вывод значения из
+# фактического interval_days вместо литерала.
+
+
+def _sql() -> str:
+    return _MIGRATION_215.read_text(encoding="utf-8")
+
+
+def _executable_sql() -> str:
+    """SQL без построчных `--`-комментариев — только исполняемый код."""
+    lines = []
+    for raw in _sql().splitlines():
+        code = raw.split("--", 1)[0]
+        if code.strip():
+            lines.append(code)
+    return "\n".join(lines)
+
+
+def _flat() -> str:
+    return re.sub(r"\s+", " ", _executable_sql()).strip().lower()
+
+
+def test_migration_215_is_transactional_and_data_only() -> None:
+    flat = _flat()
+    assert flat.startswith("begin;")
+    assert flat.endswith("commit;")
+    for ddl in ("create ", "alter ", "drop ", "truncate "):
+        assert ddl not in flat, f"миграция только про данные, найден DDL: {ddl!r}"
+
+
+def test_migration_215_targets_only_avito_full_load() -> None:
+    """Точное равенство source, НЕ подстрока — 'avito_full_load_exhaustive' рядом."""
+    flat = _flat()
+    assert "where source = 'avito_full_load'" in flat
+    assert "avito_full_load_exhaustive" not in _executable_sql()
+    assert " like " not in flat
+
+
+def test_migration_215_derives_window_from_cadence() -> None:
+    """Окно выводится из interval_days строки, а не хардкодится числом."""
+    flat = _flat()
+    assert "interval_days" in flat
+    assert "greatest(" in flat
+    assert re.search(r"'incremental_days',\s*greatest", flat) is not None
+
+
+def test_migration_215_merges_params_not_overwrites() -> None:
+    """`||` мерджит ключ в default_params — соседние ключи обязаны выжить."""
+    flat = _flat()
+    assert "default_params ||" in flat or "default_params\n||" in _executable_sql().lower()
+    assert "set default_params = jsonb_build_object" not in flat
+
+
+def test_migration_215_has_no_psycopg_cast_trap() -> None:
+    """Repo-конвенция: CAST(x AS type), не x::type."""
+    assert "::" not in _executable_sql()
+    assert "cast(" in _flat()

tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py

@@ -519,6 +519,25 @@ async def _job_avito_full_load(
 ) -> None:
     _incremental_days = params.get("incremental_days")
     incremental_days = int(_incremental_days) if _incremental_days is not None else None
+    # Окно ретроспективы не может быть уже такта (#2674). incremental_days задаёт
+    # since=today-N, а прогон повторяется раз в interval_days суток → при N < interval_days
+    # объявления, поднятые в дни (D, D+interval_days-N), не попадают НИ в один прогон.
+    # Ровно это и случилось: миграция 206 перевела источник на interval_days=7, оставив
+    # incremental_days=2 из миграции 129 → 3 календарных дня из 7 в поле зрения.
+    # Два независимых литерала, которые обязаны совпадать, однажды уже разъехались —
+    # поэтому расхождение чиним здесь, а не только данными.
+    interval_days = max(1, int(params.get("interval_days", 1)))
+    if incremental_days is not None and incremental_days < interval_days:
+        logger.warning(
+            "avito_full_load: окно ретроспективы incremental_days=%d уже такта "
+            "interval_days=%d — расширяю до %d, иначе %d суток объявлений не видит "
+            "ни один прогон",
+            incremental_days,
+            interval_days,
+            interval_days,
+            interval_days - incremental_days,
+        )
+        incremental_days = interval_days
     await run_avito_full_load(
         db,
         run_id=run_id,

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py

@@ -505,9 +505,9 @@ class AvitoScraper(BaseScraper):
         # условным на avito_proxy_rotate_url, который больше не существует).
         max_rot = 0
         rot_done = 0
-        # Независимый бюджет коротких retry на transient sidecar-ошибки (timeout/
-        # crash) — НЕ тратит бюджет ротации. Каждая ветка ниже (ротация / transient)
-        # уменьшает свой счётчик → цикл всегда движется к условию исчерпания.
+        # Бюджет коротких backoff-retry на ЛЮБУЮ sidecar-ошибку (timeout/crash И
+        # soft-ban/503) — НЕ тратит бюджет ротации. Каждая ветка ниже (ротация /
+        # backoff) уменьшает свой счётчик → цикл всегда движется к исчерпанию.
         transient_left = _AVITO_SIDECAR_TRANSIENT_RETRIES
         while True:
             try:
@@ -541,36 +541,40 @@ class AvitoScraper(BaseScraper):
                     or "browser unavailable" in lowered
                     or "proxy may be down" in lowered
                 )
-                if is_soft_ban:
+                if is_soft_ban and rot_done < max_rot and await self._rotate_ip():
                     # Soft proxy rate-limit / прокси недоступен → свежий exit-IP.
-                    if rot_done < max_rot and await self._rotate_ip():
-                        rot_done += 1
-                        logger.info(
-                            "avito page=%d sidecar soft-ban (status=%d) — "
-                            "retry after rotation #%d: %s",
-                            page,
-                            status,
-                            rot_done,
-                            error_text[:200],
-                        )
-                        continue
-                else:
-                    # Transient (Page.goto Timeout / Target closed / прочие 5xx) →
-                    # короткий backoff-retry того же url БЕЗ ротации.
-                    if transient_left > 0:
-                        transient_left -= 1
-                        backoff = _AVITO_SIDECAR_TRANSIENT_BACKOFF_SEC + random.uniform(0, 0.5)
-                        logger.warning(
-                            "avito page=%d sidecar transient (status=%d) — "
-                            "retry in %.1fs (left=%d): %s",
-                            page,
-                            status,
-                            backoff,
-                            transient_left,
-                            error_text[:200],
-                        )
-                        await asyncio.sleep(backoff)
-                        continue
+                    rot_done += 1
+                    logger.info(
+                        "avito page=%d sidecar soft-ban (status=%d) — retry after rotation #%d: %s",
+                        page,
+                        status,
+                        rot_done,
+                        error_text[:200],
+                    )
+                    continue
+                # Backoff-retry того же url БЕЗ ротации. Раньше эта ветка стояла в
+                # `else:` к soft-ban и была для него НЕДОСТИЖИМА, а бюджет ротации с
+                # #2616 шага 2 равен нулю (max_rot=0, _rotate_ip() всегда False) —
+                # то есть 503 «browser unavailable (proxy may be down)» от сайдкара
+                # не ретраился ВООБЩЕ, ни разу (#2674). Сайдкар на таком 503 сам
+                # поднимает фоновый retry launch'а браузера, так что повтор через
+                # пару секунд обычно проходит; вместо этого один блип сайдкара
+                # съедал бакет, четыре подряд (_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED)
+                # — весь прогон. Теперь бюджет transient общий для обеих причин.
+                if transient_left > 0:
+                    transient_left -= 1
+                    backoff = _AVITO_SIDECAR_TRANSIENT_BACKOFF_SEC + random.uniform(0, 0.5)
+                    logger.warning(
+                        "avito page=%d sidecar %s (status=%d) — retry in %.1fs (left=%d): %s",
+                        page,
+                        "soft-ban" if is_soft_ban else "transient",
+                        status,
+                        backoff,
+                        transient_left,
+                        error_text[:200],
+                    )
+                    await asyncio.sleep(backoff)
+                    continue
                 # Оба бюджета (ротация + transient) исчерпаны → graceful soft-ban,
                 # чтобы run_avito_full_load/city_sweep ушли в mark_banned (partial
                 # preserved), а НЕ в generic except → mark_failed. Намеренно