diff --git a/backend/app/services/forecasting/macro_series.py b/backend/app/services/forecasting/macro_series.py index fd8e3874..2143e5cc 100644 --- a/backend/app/services/forecasting/macro_series.py +++ b/backend/app/services/forecasting/macro_series.py @@ -371,12 +371,20 @@ def _month_grid(start: date, end: date) -> list[date]: def _query_key_rate_monthly(db: Session, *, since: date) -> dict[date, float]: - """Ресэмпл дневного key_rate (region 'rf') → {month1st: value}. Graceful → {}.""" + """Ресэмпл дневного key_rate (region 'rf') → {month1st: value}. Graceful → {}. + + SAVEPOINT (#2464 cluster A finding #2): `db` — общая §22-сессия отчёта; при сбое + этого запроса БЕЗ SAVEPOINT транзакция Postgres остаётся aborted и следующие + запросы get_monthly_macro (inflation, mortgage) + все ПОСЛЕДУЮЩИЕ §9.x-слои на + той же сессии тоже падают. `with db.begin_nested():` откатывает ТОЛЬКО этот + SAVEPOINT (ROLLBACK TO SAVEPOINT), внешняя транзакция остаётся рабочей. + """ try: - rows = db.execute( - _KEY_RATE_MONTHLY_SQL, - {"itype": "key_rate", "region": "rf", "since": since}, - ).all() + with db.begin_nested(): + rows = db.execute( + _KEY_RATE_MONTHLY_SQL, + {"itype": "key_rate", "region": "rf", "since": since}, + ).all() except Exception: logger.exception("get_monthly_macro: key_rate query failed") return {} @@ -389,9 +397,14 @@ def _query_inflation_monthly(db: Session, *, since: date) -> dict[date, float]: Ряд УЖЕ месячный (obs_date = 1-е число, залит cbr_macro_sync) — берём как есть через reuse get_macro_series (свой SQL не пишем). _month_start — страховка. Сбой/пустой ряд → {} (НЕ crash), inflation_yoy тогда None по всей сетке. + + SAVEPOINT (#2464 cluster A finding #2): см. `_query_key_rate_monthly` — та же + общая §22-сессия, тот же риск отравления транзакции для последующих запросов + (mortgage-поля + §9.x-слои). `with db.begin_nested():` изолирует сбой в SAVEPOINT. """ try: - series = get_macro_series(db, "inflation_yoy", region="rf", since=since) + with db.begin_nested(): + series = get_macro_series(db, "inflation_yoy", region="rf", since=since) except Exception: logger.exception("get_monthly_macro: inflation_yoy query failed") return {} @@ -404,11 +417,20 @@ def _query_mortgage_monthly(db: Session, *, since: date) -> dict[str, dict[date, Возвращает {field: {month1st: value}}. obs_date уже нормализован к 1-му числу в backfill, но _month_start применяем повторно (страховка). Сбой одного ряда не валит остальные (graceful: пустой подсловарь). + + SAVEPOINT (#2464 cluster A finding #2): все 5 полей читаются на ОДНОЙ `db`-Session + (get_monthly_macro вызывается внутри общей §22-сессии отчёта). Без SAVEPOINT сбой + ОДНОГО поля оставляет транзакцию Postgres aborted — каждое СЛЕДУЮЩЕЕ поле в этом + же цикле тоже падает (хотя его данные были бы доступны), а `except` здесь молча + отдаёт [] по каждому, маскируя каскад под «нормальную» построчную деградацию. + `with db.begin_nested():` — SAVEPOINT на КАЖДОЕ поле: сбой откатывает только его + SAVEPOINT (ROLLBACK TO SAVEPOINT), сессия остаётся рабочей для следующего поля. """ out: dict[str, dict[date, float]] = {} for indicator_type, field in _MORTGAGE_FIELDS: try: - series = get_macro_series(db, indicator_type, region="sverdl", since=since) + with db.begin_nested(): + series = get_macro_series(db, indicator_type, region="sverdl", since=since) except Exception: logger.exception("get_monthly_macro: mortgage series %s failed", indicator_type) series = [] diff --git a/backend/app/services/forecasting/orchestrator.py b/backend/app/services/forecasting/orchestrator.py index 05e07a9f..6675f780 100644 --- a/backend/app/services/forecasting/orchestrator.py +++ b/backend/app/services/forecasting/orchestrator.py @@ -124,7 +124,7 @@ def _primary_horizon(horizons: Sequence[int]) -> int: return horizons[0] if horizons else _PREFERRED_PRIMARY_HORIZON -def _safe_call(label: str, fn: Any) -> Any: +def _safe_call(label: str, db: Session, fn: Any) -> Any: """Вызвать §9.x-сервис graceful: сбой → None + logger.exception (не crash отчёта). Зеркало product_scoring._safe_call: любой §9.x-слой может бросить (тонкие данные / нет @@ -133,15 +133,29 @@ def _safe_call(label: str, fn: Any) -> Any: широкий Exception (изоляция одного слоя от отчёта) с ОБЯЗАТЕЛЬНЫМ logger.exception — НЕ молчаливое глотание. §9.x уже graceful внутри; это belt-and-suspenders на шве. + SAVEPOINT (#2464 cluster A finding #1): все §9.x-слои шарят ОДИН `db`-Session на + отчёт (module docstring `forecast_request_cache.py`). Без обёртки сбойный + `db.execute` внутри слоя оставляет транзакцию Postgres в состоянии aborted + («current transaction is aborted, commands ignored until end of transaction + block») — КАЖДЫЙ последующий слой на той же сессии тоже падает, хотя его данные + были бы доступны. `with db.begin_nested():` заводит SAVEPOINT НА ВЕСЬ вызов слоя + (слой может делать несколько `db.execute` внутри себя — например §9.6 внутри + §9.8/§11); при исключении SAVEPOINT откатывается автоматически (ROLLBACK TO + SAVEPOINT), внешняя транзакция остаётся рабочей для следующего слоя. НЕ + `db.rollback()` — тот откатил бы ВСЮ внешнюю транзакцию (см. `backend.md` § + SAVEPOINT pattern, established anti-pattern). + Args: label: имя слоя для лога (диагностика какой §9.x-вызов деградировал). + db: общая §22-сессия (для SAVEPOINT вокруг вызова слоя). fn: нулевой-аргумент thunk вокруг §9.x-вызова. Returns: Результат `fn()` или None при исключении. """ try: - return fn() + with db.begin_nested(): + return fn() except Exception: logger.exception("orchestrator: §9.x layer %s failed → section degraded", label) return None @@ -334,21 +348,25 @@ def _build_site_finder_report_impl( # ── 2. §9.x-слои, каждый graceful через _safe_call (ГЕТЕРОГЕННЫЕ сигнатуры) ── market_metrics = _safe_call( "market_metrics", + db, lambda: compute_market_metrics(db, district=district, premise_kind=_PREMISE_KIND), ) supply_rows = _safe_call( "supply_layers", + db, lambda: compute_all_layers(db, district=district, premise_kind=_PREMISE_KIND), ) supply_layers = _summarize_supply_layers(supply_rows) future_supply = _safe_call( "future_supply", + db, lambda: compute_future_supply_pressure( db, district=district, horizon_months=primary, premise_kind=_PREMISE_KIND ), ) forecasts = _safe_call( "demand_supply_forecast", + db, lambda: compute_demand_supply_forecast( db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list ), @@ -359,6 +377,7 @@ def _build_site_finder_report_impl( # _safe_call оборачивает любой сбой → None → штатно деградируем (collapse=False). scenarios_result = _safe_call( "scenarios", + db, lambda: compute_scenarios( db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list ), @@ -371,18 +390,21 @@ def _build_site_finder_report_impl( scenarios, scenarios_collapsed, scenarios_collapse_reason = scenarios_result product_scores = _safe_call( "score_card", + db, lambda: compute_score_card( db, spec=spec, district=district, cad_num=cad_num, horizon_months=primary ), ) special_indices = _safe_call( "special_indices", + db, lambda: compute_special_indices( db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list ), ) recommendation_overlay = _safe_call( "forecast_overlay", + db, lambda: build_forecast_overlay( db, district=district, @@ -393,7 +415,7 @@ def _build_site_finder_report_impl( ) # ── Макро-свежесть (audit MEDIUM): только лог; проводка в отчёт — 3b ───────── - macro = _safe_call("monthly_macro", lambda: get_monthly_macro(db)) + macro = _safe_call("monthly_macro", db, lambda: get_monthly_macro(db)) macro_as_of = _macro_as_of(macro) if macro_as_of is not None: logger.info( diff --git a/backend/app/services/forecasting/sales_series.py b/backend/app/services/forecasting/sales_series.py index a5c04295..c0c2630b 100644 --- a/backend/app/services/forecasting/sales_series.py +++ b/backend/app/services/forecasting/sales_series.py @@ -549,6 +549,12 @@ def _query_source_a( Graceful → {} при сбое/пустых данных. price_bucket в spec для Source A игнорируется (агрегат не несёт per-lot цены) — фиксируется логом. + + SAVEPOINT (#2464 cluster A finding #4): `db` — общая §22-сессия отчёта (может + переиспользоваться другими §9.x-слоями после этого вызова). Без SAVEPOINT сбойный + `db.execute` оставляет транзакцию Postgres aborted — все ПОСЛЕДУЮЩИЕ запросы на + той же сессии тоже падают. `with db.begin_nested():` откатывает ТОЛЬКО этот + SAVEPOINT при сбое, оставляя внешнюю транзакцию рабочей. """ if spec.price_bucket is not None: logger.info( @@ -567,7 +573,8 @@ def _query_source_a( "room_bucket": spec.room_bucket, } try: - rows = db.execute(_SOURCE_A_SQL, params).mappings().all() + with db.begin_nested(): + rows = db.execute(_SOURCE_A_SQL, params).mappings().all() except Exception: logger.exception("build_sales_series: source A query failed") return {} @@ -582,6 +589,9 @@ def _query_source_b( Graceful → {} при сбое/пустых данных. Передаёт bucket-пороги/-метки в SQL (зеркало pure-helpers), чтобы room×area / price сегментация считалась тем же правилом и в БД, и в Python. + + SAVEPOINT (#2464 cluster A finding #4): см. `_query_source_a` — та же общая + §22-сессия, тот же риск отравления транзакции для последующих слоёв/запросов. """ # district (админ-имя ЕКБ) → набор informal микро (objective_lots хранит микро). # None → EKB-wide (без district-фильтра). @@ -613,7 +623,8 @@ def _query_source_b( "p_unknown": PRICE_BUCKET_UNKNOWN, } try: - rows = db.execute(_SOURCE_B_SQL, params).mappings().all() + with db.begin_nested(): + rows = db.execute(_SOURCE_B_SQL, params).mappings().all() except Exception: logger.exception("build_sales_series: source B query failed") return {} diff --git a/backend/app/services/forecasting/special_indices.py b/backend/app/services/forecasting/special_indices.py index b34fc5cf..58142935 100644 --- a/backend/app/services/forecasting/special_indices.py +++ b/backend/app/services/forecasting/special_indices.py @@ -994,16 +994,29 @@ def _query_parcel_centroid(db: Session, *, cad_num: str) -> tuple[float, float] Нет геометрии / сбой → None (гео-веса упадут на floor; overlap считается по остальным осям, индекс НЕ деградирует целиком). Параметризовано (psycopg v3). Детерминированно. + + SAVEPOINT НА ВНУТРЕННЕМ swallow-сайте (#2464 cluster A, RELEASE-trap): этот helper + зовётся из `_build_cannibalization` — builder, обёрнутый внешним SAVEPOINT в + `compute_special_indices._run`. БЕЗ собственного `db.begin_nested():` сбойный + `db.execute` тут проглатывается локально (→ None), оставляя транзакцию Postgres + aborted; тогда ВНЕШНИЙ `_run`-savepoint выходит «нормально» и делает RELEASE + SAVEPOINT, а RELEASE в aborted-tx САМ падает (Postgres в aborted допускает только + ROLLBACK / ROLLBACK TO SAVEPOINT) — внешний except ловит уже RELEASE-ошибку, но + транзакция так и не откачена → отравление каскадит в следующий §25-индекс. Savepoint + ИМЕННО ЗДЕСЬ (в точке перехвата) откатывает сбой (ROLLBACK TO SAVEPOINT), поэтому + внешний RELEASE проходит. Тот же принцип, что saturation.py: savepoint у db.execute + в точке, где ошибка ловится. """ try: - row = ( - db.execute( - _PARCEL_CENTROID_SQL, - {"cad_num": cad_num, "quarter": _quarter_from_cad(cad_num)}, + with db.begin_nested(): + row = ( + db.execute( + _PARCEL_CENTROID_SQL, + {"cad_num": cad_num, "quarter": _quarter_from_cad(cad_num)}, + ) + .mappings() + .first() ) - .mappings() - .first() - ) except Exception: logger.exception("cannibalization: centroid query failed for cad_num=%s", cad_num) return None @@ -1657,9 +1670,20 @@ def compute_special_indices( segment = spec.as_dict() def _run(key: str, builder: Any) -> SpecialIndex: - """Выполнить builder одного индекса в собственном try/except (graceful).""" + """Выполнить builder одного индекса в собственном try/except (graceful). + + SAVEPOINT (#2464 cluster A finding #3): все шесть builder'ов делят ОДНУ `db`- + Session (общая сессия §22-отчёта). Без SAVEPOINT сбойный `db.execute` внутри + builder'а оставляет транзакцию Postgres aborted — КАЖДЫЙ следующий индекс на + той же сессии тоже падает (хотя его данные были бы доступны), что маскируется + под шесть независимых деградаций. `with db.begin_nested():` заводит SAVEPOINT + на ВЕСЬ builder() (некоторые builder'ы делают несколько db.execute внутри + себя) — сбой откатывает только его SAVEPOINT, сессия остаётся рабочей для + следующего индекса. + """ try: - return builder() # type: ignore[no-any-return] + with db.begin_nested(): + return builder() # type: ignore[no-any-return] except Exception: # Сбой одного индекса НЕ валит карточку: деградация-None, остальные считаются. logger.exception( diff --git a/backend/app/services/site_finder/district_resolver.py b/backend/app/services/site_finder/district_resolver.py index b8b487fa..ee309f8a 100644 --- a/backend/app/services/site_finder/district_resolver.py +++ b/backend/app/services/site_finder/district_resolver.py @@ -68,12 +68,22 @@ def _admin_names(db: Session) -> frozenset[str]: На ошибке БД возвращает пустой set (logged) — резолвер тогда трактует ЛЮБОЙ вход как raw-микро (legacy pass-through), что безопаснее, чем уронить /analyze. + + SAVEPOINT НА ВНУТРЕННЕМ swallow-сайте (#2464 cluster A, RELEASE-trap): резолвер + зовётся §22-форсайт-билдерами (sales_series `_query_source_a/_b`, + `_query_artificial_demand`) на ОБЩЕЙ сессии отчёта, которые сами обёрнуты во + внешние SAVEPOINT'ы. БЕЗ `db.begin_nested():` здесь сбойный `db.execute` + проглатывается (→ frozenset()), оставляя транзакцию aborted; тогда внешний + SAVEPOINT при RELEASE падает (RELEASE запрещён в aborted-tx) и отравление + каскадит. Savepoint в точке перехвата откатывает сбой (ROLLBACK TO SAVEPOINT), + сессия остаётся рабочей. """ global _ADMIN_NAMES_CACHE if _ADMIN_NAMES_CACHE is not None: return _ADMIN_NAMES_CACHE try: - rows = db.execute(_ADMIN_NAMES_SQL).scalars().all() + with db.begin_nested(): + rows = db.execute(_ADMIN_NAMES_SQL).scalars().all() _ADMIN_NAMES_CACHE = frozenset(str(r) for r in rows if r) except Exception: # Не кешируем пустой результат-от-сбоя: следующий вызов попробует снова. @@ -122,8 +132,13 @@ def resolve_objective_districts(db: Session, district: str | None) -> list[str] return [district] # district — официальное админ-имя: разворачиваем в чистые ЕКБ-микро. + # SAVEPOINT на внутреннем swallow-сайте (#2464 cluster A, RELEASE-trap): см. + # `_admin_names` — тот же общий-сессионный вызов из §22-билдеров под внешним + # SAVEPOINT. `db.begin_nested():` откатывает сбой в точке перехвата, чтобы + # внешний RELEASE не упал в aborted-tx (RELEASE запрещён в aborted). try: - rows = db.execute(_MICROS_FOR_ADMIN_SQL, {"d": district}).scalars().all() + with db.begin_nested(): + rows = db.execute(_MICROS_FOR_ADMIN_SQL, {"d": district}).scalars().all() micros = [str(m) for m in rows] except Exception: logger.exception( diff --git a/backend/tests/services/forecasting/test_macro_series.py b/backend/tests/services/forecasting/test_macro_series.py index 8ecc6093..859aa315 100644 --- a/backend/tests/services/forecasting/test_macro_series.py +++ b/backend/tests/services/forecasting/test_macro_series.py @@ -13,18 +13,25 @@ psycopg v3 правило проверяется явно: bind-параметр from __future__ import annotations +import contextlib import datetime as dt import os +from collections.abc import Iterator +from typing import Any from unittest.mock import MagicMock, patch os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") from app.services.forecasting.macro_series import ( _CONFOUNDER_DATES, + _MORTGAGE_FIELDS, _REGIME_EPS_PP, MonthlyMacro, _carry_forward, _month_grid, + _query_inflation_monthly, + _query_key_rate_monthly, + _query_mortgage_monthly, _shift_months, classify_regime, get_monthly_macro, @@ -476,3 +483,110 @@ class TestGetMonthlyMacroGraceful: out = get_monthly_macro(db, months_back=1) assert len(out) == 2 assert all(m.mortgage_rate_weighted is None for m in out) + + +# ────────────────────────────────────────────────────────────────────────────── +# SAVEPOINT (#2464 cluster A finding #2, `_query_mortgage_monthly`): сбой ОДНОГО +# mortgage-поля НЕ должен «отравить» общую §22-сессию для остальных 4 полей / для +# последующих §9.x-слоёв, переиспользующих ту же `db`-Session. Без `db.begin_nested()` +# сбойный `db.execute` (внутри `get_macro_series`) оставляет транзакцию Postgres +# aborted — каждый ПОСЛЕДУЮЩИЙ execute на той же сессии тоже падал бы. +# ────────────────────────────────────────────────────────────────────────────── + + +class _FakeMortgageResult: + """Мини-результат db.execute() — покрывает `.all()` (см. `get_macro_series`).""" + + def all(self) -> list[Any]: + return [] + + +class _PoisonableSession: + """Fake Session, симулирующая Postgres 'aborted transaction' семантику. + + `execute()` бросает РОВНО один раз (на первое поле) и, как реальный Postgres, + «отравляет» сессию: БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()` (даже + несвязанный) тоже бросает — `current transaction is aborted`. `begin_nested()` + зеркалит `Session.begin_nested()`: при исключении откатывает ТОЛЬКО SAVEPOINT + (сбрасывает poison-флаг) и пробрасывает исключение дальше — как настоящий + `ROLLBACK TO SAVEPOINT`, НЕ общий `db.rollback()` (откатил бы всю внешнюю + транзакцию). + """ + + def __init__(self) -> None: + self.poisoned = False + self._raised = False + self.calls = 0 + + def execute(self, *args: Any, **kwargs: Any) -> _FakeMortgageResult: + self.calls += 1 + if self.poisoned: + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + if not self._raised: + self._raised = True + self.poisoned = True + raise RuntimeError("simulated mortgage field query failure") + return _FakeMortgageResult() + + @contextlib.contextmanager + def begin_nested(self) -> Iterator[None]: + try: + yield + except Exception: + # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT, + # сессия остаётся рабочей для следующего поля. + self.poisoned = False + raise + + +class TestQueryMortgageMonthlySavepoint: + """`_query_mortgage_monthly` оборачивает КАЖДОЕ поле в `db.begin_nested()`.""" + + def test_first_field_failure_leaves_session_usable_for_later_fields(self) -> None: + db = _PoisonableSession() + out = _query_mortgage_monthly(db, since=dt.date(2024, 1, 1)) + # все 5 полей присутствуют в результате (контракт функции не изменён). + assert set(out.keys()) == {field for _itype, field in _MORTGAGE_FIELDS} + # сбойное (первое) поле → пустой fallback-словарь, НЕ crash. + first_field = _MORTGAGE_FIELDS[0][1] + assert out[first_field] == {} + # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ + # сбойного поля проходит без 'current transaction aborted' (не poisoned). + db.execute("SELECT 1") + + +class TestQueryKeyRateMonthlySavepoint: + """`_query_key_rate_monthly` оборачивает db.execute в `db.begin_nested()`. + + Тот же poisoning-класс, что и mortgage-цикл: key_rate — ПЕРВЫЙ запрос + get_monthly_macro; без SAVEPOINT его сбой отравил бы транзакцию для inflation + + mortgage + всех последующих §9.x-слоёв на общей §22-сессии. + """ + + def test_db_failure_leaves_session_usable(self) -> None: + db = _PoisonableSession() + out = _query_key_rate_monthly(db, since=dt.date(2024, 1, 1)) + # сбой запроса → пустой ресэмпл (fallback, НЕ crash). + assert out == {} + # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ + # сбоя проходит без 'current transaction aborted' (не poisoned). + db.execute("SELECT 1") + + +class TestQueryInflationMonthlySavepoint: + """`_query_inflation_monthly` оборачивает db.execute (через get_macro_series) в SAVEPOINT. + + Тот же poisoning-класс: inflation — ВТОРОЙ запрос get_monthly_macro; без SAVEPOINT + его сбой отравил бы транзакцию для mortgage-полей + последующих §9.x-слоёв. + """ + + def test_db_failure_leaves_session_usable(self) -> None: + db = _PoisonableSession() + out = _query_inflation_monthly(db, since=dt.date(2024, 1, 1)) + # сбой запроса → пустой ряд (fallback, НЕ crash). + assert out == {} + # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ + # сбоя проходит без 'current transaction aborted' (не poisoned). + db.execute("SELECT 1") diff --git a/backend/tests/services/forecasting/test_orchestrator.py b/backend/tests/services/forecasting/test_orchestrator.py index daf2b83f..72d11cff 100644 --- a/backend/tests/services/forecasting/test_orchestrator.py +++ b/backend/tests/services/forecasting/test_orchestrator.py @@ -371,3 +371,101 @@ class TestNormalizeDistrict: # list / int / прочие неподдерживаемые типы → None + warning. assert _normalize_district([1, 2, 3]) is None assert _normalize_district(42) is None + + +# ────────────────────────────────────────────────────────────────────────────── +# SAVEPOINT (#2464 cluster A finding #1, orchestrator.py `_safe_call`): один сбойный +# §9.x-слой НЕ должен «отравить» общую §22-сессию для ОСТАЛЬНЫХ слоёв. Без +# `db.begin_nested()` сбойный `db.execute` внутри слоя оставляет транзакцию Postgres +# aborted («current transaction is aborted, commands ignored until end of +# transaction block») — каждый ПОСЛЕДУЮЩИЙ слой на той же сессии тоже падал бы. +# ────────────────────────────────────────────────────────────────────────────── + + +class _FakeResult: + """Мини-результат db.execute() — покрывает `.mappings().first()`/`.all()`.""" + + def mappings(self) -> _FakeResult: + return self + + def first(self) -> None: + return None + + def all(self) -> list[Any]: + return [] + + +class _PoisonableSession: + """Fake Session, симулирующая Postgres 'aborted transaction' семантику. + + `execute()` бросает РОВНО один раз (на первый вызов) и, как реальный Postgres, + «отравляет» сессию: БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()` (даже + совершенно несвязанный) тоже бросает — `current transaction is aborted`. + `begin_nested()` зеркалит `Session.begin_nested()`: при исключении откатывает + ТОЛЬКО SAVEPOINT (сбрасывает poison-флаг) и пробрасывает исключение дальше — как + настоящий `ROLLBACK TO SAVEPOINT`, НЕ общий `db.rollback()` (который откатил бы + всю внешнюю транзакцию — established anti-pattern, `backend.md`). + """ + + def __init__(self) -> None: + self.poisoned = False + self._raised = False + self.calls = 0 + + def execute(self, *args: Any, **kwargs: Any) -> _FakeResult: + self.calls += 1 + if self.poisoned: + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + if not self._raised: + self._raised = True + self.poisoned = True + raise RuntimeError("simulated §9.x layer query failure") + return _FakeResult() + + @contextlib.contextmanager + def begin_nested(self) -> Iterator[None]: + try: + yield + except Exception: + # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT, + # сессия остаётся рабочей для следующего слоя. + self.poisoned = False + raise + + +class TestSafeCallSavepointDoesNotPoisonSession: + """`_safe_call` оборачивает вызов слоя в `db.begin_nested()` (SAVEPOINT).""" + + def test_layer_db_failure_leaves_session_usable_for_later_layers(self) -> None: + db = _PoisonableSession() + mocks = _patched_layers() + # market_metrics — единственный слой в этом тесте, реально трогающий `db`: + # его мок сам вызывает db.execute (через _PoisonableSession бросает один раз). + # Остальные слои остаются sentinel-моками из _patched_layers() (db не трогают). + mocks["compute_market_metrics"] = MagicMock( + side_effect=lambda *a, **kw: db.execute("SELECT 1") + ) + captured: dict[str, Any] = {} + + def _fake_assemble(analyze: dict[str, Any], **kwargs: Any) -> SiteFinderReport: + captured.update(kwargs) + return SiteFinderReport() + + with _patch_layers(mocks, _fake_assemble): + report = build_site_finder_report( + db, + analyze=_analyze([{"obj_class": "Комфорт"}]), + cad_num="66:41:0702048:27", + district="Ленинский", + horizons=[6, 12, 18], + ) + assert isinstance(report, SiteFinderReport) + # упавший слой → None в слоте (fallback), остальные слои — sentinel'ы, целы. + assert captured["market_metrics"] is None + assert captured["scenarios"] == ["SC"] + assert captured["forecasts"] == ["F6", "F12"] + # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ + # сбойного слоя проходит без 'current transaction aborted' (не poisoned). + db.execute("SELECT 1") diff --git a/backend/tests/services/forecasting/test_sales_series.py b/backend/tests/services/forecasting/test_sales_series.py index 2b9ca881..afc7c1bd 100644 --- a/backend/tests/services/forecasting/test_sales_series.py +++ b/backend/tests/services/forecasting/test_sales_series.py @@ -15,10 +15,12 @@ psycopg v3 правило проверяется явно: bind-параметр from __future__ import annotations +import contextlib import datetime as dt import math import os from collections.abc import Iterator +from typing import Any from unittest.mock import MagicMock, patch import pytest @@ -667,6 +669,85 @@ class TestBuildSalesSeriesGraceful: assert out.units == [0] +# ────────────────────────────────────────────────────────────────────────────── +# SAVEPOINT (#2464 cluster A finding #4, `_query_source_a`/`_query_source_b`): сбой +# db.execute НЕ должен «отравить» общую §22-сессию для ПОСЛЕДУЮЩИХ слоёв/запросов, +# переиспользующих ту же `db`-Session. Без `db.begin_nested()` сбойный `db.execute` +# оставляет транзакцию Postgres aborted — каждый ПОСЛЕДУЮЩИЙ execute на той же сессии +# тоже падал бы. +# ────────────────────────────────────────────────────────────────────────────── + + +class _FakeSalesResult: + """Мини-результат db.execute() — покрывает `.mappings().all()`.""" + + def mappings(self) -> _FakeSalesResult: + return self + + def all(self) -> list[Any]: + return [] + + +class _PoisonableSession: + """Fake Session, симулирующая Postgres 'aborted transaction' семантику. + + `execute()` бросает РОВНО один раз и, как реальный Postgres, «отравляет» сессию: + БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()` (даже несвязанный) тоже + бросает — `current transaction is aborted`. `begin_nested()` зеркалит + `Session.begin_nested()`: при исключении откатывает ТОЛЬКО SAVEPOINT (сбрасывает + poison-флаг) и пробрасывает исключение дальше — как настоящий `ROLLBACK TO + SAVEPOINT`, НЕ общий `db.rollback()` (откатил бы всю внешнюю транзакцию). + """ + + def __init__(self) -> None: + self.poisoned = False + self._raised = False + self.calls = 0 + + def execute(self, *args: Any, **kwargs: Any) -> _FakeSalesResult: + self.calls += 1 + if self.poisoned: + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + if not self._raised: + self._raised = True + self.poisoned = True + raise RuntimeError("simulated sales query failure") + return _FakeSalesResult() + + @contextlib.contextmanager + def begin_nested(self) -> Iterator[None]: + try: + yield + except Exception: + # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT, + # сессия остаётся рабочей для последующих запросов. + self.poisoned = False + raise + + +class TestSourceQuerySavepointDoesNotPoisonSession: + """`_query_source_a`/`_query_source_b` оборачивают db.execute в SAVEPOINT.""" + + def test_source_a_failure_leaves_session_usable(self) -> None: + db = _PoisonableSession() + out = build_sales_series(db, spec=SegmentSpec(), source="corpus_room_month", months_back=1) + # сбой запроса → zero-filled сетка, low confidence (fallback, НЕ crash). + assert out.confidence == "low" + assert out.units == [0, 0] + # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ + # сбоя проходит без 'current transaction aborted' (не poisoned). + db.execute("SELECT 1") + + def test_source_b_failure_leaves_session_usable(self) -> None: + db = _PoisonableSession() + out = build_sales_series(db, spec=SegmentSpec(), source="objective_lots", months_back=1) + assert out.confidence == "low" + assert out.units == [0, 0] + db.execute("SELECT 1") + + class TestDistrictResolution: """Step 2: spec.district (админ-имя ЕКБ) резолвится в МИКРО-набор в SQL-фильтре. diff --git a/backend/tests/services/forecasting/test_special_indices.py b/backend/tests/services/forecasting/test_special_indices.py index 6bade42b..92591748 100644 --- a/backend/tests/services/forecasting/test_special_indices.py +++ b/backend/tests/services/forecasting/test_special_indices.py @@ -16,6 +16,7 @@ test_market_metrics. from __future__ import annotations +import contextlib from datetime import date from typing import Any from unittest.mock import MagicMock, patch @@ -65,6 +66,7 @@ from app.services.forecasting.special_indices import ( _price_bucket_to_band, _price_overlap, _query_artificial_demand, + _query_parcel_centroid, _timing_overlap, _unit_mix_similarity, _void_index, @@ -1570,6 +1572,178 @@ class TestComputeSpecialIndicesGraceful: assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.25) +# ────────────────────────────────────────────────────────────────────────────── +# SAVEPOINT (#2464 cluster A finding #3, `_run` внутри `compute_special_indices`): +# сбой ОДНОГО §25-индекса НЕ должен «отравить» общую §22-сессию для ОСТАЛЬНЫХ +# индексов/слоёв, переиспользующих ту же `db`-Session. Без `db.begin_nested()` +# сбойный `db.execute` (внутри `_query_artificial_demand`) оставляет транзакцию +# Postgres aborted — каждый ПОСЛЕДУЮЩИЙ execute на той же сессии тоже падал бы. +# ────────────────────────────────────────────────────────────────────────────── + + +class _FakeIndicesResult: + """Мини-результат db.execute() — покрывает `.mappings().first()`.""" + + def mappings(self) -> _FakeIndicesResult: + return self + + def first(self) -> None: + return None + + +class _PoisonableSession: + """Fake Session, симулирующая Postgres 'aborted transaction' семантику. + + `execute()` бросает РОВНО один раз (Artificial Demand SQL) и, как реальный + Postgres, «отравляет» сессию: БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()` + (даже несвязанный) тоже бросает — `current transaction is aborted`. + `begin_nested()` зеркалит `Session.begin_nested()`: при исключении откатывает + ТОЛЬКО SAVEPOINT (сбрасывает poison-флаг) и пробрасывает исключение дальше — как + настоящий `ROLLBACK TO SAVEPOINT`, НЕ общий `db.rollback()`. + """ + + def __init__(self) -> None: + self.poisoned = False + self._raised = False + self.calls = 0 + + def execute(self, *args: Any, **kwargs: Any) -> _FakeIndicesResult: + self.calls += 1 + if self.poisoned: + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + if not self._raised: + self._raised = True + self.poisoned = True + raise RuntimeError("simulated artificial_demand query failure") + return _FakeIndicesResult() + + @contextlib.contextmanager + def begin_nested(self): # type: ignore[no-untyped-def] + try: + yield + except Exception: + # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT, + # сессия остаётся рабочей для следующего индекса. + self.poisoned = False + raise + + +class TestRunSavepointDoesNotPoisonSession: + """`_run` (внутренний gate compute_special_indices) оборачивает builder в SAVEPOINT.""" + + def test_artificial_demand_db_failure_leaves_session_usable(self) -> None: + db = _PoisonableSession() + # district=None/cad_num=None → Cannibalization/Competitor Strength деградируют + # БЕЗ обращения к db (см. test_no_cad_num_degrades_competitor_indices); own + # portfolio патчим в [] (own_developer_ids-гейт не задействуем) — единственный + # builder, реально трогающий db в этом тесте, это Artificial Demand. + with _full_stack_patch(), patch(f"{_MOD}.get_own_portfolio", return_value=[]): + card = compute_special_indices(db, spec=_SPEC, district=None, cad_num=None) + # упавший индекс деградировал в fallback (НЕ крашнул карточку). + assert card.indices[KEY_ARTIFICIAL_DEMAND].method == _METHOD_UNAVAILABLE + assert card.indices[KEY_ARTIFICIAL_DEMAND].value is None + assert len(card.indices) == 6 + # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ + # сбойного индекса проходит без 'current transaction aborted' (не poisoned). + db.execute("SELECT 1") + + +# ────────────────────────────────────────────────────────────────────────────── +# RELEASE-SAVEPOINT trap (#2464 cluster A, second-order): `_query_parcel_centroid` — +# ВНУТРЕННИЙ helper, зовётся из `_build_cannibalization` ПОД внешним SAVEPOINT +# `_run`. Если он глотает db-сбой БЕЗ своего savepoint, tx остаётся aborted; внешний +# `_run`-savepoint выходит «нормально» → RELEASE SAVEPOINT → RELEASE в aborted-tx САМ +# падает (Postgres допускает лишь ROLLBACK / ROLLBACK TO SAVEPOINT) → внешний except +# ловит RELEASE-ошибку, но tx не откачена → отравление каскадит в следующий индекс. +# Тест воспроизводит ИМЕННО RELEASE-trap: `_ReleaseTrapSession` моделирует, что +# RELEASE в aborted-tx падает (без внутреннего savepoint тест краснеет). +# ────────────────────────────────────────────────────────────────────────────── + + +class _FakeReleaseResult: + """Мини-результат db.execute() — покрывает `.mappings().first()`.""" + + def mappings(self) -> _FakeReleaseResult: + return self + + def first(self) -> None: + return None + + +class _ReleaseTrapSession: + """Fake Session, моделирующая Postgres SAVEPOINT/RELEASE/ROLLBACK в aborted-tx. + + `execute()` бросает РОВНО раз (помечает tx aborted). `begin_nested()`: + • ВХОД в poisoned-tx → SAVEPOINT запрещён → бросает; + • ВЫХОД с исключением → ROLLBACK TO SAVEPOINT (снимает poison), пробрасывает; + • ВЫХОД нормально → RELEASE SAVEPOINT, и если poisoned — RELEASE ПАДАЕТ (trap). + """ + + def __init__(self) -> None: + self.poisoned = False + self._raised = False + self.calls = 0 + + def execute(self, *args: Any, **kwargs: Any) -> _FakeReleaseResult: + self.calls += 1 + if self.poisoned: + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + if not self._raised: + self._raised = True + self.poisoned = True + raise RuntimeError("simulated centroid query failure") + return _FakeReleaseResult() + + @contextlib.contextmanager + def begin_nested(self): # type: ignore[no-untyped-def] + if self.poisoned: + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + try: + yield + except Exception: + self.poisoned = False # ROLLBACK TO SAVEPOINT + raise + else: + if self.poisoned: + # RELEASE SAVEPOINT в aborted-tx ПАДАЕТ — тот самый trap. + raise RuntimeError( + "current transaction is aborted, " + "commands ignored until end of transaction block" + ) + + +def _outer_savepoint_call(db: Any, fn: Any, *, fallback: Any) -> Any: + """Зеркало `_run`/`_safe_call`: fn() под ВНЕШНИМ savepoint (RELEASE на норм-выходе).""" + try: + with db.begin_nested(): + return fn() + except Exception: + return fallback + + +class TestReleaseSavepointTrapParcelCentroid: + """`_query_parcel_centroid` НЕ отравляет внешний SAVEPOINT (RELEASE-trap).""" + + def test_centroid_failure_survives_outer_release(self) -> None: + db = _ReleaseTrapSession() # первый execute (centroid) падает + result = _outer_savepoint_call( + db, + lambda: _query_parcel_centroid(db, cad_num="66:41:0303161:123"), + fallback="OUTER_FALLBACK", + ) + # centroid проглочен → его fallback (None), НЕ внешний OUTER_FALLBACK: значит + # внутренний savepoint откатил сбой и внешний RELEASE прошёл чисто. + assert result is None + # КЛЮЧЕВОЙ ассерт: сессия НЕ отравлена — follow-up execute без 'aborted'. + db.execute("SELECT 1") + + # ────────────────────────────────────────────────────────────────────────────── # §25.3 Cannibalization dispatch — TRUE own-portfolio vs PROXY fallback # ────────────────────────────────────────────────────────────────────────────── diff --git a/backend/tests/services/site_finder/test_district_resolver.py b/backend/tests/services/site_finder/test_district_resolver.py index d797a6db..bdc5d808 100644 --- a/backend/tests/services/site_finder/test_district_resolver.py +++ b/backend/tests/services/site_finder/test_district_resolver.py @@ -17,7 +17,10 @@ admin-names (.scalars().all()) + micro-lookup (.scalars().all()). КРИТИЧН from __future__ import annotations +import contextlib import os +from collections.abc import Callable, Iterator +from typing import Any from unittest.mock import MagicMock os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") @@ -26,6 +29,7 @@ import pytest from app.services.site_finder import district_resolver from app.services.site_finder.district_resolver import ( + _admin_names, _reset_admin_cache, resolve_objective_districts, ) @@ -220,3 +224,129 @@ class TestResolverSqlShape: # 8 админов = ekb_districts с непустым geom ('не определён' без geom исключён). sql = str(district_resolver._ADMIN_NAMES_SQL) assert "geom IS NOT NULL" in sql + + +# ────────────────────────────────────────────────────────────────────────────── +# RELEASE-SAVEPOINT trap (#2464 cluster A, second-order): резолвер зовётся §22- +# билдерами (sales_series `_query_source_a/_b`, `_query_artificial_demand`) на общей +# сессии отчёта ПОД внешним SAVEPOINT. Если внутренний swallow (_admin_names / +# micro-lookup) съедает db-сбой БЕЗ своего savepoint, транзакция остаётся aborted; +# внешний savepoint выходит «нормально» и делает RELEASE SAVEPOINT, который САМ падает +# в aborted-tx (Postgres допускает лишь ROLLBACK / ROLLBACK TO SAVEPOINT) — внешний +# except ловит RELEASE-ошибку, но tx так и не откачена → отравление каскадит. Fix = +# savepoint В ТОЧКЕ ПЕРЕХВАТА. Тест воспроизводит ИМЕННО RELEASE-trap (не плоский +# swallow): `_ReleaseTrapSession` моделирует, что RELEASE в aborted-tx падает. +# ────────────────────────────────────────────────────────────────────────────── + + +class _FakeResult: + """Мини-результат db.execute() — покрывает `.scalars().all()` и `.mappings().first()`.""" + + def __init__(self, rows: list[Any]) -> None: + self._rows = list(rows) + + def scalars(self) -> _FakeResult: + return self + + def mappings(self) -> _FakeResult: + return self + + def all(self) -> list[Any]: + return list(self._rows) + + def first(self) -> Any: + return self._rows[0] if self._rows else None + + +class _ReleaseTrapSession: + """Fake Session, моделирующая Postgres SAVEPOINT/RELEASE/ROLLBACK в aborted-tx. + + - `execute()` на `fail_on_call`-м вызове бросает РОВНО раз и помечает tx aborted + (poisoned). Пока poisoned — любой execute бросает 'current transaction is aborted'. + - `begin_nested()` моделирует SAVEPOINT: + • ВХОД в poisoned-tx → SAVEPOINT запрещён → бросает (как реальный Postgres); + • ВЫХОД с исключением → ROLLBACK TO SAVEPOINT (снимает poison), пробрасывает; + • ВЫХОД нормально → RELEASE SAVEPOINT, и если tx poisoned — RELEASE ПАДАЕТ + (это и есть RELEASE-trap; без внутреннего savepoint тест краснеет). + """ + + def __init__( + self, *, fail_on_call: int = 1, results: dict[int, list[Any]] | None = None + ) -> None: + self.poisoned = False + self.calls = 0 + self.fail_on_call = fail_on_call + self._results = results or {} + + def execute(self, *args: Any, **kwargs: Any) -> _FakeResult: + self.calls += 1 + if self.poisoned: + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + if self.calls == self.fail_on_call: + self.poisoned = True + raise RuntimeError("simulated inner helper query failure") + return _FakeResult(self._results.get(self.calls, [])) + + @contextlib.contextmanager + def begin_nested(self) -> Iterator[None]: + if self.poisoned: + # SAVEPOINT запрещён в aborted-tx. + raise RuntimeError( + "current transaction is aborted, commands ignored until end of transaction block" + ) + try: + yield + except Exception: + # ROLLBACK TO SAVEPOINT — разрешён в aborted, снимает poison. + self.poisoned = False + raise + else: + if self.poisoned: + # RELEASE SAVEPOINT в aborted-tx ПАДАЕТ — тот самый trap. + raise RuntimeError( + "current transaction is aborted, " + "commands ignored until end of transaction block" + ) + + +def _outer_savepoint_call(db: Any, fn: Callable[[], Any], *, fallback: Any) -> Any: + """Зеркало `_run`/`_safe_call`: обернуть fn() во ВНЕШНИЙ savepoint (RELEASE на норм-выходе). + + Именно этот RELEASE падает в aborted-tx, если внутренний helper отравил сессию без + своего savepoint → возвращается `fallback` (сигнал, что trap сработал). + """ + try: + with db.begin_nested(): + return fn() + except Exception: + return fallback + + +class TestReleaseSavepointTrap: + """Внутренние swallow-сайты резолвера НЕ отравляют внешний SAVEPOINT (RELEASE-trap).""" + + def test_admin_names_failure_survives_outer_release(self) -> None: + _reset_admin_cache() + db = _ReleaseTrapSession(fail_on_call=1) # admin-names lookup падает + result = _outer_savepoint_call(db, lambda: _admin_names(db), fallback="OUTER_FALLBACK") + # _admin_names вернул СВОЙ fallback (пустой frozenset) — внешний RELEASE прошёл, + # значит внутренний savepoint откатил сбой (иначе result был бы OUTER_FALLBACK). + assert result == frozenset() + # КЛЮЧЕВОЙ ассерт: сессия НЕ отравлена — follow-up execute без 'aborted'. + db.execute("SELECT 1") + + def test_micro_lookup_failure_survives_outer_release(self) -> None: + _reset_admin_cache() + # call 1 = admin-names (успех, "Кировский" распознан); call 2 = micro-lookup (падает). + db = _ReleaseTrapSession(fail_on_call=2, results={1: ["Кировский"]}) + result = _outer_savepoint_call( + db, + lambda: resolve_objective_districts(db, "Кировский"), + fallback="OUTER_FALLBACK", + ) + # micro-lookup проглочен → EKB-wide fallback (None), НЕ внешний OUTER_FALLBACK. + assert result is None + # КЛЮЧЕВОЙ ассерт: сессия НЕ отравлена — follow-up execute без 'aborted'. + db.execute("SELECT 1")