fix(site_finder): SAVEPOINT-isolate supply/market db.execute + kill 2 bare rollbacks (#2464)

Convert two bare db.rollback() calls on the shared per-request session
(best_layouts _fetch_supply_only_layouts + Step 2b core-attribution) to
db.begin_nested() — a bare rollback there discards ALL uncommitted work of
the request and orphans the outer SessionTransaction. Also wrap the swallow
sites in market_metrics (_query_stock, _query_sales_window, _price_sensitivity,
compute_offer_price_trend), best_layouts step-5 supply, future_supply
_query_supply_rows, own_portfolio _query_current/_query_future, and
custom_pois get_overlaps_for_scoring so a failed statement no longer poisons
the shared tx for later steps. Per-site regression tests assert fallback
preserved, begin_nested used, bare rollback NOT called, and session usable
after failure.

Refs #2464 (cluster A session-poisoning, Wave 2).
EOF2
echo "HANDLED"

backend/app/services/site_finder/best_layouts.py

@@ -971,28 +971,31 @@ def _fetch_supply_only_layouts(
 
     Returns:
         Список SupplyOnlyLayoutRow (может быть пустым). Ошибка SQL не роняет ответ —
-        логируется, откатывается транзакция, возвращается [] (блок аддитивен).
+        логируется, savepoint откатывается, возвращается [] (блок аддитивен).
     """
     if not project_names:
         return []
     try:
-        rows = (
-            db.execute(
-                _SUPPLY_ONLY_LOTS_SQL,
-                {
-                    "names": project_names,
-                    "premise_kind": _SUPPLY_ONLY_PREMISE_KIND,
-                    "price_min": _SUPPLY_ONLY_PRICE_MIN,
-                    "price_max": _SUPPLY_ONLY_PRICE_MAX,
-                },
+        # SAVEPOINT — db shared с caller (GET .../best-layouts / §22-отчёт,
+        # Depends(get_db)). Голый db.rollback() тут НЕЛЬЗЯ: он откатил бы ВСЮ outer
+        # tx (всю несохранённую работу этого запроса) и orphan'ит outer
+        # SessionTransaction (см. velocity.py:170). begin_nested откатывает ТОЛЬКО
+        # этот savepoint — outer tx остаётся usable, блок аддитивно деградирует в [].
+        with db.begin_nested():
+            rows = (
+                db.execute(
+                    _SUPPLY_ONLY_LOTS_SQL,
+                    {
+                        "names": project_names,
+                        "premise_kind": _SUPPLY_ONLY_PREMISE_KIND,
+                        "price_min": _SUPPLY_ONLY_PRICE_MIN,
+                        "price_max": _SUPPLY_ONLY_PRICE_MAX,
+                    },
+                )
+                .mappings()
+                .all()
             )
-            .mappings()
-            .all()
-        )
     except Exception:
-        # Аддитивный блок — не роняем весь ответ. rollback: следующего execute нет,
-        # но транзакция в aborted-state сломала бы commit/дальнейшую работу caller'а.
-        db.rollback()
         logger.exception(
             "best_layouts: supply-only lots query failed for %d projects — empty block",
             len(project_names),
@@ -1120,28 +1123,37 @@ def get_best_layouts(
     # больше не теряет сделки, привязанные к 1-й через mapping).
     core_group_to_project: dict[str, str] = {}
     try:
-        corpus_names = [str(row[0]) for row in db.execute(_OBJECTIVE_PROJECT_NAMES_SQL).all()]
-        lots_names = [str(row[0]) for row in db.execute(_OBJECTIVE_LOTS_PROJECT_NAMES_SQL).all()]
-        project_names = _merged_project_name_pool(corpus_names, lots_names)
-        # Два прохода: сперва кандидаты по ядру БЕЗ dev-confirm — чтобы developers
-        # тянуть ТОЛЬКО для них (0-3 имени, index probe вместо 3.8с-агрегата).
-        preliminary = match_groups_to_projects(complex_groups, project_names, None)
-        candidate_names = sorted(set(preliminary.values()))
-        developers_by_project: dict[str, list[str]] = {}
-        if candidate_names:
-            developers_by_project = {
-                str(row[0]): [str(d) for d in (row[1] or []) if d]
-                for row in db.execute(_OBJECTIVE_DEVELOPERS_SQL, {"names": candidate_names}).all()
-            }
-        core_group_to_project = match_groups_to_projects(
-            complex_groups, project_names, developers_by_project
-        )
+        # SAVEPOINT — db shared с caller (GET .../best-layouts / §22-отчёт,
+        # Depends(get_db)). Три db.execute здесь перемежаются с ЧИСТО-Python
+        # match_groups_to_projects / _merged_project_name_pool (без db) → оборачиваем
+        # весь блок одним savepoint. Голый db.rollback() НЕЛЬЗЯ: он откатил бы ВСЮ
+        # outer tx + orphan'ит outer SessionTransaction (см. velocity.py:170).
+        # begin_nested откатывает только этот savepoint → следующий execute (velocity
+        # SQL, Step 3) не упадёт InFailedSqlTransaction, «fallback на mapping-only»
+        # состоится.
+        with db.begin_nested():
+            corpus_names = [str(row[0]) for row in db.execute(_OBJECTIVE_PROJECT_NAMES_SQL).all()]
+            lots_names = [
+                str(row[0]) for row in db.execute(_OBJECTIVE_LOTS_PROJECT_NAMES_SQL).all()
+            ]
+            project_names = _merged_project_name_pool(corpus_names, lots_names)
+            # Два прохода: сперва кандидаты по ядру БЕЗ dev-confirm — чтобы developers
+            # тянуть ТОЛЬКО для них (0-3 имени, index probe вместо 3.8с-агрегата).
+            preliminary = match_groups_to_projects(complex_groups, project_names, None)
+            candidate_names = sorted(set(preliminary.values()))
+            developers_by_project: dict[str, list[str]] = {}
+            if candidate_names:
+                developers_by_project = {
+                    str(row[0]): [str(d) for d in (row[1] or []) if d]
+                    for row in db.execute(
+                        _OBJECTIVE_DEVELOPERS_SQL, {"names": candidate_names}
+                    ).all()
+                }
+            core_group_to_project = match_groups_to_projects(
+                complex_groups, project_names, developers_by_project
+            )
     except Exception:
         # Не роняем весь ответ — core-attribution аддитивна к mapping-источнику.
-        # rollback: при реальной DB-ошибке транзакция в aborted-state — без отката
-        # следующий execute (velocity SQL) упал бы InFailedSqlTransaction, и
-        # «fallback на mapping-only» не состоялся бы.
-        db.rollback()
         logger.exception(
             "best_layouts: core-attribution failed for cad_num=%s — fallback to mapping-only",
             cad_num,
@@ -1204,18 +1216,24 @@ def get_best_layouts(
     # 0 мес» / «% продано —». Per-object latest snapshot встроен в _SUPPLY_BATCH_SQL
     # (flats_latest CTE) → отдельный pre-compute MAX больше не нужен.
     try:
-        supply_rows = (
-            db.execute(
-                _SUPPLY_BATCH_SQL,
-                {
-                    "center_lon": center_lon,
-                    "center_lat": center_lat,
-                    "radius_m": radius_m,
-                },
+        # SAVEPOINT — db shared с caller (GET .../best-layouts, Depends(get_db), sync
+        # get_best_layouts мостится через run_in_threadpool). Голый db.rollback() тут
+        # НЕЛЬЗЯ — orphan'ит outer SessionTransaction (см. velocity.py:170); begin_nested
+        # откатывает только эту savepoint, оставляя db usable для Step 9b
+        # (_fetch_supply_only_layouts) на этой же сессии.
+        with db.begin_nested():
+            supply_rows = (
+                db.execute(
+                    _SUPPLY_BATCH_SQL,
+                    {
+                        "center_lon": center_lon,
+                        "center_lat": center_lat,
+                        "radius_m": radius_m,
+                    },
+                )
+                .mappings()
+                .all()
             )
-            .mappings()
-            .all()
-        )
     except Exception:
         logger.warning("best_layouts: supply query failed, supply=0 fallback")
         supply_rows = []

backend/app/services/site_finder/custom_pois.py

@@ -251,14 +251,19 @@ def get_overlaps_for_scoring(
     """
     _parcel_cad = parcel_cad or ""
     try:
-        rows = (
-            db.execute(
-                text(_OVERLAPS_SQL),
-                {"wkt": parcel_geom_wkt, "user_id": user_id, "parcel_cad": _parcel_cad},
+        # SAVEPOINT — db shared с caller (POST .../analyze, Depends(get_db)). Голый
+        # db.rollback() orphan'ит outer SessionTransaction (см. velocity.py:170) —
+        # begin_nested откатывает только эту savepoint, оставляя db usable для
+        # остальных запросов analyze_parcel на этой же сессии.
+        with db.begin_nested():
+            rows = (
+                db.execute(
+                    text(_OVERLAPS_SQL),
+                    {"wkt": parcel_geom_wkt, "user_id": user_id, "parcel_cad": _parcel_cad},
+                )
+                .mappings()
+                .all()
             )
-            .mappings()
-            .all()
-        )
         return [
             {
                 "id": int(r["id"]),

backend/app/services/site_finder/future_supply.py

@@ -74,6 +74,7 @@ def _future_supply_key(
     """
     return (district, horizon_months, premise_kind)
 
+
 Confidence = Literal["high", "medium", "low"]
 
 # Слои склада предложения (м.125): 1=open (в продаже), 2=hidden (запас ПД),
@@ -439,7 +440,12 @@ def _query_supply_rows(db: Session, params: Mapping[str, Any]) -> Sequence[Mappi
     (worker ещё не наполнил м.125) или сбой БД не должны валить расчёт индекса.
     """
     try:
-        return db.execute(_SUPPLY_LAYERS_SQL, dict(params)).mappings().all()
+        # SAVEPOINT — db shared с caller (compute_future_supply_pressure в цепочке
+        # §22-форсайта / analyze). Голый db.rollback() orphan'ит outer SessionTransaction
+        # (см. velocity.py:170) — begin_nested откатывает только эту savepoint, оставляя
+        # db usable для последующих db.execute на этой же сессии.
+        with db.begin_nested():
+            return db.execute(_SUPPLY_LAYERS_SQL, dict(params)).mappings().all()
     except Exception:
         logger.exception(
             "future_supply: supply-layers view query failed (district=%s)",

backend/app/services/site_finder/market_metrics.py

@@ -613,7 +613,13 @@ def compute_market_metrics(
 def _query_stock(db: Session, params: Mapping[str, Any]) -> dict[str, int]:
     """Текущий сток. На ошибке/пустых данных → все счётчики 0 (graceful)."""
     try:
-        row = db.execute(_STOCK_SQL, dict(params)).mappings().first()
+        # SAVEPOINT — db shared с caller (compute_market_metrics в цепочке §22-форсайта
+        # / analyze). Голый db.rollback() orphan'ит outer SessionTransaction (см.
+        # velocity.py:170) — begin_nested откатывает только эту savepoint, outer tx
+        # остаётся usable для последующих db.execute (_query_sales_window/_price_sensitivity)
+        # на этой же сессии.
+        with db.begin_nested():
+            row = db.execute(_STOCK_SQL, dict(params)).mappings().first()
     except Exception:
         logger.exception(
             "market_metrics: stock query failed (districts=%s)", params.get("districts")
@@ -650,7 +656,12 @@ def _query_sales_window(
     затирать прежние счётчики. На ошибке/пусто → (0, 0.0, {}).
     """
     try:
-        rows = db.execute(_SALES_WINDOW_SQL, dict(params)).mappings().all()
+        # SAVEPOINT — db shared с caller (см. _query_stock выше). CRITICAL: без него
+        # сбой здесь оставляет транзакцию aborted, и следующий db.execute в этой же
+        # цепочке (_price_sensitivity → _elasticity_coef) падает "current transaction
+        # is aborted" вместо честного graceful-fallback.
+        with db.begin_nested():
+            rows = db.execute(_SALES_WINDOW_SQL, dict(params)).mappings().all()
     except Exception:
         logger.exception(
             "market_metrics: sales-window query failed (districts=%s)", params.get("districts")
@@ -700,14 +711,19 @@ def _price_sensitivity(
     micros = resolve_objective_districts(db, district)
     elasticity_window = max(window_months, 24)
     try:
-        elast = _elasticity_coef(
-            db,
-            region_code=_EKB_REGION_CODE,
-            district_name=district,
-            target_class=None,
-            elasticity_window_months=elasticity_window,
-            districts=micros if micros is not None else [],
-        )
+        # SAVEPOINT — db shared с caller (compute_market_metrics в цепочке §22-форсайта
+        # / analyze). _elasticity_coef делает свой db.execute внутри без try/except —
+        # оборачиваем ВЕСЬ вызов, чтобы сбой откатывал только эту savepoint (не outer
+        # SessionTransaction, см. velocity.py:170), и db оставалась usable дальше.
+        with db.begin_nested():
+            elast = _elasticity_coef(
+                db,
+                region_code=_EKB_REGION_CODE,
+                district_name=district,
+                target_class=None,
+                elasticity_window_months=elasticity_window,
+                districts=micros if micros is not None else [],
+            )
     except Exception:
         logger.exception(
             "market_metrics: elasticity reuse failed (district=%s micros=%s)",
@@ -877,7 +893,12 @@ def compute_offer_price_trend(
         "price_max": _OFFER_PRICE_MAX,
     }
     try:
-        rows = db.execute(_OFFER_TREND_SQL, params).mappings().all()
+        # SAVEPOINT — db shared с caller (POST .../analyze / §22-отчёт, Depends(get_db)).
+        # 8с-гвард (parcels.py) может отменить этот запрос (QueryCanceled) → без savepoint
+        # транзакция остаётся aborted и следующий db.execute падает "current transaction
+        # is aborted". begin_nested откатывает только этот savepoint, db остаётся usable.
+        with db.begin_nested():
+            rows = db.execute(_OFFER_TREND_SQL, params).mappings().all()
     except Exception as exc:
         # Ожидаемый путь: 8с-гвард (parcels.py) отменяет запрос на холодном кэше
         # (~2/сутки, профилирование 2026-07-04) — блок аддитивный, analyze успешен.

backend/app/services/site_finder/own_portfolio.py

@@ -169,7 +169,12 @@ def _query_current(db: Session) -> list[OwnProject]:
         )
         return []
     try:
-        rows = db.execute(_CURRENT_SQL, {"ids": ids}).mappings().all()
+        # SAVEPOINT — db shared с caller (get_own_portfolio в цепочке §25.3 / analyze).
+        # Голый db.rollback() orphan'ит outer SessionTransaction (см. velocity.py:170) —
+        # begin_nested откатывает только эту savepoint, оставляя db usable для
+        # _query_future (следующий вызов на этой же сессии в get_own_portfolio).
+        with db.begin_nested():
+            rows = db.execute(_CURRENT_SQL, {"ids": ids}).mappings().all()
     except Exception:
         logger.exception("own_portfolio: current (domrf) query failed → []")
         return []
@@ -200,7 +205,11 @@ def _query_future(db: Session) -> list[OwnProject]:
     Сбой/отсутствие таблицы (cold-start до миграции 148) → [] (НЕ crash).
     """
     try:
-        rows = db.execute(_FUTURE_SQL).mappings().all()
+        # SAVEPOINT — db shared с caller (см. _query_current выше). begin_nested
+        # откатывает только эту savepoint (НЕ голый db.rollback(), см. velocity.py:170),
+        # оставляя db usable для остальных потребителей той же сессии (§25.3 / analyze).
+        with db.begin_nested():
+            rows = db.execute(_FUTURE_SQL).mappings().all()
     except Exception:
         logger.exception("own_portfolio: future (own_planned_project) query failed → []")
         return []

backend/tests/api/v1/test_custom_pois.py

@@ -599,3 +599,94 @@ def test_service_update_custom_poi_no_commit_when_not_found() -> None:
 
     db.commit.assert_not_called()
     assert result is None
+
+
+# ── SAVEPOINT regression (#2464 cluster A) ─────────────────────────────────────
+#
+# get_overlaps_for_scoring: db shared с caller (POST .../analyze, Depends(get_db)).
+# Раньше db.execute-сбой ловился в bare ``except Exception as e`` без
+# SAVEPOINT/rollback — на реальном Postgres это отравляет db shared с caller:
+# следующий db.execute на этой же сессии (остальные шаги analyze_parcel) падает
+# "current transaction is aborted". Фикс — db.execute обёрнут в
+# ``with db.begin_nested():``. MagicMock не эмулирует реальный aborted-transaction
+# Postgres — тесты проверяют (1) graceful [] fallback при сбое, (2) что
+# begin_nested() реально вызван, (3) что db остаётся usable для следующего вызова
+# на той же mock-сессии.
+
+
+def _overlap_row(data: dict[str, Any]) -> MagicMock:
+    m = MagicMock()
+    m.__getitem__ = lambda self, k: data[k]
+    return m
+
+
+def test_get_overlaps_for_scoring_returns_shaped_rows() -> None:
+    """Happy-path: db.execute успешен → список dict с ожидаемыми ключами."""
+    from app.services.site_finder.custom_pois import get_overlaps_for_scoring
+
+    row = _overlap_row(
+        {
+            "id": 42,
+            "name": "Мой парк",
+            "category": "park",
+            "weight": 2.0,
+            "lon": 60.607,
+            "lat": 56.840,
+            "distance_m": 500.0,
+        }
+    )
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = [row]
+
+    out = get_overlaps_for_scoring(db, "POINT(60.6 56.8)", _SESSION, parcel_cad="66:41:0:1")
+
+    assert out == [
+        {
+            "id": 42,
+            "name": "Мой парк",
+            "category": "park",
+            "weight": 2.0,
+            "lon": 60.607,
+            "lat": 56.840,
+            "distance_m": 500.0,
+        }
+    ]
+
+
+def test_get_overlaps_for_scoring_uses_begin_nested_savepoint() -> None:
+    """Фикс: db.execute обёрнут в db.begin_nested(), не голый вызов."""
+    from app.services.site_finder.custom_pois import get_overlaps_for_scoring
+
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = []
+
+    get_overlaps_for_scoring(db, "POINT(60.6 56.8)", _SESSION)
+
+    assert db.begin_nested.call_count == 1
+
+
+def test_get_overlaps_for_scoring_query_failure_degrades_to_empty_list() -> None:
+    """Сбой db.execute → [] (graceful), не крашит analyze_parcel."""
+    from app.services.site_finder.custom_pois import get_overlaps_for_scoring
+
+    db = MagicMock()
+    db.execute.side_effect = RuntimeError("simulated DB failure")
+
+    assert get_overlaps_for_scoring(db, "POINT(60.6 56.8)", _SESSION) == []
+
+
+def test_get_overlaps_for_scoring_session_usable_after_failure() -> None:
+    """db.execute сбоит один раз, но db остаётся usable для следующего вызова на
+    этой же сессии (имитирует custom-POI overlaps → остальные шаги analyze_parcel
+    на одной shared Session)."""
+    from app.services.site_finder.custom_pois import get_overlaps_for_scoring
+
+    db = MagicMock()
+    db.execute.side_effect = RuntimeError("boom")
+    assert get_overlaps_for_scoring(db, "POINT(60.6 56.8)", _SESSION) == []
+
+    db.execute.side_effect = None
+    next_result = MagicMock()
+    next_result.mappings.return_value.all.return_value = [{"marker": "ok"}]
+    db.execute.return_value = next_result
+    assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]

backend/tests/services/site_finder/test_best_layouts.py

@@ -40,6 +40,7 @@ from app.services.site_finder.best_layouts import (
     MAX_BUCKET_SHARE_PCT,
     NAMELESS_CLUSTER_M,
     _cap_and_redistribute,
+    _fetch_supply_only_layouts,
     get_best_layouts,
     group_radius_objects,
     match_groups_to_projects,
@@ -1224,3 +1225,193 @@ def test_supply_only_sql_uses_price_sanity_and_on_sale_filter() -> None:
     assert "'studio'" in sql_text and "'4+'" in sql_text
     # psycopg3: CAST(:x AS type), НЕ :x::type
     assert "::text[]" not in sql_text.replace("CAST(:names AS text[])", "")
+
+
+# ── SAVEPOINT regression (#2464 cluster A) ────────────────────────────────────
+#
+# Step 5 (_SUPPLY_BATCH_SQL) db shared с caller (GET .../best-layouts,
+# Depends(get_db); sync get_best_layouts мостится через run_in_threadpool). Раньше
+# db.execute-сбой ловился в bare ``except Exception`` без SAVEPOINT/rollback — на
+# реальном Postgres это отравляет db shared с caller: следующий db.execute на этой
+# же сессии (напр. Step 9b _fetch_supply_only_layouts) падает "current transaction
+# is aborted". Фикс — db.execute обёрнут в ``with db.begin_nested():``. MagicMock не
+# эмулирует реальный aborted-transaction Postgres — тесты проверяют (1) graceful
+# supply=0 fallback при сбое, (2) что begin_nested() реально вызван вокруг именно
+# этого db.execute, (3) что db остаётся usable для последующих вызовов на этой же
+# mock-сессии (get_best_layouts не крашит, ранжирование по-прежнему строится).
+
+
+def test_supply_query_failure_degrades_to_zero_supply() -> None:
+    """Сбой Step 5 (_SUPPLY_BATCH_SQL) → supply_units_in_radius=0 для всех строк,
+    остальной pipeline (velocity/ranking) не ломается."""
+    id_rows = [_obj_id_row(1)]
+    vel_rows = [_vel_row("2", deals_window=48.0, avg_area=55.0, obj_ids=[1])]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
+
+    real_dispatch = db.execute.side_effect
+
+    def _execute_raise_on_supply(stmt, params=None, *args, **kwargs):
+        if stmt is _SUPPLY_BATCH_SQL:
+            raise RuntimeError("simulated DB failure")
+        return real_dispatch(stmt, params, *args, **kwargs)
+
+    db.execute.side_effect = _execute_raise_on_supply
+
+    req = _request(time_window="last_quarter")
+    resp = get_best_layouts(db, CAD_NUM, req)
+
+    assert len(resp.top_layouts) == 1
+    assert resp.top_layouts[0].supply_units_in_radius == 0
+    assert resp.top_layouts[0].sold_pct_of_supply is None
+    assert resp.top_layouts[0].is_oversold is False
+    # velocity per-month всё ещё честно посчитан (сбой изолирован в supply-блоке).
+    assert resp.top_layouts[0].velocity_per_month == pytest.approx(16.0, rel=1e-3)
+
+
+def test_supply_query_uses_begin_nested_savepoint() -> None:
+    """Фикс: Step 5 запрос обёрнут в db.begin_nested(), не голый db.execute."""
+    id_rows = [_obj_id_row(1)]
+    vel_rows = [_vel_row("2", deals_window=48.0, avg_area=55.0, obj_ids=[1])]
+    supply_rows = [_supply_row("2", "40-60", 80)]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows, supply_rows=supply_rows)
+
+    req = _request(time_window="last_quarter")
+    get_best_layouts(db, CAD_NUM, req)
+
+    assert db.begin_nested.call_count >= 1
+
+
+def test_supply_query_failure_session_usable_for_supply_only_step() -> None:
+    """db остаётся usable после Step 5-сбоя: Step 9b (_fetch_supply_only_layouts,
+    отдельный db.execute на той же сессии) по-прежнему выполняется без исключения.
+
+    Сетап зеркалит test_supply_only_group_without_velocity_appears_in_block: две
+    группы, «Траектория» с velocity → top_layouts, «Символ» без velocity → должна
+    дойти до Step 9b (_fetch_supply_only_layouts) на той же db, что и Step 5.
+    """
+    id_rows = [
+        _obj_id_row(1, "ЖК «Траектория»", "Брусника", 56.85, 60.60),
+        _obj_id_row(2, "ЖК «Символ»", "Эфес", 56.851, 60.601),
+    ]
+    vel_rows = [_vel_row("2", deals_window=30.0, obj_ids=[], project_names=["Траектория"])]
+    supply_only_rows = [_supply_only_row("Символ", room_bucket="1", area_bin="25-40")]
+    db = _make_db(
+        coord=_coord_row(),
+        id_rows=id_rows,
+        vel_rows=vel_rows,
+        project_names=["Траектория", "Символ"],
+        developers_by_project={},
+        supply_only_rows=supply_only_rows,
+    )
+
+    real_dispatch = db.execute.side_effect
+
+    def _execute_raise_on_supply(stmt, params=None, *args, **kwargs):
+        if stmt is _SUPPLY_BATCH_SQL:
+            raise RuntimeError("simulated DB failure")
+        return real_dispatch(stmt, params, *args, **kwargs)
+
+    db.execute.side_effect = _execute_raise_on_supply
+
+    req = _request(time_window="last_quarter")
+    # Не должно бросить InFailedSqlTransaction-подобное исключение — Step 9b
+    # выполняется на той же db без проблем (в mock — просто control-flow guard).
+    resp = get_best_layouts(db, CAD_NUM, req)
+    assert resp.top_layouts[0].supply_units_in_radius == 0
+    # Step 9b реально выполнился на этой же db — supply-only блок не пуст.
+    assert len(resp.supply_only_layouts) == 1
+    assert resp.supply_only_layouts[0].project_name == "Символ"
+
+
+# ── SAVEPOINT regression: ex-bare-rollback sites (#2464 cluster A) ──────────────
+#
+# ДВА места в best_layouts.py раньше делали ГОЛЫЙ db.rollback() на SHARED сессии в
+# except после сбоя db.execute — это ХУЖЕ простого swallow: bare rollback на общей
+# сессии откатывает ВСЮ outer-транзакцию /analyze-запроса (всю несохранённую работу
+# ранее в этом же запросе) и orphan'ит outer SessionTransaction (velocity.py:170).
+# Фикс — savepoint: db.execute обёрнут в ``with db.begin_nested():``, а bare
+# db.rollback() УДАЛЁН (auto-rollback savepoint'а откатывает только сбойный statement,
+# не весь запрос). Тесты проверяют: (a) документированный fallback сохранён, (b)
+# begin_nested() использован И db.rollback() НЕ вызван (весь смысл — outer tx выживает),
+# (c) follow-up db.execute на той же сессии проходит.
+
+
+class TestFetchSupplyOnlyLayoutsSavepoint:
+    """Ex-rollback site #1: _fetch_supply_only_layouts (Step 9b)."""
+
+    def test_uses_begin_nested_savepoint(self) -> None:
+        db = MagicMock()
+        db.execute.return_value.mappings.return_value.all.return_value = []
+        _fetch_supply_only_layouts(db, ["ЖК Резерв"])
+        assert db.begin_nested.call_count == 1
+
+    def test_query_failure_degrades_to_empty_and_no_bare_rollback(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("simulated DB failure")
+        out = _fetch_supply_only_layouts(db, ["ЖК Резерв"])
+        assert out == []  # документированный fallback
+        assert db.begin_nested.call_count == 1  # savepoint использован
+        db.rollback.assert_not_called()  # bare rollback удалён — outer tx выживает
+
+    def test_empty_project_names_skips_db_entirely(self) -> None:
+        # Пустой вход → SQL не выполняется, savepoint не открывается (штатный ранний []).
+        db = MagicMock()
+        assert _fetch_supply_only_layouts(db, []) == []
+        db.execute.assert_not_called()
+        db.begin_nested.assert_not_called()
+
+    def test_session_usable_after_failure(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("boom")
+        assert _fetch_supply_only_layouts(db, ["ЖК Резерв"]) == []
+
+        db.execute.side_effect = None
+        next_result = MagicMock()
+        next_result.mappings.return_value.all.return_value = [{"marker": "ok"}]
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]
+
+
+def test_core_attribution_failure_uses_savepoint_no_bare_rollback() -> None:
+    """Ex-rollback site #2: Step 2b core-attribution db.execute сбоит → SAVEPOINT
+    (не bare rollback) → get_best_layouts деградирует в mapping-only.
+
+    Раньше bare db.rollback() здесь откатил бы ВСЮ outer tx /analyze-запроса. Теперь
+    begin_nested откатывает только сбойный savepoint, а velocity через obj_id (mapping)
+    по-прежнему считается на той же сессии → top_layouts не пуст, outer tx выживает.
+    """
+    id_rows = [_obj_id_row(1)]
+    vel_rows = [_vel_row("2", deals_window=48.0, avg_area=55.0, obj_ids=[1])]
+    supply_rows = [_supply_row("2", "40-60", 80)]
+    db = _make_db(
+        coord=_coord_row(),
+        id_rows=id_rows,
+        vel_rows=vel_rows,
+        supply_rows=supply_rows,
+        project_names=["Символ"],
+        developers_by_project={},
+    )
+
+    real_dispatch = db.execute.side_effect
+
+    def _raise_on_project_names(stmt, params=None, *args, **kwargs):
+        # Первый из трёх Step 2b запросов — сбой на нём отравил бы сессию без savepoint.
+        if stmt is _OBJECTIVE_PROJECT_NAMES_SQL:
+            raise RuntimeError("simulated DB failure")
+        return real_dispatch(stmt, params, *args, **kwargs)
+
+    db.execute.side_effect = _raise_on_project_names
+
+    req = _request(time_window="last_quarter")
+    resp = get_best_layouts(db, CAD_NUM, req)
+
+    # savepoint использован, bare rollback НЕ вызван (outer tx /analyze выживает).
+    assert db.begin_nested.call_count >= 1
+    db.rollback.assert_not_called()
+    # fallback на mapping-only: velocity через obj_id всё равно посчитан на той же
+    # сессии (Step 3 velocity execute не заблокирован aborted-транзакцией).
+    assert len(resp.top_layouts) == 1
+    assert resp.top_layouts[0].room_bucket == "2"
+    assert resp.top_layouts[0].velocity_per_month == pytest.approx(16.0, rel=1e-3)
+    # core-match не дал имён (сбой) → matched-only supply-блок пуст, но ответ валиден.
+    assert resp.data_quality.groups_matched_by_core == 0

backend/tests/services/site_finder/test_future_supply.py

@@ -37,6 +37,7 @@ from app.services.site_finder.future_supply import (
     _monthly_absorption_units,
     _months_between,
     _months_of_pressure,
+    _query_supply_rows,
     _round_or_none,
     _saturating_index,
     compute_future_supply_pressure,
@@ -635,3 +636,61 @@ class TestComputeGraceful:
             res = compute_future_supply_pressure(db, district=None)
         assert isinstance(res, FutureSupplyPressure)
         assert res.district is None
+
+
+# ── SAVEPOINT regression (#2464 cluster A) ─────────────────────────────────────
+#
+# db shared с caller (compute_future_supply_pressure в цепочке §22-форсайта /
+# analyze). Раньше _query_supply_rows ловил db.execute-сбой в bare
+# ``except Exception`` без SAVEPOINT/rollback — на реальном Postgres это отравляет
+# db shared с caller: следующий db.execute на этой же сессии (напр.
+# compute_market_metrics, вызываемый следом в compute_future_supply_pressure)
+# падает "current transaction is aborted". Фикс — db.execute обёрнут в
+# ``with db.begin_nested():``. MagicMock не эмулирует реальный aborted-transaction
+# Postgres — тесты проверяют (1) graceful [] fallback при сбое, (2) что
+# begin_nested() реально вызван, (3) что db остаётся usable для следующего вызова
+# на той же mock-сессии.
+
+
+class TestQuerySupplyRowsSavepoint:
+    def test_uses_begin_nested_savepoint(self) -> None:
+        db = MagicMock()
+        db.execute.return_value.mappings.return_value.all.return_value = []
+        _query_supply_rows(db, {"district": "X", "names": ["X"]})
+        assert db.begin_nested.call_count == 1
+
+    def test_query_failure_degrades_to_empty_list(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("simulated DB failure")
+        assert _query_supply_rows(db, {"district": "X", "names": ["X"]}) == []
+
+    def test_session_usable_after_failure(self) -> None:
+        """db.execute сбоит один раз (supply-layers view), но db остаётся usable
+        для следующего вызова на этой же сессии (имитирует
+        _query_supply_rows → compute_market_metrics в compute_future_supply_pressure)."""
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("boom")
+        assert _query_supply_rows(db, {"district": "X", "names": ["X"]}) == []
+
+        db.execute.side_effect = None
+        next_result = MagicMock()
+        next_result.mappings.return_value.all.return_value = [{"marker": "ok"}]
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]
+
+
+class TestComputeFutureSupplyPressureSessionUsableAfterFailure:
+    """End-to-end: view-запрос сбоит → SAVEPOINT (не bare rollback) → reused
+    compute_market_metrics (отдельный db.execute на той же сессии) не задет."""
+
+    def test_view_failure_does_not_poison_market_metrics_call(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("view gone")
+        with _patched(metrics=_metrics_stub(months_of_supply=6.0, n_available=120)) as (mm, _rs):
+            res = compute_future_supply_pressure(db, district="Сбой")
+        assert db.begin_nested.call_count >= 1
+        # compute_market_metrics (reused) вызван несмотря на сбой view-запроса —
+        # сессия не отравлена (в реальном Postgres это была бы "aborted transaction").
+        assert mm.called
+        assert res.hidden_units == 0
+        assert res.index == 0.0

backend/tests/services/site_finder/test_market_metrics.py

@@ -33,6 +33,8 @@ from app.services.site_finder.market_metrics import (
     _months_of_supply,
     _overstock_index,
     _price_sensitivity,
+    _query_sales_window,
+    _query_stock,
     _room_bucket,
     _sell_through_pct,
     compute_market_metrics,
@@ -1010,3 +1012,161 @@ class TestOfferPriceTrendCompute:
         assert d["delta_pct"] == 3.3
         assert d["lots_latest"] == 12
         assert d["complexes_covered"] == 1
+
+
+# ── SAVEPOINT regression (#2464 cluster A) ─────────────────────────────────────
+#
+# db shared с caller (compute_market_metrics в цепочке §22-форсайта / analyze).
+# Раньше _query_stock/_query_sales_window/_price_sensitivity ловили db.execute-сбой
+# в bare ``except Exception`` без SAVEPOINT/rollback — на реальном Postgres это
+# отравляет db shared с caller: следующий db.execute на этой же сессии падает
+# "current transaction is aborted". Фикс — db.execute (и, для _price_sensitivity,
+# весь вызов _elasticity_coef) обёрнут в ``with db.begin_nested():``. MagicMock не
+# эмулирует реальный aborted-transaction Postgres — тесты проверяют (1) graceful
+# fallback при сбое, (2) что begin_nested() реально вызван, (3) что db остаётся
+# usable для следующего вызова на той же mock-сессии.
+
+
+class TestQueryStockSavepoint:
+    def test_uses_begin_nested_savepoint(self) -> None:
+        db = MagicMock()
+        db.execute.return_value.mappings.return_value.first.return_value = None
+        _query_stock(db, {"districts": ["X"]})
+        assert db.begin_nested.call_count == 1
+
+    def test_query_failure_degrades_to_zero_counts(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("simulated DB failure")
+        out = _query_stock(db, {"districts": ["X"]})
+        assert out == {
+            "n_lots": 0,
+            "n_sold": 0,
+            "n_available": 0,
+            "obj_count": 0,
+            "n_long_unsold": 0,
+        }
+
+    def test_session_usable_after_failure(self) -> None:
+        """db.execute сбоит один раз (stock), но db остаётся usable для следующего
+        (не связанного) вызова на этой же сессии — имитирует
+        _query_stock → _query_sales_window → _price_sensitivity chain."""
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("boom")
+        out = _query_stock(db, {"districts": ["X"]})
+        assert out["n_lots"] == 0
+
+        db.execute.side_effect = None
+        next_result = MagicMock()
+        next_result.mappings.return_value.all.return_value = [{"marker": "ok"}]
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]
+
+
+class TestQuerySalesWindowSavepoint:
+    def test_uses_begin_nested_savepoint(self) -> None:
+        db = MagicMock()
+        db.execute.return_value.mappings.return_value.all.return_value = []
+        _query_sales_window(db, {"districts": ["X"]})
+        assert db.begin_nested.call_count == 1
+
+    def test_query_failure_degrades_to_empty(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("simulated DB failure")
+        units, area, by_room = _query_sales_window(db, {"districts": ["X"]})
+        assert (units, area, by_room) == (0, 0.0, {})
+
+    def test_session_usable_after_failure(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("boom")
+        units, _area, _by_room = _query_sales_window(db, {"districts": ["X"]})
+        assert units == 0
+
+        db.execute.side_effect = None
+        next_result = MagicMock()
+        next_result.mappings.return_value.all.return_value = [{"marker": "ok"}]
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]
+
+
+class TestPriceSensitivitySavepoint:
+    """_price_sensitivity делает db.execute КОСВЕННО (через _elasticity_coef) —
+    SAVEPOINT обёрнут вокруг всего вызова, а не голого db.execute."""
+
+    def test_uses_begin_nested_savepoint(self, _patch_resolver: MagicMock) -> None:
+        _patch_resolver.side_effect = lambda _db, d: [d] if d is not None else None
+        db = MagicMock()
+        with patch(_ELAST, return_value={"elasticity": -1.4, "source": "regression"}):
+            _price_sensitivity(db, district="Автовокзал", window_months=6)
+        assert db.begin_nested.call_count == 1
+
+    def test_elasticity_failure_still_degrades_gracefully(self, _patch_resolver: MagicMock) -> None:
+        _patch_resolver.side_effect = lambda _db, d: [d] if d is not None else None
+        db = MagicMock()
+        with patch(_ELAST, side_effect=RuntimeError("simulated DB failure")):
+            value, source = _price_sensitivity(db, district="Автовокзал", window_months=6)
+        assert (value, source) == (None, None)
+        assert db.begin_nested.call_count == 1
+
+    def test_session_usable_after_failure(self, _patch_resolver: MagicMock) -> None:
+        """Сбой elasticity-цепочки не отравляет db — следующий вызов на этой же
+        сессии (например, offer-price-trend) отрабатывает нормально."""
+        _patch_resolver.side_effect = lambda _db, d: [d] if d is not None else None
+        db = MagicMock()
+        with patch(_ELAST, side_effect=RuntimeError("boom")):
+            _price_sensitivity(db, district="Автовокзал", window_months=6)
+
+        next_result = MagicMock()
+        next_result.mappings.return_value.all.return_value = [{"marker": "ok"}]
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]
+
+
+class TestComputeMarketMetricsSessionUsableAfterFirstFailure:
+    """End-to-end (via compute_market_metrics): stock-запрос сбоит → SAVEPOINT
+    (не bare rollback) → sales-window запрос по-прежнему выполняется на той же
+    сессии (не блокируется "current transaction is aborted")."""
+
+    def test_stock_failure_does_not_poison_sales_window(self, _patch_resolver: MagicMock) -> None:
+        db = MagicMock()
+        sales_result = MagicMock()
+        sales_result.mappings.return_value.all.return_value = _FULL_SALES
+        db.execute.side_effect = [RuntimeError("boom on stock"), sales_result]
+        with patch(_ELAST, return_value={"elasticity": -1.4, "source": "regression"}):
+            m = compute_market_metrics(db, district="Автовокзал", window_months=6)
+        assert db.begin_nested.call_count >= 1
+        assert m.n_lots == 0  # stock fallback (graceful, не crash)
+        assert m.confidence == "low"
+        # sales-window execute РЕАЛЬНО выполнился на той же db (2-й вызов не был
+        # заблокирован сбоем 1-го) — session осталась usable после stock-savepoint.
+        assert db.execute.call_count >= 2
+
+
+class TestOfferPriceTrendSavepoint:
+    """compute_offer_price_trend: db shared с caller (POST .../analyze / §22-отчёт).
+    db.execute обёрнут в begin_nested — 8с-гвард QueryCanceled не отравляет сессию."""
+
+    def test_uses_begin_nested_savepoint(self) -> None:
+        db = _mock_offer_db([])
+        compute_offer_price_trend(db, center_lat=56.83, center_lon=60.6)
+        assert db.begin_nested.call_count == 1
+
+    def test_query_failure_degrades_and_no_bare_rollback(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("query canceled (8s guard)")
+        res = compute_offer_price_trend(db, center_lat=56.83, center_lon=60.6)
+        assert res.points == []  # документированный fallback
+        assert res.has_data is False
+        assert db.begin_nested.call_count == 1
+        db.rollback.assert_not_called()  # никакого bare rollback на shared сессии
+
+    def test_session_usable_after_failure(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("db down")
+        res = compute_offer_price_trend(db, center_lat=56.83, center_lon=60.6)
+        assert res.points == []
+
+        db.execute.side_effect = None
+        next_result = MagicMock()
+        next_result.mappings.return_value.all.return_value = [{"marker": "ok"}]
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]

backend/tests/services/site_finder/test_own_portfolio.py

@@ -23,7 +23,12 @@ from typing import Any
 from unittest.mock import MagicMock
 
 from app.services.site_finder import own_portfolio
-from app.services.site_finder.own_portfolio import OwnProject, get_own_portfolio
+from app.services.site_finder.own_portfolio import (
+    OwnProject,
+    _query_current,
+    _query_future,
+    get_own_portfolio,
+)
 
 
 def _row(data: dict[str, Any]) -> MagicMock:
@@ -272,3 +277,81 @@ def test_future_query_failure_returns_empty_gracefully(monkeypatch) -> None:
     out = get_own_portfolio(db)
 
     assert out == []
+
+
+# ── SAVEPOINT regression (#2464 cluster A) ───────────────────────────────────────
+#
+# db shared с caller (get_own_portfolio в цепочке §25.3 / analyze). Раньше
+# _query_current/_query_future ловили db.execute-сбой в bare ``except Exception``
+# без SAVEPOINT/rollback — на реальном Postgres это отравляет db shared с caller:
+# следующий db.execute на этой же сессии падает "current transaction is aborted".
+# Фикс — db.execute обёрнут в ``with db.begin_nested():``. MagicMock не эмулирует
+# реальный aborted-transaction Postgres — тесты проверяют (1) graceful fallback при
+# сбое, (2) что begin_nested() реально вызван вокруг db.execute, (3) что db остаётся
+# usable для следующего вызова на той же mock-сессии.
+
+
+class TestQueryCurrentSavepoint:
+    def test_uses_begin_nested_savepoint(self, monkeypatch) -> None:
+        monkeypatch.setattr(own_portfolio.settings, "own_developer_ids", [12345])
+        db = MagicMock()
+        db.execute.return_value = _result([])
+        _query_current(db)
+        assert db.begin_nested.call_count == 1
+
+    def test_query_failure_degrades_to_empty_list(self, monkeypatch) -> None:
+        monkeypatch.setattr(own_portfolio.settings, "own_developer_ids", [12345])
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("simulated DB failure")
+        assert _query_current(db) == []
+
+    def test_session_usable_after_failure(self, monkeypatch) -> None:
+        """db.execute сбоит один раз, но db остаётся usable для следующего вызова
+        на этой же сессии (имитирует _query_current → _query_future в
+        get_own_portfolio на одной shared Session)."""
+        monkeypatch.setattr(own_portfolio.settings, "own_developer_ids", [12345])
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("boom")
+        assert _query_current(db) == []
+
+        db.execute.side_effect = None
+        next_result = _result([{"marker": "ok"}])
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]
+
+
+class TestQueryFutureSavepoint:
+    def test_uses_begin_nested_savepoint(self) -> None:
+        db = MagicMock()
+        db.execute.return_value = _result([])
+        _query_future(db)
+        assert db.begin_nested.call_count == 1
+
+    def test_query_failure_degrades_to_empty_list(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("simulated DB failure")
+        assert _query_future(db) == []
+
+    def test_session_usable_after_failure(self) -> None:
+        db = MagicMock()
+        db.execute.side_effect = RuntimeError("boom")
+        assert _query_future(db) == []
+
+        db.execute.side_effect = None
+        next_result = _result([{"marker": "ok"}])
+        db.execute.return_value = next_result
+        assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}]
+
+
+def test_get_own_portfolio_current_failure_savepoint_isolated(monkeypatch) -> None:
+    """End-to-end (via get_own_portfolio): current db.execute сбоит → SAVEPOINT
+    (не bare rollback) → future по-прежнему считается на той же сессии."""
+    monkeypatch.setattr(own_portfolio.settings, "own_developer_ids", [12345])
+    db = MagicMock()
+    db.execute.side_effect = [RuntimeError("boom on domrf"), _result([_future_row()])]
+
+    out = get_own_portfolio(db)
+
+    assert db.begin_nested.call_count >= 1
+    assert len(out) == 1
+    assert out[0].source == "future"