From eec62c4e1cb8699729984d81ccb45f6cfde1ebd2 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 6 Aug 2026 16:19:18 +0500 Subject: [PATCH] =?UTF-8?q?fix(tradein/scraper):=20=D1=81=D0=B8=D0=B3?= =?UTF-8?q?=D0=BD=D0=B0=D0=BB=20=D0=B6=D0=B8=D0=B2=D0=BE=D1=81=D1=82=D0=B8?= =?UTF-8?q?=20=D0=B8=D0=B7=20=D1=81=D0=B5=D1=80=D0=B5=D0=B4=D0=B8=D0=BD?= =?UTF-8?q?=D1=8B=20=D0=B1=D0=B0=D1=82=D1=87=D0=B0=20=E2=80=94=20=D0=B6?= =?UTF-8?q?=D0=B8=D0=B2=D1=8B=D0=B5=20=D0=BF=D1=80=D0=BE=D0=B3=D0=BE=D0=BD?= =?UTF-8?q?=D1=8B=20=D0=BF=D0=B5=D1=80=D0=B5=D1=81=D1=82=D0=B0=D1=8E=D1=82?= =?UTF-8?q?=20=D1=87=D0=B8=D1=81=D0=BB=D0=B8=D1=82=D1=8C=D1=81=D1=8F=20?= =?UTF-8?q?=D0=B7=D0=B0=D0=B2=D0=B8=D1=81=D1=88=D0=B8=D0=BC=D0=B8=20(#2725?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 48 +++- .../app/tasks/cian_history_backfill.py | 20 ++ .../app/tasks/newbuilding_enrich_backfill.py | 25 +- .../tasks/test_newbuilding_enrich_backfill.py | 6 +- .../tests/test_2725_heartbeat_in_batch.py | 225 ++++++++++++++++++ 5 files changed, 309 insertions(+), 15 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_2725_heartbeat_in_batch.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 3e715dd3..b1a43824 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/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 = { - "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), - } + counters = {**_counters(result), "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", diff --git a/tradein-mvp/backend/app/tasks/cian_history_backfill.py b/tradein-mvp/backend/app/tasks/cian_history_backfill.py index d76f2be9..b6c5907b 100644 --- a/tradein-mvp/backend/app/tasks/cian_history_backfill.py +++ b/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, diff --git a/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py b/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py index b6f5dda1..e46fe9d9 100644 --- a/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py +++ b/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 " - "WHERE house_id = CAST(:h AS bigint)" + "SELECT COUNT(*) FROM house_reliability_checks 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() diff --git a/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py b/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py index 16c32b25..4df0d729 100644 --- a/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py +++ b/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() diff --git a/tradein-mvp/backend/tests/test_2725_heartbeat_in_batch.py b/tradein-mvp/backend/tests/test_2725_heartbeat_in_batch.py new file mode 100644 index 00000000..a0450aa2 --- /dev/null +++ b/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