perf(tradein): снять блокировку event loop на POST /estimate (#2207)

async estimate_quality выпускал ~28 синхронных psycopg PostGIS-запросов
прямо на event loop; прод — один uvicorn worker → одна медленная оценка
(холодный кэш, wide-radius fallback) стопила ВСЕ конкурентные запросы,
включая /health (флап healthcheck).

- 15 sync DB-кластеров обёрнуты в await asyncio.to_thread (паттерн
  geocoder.py), строго последовательно — Session не используется конкурентно
- SAVEPOINT-блок (_backfill_house_fias) целиком внутри одного потока,
  транзакционные точки не сдвинуты; INSERT+IMV-UPDATE+commit сгруппированы
  атомарно в один to_thread
- sync-сигнатуры внутренностей (_price_from_inputs, _fetch_analogs, ...)
  не тронуты — backtest-harness и frozen regression gate зелёные
- тест неблокирования: sleep(0.3) в _fetch_analogs, конкурентная корутина
  завершается за <0.25s

Refs #2207

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

@@ -2651,7 +2651,9 @@ async def estimate_quality(
     if geo is None:
         # Без координат не можем искать через PostGIS. Возвращаем low confidence.
         logger.warning("geocode failed for %s — returning low-confidence estimate", payload.address)
-        return _empty_estimate(payload, db, reason="address_not_geocoded", created_by=created_by)
+        return await asyncio.to_thread(
+            _empty_estimate, payload, db, reason="address_not_geocoded", created_by=created_by
+        )
 
     # 1b. DaData enrichment (PR Q1) — on-demand cleanup для target адреса.
     # Best-effort: graceful None при отсутствии credentials / quota / fail.
@@ -2674,7 +2676,8 @@ async def estimate_quality(
     dadata_fias = dadata.house_fias_id if dadata else None
     match_fias = payload.target_fias_id or dadata_fias
     try:
-        match = match_house_readonly(
+        match = await asyncio.to_thread(
+            match_house_readonly,
             db,
             address=(dadata.canonical_address if dadata else None) or geo.full_address,
             lat=geo.lat,
@@ -2693,7 +2696,8 @@ async def estimate_quality(
             # Guard `house_fias_id IS NULL` — никогда не перетираем curated id.
             # SAVEPOINT изолирует сбой UPDATE от внешней estimate-транзакции.
             if dadata_fias and dadata is not None:
-                try:
+
+                def _backfill_house_fias() -> None:
                     with db.begin_nested():
                         db.execute(
                             text(
@@ -2711,6 +2715,9 @@ async def estimate_quality(
                                 "hid": target_house_id,
                             },
                         )
+
+                try:
+                    await asyncio.to_thread(_backfill_house_fias)
                 except Exception as exc:  # pragma: no cover — defensive
                     logger.warning("estimate fias back-fill failed (graceful): %s", exc)
     except Exception as exc:  # pragma: no cover — defensive
@@ -2751,7 +2758,8 @@ async def estimate_quality(
 
     if cohort_range is not None:
         cy_min, cy_max = cohort_range
-        listings_tier0, _, analog_tier = _fetch_analogs(
+        listings_tier0, _, analog_tier = await asyncio.to_thread(
+            _fetch_analogs,
             db,
             lat=geo.lat,
             lon=geo.lon,
@@ -2775,7 +2783,8 @@ async def estimate_quality(
         fallback_used = False
     else:
         # Tier 0 пуст/мал — graceful fallback на Tier A без cohort
-        listings, fallback_used, analog_tier = _fetch_analogs(
+        listings, fallback_used, analog_tier = await asyncio.to_thread(
+            _fetch_analogs,
             db,
             lat=geo.lat,
             lon=geo.lon,
@@ -2791,7 +2800,8 @@ async def estimate_quality(
     area_widened = False
 
     if len(listings) < 5:
-        listings_wide, _, analog_tier_wide = _fetch_analogs(
+        listings_wide, _, analog_tier_wide = await asyncio.to_thread(
+            _fetch_analogs,
             db,
             lat=geo.lat,
             lon=geo.lon,
@@ -2812,7 +2822,8 @@ async def estimate_quality(
     # Tier C: если даже на 2км мало — расширяем area tolerance до ±25%
     # (актуально для отдалённых районов / новостроек с нестандартной планировкой)
     if len(listings) < 3:
-        listings_widearea, _, analog_tier_wa = _fetch_analogs(
+        listings_widearea, _, analog_tier_wa = await asyncio.to_thread(
+            _fetch_analogs,
             db,
             lat=geo.lat,
             lon=geo.lon,
@@ -2835,7 +2846,8 @@ async def estimate_quality(
     # ── PRE-FETCH: dkp_raw (hoisted before _price_from_inputs) ──────────────
     # #1795: ДКП-коридор фетчим ДО вызова _price_from_inputs, чтобы
     # corridor_high был доступен для Tier C-гейта и soft-клампа headline.
-    dkp_raw = _fetch_dkp_corridor(
+    dkp_raw = await asyncio.to_thread(
+        _fetch_dkp_corridor,
         db,
         address=geo.full_address,
         rooms=payload.rooms,
@@ -2878,7 +2890,7 @@ async def estimate_quality(
             label="yandex_valuation",
         )
         if yandex_val is not None:
-            saved_hist = _save_yandex_history_items(db, yandex_val)
+            saved_hist = await asyncio.to_thread(_save_yandex_history_items, db, yandex_val)
             logger.info(
                 "yandex_valuation: history items processed=%d saved=%d"
                 " (house_id=NULL — matching deferred)",
@@ -2931,7 +2943,8 @@ async def estimate_quality(
         and payload.area_m2
         and (payload.address or (geo is not None and geo.full_address))
     ):
-        _anchor_comps_pre, _anchor_tier_pre = _fetch_anchor_comps(
+        _anchor_comps_pre, _anchor_tier_pre = await asyncio.to_thread(
+            _fetch_anchor_comps,
             db,
             address=payload.address or geo.full_address,
             target_house_id=target_house_id,
@@ -2944,7 +2957,8 @@ async def estimate_quality(
         _anchor_comps_pre, _anchor_tier_pre = [], None
 
     # ── Pre-fetch: house IMV anchor ONCE (used for both blend and display) ───
-    imv_anchor_data = _fetch_house_imv_anchor(
+    imv_anchor_data = await asyncio.to_thread(
+        _fetch_house_imv_anchor,
         db,
         target_house_id=target_house_id,
         rooms=payload.rooms,
@@ -2975,7 +2989,8 @@ async def estimate_quality(
         )
 
     # ── Deterministic pricing orchestration ──────────────────────────────────
-    pr = _price_from_inputs(
+    pr = await asyncio.to_thread(
+        _price_from_inputs,
         listings=listings,
         area_m2=payload.area_m2,
         rooms=payload.rooms,
@@ -3026,7 +3041,8 @@ async def estimate_quality(
     # 5. Deals — ДКП-only sales (вторичка) из rosreestr_deals.
     #    Importer фильтрует doc_type='ДКП' (PR-A 2026-05-24), ДДУ застройщиков
     #    исключены — больше не скёюят median вторички ~110-120 К/м².
-    deals = _fetch_deals(
+    deals = await asyncio.to_thread(
+        _fetch_deals,
         db,
         lat=geo.lat,
         lon=geo.lon,
@@ -3077,119 +3093,124 @@ async def estimate_quality(
         if dadata is not None and dadata.metro
         else None
     )
-    db.execute(
-        text(
-            """
-            INSERT INTO trade_in_estimates (
-                id, address, lat, lon,
-                area_m2, rooms, floor, total_floors,
-                year_built, house_type, repair_state, has_balcony,
-                ownership_type, has_mortgage, client_name, client_phone,
-                median_price, range_low, range_high, median_price_per_m2,
-                confidence, confidence_explanation, n_analogs,
-                analogs, actual_deals,
-                sources_used, data_freshness_minutes,
-                canonical_address, house_cadnum, house_fias_id,
-                dadata_qc_geo, dadata_qc_house, dadata_metro,
-                expected_sold_price, expected_sold_range_low,
-                expected_sold_range_high, expected_sold_per_m2,
-                asking_to_sold_ratio, ratio_basis,
-                created_by,
-                expires_at
-            ) VALUES (
-                CAST(:id AS uuid),
-                :address, :lat, :lon,
-                :area, :rooms, :floor, :total_floors,
-                :year_built, :house_type, :repair_state, :has_balcony,
-                :ownership_type, :has_mortgage, :client_name, :client_phone,
-                :median_price, :range_low, :range_high, :median_ppm2,
-                :confidence, :explanation, :n_analogs,
-                CAST(:analogs_json AS jsonb),
-                CAST(:deals_json AS jsonb),
-                CAST(:sources_json AS jsonb),
-                :freshness,
-                :canonical_address, :house_cadnum, :house_fias_id,
-                :dadata_qc_geo, :dadata_qc_house,
-                CAST(:dadata_metro_json AS jsonb),
-                :expected_sold_price, :expected_sold_range_low,
-                :expected_sold_range_high, :expected_sold_per_m2,
-                :asking_to_sold_ratio, :ratio_basis,
-                :created_by,
-                :expires_at
-            )
-            """
-        ),
-        {
-            "id": str(estimate_id),
-            "address": geo.full_address,
-            "lat": geo.lat,
-            "lon": geo.lon,
-            "area": payload.area_m2,
-            "rooms": payload.rooms,
-            "floor": payload.floor,
-            "total_floors": payload.total_floors,
-            "year_built": target_year,
-            "house_type": target_house_type,
-            "repair_state": payload.repair_state,
-            "has_balcony": payload.has_balcony,
-            "ownership_type": payload.ownership_type,
-            "has_mortgage": payload.has_mortgage,
-            "client_name": payload.client_name,
-            "client_phone": payload.client_phone,
-            "median_price": median_price,
-            "range_low": range_low,
-            "range_high": range_high,
-            "median_ppm2": int(median_ppm2),
-            "confidence": confidence,
-            "explanation": explanation,
-            "n_analogs": n_analogs,
-            "analogs_json": json.dumps(
-                [a.model_dump(mode="json") for a in analogs_lots], ensure_ascii=False
-            ),
-            "deals_json": json.dumps(
-                [a.model_dump(mode="json") for a in deals_lots], ensure_ascii=False
-            ),
-            # #2087 (M1): персистим КАНОНИЧЕСКИЙ sources_used (листинговые из
-            # analogs ∪ оценочные флаги) — тот же массив, что в POST-ответе. GET
-            # рехайдрейтит его же (валид. флаги читаются из этой колонки), history
-            # и PDF получают консистентный «X/7» без quarter_index/радиусного шума.
-            "sources_json": json.dumps(sources_used, ensure_ascii=False),
-            "freshness": freshness_pre,
-            "canonical_address": dadata.canonical_address if dadata else None,
-            "house_cadnum": dadata.house_cadnum if dadata else None,
-            "house_fias_id": dadata.house_fias_id if dadata else None,
-            "dadata_qc_geo": dadata.qc_geo if dadata else None,
-            "dadata_qc_house": dadata.qc_house if dadata else None,
-            "dadata_metro_json": dadata_metro_json,
-            "expected_sold_price": expected_sold_price,
-            "expected_sold_range_low": expected_sold_range_low,
-            "expected_sold_range_high": expected_sold_range_high,
-            "expected_sold_per_m2": expected_sold_per_m2,
-            "asking_to_sold_ratio": asking_to_sold_ratio,
-            "ratio_basis": ratio_basis,
-            "created_by": created_by,
-            "expires_at": expires_at,
-        },
-    )
 
-    # Link saved IMV evaluation к этому estimate_id атомарно с основным INSERT
-    # (closes finding #4 from 2026-05-24 audit — prior code committed estimate first,
-    # then UPDATEd IMV in a separate tx, racing against concurrent estimators
-    # sharing the same cache_key).
-    if imv_eval is not None:
+    # #2207: INSERT+commit сгруппированы и вынесены с event loop (to_thread).
+    def _persist_estimate_and_commit() -> None:
         db.execute(
             text(
                 """
-                UPDATE avito_imv_evaluations
-                SET estimate_id = CAST(:estimate_id AS uuid)
-                WHERE cache_key = :cache_key
-                  AND (estimate_id IS NULL OR estimate_id = CAST(:estimate_id AS uuid))
+                INSERT INTO trade_in_estimates (
+                    id, address, lat, lon,
+                    area_m2, rooms, floor, total_floors,
+                    year_built, house_type, repair_state, has_balcony,
+                    ownership_type, has_mortgage, client_name, client_phone,
+                    median_price, range_low, range_high, median_price_per_m2,
+                    confidence, confidence_explanation, n_analogs,
+                    analogs, actual_deals,
+                    sources_used, data_freshness_minutes,
+                    canonical_address, house_cadnum, house_fias_id,
+                    dadata_qc_geo, dadata_qc_house, dadata_metro,
+                    expected_sold_price, expected_sold_range_low,
+                    expected_sold_range_high, expected_sold_per_m2,
+                    asking_to_sold_ratio, ratio_basis,
+                    created_by,
+                    expires_at
+                ) VALUES (
+                    CAST(:id AS uuid),
+                    :address, :lat, :lon,
+                    :area, :rooms, :floor, :total_floors,
+                    :year_built, :house_type, :repair_state, :has_balcony,
+                    :ownership_type, :has_mortgage, :client_name, :client_phone,
+                    :median_price, :range_low, :range_high, :median_ppm2,
+                    :confidence, :explanation, :n_analogs,
+                    CAST(:analogs_json AS jsonb),
+                    CAST(:deals_json AS jsonb),
+                    CAST(:sources_json AS jsonb),
+                    :freshness,
+                    :canonical_address, :house_cadnum, :house_fias_id,
+                    :dadata_qc_geo, :dadata_qc_house,
+                    CAST(:dadata_metro_json AS jsonb),
+                    :expected_sold_price, :expected_sold_range_low,
+                    :expected_sold_range_high, :expected_sold_per_m2,
+                    :asking_to_sold_ratio, :ratio_basis,
+                    :created_by,
+                    :expires_at
+                )
                 """
             ),
-            {"estimate_id": str(estimate_id), "cache_key": imv_eval.cache_key},
+            {
+                "id": str(estimate_id),
+                "address": geo.full_address,
+                "lat": geo.lat,
+                "lon": geo.lon,
+                "area": payload.area_m2,
+                "rooms": payload.rooms,
+                "floor": payload.floor,
+                "total_floors": payload.total_floors,
+                "year_built": target_year,
+                "house_type": target_house_type,
+                "repair_state": payload.repair_state,
+                "has_balcony": payload.has_balcony,
+                "ownership_type": payload.ownership_type,
+                "has_mortgage": payload.has_mortgage,
+                "client_name": payload.client_name,
+                "client_phone": payload.client_phone,
+                "median_price": median_price,
+                "range_low": range_low,
+                "range_high": range_high,
+                "median_ppm2": int(median_ppm2),
+                "confidence": confidence,
+                "explanation": explanation,
+                "n_analogs": n_analogs,
+                "analogs_json": json.dumps(
+                    [a.model_dump(mode="json") for a in analogs_lots], ensure_ascii=False
+                ),
+                "deals_json": json.dumps(
+                    [a.model_dump(mode="json") for a in deals_lots], ensure_ascii=False
+                ),
+                # #2087 (M1): персистим КАНОНИЧЕСКИЙ sources_used (листинговые из
+                # analogs ∪ оценочные флаги) — тот же массив, что в POST-ответе. GET
+                # рехайдрейтит его же (валид. флаги читаются из этой колонки), history
+                # и PDF получают консистентный «X/7» без quarter_index/радиусного шума.
+                "sources_json": json.dumps(sources_used, ensure_ascii=False),
+                "freshness": freshness_pre,
+                "canonical_address": dadata.canonical_address if dadata else None,
+                "house_cadnum": dadata.house_cadnum if dadata else None,
+                "house_fias_id": dadata.house_fias_id if dadata else None,
+                "dadata_qc_geo": dadata.qc_geo if dadata else None,
+                "dadata_qc_house": dadata.qc_house if dadata else None,
+                "dadata_metro_json": dadata_metro_json,
+                "expected_sold_price": expected_sold_price,
+                "expected_sold_range_low": expected_sold_range_low,
+                "expected_sold_range_high": expected_sold_range_high,
+                "expected_sold_per_m2": expected_sold_per_m2,
+                "asking_to_sold_ratio": asking_to_sold_ratio,
+                "ratio_basis": ratio_basis,
+                "created_by": created_by,
+                "expires_at": expires_at,
+            },
         )
 
-    db.commit()
+        # Link saved IMV evaluation к этому estimate_id атомарно с основным INSERT
+        # (closes finding #4 from 2026-05-24 audit — prior code committed estimate first,
+        # then UPDATEd IMV in a separate tx, racing against concurrent estimators
+        # sharing the same cache_key).
+        if imv_eval is not None:
+            db.execute(
+                text(
+                    """
+                    UPDATE avito_imv_evaluations
+                    SET estimate_id = CAST(:estimate_id AS uuid)
+                    WHERE cache_key = :cache_key
+                      AND (estimate_id IS NULL OR estimate_id = CAST(:estimate_id AS uuid))
+                    """
+                ),
+                {"estimate_id": str(estimate_id), "cache_key": imv_eval.cache_key},
+            )
+
+        db.commit()
+
+    await asyncio.to_thread(_persist_estimate_and_commit)
 
     logger.info(
         "estimate: id=%s addr=%s rooms=%d area=%.1f → median=%d (n=%d, conf=%s)%s%s",
@@ -3211,7 +3232,8 @@ async def estimate_quality(
     last_scraped_at = _compute_last_scraped_at(metadata_lots)
     # Месячный ₽/м² тренд целевого дома (web TREND chart) — best-effort, None если нет данных.
     # #audit-3: передаём freshness_months из настроек — исключаем устаревшие items.
-    price_trend_raw = _fetch_price_trend(
+    price_trend_raw = await asyncio.to_thread(
+        _fetch_price_trend,
         db,
         target_house_id=target_house_id,
         freshness_months=settings.estimate_price_trend_max_age_months,
@@ -3228,9 +3250,11 @@ async def estimate_quality(
     # median/expected_sold/ranges. Best-effort: degrade в (False, None, None) при
     # отсутствии таблиц. premium_building_class — класс из curated (None для
     # невыверенных MV-домов).
-    premium_building, premium_building_median_ppm2, premium_building_class = _is_premium_building(
-        db, target_house_id
-    )
+    (
+        premium_building,
+        premium_building_median_ppm2,
+        premium_building_class,
+    ) = await asyncio.to_thread(_is_premium_building, db, target_house_id)
 
     # #audit-2: структурный analog_tier — стабильный enum для фронта.
     # anchor-путь: anchor_tier "A" → "same_building", "C" → "micro_radius".

tradein-mvp/backend/tests/test_estimator_event_loop_2207.py

@@ -0,0 +1,105 @@
+"""#2207 — estimate_quality не должен блокировать event loop.
+
+`estimate_quality` выпускает ~28 СИНХРОННЫХ psycopg PostGIS-запросов (multi-tier
+analogs ×4, deals, price_trend, anchor comps, INSERT+commit и т.д.). На проде это
+один uvicorn worker → одна медленная оценка раньше стопила ВЕСЬ event loop,
+включая /health. Фикс #2207: каждый sync-кластер вынесен через
+``asyncio.to_thread(...)``.
+
+Тест патчит ОДИН тяжёлый sync-хелпер (`_fetch_analogs`) на блокирующий
+``time.sleep(0.3)`` и запускает `estimate_quality` конкурентно с лёгкой
+корутиной (``asyncio.sleep(0.05)``). Если бы loop блокировался, лёгкая корутина
+не смогла бы возобновиться до конца sync-сна; мы ассертим, что она завершается
+у ~0.05с — задолго до оценки, — что доказывает свободный loop.
+
+Стиль зеркалит tests/test_estimator_client_coords.py.
+"""
+
+import asyncio
+import contextlib
+import os
+import time
+
+# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+def _make_payload():
+    from app.schemas.trade_in import TradeInEstimateInput
+
+    return TradeInEstimateInput(
+        address="ЕКБ, ул. Учителей, 18",
+        area_m2=38.8,
+        rooms=1,
+        floor=4,
+        total_floors=16,
+        # in-EKB coords → geocode() пропускается (Variant A), сеть не трогаем.
+        lat=56.838,
+        lon=60.595,
+    )
+
+
+# Тяжёлый sync-хелпер: имитирует медленный PostGIS-запрос блокирующим сном.
+def _slow_fetch_analogs(*_args, **_kwargs):
+    time.sleep(0.3)
+    return ([], False, "W")
+
+
+def _downstream_patches():
+    """Оффлайн-моки так, чтобы estimate_quality дошёл до конца детерминированно."""
+    return (
+        patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
+        patch("app.services.estimator.match_house_readonly", return_value=None),
+        patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+        patch("app.services.estimator._fetch_analogs", side_effect=_slow_fetch_analogs),
+        patch("app.services.estimator._fetch_deals", return_value=[]),
+        patch("app.services.estimator._get_or_fetch_imv_cached", new=AsyncMock(return_value=None)),
+        patch(
+            "app.services.estimator._get_or_fetch_yandex_valuation_cached",
+            new=AsyncMock(return_value=None),
+        ),
+        patch(
+            "app.services.estimator.estimate_via_cian_valuation",
+            new=AsyncMock(return_value=None),
+        ),
+        patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
+    )
+
+
+def test_estimate_quality_does_not_block_event_loop() -> None:
+    """Медленный sync-путь оценки не должен starve'ить event loop."""
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    payload = _make_payload()
+
+    async def _run() -> dict[str, float]:
+        with contextlib.ExitStack() as stack:
+            for cm in _downstream_patches():
+                stack.enter_context(cm)
+
+            timings: dict[str, float] = {}
+            start = time.perf_counter()
+
+            async def light() -> None:
+                await asyncio.sleep(0.05)
+                timings["light"] = time.perf_counter() - start
+
+            async def heavy() -> None:
+                await estimate_quality(payload, db)
+                timings["estimate"] = time.perf_counter() - start
+
+            await asyncio.gather(light(), heavy())
+            return timings
+
+    timings = asyncio.run(_run())
+
+    # Лёгкая корутина завершилась ДО оценки в целом.
+    assert timings["light"] < timings["estimate"], timings
+    # И — ключевое — возобновилась у ~0.05с, а НЕ после 0.3с sync-сна
+    # (было бы >=0.3, если бы loop блокировался). Потолок с запасом на CI-jitter.
+    assert timings["light"] < 0.25, timings
+    # Оценка реально ждала sync-работу в потоке(ах): >=0.3с (>=1 sleep).
+    assert timings["estimate"] >= 0.3, timings