fix(site-finder): isolate zone-regulation cache write off the request session
get_or_fetch_zone_regulation no longer commits the shared /analyze transaction mid-request on cache-miss. The upsert now runs in a fresh short-lived SessionLocal() (commit/rollback/close), mirroring the zone_regulation_refresh Celery task. Read-back still uses the shared session (READ COMMITTED sees the committed row), so caller return shape is unchanged. Resolves the #1850 item-4 SMELL (the outer analyze tx is designed to commit exactly once via persist_analysis_run). Refs #1850
This commit is contained in:
parent
d2773e0c91
commit
7eee75f657
2 changed files with 184 additions and 11 deletions
|
|
@ -38,6 +38,7 @@ from sqlalchemy import text
|
|||
from sqlalchemy.exc import OperationalError, ProgrammingError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.db import SessionLocal
|
||||
from app.services.scrapers.ekb_geoportal_client import (
|
||||
EKBFeature,
|
||||
EKBGeoportalClient,
|
||||
|
|
@ -337,17 +338,25 @@ def get_or_fetch_zone_regulation(
|
|||
return None
|
||||
if reg is None or not reg.zone_index:
|
||||
return None
|
||||
upsert_zone_regulation(db, reg, city=city)
|
||||
# SMELL (#1850, item 4): db.commit() здесь коммитит ВНЕШНЮЮ /analyze-транзакцию
|
||||
# mid-request (она в норме идёт на SAVEPOINT'ах). upsert_zone_regulation пишет внутри
|
||||
# begin_nested() (SAVEPOINT), но релиз SAVEPOINT не персистит durable — нужен commit
|
||||
# внешней tx. Изолировать апсёрт в собственную короткую транзакцию здесь нельзя без
|
||||
# смены семантики резолвера (он получает SHARED Session от хэндлера, не владеет её
|
||||
# границами); отдельная сессия/коннект рискует потерять только что записанный регламент.
|
||||
# ОСТАВЛЕНО как есть: cache-miss редок (Fix 1 мемоизирует zone_index, кэш ЕКБ прогрет ~33
|
||||
# зоны), поэтому фактический коммит почти не случается в hot-пути. Корректный фикс —
|
||||
# отдельная write-сессия для апсёрта — отдельной задачей, чтобы не рисковать резолвером.
|
||||
db.commit()
|
||||
# #1850 item 4: апсёрт изолирован в СОБСТВЕННУЮ короткую write-сессию (SessionLocal()),
|
||||
# чтобы durable-запись кэша НЕ трогала границы транзакции вызывающего. Резолвер получает
|
||||
# SHARED Session от хэндлера (/analyze коммитит её РОВНО один раз в persist_analysis_run),
|
||||
# поэтому db.commit() здесь раньше преждевременно коммитил весь in-flight analyze-ран.
|
||||
# Свежая сессия коммитит самостоятельно → под READ COMMITTED последующий read-back на
|
||||
# SHARED db видит только что записанную строку. Канонический паттерн — как в
|
||||
# workers/tasks/zone_regulation_refresh.py (SessionLocal → upsert → commit → finally close).
|
||||
write_db = SessionLocal()
|
||||
try:
|
||||
params_numeric = upsert_zone_regulation(write_db, reg, city=city)
|
||||
if params_numeric is not None:
|
||||
write_db.commit()
|
||||
else:
|
||||
write_db.rollback() # graceful: таблица недоступна / нечего персистить
|
||||
except Exception as exc:
|
||||
logger.warning("get_or_fetch_zone_regulation: cache write failed: %s", exc)
|
||||
write_db.rollback()
|
||||
finally:
|
||||
write_db.close()
|
||||
return get_cached_zone_regulation(db, reg.zone_index, city=city)
|
||||
|
||||
|
||||
|
|
|
|||
164
backend/tests/services/test_zone_regulation_write_session.py
Normal file
164
backend/tests/services/test_zone_regulation_write_session.py
Normal file
|
|
@ -0,0 +1,164 @@
|
|||
"""Тесты изолированной write-сессии в get_or_fetch_zone_regulation (#1850 item 4).
|
||||
|
||||
На cache-MISS резолвер пишет регламент в кэш в СОБСТВЕННОЙ короткой сессии (SessionLocal()),
|
||||
а НЕ коммитит SHARED-сессию вызывающего (/analyze коммитит её ровно один раз в конце через
|
||||
persist_analysis_run). Проверяем:
|
||||
- happy path: shared db.commit НИКОГДА не вызывается, fresh write-session коммитит РОВНО раз,
|
||||
сессия закрыта; read-back идёт по SHARED db (форма результата сохранена для callers);
|
||||
- graceful: upsert_zone_regulation вернул None (таблица недоступна) → fresh session rollback,
|
||||
shared db не тронута, функция возвращает результат read-back без падения.
|
||||
|
||||
Сеть/БД не дёргаются: фейковый клиент + monkeypatch SessionLocal (recording-фейк) +
|
||||
отдельная фейковая SHARED Session, считающая commit/rollback.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services.scrapers.ekb_geoportal_client import ZoneRegulation
|
||||
from app.services.site_finder import zone_regulation as zr
|
||||
|
||||
|
||||
def _reg(idx: str) -> ZoneRegulation:
|
||||
return ZoneRegulation(
|
||||
zone_index=idx,
|
||||
zone_full_name=f"{idx} зона",
|
||||
main_vri=[],
|
||||
conditional_vri=[],
|
||||
auxiliary_vri=[],
|
||||
limit_params=[],
|
||||
)
|
||||
|
||||
|
||||
class _MissThenResolveClient:
|
||||
"""Фейковый WFS-клиент: индекс зоны есть, но кэш промахивается → live-резолв регламента."""
|
||||
|
||||
def __init__(self, zone_index: str) -> None:
|
||||
self._zone_index = zone_index
|
||||
self.index_calls = 0
|
||||
self.regulation_calls = 0
|
||||
|
||||
def zone_index_at(self, lon: float, lat: float) -> str:
|
||||
self.index_calls += 1
|
||||
return self._zone_index
|
||||
|
||||
def zone_regulation_at(self, lon: float, lat: float) -> ZoneRegulation:
|
||||
self.regulation_calls += 1
|
||||
return _reg(self._zone_index)
|
||||
|
||||
|
||||
class _SharedDB:
|
||||
"""Фейковая SHARED Session вызывающего: считает commit/rollback, ничего не пишет.
|
||||
|
||||
commit() должен остаться НЕТРОНУТЫМ — фикс #1850 item 4 запрещает преждевременный
|
||||
коммит in-flight analyze-транзакции из резолвера.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.commits = 0
|
||||
self.rollbacks = 0
|
||||
|
||||
def commit(self) -> None:
|
||||
self.commits += 1
|
||||
|
||||
def rollback(self) -> None:
|
||||
self.rollbacks += 1
|
||||
|
||||
|
||||
class _WriteDB:
|
||||
"""Фейковая короткоживущая write-сессия (то, что отдаёт monkeypatched SessionLocal)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.commits = 0
|
||||
self.rollbacks = 0
|
||||
self.closed = False
|
||||
|
||||
def commit(self) -> None:
|
||||
self.commits += 1
|
||||
|
||||
def rollback(self) -> None:
|
||||
self.rollbacks += 1
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_memo() -> Any:
|
||||
"""Чистый memo до/после теста (процесс-локальный кэш течёт между тестами)."""
|
||||
zr._reset_zone_index_memo()
|
||||
yield
|
||||
zr._reset_zone_index_memo()
|
||||
|
||||
|
||||
def test_cache_miss_writes_in_isolated_session(monkeypatch: Any) -> None:
|
||||
"""Cache-miss: shared db.commit НЕ вызывается, fresh write-session коммитит РОВНО раз."""
|
||||
write_db = _WriteDB()
|
||||
created: list[_WriteDB] = []
|
||||
|
||||
def _session_local() -> _WriteDB:
|
||||
created.append(write_db)
|
||||
return write_db
|
||||
|
||||
monkeypatch.setattr(zr, "SessionLocal", _session_local)
|
||||
upserts: list[ZoneRegulation] = []
|
||||
# upsert вернул params_numeric (не None) → write-сессия должна закоммитить.
|
||||
monkeypatch.setattr(
|
||||
zr, "upsert_zone_regulation", lambda db, reg, **kw: upserts.append(reg) or {"x": [1]}
|
||||
)
|
||||
# get_cached дёргается на SHARED db ДВАЖДЫ: 1) до fetch (cache MISS → None, форсит резолв),
|
||||
# 2) read-back после commit write-сессии (READ COMMITTED видит строку → dict для callers).
|
||||
cache_reads: list[Any] = []
|
||||
|
||||
def _cached(db: Any, zone_index: str | None, **kw: Any) -> dict[str, Any] | None:
|
||||
cache_reads.append(db)
|
||||
assert db is shared # резолвер читает кэш ИМЕННО по SHARED db, не по write-сессии
|
||||
return None if len(cache_reads) == 1 else {"zone_index": zone_index}
|
||||
|
||||
monkeypatch.setattr(zr, "get_cached_zone_regulation", _cached)
|
||||
|
||||
shared = _SharedDB()
|
||||
client = _MissThenResolveClient("Ц-1")
|
||||
result = zr.get_or_fetch_zone_regulation(shared, 60.6, 56.8, client=client) # type: ignore[arg-type]
|
||||
|
||||
# SHARED транзакция НЕ тронута резолвером.
|
||||
assert shared.commits == 0
|
||||
assert shared.rollbacks == 0
|
||||
# Свежая write-сессия закоммитила РОВНО один раз и закрыта.
|
||||
assert len(created) == 1
|
||||
assert write_db.commits == 1
|
||||
assert write_db.rollbacks == 0
|
||||
assert write_db.closed is True
|
||||
# Апсёрт ушёл в write-сессию, не в shared.
|
||||
assert upserts == [_reg("Ц-1")]
|
||||
# get_cached на SHARED db: cache-check (miss) + read-back (hit) = 2 чтения.
|
||||
assert len(cache_reads) == 2
|
||||
# Read-back вернул кэшированную форму (dict) для callers parcels.py / ird_analyze.py.
|
||||
assert result == {"zone_index": "Ц-1"}
|
||||
|
||||
|
||||
def test_cache_miss_table_missing_rolls_back_gracefully(monkeypatch: Any) -> None:
|
||||
"""upsert вернул None (таблица недоступна) → fresh session rollback, shared не тронута."""
|
||||
write_db = _WriteDB()
|
||||
monkeypatch.setattr(zr, "SessionLocal", lambda: write_db)
|
||||
# upsert_zone_regulation graceful-возврат None (OperationalError/ProgrammingError внутри).
|
||||
monkeypatch.setattr(zr, "upsert_zone_regulation", lambda db, reg, **kw: None)
|
||||
# Кэш недоступен и для read-back → None (форма как при недоступной таблице у callers).
|
||||
monkeypatch.setattr(zr, "get_cached_zone_regulation", lambda *a, **kw: None)
|
||||
|
||||
shared = _SharedDB()
|
||||
client = _MissThenResolveClient("Ж-2")
|
||||
result = zr.get_or_fetch_zone_regulation(shared, 60.6, 56.8, client=client) # type: ignore[arg-type]
|
||||
|
||||
# Write-сессия откатилась (нечего персистить), не коммитила, закрыта.
|
||||
assert write_db.commits == 0
|
||||
assert write_db.rollbacks == 1
|
||||
assert write_db.closed is True
|
||||
# SHARED транзакция вообще не тронута.
|
||||
assert shared.commits == 0
|
||||
assert shared.rollbacks == 0
|
||||
# Функция отработала gracefully (read-back None при недоступном кэше).
|
||||
assert result is None
|
||||
Loading…
Add table
Reference in a new issue