fix(tradein/estimator): savepoint+retry вокруг ratio-lookup — transient не гасит expected_sold (#2265)

tradein-mvp/backend/app/services/estimator.py

@@ -263,46 +263,63 @@ def _get_asking_sold_ratio(
 
     ratio: float | None = None
     basis: str | None = None
-    try:
+    # #2265 D2: SAVEPOINT + 1 retry вокруг ratio-lookup. Транзиентный сбой —
-        row = db.execute(
+    # обычно poisoned tx (InFailedSqlTransaction) от вышестоящего graceful-except:
-            text(
+    # первый же SELECT падает, и раньше это гасило expected_sold для КОНКРЕТНОЙ
-                """
+    # оценки (3/224 за 14 дней при живых ratio-бакетах). Оборачиваем в begin_nested:
-                SELECT ratio, basis FROM asking_to_sold_ratios
+    # при сбое savepoint откатывается, tx чистится (db.rollback), retry идёт на
-                WHERE rooms_bucket = CAST(:b AS int) AND district = ''
+    # здоровом соединении. begin_nested также локализует benign-сбой (таблицы нет
-                """
+    # на свежей/старой БД, миграция 080 не применена) — без отравления outer tx.
-            ),
+    last_exc: Exception | None = None
-            {"b": bucket},
+    ok = False
-        ).fetchone()
+    for attempt in range(2):
-        if row is None:
-            # Бакет тонкий (n<30 при seed'е) или отсутствует → global fallback (-1).
-            row = db.execute(
-                text(
-                    """
-                    SELECT ratio, basis FROM asking_to_sold_ratios
-                    WHERE rooms_bucket = -1 AND district = ''
-                    """
-                ),
-            ).fetchone()
-        if row is not None and row.ratio is not None:
-            ratio = float(row.ratio)
-            basis = row.basis
-    except Exception as exc:
-        # Таблицы может не быть на свежей/старой БД (миграция 080 не применена),
-        # либо транзакция в сбойном состоянии — graceful: без sold-коррекции.
-        # ОБЯЗАТЕЛЬНО rollback: неудачный SELECT помечает транзакцию
-        # InFailedSqlTransaction, и без отката следующий statement упал бы → 500.
-        #
-        # НЕ кэшируем этот None: ошибка транзиентна (poisoned tx от вышестоящего
-        # graceful-except, миг. лаг, коннект-хиккап). Раньше строка кэша ниже
-        # писалась безусловно → один сбой отравлял _asking_sold_ratio_cache[bucket]
-        # = (None, None) на весь TTL (300с) и молча гасил expected_sold («ожидаемая
-        # цена сделки» → «—») для ВСЕХ оценок этого rooms-бакета на воркере до
-        # истечения TTL. Ранний return без записи в кэш → следующая оценка ретраит.
-        logger.debug("asking_to_sold_ratio lookup skipped (graceful, NOT cached): %s", exc)
         try:
-            db.rollback()
+            with db.begin_nested():
-        except Exception:
+                row = db.execute(
-            pass
+                    text(
+                        """
+                        SELECT ratio, basis FROM asking_to_sold_ratios
+                        WHERE rooms_bucket = CAST(:b AS int) AND district = ''
+                        """
+                    ),
+                    {"b": bucket},
+                ).fetchone()
+                if row is None:
+                    # Бакет тонкий (n<30 при seed'е) или отсутствует → global (-1).
+                    row = db.execute(
+                        text(
+                            """
+                            SELECT ratio, basis FROM asking_to_sold_ratios
+                            WHERE rooms_bucket = -1 AND district = ''
+                            """
+                        ),
+                    ).fetchone()
+                if row is not None and row.ratio is not None:
+                    ratio = float(row.ratio)
+                    basis = row.basis
+            ok = True
+            break
+        except Exception as exc:
+            last_exc = exc
+            # begin_nested откатывает СВОЙ savepoint; но если outer tx уже была
+            # poisoned до savepoint — сам SAVEPOINT не встаёт → чистим всю tx,
+            # чтобы retry шёл на здоровом соединении.
+            try:
+                db.rollback()
+            except Exception:
+                pass
+            if attempt == 0:
+                logger.debug("asking_to_sold_ratio lookup failed, retrying on clean tx: %s", exc)
+                continue
+
+    if not ok:
+        # Оба захода упали — graceful без sold-коррекции. НЕ кэшируем этот None:
+        # ошибка транзиентна (poisoned tx, миг. лаг, коннект-хиккап). Раньше строка
+        # кэша ниже писалась безусловно → один сбой отравлял
+        # _asking_sold_ratio_cache[bucket] = (None, None) на весь TTL (300с) и молча
+        # гасил expected_sold для ВСЕХ оценок бакета до истечения TTL (#2175). Ранний
+        # return без записи в кэш → следующая оценка ретраит.
+        logger.debug("asking_to_sold_ratio lookup skipped (graceful, NOT cached): %s", last_exc)
         return None, None
 
     # Кэшируем ТОЛЬКО успешный lookup. ratio может быть None (строки нет —

tradein-mvp/backend/tests/test_asking_sold_ratio_cache_poison.py

@@ -12,6 +12,8 @@ No DB: fake-Session (monkeypatched .execute/.rollback). DATABASE_URL нужен
 """
 
 import os
+from collections.abc import Iterator
+from contextlib import contextmanager
 from types import SimpleNamespace
 from typing import Any
 
@@ -37,6 +39,11 @@ class _OkDB:
         self._row = SimpleNamespace(ratio=ratio, basis=basis)
         self.rolled_back = False
 
+    @contextmanager
+    def begin_nested(self) -> Iterator[Any]:
+        # #2265 D2: ratio-lookup обёрнут в SAVEPOINT — моделируем CM.
+        yield SimpleNamespace()
+
     def execute(self, *_a: Any, **_k: Any) -> _Result:
         return _Result(self._row)
 
@@ -50,6 +57,10 @@ class _BoomDB:
     def __init__(self) -> None:
         self.rolled_back = False
 
+    @contextmanager
+    def begin_nested(self) -> Iterator[Any]:
+        yield SimpleNamespace()
+
     def execute(self, *_a: Any, **_k: Any) -> _Result:
         raise RuntimeError("InFailedSqlTransaction")
 

tradein-mvp/backend/tests/test_estimator_ratio_savepoint_retry_2265.py

@@ -0,0 +1,103 @@
+"""Unit tests for SAVEPOINT + retry вокруг ratio-lookup (#2265 D2).
+
+Транзиентный сбой первого ratio-SELECT (обычно poisoned tx,
+InFailedSqlTransaction, от вышестоящего graceful-except) раньше гасил
+expected_sold для КОНКРЕТНОЙ оценки (3/224 за 14 дней при живых бакетах).
+Фикс: begin_nested + 1 retry на чистом savepoint. Проверяем:
+  - первый lookup падает → второй проходит → возвращается ratio (recover);
+  - оба захода падают → (None, None) и кэш НЕ отравлен (#2175 держится);
+  - happy-path — ровно один begin_nested, без rollback.
+
+No DB / network: session — лёгкий fake с side-effect'ами на execute.
+DATABASE_URL нужен только для импорта app.core.config.Settings.
+"""
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from contextlib import contextmanager
+from typing import Any
+from unittest.mock import MagicMock
+
+from app.services import estimator
+
+
+class _Row:
+    """Строка результата с полями .ratio/.basis (как row из fetchone())."""
+
+    def __init__(self, ratio: float | None, basis: str | None) -> None:
+        self.ratio = ratio
+        self.basis = basis
+
+
+class _FakeSession:
+    """Session-заглушка: execute отдаёт side-effect'ы по очереди.
+
+    Exception в списке → execute бросает; иначе — result, чей fetchone() вернёт
+    заготовленную строку. begin_nested — CM, rollback — счётчик.
+    """
+
+    def __init__(self, execute_effects: list[Any]) -> None:
+        self._effects = list(execute_effects)
+        self.rollback_count = 0
+        self.begin_nested_count = 0
+
+    @contextmanager
+    def begin_nested(self):  # type: ignore[no-untyped-def]
+        self.begin_nested_count += 1
+        yield MagicMock()
+
+    def execute(self, *_args: Any, **_kwargs: Any) -> Any:
+        eff = self._effects.pop(0)
+        if isinstance(eff, Exception):
+            raise eff
+        result = MagicMock()
+        result.fetchone.return_value = eff
+        return result
+
+    def rollback(self) -> None:
+        self.rollback_count += 1
+
+
+def test_ratio_retry_recovers_after_first_lookup_fails() -> None:
+    estimator._asking_sold_ratio_cache.clear()
+    poisoned = RuntimeError("current transaction is aborted (InFailedSqlTransaction)")
+    db = _FakeSession([poisoned, _Row(0.84, "per_rooms")])
+
+    ratio, basis = estimator._get_asking_sold_ratio(db, rooms=1)  # type: ignore[arg-type]
+
+    assert ratio == 0.84
+    assert basis == "per_rooms"
+    assert db.begin_nested_count == 2  # первый заход упал, второй прошёл
+    assert db.rollback_count >= 1  # чистка tx перед retry
+    assert 1 in estimator._asking_sold_ratio_cache  # успех кэшируется
+
+
+def test_ratio_both_attempts_fail_returns_none_and_not_cached() -> None:
+    estimator._asking_sold_ratio_cache.clear()
+    db = _FakeSession([RuntimeError("boom-1"), RuntimeError("boom-2")])
+
+    ratio, basis = estimator._get_asking_sold_ratio(db, rooms=2)  # type: ignore[arg-type]
+
+    assert ratio is None
+    assert basis is None
+    assert 2 not in estimator._asking_sold_ratio_cache  # транзиент НЕ отравляет кэш (#2175)
+    assert db.begin_nested_count == 2
+    assert db.rollback_count >= 2
+
+
+def test_ratio_happy_path_single_attempt_no_rollback() -> None:
+    estimator._asking_sold_ratio_cache.clear()
+    db = _FakeSession([_Row(0.9, "global_fallback")])
+
+    ratio, basis = estimator._get_asking_sold_ratio(db, rooms=0)  # type: ignore[arg-type]
+
+    assert ratio == 0.9
+    assert basis == "global_fallback"
+    assert db.begin_nested_count == 1  # ни одного retry
+    assert db.rollback_count == 0
+    cached = estimator._asking_sold_ratio_cache.get(0)
+    assert cached is not None
+    assert cached[0] == 0.9
+    assert cached[1] == "global_fallback"