diff --git a/tradein-mvp/backend/app/api/v1/trade_in.py b/tradein-mvp/backend/app/api/v1/trade_in.py index bc05717c..512785ab 100644 --- a/tradein-mvp/backend/app/api/v1/trade_in.py +++ b/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). diff --git a/tradein-mvp/backend/app/core/db.py b/tradein-mvp/backend/app/core/db.py index cd18b717..dd260dce 100644 --- a/tradein-mvp/backend/app/core/db.py +++ b/tradein-mvp/backend/app/core/db.py @@ -16,6 +16,25 @@ 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. + # + # pool_size оставлен дефолтным (5): это ПОСТОЯННО открытые коннекты, а в покое + # прод держит 5-6 (замер 11.09). Растёт только overflow — коннекты пика, + # которые пул закрывает сам. Потолок процесса: 5 + 15 = 20; воркер один + # (docker-compose.prod.yml, uvicorn без --workers), Postgres max_connections=100. + max_overflow=15, + # Дефолтные 30 с ожидания коннекта — вчетверо больше любого бюджета внешнего + # источника в эстиматоре (8 с, `estimate_*_timeout_s`). Такой чекаут нельзя + # прервать `asyncio.wait_for`: он занимает поток `asyncio.to_thread` целиком, а + # пул потоков сам конечен (min(32, cpu+4)) — исчерпанный пул коннектов так + # превращается в исчерпанный пул потоков. 5 с < бюджета источника: занятый пул + # деградирует ОДИН источник, а не весь запрос. + pool_timeout=5, ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, expire_on_commit=False) diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 12f0aa31..75bdfbc9 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -695,6 +695,43 @@ 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'ах, а в целых фетчах. + + ЧТО ЭТО МЕНЯЕТ ДЛЯ ВЫЗЫВАЮЩЕГО: незакоммиченная работа его транзакции + (например 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 + + return await asyncio.to_thread(_run) + + async def _get_or_fetch_imv_cached( db: Session, *, @@ -731,10 +768,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 +786,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 +848,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 +884,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 +939,10 @@ _DEFERRED_REFRESH_TASKS: set[asyncio.Task[None]] = set() # Подобран под текущий прод: 1 воркер uvicorn, mem_limit 768m, max_connections # 100 у Postgres. Догрузка почти всё время ждёт чужой HTTP, поэтому небольшой # потолок не занижает пропускную способность прогрева. +# +# Коннект к БД задача держит только на самих SELECT/UPSERT кэша, а не весь фетч +# (`_db_step`, #3408): до правки одна догрузка Яндекса удерживала соединение +# 8.5 с (замер на проде 11.09), то есть восемь таких задач выедали пул целиком. _MAX_DEFERRED_REFRESH_TASKS = 8 @@ -976,10 +1024,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 +1038,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 +1117,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,16 +1141,16 @@ 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], diff --git a/tradein-mvp/backend/tests/test_3408_estimator_db_off_loop.py b/tradein-mvp/backend/tests/test_3408_estimator_db_off_loop.py new file mode 100644 index 00000000..68db9a4a --- /dev/null +++ b/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)" + ) diff --git a/tradein-mvp/backend/tests/test_3408_pool_ceiling.py b/tradein-mvp/backend/tests/test_3408_pool_ceiling.py new file mode 100644 index 00000000..9cbde0ef --- /dev/null +++ b/tradein-mvp/backend/tests/test_3408_pool_ceiling.py @@ -0,0 +1,54 @@ +"""Пул коннектов не меньше суммы потолков одновременности процесса (#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 + +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 test_pool_ceiling_covers_declared_concurrency() -> None: + declared = _ESTIMATE_CONCURRENCY + _SUGGEST_CONCURRENCY + _MAX_DEFERRED_REFRESH_TASKS + ceiling = engine.pool.size() + engine.pool._max_overflow + + 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`, + которых тоже конечное число) вместо того, чтобы деградировать один источник. + """ + 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}с" + )