Merge pull request 'fix(mera/estimate): коннект БД не живёт через внешний HTTP; потолок пула ≥ суммы потолков одновременности (#3083, #3408)' (#3444) from fix/3083-estimate-throughput into main

tradein-mvp/backend/app/api/v1/trade_in.py

@@ -79,8 +79,10 @@ _estimate_limiter = SlidingWindowLimiter(
 # внешние тиры; пила одновременных оценок выедает пул и тормозит весь /api/v1/*.
 # Образец — public/mera.py::_suggest_slots (4 слота на секундное автодополнение).
 #
-# 4 слота: вместе с 4 слотами suggest — 8 одновременно удерживаемых соединений
-# из 15 возможных, остаток пула остаётся прочим ручкам. Ожидание слота 5с ≈ две
+# 4 слота: вместе с 4 слотами suggest — 8 одновременно удерживаемых соединений.
+# Пул под это заведомо шире: 5+15=20 на процесс, и потолок пула держится не
+# меньше СУММЫ объявленных потолков одновременности, включая 8 фоновых догрузок
+# (core/db.py + tests/test_3408_pool_ceiling.py, #3408). Ожидание слота 5с ≈ две
 # длительности оценки: если за это время слот не освободился, очередь глубока и
 # честный ответ — быстрый 429 с Retry-After, а не растущая очередь (очередь под
 # нагрузкой — те же занятые соединения плюс таймаут у клиента; mera.py:117-127).

tradein-mvp/backend/app/core/db.py

@@ -16,6 +16,41 @@ engine = create_engine(
     # НЕ закрывает: текст ошибки самого драйвера (Postgres DETAIL со значением)
     # и сырые psycopg-подключения мимо движков — это отдельный класс.
     hide_parameters=True,
+    # #3408 п.1. Потолок пула обязан быть НЕ МЕНЬШЕ суммы потолков одновременности,
+    # которые сам же процесс и объявляет: 4 оценки (`api/v1/trade_in`
+    # `_ESTIMATE_CONCURRENCY`) + 4 подсказки (`api/public/mera` `_SUGGEST_CONCURRENCY`)
+    # + 8 фоновых догрузок (`services/estimator` `_MAX_DEFERRED_REFRESH_TASKS`) = 16.
+    # Дефолт SQLAlchemy 5+10=15 меньше этой суммы, то есть исчерпать пул можно
+    # штатной работой, не абузом. Гейт — tests/test_3408_pool_ceiling.py.
+    #
+    # Инвариант «пул >= суммы объявленных потолков» — это ПОЛ, а не гарантия:
+    # сумма считает по одному коннекту на объявленную единицу работы, а коннект
+    # держит и всё остальное — любая ручка с `Depends(get_db)` тоже. Глобального
+    # обработчика `sqlalchemy.exc.TimeoutError` в app/main.py нет, поэтому на
+    # исчерпанном пуле соседние ручки отдают 500 (после ожидания `pool_timeout`),
+    # а не медленный 200.
+    #
+    # pool_size оставлен дефолтным (5): это ПОСТОЯННО открытые коннекты, а в покое
+    # прод держит 5-6 (замер 11.09). Растёт только overflow — коннекты пика,
+    # которые пул закрывает сам. Потолок процесса: 5 + 15 = 20; воркер один
+    # (docker-compose.prod.yml, uvicorn без --workers), Postgres max_connections=100.
+    max_overflow=15,
+    # #3408 п.2. Дефолтные 30 с ожидания коннекта длиннее ЛЮБОГО бюджета внешнего
+    # источника в эстиматоре: 8 с (Yandex / Cian / house_meta), 12 с (geocode),
+    # 20 с (Avito IMV — `estimate_avito_imv_timeout_s`, config.py:852). Сам чекаут
+    # прервать `asyncio.wait_for` не может: он занимает поток `asyncio.to_thread`
+    # целиком, а пул потоков конечен (min(32, cpu+4)) — исчерпанный пул коннектов
+    # так превращается в исчерпанный пул потоков. 5 с короче самого КОРОТКОГО
+    # бюджета: занятый пул деградирует ОДИН источник, а не весь запрос.
+    #
+    # Отдельный коммит в конце ветки намеренно (ревью PR #3444): это единственная
+    # правка, которая меняет режим отказа с «медленно» на «быстро с ошибкой», и
+    # едет она во ВСЕ сервисы образа — backend, scraper, tgbot
+    # (tradein-mvp/docker-compose.prod.yml). За 29 ч логов исчерпания пула не было
+    # ни разу, то есть новое значение на проде пока не на чем проверить.
+    # ТРИГГЕР ОТКАТА на 30 с: любое `QueuePool limit ... timed out` в логах
+    # бэкенда ЛИБО рост failed+zombie в `scrape_runs` после деплоя.
+    pool_timeout=5,
 )
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, expire_on_commit=False)
 

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

@@ -695,6 +695,68 @@ YANDEX_VALUATION_DEFAULT_CATEGORY = "APARTMENT"
 YANDEX_VALUATION_DEFAULT_TYPE = "SELL"
 
 
+async def _db_step[T](db: Session, work: Callable[[], T]) -> T:
+    """Синхронный шаг БД внутри async-функции: в потоке, с возвратом коннекта и без сирот.
+
+    Три вещи сразу, потому что все про одно и то же — `Session` из `get_db`
+    (#3408 п.1/п.2):
+
+    1. `db.execute()` — блокирующий вызов. На loop'е он держит ВЕСЬ инстанс
+       (один воркер uvicorn, #3083) на время чекаута коннекта: при занятом пуле
+       это `pool_timeout` секунд, и никакой `asyncio.wait_for` его не прервёт.
+       В потоке ждёт поток, а loop продолжает обслуживать остальных.
+    2. `commit()`/`rollback()` в конце — не про данные, а про КОННЕКТ: сессия
+       отдаёт его в пул только на завершении транзакции. Без этого коннект,
+       взятый ради одного SELECT'а кэша, живёт до конца функции — включая
+       ожидание чужого HTTP (замер 11.09: 8.5 с на догрузку Яндекса), и пиковый
+       спрос считается не в коротких SELECT'ах, а в целых фетчах.
+    3. Отмена шага ДОЖИДАЕТСЯ потока (`asyncio.shield` ниже). Поток отменить
+       нельзя, а сессия у шага не своя — общая со всем остальным запросом.
+
+    ЧТО ЭТО МЕНЯЕТ ДЛЯ ВЫЗЫВАЮЩЕГО: незакоммиченная работа его транзакции
+    (например fill-null backfill ФИАСа в `estimate_quality`) фиксируется здесь,
+    а не в первом commit'е ниже по коду. Новым классом поведения это не
+    является: обе функции-источника и так коммитят ЧУЖУЮ сессию сразу после
+    фетча (`save_imv_evaluation`, UPSERT в `external_valuations`) — сдвигается
+    момент, а не факт. Ошибка внутри `work` → `rollback` + проброс наверх, как
+    и раньше: гасят её существующие `except` вокруг вызова.
+    """
+
+    def _run() -> T:
+        try:
+            result = work()
+        except Exception:
+            db.rollback()
+            raise
+        db.commit()
+        return result
+
+    step = asyncio.ensure_future(asyncio.to_thread(_run))
+    try:
+        return await asyncio.shield(step)
+    except asyncio.CancelledError:
+        # Отмена по бюджету источника (`_with_budget` = `asyncio.wait_for`) поток НЕ
+        # останавливает: у `to_thread` отменяется только ожидание со стороны loop'а.
+        # Корутина умирает, поток продолжает работать с ТОЙ ЖЕ `Session`, а вызывающий
+        # тем временем идёт дальше — следующий источник, `_fetch_anchor_comps`,
+        # `_persist_estimate_and_commit` — ПО ТОЙ ЖЕ сессии. Два потока в одной сессии
+        # дают «another operation is in progress» / InvalidRequestError на следующем
+        # шаге БД: у источников такую ошибку глушит `except` вокруг вызова, у персиста
+        # оценки не глушит ничего — это 500 и потерянная оценка клиента, причём ровно
+        # под нагрузкой, ради которой правка и делается.
+        #
+        # Поэтому отмену пробрасываем ПОСЛЕ того, как поток отпустил сессию. Цена —
+        # бюджет источника переезжает на длину ОДНОГО шага БД (чекаут ≤ `pool_timeout`
+        # плюс сам запрос), а не на длину фетча, ради которой бюджет и заведён.
+        # Гейт — tests/test_3408_db_step_cancel_orphan.py.
+        await asyncio.wait([step])
+        if not step.cancelled() and step.exception() is not None:
+            # Результата уже никто не ждёт: без явного чтения asyncio напечатает
+            # «Task exception was never retrieved» вообще без контекста.
+            logger.warning("db-шаг упал уже после отмены по бюджету: %s", step.exception())
+        raise
+
+
 async def _get_or_fetch_imv_cached(
     db: Session,
     *,
@@ -731,10 +793,12 @@ async def _get_or_fetch_imv_cached(
             has_loggia,
         )
 
-        existing = (
-            db.execute(
-                text(
-                    """
+        existing = await _db_step(
+            db,
+            lambda: (
+                db.execute(
+                    text(
+                        """
                 SELECT id, cache_key, address, rooms, area_m2, floor, floor_at_home,
                        house_type, renovation_type, has_balcony, has_loggia,
                        lat, lon, geo_hash, avito_address_id, avito_location_id,
@@ -747,11 +811,12 @@ async def _get_or_fetch_imv_cached(
                 ORDER BY fetched_at DESC
                 LIMIT 1
                 """
-                ),
-                {"ck": cache_key, "ttl_hours": IMV_CACHE_TTL_HOURS},
-            )
-            .mappings()
-            .first()
+                    ),
+                    {"ck": cache_key, "ttl_hours": IMV_CACHE_TTL_HOURS},
+                )
+                .mappings()
+                .first()
+            ),
         )
 
         if existing is not None:
@@ -808,7 +873,9 @@ async def _get_or_fetch_imv_cached(
             # release в finally на всех выходах (исключение/таймаут — тоже).
             proxy_provider=RealProxyProvider(),
         )
-        save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
+        await _db_step(
+            db, lambda: save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
+        )
         logger.info(
             "imv: fresh recommended=%d range=(%d, %d) count=%d",
             result.recommended_price,
@@ -842,7 +909,9 @@ async def _get_or_fetch_imv_cached(
                     config=RealScraperConfig(),
                     proxy_provider=RealProxyProvider(),  # #3386, см. первый вызов выше
                 )
-                save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
+                await _db_step(
+                    db, lambda: save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
+                )
                 logger.info(
                     "imv: retry OK recommended=%d range=(%d, %d) count=%d",
                     result.recommended_price,
@@ -895,6 +964,19 @@ _DEFERRED_REFRESH_TASKS: set[asyncio.Task[None]] = set()
 # Подобран под текущий прод: 1 воркер uvicorn, mem_limit 768m, max_connections
 # 100 у Postgres. Догрузка почти всё время ждёт чужой HTTP, поэтому небольшой
 # потолок не занижает пропускную способность прогрева.
+#
+# Фоновых источника ДВА, и по удержанию коннекта они РАЗНЫЕ (#3408):
+#   • yandex_valuation — коннект только на самих SELECT/UPSERT кэша (`_db_step`):
+#     до правки одна догрузка держала соединение 8.5 с (замер на проде 11.09),
+#     то есть восемь таких задач выедали пул целиком;
+#   • cian_valuation — коннект на ВЕСЬ фетч (HTTP-таймаут 25 с): транзакцию
+#     открывают `_load_from_cache` / `load_session` на входе, а закрывает её
+#     `db.commit()` в самом конце (scraper_kit providers/cian/valuation.py:163,
+#     176, 595). Известный остаток: функция общая со скраппером
+#     (`app/tasks/cian_history_backfill.py:458`), где коммит в середине менял бы
+#     семантику батча, — нужен отдельный опт-ин путь, а не правка на месте.
+# На ПОТОЛОК пула остаток не влияет (задача держит один коннект независимо от
+# того, как долго) — только на среднюю занятость пула.
 _MAX_DEFERRED_REFRESH_TASKS = 8
 
 
@@ -976,10 +1058,12 @@ async def _get_or_fetch_yandex_valuation_cached(
 
     # Cache lookup
     try:
-        cached = (
-            db.execute(
-                text(
-                    """
+        cached = await _db_step(
+            db,
+            lambda: (
+                db.execute(
+                    text(
+                        """
                 SELECT raw_payload, fetched_at
                 FROM external_valuations
                 WHERE source = 'yandex_valuation'
@@ -988,11 +1072,12 @@ async def _get_or_fetch_yandex_valuation_cached(
                 ORDER BY fetched_at DESC
                 LIMIT 1
                 """
-                ),
-                {"ck": cache_key},
-            )
-            .mappings()
-            .first()
+                    ),
+                    {"ck": cache_key},
+                )
+                .mappings()
+                .first()
+            ),
         )
     except Exception as e:
         logger.warning("yandex_valuation: cache lookup failed: %s", e)
@@ -1066,9 +1151,11 @@ async def _get_or_fetch_yandex_valuation_cached(
 
     # Save to cache (UPSERT on (source, cache_key))
     try:
-        db.execute(
-            text(
-                """
+        await _db_step(
+            db,
+            lambda: db.execute(
+                text(
+                    """
                 INSERT INTO external_valuations (
                     source, cache_key, address,
                     house_id,
@@ -1088,24 +1175,25 @@ async def _get_or_fetch_yandex_valuation_cached(
                         fetched_at = NOW(),
                         expires_at = NOW() + (:ttl_hours || ' hours')::interval
                 """
+                ),
+                {
+                    "ck": cache_key,
+                    "addr": address,
+                    "hid": house_id,
+                    "payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False),
+                    "ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS,
+                },
             ),
-            {
-                "ck": cache_key,
-                "addr": address,
-                "hid": house_id,
-                "payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False),
-                "ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS,
-            },
         )
-        db.commit()
         logger.info(
             "yandex_valuation: fresh fetch saved key=%s items=%d",
             cache_key[:8],
             len(result.history_items),
         )
     except Exception as e:
+        # Откат уже сделан в потоке (`_db_step` → `except` → `db.rollback()`), а здесь
+        # это был бы блокирующий вызов на event loop'е — ровно то, что правка убирает.
         logger.warning("yandex_valuation: cache save failed (continuing): %s", e)
-        db.rollback()
 
     return result
 

tradein-mvp/backend/tests/test_3408_db_step_cancel_orphan.py

@@ -0,0 +1,82 @@
+"""Отмена шага БД по бюджету не оставляет ОСИРОТЕВШИЙ поток в чужой сессии (#3408).
+
+`_with_budget` — это `asyncio.wait_for`, а `asyncio.to_thread` отменить нельзя:
+ожидание со стороны loop'а снимается, поток продолжает работать. Сессия у шага не
+своя — та же самая, с которой запрос идёт дальше: следующий источник,
+`_fetch_anchor_comps`, `_persist_estimate_and_commit`. Два потока в одной `Session`
+дают «another operation is in progress» / InvalidRequestError на СЛЕДУЮЩЕМ шаге,
+и у персиста оценки этой ошибки не ловит никто — 500 и потерянная оценка клиента.
+
+Меряем значение, а не форму: «следующий шаг не вошёл в сессию, пока сирота не
+закончил». Следующий шаг здесь — ГОЛЫЙ `asyncio.to_thread(db...)`, как
+`_persist_estimate_and_commit` (estimator.py), а не ещё один `_db_step`: защита,
+которая живёт только внутри `_db_step`, ровно того пострадавшего и не закрывает.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import threading
+import time
+from typing import Any
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import estimator as est
+
+# Шаг БД заметно длиннее бюджета — окно, в котором сирота ещё работает.
+_STEP_S = 0.3
+_BUDGET_S = 0.05
+
+
+class _ConcurrencyProbeSession:
+    """Session-дублёр, который считает ОДНОВРЕМЕННЫЕ входы в сессию."""
+
+    def __init__(self) -> None:
+        self._guard = threading.Lock()
+        self._inside = 0
+        self.conflicts = 0
+        self.executed = 0
+        self.commits = 0
+
+    def execute(self, *_a: Any, **_kw: Any) -> Any:
+        with self._guard:
+            self._inside += 1
+            self.executed += 1
+            if self._inside > 1:
+                self.conflicts += 1
+        try:
+            time.sleep(_STEP_S)
+        finally:
+            with self._guard:
+                self._inside -= 1
+        return self
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def rollback(self) -> None:  # pragma: no cover — на этом пути не звучит
+        pass
+
+
+async def test_budget_cancel_does_not_leave_orphan_in_session() -> None:
+    db = _ConcurrencyProbeSession()
+
+    degraded = await est._with_budget(
+        est._db_step(db, lambda: db.execute("cache lookup")),  # type: ignore[arg-type]
+        _BUDGET_S,
+        label="проба",
+    )
+    # Бюджет истёк — источник деградировал в None, вызывающий идёт дальше.
+    assert degraded is None, "бюджет не сработал — тест ничего не проверил"
+
+    # Следующий шаг ТОГО ЖЕ запроса по ТОЙ ЖЕ сессии (образец — persist оценки).
+    await asyncio.to_thread(db.execute, "persist estimate")
+
+    assert db.executed == 2, f"звучали не оба шага (executed={db.executed})"
+    assert db.conflicts == 0, (
+        "следующий шаг вошёл в сессию, пока осиротевший поток ещё работал в ней: "
+        "два потока в одной Session → «another operation is in progress» на персисте"
+    )
+    assert db.commits >= 1, "сирота не завершил транзакцию — коннект не вернулся в пул"

tradein-mvp/backend/tests/test_3408_estimator_db_off_loop.py

@@ -0,0 +1,213 @@
+"""Синхронная БД внутри async-источников эстиматора: не на loop'е и не через фетч (#3408).
+
+`/estimate` публичен (meraocenka.ru) и крутится ОДНИМ воркером uvicorn (#3083), поэтому
+у `db.execute()` прямо на event loop'е цена не «медленнее на миллисекунды», а «весь
+инстанс не отвечает»: чекаут коннекта из занятого пула ждёт `pool_timeout`, и
+`asyncio.wait_for` (`_with_budget`) синхронный вызов прервать не может.
+
+Что меряют тесты (значение, а не форму вызова):
+  (а) пока `db.execute` «ходит в БД» 0.3 с, соседняя корутина тикает — на origin/main
+      тиков ноль, loop стоит целиком (обе функции-источника: Яндекс и IMV);
+  (б) на время внешнего HTTP коннект ОТДАН пулу: пул из ОДНОГО коннекта, и второй
+      чекаут во время фетча обязан состояться. На origin/main транзакция, открытая
+      SELECT'ом кэша, живёт весь фетч (замер на проде 11.09: 8.5 с на одну догрузку
+      Яндекса при потолке 8 одновременных) — второй чекаут падает по таймауту.
+
+NB к (б): на sqlite SELECT кэша падает (Postgres-синтаксис `NOW()`), и это ровно тот
+путь, что и на успехе: коннект удерживается транзакцией до `commit`/`rollback`
+независимо от того, вернул ли SELECT строки. Проверяется «коннект свободен во время
+await», а не «SELECT отработал».
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+from typing import Any
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import TimeoutError as SATimeoutError
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import QueuePool
+
+from app.services import estimator as est
+
+# Столько «занимает поход в БД». 0.3 с — заметно больше планировочного шума и
+# заметно меньше любого таймаута теста.
+_DB_SLEEP_S = 0.3
+
+
+class _SlowSession:
+    """Session-дублёр: `execute` блокирует ПОТОК, как настоящий чекаут+SELECT."""
+
+    def __init__(self) -> None:
+        self.executed = 0
+        self.commits = 0
+        self.rollbacks = 0
+
+    def execute(self, *_a: Any, **_kw: Any) -> Any:
+        self.executed += 1
+        time.sleep(_DB_SLEEP_S)
+        return self
+
+    def mappings(self) -> Any:
+        return self
+
+    def first(self) -> None:
+        return None
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def rollback(self) -> None:
+        self.rollbacks += 1
+
+
+async def _ticks_during(call: Any) -> tuple[int, Any]:
+    """Сколько раз соседняя корутина успела проснуться, пока шёл `call`."""
+    ticks = 0
+
+    async def _ticker() -> None:
+        nonlocal ticks
+        while True:
+            await asyncio.sleep(0)
+            ticks += 1
+
+    task = asyncio.create_task(_ticker())
+    await asyncio.sleep(0)  # дать тикеру стартовать
+    before = ticks
+    result = await call
+    during = ticks - before
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+    return during, result
+
+
+# ── (а) sync-БД не держит event loop ────────────────────────────────────────
+
+
+async def test_yandex_cache_lookup_does_not_block_event_loop() -> None:
+    db = _SlowSession()
+
+    during, result = await _ticks_during(
+        est._get_or_fetch_yandex_valuation_cached(
+            db,  # type: ignore[arg-type]
+            address="Екатеринбург, улица Малышева, 84",
+            fetch_on_miss=False,  # после промаха функция возвращает None сразу
+        )
+    )
+
+    assert db.executed == 1, "SELECT кэша не звучал — тест ничего не проверил"
+    assert result is None
+    # Свободный loop успевает десятки тысяч тиков за 0.3 с; заблокированный — ноль.
+    assert during > 100, f"loop простоял весь SELECT кэша: тиков всего {during}"
+
+
+async def test_imv_cache_lookup_does_not_block_event_loop(monkeypatch: Any) -> None:
+    db = _SlowSession()
+
+    async def _no_fetch(**_kw: Any) -> None:
+        raise est.IMVTransientError("фетч в этом тесте не участвует")
+
+    monkeypatch.setattr(est, "evaluate_via_imv", _no_fetch)
+
+    during, result = await _ticks_during(
+        est._get_or_fetch_imv_cached(
+            db,  # type: ignore[arg-type]
+            address="Екатеринбург, улица Малышева, 84",
+            rooms=2,
+            area_m2=54.0,
+            floor=3,
+            floor_at_home=9,
+            house_type="panel",
+            renovation_type="cosmetic",
+            has_balcony=True,
+            has_loggia=False,
+        )
+    )
+
+    assert db.executed == 1, "SELECT кэша не звучал — тест ничего не проверил"
+    assert result is None
+    assert during > 100, f"loop простоял весь SELECT кэша: тиков всего {during}"
+
+
+# ── (б) коннект не удерживается через внешний HTTP ──────────────────────────
+
+
+class _FetchProbeScraper:
+    """Дублёр YandexValuationScraper: во время «фетча» пробует взять второй коннект."""
+
+    second_checkout_ok: bool | None = None
+
+    def __init__(self, engine: Any) -> None:
+        self._engine = engine
+
+    def __call__(self, *_a: Any, **_kw: Any) -> _FetchProbeScraper:
+        return self
+
+    async def __aenter__(self) -> _FetchProbeScraper:
+        return self
+
+    async def __aexit__(self, *_a: Any) -> None:
+        return None
+
+    async def fetch_house_history(self, **_kw: Any) -> None:
+        await asyncio.sleep(0)
+        try:
+            with self._engine.connect() as conn:
+                conn.execute(text("SELECT 1"))
+            type(self).second_checkout_ok = True
+        except SATimeoutError:
+            type(self).second_checkout_ok = False
+        return None  # «дом не найден» — функция выходит до UPSERT
+
+
+@pytest.fixture
+def one_connection_engine() -> Any:
+    """Пул ровно из одного коннекта: занятый коннект видно по таймауту чекаута."""
+    engine = create_engine(
+        "sqlite://",
+        # sqlite по умолчанию берёт SingletonThreadPool — у него нет ни
+        # max_overflow, ни таймаута; нам нужен ровно тот пул, что на проде.
+        poolclass=QueuePool,
+        pool_size=1,
+        max_overflow=0,
+        pool_timeout=0.25,
+        connect_args={"check_same_thread": False},
+    )
+    try:
+        yield engine
+    finally:
+        engine.dispose()
+
+
+async def test_yandex_fetch_does_not_hold_pool_connection(
+    monkeypatch: Any, one_connection_engine: Any
+) -> None:
+    probe = _FetchProbeScraper(one_connection_engine)
+    _FetchProbeScraper.second_checkout_ok = None
+    monkeypatch.setattr(est, "YandexValuationScraper", probe)
+
+    db = sessionmaker(bind=one_connection_engine, expire_on_commit=False)()
+    try:
+        result = await est._get_or_fetch_yandex_valuation_cached(
+            db, address="Екатеринбург, улица Малышева, 84"
+        )
+    finally:
+        db.close()
+
+    assert result is None
+    assert _FetchProbeScraper.second_checkout_ok is not None, (
+        "фетч не звучал — тест ничего не проверил"
+    )
+    assert _FetchProbeScraper.second_checkout_ok is True, (
+        "коннект удерживается транзакцией всё время внешнего HTTP — "
+        "пик спроса на пул считается фетчами, а не SELECT'ами (#3408 п.1)"
+    )

tradein-mvp/backend/tests/test_3408_pool_ceiling.py

@@ -0,0 +1,71 @@
+"""Пул коннектов не меньше суммы потолков одновременности процесса (#3408 п.1).
+
+Процесс сам объявляет, сколько одновременной работы он допускает: 4 оценки, 4
+подсказки, 8 фоновых догрузок. Каждая из этих единиц работы держит СВОЮ сессию.
+Если сумма больше пула, исчерпать пул можно штатной нагрузкой — и упереться не в
+ресурс, а в `QueuePool limit ... timed out`, причём на публичной ручке.
+
+На origin/main сумма 16 против дефолта SQLAlchemy 5+10=15 — тест красный.
+
+Гейт нужен не ради текущих чисел, а ради будущих: поднять `_ESTIMATE_CONCURRENCY`
+или `_MAX_DEFERRED_REFRESH_TASKS`, не поднимая пула, станет видно здесь.
+Добавится воркер uvicorn (#3083) — числа per-process не меняются, но суммарный
+потолок коннектов умножается на число воркеров; это отдельное решение, не это.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.api.public.mera import _SUGGEST_CONCURRENCY
+from app.api.v1.trade_in import _ESTIMATE_CONCURRENCY
+from app.core.db import engine
+from app.services.estimator import _MAX_DEFERRED_REFRESH_TASKS
+
+
+def _max_overflow(pool: Any) -> int:
+    """`max_overflow` у QueuePool публичного геттера не имеет (в отличие от
+    `size()` / `timeout()`), поэтому читаем приватное поле — но с явным падением:
+    молча переименуется в новой SQLAlchemy → гейт бы просто перестал что-либо
+    проверять (AttributeError упал бы ошибкой теста, а вот подмена значения на
+    дефолт — нет).
+    """
+    assert hasattr(pool, "_max_overflow"), (
+        f"{type(pool).__name__} больше не хранит `_max_overflow` — SQLAlchemy сменила "
+        "API пула, потолок коннектов читать больше нечем: почини гейт, а не удаляй его"
+    )
+    return int(pool._max_overflow)
+
+
+def test_pool_ceiling_covers_declared_concurrency() -> None:
+    declared = _ESTIMATE_CONCURRENCY + _SUGGEST_CONCURRENCY + _MAX_DEFERRED_REFRESH_TASKS
+    ceiling = engine.pool.size() + _max_overflow(engine.pool)
+
+    assert ceiling >= declared, (
+        f"пул {ceiling} меньше суммы потолков одновременности {declared} "
+        f"({_ESTIMATE_CONCURRENCY} оценок + {_SUGGEST_CONCURRENCY} подсказок + "
+        f"{_MAX_DEFERRED_REFRESH_TASKS} фоновых догрузок) — штатная работа исчерпает пул"
+    )
+
+
+def test_pool_checkout_wait_shorter_than_source_budget() -> None:
+    """Ожидание коннекта короче бюджета внешнего источника.
+
+    Иначе занятый пул съедает весь бюджет запроса (и поток `asyncio.to_thread`,
+    которых тоже конечное число) вместо того, чтобы деградировать один источник.
+    Сравниваем с самым КОРОТКИМ бюджетом (8 с Yandex/Cian): geocode 12 с и IMV
+    20 с длиннее, их этот же потолок покрывает с запасом.
+    """
+    from app.core.config import settings
+
+    budget = min(
+        settings.estimate_yandex_valuation_timeout_s,
+        settings.estimate_cian_valuation_timeout_s,
+    )
+
+    assert engine.pool.timeout() < budget, (
+        f"pool_timeout {engine.pool.timeout()}с >= бюджета источника {budget}с"
+    )