From e3adcd47be1238656a99dffa89c7759dc027de3d Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 27 Jun 2026 23:19:26 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein/avito):=20hard-timeout=20on=20detai?= =?UTF-8?q?l-backfill=20fetch=20=E2=80=94=20prevent=20zombie-hang=20(#1950?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prod run 423 завис на 7.7ч (healthy ~2.4ч) → reaped как zombie → detail-backlog не уменьшился (21456 листингов ждут detail, 29.3% enriched). Корень: fetch_detail(...) в loop'е без hard-timeout. Один зависший fetch (camoufox/browser hang или curl-stall) блокировал loop навсегда → budget-guard (раз в итерацию) не срабатывал → heartbeat (раз в 25 attempts) не обновлялся → run zombie. Вторичный hang-вектор: _rotate_ip() на блок — тоже без timeout. Fix: - asyncio.wait_for(fetch_detail, timeout=avito_detail_fetch_timeout_s=90s): зависший fetch отменяется → TimeoutError → листинг failed, loop идёт дальше. TimeoutError ловится ДО общего Exception (TimeoutError ⊂ OSError ⊂ Exception), порядок: AvitoBlocked/RateLimited → TimeoutError → Exception. - bound rotate: asyncio.wait_for(_rotate_ip, timeout=settle+30s) + try/except — зависший changeip не роняет run. - config: avito_detail_fetch_timeout_s (ENV AVITO_DETAIL_FETCH_TIMEOUT_S). - admin: POST /scrape/avito-detail-backfill (batch_size, budget_sec) — on-demand прогон/верификация (основной schedule window-gated 09-12 UTC). Восстанавливает throughput detail-очереди (точные coords #1967 + house_type/kitchen). --- tradein-mvp/backend/app/api/v1/admin.py | 40 +++++++++++++++ tradein-mvp/backend/app/core/config.py | 10 ++++ .../app/tasks/avito_detail_backfill.py | 51 +++++++++++++++++-- .../tests/tasks/test_avito_detail_backfill.py | 51 +++++++++++++++++++ 4 files changed, 147 insertions(+), 5 deletions(-) diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 6abd1c7d..1720ccf7 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -49,6 +49,7 @@ from app.services.scrapers.yandex_detail import YandexDetailScraper from app.services.scrapers.yandex_newbuilding import YandexNewbuildingScraper from app.services.scrapers.yandex_realty import YandexRealtyScraper from app.services.scrapers.yandex_valuation import YandexValuationScraper +from app.tasks.avito_detail_backfill import run_avito_detail_backfill from app.tasks.geocode_missing import GeocodeBackfillResult, geocode_missing_listings logger = logging.getLogger(__name__) @@ -592,6 +593,45 @@ async def scrape_avito_detail( return {"ok": True, "item_id": enrichment.item_id, "updated": updated} +@router.post("/scrape/avito-detail-backfill") +async def scrape_avito_detail_backfill( + db: Annotated[Session, Depends(get_db)], + batch_size: int = Query(default=200, ge=1, le=2000), + budget_sec: float = Query(default=600.0, ge=30.0, le=3600.0), +) -> dict[str, object]: + """On-demand прогон detail-backfill detail-очереди avito (#1950). + + Запускает `run_avito_detail_backfill` синхронно (await) на одном snapshot'е + из batch_size pending-листингов (detail_enriched_at IS NULL, активные ЕКБ), + ограниченный budget_sec. Создаёт scrape_runs-строку (source='avito_detail_backfill') + для observability — статус виден в общем списке runs. + + Нужен для ручной верификации/прогона: основной schedule window-gated (09-12 UTC), + вне окна его не запустить. Не для постоянного использования — backlog нормально + рассасывается ночным schedule. + + Returns counters {'attempted','enriched','blocked','failed','duration_sec'} + run_id. + """ + if has_running_run(db, "avito_detail_backfill"): + raise HTTPException( + status_code=409, + detail=( + "avito detail-backfill уже выполняется — одновременно допустим только один " + "(параллельные раны делят прокси-IP → Avito бан). Дождитесь завершения." + ), + ) + params = {"batch_size": batch_size, "budget_sec": budget_sec} + run_id = runs_mod.create_run(db, source="avito_detail_backfill", params=params) + try: + result = await run_avito_detail_backfill(db, run_id=run_id, params=params) + except Exception as e: + logger.exception("avito-detail-backfill: run_id=%d crashed", run_id) + raise HTTPException(status_code=502, detail=f"backfill failed: {e}") from e + + logger.info("avito-detail-backfill ok: run_id=%d %s", run_id, result.to_dict()) + return {"ok": True, "run_id": run_id, "counters": result.to_dict()} + + @router.post("/scrape/avito-imv") async def scrape_avito_imv( db: Annotated[Session, Depends(get_db)], diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index ff7c2f9d..46c273f9 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -501,6 +501,16 @@ class Settings(BaseSettings): default=True, validation_alias="AVITO_DETAIL_BACKFILL_USE_CURL" ) + # #1950: hard-timeout на один detail-fetch внутри avito_detail_backfill. Зависший + # fetch_detail (camoufox/browser hang или curl-stall) блокирует loop навсегда → + # budget-guard (раз в итерацию) не срабатывает → heartbeat не обновляется → run + # reaped как zombie (run 423 завис 7.7ч). asyncio.wait_for(fetch, timeout=X) + # отменяет зависший fetch → TimeoutError → листинг failed, loop идёт дальше. + # 90s щедрее нормального detail-fetch (~10-15s). ENV: AVITO_DETAIL_FETCH_TIMEOUT_S. + avito_detail_fetch_timeout_s: float = Field( + default=90.0, validation_alias="AVITO_DETAIL_FETCH_TIMEOUT_S" + ) + # ── #884/#905/#1805: BrowserFetcher — HTTP-клиент к tradein-browser ───────── # scraper_fetch_mode: "browser" (дефолт с #1805 — HTTP POST к tradein-browser # /fetch через per-provider camoufox + ротирующий backconnect-прокси) или diff --git a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py index e4ebb81c..a6f2e625 100644 --- a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py +++ b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py @@ -86,6 +86,13 @@ async def run_avito_detail_backfill( request_delay_sec = float(params.get("request_delay_sec", 6.0)) max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5)) + # #1950: hard-timeout'ы на блокирующие await'ы внутри loop'а. Без них один + # зависший fetch_detail/rotate ронял весь run в zombie (run 423 завис 7.7ч). + # Локальные float'ы (а не settings.* напрямую) — чтобы wait_for/format получали + # число даже когда settings замокан MagicMock'ом в тестах. + fetch_timeout_s = float(settings.avito_detail_fetch_timeout_s) + rotate_timeout_s = float(settings.avito_proxy_rotate_settle_s) + 30.0 + counters = AvitoDetailBackfillResult() current_counters: dict[str, int] = counters.to_dict() @@ -212,10 +219,16 @@ async def run_avito_detail_backfill( item_url = urlparse(source_url).path if source_url.startswith("http") else source_url try: - enrichment = await fetch_detail( - item_url, - cffi_session=session, - browser_fetcher=browser_fetcher, + # #1950: hard-timeout — зависший fetch (browser hang / curl-stall) не + # должен блокировать loop навсегда (иначе budget-guard/heartbeat молчат + # и run reaped как zombie). wait_for отменяет fetch → TimeoutError. + enrichment = await asyncio.wait_for( + fetch_detail( + item_url, + cffi_session=session, + browser_fetcher=browser_fetcher, + ), + timeout=fetch_timeout_s, ) if save_detail_enrichment(db, enrichment): counters.enriched += 1 @@ -233,7 +246,18 @@ async def run_avito_detail_backfill( consecutive_blocks, e, ) - await scraper._rotate_ip() + # #1950: bound rotate — _rotate_ip ждёт settle + до 3 changeip-попыток; + # на блокирующем changeip (зависшее соединение) без timeout loop виснет. + try: + await asyncio.wait_for(scraper._rotate_ip(), timeout=rotate_timeout_s) + except Exception: + logger.warning( + "avito_detail_backfill: run_id=%d rotate_ip timed out/failed " + "(>%.0fs) -- continuing", + run_id, + rotate_timeout_s, + exc_info=True, + ) if consecutive_blocks >= max_consecutive_blocks: logger.error( "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, " @@ -245,6 +269,23 @@ async def run_avito_detail_backfill( ) break + except TimeoutError: + # asyncio.wait_for → TimeoutError (py3.12: asyncio.TimeoutError — alias). + # Ловим ДО общего Exception (TimeoutError ⊂ OSError ⊂ Exception). Зависший + # fetch отменён → листинг failed, переходим к следующему (loop не зависает, + # run не zombie #1950). Не считаем soft-блоком: rotate не дёргаем. + counters.failed += 1 + logger.warning( + "avito_detail_backfill: run_id=%d listing %s TIMEOUT (>%.0fs) -- skip", + run_id, + source_url, + fetch_timeout_s, + ) + try: + db.rollback() + except Exception: + pass + except Exception as e: counters.failed += 1 logger.warning( diff --git a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py index fd4fcd05..5e1c66d7 100644 --- a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py +++ b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import os import sys from unittest.mock import AsyncMock, MagicMock, patch @@ -272,6 +273,56 @@ async def test_backfill_fetch_exception_continues() -> None: runs.mark_done.assert_called_once() +@pytest.mark.asyncio +async def test_backfill_fetch_timeout_skips_and_continues() -> None: + """#1950: один fetch_detail зависает дольше hard-timeout → этот листинг failed, + loop НЕ зависает, переходит к следующему, run завершается mark_done (не zombie). + + Регрессия run 423 (завис 7.7ч → reaped): fetch_detail без timeout блокировал + loop навсегда. asyncio.wait_for(timeout) отменяет зависший fetch → TimeoutError. + """ + snapshot = _make_snapshot(2) + db = _mock_db(snapshot) + runs = MagicMock() + mock_enrichment = MagicMock() + + call_urls: list[str] = [] + + async def _fetch(url: str, *, cffi_session: object = None, browser_fetcher: object = None): + call_urls.append(url) + if len(call_urls) == 1: + await asyncio.Event().wait() # висит вечно → wait_for отменит по timeout + return mock_enrichment + + # Короткий timeout (50ms) — тест не ждёт реальные 90s; rotate-settle тоже 0. + fake_settings = MagicMock( + scraper_fetch_mode="cffi", + avito_detail_fetch_timeout_s=0.05, + avito_proxy_rotate_settle_s=0.0, + ) + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, _fetch), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + ): + result = await run_avito_detail_backfill( + db, run_id=11, params={"batch_size": 10, "budget_sec": 3600} + ) + + # Первый листинг — failed (timeout), второй — enriched. Run завершён mark_done. + assert result.failed == 1 + assert result.enriched == 1 + assert result.attempted == 2 + assert len(call_urls) == 2, "loop должен дойти до второго листинга, а не зависнуть" + db.rollback.assert_called() + runs.mark_done.assert_called_once() + runs.mark_failed.assert_not_called() + + _BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"