diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 144fdc78..b8b1b447 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/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". diff --git a/tradein-mvp/backend/tests/test_estimator_event_loop_2207.py b/tradein-mvp/backend/tests/test_estimator_event_loop_2207.py new file mode 100644 index 00000000..22c9adb8 --- /dev/null +++ b/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