fix(tradein/scraper): сигнал живости из середины батча — живые прогоны перестают числиться зависшими (#2725)

reap_zombies меряет scrape_runs.heartbeat_at с порогом 6 ч, а cian_history_backfill
слал heartbeat ровно один раз — ДО батча. Прогон, работающий дольше шести часов,
помечался 'zombie' независимо от того, жив он или висит.

Замер на проде 2026-08-06: 6 прогонов cian_history_backfill в статусе 'zombie', у всех
шести сдвиг heartbeat 16-32 мс (единственный стартовый вызов) и финал ровно на
started_at + 6.00 ч. Живыми они были: внутри окна пятерых писались строки
offer_price_history с source='cian' (98/523/38/60/83 — плановый писатель этих строк
только этот батч), у прогона 304 последняя строка легла через 5.40 ч после старта.
Штатная длительность источника доходит до 5.06 ч (прогон 346).

Чинится сигнал, а не критерий: пометка 'zombie' снимает running-блокировку источника
(has_running_run), и ослабление критерия заперло бы источник на зависшем прогоне
навсегда. Плюс mark_done апдейтит WHERE status='running' — после ложной пометки финал
прогона становится no-op, отсюда нулевые counters у всех шести строк.

backfill_cian_history и backfill_newbuilding_enrichment получают колбэк on_progress,
дёргаемый на каждой сущности; планировщик переливает его в update_heartbeat вместе с
текущими счётчиками (best-effort — сбой heartbeat не роняет уже идущую работу).
newbuilding_enrich лечится тем же шаблоном превентивно: тот же единственный стартовый
вызов, на проде пока max 1.08 ч при limit=25, но полный прогон — 318 домов по ~2.6 мин.

Шаблон уже был в репозитории (house_imv_backfill, #1363) — здесь он доехал до двух
оставшихся долгоживущих задач. listing_source_snapshot (34 зомби) в правку не входит:
там вся работа — два set-based statement'а, heartbeat невозможен по построению,
лечилось иначе в #2607 (statement_timeout).

Refs #2725

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

@@ -76,7 +76,16 @@ async def _execute_cian_backfill(
     """Orchestrate Cian history backfill with heartbeat + checkpoint.
 
     Wraps backfill_cian_history(), updating scrape_runs counters (via update_heartbeat)
-    before and after the batch call for zombie-detection visibility.
+    НА КАЖДОЙ сущности батча, а не только до и после него (#2725). Раньше сигнал
+    живости слался ровно один раз — до батча, — а `reap_zombies` меряет именно
+    heartbeat_at с порогом 6 ч, и добивал живые прогоны строго на 6-м часу: 6 прод-
+    прогонов этого источника помечены 'zombie' со сдвигом heartbeat 16-32 мс, при том
+    что у пятерых внутри окна писались строки offer_price_history (у прогона 304 — до
+    5.4 ч после старта), а штатная длительность источника доходит до 5.06 ч (346).
+    Цена ошибки не косметическая: mark_done апдейтит WHERE status='running', так что
+    после ложной пометки собственный финал прогона становится no-op (отсюда нулевые
+    counters у всех шести), а has_running_run перестаёт видеть прогон и следующий тик
+    может запустить второй такой же батч поверх работающего.
 
     Checkpoint/resume semantics: backfill_cian_history() queries rows WHERE history IS
     NULL via LEFT JOIN — so re-running after a partial completion naturally skips
@@ -85,9 +94,33 @@ async def _execute_cian_backfill(
     Params (from default_params jsonb):
         batch_size: int — rows per run (listings + houses counted separately).
     """
-    from app.tasks.cian_history_backfill import backfill_cian_history
+    from app.tasks.cian_history_backfill import CianBackfillResult, backfill_cian_history
 
     batch_size = int(params.get("batch_size", 100))
+
+    def _counters(result: CianBackfillResult) -> dict[str, int]:
+        return {
+            "listings_processed": result.listings_processed,
+            "listings_succeeded": result.listings_succeeded,
+            "listings_failed": result.listings_failed_fetch + result.listings_failed_save,
+            "houses_processed": result.houses_processed,
+            "houses_succeeded": result.houses_succeeded,
+            "houses_failed": result.houses_failed_fetch + result.houses_failed_save,
+        }
+
+    def _heartbeat(progress: CianBackfillResult) -> None:
+        """Сигнал живости из середины батча. Best-effort: сбой heartbeat не должен
+        ронять уже идущую работу — прогон в худшем случае вернётся к прежнему
+        поведению (пометка 'zombie' на 6-м часу)."""
+        try:
+            runs_mod.update_heartbeat(db, run_id, _counters(progress))
+        except Exception:
+            logger.warning(
+                "scheduler: cian_history_backfill run_id=%d heartbeat failed (ignored)",
+                run_id,
+                exc_info=True,
+            )
+
     counters: dict[str, int] = {
         "listings_processed": 0,
         "listings_succeeded": 0,
@@ -106,17 +139,10 @@ async def _execute_cian_backfill(
             do_listings=True,
             do_houses=True,
             do_valuations=False,
+            on_progress=_heartbeat,
         )
 
-        counters = {
+        counters = {**_counters(result), "duration_sec": int(result.duration_sec)}
-            "listings_processed": result.listings_processed,
-            "listings_succeeded": result.listings_succeeded,
-            "listings_failed": result.listings_failed_fetch + result.listings_failed_save,
-            "houses_processed": result.houses_processed,
-            "houses_succeeded": result.houses_succeeded,
-            "houses_failed": result.houses_failed_fetch + result.houses_failed_save,
-            "duration_sec": int(result.duration_sec),
-        }
         runs_mod.mark_done(db, run_id, counters)
         logger.info(
             "scheduler: cian_history_backfill run_id=%d done — listings=%d/%d houses=%d/%d %.1fs",

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

@@ -18,6 +18,14 @@
      Requires migration 071_houses_cian_zhk_url.sql (cian_zhk_url column).
 
 Rate limit: scraper_settings.get_scraper_delay('cian') between requests.
+
+Сигнал живости (#2725): батч дёргает `on_progress` на КАЖДОЙ сущности — caller
+переливает это в scrape_runs.heartbeat_at. Пока колбэка не было, планировщик слал
+heartbeat один раз ДО батча, а `reap_zombies` меряет ровно heartbeat с порогом 6 ч —
+и добивал живые прогоны строго на 6-м часу (6 прод-прогонов, у пятерых внутри окна
+писались строки offer_price_history, у одного — до 5.4 ч после старта). Ослаблять
+критерий нельзя: пометка 'zombie' снимает running-блокировку источника
+(`has_running_run`), без неё зависший прогон запер бы источник навсегда.
 """
 
 from __future__ import annotations
@@ -25,6 +33,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import time
+from collections.abc import Callable
 from dataclasses import dataclass, field
 
 from scraper_kit.browser_fetcher import BrowserFetcher
@@ -68,6 +77,7 @@ async def backfill_cian_history(
     do_houses: bool = True,
     do_valuations: bool = False,
     dry_run: bool = False,
+    on_progress: Callable[[CianBackfillResult], None] | None = None,
 ) -> CianBackfillResult:
     """Iterate Cian listings + houses with missing history, fetch+save.
 
@@ -82,6 +92,10 @@ async def backfill_cian_history(
         do_valuations: process Cian Valuation Calculator batch (external_valuations backfill).
             Default False — opt-in because each call hits Cian auth-gated API.
         dry_run: skip all fetch+save; only count and log pending rows.
+        on_progress: колбэк живости (#2725) — вызывается на каждой сущности ЛЮБОГО из
+            трёх блоков, до её обработки, с текущим (мутируемым) result. Caller пишет
+            heartbeat; исключения колбэка — на его совести (планировщик глушит их сам),
+            здесь они прервали бы батч.
 
     Returns:
         CianBackfillResult with per-domain counters + total wall-clock duration.
@@ -120,6 +134,8 @@ async def backfill_cian_history(
                     listing_id: int = row["id"]
                     source_url: str = row["source_url"]
                     result.listings_processed += 1
+                    if on_progress is not None:
+                        on_progress(result)
 
                     enrichment = None
                     try:
@@ -209,6 +225,8 @@ async def backfill_cian_history(
                 house_id: int = hrow["id"]
                 zhk_url: str = hrow["cian_zhk_url"]
                 result.houses_processed += 1
+                if on_progress is not None:
+                    on_progress(result)
 
                 enrichment = None
                 try:
@@ -291,6 +309,8 @@ async def backfill_cian_history(
         else:
             for row in rows:
                 result.valuations_processed += 1
+                if on_progress is not None:
+                    on_progress(result)
                 try:
                     cval = await estimate_via_cian_valuation(
                         db,

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

@@ -66,6 +66,7 @@ import json
 import logging
 import random
 import time
+from collections.abc import Callable
 from dataclasses import dataclass, field, fields
 
 from sqlalchemy import text
@@ -306,8 +307,7 @@ def _house_enrichment_counts(db: Session, house_id: int) -> tuple[int, int, int]
     rc = int(
         db.execute(
             text(
-                "SELECT COUNT(*) FROM house_reliability_checks "
+                "SELECT COUNT(*) FROM house_reliability_checks WHERE house_id = CAST(:h AS bigint)"
-                "WHERE house_id = CAST(:h AS bigint)"
             ),
             {"h": house_id},
         ).scalar_one()
@@ -328,6 +328,7 @@ async def backfill_newbuilding_enrichment(
     force: bool = False,
     request_delay_sec: float | None = None,
     dry_run: bool = False,
+    on_progress: Callable[[NewbuildingEnrichBackfillResult], None] | None = None,
 ) -> NewbuildingEnrichBackfillResult:
     """Backfill the 3 newbuilding-enrichment tables over cian_newbuilding houses.
 
@@ -347,6 +348,11 @@ async def backfill_newbuilding_enrichment(
             (default 5s). Applied with ±20% jitter; anti-bot politeness. A house needing
             a resolve incurs TWO delays (resolve fetch + enrich fetch).
         dry_run: count the population + log the pending list, fetch nothing, write nothing.
+        on_progress: колбэк живости (#2725) — вызывается на каждом доме с текущим
+            (мутируемым) result; caller пишет scrape_runs.heartbeat_at. Без него
+            heartbeat уходил один раз до цикла, а `reap_zombies` меряет именно его:
+            дом обходится за ~2.6 мин, и на limit'е порядка 140 (полный прогон — 318
+            домов, см. выше) прогон переваливал бы 6-часовой порог живым.
 
     Returns:
         NewbuildingEnrichBackfillResult with population sizing, per-house outcome
@@ -415,6 +421,8 @@ async def backfill_newbuilding_enrichment(
         zhk_url: str | None = row["cian_zhk_url"]
         ext_id: str | None = row["ext_id"]
         result.processed += 1
+        if on_progress is not None:
+            on_progress(result)
 
         # Idempotency fast-path: with force=False the SELECT already excludes enriched
         # houses (price_dynamics + reliability present), so this branch is a belt-and-
@@ -696,6 +704,18 @@ async def run_newbuilding_enrich(
         "failed_save": 0,
     }
 
+    def _heartbeat(progress: NewbuildingEnrichBackfillResult) -> None:
+        """Сигнал живости из середины цикла (#2725). Best-effort — сбой heartbeat не
+        должен ронять уже идущий обход."""
+        try:
+            runs_mod.update_heartbeat(db, run_id, progress.to_dict())
+        except Exception:
+            logger.warning(
+                "scheduler: newbuilding_enrich run_id=%d heartbeat failed (ignored)",
+                run_id,
+                exc_info=True,
+            )
+
     try:
         runs_mod.update_heartbeat(db, run_id, counters)
 
@@ -704,6 +724,7 @@ async def run_newbuilding_enrich(
             limit=limit,
             force=force,
             request_delay_sec=request_delay_sec,
+            on_progress=_heartbeat,
         )
 
         counters = result.to_dict()

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

@@ -541,7 +541,8 @@ async def test_run_wrapper_marks_done_and_passes_params(monkeypatch: pytest.Monk
     """run_newbuilding_enrich emits heartbeat → delegates with parsed params → mark_done."""
     seen: dict = {}
 
-    async def _fake_backfill(_db, *, limit, force, request_delay_sec):
+    async def _fake_backfill(_db, *, limit, force, request_delay_sec, on_progress=None):
+        # on_progress — сигнал живости внутрь цикла (#2725); здесь только принимаем.
         seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
         return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_rows=2)
 
@@ -575,7 +576,8 @@ async def test_run_wrapper_defaults_when_params_empty(monkeypatch: pytest.Monkey
     """Empty default_params → limit=25, force=False, request_delay_sec=None (→ scraper delay)."""
     seen: dict = {}
 
-    async def _fake_backfill(_db, *, limit, force, request_delay_sec):
+    async def _fake_backfill(_db, *, limit, force, request_delay_sec, on_progress=None):
+        # on_progress — сигнал живости внутрь цикла (#2725); здесь только принимаем.
         seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
         return NewbuildingEnrichBackfillResult()
 

tradein-mvp/backend/tests/test_2725_heartbeat_in_batch.py

@@ -0,0 +1,225 @@
+"""#2725: сигнал живости слался один раз — до батча, — и живые прогоны reap'ились.
+
+Что было. `_execute_cian_backfill` дёргал `update_heartbeat` ровно один раз, ДО
+`backfill_cian_history()`, а сам батч (до 100 объявлений + 37 домов, каждое — fetch
+через браузер + пауза ~5 с) heartbeat не трогал. `reap_zombies` меряет именно
+`heartbeat_at` с порогом ZOMBIE_THRESHOLD_HOURS = 6 ч → прогон помечался 'zombie'
+строго на 6-м часу независимо от того, работает он или висит.
+
+Прод-замер 2026-08-06: 6 прогонов `cian_history_backfill` со статусом 'zombie', у всех
+шести сдвиг heartbeat 16-32 мс (= единственный стартовый вызов) и финал ровно на
+started_at + 6.00 ч. Живыми они при этом были: внутри окна пятерых писались строки
+offer_price_history с source='cian' (98/523/38/60/83 — плановый писатель этих строк
+только этот батч), у прогона 304 последняя строка легла через 5.40 ч после старта.
+Штатная длительность источника доходит до 5.06 ч (прогон 346, counters.duration_sec
+18230) — то есть источник ходит вплотную к порогу.
+
+Почему чинится сигнал, а не критерий: пометка 'zombie' снимает running-блокировку
+источника (`has_running_run` видит только status='running'), и без неё зависший
+прогон запер бы источник навсегда. Плюс `mark_done` апдейтит `WHERE status='running'`,
+поэтому после ложной пометки собственный финал прогона — no-op (отсюда нулевые
+counters у всех шести строк).
+
+Фальсификация: на старом коде тест 2 падает — планировщик не передавал `on_progress`,
+батч heartbeat не двигал, и к концу 7-часовой работы возраст сигнала = 7 ч > 6 ч.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.orchestration.scheduler import ZOMBIE_THRESHOLD_HOURS
+
+from app.services import scheduler as sched_mod
+from app.tasks import cian_history_backfill
+
+
+def _would_be_reaped(heartbeat_at: datetime, now: datetime) -> bool:
+    """Критерий reap_zombies дословно: heartbeat старше порога → 'zombie'."""
+    return heartbeat_at < now - timedelta(hours=ZOMBIE_THRESHOLD_HOURS)
+
+
+class _FakeBrowserFetcher:
+    def __init__(self, **kwargs: Any) -> None:
+        pass
+
+    async def __aenter__(self) -> _FakeBrowserFetcher:
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+
+# ── 1. Батч сообщает о продвижении на каждой сущности ────────────────────────
+async def test_batch_reports_progress_per_entity() -> None:
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.side_effect = [
+        [
+            {"id": 1, "source_url": "https://cian.ru/1"},
+            {"id": 2, "source_url": "https://cian.ru/2"},
+        ],
+        [{"id": 10, "cian_zhk_url": "https://cian.ru/zhk-10"}],
+    ]
+    seen: list[tuple[int, int]] = []
+
+    with (
+        patch.object(cian_history_backfill, "BrowserFetcher", _FakeBrowserFetcher),
+        patch.object(
+            cian_history_backfill,
+            "fetch_detail",
+            AsyncMock(return_value=SimpleNamespace(price_changes=[])),
+        ),
+        patch.object(cian_history_backfill, "save_detail_enrichment", MagicMock()),
+        patch(
+            "scraper_kit.providers.cian.newbuilding.fetch_newbuilding",
+            AsyncMock(return_value=SimpleNamespace()),
+        ),
+        patch("scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment", MagicMock()),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        result = await cian_history_backfill.backfill_cian_history(
+            db,
+            do_listings=True,
+            do_houses=True,
+            do_valuations=False,
+            on_progress=lambda r: seen.append((r.listings_processed, r.houses_processed)),
+        )
+
+    # По одному сигналу на каждую сущность обоих блоков, счётчики растут.
+    assert seen == [(1, 0), (2, 0), (2, 1)]
+    assert result.listings_processed == 2
+    assert result.houses_processed == 1
+
+
+# ── 2. Долгий прогон с продвигающимся heartbeat не помечается зависшим ───────
+async def test_long_run_with_advancing_heartbeat_is_not_reaped() -> None:
+    """7 часов работы, час на сущность: планировщик обязан двигать heartbeat."""
+    t0 = datetime(2026, 6, 26, 4, 47, tzinfo=UTC)
+    clock = SimpleNamespace(now=t0)
+    beats: list[datetime] = []
+
+    async def _fake_batch(db: Any, **kwargs: Any) -> Any:
+        on_progress = kwargs.get("on_progress")
+        result = cian_history_backfill.CianBackfillResult()
+        for _ in range(7):  # 7 сущностей по часу — дольше 6-часового порога
+            clock.now += timedelta(hours=1)
+            result.listings_processed += 1
+            if on_progress is not None:
+                on_progress(result)
+        result.duration_sec = 7 * 3600
+        return result
+
+    fake_runs = SimpleNamespace(
+        update_heartbeat=lambda db, run_id, counters: beats.append(clock.now),
+        mark_done=MagicMock(),
+        mark_failed=MagicMock(),
+    )
+
+    with (
+        patch.object(sched_mod, "runs_mod", fake_runs),
+        patch.object(cian_history_backfill, "backfill_cian_history", _fake_batch),
+    ):
+        await sched_mod._execute_cian_backfill(MagicMock(), run_id=1, params={})
+
+    assert len(beats) == 8, "стартовый сигнал + по одному на сущность"
+    reaped = _would_be_reaped(beats[-1], clock.now)
+    assert not reaped, "живой прогон с продвигающимся heartbeat не должен reap'иться"
+    assert fake_runs.mark_done.called
+
+
+# ── 3. Прогон без продвижения — помечается (контроль критерия) ───────────────
+async def test_long_run_without_advancing_heartbeat_is_reaped() -> None:
+    """Тот же прогон, но батч сигнала не шлёт — критерий обязан сработать."""
+    t0 = datetime(2026, 6, 26, 4, 47, tzinfo=UTC)
+    clock = SimpleNamespace(now=t0)
+    beats: list[datetime] = []
+
+    async def _mute_batch(db: Any, **kwargs: Any) -> Any:
+        clock.now += timedelta(hours=7)  # работает, но молча
+        return cian_history_backfill.CianBackfillResult()
+
+    fake_runs = SimpleNamespace(
+        update_heartbeat=lambda db, run_id, counters: beats.append(clock.now),
+        mark_done=MagicMock(),
+        mark_failed=MagicMock(),
+    )
+
+    with (
+        patch.object(sched_mod, "runs_mod", fake_runs),
+        patch.object(cian_history_backfill, "backfill_cian_history", _mute_batch),
+    ):
+        await sched_mod._execute_cian_backfill(MagicMock(), run_id=1, params={})
+
+    assert beats == [t0], "единственный сигнал — стартовый"
+    assert _would_be_reaped(beats[-1], clock.now)
+
+
+# ── 4. Сбой heartbeat не роняет уже идущую работу ────────────────────────────
+async def test_heartbeat_failure_does_not_abort_the_batch() -> None:
+    processed: list[int] = []
+    calls = {"n": 0}
+
+    def _flaky_heartbeat(db: Any, run_id: int, counters: dict[str, int]) -> None:
+        calls["n"] += 1
+        if calls["n"] > 1:  # стартовый прошёл, дальше БД отвалилась
+            raise Exception("DB gone")
+
+    async def _fake_batch(db: Any, **kwargs: Any) -> Any:
+        on_progress = kwargs["on_progress"]
+        result = cian_history_backfill.CianBackfillResult()
+        for _ in range(3):
+            result.listings_processed += 1
+            on_progress(result)  # обязан проглотить исключение внутри себя
+            processed.append(result.listings_processed)
+        return result
+
+    fake_runs = SimpleNamespace(
+        update_heartbeat=_flaky_heartbeat,
+        mark_done=MagicMock(),
+        mark_failed=MagicMock(),
+    )
+
+    with (
+        patch.object(sched_mod, "runs_mod", fake_runs),
+        patch.object(cian_history_backfill, "backfill_cian_history", _fake_batch),
+    ):
+        await sched_mod._execute_cian_backfill(MagicMock(), run_id=1, params={})
+
+    assert processed == [1, 2, 3]
+    assert fake_runs.mark_done.called
+
+
+# ── 5. Тот же дефект у newbuilding_enrich — сигнал прокинут ──────────────────
+async def test_newbuilding_enrich_passes_progress_callback() -> None:
+    from app.tasks import newbuilding_enrich_backfill as nb
+
+    beats: list[dict[str, int]] = []
+
+    async def _fake_backfill(db: Any, **kwargs: Any) -> Any:
+        on_progress = kwargs.get("on_progress")
+        assert on_progress is not None, "планировщик обязан прокинуть сигнал живости"
+        result = nb.NewbuildingEnrichBackfillResult()
+        result.processed += 1
+        on_progress(result)
+        return result
+
+    fake_runs = SimpleNamespace(
+        update_heartbeat=lambda db, run_id, counters: beats.append(counters),
+        mark_done=MagicMock(),
+        mark_failed=MagicMock(),
+    )
+
+    with (
+        patch.object(nb, "runs_mod", fake_runs),
+        patch.object(nb, "backfill_newbuilding_enrichment", _fake_backfill),
+    ):
+        await nb.run_newbuilding_enrich(MagicMock(), run_id=1, params={})
+
+    assert len(beats) == 2, "стартовый сигнал + сигнал из середины цикла"
+    assert beats[-1]["processed"] == 1