diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 4be0645b..766959fb 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -384,6 +384,13 @@ class Settings(BaseSettings): # ENV: CIAN_PROXY_MAX_ROTATIONS. cian_proxy_max_rotations: int = 4 + # #1949: per-fetch hard-cancel для browser-fetch'ей внутри bucket-gather Cian full_load. + # asyncio.wait_for(fetch, timeout=X) отменяет зависший fetch → TimeoutError → + # _fetch_page_html ловит как Exception → None → _one_page → [] → gather завершается. + # 0 = отключить (старое поведение). Дефолт 90.0s щедрее nормального fetch ~12s. + # ENV: CIAN_FULL_LOAD_PER_FETCH_TIMEOUT_S. + cian_full_load_per_fetch_timeout_s: float = 90.0 + @property def cian_proxy_url(self) -> str | None: """Прокси для Cian-скраперов. CIAN_PROXY_URL > scraper_proxy_url (fallback).""" diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 6141710e..aa38b5ff 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -33,6 +33,7 @@ import random from contextlib import AsyncExitStack from dataclasses import dataclass, field, fields from datetime import date, timedelta +from typing import Any from urllib.parse import urlparse from curl_cffi.requests import AsyncSession @@ -2535,10 +2536,61 @@ async def run_cian_full_load( """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + # #1949: фоновый heartbeat — обновляет heartbeat_at каждые 60s независимо от on_bucket. + # Без него зависший asyncio.gather внутри bucket-loop замораживает heartbeat на часы + # → reap_zombies помечает живой run zombie. Отменяется в finally при любом завершении. + async def _background_heartbeat() -> None: + """Периодический heartbeat каждые 60s, не завязанный на прогресс бакетов.""" + while True: + await asyncio.sleep(60.0) + try: + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + except Exception: + pass + + _hb_task: asyncio.Task[None] | None = None try: + _hb_task = asyncio.create_task(_background_heartbeat()) + async with CianScraper() as scraper: scraper.request_delay_sec = request_delay_sec + # #1949: per-fetch hard-cancel для browser-fetch'ей внутри bucket-gather. + # BrowserFetcher.fetch имеет httpx-timeout 120s, но browser-сервис иногда + # виснет так что httpx не получает ответ → gather морозит → heartbeat стоп. + # Оборачиваем _browser.fetch в asyncio.wait_for(timeout=X) через monkey-patch: + # TimeoutError → _fetch_page_html ловит как Exception → None → _one_page → [] + # → gather завершается нормально → on_bucket вызывается → heartbeat обновляется. + # За флагом cian_full_load_per_fetch_timeout_s (0 = отключить). + _fetch_timeout = settings.cian_full_load_per_fetch_timeout_s + if _fetch_timeout > 0 and scraper._browser is not None: + _orig_fetch = scraper._browser.fetch + + async def _timed_fetch( + url: str, + _f: Any = _orig_fetch, + _t: float = _fetch_timeout, + _rid: int = run_id, + ) -> str: + try: + return await asyncio.wait_for(_f(url), timeout=_t) + except TimeoutError: + logger.warning( + "cian-full-load run_id=%d: browser-fetch timed out " + "after %.0fs — url=%s", + _rid, + _t, + url, + ) + raise + + scraper._browser.fetch = _timed_fetch # type: ignore[method-assign] + logger.info( + "cian-full-load run_id=%d: per-fetch timeout %.0fs enabled", + run_id, + _fetch_timeout, + ) + await scraper.fetch_all_secondary( price_cap_per_bucket=price_cap_per_bucket, concurrency=concurrency, @@ -2652,6 +2704,15 @@ async def run_cian_full_load( scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise + finally: + # #1949: отменяем фоновый heartbeat при любом завершении (return / raise / cancel). + if _hb_task is not None: + _hb_task.cancel() + try: + await _hb_task + except asyncio.CancelledError: + pass + # ── Yandex exhaustive full load (region-wide, без anchor'ов) ────────────────── diff --git a/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py b/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py index 07313c56..e659e375 100644 --- a/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py +++ b/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py @@ -6,15 +6,26 @@ Fix A (#1950): _rotate_proxy_ip теперь делает до proxy_rotate_atte Fix B (#1950): run_avito_city_sweep при AvitoBlockedError/RateLimited с lots_inserted+lots_updated>0 вызывает mark_done (не mark_banned) когда avito_serp_ok_not_banned=True. + +Fix C (#1949): run_cian_full_load применяет monkey-patch _browser.fetch с + asyncio.wait_for(timeout=cian_full_load_per_fetch_timeout_s). + +Fix D (#1949): run_cian_full_load создаёт фоновый heartbeat-task и отменяет + его в finally при любом завершении. """ from __future__ import annotations +import asyncio +import os +import time from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + from app.services.scrapers.avito_exceptions import AvitoBlockedError from app.services.scrapers.base import ScrapedLot @@ -334,3 +345,214 @@ async def test_avito_city_sweep_flag_off_still_marks_banned() -> None: mock_runs.mark_banned.called ), "avito_serp_ok_not_banned=False → mark_banned (backward-compat)" assert not mock_runs.mark_done.called + + +# ────────────────────────────────────────────────────────────────────────────── +# Fix C: cian per-fetch wait_for (#1949) +# ────────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_cian_full_load_hanging_fetch_cancelled_by_timeout() -> None: + """#1949-C: browser-fetch зависший навсегда отменяется per-fetch timeout. + + Monkey-patch заменяет _browser.fetch на _timed_fetch(wait_for timeout=0.05s). + Зависший fetch отменяется через ~50ms → TimeoutError → run завершается. + """ + from app.services.scrape_pipeline import run_cian_full_load + + fetch_was_called = False + + async def hanging_fetch(url: str) -> str: + nonlocal fetch_was_called + fetch_was_called = True + await asyncio.Event().wait() # висит вечно + return "" + + mock_browser = MagicMock() + mock_browser.fetch = hanging_fetch + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = mock_browser + mock_scraper.request_delay_sec = 0.0 + + # fetch_all_secondary симулирует _fetch_page_html: вызывает fetch и ловит Exception + async def fake_fetch_all_secondary(**kwargs: object) -> None: + try: + # После monkey-patch _browser.fetch = _timed_fetch (с timeout=0.05s) + await mock_scraper._browser.fetch("https://ekb.cian.ru/sale/flat/1/") + except Exception: + pass # TimeoutError → аналог _fetch_page_html → returns None + + mock_scraper.fetch_all_secondary = fake_fetch_all_secondary + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + t0 = time.monotonic() + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.05 # 50ms — быстрый timeout для теста + s.cian_full_load_detail_top_n = 0 + + await run_cian_full_load(mock_db, run_id=1) + + elapsed = time.monotonic() - t0 + + assert fetch_was_called, "hanging_fetch должен быть вызван через monkey-patched _browser.fetch" + assert ( + elapsed < 3.0 + ), f"run должен завершиться быстро (fetch timeout 50ms), занял {elapsed:.2f}s" + assert mock_runs.mark_done.called, "run_cian_full_load должен завершиться mark_done" + + +@pytest.mark.asyncio +async def test_cian_full_load_per_fetch_timeout_disabled_when_zero() -> None: + """#1949-C: cian_full_load_per_fetch_timeout_s=0 → monkey-patch НЕ применяется.""" + from app.services.scrape_pipeline import run_cian_full_load + + original_fetch = AsyncMock(return_value="") + + mock_browser = MagicMock() + mock_browser.fetch = original_fetch + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = mock_browser + mock_scraper.request_delay_sec = 0.0 + mock_scraper.fetch_all_secondary = AsyncMock(return_value=None) + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.0 # отключено + s.cian_full_load_detail_top_n = 0 + + await run_cian_full_load(mock_db, run_id=2) + + # Когда timeout=0 (отключён), monkey-patch не применяется → fetch остаётся original + assert ( + mock_browser.fetch is original_fetch + ), "cian_full_load_per_fetch_timeout_s=0 → fetch НЕ должен быть заменён monkey-patch" + + +# ────────────────────────────────────────────────────────────────────────────── +# Fix D: cian background heartbeat created and cancelled (#1949) +# ────────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_cian_full_load_background_heartbeat_created_and_cancelled() -> None: + """#1949-D: background heartbeat task создаётся и отменяется в finally.""" + from app.services.scrape_pipeline import run_cian_full_load + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = None # per-fetch timeout не применяется + mock_scraper.fetch_all_secondary = AsyncMock(return_value=None) + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + created_tasks: list[asyncio.Task[None]] = [] + _real_create_task = asyncio.create_task + + def _tracking_create_task(coro: Any, *, name: str | None = None) -> asyncio.Task[None]: + task = _real_create_task(coro, name=name) + created_tasks.append(task) + return task # type: ignore[return-value] + + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch( + "app.services.scrape_pipeline.asyncio.create_task", + side_effect=_tracking_create_task, + ), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.0 + s.cian_full_load_detail_top_n = 0 + + await run_cian_full_load(mock_db, run_id=3) + + assert ( + len(created_tasks) >= 1 + ), "asyncio.create_task должен быть вызван хотя бы раз (background heartbeat)" + for task in created_tasks: + assert ( + task.cancelled() or task.done() + ), f"Heartbeat task должен быть отменён в finally: {task}" + + +@pytest.mark.asyncio +async def test_cian_full_load_heartbeat_cancelled_on_exception() -> None: + """#1949-D: heartbeat task отменяется в finally даже при исключении в run.""" + from app.services.scrape_pipeline import run_cian_full_load + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = None + mock_scraper.fetch_all_secondary = AsyncMock(side_effect=RuntimeError("fatal-test-error")) + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + created_tasks: list[asyncio.Task[None]] = [] + _real_create_task = asyncio.create_task + + def _tracking_create_task(coro: Any, *, name: str | None = None) -> asyncio.Task[None]: + task = _real_create_task(coro, name=name) + created_tasks.append(task) + return task # type: ignore[return-value] + + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch( + "app.services.scrape_pipeline.asyncio.create_task", + side_effect=_tracking_create_task, + ), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.0 + s.cian_full_load_detail_top_n = 0 + + with pytest.raises(RuntimeError, match="fatal-test-error"): + await run_cian_full_load(mock_db, run_id=4) + + assert len(created_tasks) >= 1, "heartbeat task должен быть создан" + for task in created_tasks: + assert ( + task.cancelled() or task.done() + ), f"Heartbeat task должен быть отменён в finally даже при RuntimeError: {task}" + assert mock_runs.mark_failed.called, "mark_failed должен быть вызван при RuntimeError"