Merge pull request 'fix(scrapers): geocode-missing listings nightly scheduler — listings geom coverage backfill (all sources)' (#1549) from fix/geocode-missing-listings-scheduler into main

Reviewed-on: #1549

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

@@ -51,6 +51,11 @@ Sources:
                               enable manually after deploy verified; window 02:00-05:00 UTC)
   - cian_city_sweep         → run_cian_city_sweep (newbuilding Phase 4, #973; shipped DORMANT —
                               proxy dead, enable manually after proxy restored)
+  - geocode_missing_listings → run_geocode_missing_listings
+                              (tasks/geocode_missing.py; nightly backfill listings.lat/lon для
+                              всех источников с NULL coords; address-dedup batch loop с
+                              wall-clock budget; window 06:00-09:00 UTC — после city sweeps
+                              ~03-04 UTC и deals OS-cron 05:30/05:45 UTC)
 """
 
 from __future__ import annotations
@@ -553,6 +558,45 @@ async def trigger_yandex_address_backfill_run(
     return run_id
 
 
+async def trigger_geocode_missing_listings_run(
+    db: Session, schedule_row: dict[str, Any]
+) -> int | None:
+    """Создать scrape_runs + launch run_geocode_missing_listings в asyncio.create_task.
+
+    Nightly backfill listings.lat/lon для всех источников с NULL coords (#1: geom-coverage gap).
+    Адрес-dedup batch loop с wall-clock budget (default 30 мин).
+
+    Root cause: scrapers сохраняют listings с lat=lon=NULL — «геокод подтянет cron»,
+    но cron геокодирует только deals, не listings. Эта функция замыкает петлю: нигде
+    снаружи нет OS cron для listings — только этот планировщик.
+
+    Mirrors trigger_yandex_address_backfill_run: claim run → create_task → mark_done/failed.
+
+    Returns run_id или 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.geocode_missing import run_geocode_missing_listings
+
+            await run_geocode_missing_listings(run_db, run_id=run_id, params=params)
+        except Exception:
+            logger.exception("scheduler: run_geocode_missing_listings 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 geocode_missing_listings run_id=%d", run_id)
+    return run_id
+
+
 async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task)."""
     run_id = _claim_run(db, schedule_row)
@@ -1191,6 +1235,8 @@ async def scheduler_loop() -> None:
                         await trigger_newbuilding_enrich_run(db, sch)
                     elif source == "yandex_newbuilding_sweep":
                         await trigger_yandex_newbuilding_sweep_run(db, sch)
+                    elif source == "geocode_missing_listings":
+                        await trigger_geocode_missing_listings_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

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

@@ -2,7 +2,8 @@
 
 Запускается:
   - Manual через POST /admin/scrape/geocode-missing-listings
-  - Future: cron job (Celery beat если bootstrap'нут / OS cron)
+  - Scheduled: nightly via scrape_schedules (source='geocode_missing_listings', migration 110)
+    — wired into in-app scheduler, window 06:00-09:00 UTC.
 
 Pattern: dedup по address (1 unique address → 1 geocode call → UPDATE all listings).
 Rate limit: Nominatim 1 req/sec. Yandex 25K/day если YANDEX_GEOCODER_API_KEY set.
@@ -11,6 +12,8 @@ Rate limit: Nominatim 1 req/sec. Yandex 25K/day если YANDEX_GEOCODER_API_KEY
   - Этот модуль группирует по address → меньше API calls (dedup).
   - Поддерживает all sources включая Avito (после PR #487 убрали jitter).
   - Возвращает GeocodeBackfillResult с детальными counters.
+  - Loop-safe: SELECT фильтрует geocode_tried_at IS NULL OR tried_at < 7 days;
+    при geocode failure помечает tried_at=NOW() → адрес не переотбирается в этом же run.
 """
 
 from __future__ import annotations
@@ -22,6 +25,7 @@ from dataclasses import dataclass, field
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services import scrape_runs as runs_mod
 from app.services.estimator import _geocode_is_coarse
 from app.services.geocoder import geocode
 
