From a8a21634720a4f0dd707c640a2c55ea26db9a843 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Mon, 10 Aug 2026 13:17:45 +0500 Subject: [PATCH] =?UTF-8?q?fix(tradein/scraper):=20=D1=81=D0=B2=D0=BE?= =?UTF-8?q?=D0=B4=D0=BA=D0=B0=20=C2=AB=D1=87=D1=82=D0=BE=20=D1=81=D0=B5?= =?UTF-8?q?=D0=B9=D1=87=D0=B0=D1=81=20=D0=BD=D0=B5=20=D1=81=D0=BE=D0=B1?= =?UTF-8?q?=D0=B8=D1=80=D0=B0=D0=B5=D1=82=C2=BB=20=E2=80=94=20=D0=BB=D0=B5?= =?UTF-8?q?=D1=81=D1=82=D0=BD=D0=B8=D1=86=D0=B0=20=D0=B2=D0=B5=D1=85=20?= =?UTF-8?q?=D0=BD=D0=B5=20=D0=B2=D0=B8=D0=B4=D0=B8=D1=82=20=D1=81=D1=82?= =?UTF-8?q?=D1=80=D0=B8=D0=BA=200=20(#2670)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #2720 вылечил «алерт ровно один раз за серию» (вехи 3, 6, 12, 24, 48…), но лестница шагает по подряд идущим завершённым failed/banned прогонам. Прод 2026-08-10: шесть источников не собирали дольше 3× своего такта, и трое худших лестнице недоступны — причём лишь один из трёх из-за редких вех: cian_history_backfill 42.1 сут, стрик 0 — прогонов нет с 30.06 вовсе avito_full_load_exhaustive 49.5 сут, стрик 0 — 5 банов обнулил 'cancelled' 09.08 avito_full_load 37.7 сут, стрик 31 — веха 48 при interval_days=7 → 119 сут Уплотнение вех чинит только третью строку: у первых двух стрик равен нулю, уплотнять нечего. Поэтому добавлена сводка, считающая КАЛЕНДАРНЫЙ возраст последнего успешного прогона по scrape_schedules: раз в сутки одно событие со списком источников, у которых он старше 3× их такта. Просрочка меряется в тактах, а не в сутках — 24 сут без сбора у 28-суточного rosreestr_quarter_poll норма, а у суточного domclick_city_sweep авария. Лестница не тронута: она отвечает на другой вопрос («что сломалось только что»). Заодно: шапка модуля утверждала, что прод крутит legacy app.services.scheduler. Эта ветка удалена в #2397 Part C — строка пережила собственную правду и посылала правку сторожей не в тот файл. --- .../tests/test_2670_stale_source_digest.py | 199 ++++++++++++++++++ .../scraper_kit/orchestration/scheduler.py | 150 ++++++++++++- 2 files changed, 347 insertions(+), 2 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_2670_stale_source_digest.py diff --git a/tradein-mvp/backend/tests/test_2670_stale_source_digest.py b/tradein-mvp/backend/tests/test_2670_stale_source_digest.py new file mode 100644 index 00000000..baf2ec4a --- /dev/null +++ b/tradein-mvp/backend/tests/test_2670_stale_source_digest.py @@ -0,0 +1,199 @@ +"""#2670 (остаток): лестница напоминаний не отвечает на вопрос «что сломано сейчас». + +#2720 вылечил «алерт ровно один раз за серию»: теперь вехи 3, 6, 12, 24, 48… Но лестница +шагает по ПОДРЯД ИДУЩИМ завершённым failed/banned прогонам, а на проде 2026-08-10 три +самых залежавшихся источника из шести просроченных ей недоступны — и лишь один из трёх +из-за редких вех: + + ┌────────────────────────────┬────────┬───────┬──────────────────────────────────────┐ + │ источник │ стрик │ сут. │ когда напомнит лестница │ + ├────────────────────────────┼────────┼───────┼──────────────────────────────────────┤ + │ avito_full_load_exhaustive │ 0 │ 49.5 │ никогда: 5 банов обнулил 'cancelled' │ + │ cian_history_backfill │ 0 │ 42.1 │ никогда: прогонов нет с 30.06 │ + │ avito_full_load │ 31 │ 37.7 │ веха 48 → +17 прогонов × 7 сут = 119 │ + │ avito_detail_backfill │ 5 │ 5.2 │ веха 6 → завтра │ + │ domclick_city_sweep │ 5 │ 5.1 │ веха 6 → завтра │ + │ domclick_detail_backfill │ 4 │ 5.0 │ веха 6 → послезавтра │ + └────────────────────────────┴────────┴───────┴──────────────────────────────────────┘ + +Уплотнение вех (3,4,5,6…) чинит ТОЛЬКО третью строку: у первых двух стрик равен нулю, +уплотнять нечего — «замолчал» там означает «перестал производить прогоны», а не «серия +длиннее последней вехи». Поэтому остаток задачи закрывает сводка, считающая КАЛЕНДАРНЫЙ +возраст последнего успеха, а не длину серии. + +Фальсификация: на коде до этой правки `emit_stale_digest`/`stale_sources` не существует +(ImportError на сборе тестов) — сводки нет ни в каком виде. Тест +`test_ladder_is_silent_for_the_worst_two` — КОНТРОЛЬ: он зелёный и до, и после правки и +показывает ровно то, чего сводка не заменяет, а добавляет: лестница на этих двух молчит. +""" + +from __future__ import annotations + +import os +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.orchestration import runs as kit_runs +from scraper_kit.orchestration import scheduler as sched + +NOW = datetime(2026, 8, 10, 8, 0, tzinfo=UTC) + + +def _row(source: str, interval_days: Any, age_days: float, never_ok: bool = False) -> Any: + """Строка `_STALE_SOURCES_SQL`: last_ok уже схлопнут в `since` через COALESCE.""" + return SimpleNamespace( + source=source, + interval_days=interval_days, + since=NOW - timedelta(days=age_days), + never_ok=never_ok, + ) + + +# Снимок прода 2026-08-10 08:00 UTC: все 52 включённых расписания не влезают, взяты все +# просроченные + четыре контрольных, каждое из которых мимо порога по своей причине. +PROD_ROWS = [ + _row("cian_history_backfill", None, 42.1), # такт по умолчанию (daily) + _row("avito_full_load_exhaustive", 7, 49.5), + _row("avito_full_load", 7, 37.7), + _row("avito_detail_backfill", None, 5.2), + _row("domclick_city_sweep", None, 5.1), + _row("domclick_detail_backfill", None, 5.0), + # ── контроль: НЕ просрочены ── + _row("rosreestr_quarter_poll", 28, 24.0), # 24 сут при такте 28 — норма + _row("sber_index_pull", 7, 4.1), + _row("avito_city_sweep", None, 1.1), + _row("proxy_healthcheck", None, 0.02), +] + +# Порядок — по числу ПРОПУЩЕННЫХ ТАКТОВ (age/interval), а не по календарю: 42 суток +# у суточного backfill'а = 42 пропущенных такта, 49.5 у недельного = 7. +PROD_STALE = [ + "cian_history_backfill", # 42.1 / 1 + "avito_full_load_exhaustive", # 49.5 / 7 = 7.07 + "avito_full_load", # 37.7 / 7 = 5.39 + "avito_detail_backfill", # 5.2 / 1 + "domclick_city_sweep", # 5.1 / 1 + "domclick_detail_backfill", # 5.0 / 1 +] + + +@pytest.fixture(autouse=True) +def _reset_digest_clock() -> Any: + """Выпуск сводки помнится в памяти модуля — сбрасываем между тестами.""" + sched._last_stale_digest_at = None + yield + sched._last_stale_digest_at = None + + +def _db(rows: list[Any]) -> MagicMock: + db = MagicMock() + db.execute.return_value.fetchall.return_value = rows + return db + + +# ── 1. Чистая логика порога ────────────────────────────────────────────────── + + +def test_stale_sources_names_exactly_the_prod_six() -> None: + """Шесть просроченных из десяти, порядок — по числу пропущенных ТАКТОВ, не суток.""" + stale = sched.stale_sources(PROD_ROWS, NOW) + assert [s.source for s in stale] == PROD_STALE + + +def test_quarterly_source_is_not_stale_at_24_days() -> None: + """Порог считается в тактах: 24 сут для 28-суточного poll'а — не просрочка.""" + assert sched.stale_sources([_row("rosreestr_quarter_poll", 28, 24.0)], NOW) == [] + # …а 85 суток (>3×28) — уже просрочка. + assert [s.source for s in sched.stale_sources([_row("q", 28, 85.0)], NOW)] == ["q"] + + +@pytest.mark.parametrize("raw", [None, "null", "", "abc", 0, -5]) +def test_broken_interval_falls_back_to_daily(raw: Any) -> None: + """`interval_days: null` и мусор → такт 1 сут, как у compute_next_run_at.""" + assert sched._schedule_interval_days(raw) == 1 + + +def test_never_successful_source_is_reported_with_a_flag() -> None: + """Расписание без единого 'done' считается от created_at и помечается явно.""" + (only,) = sched.stale_sources([_row("brand_new", 1, 9.0, never_ok=True)], NOW) + assert only.never_ok is True + + +# ── 2. Выпуск сводки ───────────────────────────────────────────────────────── + + +def test_digest_emits_one_event_listing_all_stale_sources() -> None: + sentry = MagicMock() + with patch.object(sched, "sentry_sdk", sentry): + stale = sched.emit_stale_digest(_db(PROD_ROWS), now=NOW) + assert [s.source for s in stale] == PROD_STALE + sentry.capture_message.assert_called_once() + msg = sentry.capture_message.call_args[0][0] + assert msg.startswith("6 scraper sources are stale") + for name in PROD_STALE: + assert name in msg + assert "rosreestr_quarter_poll" not in msg + + +def test_digest_covers_the_two_sources_the_ladder_cannot_reach() -> None: + """Главное свойство: стрик 0 не мешает сводке — она меряет календарь, а не серию.""" + sentry = MagicMock() + with patch.object(sched, "sentry_sdk", sentry): + stale = sched.emit_stale_digest(_db(PROD_ROWS), now=NOW) + zero_streak = {"avito_full_load_exhaustive", "cian_history_backfill"} + assert zero_streak <= {s.source for s in stale} + + +def test_digest_is_quiet_when_everything_is_fresh() -> None: + sentry = MagicMock() + fresh = [_row("avito_city_sweep", None, 1.1), _row("sber_index_pull", 7, 4.1)] + with patch.object(sched, "sentry_sdk", sentry): + assert sched.emit_stale_digest(_db(fresh), now=NOW) == [] + sentry.capture_message.assert_not_called() + + +def test_digest_is_daily_not_per_tick() -> None: + """Планировщик тикает раз в минуту; сводка обязана выходить раз в сутки.""" + sentry = MagicMock() + db = _db(PROD_ROWS) + with patch.object(sched, "sentry_sdk", sentry): + sched.emit_stale_digest(db, now=NOW) + sched.emit_stale_digest(db, now=NOW + timedelta(minutes=1)) + sched.emit_stale_digest(db, now=NOW + timedelta(hours=23)) + assert sentry.capture_message.call_count == 1 + sched.emit_stale_digest(db, now=NOW + timedelta(hours=24, minutes=1)) + assert sentry.capture_message.call_count == 2 + + +def test_digest_failure_never_breaks_the_tick() -> None: + """Сводка — best-effort: упавший запрос не имеет права уронить тик планировщика.""" + db = MagicMock() + db.execute.side_effect = RuntimeError("db down") + with patch.object(sched, "sentry_sdk", MagicMock()): + assert sched.emit_stale_digest(db, now=NOW) == [] + + +# ── 3. Контроль: что именно сводка ДОБАВЛЯЕТ к лестнице ────────────────────── + + +@pytest.mark.parametrize( + ("name", "streak"), + [("avito_full_load_exhaustive", 0), ("cian_history_backfill", 0), ("avito_full_load", 31)], +) +def test_ladder_is_silent_for_the_worst_two(name: str, streak: int) -> None: + """КОНТРОЛЬ (зелёный и до правки): у трёх худших источников лестница молчит. + + Стрик 0 — прогонов нет / серию обнулил 'cancelled'; стрик 31 — между вехами 24 и 48. + """ + rows = [SimpleNamespace(status="banned") for _ in range(streak)] + rows += [SimpleNamespace(status="done") for _ in range(3)] + sentry = MagicMock() + with patch.object(kit_runs, "sentry_sdk", sentry): + kit_runs._alert_if_consecutive_failures(_db(rows), name) + sentry.capture_message.assert_not_called() diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py index 0ddd275e..06d778a8 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py @@ -17,8 +17,11 @@ Kit-native (sweep-оркестраторы, уже перенесённые в ` осталось в `app` (rosreestr_dkp / sber_index / deactivate_stale / *_backfill / …), инжектируются извне как `Handler` через `build_registry(product_handlers=...)`. -Боевой рантайм (scraper-контейнер) по-прежнему крутит старый `app.services.scheduler` — -это COPY, не MOVE. Переключение — отдельный поздний strangler-шаг. +Боевой рантайм (scraper-контейнер, `python -m app.scheduler_main`) крутит ИМЕННО ЭТОТ +loop: #2397 Part C удалил legacy-ветку `app.services.scheduler.scheduler_loop`, и +`_run_kit_scheduler` остался единственным путём. Строка «по-прежнему крутит старый +app.services.scheduler» жила здесь после того, как перестала быть правдой, и посылала +правку сторожей не в тот файл (#2670). Критичная concurrency-логика (`_claim_run` advisory-lock + double-check, `reap_zombies` порог, heartbeat, SIGTERM-drain) перенесена ДОСЛОВНО — тот же SQL, то же ветвление. @@ -36,6 +39,11 @@ from typing import TYPE_CHECKING, Any from sqlalchemy import text +try: # sentry опционален — kit standalone-импортируем, sentry-sdk не в его зависимостях + import sentry_sdk +except ImportError: # pragma: no cover - в проде backend-env sentry_sdk присутствует + sentry_sdk = None # type: ignore[assignment] + from scraper_kit.orchestration import runs as _kit_runs from scraper_kit.orchestration.pipeline import ( get_city_anchors, @@ -78,6 +86,141 @@ SKIP_CONCURRENT_CLAIM = "concurrent_claim" SKIP_RUNNING_UNDER_LOCK = "running_appeared_under_lock" SKIP_UNKNOWN_SOURCE = "unknown_source" +# ── сводка «что сейчас не собирает» (#2670, второй пункт задачи) ───────────── +# Лестница напоминаний из #2720 считает ПОДРЯД ИДУЩИЕ неудачные ПРОГОНЫ. Прод +# 2026-08-10: шесть источников не имели успешного прогона дольше 3× своего такта, и +# трое худших из них лестнице недоступны ПО ПОСТРОЕНИЮ, а не из-за редких вех: +# +# cian_history_backfill 42.1 сут без успеха, стрик 0 — с 30.06 прогонов нет +# вовсе (сегодняшний единственный — 'skipped', +# cian_cookies_expired), а лестница шагает только по +# завершённым failed/banned; +# avito_full_load_exhaustive 49.5 сут без успеха, стрик 0 — 5 банов подряд обнулил +# один 'cancelled' 09.08 (деплой убил бегущий прогон); +# avito_full_load 37.7 сут без успеха, стрик 31 — веха 48 при такте +# interval_days=7 наступит через 17 прогонов ≈ 119 суток. +# +# Уплотнение вех чинит только третий случай: у первых двух стрик равен нулю, уплотнять +# нечего. Поэтому сводка не «ещё один сторож помельче», а ЕДИНСТВЕННЫЙ ответ на вопрос +# «что сломано сейчас»: она считает КАЛЕНДАРНЫЙ возраст последнего успеха, поэтому +# видит и молчащий источник, и обнулённый стрик, и редкую веху. Лестница остаётся как +# была — она отвечает на другой вопрос («что сломалось только что») и стоит дёшево. +STALE_DIGEST_INTERVAL_FACTOR = 3 +STALE_DIGEST_PERIOD_H = 24 + +# Возраст последнего УСПЕШНОГО ('done') прогона на каждое включённое расписание. +# COALESCE(last_ok, created_at): у расписания без единого успеха отсчёт идёт от его +# создания — иначе «никогда не собирал» выглядело бы как «нет данных, судить нечем». +_STALE_SOURCES_SQL = text(""" + SELECT sch.source, + sch.default_params->>'interval_days' AS interval_days, + COALESCE( + (SELECT max(r.finished_at) FROM scrape_runs r + WHERE r.source = sch.source AND r.status = 'done'), + sch.created_at + ) AS since, + (NOT EXISTS (SELECT 1 FROM scrape_runs r + WHERE r.source = sch.source AND r.status = 'done')) AS never_ok + FROM scrape_schedules sch + WHERE sch.enabled +""") + +# ponytail: последний выпуск сводки помнится В ПАМЯТИ процесса, поэтому рестарт +# scheduler'а (деплой) даёт лишний выпуск. Осознанный размен: альтернатива — таблица +# состояния (миграция) ради анти-спама у механизма, который и заводится ПРОТИВ +# молчания. Понадобится точность — переносить в scrape_runs строкой своего source'а. +_last_stale_digest_at: datetime | None = None + + +@dataclass(frozen=True) +class StaleSource: + """Источник, не собиравший дольше STALE_DIGEST_INTERVAL_FACTOR× своего такта.""" + + source: str + interval_days: int + age_days: float + never_ok: bool + + +def _schedule_interval_days(raw: Any) -> int: + """default_params.interval_days → такт в сутках; всё непонятное → 1 (как у claim'а). + + Тот же дефолт, что у `compute_next_run_at` (interval_days=1 == daily): порог сводки + обязан считаться из ТОГО ЖЕ числа, которым расписание себя двигает, иначе «просрочен» + будет мерить не тот такт. `"interval_days": null` в jsonb приезжает сюда None. + """ + try: + return max(1, int(raw)) + except (TypeError, ValueError): + return 1 + + +def stale_sources(rows: list[Any], now: datetime) -> list[StaleSource]: + """Чистая часть сводки: какие расписания просрочены и на сколько (свежие — внизу). + + Просрочка меряется в ТАКТАХ, а не в сутках: у rosreestr_quarter_poll такт 28 суток, + и 24 суток без сбора для него норма, а для суточного domclick_city_sweep — авария. + Сортировка по числу пропущенных тактов, а не по календарю, по той же причине. + """ + stale: list[StaleSource] = [] + for row in rows: + interval = _schedule_interval_days(row.interval_days) + age_days = (now - row.since).total_seconds() / 86400.0 + if age_days > STALE_DIGEST_INTERVAL_FACTOR * interval: + stale.append( + StaleSource( + source=row.source, + interval_days=interval, + age_days=age_days, + never_ok=bool(row.never_ok), + ) + ) + stale.sort(key=lambda s: s.age_days / s.interval_days, reverse=True) + return stale + + +def emit_stale_digest(db: Session, *, now: datetime | None = None) -> list[StaleSource]: + """Раз в STALE_DIGEST_PERIOD_H часов — одно событие «что сейчас не собирает». + + Возвращает список просроченных источников (пустой — либо всё свежо, либо выпуск ещё + не подошёл по времени). Best-effort, как и оба сторожа в runs.py: сводка не имеет + права уронить тик планировщика. + """ + global _last_stale_digest_at + now = now or datetime.now(UTC) + if _last_stale_digest_at is not None and now - _last_stale_digest_at < timedelta( + hours=STALE_DIGEST_PERIOD_H + ): + return [] + try: + stale = stale_sources(list(db.execute(_STALE_SOURCES_SQL).fetchall()), now) + _last_stale_digest_at = now + if not stale: + logger.info("scheduler: stale-digest — просроченных источников нет (#2670)") + return [] + details = ", ".join( + f"{s.source} {s.age_days:.1f}d/{s.interval_days}d" + + (" (успеха не было ни разу)" if s.never_ok else "") + for s in stale + ) + logger.error( + "scheduler: %d источников не собирают дольше %d× своего такта — %s (#2670)", + len(stale), + STALE_DIGEST_INTERVAL_FACTOR, + details, + ) + if sentry_sdk is not None: + sentry_sdk.capture_message( + f"{len(stale)} scraper sources are stale (no successful run for more than " + f"{STALE_DIGEST_INTERVAL_FACTOR}× their schedule interval): {details}", + level="error", + ) + return stale + except Exception: + logger.exception("scheduler: stale-digest failed") + return [] + + # ── типы job/handler ───────────────────────────────────────────────────────── # Job получает свежую сессию (открыта `_dispatch`), run_id, params и весь контекст # (config/matcher/enrichment/runs) — чтобы иметь доступ к инжектированным зависимостям. @@ -759,6 +902,9 @@ async def scheduler_loop(ctx: SchedulerContext, registry: Mapping[str, Handler]) try: # Reap zombies first reap_zombies(db) + # #2670: раз в сутки — сводка «что сейчас не собирает». Календарная, а + # не по стрику: два самых залежавшихся источника прода имеют стрик 0. + emit_stale_digest(db) # Process due schedules due = get_due_schedules(db) for sch in due: