From c52c2c4063517257d41481c2cc57ebf149f66d51 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 7 Jul 2026 17:32:32 +0500 Subject: [PATCH] fix(site_finder): SAVEPOINT-isolate DB errors across cluster A (#2464) developer_attribution, connection_capacity_lookup (gas/heat helpers), competitors, saturation, supply_layers, and pat_lookup swallowed DB errors without begin_nested()/rollback(), leaving the shared analyze-request Session aborted for every subsequent db.execute (silently breaking persist_analysis_run and the forecast enqueue). zone_regulation's backfill_ekb_zone_regulations gets a plain rollback() since it only ever runs on its own owned Celery-task session. Refs #2464 (cluster A session-poisoning). --- .../app/services/site_finder/competitors.py | 46 +++-- .../site_finder/connection_capacity_lookup.py | 178 +++++++++--------- .../site_finder/developer_attribution.py | 22 ++- .../app/services/site_finder/pat_lookup.py | 7 +- .../app/services/site_finder/saturation.py | 10 +- .../app/services/site_finder/supply_layers.py | 11 +- .../services/site_finder/zone_regulation.py | 12 +- .../tests/api/v1/test_parcel_competitors.py | 57 ++++++ .../test_connection_capacity_savepoint.py | 138 ++++++++++++++ .../site_finder/test_developer_attribution.py | 30 +++ .../services/site_finder/test_pat_subzones.py | 32 ++++ .../site_finder/test_supply_layers.py | 22 +++ .../services/test_zone_regulation_backfill.py | 17 +- backend/tests/test_saturation.py | 58 ++++-- 14 files changed, 514 insertions(+), 126 deletions(-) create mode 100644 backend/tests/services/site_finder/test_connection_capacity_savepoint.py diff --git a/backend/app/services/site_finder/competitors.py b/backend/app/services/site_finder/competitors.py index c65d5e20..8a19ef14 100644 --- a/backend/app/services/site_finder/competitors.py +++ b/backend/app/services/site_finder/competitors.py @@ -799,7 +799,12 @@ def get_competitors( avg_price_map: dict[int, float] = {} price_source_map: dict[int, str] = {} try: - price_rows = db.execute(_AVG_PRICE_SQL, {"obj_ids": obj_ids}).mappings().all() + # SAVEPOINT — db shared с caller (POST .../competitors, Depends(get_db)). Голый + # db.rollback() orphan'ит outer SessionTransaction (velocity.py:170) — тем более + # здесь ЕЩЁ ДВА db.execute идут следом в этой же функции (sold-count, + # objective-fallback) на том же db; без savepoint сбой здесь отравил бы их тоже. + with db.begin_nested(): + price_rows = db.execute(_AVG_PRICE_SQL, {"obj_ids": obj_ids}).mappings().all() for r in price_rows: oid = int(r["obj_id"]) price = _row_get(r, "avg_price_per_m2") @@ -818,11 +823,16 @@ def get_competitors( # flats_sold → None → нейтральный stage. sold_count_map: dict[int, int] = {} try: - sold_rows = ( - db.execute(_SOLD_COUNT_SQL, {"obj_ids": obj_ids, "premise_kind": _SOLD_PREMISE_KIND}) - .mappings() - .all() - ) + # SAVEPOINT — db shared, см. avg_price блок выше (иначе отравляет objective + # price fallback ниже на том же db). + with db.begin_nested(): + sold_rows = ( + db.execute( + _SOLD_COUNT_SQL, {"obj_ids": obj_ids, "premise_kind": _SOLD_PREMISE_KIND} + ) + .mappings() + .all() + ) for r in sold_rows: oid = int(r["obj_id"]) sold = _row_get(r, "flats_sold") @@ -838,17 +848,21 @@ def get_competitors( missing_price_ids = [oid for oid in obj_ids if oid not in avg_price_map] if missing_price_ids: try: - obj_price_rows = ( - db.execute( - _OBJECTIVE_PRICE_FALLBACK_SQL, - { - "obj_ids": missing_price_ids, - "velocity_match_radius_m": _VELOCITY_MATCH_RADIUS_M, - }, + # SAVEPOINT — db shared, см. avg_price блок выше. Последний из трёх + # db.execute в get_competitors — тоже под savepoint, чтобы дальнейшие + # запросы analyze на этом же db (persist_analysis_run и т.п.) не отравились. + with db.begin_nested(): + obj_price_rows = ( + db.execute( + _OBJECTIVE_PRICE_FALLBACK_SQL, + { + "obj_ids": missing_price_ids, + "velocity_match_radius_m": _VELOCITY_MATCH_RADIUS_M, + }, + ) + .mappings() + .all() ) - .mappings() - .all() - ) for r in obj_price_rows: oid = int(r["obj_id"]) price = _row_get(r, "median_price_per_m2") diff --git a/backend/app/services/site_finder/connection_capacity_lookup.py b/backend/app/services/site_finder/connection_capacity_lookup.py index 155f5494..2c43e552 100644 --- a/backend/app/services/site_finder/connection_capacity_lookup.py +++ b/backend/app/services/site_finder/connection_capacity_lookup.py @@ -260,25 +260,29 @@ def _query_gas_city_grs(db: Session) -> dict: """ outlet_counts = _query_gas_outlet_counts(db) try: - rows = ( - db.execute( - text(""" - SELECT grs_name, - SUM(design_capacity_th_m3_h) AS design_sum, - SUM(free_capacity_th_m3_h) AS free_sum, - AVG(free_capacity_pct) AS pct_avg, - MAX(upgrade_due) AS upgrade_due, - COUNT(*) AS outputs_count - FROM gas_grs_capacity - WHERE grs_name ILIKE '%свердловск%' - OR grs_name ILIKE '%екатеринбург%' - GROUP BY grs_name - ORDER BY grs_name - """) + # SAVEPOINT — db shared с caller (GET .../connection-capacity, Depends(get_db)). + # Голый db.rollback() orphan'ит outer SessionTransaction (см. velocity.py:170) — + # begin_nested откатывает только эту savepoint, outer tx остаётся usable. + with db.begin_nested(): + rows = ( + db.execute( + text(""" + SELECT grs_name, + SUM(design_capacity_th_m3_h) AS design_sum, + SUM(free_capacity_th_m3_h) AS free_sum, + AVG(free_capacity_pct) AS pct_avg, + MAX(upgrade_due) AS upgrade_due, + COUNT(*) AS outputs_count + FROM gas_grs_capacity + WHERE grs_name ILIKE '%свердловск%' + OR grs_name ILIKE '%екатеринбург%' + GROUP BY grs_name + ORDER BY grs_name + """) + ) + .mappings() + .all() ) - .mappings() - .all() - ) except Exception as e: # Таблица может не существовать до применения миграции 181 (schema-first). # Блок аддитивный — деградируем в пустой, не роняя весь эндпоинт. Счётчики @@ -333,25 +337,27 @@ def _query_gas_outlet_counts(db: Session) -> dict: Таблицы может не быть (миграция 184 не применена) → все нули (graceful, аддитивно). """ try: - row = ( - db.execute( - text(""" - SELECT COUNT(*) AS outlets_total, - COUNT(*) FILTER ( - WHERE free_capacity_mln_m3 < 0 - ) AS outlets_deficit, - COUNT(*) FILTER ( - WHERE free_capacity_needs_calc - ) AS outlets_needs_calc - FROM gas_grs_outlet_points - WHERE period_month = ( - SELECT MAX(period_month) FROM gas_grs_outlet_points - ) - """) + # SAVEPOINT — db shared с caller (см. _query_gas_city_grs выше). + with db.begin_nested(): + row = ( + db.execute( + text(""" + SELECT COUNT(*) AS outlets_total, + COUNT(*) FILTER ( + WHERE free_capacity_mln_m3 < 0 + ) AS outlets_deficit, + COUNT(*) FILTER ( + WHERE free_capacity_needs_calc + ) AS outlets_needs_calc + FROM gas_grs_outlet_points + WHERE period_month = ( + SELECT MAX(period_month) FROM gas_grs_outlet_points + ) + """) + ) + .mappings() + .first() ) - .mappings() - .first() - ) except Exception as e: # Таблица может не существовать до применения миграции 184 (schema-first). # Счётчики аддитивны — деградируем в нули, не роняя весь эндпоинт. @@ -384,39 +390,41 @@ def _query_gas_outlet_points(db: Session, parcel_wkt: str) -> list[dict]: distance_m, lat, lon}] — отсортировано по distance ASC. """ try: - rows = ( - db.execute( - text(""" - SELECT outlet_name, consumer_type, free_capacity_mln_m3, - free_capacity_needs_calc AS needs_calc, - ST_Distance( - geom::geography, - ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography - ) AS distance_m, - ST_Y(geom) AS lat, - ST_X(geom) AS lon - FROM gas_grs_outlet_points - WHERE geom IS NOT NULL - AND period_month = ( - SELECT MAX(period_month) FROM gas_grs_outlet_points - ) - AND ST_DWithin( - geom::geography, - ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography, - CAST(:radius_m AS float) - ) - ORDER BY distance_m ASC - LIMIT CAST(:limit AS int) - """), - { - "wkt": parcel_wkt, - "radius_m": _GAS_OUTLET_RADIUS_M, - "limit": _GAS_OUTLET_LIMIT, - }, + # SAVEPOINT — db shared с caller (см. _query_gas_city_grs выше). + with db.begin_nested(): + rows = ( + db.execute( + text(""" + SELECT outlet_name, consumer_type, free_capacity_mln_m3, + free_capacity_needs_calc AS needs_calc, + ST_Distance( + geom::geography, + ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography + ) AS distance_m, + ST_Y(geom) AS lat, + ST_X(geom) AS lon + FROM gas_grs_outlet_points + WHERE geom IS NOT NULL + AND period_month = ( + SELECT MAX(period_month) FROM gas_grs_outlet_points + ) + AND ST_DWithin( + geom::geography, + ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography, + CAST(:radius_m AS float) + ) + ORDER BY distance_m ASC + LIMIT CAST(:limit AS int) + """), + { + "wkt": parcel_wkt, + "radius_m": _GAS_OUTLET_RADIUS_M, + "limit": _GAS_OUTLET_LIMIT, + }, + ) + .mappings() + .all() ) - .mappings() - .all() - ) except Exception as e: # Таблица может не существовать до применения миграции 184 (schema-first). # Слой аддитивный — деградируем в пустой, не роняя весь эндпоинт. @@ -451,23 +459,25 @@ def _query_heat_latest(db: Session) -> dict: "total_reserve_gcal_h": float|None}. """ try: - rows = ( - db.execute( - text(""" - SELECT h.org, h.system_name, h.reserve_gcal_h, h.period - FROM heat_system_reserves h - WHERE h.period IS NOT NULL - AND h.period = ( - SELECT MAX(h2.period) FROM heat_system_reserves h2 - WHERE h2.period IS NOT NULL - AND h2.org = h.org - ) - ORDER BY h.org, h.system_name - """) + # SAVEPOINT — db shared с caller (см. _query_gas_city_grs выше). + with db.begin_nested(): + rows = ( + db.execute( + text(""" + SELECT h.org, h.system_name, h.reserve_gcal_h, h.period + FROM heat_system_reserves h + WHERE h.period IS NOT NULL + AND h.period = ( + SELECT MAX(h2.period) FROM heat_system_reserves h2 + WHERE h2.period IS NOT NULL + AND h2.org = h.org + ) + ORDER BY h.org, h.system_name + """) + ) + .mappings() + .all() ) - .mappings() - .all() - ) except Exception as e: # Таблица может не существовать до применения миграции 182 (schema-first). # Блок аддитивный — деградируем в пустой, не роняя весь эндпоинт. diff --git a/backend/app/services/site_finder/developer_attribution.py b/backend/app/services/site_finder/developer_attribution.py index f4523917..9b818660 100644 --- a/backend/app/services/site_finder/developer_attribution.py +++ b/backend/app/services/site_finder/developer_attribution.py @@ -142,14 +142,22 @@ def get_developer_attribution( - БД-ошибке, включая ещё не задеплоенную миграцию 149 (analyze не падает). """ try: - rows = ( - db.execute( - _DEVELOPER_FOR_PARCEL_SQL, - {"cad_num": cad_num, "radius": radius_m}, + # SAVEPOINT — db получена от caller (shared request-scoped Session, + # analyze_parcel в parcels.py). Голый db.rollback() здесь НЕЛЬЗЯ — он + # orphan'ит outer SessionTransaction и роняет persist_analysis_run сразу + # после (см. velocity.py:170 / connection_capacity_lookup._query_nearby_ + # network_zones). Ошибка внутри begin_nested() откатывает ТОЛЬКО эту + # savepoint до propagate — outer-транзакция остаётся usable для + # последующих db.execute (#2464 cluster A finding 1). + with db.begin_nested(): + rows = ( + db.execute( + _DEVELOPER_FOR_PARCEL_SQL, + {"cad_num": cad_num, "radius": radius_m}, + ) + .mappings() + .all() ) - .mappings() - .all() - ) except (OperationalError, ProgrammingError, DataError) as exc: # Миграция 149 ещё не задеплоена / БД-ошибка — graceful degrade. logger.warning( diff --git a/backend/app/services/site_finder/pat_lookup.py b/backend/app/services/site_finder/pat_lookup.py index 32b1f6d1..2ce9586f 100644 --- a/backend/app/services/site_finder/pat_lookup.py +++ b/backend/app/services/site_finder/pat_lookup.py @@ -45,7 +45,12 @@ def parcel_pat_subzones(db: Session, parcel_wkt: str | None) -> list[dict[str, A return [] try: - rows = db.execute(_PAT_INTERSECTS_SQL, {"parcel_wkt": parcel_wkt}).mappings().all() + # SAVEPOINT — db shared с caller (analyze_parcel, shared request-scoped Session). + # Голый db.rollback() orphan'ит outer SessionTransaction (см. velocity.py:170) — + # begin_nested откатывает только эту savepoint, не мешая persist_analysis_run + # и прочим db.execute дальше по /analyze (#2464 cluster A finding 7). + with db.begin_nested(): + rows = db.execute(_PAT_INTERSECTS_SQL, {"parcel_wkt": parcel_wkt}).mappings().all() except (OperationalError, ProgrammingError) as exc: # Таблица ещё не задеплоена / структура изменилась — graceful degrade. logger.warning("parcel_pat_subzones: pat_subzones недоступна, skip: %s", exc) diff --git a/backend/app/services/site_finder/saturation.py b/backend/app/services/site_finder/saturation.py index 4e6d4a0a..0c2faa51 100644 --- a/backend/app/services/site_finder/saturation.py +++ b/backend/app/services/site_finder/saturation.py @@ -237,7 +237,15 @@ def compute_district_saturation(db: Session, district_name: str) -> dict[str, An psycopg v3: bind через CAST(:dn AS text). Read-only. """ try: - row = db.execute(_DISTRICT_SATURATION_SQL, {"dn": district_name}).mappings().first() + # SAVEPOINT — db shared с caller (analyze_parcel, parcels.py). Caller УЖЕ + # оборачивает вызов этой функции в свой db.begin_nested(), но это НЕ спасает: + # если db.execute здесь падает и мы её же тут глотаем (return None, не re-raise), + # исключение никогда не долетает до caller-savepoint → её ROLLBACK TO SAVEPOINT + # никогда не срабатывает, а сам Postgres-connection уже aborted. Нужна СОБСТВЕННАЯ + # savepoint здесь — откатывается на выходе из `with` ДО того, как исключение + # поймано ниже, оставляя outer-транзакцию (caller'а) usable. + with db.begin_nested(): + row = db.execute(_DISTRICT_SATURATION_SQL, {"dn": district_name}).mappings().first() except Exception: logger.exception("saturation: query failed (district=%s)", district_name) return None diff --git a/backend/app/services/site_finder/supply_layers.py b/backend/app/services/site_finder/supply_layers.py index 8444a66f..a83623fa 100644 --- a/backend/app/services/site_finder/supply_layers.py +++ b/backend/app/services/site_finder/supply_layers.py @@ -570,9 +570,18 @@ def _safe_rows( Зеркало market_metrics._query_* — тонкие данные/сбой БД не должны валить расчёт остальных слоёв или worker'а целиком. + + SAVEPOINT (``db.begin_nested()``): ``compute_all_layers`` вызывает эту функцию 3× + (L1/L2/L3) на ОДНОЙ Session — без savepoint сбой L1 отравил бы транзакцию и L2/L3 + молча получили бы [] тоже (плюс любой db.execute ПОСЛЕ compute_all_layers на этом же + db, включая caller'а — worker owns SessionLocal(), но при вызове из shared + request-scoped db, напр. forecasting/orchestrator.build_site_finder_report, + отравился бы весь оставшийся запрос). Голый db.rollback() здесь НЕЛЬЗЯ — см. + velocity.py:170. """ try: - return db.execute(sql, dict(params)).mappings().all() + with db.begin_nested(): + return db.execute(sql, dict(params)).mappings().all() except Exception: # L1 фильтрует по resolved-микро (`districts`); L2/L3 — по админ `district`. district_dbg = params.get("district", params.get("districts")) diff --git a/backend/app/services/site_finder/zone_regulation.py b/backend/app/services/site_finder/zone_regulation.py index bcd7230a..cf5fdb6d 100644 --- a/backend/app/services/site_finder/zone_regulation.py +++ b/backend/app/services/site_finder/zone_regulation.py @@ -470,9 +470,15 @@ def backfill_ekb_zone_regulations( upserted += 1 except Exception as exc: # Per-zone изоляция: один сбойный фетч/апсёрт не должен ронять весь прогон. - logger.warning( - "backfill_ekb_zone_regulations: зона idx=%s failed: %s", zone_index, exc - ) + # rollback() (не begin_nested): db здесь — СОБСТВЕННАЯ сессия таска + # backfill_zone_regulations.py (SessionLocal() + try/finally close), не + # shared request-scoped Session — orphan'ить нечего, следующая итерация + # цикла (get_cached_zone_regulation / upsert_zone_regulation) на этом же db + # иначе поймает "current transaction is aborted" после сбойного db.commit() + # выше (#2464 cluster A finding 6; сиблинг — write_db.rollback() в + # get_or_fetch_zone_regulation, тоже owned-сессия). + db.rollback() + logger.warning("backfill_ekb_zone_regulations: зона idx=%s failed: %s", zone_index, exc) failed += 1 continue diff --git a/backend/tests/api/v1/test_parcel_competitors.py b/backend/tests/api/v1/test_parcel_competitors.py index ca3d4332..d49138ab 100644 --- a/backend/tests/api/v1/test_parcel_competitors.py +++ b/backend/tests/api/v1/test_parcel_competitors.py @@ -545,6 +545,63 @@ def test_competitors_no_price_anywhere_is_none() -> None: app.dependency_overrides.clear() +# ── SAVEPOINT-регрессия (#2464 cluster A finding 3) ────────────────────────── + + +def test_competitors_avg_price_failure_does_not_poison_later_queries() -> None: + """avg_price query падает → sold-count и objective-fallback (следом на ТОЙ ЖЕ + Session) всё равно отрабатывают, а не ловят "current transaction is aborted". + + Раньше avg_price/sold-count/objective-fallback ловили db.execute-сбой без + SAVEPOINT (db.begin_nested()) — сбой ЛЮБОГО из трёх отравлял shared db для + ОСТАЛЬНЫХ двух (и для persist_analysis_run дальше по /analyze). Фикс оборачивает + каждый в свою begin_nested(); здесь ловим именно avg_price и проверяем, что + objective-fallback (третий execute) всё равно наполняет цену. + """ + coord_result = MagicMock() + coord_result.mappings.return_value.first.return_value = _coord_row() + obj_result = MagicMock() + obj_result.mappings.return_value.all.return_value = [_obj_row(obj_id=1)] + sold_result = MagicMock() + sold_result.mappings.return_value.all.return_value = [] + obj_price_result = MagicMock() + obj_price_result.mappings.return_value.all.return_value = [ + _obj_price_row(obj_id=1, price=142_000.0) + ] + + db = MagicMock() + db.execute.side_effect = [ + coord_result, + obj_result, + RuntimeError("simulated avg_price DB failure"), # avg_price query сбоит + sold_result, + obj_price_result, # objective-fallback ДОЛЖЕН всё равно выполниться + ] + + from app.core.db import get_db + + app.dependency_overrides[get_db] = _override_db(db) + try: + client = TestClient(app) + resp = client.post( + "/api/v1/parcels/66:41:0303161:5/competitors", + json={"radius_km": 1.0, "time_window": "last_quarter"}, + ) + assert resp.status_code == 200, resp.text + comp = resp.json()["competitors"][0] + # avg_price сбойнул (без domrf-цены), но objective-fallback наполнил её — + # это возможно ТОЛЬКО если сбой avg_price не отравил сессию для fallback. + assert comp["avg_price_per_m2"] == pytest.approx(142_000.0) + assert comp["price_source"] == "objective" + # Все 5 запланированных execute реально дошли (sold-count и fallback не + # были пропущены/не упали каскадом из-за отравленной сессии). + assert db.execute.call_count == 5 + # avg_price/sold-count/objective-fallback — каждый под своим SAVEPOINT. + assert db.begin_nested.call_count == 3 + finally: + app.dependency_overrides.clear() + + def test_competitors_multi_object_radius_yields_per_object_prices() -> None: """Регрессия #2445 A5 (byte-identical к best_layouts #1956): несколько конкурентов в радиусе → КАЖДЫЙ получает свою domrf-цену, а не «1 объект на diff --git a/backend/tests/services/site_finder/test_connection_capacity_savepoint.py b/backend/tests/services/site_finder/test_connection_capacity_savepoint.py new file mode 100644 index 00000000..ee721903 --- /dev/null +++ b/backend/tests/services/site_finder/test_connection_capacity_savepoint.py @@ -0,0 +1,138 @@ +"""SAVEPOINT-регрессия для connection_capacity_lookup.py (#2464 cluster A finding 2). + +gas/heat/gas-outlet query helpers (_query_gas_city_grs, _query_gas_outlet_counts, +_query_gas_outlet_points, _query_heat_latest) когда-то ловили db.execute-сбой в +bare ``except Exception`` без SAVEPOINT/rollback — на реальном Postgres это отравляло +db shared с caller (GET .../connection-capacity, Depends(get_db)): следующий db.execute +в этом же запросе падал с "current transaction is aborted". Сиблинг +``_query_nearby_network_zones`` (в этом же файле) уже был правильным (db.begin_nested()) +— эти тесты проверяют, что остальные функции приведены в соответствие. + +MagicMock-сессия не эмулирует реальный aborted-transaction Postgres — тесты +проверяют (1) graceful fallback при сбое, (2) что begin_nested() реально вызван +вокруг db.execute (SAVEPOINT-обёртка присутствует в коде), (3) что db остаётся +usable для следующего вызова на том же mock-сессии (call_count растёт, ошибка не +проглатывает control flow). +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services.site_finder.connection_capacity_lookup import ( + _query_gas_city_grs, + _query_gas_outlet_counts, + _query_gas_outlet_points, + _query_heat_latest, +) + +_WKT = "POLYGON((60.8 56.7, 60.9 56.7, 60.9 56.8, 60.8 56.8, 60.8 56.7))" + + +def _mock_db_raising_once(final_rows: list[dict] | None = None) -> MagicMock: + """db.execute кидает на 1-м вызове, на 2-м успешно отдаёт final_rows.""" + db = MagicMock() + ok_result = MagicMock() + ok_result.mappings.return_value.all.return_value = final_rows or [] + ok_result.mappings.return_value.first.return_value = (final_rows or [None])[0] + db.execute.side_effect = [RuntimeError("simulated DB failure"), ok_result] + return db + + +class TestGasCityGrsSavepoint: + def test_query_failure_degrades_gracefully(self) -> None: + db = MagicMock() + db.execute.side_effect = RuntimeError("gas_grs_capacity недоступна") + out = _query_gas_city_grs(db) + assert out["city_grs"] == [] + assert out["total_free_th_m3_h"] is None + + def test_uses_begin_nested_savepoint(self) -> None: + """Фикс: query обёрнут в db.begin_nested(), не голый db.execute.""" + db = MagicMock() + db.execute.return_value.mappings.return_value.all.return_value = [] + db.execute.return_value.mappings.return_value.first.return_value = None + _query_gas_city_grs(db) + # 2 execute на функцию: _query_gas_outlet_counts (city-level счётчики) + + # сам gas_grs_capacity query — оба теперь под begin_nested. + assert db.begin_nested.call_count >= 1 + + def test_session_usable_after_failure(self) -> None: + """db.execute сбоит один раз (city_grs), но db остаётся usable для + следующего (не связанного) вызова на этой же сессии.""" + db = _mock_db_raising_once() + # outlet_counts вызывается ПЕРВЫМ внутри _query_gas_city_grs — заставим его + # тоже отдать пустой результат нормально, а основной query — кинуть. + db.execute.side_effect = [ + MagicMock( + mappings=MagicMock(return_value=MagicMock(first=lambda: None)) + ), # outlet_counts + RuntimeError("main query failed"), # gas_grs_capacity query + ] + out = _query_gas_city_grs(db) + assert out["city_grs"] == [] + # Следующий вызов на том же db.execute отрабатывает (session не отравлена). + next_result = MagicMock() + next_result.mappings.return_value.all.return_value = [{"marker": "ok"}] + db.execute.side_effect = None + db.execute.return_value = next_result + assert db.execute("SELECT 1").mappings().all() == [{"marker": "ok"}] + + +class TestGasOutletCountsSavepoint: + def test_query_failure_degrades_to_zeros(self) -> None: + db = MagicMock() + db.execute.side_effect = RuntimeError("gas_grs_outlet_points недоступна") + out = _query_gas_outlet_counts(db) + assert out == {"outlets_total": 0, "outlets_deficit": 0, "outlets_needs_calc": 0} + + def test_uses_begin_nested_savepoint(self) -> None: + db = MagicMock() + db.execute.return_value.mappings.return_value.first.return_value = None + _query_gas_outlet_counts(db) + assert db.begin_nested.call_count == 1 + + +class TestGasOutletPointsSavepoint: + def test_query_failure_degrades_to_empty_list(self) -> None: + db = MagicMock() + db.execute.side_effect = RuntimeError("gas_grs_outlet_points (радиус) недоступна") + assert _query_gas_outlet_points(db, _WKT) == [] + + def test_uses_begin_nested_savepoint(self) -> None: + db = MagicMock() + db.execute.return_value.mappings.return_value.all.return_value = [] + _query_gas_outlet_points(db, _WKT) + assert db.begin_nested.call_count == 1 + + +class TestHeatLatestSavepoint: + def test_query_failure_degrades_gracefully(self) -> None: + db = MagicMock() + db.execute.side_effect = RuntimeError("heat_system_reserves недоступна") + out = _query_heat_latest(db) + assert out == {"systems": [], "total_reserve_gcal_h": None} + + def test_uses_begin_nested_savepoint(self) -> None: + db = MagicMock() + db.execute.return_value.mappings.return_value.all.return_value = [] + _query_heat_latest(db) + assert db.begin_nested.call_count == 1 + + def test_session_usable_after_failure(self) -> None: + """execute сбоит один раз, db остаётся usable для следующего вызова — + имитирует heat → gas_outlet_points → nearby_network_zones в + get_connection_capacity() на одной shared Session.""" + db = MagicMock() + db.execute.side_effect = RuntimeError("boom") + out = _query_heat_latest(db) + assert out["systems"] == [] + + db.execute.side_effect = None + ok_result = MagicMock() + ok_result.mappings.return_value.all.return_value = [{"marker": "next-query-ok"}] + db.execute.return_value = ok_result + assert db.execute("SELECT 1").mappings().all() == [{"marker": "next-query-ok"}] diff --git a/backend/tests/services/site_finder/test_developer_attribution.py b/backend/tests/services/site_finder/test_developer_attribution.py index e2048e73..c16834c2 100644 --- a/backend/tests/services/site_finder/test_developer_attribution.py +++ b/backend/tests/services/site_finder/test_developer_attribution.py @@ -289,3 +289,33 @@ def test_unexpected_error_also_degrades_to_none() -> None: db.execute.side_effect = RuntimeError("boom") result = get_developer_attribution(db, "66:41:0303161:1") assert result is None + + +# ── SAVEPOINT-регрессия (#2464 cluster A finding 1) ────────────────────────── + + +def test_db_error_uses_savepoint_and_session_stays_usable() -> None: + """Фикс: db.execute обёрнут в db.begin_nested() (SAVEPOINT), НЕ голый rollback. + + db здесь — shared request-scoped Session (Depends(get_db) в parcels.py analyze_parcel); + без SAVEPOINT сбой fn_developer_for_parcel отравил бы persist_analysis_run сразу + после (parcels.py:4116). Мокаем db.execute: 1-й вызов (внутри + get_developer_attribution) кидает, 2-й (имитация persist_analysis_run следом на + том же db) отрабатывает нормально. + """ + db = MagicMock() + ok_result = MagicMock() + ok_result.mappings.return_value.all.return_value = [{"marker": "persist-ok"}] + db.execute.side_effect = [ + ProgrammingError("stmt", {}, Exception("fn_developer_for_parcel does not exist")), + ok_result, + ] + + result = get_developer_attribution(db, "66:41:0303161:1") + + assert result is None + assert db.begin_nested.call_count == 1 # SAVEPOINT вокруг резолвера + # Сессия осталась usable для следующего (не связанного) db.execute. + next_call = db.execute("SELECT 1") + assert next_call.mappings().all() == [{"marker": "persist-ok"}] + assert db.execute.call_count == 2 diff --git a/backend/tests/services/site_finder/test_pat_subzones.py b/backend/tests/services/site_finder/test_pat_subzones.py index 8401559b..826f8ef0 100644 --- a/backend/tests/services/site_finder/test_pat_subzones.py +++ b/backend/tests/services/site_finder/test_pat_subzones.py @@ -8,6 +8,7 @@ from __future__ import annotations import importlib import pathlib +from contextlib import contextmanager from typing import Any import pytest @@ -211,6 +212,10 @@ class _Result: class _FakeDB: + """begin_nested — реальный (не-swallow) CM: зеркалит SAVEPOINT-фикс #2464 + cluster A finding 7 (parcel_pat_subzones); исключение из db.execute должно + долетать до except в проде, не глушиться на выходе из ``with``.""" + def __init__( self, rows: list[dict[str, Any]] | None = None, @@ -218,8 +223,14 @@ class _FakeDB: ) -> None: self._rows = rows or [] self._raise = raise_exc + self.execute_calls = 0 + + @contextmanager + def begin_nested(self): # type: ignore[no-untyped-def] + yield def execute(self, sql: Any, params: dict[str, Any] | None = None) -> _Result: + self.execute_calls += 1 if self._raise is not None: raise self._raise return _Result(self._rows) @@ -270,6 +281,27 @@ def test_pat_lookup_graceful_programming_error() -> None: assert result == [] +def test_pat_lookup_db_error_leaves_session_usable_for_next_query() -> None: + """#2464 cluster A finding 7: db shared с caller (analyze_parcel) — сбой здесь + не должен отравлять session для persist_analysis_run и прочих db.execute дальше + по /analyze. Симулируем: execute кидает ОДИН раз, затем следующий (не связанный) + execute на ТОЙ ЖЕ session отрабатывает нормально.""" + from sqlalchemy.exc import OperationalError + + db = _FakeDB(raise_exc=OperationalError("stmt", {}, Exception("relation does not exist"))) + result = parcel_pat_subzones(db, _WKT) + assert result == [] + assert db.execute_calls == 1 + + # Сессия осталась usable: сбрасываем "raise" (имитируя, что savepoint откатился + # и не поломала outer-транзакцию) и проверяем, что db.execute снова отрабатывает. + db._raise = None + db._rows = [{"subzone_no": 3, "name": "N", "restriction": "R", "aerodrome": "A"}] + next_result = parcel_pat_subzones(db, _WKT) + assert next_result == [{"subzone_no": 3, "name": "N", "restriction": "R", "aerodrome": "A"}] + assert db.execute_calls == 2 + + # ── Резолв пути к JSON (#1150) ────────────────────────────────────────────── diff --git a/backend/tests/services/site_finder/test_supply_layers.py b/backend/tests/services/site_finder/test_supply_layers.py index 048ab910..0e11e83f 100644 --- a/backend/tests/services/site_finder/test_supply_layers.py +++ b/backend/tests/services/site_finder/test_supply_layers.py @@ -704,3 +704,25 @@ class TestComputeAllLayers: db.execute.side_effect = [l1_fail, l2_ok, l3_ok] out = compute_all_layers(db, district="A") assert out == [] + + def test_l1_failure_uses_savepoint_not_bare_rollback(self) -> None: + """#2464 cluster A finding 5: _safe_rows оборачивает db.execute в begin_nested() + (не голый db.rollback()) — L1 падает, но L2/L3 execute на ТОЙ ЖЕ Session должны + быть достижимы (session не отравлена/не orphan'ена outer-транзакция).""" + db = MagicMock() + l1_fail = RuntimeError("l1 down") + l2_ok = MagicMock() + l2_ok.mappings.return_value.all.return_value = [ + {"district_name": "A", "n_objects_total": 1, "n_with_free_flats": 1, "hidden_units": 1} + ] + l3_ok = MagicMock() + l3_ok.mappings.return_value.all.return_value = [] + db.execute.side_effect = [l1_fail, l2_ok, l3_ok] + + out = compute_all_layers(db, district="A") + + # L1 деградировал в [], но L2 (следующий вызов на этом же db) успешно вернул строку — + # begin_nested() внутри _safe_rows не даёт сбою L1 отравить L2/L3. + assert db.execute.call_count == 3 + assert db.begin_nested.call_count == 3 # SAVEPOINT на каждый из L1/L2/L3 + assert any(r.layer == 2 for r in out) diff --git a/backend/tests/services/test_zone_regulation_backfill.py b/backend/tests/services/test_zone_regulation_backfill.py index a17b193b..17720746 100644 --- a/backend/tests/services/test_zone_regulation_backfill.py +++ b/backend/tests/services/test_zone_regulation_backfill.py @@ -63,14 +63,22 @@ class _FakeClient: class _DB: - """Фейковая Session: считает commit'ы, ничего не пишет.""" + """Фейковая Session: считает commit'ы/rollback'и, ничего не пишет. + + rollback() — owned-session (Celery-таск backfill_zone_regulations.py владеет + SessionLocal() целиком) fallback на per-zone сбое (#2464 cluster A finding 6). + """ def __init__(self) -> None: self.commits = 0 + self.rollbacks = 0 def commit(self) -> None: self.commits += 1 + def rollback(self) -> None: + self.rollbacks += 1 + @pytest.fixture def _no_sleep(monkeypatch: Any) -> None: @@ -146,12 +154,17 @@ def test_per_zone_error_isolation(monkeypatch: Any, _no_sleep: None) -> None: zr, "upsert_zone_regulation", lambda db, reg, **kw: upserts.append(reg) or {} ) - result = zr.backfill_ekb_zone_regulations(_DB(), client=client, rate_delay_s=0.0) # type: ignore[arg-type] + db = _DB() + result = zr.backfill_ekb_zone_regulations(db, client=client, rate_delay_s=0.0) # type: ignore[arg-type] assert result["unique_zones"] == 2 assert result["failed"] == 1 # первая зона упала assert result["upserted"] == 1 # вторая прошла → batch не остановился assert len(client.resolved) == 2 # дошли до обеих зон + # #2464 cluster A finding 6: сбойная зона откатывает СВОЮ (owned) сессию — + # иначе следующая итерация цикла поймала бы "current transaction is aborted". + assert db.rollbacks == 1 + assert db.commits == 1 # только вторая (успешная) зона закоммитилась def test_limit_caps_fetches(monkeypatch: Any, _no_sleep: None) -> None: diff --git a/backend/tests/test_saturation.py b/backend/tests/test_saturation.py index eba19e2f..c954439b 100644 --- a/backend/tests/test_saturation.py +++ b/backend/tests/test_saturation.py @@ -5,6 +5,7 @@ DB-слой compute_district_saturation — через минимальный mo (тот же приём, что в test_poi_score.py). """ +from contextlib import contextmanager from datetime import date import pytest @@ -67,28 +68,21 @@ def test_multiplier_none_and_unknown_category_neutral(): def test_provision_unknown_category_none(): assert ( - compute_provision_ratio( - poi_count=5, population=100000, age_share=0.1, category="park" - ) + compute_provision_ratio(poi_count=5, population=100000, age_share=0.1, category="park") is None ) def test_provision_zero_population_none(): assert ( - compute_provision_ratio( - poi_count=5, population=0, age_share=0.1, category="school" - ) - is None + compute_provision_ratio(poi_count=5, population=0, age_share=0.1, category="school") is None ) def test_provision_missing_age_share_none(): """Школа/детсад без age_share → None (когорту не оценить).""" assert ( - compute_provision_ratio( - poi_count=5, population=100000, age_share=None, category="school" - ) + compute_provision_ratio(poi_count=5, population=100000, age_share=None, category="school") is None ) @@ -147,12 +141,21 @@ class _MockResult: class _MockDb: - """Минимальный мок SQLAlchemy Session (как в test_poi_score.py).""" + """Минимальный мок SQLAlchemy Session (как в test_poi_score.py). + + ``begin_nested`` — реальный (не-swallow) context manager, зеркалит SAVEPOINT из + ``compute_district_saturation`` (#2464 cluster A finding 4): исключение внутри + ``with`` должно долетать до ``except`` в проде, не глушиться на выходе из CM. + """ def __init__(self, row: dict | None, *, raise_on_execute: bool = False) -> None: self._row = row self._raise = raise_on_execute + @contextmanager + def begin_nested(self): # type: ignore[no-untyped-def] + yield + def execute(self, *_args: object, **_kwargs: object) -> _MockResult: if self._raise: raise RuntimeError("simulated DB failure") @@ -187,6 +190,39 @@ def test_saturation_db_error_returns_none(): assert compute_district_saturation(db, "Чкаловский") is None +def test_saturation_db_error_leaves_session_usable_for_next_query(): + """#2464 cluster A finding 4: сбой query здесь не должен отравлять session. + + Симулируем реальный сценарий: db.execute кидает ОДИН раз (внутри + compute_district_saturation), затем на ТОЙ ЖЕ session успешно отрабатывает + следующий (не связанный) запрос — как это было бы с persist_analysis_run + сразу после saturation-блока в parcels.py analyze_parcel. Без SAVEPOINT + (begin_nested) второй execute на реальном Postgres упал бы с "current + transaction is aborted, commands ignored until end of transaction block". + """ + + class _FlakyDb: + def __init__(self) -> None: + self.calls = 0 + + @contextmanager + def begin_nested(self): # type: ignore[no-untyped-def] + yield + + def execute(self, *_args: object, **_kwargs: object) -> _MockResult: + self.calls += 1 + if self.calls == 1: + raise RuntimeError("simulated DB failure") + return _MockResult({"marker": "next-query-succeeded"}) + + db = _FlakyDb() + assert compute_district_saturation(db, "Чкаловский") is None + # Сессия осталась usable: следующий (не связанный) db.execute отрабатывает. + result = db.execute("SELECT 1") + assert result.mappings().first() == {"marker": "next-query-succeeded"} + assert db.calls == 2 + + def test_saturation_shape_and_flags(): db = _MockDb(_chkalovsky_row()) out = compute_district_saturation(db, "Чкаловский")