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..1214a3eb 100644 --- a/tradein-mvp/backend/app/core/db.py +++ b/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) diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 12f0aa31..b757e7cd 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/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 diff --git a/tradein-mvp/backend/tests/test_3408_db_step_cancel_orphan.py b/tradein-mvp/backend/tests/test_3408_db_step_cancel_orphan.py new file mode 100644 index 00000000..e58c449e --- /dev/null +++ b/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, "сирота не завершил транзакцию — коннект не вернулся в пул" 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..2c220834 --- /dev/null +++ b/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}с" + )