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

@@ -0,0 +1,85 @@
+-- 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):
+--   * Замер по ОДНОМУ прогону (run 2990, exhaustive 2026-08-02, 2317 строк
+--     снапшота / 2184 с датой) — сколько лотов достаёт окно шириной W суток:
+--       W=2 -> 147      W=6 -> 446      W=7 -> 1275     W=12 -> 1278
+--     Три вывода. Окно 2 теряет в 8.7 раза (147 против 1275). Шестёрка теряет
+--     две трети семёрки (446 против 1275) — это обрыв, а не экономия. И 7..12 —
+--     ПЛАТО: +3 лота на пять дополнительных суток окна, то есть семёрка стоит
+--     на самой дешёвой точке плато, а не является компромиссом.
+--   * Обрыв на 7 и плато за ним — свойство НЕ Avito, а КВАНТОВАНИЯ нашего же
+--     парсера относительных дат (_parse_relative_date, providers/avito/serp.py):
+--     «неделю назад» -> ровно today-7, «две недели назад» -> ровно today-14.
+--     Возрасты 8..13 по этому пути недостижимы — на проде их 3 лота из 2184
+--     (это и есть плато). Значит бакет «возраст 7» — не «поднятые ровно неделю
+--     назад», а ВСЁ, чему реально от 7 до 13 суток. W=6 режет не по пику
+--     распределения, а по ГРАНИЦЕ КВАНТОВАНИЯ и теряет бакет целиком.
+--   * Тот же бакет в общей выборке: (last_seen_at::date - listing_date) по
+--     avito за 20 суток — возраст 0-2 = 594, возраст ровно 7 = 1212 из 3714
+--     датированных наблюдений (32.6%).
+--   * Полосы соседних прогонов [D-7, D] и [D, D+7] смыкаются с суточным
+--     перехлёстом, который покрывает дрейф расписания (замер: last_run
+--     2026-08-03 13:37 -> next_run_at 2026-08-10 14:16 = +7 суток 39 минут).
+--
+-- ЦЕНА ПО ЗАПРОСАМ растёт НЕ пропорционально лотам. Стоимость бакета —
+-- ceil(свежих / 50) страниц с полом в 1-2 страницы; при окне 7 на бакет
+-- приходится ~15-20 свежих (1275 лотов на 77 бакетов = 7 комнатностей x 11
+-- ценовых seed-брекетов) — МЕНЬШЕ одной страницы. Большинство бакетов как
+-- стояло на 1-2 страницах, так и останется, глубже пойдут только плотные.
+-- Верхняя граница честная и продом уже пережитая: полный обход без отсечки
+-- вообще — 6 ч 59 мин (run 295) и 2 ч 34 мин (run 2990); окно 13-15 UTC
+-- ограничивает только СТАРТ прогона, не длительность.
+--
+-- ВОЗВРАЩАЕТЕ ЕЖЕДНЕВНЫЙ ТАКТ — ВЕРНИТЕ И ОКНО. Ни эта миграция (GREATEST
+-- только расширяет), ни планировщик (расширяет до такта, не сужает) окно НЕ
+-- сузят. interval_days 7 -> 1 при incremental_days = 7 даст ежедневный прогон
+-- с семисуточной глубиной, то есть восьмикратный охват КАЖДЫЙ день. Такт и
+-- окно менять одной правкой.
+--
+-- Значение НЕ хардкодим числом 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,225 @@
+"""#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_null_interval_days_does_not_crash() -> None:
+    """`"interval_days": null` в jsonb приезжает сюда как None, а int(None) — TypeError."""
+    assert await _incremental_days_passed({"interval_days": None, "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,28 @@ 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 в поле зрения.
+    # Два независимых литерала, которые обязаны совпадать, однажды уже разъехались —
+    # поэтому расхождение чиним здесь, а не только данными.
+    # None-safe так же, как incremental_days выше: `"interval_days": null` в jsonb
+    # приезжает сюда как None, а int(None) — TypeError.
+    _interval_days = params.get("interval_days")
+    interval_days = max(1, int(_interval_days)) if _interval_days is not None else 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. Намеренно