Merge pull request 'feat(scrapers): ночной avito detail-enrichment backfill для legacy listings (Closes #1551)' (#1552) from feat/avito-detail-backfill into main

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:

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)

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;

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()