From 816bb3d3bd6d61b48e539464d6aa97c30942783d Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 13 Jun 2026 13:04:58 +0500 Subject: [PATCH] =?UTF-8?q?fix(workers):=20re-raise=20SoftTimeLimitExceede?= =?UTF-8?q?d=20=D0=B2=20nspd=5Fgeo,=20set=20paused=20(#1235)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/workers/tasks/nspd_geo.py | 52 +++++++ backend/tests/workers/test_nspd_geo.py | 181 +++++++++++++++++++++++++ 2 files changed, 233 insertions(+) diff --git a/backend/app/workers/tasks/nspd_geo.py b/backend/app/workers/tasks/nspd_geo.py index 97462ad3..418c576f 100644 --- a/backend/app/workers/tasks/nspd_geo.py +++ b/backend/app/workers/tasks/nspd_geo.py @@ -23,6 +23,7 @@ import json import logging from typing import Any +from celery.exceptions import SoftTimeLimitExceeded from sqlalchemy import text from sqlalchemy.orm import Session @@ -492,6 +493,52 @@ def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]: return {"job_id": job_id, "paused": True, "reason": "waf"} continue + except SoftTimeLimitExceeded: + # Celery soft_time_limit=6h истёк — graceful shutdown ДО generic except. + # Issue #1235: `except (NspdLiteError, Exception)` ниже проглатывал бы это + # исключение (SoftTimeLimitExceeded унаследован от Exception) → таска + # продолжала бы цикл неограниченно дольше 6ч, занимая слот geo-очереди. + # Контракт: коммитим финальный heartbeat + статусы targets, помечаем job + # 'paused' (auto-resume через cleanup_zombies/worker_ready), re-raise чтобы + # Celery освободил воркер и зарегистрировал лимит. + logger.warning( + "process_nspd_geo_job soft time limit exceeded: job=%s done=%d failed=%d", + job_id, + done, + failed, + ) + try: + _heartbeat( + db, + job_id, + targets_done=done, + targets_failed=failed, + targets_skipped=skipped, + requests_count=n_requests, + waf_blocked_count=n_waf, + ) + _finish_job( + db, + job_id, + "paused", + error=f"soft_time_limit exceeded after {done + failed} targets", + ) + _log( + db, + job_id, + "warn", + "paused", + f"job paused: Celery soft_time_limit (6h) reached " + f"after {done} done / {failed} failed", + ) + except Exception as flush_err: + # Финальный flush не должен скрыть SoftTimeLimitExceeded — лог + re-raise. + logger.exception( + "process_nspd_geo_job: failed to flush state on soft-limit: %s", + flush_err, + ) + raise + except (NspdLiteError, Exception) as e: logger.warning("fetch failed for %s: %s", cad, e) _log(db, job_id, "warn", "fetch_fail", str(e)[:300], cad_num=cad) @@ -556,6 +603,11 @@ def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]: "requests": n_requests, "waf_blocked": n_waf, } + except SoftTimeLimitExceeded: + # Inner handler уже коммитнул heartbeat + статус 'paused' + лог. + # Не перетираем 'paused' на 'failed' через outer fallback — просто пробрасываем, + # чтобы Celery освободил worker и зарегистрировал срабатывание soft_time_limit. + raise except Exception as e: logger.exception("process_nspd_geo_job crashed: job=%s: %s", job_id, e) try: diff --git a/backend/tests/workers/test_nspd_geo.py b/backend/tests/workers/test_nspd_geo.py index 7ea5816f..2cbf6802 100644 --- a/backend/tests/workers/test_nspd_geo.py +++ b/backend/tests/workers/test_nspd_geo.py @@ -14,6 +14,9 @@ from __future__ import annotations from typing import Any from unittest.mock import MagicMock +import pytest +from celery.exceptions import SoftTimeLimitExceeded + from app.workers.tasks import nspd_geo as nspd_geo_mod from app.workers.tasks.nspd_geo import ( _ZOMBIE_PAUSED_THRESHOLD, @@ -22,6 +25,7 @@ from app.workers.tasks.nspd_geo import ( _save_parcel, _save_quarter, cleanup_zombies, + process_nspd_geo_job, ) # ── Фикстуры payload (форма ответа _persist_target: {"data": {"features": [...]}}) ── @@ -414,3 +418,180 @@ def test_cleanup_zombies_paused_threshold_strictly_greater_than_running( # Sanity: SQL действительно отправлен с двумя разными порогами _sql, params = captured[0] assert params["running_threshold"] != params["paused_threshold"] + + +# ── SoftTimeLimitExceeded handler (issue #1235) ───────────────────────────── + + +class _JobFetchResult: + """sqlalchemy-like Result, возвращающий заданный mapping для SELECT FROM nspd_geo_jobs.""" + + def __init__(self, row: dict[str, Any] | None) -> None: + self._row = row + + def mappings(self) -> _JobFetchResult: + return self + + def first(self) -> dict[str, Any] | None: + return self._row + + def scalar_one(self) -> Any: + return 1 + + +def _build_soft_limit_session( + raise_in: str = "fetch", +) -> tuple[MagicMock, list[tuple[str, dict[str, Any]]]]: + """Fake SessionLocal() для теста graceful shutdown по SoftTimeLimitExceeded. + + Контракт: первый SELECT ... FROM nspd_geo_jobs WHERE job_id=... → job row, + UPDATE _start_job → OK, INSERT nspd_geo_log → OK, второй SELECT (target loop) → + pending target row. Затем fetcher кидает SoftTimeLimitExceeded. + + После raise — handler должен (a) выполнить UPDATE heartbeat + counters, + (b) UPDATE nspd_geo_jobs SET status='paused', (c) re-raise. + """ + captured: list[tuple[str, dict[str, Any]]] = [] + job_row = { + "job_id": 42, + "rate_ms": 0, # отключаем sleep между итерациями для теста + "targets_done": 0, + "targets_failed": 0, + "targets_skipped": 0, + "requests_count": 0, + "waf_blocked_count": 0, + } + target_row = { + "target_id": 7, + "cad_num": "66:41:0610029:83", + "thematic_id": 1, + "attempts": 0, + } + call_state = {"job_selects": 0, "target_selects": 0} + + def _execute(stmt: Any, params: Any = None) -> Any: + sql = str(stmt) + captured.append((sql, params or {})) + if "FROM nspd_geo_jobs" in sql and "SELECT" in sql: + call_state["job_selects"] += 1 + return _JobFetchResult(job_row) + if "FROM nspd_geo_targets" in sql and "SELECT" in sql: + call_state["target_selects"] += 1 + # Первый SELECT — отдаём pending target. Дальше — None (loop break), + # но до второй итерации не дойдём — fetcher кинет SoftTimeLimitExceeded. + if call_state["target_selects"] == 1: + return _JobFetchResult(target_row) + return _JobFetchResult(None) + if "INSERT INTO nspd_geo_jobs" in sql: + return _JobFetchResult({"job_id": 42}) # для scalar_one + return MagicMock() + + session = MagicMock() + session.execute = _execute + session.commit = MagicMock() + session.rollback = MagicMock() + session.close = MagicMock() + return MagicMock(return_value=session), captured + + +def test_soft_time_limit_exceeded_reraises_not_swallowed(monkeypatch: Any) -> None: + """Issue #1235 root: `except (NspdLiteError, Exception)` проглатывало SoftTimeLimitExceeded. + + После fix — explicit `except SoftTimeLimitExceeded: raise` ДО generic ветки → + исключение пробрасывается наружу process_nspd_geo_job, Celery освобождает воркер. + """ + factory, _captured = _build_soft_limit_session() + monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory) + + def _raise_soft_limit(*_args: Any, **_kwargs: Any) -> Any: + raise SoftTimeLimitExceeded() + + monkeypatch.setattr(nspd_geo_mod, "fetch_via_rosreestr2coord", _raise_soft_limit) + + with pytest.raises(SoftTimeLimitExceeded): + process_nspd_geo_job(42) + + +def test_soft_time_limit_exceeded_sets_paused_status(monkeypatch: Any) -> None: + """Handler коммитит status='paused' перед re-raise — иначе job навечно 'running' + с stale heartbeat → cleanup_zombies ре-enqueue'ит через running-threshold (6 мин), + но мы хотим явный paused + 30 мин cooldown. + """ + factory, captured = _build_soft_limit_session() + monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory) + monkeypatch.setattr( + nspd_geo_mod, + "fetch_via_rosreestr2coord", + MagicMock(side_effect=SoftTimeLimitExceeded()), + ) + + with pytest.raises(SoftTimeLimitExceeded): + process_nspd_geo_job(42) + + # Должен быть UPDATE с status = :st где :st == 'paused' + paused_updates = [ + (sql, params) + for sql, params in captured + if "UPDATE nspd_geo_jobs" in sql and "status = :st" in sql and params.get("st") == "paused" + ] + assert paused_updates, ( + "SoftTimeLimitExceeded handler должен выставить status='paused' до re-raise; " + f"captured updates: {[(s[:80], p) for s, p in captured if 'UPDATE nspd_geo_jobs' in s]}" + ) + + +def test_soft_time_limit_exceeded_flushes_heartbeat(monkeypatch: Any) -> None: + """Handler коммитит финальный heartbeat (counters + timestamp) до raise. + + Иначе после паузы job выглядит в UI с устаревшими счётчиками — admin не видит + сколько таргетов реально обработано. + """ + factory, captured = _build_soft_limit_session() + monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory) + monkeypatch.setattr( + nspd_geo_mod, + "fetch_via_rosreestr2coord", + MagicMock(side_effect=SoftTimeLimitExceeded()), + ) + + with pytest.raises(SoftTimeLimitExceeded): + process_nspd_geo_job(42) + + heartbeat_updates = [ + (sql, params) + for sql, params in captured + if "heartbeat_at = NOW()" in sql and "targets_done" in sql + ] + assert heartbeat_updates, ( + "SoftTimeLimitExceeded handler должен flush'нуть heartbeat с counters перед raise" + ) + + +def test_soft_time_limit_exceeded_does_not_overwrite_paused_with_failed( + monkeypatch: Any, +) -> None: + """Outer `except Exception` НЕ должен затереть 'paused' на 'failed' для SoftTimeLimitExceeded. + + Issue #1235 corollary: если inner handler выставил 'paused' и raise — outer + fallback не должен мгновенно переписать status='failed' через _finish_job. + """ + factory, captured = _build_soft_limit_session() + monkeypatch.setattr(nspd_geo_mod, "SessionLocal", factory) + monkeypatch.setattr( + nspd_geo_mod, + "fetch_via_rosreestr2coord", + MagicMock(side_effect=SoftTimeLimitExceeded()), + ) + + with pytest.raises(SoftTimeLimitExceeded): + process_nspd_geo_job(42) + + failed_updates = [ + params + for sql, params in captured + if "UPDATE nspd_geo_jobs" in sql and "status = :st" in sql and params.get("st") == "failed" + ] + assert not failed_updates, ( + "outer except не должен затирать 'paused' на 'failed' для SoftTimeLimitExceeded; " + f"got failed updates: {failed_updates}" + )