fix(workers): cadastre permanent-fail targets_failed + beat cleanup (#1236)

backend/app/workers/beat_schedule.py

@@ -231,6 +231,16 @@ def build_beat_schedule() -> dict:
         "options": {"queue": "celery"},
     }
 
+    # Bug #1236: парный cleanup для cadastre_jobs. Переводит застрявшие
+    # 'running' jobs (heartbeat > 5 мин) в 'paused', чтобы оператор мог
+    # resume через admin API. Раз в 2 минуты — cadastre fanout меньше nspd_geo,
+    # per-minute опрос DB не оправдан.
+    schedule["cadastre-zombie-cleanup"] = {
+        "task": "tasks.cadastre.cleanup_zombies",
+        "schedule": _parse_cron("*/2 * * * *"),
+        "options": {"queue": "geo"},
+    }
+
     # OSM POI sync — еженедельно в понедельник в 03:00 МСК
     schedule["poi-sync-weekly"] = {
         "task": "tasks.poi_sync.sync_osm_poi_ekb",

backend/app/workers/tasks/scrape_cadastre.py

@@ -265,13 +265,25 @@ def bulk_harvest_quarter_task(self: Any, quarter: str, job_id: int) -> dict[str,
             job_id,
             e,
         )
+        # Bug #1236: финальная (после исчерпания autoretry) попытка обязана
+        # учесть квартал в targets_failed и попытаться закрыть job. Иначе
+        # done+failed < total → _maybe_finish_job не срабатывает → cadastre_jobs
+        # вечно 'running' с протухшим heartbeat. Промежуточные autoretry-попытки
+        # (retries < max_retries) счётчики НЕ трогают — Celery re-enqueue'ит task,
+        # финальный исход учтётся ровно один раз.
+        max_retries = getattr(self, "max_retries", 0) or 0
+        current_retries = getattr(getattr(self, "request", None), "retries", 0) or 0
+        is_final_attempt = current_retries >= max_retries
         try:
             _update_job(
                 db,
                 job_id,
                 last_error=f"{type(e).__name__}: {e}"[:500],
+                targets_failed_inc=1 if is_final_attempt else 0,
                 heartbeat=True,
             )
+            if is_final_attempt:
+                _maybe_finish_job(db, job_id)
             db.commit()
         except Exception as inner_e:
             logger.warning("failed to update job state: %s", inner_e)

backend/tests/workers/test_scrape_cadastre.py

@@ -1,14 +1,20 @@
 """Тесты для backend/app/workers/tasks/scrape_cadastre.py.
 
-Покрывает Fix H — per-quarter freshness skip в enqueue_cadastre_harvest.
+Покрывает:
+- Fix H — per-quarter freshness skip в enqueue_cadastre_harvest.
+- Bug #1236 — генерик-фейл инкрементит targets_failed только на финальной
+  попытке + регистрирует cleanup_cadastre_zombies в beat_schedule.
 """
 
 from __future__ import annotations
 
 import datetime as _dt
+from types import SimpleNamespace
 from typing import Any
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 
 def _make_mock_job(quarters: list[str], raw_targets_extra: dict[str, Any] | None = None) -> dict:
     """Mock cadastre_jobs row."""
@@ -130,3 +136,136 @@ def test_enqueue_freshness_threshold_passed_in_sql() -> None:
     # Threshold должен быть приблизительно 2h назад (±5s tolerance)
     delta = now - threshold
     assert _dt.timedelta(hours=1, minutes=59) < delta < _dt.timedelta(hours=2, minutes=1)
+
+
+# ── Bug #1236: permanent non-WAF fail counter ──────────────────────────────
+
+
+def _make_counter_db() -> tuple[MagicMock, list[dict[str, Any]]]:
+    """Mock db, capturing UPDATE cadastre_jobs param dicts in chronological order."""
+    captured: list[dict[str, Any]] = []
+    db = MagicMock()
+
+    def execute_router(stmt: Any, params: Any = None) -> MagicMock:
+        sql = str(stmt).upper()
+        result = MagicMock()
+        if "UPDATE CADASTRE_JOBS" in sql:
+            captured.append({"sql": sql, "params": params or {}})
+            result.rowcount = 1
+        elif "SELECT TARGETS_TOTAL" in sql:
+            # _maybe_finish_job: эмулируем done+failed < total → finish не дёрнется,
+            # тест проверяет ТОЛЬКО факт вызова + targets_failed_inc.
+            result.mappings.return_value.first.return_value = {
+                "targets_total": 10,
+                "targets_done": 0,
+                "targets_failed": 1,
+                "status": "running",
+            }
+        else:
+            result.mappings.return_value.first.return_value = None
+            result.scalars.return_value.all.return_value = []
+        return result
+
+    db.execute = execute_router
+    return db, captured
+
+
+def test_bulk_harvest_quarter_final_fail_increments_targets_failed() -> None:
+    """Bug #1236: на финальной (retries == max_retries) попытке non-WAF фейл
+    инкрементит targets_failed_inc=1 и вызывает _maybe_finish_job."""
+    from app.workers.tasks import scrape_cadastre
+
+    db, captured = _make_counter_db()
+    # Эмулируем `self` Celery task: финальная попытка autoretry.
+    self_mock = SimpleNamespace(
+        max_retries=3,
+        request=SimpleNamespace(retries=3),
+    )
+
+    finish_called: list[int] = []
+
+    def fake_finish(_db: Any, job_id: int) -> None:
+        finish_called.append(job_id)
+
+    with (
+        patch.object(scrape_cadastre, "SessionLocal", return_value=db),
+        patch.object(scrape_cadastre, "_maybe_finish_job", side_effect=fake_finish),
+        patch.object(
+            scrape_cadastre,
+            "_is_job_cancelled",
+            return_value=False,
+        ),
+        patch(
+            "app.services.cadastre.bulk_harvest.harvest_quarter",
+            side_effect=RuntimeError("boom"),
+        ),
+    ):
+        with pytest.raises(RuntimeError):
+            # Вызываем underlying function через .run (не apply_async).
+            scrape_cadastre.bulk_harvest_quarter_task.__wrapped__.__func__(
+                self_mock, "66:41:0001001", 42
+            )
+
+    # В except-ветке ровно один UPDATE с targets_failed_inc=1 и last_error.
+    fail_updates = [
+        c for c in captured if c["params"].get("failed_inc") == 1 and "last_error" in c["params"]
+    ]
+    assert len(fail_updates) == 1, f"expected single fail-update, got: {captured}"
+    assert fail_updates[0]["params"]["id"] == 42
+    assert finish_called == [42], "_maybe_finish_job must be called on final attempt"
+
+
+def test_bulk_harvest_quarter_intermediate_fail_does_not_increment() -> None:
+    """Bug #1236: промежуточная (retries < max_retries) попытка НЕ трогает
+    targets_failed_inc и НЕ вызывает _maybe_finish_job — Celery re-enqueue'ит."""
+    from app.workers.tasks import scrape_cadastre
+
+    db, captured = _make_counter_db()
+    self_mock = SimpleNamespace(
+        max_retries=3,
+        request=SimpleNamespace(retries=1),  # ещё две попытки впереди
+    )
+
+    finish_called: list[int] = []
+
+    with (
+        patch.object(scrape_cadastre, "SessionLocal", return_value=db),
+        patch.object(
+            scrape_cadastre,
+            "_maybe_finish_job",
+            side_effect=lambda _db, jid: finish_called.append(jid),
+        ),
+        patch.object(scrape_cadastre, "_is_job_cancelled", return_value=False),
+        patch(
+            "app.services.cadastre.bulk_harvest.harvest_quarter",
+            side_effect=RuntimeError("transient"),
+        ),
+    ):
+        with pytest.raises(RuntimeError):
+            scrape_cadastre.bulk_harvest_quarter_task.__wrapped__.__func__(
+                self_mock, "66:41:0001002", 43
+            )
+
+    fail_updates = [c for c in captured if c["params"].get("failed_inc")]
+    assert fail_updates == [], "intermediate retry must NOT bump targets_failed"
+    assert finish_called == [], "intermediate retry must NOT call _maybe_finish_job"
+
+
+# ── Bug #1236: beat schedule registers cadastre-zombie-cleanup ─────────────
+
+
+def test_beat_schedule_includes_cadastre_zombie_cleanup() -> None:
+    """Bug #1236: cleanup_cadastre_zombies должен быть зарегистрирован в beat,
+    иначе застрявший в 'running' job никто не переведёт в 'paused'."""
+    from app.workers import beat_schedule
+
+    # build_beat_schedule читает DB → мокаем _build_beat_schedule_from_db,
+    # чтобы тест не зависел от живого Postgres.
+    with patch.object(beat_schedule, "_build_beat_schedule_from_db", return_value={}):
+        schedule = beat_schedule.build_beat_schedule()
+
+    assert "cadastre-zombie-cleanup" in schedule, (
+        f"cadastre-zombie-cleanup отсутствует в beat: {sorted(schedule.keys())}"
+    )
+    entry = schedule["cadastre-zombie-cleanup"]
+    assert entry["task"] == "tasks.cadastre.cleanup_zombies"