@@ -70,7 +74,12 @@ async def geocode_missing_listings(
     start = time.monotonic()
     result = GeocodeBackfillResult()
 
-    # 1. Найти top-N адресов с NULL coords (DESC by occurrence count)
+    # 1. Найти top-N адресов с NULL coords (DESC by occurrence count).
+    # Фильтруем адреса, по которым геокодер уже пробовал и не нашёл — они помечены
+    # geocode_tried_at. Повторяем попытку только если tried_at старше 7 дней (возможен
+    # переезд адреса в кэше или смена провайдера), либо tried_at IS NULL (ещё не пробовали).
+    # Это делает функцию loop-safe: при вызове несколько раз в одном прогоне
+    # failed-адреса не переотбираются бесконечно.
     rows = (
         db.execute(
             text(
@@ -80,6 +89,8 @@ async def geocode_missing_listings(
             WHERE lat IS NULL
               AND address IS NOT NULL
               AND length(trim(address)) >= 5
+              AND (geocode_tried_at IS NULL
+                   OR geocode_tried_at < NOW() - INTERVAL '7 days')
             GROUP BY address
             ORDER BY listings_count DESC, address ASC
             LIMIT :limit
@@ -114,6 +125,17 @@ async def geocode_missing_listings(
         except Exception as exc:
             logger.warning("geocode_missing: geocode raised for '%s': %s", address[:60], exc)
             result.addresses_failed += 1
+            if not dry_run:
+                # Пометить tried_at чтобы адрес не переотбирался в следующих batch'ах
+                # этого же прогона (loop-safe backoff 7 дней).
+                db.execute(
+                    text(
+                        "UPDATE listings SET geocode_tried_at = NOW()"
+                        " WHERE address = :addr AND lat IS NULL"
+                    ),
+                    {"addr": address},
+                )
+                db.commit()
             continue
 
         if geo is None:
@@ -123,6 +145,16 @@ async def geocode_missing_listings(
                 address[:60],
                 listings_count,
             )
+            if not dry_run:
+                # Пометить tried_at — geocoder не нашёл адрес, backoff 7 дней.
+                db.execute(
+                    text(
+                        "UPDATE listings SET geocode_tried_at = NOW()"
+                        " WHERE address = :addr AND lat IS NULL"
+                    ),
+                    {"addr": address},
+                )
+                db.commit()
             continue
 
         if geo.provider == "cache":
@@ -150,12 +182,13 @@ async def geocode_missing_listings(
         precision: str | None = "city" if _geocode_is_coarse(geo) else None
 
         # UPDATE listings — PostGIS trigger (listings_set_geom_trg) обновит geom автоматически.
-        # geo_precision проставляется одновременно с координатами.
+        # geo_precision и geocode_tried_at проставляются одновременно с координатами.
         update_result = db.execute(
             text(
                 """
                 UPDATE listings
-                SET lat = :lat, lon = :lon, geo_precision = :precision
+                SET lat = :lat, lon = :lon, geo_precision = :precision,
+                    geocode_tried_at = NOW()
                 WHERE address = :addr AND lat IS NULL
                 """
             ),
@@ -195,3 +228,109 @@ async def geocode_missing_listings(
     )
 
     return result
+
+
+async def run_geocode_missing_listings(
+    db: Session,
+    *,
+    run_id: int,
+    params: dict,
+) -> GeocodeBackfillResult:
+    """Run-lifecycle wrapper: batch loop с wall-clock budget.
+
+    Запускается планировщиком (source='geocode_missing_listings') или вручную.
+    Гоняет geocode_missing_listings() в цикле до тех пор пока:
+      - res.addresses_total == 0 (нет pending адресов)
+      - res.addresses_total < batch_size (дренаж — последний batch меньше полного)
+      - истёк budget_sec
+
+    Params (из default_params jsonb в scrape_schedules):
+        batch_size: int  — адресов за один вызов geocode_missing_listings (default 200).
+        budget_sec: float — максимальное время прогона в секундах (default 1800 = 30 мин).
+
+    Lifecycle:
+        update_heartbeat перед loop → аккумуляция counters → mark_done / mark_failed.
+    """
+    batch_size = int(params.get("batch_size", 200))
+    budget_sec = float(params.get("budget_sec", 1800))
+
+    total = GeocodeBackfillResult()
+    counters: dict[str, int] = {
+        "checked": 0,
+        "saved": 0,
+        "skipped": 0,
+    }
+
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        start = time.monotonic()
+
+        while True:
+            res = await geocode_missing_listings(db, batch_size=batch_size, dry_run=False)
+
+            # Аккумулируем counters
+            total.addresses_total += res.addresses_total
+            total.addresses_processed += res.addresses_processed
+            total.addresses_geocoded += res.addresses_geocoded
+            total.addresses_failed += res.addresses_failed
+            total.listings_updated += res.listings_updated
+            total.cache_hits += res.cache_hits
+            total.cache_misses += res.cache_misses
+
+            counters = {
+                "checked": total.addresses_processed,
+                "saved": total.listings_updated,
+                "skipped": total.addresses_failed,
+            }
+            runs_mod.update_heartbeat(db, run_id, counters)
+
+            elapsed = time.monotonic() - start
+            if res.addresses_total == 0:
+                logger.info(
+                    "run_geocode_missing_listings: run_id=%d — нет pending адресов, завершаем",
+                    run_id,
+                )
+                break
+            if res.addresses_total < batch_size:
+                logger.info(
+                    "run_geocode_missing_listings: run_id=%d — дренаж "
+                    "(addresses_total=%d < batch_size=%d), завершаем",
+                    run_id,
+                    res.addresses_total,
+                    batch_size,
+                )
+                break
+            if elapsed > budget_sec:
+                logger.info(
+                    "run_geocode_missing_listings: run_id=%d — бюджет %.0fs исчерпан "
+                    "(elapsed=%.1fs), завершаем",
+                    run_id,
+                    budget_sec,
+                    elapsed,
+                )
+                break
+
+        total.duration_sec = time.monotonic() - start
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "run_geocode_missing_listings: run_id=%d DONE — "
+            "processed=%d geocoded=%d failed=%d listings_updated=%d duration=%.1fs",
+            run_id,
+            total.addresses_processed,
+            total.addresses_geocoded,
+            total.addresses_failed,
+            total.listings_updated,
+            total.duration_sec,
+        )
+        return total
+
+    except Exception as exc:
+        total.duration_sec = time.monotonic() - start
+        logger.exception(
+            "run_geocode_missing_listings: run_id=%d FAILED after %.1fs",
+            run_id,
+            total.duration_sec,
+        )
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/110_scrape_schedules_seed_geocode_missing_listings.sql

@@ -0,0 +1,65 @@
+-- 110_scrape_schedules_seed_geocode_missing_listings.sql
+-- Seed row для nightly geocode backfill listings.lat/lon (все источники, все города).
+--
+-- Root cause (обнаружен 2026-06-16):
+--   Scrapers (avito, yandex, n1) сохраняют поисковые карточки с lat=lon=NULL
+--   в расчёте на то что «geocode-missing cron подтянет».
+--   OS crontab геокодирует ТОЛЬКО rosreestr_deals — для listings такого cron нет.
+--   На prod: 6959 avito + 590 yandex + 125 n1 = 7674 listings с null-geom при
+--   ненулевом address; geocode_tried_at IS NULL (ни разу не пробовали с 2026-05-23).
+--   Dry-run показал 85% geocode success (51/60) — wiring нотночного scheduler'а
+--   материально поднимает геопокрытие.
+--
+-- Решение:
+--   Вместо нового cron — wire существующую geocode_missing_listings в in-app scheduler.
+--   Функция address-dedup (1 geocode call на уникальный адрес → UPDATE all listings).
+--   Loop-safe: addresses помечаются geocode_tried_at=NOW() при failure → backoff 7 дней,
+--   нет бесконечного переотбора.
+--
+-- Окно 06:00-09:00 UTC (09:00-12:00 МСК):
+--   - После city sweeps (~03-04 UTC), которые depositing новые listings с null-geom.
+--   - После deals OS-geocode cron (05:30 / 05:45 UTC) — снижаем Nominatim contention.
+--   - До open-hours peaks на Nominatim (дневные задачи других систем).
+--
+-- default_params:
+--   batch_size  — addresses за один вызов geocode_missing_listings (200 ≈ 3.5 мин Nominatim).
+--   budget_sec  — wall-clock budget одного прогона; 1800 = 30 мин (fits в 3h window).
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + 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
+(
+    'geocode_missing_listings',
+    true,                                        -- ACTIVE: нет внешней зависимости (anti-bot),
+                                                 -- только Nominatim rate-limited geocoder
+    6,
+    9,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 6)) AT TIME ZONE 'UTC',
+    '{"batch_size": 200, "budget_sec": 1800}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет 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 СберИндекс city-level price index), '
+    'rosreestr_quarter_poll (#888, monthly Rosreestr new-quarter availability check), '
+    'cian_city_sweep (dormant, #973), '
+    'yandex_newbuilding_sweep (dormant, #974), '
+    'geocode_missing_listings (#1: listings geom backfill, all sources).';
+
+COMMIT;

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

@@ -1,9 +1,10 @@
 """Tests for geocode_missing_listings batch task."""
+
 from __future__ import annotations
 
 import os
 import sys
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 # DATABASE_URL required by config before any app import.
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
@@ -12,10 +13,14 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:
 _wp_mock = MagicMock()
 sys.modules.setdefault("weasyprint", _wp_mock)
 
-import pytest
+import pytest  # noqa: E402
 
-from app.services.geocoder import GeocodeResult
-from app.tasks.geocode_missing import GeocodeBackfillResult, geocode_missing_listings
+from app.services.geocoder import GeocodeResult  # noqa: E402
+from app.tasks.geocode_missing import (  # noqa: E402
+    GeocodeBackfillResult,
+    geocode_missing_listings,
+    run_geocode_missing_listings,
+)
 
 
 def _make_geocode_result(provider: str = "nominatim") -> GeocodeResult:
@@ -142,13 +147,16 @@ async def test_geocode_missing_cache_hit_counted() -> None:
 
 @pytest.mark.asyncio
 async def test_geocode_missing_skips_failed_addresses() -> None:
-    """Geocoder возвращает None → failed++, UPDATE не вызывается."""
+    """Geocoder возвращает None → failed++, lat UPDATE не вызывается, tried_at помечается."""
     rows = [{"address": "несуществующий адрес", "listings_count": 1}]
 
     db = MagicMock()
     select_result = MagicMock()
     select_result.mappings.return_value.all.return_value = rows
-    db.execute.return_value = select_result
+    tried_at_update_result = MagicMock()
+    tried_at_update_result.rowcount = 1
+    # execute side_effect: [SELECT, UPDATE tried_at]
+    db.execute.side_effect = [select_result, tried_at_update_result]
 
     with patch(
         "app.tasks.geocode_missing.geocode",
@@ -160,8 +168,8 @@ async def test_geocode_missing_skips_failed_addresses() -> None:
     assert result.addresses_failed == 1
     assert result.addresses_geocoded == 0
     assert result.listings_updated == 0
-    # commit не должен был вызываться после failed geocode
-    db.commit.assert_not_called()
+    # commit вызывается для tried_at UPDATE (loop-safe backoff)
+    db.commit.assert_called_once()
 
 
 @pytest.mark.asyncio
@@ -209,7 +217,7 @@ async def test_geocode_missing_batch_size_respected() -> None:
 
 @pytest.mark.asyncio
 async def test_geocode_missing_geocode_exception_counted_as_failed() -> None:
-    """Geocoder raises Exception → failed++, обработка продолжается."""
+    """Geocoder raises Exception → failed++, tried_at помечается, обработка продолжается."""
     rows = [
         {"address": "ул. Сломанная, 1", "listings_count": 1},
         {"address": "ул. Рабочая, 5", "listings_count": 2},
@@ -218,9 +226,12 @@ async def test_geocode_missing_geocode_exception_counted_as_failed() -> None:
     db = MagicMock()
     select_result = MagicMock()
     select_result.mappings.return_value.all.return_value = rows
+    tried_at_result = MagicMock()
+    tried_at_result.rowcount = 1
     update_result = MagicMock()
     update_result.rowcount = 2
-    db.execute.side_effect = [select_result, update_result]
+    # SELECT → tried_at UPDATE (exception path) → lat/lon UPDATE (success path)
+    db.execute.side_effect = [select_result, tried_at_result, update_result]
 
     geo_side_effects = [RuntimeError("network error"), _make_geocode_result("nominatim")]
 
@@ -236,6 +247,168 @@ async def test_geocode_missing_geocode_exception_counted_as_failed() -> None:
     assert result.listings_updated == 2
 
 
+# ── New tests: geocode_tried_at / loop-safe / run_geocode_missing_listings ───
+
+
+@pytest.mark.asyncio
+async def test_geocode_missing_failure_marks_tried_at() -> None:
+    """geo=None → UPDATE geocode_tried_at выполняется (loop-safe backoff)."""
+    rows = [{"address": "несуществующий адрес", "listings_count": 2}]
+
+    db = MagicMock()
+    select_result = MagicMock()
+    select_result.mappings.return_value.all.return_value = rows
+    tried_at_result = MagicMock()
+    db.execute.side_effect = [select_result, tried_at_result]
+
+    with patch(
+        "app.tasks.geocode_missing.geocode",
+        new_callable=AsyncMock,
+        return_value=None,
+    ):
+        result = await geocode_missing_listings(db, batch_size=200)
+
+    assert result.addresses_failed == 1
+    assert result.listings_updated == 0
+    # Второй execute — UPDATE tried_at
+    assert db.execute.call_count == 2
+    second_call_sql = str(db.execute.call_args_list[1][0][0])
+    assert "geocode_tried_at" in second_call_sql
+    db.commit.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_geocode_missing_exception_marks_tried_at() -> None:
+    """geocode() raises → UPDATE geocode_tried_at выполняется, commit вызывается."""
+    rows = [{"address": "ул. Битая, 99", "listings_count": 1}]
+
+    db = MagicMock()
+    select_result = MagicMock()
+    select_result.mappings.return_value.all.return_value = rows
+    tried_at_result = MagicMock()
+    db.execute.side_effect = [select_result, tried_at_result]
+
+    with patch(
+        "app.tasks.geocode_missing.geocode",
+        new_callable=AsyncMock,
+        side_effect=RuntimeError("timeout"),
+    ):
+        result = await geocode_missing_listings(db, batch_size=200)
+
+    assert result.addresses_failed == 1
+    assert db.execute.call_count == 2
+    second_call_sql = str(db.execute.call_args_list[1][0][0])
+    assert "geocode_tried_at" in second_call_sql
+    db.commit.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_geocode_missing_dry_run_no_tried_at_update() -> None:
+    """dry_run=True + geo=None → tried_at НЕ обновляется, commit не вызывается."""
+    rows = [{"address": "несуществующий адрес", "listings_count": 1}]
+
+    db = MagicMock()
+    select_result = MagicMock()
+    select_result.mappings.return_value.all.return_value = rows
+    db.execute.return_value = select_result
+
+    with patch(
+        "app.tasks.geocode_missing.geocode",
+        new_callable=AsyncMock,
+        return_value=None,
+    ):
+        result = await geocode_missing_listings(db, batch_size=200, dry_run=True)
+
+    assert result.addresses_failed == 1
+    # Только SELECT, никаких UPDATE
+    assert db.execute.call_count == 1
+    db.commit.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_geocode_missing_recent_tried_at_excluded_via_where() -> None:
+    """SELECT содержит geocode_tried_at IS NULL OR tried_at < NOW() - 7d (loop-safe WHERE)."""
+    db = MagicMock()
+    select_result = MagicMock()
+    select_result.mappings.return_value.all.return_value = []
+    db.execute.return_value = select_result
+
+    with patch("app.tasks.geocode_missing.geocode", new_callable=AsyncMock):
+        await geocode_missing_listings(db, batch_size=10)
+
+    first_call = db.execute.call_args_list[0]
+    sql_text = str(first_call[0][0])
+    assert "geocode_tried_at IS NULL" in sql_text
+    assert "7 days" in sql_text
+
+
+@pytest.mark.asyncio
+async def test_run_geocode_missing_listings_terminates_on_drained() -> None:
+    """run_geocode_missing_listings завершается когда addresses_total == 0 (ничего pending)."""
+    db = MagicMock()
+    drained = GeocodeBackfillResult(addresses_total=0)
+
+    with (
+        patch(
+            "app.tasks.geocode_missing.geocode_missing_listings",
+            new_callable=AsyncMock,
+            return_value=drained,
+        ),
+        patch("app.tasks.geocode_missing.runs_mod") as mock_runs,
+    ):
+        result = await run_geocode_missing_listings(db, run_id=42, params={})
+
+    mock_runs.update_heartbeat.assert_called()
+    mock_runs.mark_done.assert_called_once()
+    mock_runs.mark_failed.assert_not_called()
+    assert result.addresses_total == 0
+
+
+@pytest.mark.asyncio
+async def test_run_geocode_missing_listings_terminates_on_partial_batch() -> None:
+    """Последний batch меньше batch_size → завершаем (дренаж)."""
+    db = MagicMock()
+    partial = GeocodeBackfillResult(
+        addresses_total=50,  # < batch_size=200
+        addresses_processed=50,
+        listings_updated=30,
+    )
+
+    with (
+        patch(
+            "app.tasks.geocode_missing.geocode_missing_listings",
+            new_callable=AsyncMock,
+            return_value=partial,
+        ),
+        patch("app.tasks.geocode_missing.runs_mod") as mock_runs,
+    ):
+        result = await run_geocode_missing_listings(db, run_id=7, params={"batch_size": 200})
+
+    mock_runs.mark_done.assert_called_once()
+    assert result.addresses_processed == 50
+    assert result.listings_updated == 30
+
+
+@pytest.mark.asyncio
+async def test_run_geocode_missing_listings_mark_failed_on_exception() -> None:
+    """При исключении внутри loop: mark_failed вызывается, re-raise."""
+    db = MagicMock()
+
+    with (
+        patch(
+            "app.tasks.geocode_missing.geocode_missing_listings",
+            new_callable=AsyncMock,
+            side_effect=RuntimeError("DB down"),
+        ),
+        patch("app.tasks.geocode_missing.runs_mod") as mock_runs,
+    ):
+        with pytest.raises(RuntimeError, match="DB down"):
+            await run_geocode_missing_listings(db, run_id=99, params={})
+
+    mock_runs.mark_failed.assert_called_once()
+    mock_runs.mark_done.assert_not_called()
+
+
 # ── Integration-style: estimator Avito exclusion removed ─────────────────────
 
 
@@ -246,9 +419,9 @@ def test_estimator_fetch_analogs_includes_avito() -> None:
     from app.services import estimator
 
     source = inspect.getsource(estimator._fetch_analogs)
-    assert "source <> 'avito'" not in source, (
-        "Avito exclusion должен быть удалён из estimator._fetch_analogs"
-    )
+    assert (
+        "source <> 'avito'" not in source
+    ), "Avito exclusion должен быть удалён из estimator._fetch_analogs"
 
 
 # ── Admin endpoint smoke (schema-only, no real DB) ────────────────────────────
@@ -302,8 +475,13 @@ def test_admin_geocode_missing_post_dry_run_endpoint_exists() -> None:
         "app.tasks.geocode_missing.geocode_missing_listings",
         new_callable=AsyncMock,
         return_value=GeocodeBackfillResult(
-            addresses_total=0, addresses_processed=0, addresses_geocoded=0,
-            addresses_failed=0, listings_updated=0, cache_hits=0, cache_misses=0,
+            addresses_total=0,
+            addresses_processed=0,
+            addresses_geocoded=0,
+            addresses_failed=0,
+            listings_updated=0,
+            cache_hits=0,
+            cache_misses=0,
             duration_sec=0.01,
         ),
     ):