fix(ptica): упавший прогон Объектива помечается failed, а не висит running вечно (#2464)

Обработчик выглядел так:

    except Exception as e:
        if run_id:
            try:
                _finish_run(db, run_id, status="failed", ...)
            except Exception:
                pass          # ← молча

Если исходный сбой был DB-level (напр. INSERT в _save_raw), транзакция остаётся в
aborted-состоянии: _finish_run падает уже на своём execute, отказ гасится голым
pass, и строка прогона навсегда остаётся в status='running'.

Замер прода 20.08: шесть таких строк висят с 17.05 — 2274 часа, 95 суток. Уборщика
зомби для objective_scrape_runs нет (в отличие от cadastre, где он есть).

Правка: rollback перед _finish_run. Сессия здесь СВОЯ (SessionLocal() в этой же
функции, close в finally), поэтому плоский rollback законен — он отбрасывает уже
провалившуюся транзакцию и ничего чужого не теряет. Плюс отказ самого _finish_run
больше не молчит: если и после rollback не прошло, это логируется — знать об этом
важнее, чем сохранить тишину.

Тест на PostgresLikeSession — двойнике с настоящей семантикой aborted-транзакции.
На MagicMock он был бы зелёным по построению. Против origin/main:

  UPDATE статуса не выполнился, журнал SQL пуст  → падает
  сессия закрывается в finally      — контроль, зелёный с обеих сторон
  исходная ошибка пробрасывается    — контроль, зелёный с обеих сторон

Второй контроль не для симметрии: ловит «починку», которая заодно погасила бы
исключение — тогда Celery считал бы упавший прогон успешным.

Шесть уже висящих строк не трогаю: правка предотвращает новые, а чистка старых —
отдельное решение (данные прода, и на них ничего не завязано: с 17.05 прошёл 71
успешный прогон).

Прогоны: tests/workers — 226 passed rc=0.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/workers/tasks/scrape_objective.py

@@ -402,10 +402,31 @@ def sync_objective_group(
         }
     except Exception as e:
         if run_id:
+            # #2464: сессия здесь МОЖЕТ быть отравлена. Исходный сбой бывает
+            # DB-level (напр. INSERT в _save_raw), и тогда транзакция остаётся в
+            # aborted-состоянии: следующий execute падает, _finish_run не проходит,
+            # а голый `except Exception: pass` ниже гасил это молча — строка прогона
+            # навсегда оставалась в status='running'. Замер прода 20.08: шесть таких
+            # строк висят с 17.05, то есть 95 суток; уборщика зомби для
+            # objective_scrape_runs нет.
+            #
+            # Сессия здесь СВОЯ (SessionLocal() выше, close в finally), поэтому
+            # плоский rollback законен: он отбрасывает уже провалившуюся транзакцию
+            # и ничего чужого не теряет.
+            try:
+                db.rollback()
+            except Exception:
+                logger.exception("sync_objective_group: rollback перед _finish_run не удался")
             try:
                 _finish_run(db, run_id, status="failed", error=f"{type(e).__name__}: {e}")
             except Exception:
-                pass
+                # Больше не молча: если и это не прошло, строка останется 'running',
+                # и знать об этом важнее, чем сохранить тишину в логе.
+                logger.exception(
+                    "sync_objective_group: не удалось пометить run_id=%s как failed —"
+                    " строка останется в status='running'",
+                    run_id,
+                )
         raise
     finally:
         db.close()

backend/tests/workers/test_2464_objective_run_not_stuck.py

@@ -0,0 +1,136 @@
+"""Упавший прогон Объектива помечается failed, а не остаётся running навсегда (#2464).
+
+Обработчик в `sync_objective_group` выглядел так:
+
+    except Exception as e:
+        if run_id:
+            try:
+                _finish_run(db, run_id, status="failed", ...)
+            except Exception:
+                pass          # ← молча
+
+Если исходный сбой был DB-level (напр. INSERT в `_save_raw`), транзакция остаётся в
+aborted-состоянии: `_finish_run` падает уже на своём execute, это гасится голым `pass`,
+и строка прогона навсегда остаётся в `status='running'`.
+
+Замер прода 20.08.2026: шесть таких строк висят с 17.05 — 2274 часа, 95 суток. Уборщика
+зомби для `objective_scrape_runs` нет (в отличие от cadastre).
+
+Проверяется на `PostgresLikeSession` — двойнике с настоящей семантикой aborted-транзакции.
+На `MagicMock` тест был бы зелёным по построению: у него нет aborted-состояния, и любой
+следующий execute «успешен».
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from tests.support.pg_session import PostgresLikeSession
+
+_RUN_ID = 42
+
+
+class _Session(PostgresLikeSession):
+    """Двойник + close() и журнал УСПЕШНО выполненного SQL."""
+
+    def __init__(self, **kw: Any) -> None:
+        super().__init__(**kw)
+        self.sql: list[str] = []
+        self.closed = False
+
+    def execute(self, statement: Any = None, *a: Any, **kw: Any):  # type: ignore[no-untyped-def]
+        res = super().execute(statement, *a, **kw)
+        self.sql.append(str(statement))
+        return res
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def _run_with_poisoned_session() -> _Session:
+    """Прогон, где _start_url травит сессию, а тело падает следом."""
+    from app.workers.tasks import scrape_objective as mod
+
+    db = _Session(fail_on=(1,))
+
+    def _fake_start(session: Any, *_a: Any, **_kw: Any) -> int:
+        # Первый же execute падает → транзакция aborted (как DB-сбой в _save_raw).
+        try:
+            session.execute("INSERT INTO objective_scrape_runs ...")
+        except RuntimeError:
+            pass
+        return _RUN_ID
+
+    class _BoomClient:
+        def __init__(self, *_a: Any, **_kw: Any) -> None:
+            raise RuntimeError("сбой сразу после старта прогона")
+
+    with (
+        # Без ключа функция выходит на первой строке — подменяем, иначе тест
+        # проверял бы ранний return, а не обработчик ошибки.
+        patch.object(mod.settings, "objective_api_key", "test-key"),
+        patch.object(mod, "SessionLocal", lambda: db),
+        patch.object(mod, "_start_run", _fake_start),
+        patch.object(mod, "ObjectiveClient", _BoomClient),
+        pytest.raises(RuntimeError),
+    ):
+        mod.sync_objective_group(group_name="test", triggered_by="unit")
+
+    return db
+
+
+def _marked_failed(db: _Session) -> bool:
+    return any("objective_scrape_runs" in s and "status" in s for s in db.sql)
+
+
+def test_failed_run_is_marked_even_on_poisoned_session() -> None:
+    """Прогон обязан получить status='failed' даже когда транзакция отравлена.
+
+    На origin/main этого UPDATE в журнале нет: execute падает на aborted-сессии,
+    и голый `except Exception: pass` гасит отказ.
+    """
+    db = _run_with_poisoned_session()
+
+    assert _marked_failed(db), (
+        "UPDATE статуса прогона не выполнился — строка осталась в status='running' "
+        f"навсегда. Выполненный SQL: {db.sql}"
+    )
+
+
+def test_session_is_closed_anyway() -> None:
+    """Контроль: сессия закрывается в finally независимо от исхода."""
+    db = _run_with_poisoned_session()
+    assert db.closed, "сессия не закрыта — утечка соединения на аварийном пути"
+
+
+def test_original_error_still_propagates() -> None:
+    """Контроль: исходная ошибка не проглатывается — она и есть причина падения.
+
+    Ловит «починку», которая заодно погасила бы исключение: тогда Celery считал бы
+    прогон успешным.
+    """
+    from app.workers.tasks import scrape_objective as mod
+
+    db = _Session(fail_on=())
+
+    class _BoomClient:
+        def __init__(self, *_a: Any, **_kw: Any) -> None:
+            raise RuntimeError("характерный текст ошибки")
+
+    with (
+        patch.object(mod.settings, "objective_api_key", "test-key"),
+        patch.object(mod, "SessionLocal", lambda: db),
+        patch.object(mod, "_start_run", lambda *_a, **_kw: _RUN_ID),
+        patch.object(mod, "ObjectiveClient", _BoomClient),
+        pytest.raises(RuntimeError, match="характерный текст ошибки"),
+    ):
+        mod.sync_objective_group(group_name="test", triggered_by="unit")
+
+    assert _marked_failed(db), "на ЗДОРОВОЙ сессии пометка тем более обязана пройти"