diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 4b2d672d..dffd5e16 100644 --- a/backend/app/workers/beat_schedule.py +++ b/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", diff --git a/backend/app/workers/tasks/scrape_cadastre.py b/backend/app/workers/tasks/scrape_cadastre.py index 4c1ec35d..9057f776 100644 --- a/backend/app/workers/tasks/scrape_cadastre.py +++ b/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) diff --git a/backend/tests/workers/test_scrape_cadastre.py b/backend/tests/workers/test_scrape_cadastre.py index fbcd6e1f..bb42a4bc 100644 --- a/backend/tests/workers/test_scrape_cadastre.py +++ b/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"