From 227e82dc01d951c68625e767021d4533230ccc55 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 16 Jun 2026 13:25:29 +0300 Subject: [PATCH] feat(tradein): avito detail-enrichment nightly backfill (#1551) - New task app/tasks/avito_detail_backfill.py with run_avito_detail_backfill() * Single snapshot SELECT at start (guarantees termination) * Same proxy/AsyncSession path as scrape_pipeline.py step 5 * Budget guard (budget_sec), consecutive block abort (mark_done not mark_failed) * rotate_ip() on every AvitoBlockedError; rollback on generic Exception (#1368) * start = time.monotonic() initialized before try so except can reference it - Scheduler wiring: trigger_avito_detail_backfill_run() + elif in scheduler_loop() - Migration 112: scrape_schedules INSERT window 09-12 UTC, batch_size=800, budget_sec=3600, request_delay_sec=6, max_consecutive_blocks=5 - 7 unit tests (no pytest-mock, unittest.mock only): all 7 passing, full CI suite 1809 passed --- tradein-mvp/backend/app/services/scheduler.py | 39 +++ .../app/tasks/avito_detail_backfill.py | 270 ++++++++++++++++++ ...e_schedules_seed_avito_detail_backfill.sql | 54 ++++ .../tests/tasks/test_avito_detail_backfill.py | 244 ++++++++++++++++ 4 files changed, 607 insertions(+) create mode 100644 tradein-mvp/backend/app/tasks/avito_detail_backfill.py create mode 100644 tradein-mvp/backend/data/sql/112_scrape_schedules_seed_avito_detail_backfill.sql create mode 100644 tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 84db6fb3..e33861ec 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -1170,6 +1170,43 @@ def import_rosreestr_dkp( raise +async def trigger_avito_detail_backfill_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Create scrape_run + launch run_avito_detail_backfill in asyncio.create_task. + + Nightly backfill detail_enriched_at for legacy avito listings (#1551). + ~15243/15917 avito listings have detail_enriched_at IS NULL -- never enriched + by city_sweep ENRICH_DETAIL (only top-N proximate per anchor are touched). + + Mirrors trigger_geocode_missing_listings_run: claim run -> create_task -> + mark_done/failed delegated to tasks/avito_detail_backfill.run_avito_detail_backfill. + + Returns run_id or None (skip -- already running). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.avito_detail_backfill import run_avito_detail_backfill + + await run_avito_detail_backfill(run_db, run_id=run_id, params=params) + except Exception: + logger.exception("scheduler: run_avito_detail_backfill crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered avito_detail_backfill run_id=%d", run_id) + return run_id + + def get_due_schedules(db: Session) -> list[dict[str, Any]]: """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW()).""" rows = ( @@ -1237,6 +1274,8 @@ async def scheduler_loop() -> None: await trigger_yandex_newbuilding_sweep_run(db, sch) elif source == "geocode_missing_listings": await trigger_geocode_missing_listings_run(db, sch) + elif source == "avito_detail_backfill": + await trigger_avito_detail_backfill_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py new file mode 100644 index 00000000..f60bde42 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py @@ -0,0 +1,270 @@ +"""Scheduled backfill: detail-enrichment for legacy avito listings (#1551). + +Nightly window 09:00-12:00 UTC (migration 112, source=avito_detail_backfill). +Offset from avito_city_sweep (~03-04 UTC) to avoid sharing proxy IP simultaneously. + +Problem: ~15243/15917 avito listings have detail_enriched_at IS NULL. +City sweep ENRICH_DETAIL only enriches top-N proximate listings per anchor. +Legacy listings (older than 2h or outside radius) are never enriched. + +Solution: single snapshot SELECT at start (guarantees termination), same proxy +session path as scrape_pipeline.py step 5. Block handling mirrors step 5: +rotate IP on every block, abort after max_consecutive_blocks (mark_done not +mark_failed -- block is temporary, retry next night via NULL detail_enriched_at). +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from dataclasses import dataclass, field +from urllib.parse import urlparse + +from curl_cffi.requests import AsyncSession +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.config import settings +from app.services import scrape_runs as runs_mod +from app.services.scrape_pipeline import _CHROME_HEADERS, _avito_proxies +from app.services.scrapers.avito import AvitoScraper +from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment +from app.services.scrapers.avito_exceptions import ( + AvitoBlockedError, + AvitoRateLimitedError, +) +from app.services.scrapers.browser_fetcher import BrowserFetcher + +logger = logging.getLogger(__name__) + +__all__ = [ + "AvitoDetailBackfillResult", + "run_avito_detail_backfill", +] + + +@dataclass +class AvitoDetailBackfillResult: + """Counters for one backfill run.""" + + attempted: int = 0 + enriched: int = 0 + blocked: int = 0 + failed: int = 0 + duration_sec: float = field(default=0.0) + + def to_dict(self) -> dict[str, int]: + return { + "attempted": self.attempted, + "enriched": self.enriched, + "blocked": self.blocked, + "failed": self.failed, + "duration_sec": int(self.duration_sec), + } + + +async def run_avito_detail_backfill( + db: Session, + *, + run_id: int, + params: dict, +) -> AvitoDetailBackfillResult: + """Backfill detail_enriched_at for legacy avito listings via mobile proxy. + + Params (from default_params jsonb in scrape_schedules): + batch_size: int -- snapshot size (SELECT LIMIT), default 800. + budget_sec: float -- wall-clock budget per run, default 3600s. + request_delay_sec: float -- delay between listings, default 6.0s. + max_consecutive_blocks: int -- abort threshold, default 5. + + Lifecycle: update_heartbeat -> snapshot -> loop with budget guard -> + mark_done (incl. partial/block-abort) / mark_failed (exception only). + """ + batch_size = int(params.get("batch_size", 800)) + budget_sec = float(params.get("budget_sec", 3600)) + request_delay_sec = float(params.get("request_delay_sec", 6.0)) + max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5)) + + counters = AvitoDetailBackfillResult() + current_counters: dict[str, int] = counters.to_dict() + + browser_mode = settings.scraper_fetch_mode == "browser" + session: AsyncSession | None = None + browser_fetcher: BrowserFetcher | None = None + own_session = False + own_browser = False + + scraper = AvitoScraper() + start = time.monotonic() + + try: + # Setup session (mirrors run_avito_pipeline lines 161-183) + if browser_mode: + browser_fetcher = BrowserFetcher() + await browser_fetcher.__aenter__() + own_browser = True + scraper._browser = browser_fetcher + else: + own_session = True + session = AsyncSession( + impersonate="chrome120", + timeout=25, + headers=_CHROME_HEADERS, + proxies=_avito_proxies(), + ) + scraper._cffi = session + + runs_mod.update_heartbeat(db, run_id, current_counters) + + # SNAPSHOT: single SELECT at start -- NOT re-selected in loop. + # Priority: is_active DESC (active first), scraped_at DESC (newest first). + snapshot = ( + db.execute( + text( + """ + SELECT id, source_url + FROM listings + WHERE source = 'avito' + AND detail_enriched_at IS NULL + AND source_url IS NOT NULL + ORDER BY is_active DESC NULLS LAST, scraped_at DESC NULLS LAST + LIMIT CAST(:batch_size AS int) + """ + ), + {"batch_size": batch_size}, + ) + .mappings() + .all() + ) + + if not snapshot: + logger.info( + "avito_detail_backfill: run_id=%d -- no pending listings " + "(detail_enriched_at IS NULL = 0), done", + run_id, + ) + runs_mod.mark_done(db, run_id, current_counters) + return counters + + logger.info( + "avito_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs " + "delay=%.1fs max_blocks=%d browser=%s)", + run_id, + len(snapshot), + budget_sec, + request_delay_sec, + max_consecutive_blocks, + browser_mode, + ) + + consecutive_blocks = 0 + do_sleep = False + + for idx, row in enumerate(snapshot): + # Budget guard + elapsed = time.monotonic() - start + if elapsed > budget_sec: + logger.info( + "avito_detail_backfill: run_id=%d -- budget %.0fs exhausted " + "(elapsed=%.1fs), stopping at #%d/%d", + run_id, + budget_sec, + elapsed, + idx, + len(snapshot), + ) + break + + # Delay before each request except the first + if do_sleep: + await asyncio.sleep(request_delay_sec) + do_sleep = True + + source_url: str = row["source_url"] + counters.attempted += 1 + + # Normalise URL -> path (mirrors scrape_pipeline.py line 311) + item_url = urlparse(source_url).path if source_url.startswith("http") else source_url + + try: + enrichment = await fetch_detail( + item_url, + cffi_session=session, + browser_fetcher=browser_fetcher, + ) + if save_detail_enrichment(db, enrichment): + counters.enriched += 1 + consecutive_blocks = 0 + + except (AvitoBlockedError, AvitoRateLimitedError) as e: + consecutive_blocks += 1 + counters.blocked += 1 + do_sleep = False + logger.warning( + "avito_detail_backfill: run_id=%d BLOCKED #%d/%d (consecutive=%d): %s", + run_id, + idx + 1, + len(snapshot), + consecutive_blocks, + e, + ) + await scraper._rotate_ip() + if consecutive_blocks >= max_consecutive_blocks: + logger.error( + "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, " + "IP rate-limited. enriched=%d attempted=%d", + run_id, + consecutive_blocks, + counters.enriched, + counters.attempted, + ) + break + + except Exception as e: + counters.failed += 1 + logger.warning( + "avito_detail_backfill: run_id=%d listing %s failed: %s", + run_id, + source_url, + e, + ) + try: + db.rollback() + except Exception: + pass + + if counters.attempted % 25 == 0: + current_counters = counters.to_dict() + runs_mod.update_heartbeat(db, run_id, current_counters) + + counters.duration_sec = time.monotonic() - start + current_counters = counters.to_dict() + runs_mod.mark_done(db, run_id, current_counters) + logger.info( + "avito_detail_backfill: run_id=%d DONE -- attempted=%d enriched=%d " + "blocked=%d failed=%d duration=%.1fs", + run_id, + counters.attempted, + counters.enriched, + counters.blocked, + counters.failed, + counters.duration_sec, + ) + return counters + + except Exception as exc: + counters.duration_sec = time.monotonic() - start + logger.exception( + "avito_detail_backfill: run_id=%d FAILED after %.1fs", + run_id, + counters.duration_sec, + ) + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters.to_dict()) + raise + + finally: + if own_session and session is not None: + await session.close() + if own_browser and browser_fetcher is not None: + await browser_fetcher.__aexit__(None, None, None) diff --git a/tradein-mvp/backend/data/sql/112_scrape_schedules_seed_avito_detail_backfill.sql b/tradein-mvp/backend/data/sql/112_scrape_schedules_seed_avito_detail_backfill.sql new file mode 100644 index 00000000..39812d4d --- /dev/null +++ b/tradein-mvp/backend/data/sql/112_scrape_schedules_seed_avito_detail_backfill.sql @@ -0,0 +1,54 @@ +-- 112_scrape_schedules_seed_avito_detail_backfill.sql +-- Nightly detail-enrichment backfill for legacy avito listings (detail_enriched_at IS NULL). +-- +-- Root cause (#1551): +-- City sweep ENRICH_DETAIL step only enriches top-N listings proximate to each anchor. +-- Legacy avito listings (older than 2h or outside radius) are never enriched. +-- On prod: 15243/15917 avito listings with detail_enriched_at IS NULL. +-- These carry no kitchen_area_m2 / ceiling_height_m / finishing / views / description. +-- +-- Solution: +-- Nightly backfill task (avito_detail_backfill) takes a SNAPSHOT of pending listings +-- (single SELECT LIMIT batch_size at start, same proxy path as scrape_pipeline step 5), +-- then iterates with budget guard. Block handling: rotate IP on every AvitoBlockedError, +-- abort after max_consecutive_blocks (mark_done, NOT mark_failed -- block is temporary). +-- +-- Window 09:00-12:00 UTC (12:00-15:00 MSK): +-- - After city sweeps (~03-04 UTC, depositing new listings). +-- - After geocode_missing_listings backfill (06:00-09:00 UTC). +-- - Offset from avito_city_sweep to avoid sharing proxy IP simultaneously. +-- +-- default_params: +-- batch_size -- listings per run (800 ~ 1.3h at 6s/listing, fits in 3h window). +-- budget_sec -- wall-clock budget; 3600 = 1 hour (safe within 3h window). +-- request_delay_sec -- anti-bot delay between requests (6s = ~600 req/hour). +-- max_consecutive_blocks -- abort threshold for consecutive IP blocks (5). +-- +-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)). +-- Idempotent: ON CONFLICT (source) DO NOTHING. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'avito_detail_backfill', + true, + 9, + 12, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 9)) AT TIME ZONE 'UTC', + '{"batch_size": 800, "budget_sec": 3600, "request_delay_sec": 6, "max_consecutive_blocks": 5}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (replaces cron-script setup). Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), yandex_address_backfill (#855, EKB pilot), sber_index_pull (#887, monthly), rosreestr_quarter_poll (#888, monthly), cian_city_sweep (dormant, #973), yandex_newbuilding_sweep (dormant, #974), geocode_missing_listings (#1: listings geom backfill, all sources), avito_detail_backfill (#1551: nightly detail-enrichment backfill for legacy avito listings).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py new file mode 100644 index 00000000..82d113cc --- /dev/null +++ b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import os +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +_wp_mock = MagicMock() +sys.modules.setdefault("weasyprint", _wp_mock) + +import pytest # noqa: E402 + +from app.tasks.avito_detail_backfill import ( # noqa: E402 + AvitoDetailBackfillResult, + run_avito_detail_backfill, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_snapshot(n: int) -> list[dict]: + return [{"id": i + 1, "source_url": f"/items/{i + 1}"} for i in range(n)] + + +def _mock_db(snapshot: list[dict]) -> MagicMock: + """Fake Session: first execute() returns snapshot via .mappings().all().""" + db = MagicMock() + sel = MagicMock() + sel.mappings.return_value.all.return_value = snapshot + db.execute.return_value = sel + return db + + +_FETCH = "app.tasks.avito_detail_backfill.fetch_detail" +_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment" +_RUNS = "app.tasks.avito_detail_backfill.runs_mod" +_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep" +_SESSION = "app.tasks.avito_detail_backfill.AsyncSession" +_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper" +_SETTINGS = "app.tasks.avito_detail_backfill.settings" + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_backfill_empty_snapshot_marks_done() -> None: + """Empty snapshot -> mark_done immediately, no fetch calls.""" + db = _mock_db([]) + runs = MagicMock() + fake_session = AsyncMock() + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=fake_session), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH) as mock_fetch, + ): + result = await run_avito_detail_backfill( + db, run_id=1, params={"batch_size": 10, "budget_sec": 60} + ) + + assert isinstance(result, AvitoDetailBackfillResult) + assert result.attempted == 0 + assert result.enriched == 0 + mock_fetch.assert_not_called() + runs.mark_done.assert_called_once() + runs.mark_failed.assert_not_called() + + +@pytest.mark.asyncio +async def test_backfill_processes_snapshot_to_completion() -> None: + """3 listings -> all fetched and enriched, mark_done called.""" + snapshot = _make_snapshot(3) + db = _mock_db(snapshot) + runs = MagicMock() + mock_enrichment = MagicMock() + mock_fetch = AsyncMock(return_value=mock_enrichment) + mock_save = MagicMock(return_value=True) + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, mock_save), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=2, params={"batch_size": 10, "budget_sec": 3600} + ) + + assert result.attempted == 3 + assert result.enriched == 3 + assert result.blocked == 0 + assert result.failed == 0 + assert mock_fetch.call_count == 3 + runs.mark_done.assert_called_once() + runs.mark_failed.assert_not_called() + + +@pytest.mark.asyncio +async def test_backfill_blocked_abort_after_max_consecutive() -> None: + """5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed), rotate_ip x5.""" + from app.services.scrapers.avito_exceptions import AvitoBlockedError + + snapshot = _make_snapshot(10) + db = _mock_db(snapshot) + runs = MagicMock() + blocked_exc = AvitoBlockedError("ip blocked") + mock_fetch = AsyncMock(side_effect=blocked_exc) + mock_scraper = MagicMock() + mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER, mock_scraper), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5} + ) + + assert result.blocked == 5 + assert result.attempted == 5 + assert result.enriched == 0 + runs.mark_done.assert_called_once() + runs.mark_failed.assert_not_called() + assert mock_scraper.return_value._rotate_ip.call_count == 5 + + +@pytest.mark.asyncio +async def test_backfill_budget_guard_stops_loop() -> None: + """Budget expired before first listing -> fetch_detail not called.""" + snapshot = _make_snapshot(5) + db = _mock_db(snapshot) + runs = MagicMock() + mono_values = iter([0.0, 999.0, 999.0]) + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch("app.tasks.avito_detail_backfill.time.monotonic", side_effect=mono_values), + patch(_FETCH) as mock_fetch, + ): + await run_avito_detail_backfill(db, run_id=4, params={"batch_size": 5, "budget_sec": 1}) + + mock_fetch.assert_not_called() + runs.mark_done.assert_called_once() + + +@pytest.mark.asyncio +async def test_backfill_top_level_exception_marks_failed() -> None: + """db.execute raises -> mark_failed called, exception re-raised.""" + db = MagicMock() + db.execute.side_effect = RuntimeError("DB connection lost") + runs = MagicMock() + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + ): + with pytest.raises(RuntimeError, match="DB connection lost"): + await run_avito_detail_backfill( + db, run_id=5, params={"batch_size": 5, "budget_sec": 60} + ) + + runs.mark_failed.assert_called_once() + runs.mark_done.assert_not_called() + + +@pytest.mark.asyncio +async def test_backfill_rotate_ip_called_on_each_block() -> None: + """1 block + 1 success -> rotate_ip called once, enriched=1.""" + from app.services.scrapers.avito_exceptions import AvitoBlockedError + + snapshot = _make_snapshot(2) + db = _mock_db(snapshot) + runs = MagicMock() + mock_enrichment = MagicMock() + blocked_exc = AvitoBlockedError("x") + mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment]) + mock_scraper = MagicMock() + mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True) + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER, mock_scraper), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=6, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5} + ) + + assert result.enriched == 1 + assert result.blocked == 1 + assert mock_scraper.return_value._rotate_ip.call_count == 1 + runs.mark_done.assert_called_once() + + +@pytest.mark.asyncio +async def test_backfill_fetch_exception_continues() -> None: + """RuntimeError on one listing -> failed++, db.rollback(), loop continues for next.""" + snapshot = _make_snapshot(2) + db = _mock_db(snapshot) + runs = MagicMock() + mock_enrichment = MagicMock() + mock_fetch = AsyncMock(side_effect=[RuntimeError("parse error"), mock_enrichment]) + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=7, params={"batch_size": 10, "budget_sec": 3600} + ) + + assert result.failed == 1 + assert result.enriched == 1 + assert result.attempted == 2 + db.rollback.assert_called() + runs.mark_done.assert_called_once()