From 73739671a98a887eaab588083550d6a38877511f Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 8 Jul 2026 11:29:31 +0500 Subject: [PATCH] fix(site_finder): SAVEPOINT-isolate supply/market db.execute + kill 2 bare rollbacks (#2464) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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" --- .../app/services/site_finder/best_layouts.py | 112 +++++----- .../app/services/site_finder/custom_pois.py | 19 +- .../app/services/site_finder/future_supply.py | 8 +- .../services/site_finder/market_metrics.py | 43 +++- .../app/services/site_finder/own_portfolio.py | 13 +- backend/tests/api/v1/test_custom_pois.py | 91 +++++++++ .../services/site_finder/test_best_layouts.py | 191 ++++++++++++++++++ .../site_finder/test_future_supply.py | 59 ++++++ .../site_finder/test_market_metrics.py | 160 +++++++++++++++ .../site_finder/test_own_portfolio.py | 85 +++++++- 10 files changed, 712 insertions(+), 69 deletions(-) diff --git a/backend/app/services/site_finder/best_layouts.py b/backend/app/services/site_finder/best_layouts.py index 0f71d933..9c6b73e9 100644 --- a/backend/app/services/site_finder/best_layouts.py +++ b/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 = [] diff --git a/backend/app/services/site_finder/custom_pois.py b/backend/app/services/site_finder/custom_pois.py index e3dba224..eee5b0c3 100644 --- a/backend/app/services/site_finder/custom_pois.py +++ b/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"]), diff --git a/backend/app/services/site_finder/future_supply.py b/backend/app/services/site_finder/future_supply.py index 983cb398..fb244323 100644 --- a/backend/app/services/site_finder/future_supply.py +++ b/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)", diff --git a/backend/app/services/site_finder/market_metrics.py b/backend/app/services/site_finder/market_metrics.py index 77b11863..d6347644 100644 --- a/backend/app/services/site_finder/market_metrics.py +++ b/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 успешен. diff --git a/backend/app/services/site_finder/own_portfolio.py b/backend/app/services/site_finder/own_portfolio.py index 3c983cc1..05f9d527 100644 --- a/backend/app/services/site_finder/own_portfolio.py +++ b/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 [] diff --git a/backend/tests/api/v1/test_custom_pois.py b/backend/tests/api/v1/test_custom_pois.py index 3d9ae0f8..2e3fd6cc 100644 --- a/backend/tests/api/v1/test_custom_pois.py +++ b/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"}] diff --git a/backend/tests/services/site_finder/test_best_layouts.py b/backend/tests/services/site_finder/test_best_layouts.py index 001c86cb..8825c9c5 100644 --- a/backend/tests/services/site_finder/test_best_layouts.py +++ b/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 diff --git a/backend/tests/services/site_finder/test_future_supply.py b/backend/tests/services/site_finder/test_future_supply.py index db56c07b..fb58788e 100644 --- a/backend/tests/services/site_finder/test_future_supply.py +++ b/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 diff --git a/backend/tests/services/site_finder/test_market_metrics.py b/backend/tests/services/site_finder/test_market_metrics.py index 8dd1928a..9d4d959c 100644 --- a/backend/tests/services/site_finder/test_market_metrics.py +++ b/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"}] diff --git a/backend/tests/services/site_finder/test_own_portfolio.py b/backend/tests/services/site_finder/test_own_portfolio.py index 9d825c1c..50fec090 100644 --- a/backend/tests/services/site_finder/test_own_portfolio.py +++ b/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" -- 2.45.3