fix(tradein/scraper-kit): scale cian anchor watchdog for proxy-pool browser fetches (#2160)

Fixed ANCHOR_TIMEOUT_SEC=240 guillotined every cian anchor when running via
USE_PROXY_POOL_BROWSER=true: each SERP fetch goes through camoufox with relaunch
on proxy swap (13-45s/page), and an anchor = 4 room-buckets x pages_per_anchor
SERP fetches + detail_top_n details + houses-enrich >> 240s. save_listings never
ran, lots_fetched=0, and after 3 consecutive skips the run was mark_banned.

Introduces _cian_anchor_timeout_s() mirroring the avito/yandex/domclick scaling
pattern: max(ANCHOR_TIMEOUT_SEC, buckets*pages*(delay+per_serp) + details*per_detail
+ houses_budget). Defaults -> 1580s worst-case watchdog (hang guard, not expected
duration). Applied to both the scraper-kit copy (strangler target) and the active
app.services.scrape_pipeline copy that run 581 actually hit.

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

@@ -175,6 +175,45 @@ _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5
 # паттерн avito_detail_backfill (max_consecutive_blocks).
 _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3
 
+# #2160: константы для расчёта watchdog-таймаута Cian city sweep. При
+# USE_PROXY_POOL_BROWSER=true каждый SERP-фетч идёт через camoufox с relaunch при смене
+# прокси (page.goto timeout 60s + overhead) = 13-45s/страница, а якорь cian = 4 room-buckets
+# × pages_per_anchor SERP-фетчей + detail_top_n detail-фетчей + houses-enrich. Фиксированный
+# ANCHOR_TIMEOUT_SEC=240 гильотинит якорь ещё в SERP-фазе → save_listings не успевает,
+# lots_fetched=0, после N подряд → mark_banned. Таймаут масштабируется:
+#   _cian_anchor_timeout = max(
+#       ANCHOR_TIMEOUT_SEC,
+#       _CIAN_ROOM_BUCKETS × pages_per_anchor × (request_delay_sec + _CIAN_PER_SERP_FETCH_S)
+#       + detail_top_n × _CIAN_PER_DETAIL_S
+#       + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0))
+# Пример (дефолты): 4×3×(5+70) + 10×50 + 180 = 900 + 500 + 180 = 1580s/anchor (worst-case,
+# это watchdog от зависания, не ожидаемая длительность).
+_CIAN_ROOM_BUCKETS: int = 4  # fetch_around_multi_room итерирует ((1,),(2,),(3,),(4,)),
+#   см. providers/cian/serp.py:215
+_CIAN_PER_SERP_FETCH_S: float = 70.0  # browser fetch через пул: page.goto 60s + camoufox relaunch
+_CIAN_PER_DETAIL_S: float = 50.0  # секунд на один detail-fetch (как avito)
+_CIAN_HOUSES_BUDGET_S: float = 180.0  # бюджет на house-enrich фазу (как avito)
+
+
+def _cian_anchor_timeout_s(
+    pages_per_anchor: int,
+    request_delay_sec: float,
+    detail_top_n: int,
+    enrich_houses: bool,
+) -> float:
+    """Масштабируемый watchdog-таймаут одного Cian anchor'а (секунды).
+
+    Чистая функция для тестируемости формулы (#2160). Никогда не опускается ниже
+    ANCHOR_TIMEOUT_SEC (max-ветка): при малых параметрах фиксированный минимум остаётся.
+    """
+    return max(
+        float(ANCHOR_TIMEOUT_SEC),
+        _CIAN_ROOM_BUCKETS * pages_per_anchor * (request_delay_sec + _CIAN_PER_SERP_FETCH_S)
+        + detail_top_n * _CIAN_PER_DETAIL_S
+        + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0),
+    )
+
+
 # Default anchors ЕКБ — 5 точек покрытия города
 EKB_ANCHORS: list[tuple[float, float, str]] = [
     (56.8400, 60.6050, "Центр"),
@@ -2108,6 +2147,27 @@ async def run_cian_city_sweep(
     consecutive_failures = 0
     cian_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep
 
+    # #2160: масштабируемый watchdog-таймаут для cian anchor'а. При proxy-pool browser каждый
+    # SERP-фетч 13-45s (camoufox relaunch), фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинит
+    # якорь в SERP-фазе. См. _cian_anchor_timeout_s + константы выше.
+    _cian_anchor_timeout = _cian_anchor_timeout_s(
+        pages_per_anchor, request_delay_sec, detail_top_n, enrich_houses
+    )
+    logger.info(
+        "cian-sweep run_id=%d: anchor_timeout=%.0fs "
+        "(base=%d, serp=%d×%d×(%.0f+%.0f)s + detail_top_n=%d×%.0fs + houses=%.0fs)",
+        run_id,
+        _cian_anchor_timeout,
+        ANCHOR_TIMEOUT_SEC,
+        _CIAN_ROOM_BUCKETS,
+        pages_per_anchor,
+        request_delay_sec,
+        _CIAN_PER_SERP_FETCH_S,
+        detail_top_n,
+        _CIAN_PER_DETAIL_S,
+        _CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0,
+    )
+
     try:
         for idx, (lat, lon, name) in enumerate(_anchors, start=1):
             # Cooperative cancel перед каждым anchor
@@ -2381,7 +2441,7 @@ async def run_cian_city_sweep(
                 )
 
             try:
-                await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC)
+                await asyncio.wait_for(_cian_anchor_phases(), timeout=_cian_anchor_timeout)
             except TimeoutError:
                 logger.warning(
                     "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
@@ -2392,7 +2452,7 @@ async def run_cian_city_sweep(
                     name,
                     lat,
                     lon,
-                    ANCHOR_TIMEOUT_SEC,
+                    _cian_anchor_timeout,
                 )
                 counters.errors_count += 1
                 consecutive_failures += 1
@@ -2413,7 +2473,7 @@ async def run_cian_city_sweep(
                         db,
                         run_id,
                         f"cian sweep aborted: {consecutive_failures} consecutive "
-                        f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)",
+                        f"anchor failures (last: anchor timeout {_cian_anchor_timeout:.0f}s)",
                         counters.to_dict(),
                     )
                     return counters

tradein-mvp/backend/tests/test_cian_city_sweep.py

@@ -785,3 +785,59 @@ def test_cian_list_runs_endpoint(app_with_admin) -> None:
     assert body[0]["run_id"] == 10
     assert body[0]["source"] == "cian_city_sweep"
     assert "2025-05-01" in body[0]["started_at"]
+
+
+# ── #2160: масштабируемый per-anchor watchdog для proxy-pool browser ──────────
+
+
+def test_cian_anchor_timeout_defaults() -> None:
+    """Формула watchdog даёт worst-case ~1580s при дефолтных параметрах.
+
+    4 room-buckets × 3 pages × (5s delay + 70s serp-fetch) + 10 detail × 50s + 180s houses
+    = 900 + 500 + 180 = 1580s. Фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинил бы якорь в
+    SERP-фазе (#2160): каждый SERP-фетч через camoufox 13-45s.
+    """
+    t = scrape_pipeline._cian_anchor_timeout_s(
+        pages_per_anchor=3,
+        request_delay_sec=5.0,
+        detail_top_n=10,
+        enrich_houses=True,
+    )
+    assert t == pytest.approx(1580.0)
+    assert t > scrape_pipeline.ANCHOR_TIMEOUT_SEC
+
+
+def test_cian_anchor_timeout_no_houses() -> None:
+    """enrich_houses=False убирает houses-бюджет (180s) из формулы."""
+    t = scrape_pipeline._cian_anchor_timeout_s(
+        pages_per_anchor=3,
+        request_delay_sec=5.0,
+        detail_top_n=10,
+        enrich_houses=False,
+    )
+    assert t == pytest.approx(1400.0)  # 900 + 500 + 0
+
+
+def test_cian_anchor_timeout_floor_at_anchor_timeout_sec() -> None:
+    """Малые параметры не опускают таймаут ниже ANCHOR_TIMEOUT_SEC (max-ветка)."""
+    t = scrape_pipeline._cian_anchor_timeout_s(
+        pages_per_anchor=0,
+        request_delay_sec=0.0,
+        detail_top_n=0,
+        enrich_houses=False,
+    )
+    assert t == float(scrape_pipeline.ANCHOR_TIMEOUT_SEC)
+
+
+def test_cian_anchor_timeout_parity_with_scraper_kit() -> None:
+    """Strangler-инвариант: backend-хелпер ≡ scraper_kit-хелпер (обе копии в синхроне)."""
+    from scraper_kit.orchestration.pipeline import (
+        _cian_anchor_timeout_s as kit_timeout_s,
+    )
+
+    for args in (
+        dict(pages_per_anchor=3, request_delay_sec=5.0, detail_top_n=10, enrich_houses=True),
+        dict(pages_per_anchor=5, request_delay_sec=9.0, detail_top_n=20, enrich_houses=False),
+        dict(pages_per_anchor=0, request_delay_sec=0.0, detail_top_n=0, enrich_houses=False),
+    ):
+        assert scrape_pipeline._cian_anchor_timeout_s(**args) == kit_timeout_s(**args)

tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py

@@ -235,6 +235,44 @@ _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5
 # прерывают enrichment-фазу. Listings из SEARCH+SAVE сохраняются всегда.
 _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3
 
+# #2160: константы для расчёта watchdog-таймаута Cian city sweep. При
+# USE_PROXY_POOL_BROWSER=true каждый SERP-фетч идёт через camoufox с relaunch при смене
+# прокси (page.goto timeout 60s + overhead) = 13-45s/страница, а якорь cian = 4 room-buckets
+# × pages_per_anchor SERP-фетчей + detail_top_n detail-фетчей + houses-enrich. Фиксированный
+# ANCHOR_TIMEOUT_SEC=240 гильотинит якорь ещё в SERP-фазе → save_listings не успевает,
+# lots_fetched=0, после N подряд → mark_banned. Таймаут масштабируется:
+#   _cian_anchor_timeout = max(
+#       ANCHOR_TIMEOUT_SEC,
+#       _CIAN_ROOM_BUCKETS × pages_per_anchor × (request_delay_sec + _CIAN_PER_SERP_FETCH_S)
+#       + detail_top_n × _CIAN_PER_DETAIL_S
+#       + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0))
+# Пример (дефолты): 4×3×(5+70) + 10×50 + 180 = 900 + 500 + 180 = 1580s/anchor (worst-case,
+# это watchdog от зависания, не ожидаемая длительность).
+_CIAN_ROOM_BUCKETS: int = 4  # fetch_around_multi_room итерирует ((1,),(2,),(3,),(4,)),
+#   см. providers/cian/serp.py:215
+_CIAN_PER_SERP_FETCH_S: float = 70.0  # browser fetch через пул: page.goto 60s + camoufox relaunch
+_CIAN_PER_DETAIL_S: float = 50.0  # секунд на один detail-fetch (как avito)
+_CIAN_HOUSES_BUDGET_S: float = 180.0  # бюджет на house-enrich фазу (как avito)
+
+
+def _cian_anchor_timeout_s(
+    pages_per_anchor: int,
+    request_delay_sec: float,
+    detail_top_n: int,
+    enrich_houses: bool,
+) -> float:
+    """Масштабируемый watchdog-таймаут одного Cian anchor'а (секунды).
+
+    Чистая функция для тестируемости формулы (#2160). Никогда не опускается ниже
+    ANCHOR_TIMEOUT_SEC (max-ветка): при малых параметрах фиксированный минимум остаётся.
+    """
+    return max(
+        float(ANCHOR_TIMEOUT_SEC),
+        _CIAN_ROOM_BUCKETS * pages_per_anchor * (request_delay_sec + _CIAN_PER_SERP_FETCH_S)
+        + detail_top_n * _CIAN_PER_DETAIL_S
+        + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0),
+    )
+
 # Default anchors ЕКБ — 5 точек покрытия города
 EKB_ANCHORS: list[tuple[float, float, str]] = [
     (56.8400, 60.6050, "Центр"),
@@ -2168,6 +2206,27 @@ async def run_cian_city_sweep(
     consecutive_failures = 0
     cian_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep
 
+    # #2160: масштабируемый watchdog-таймаут для cian anchor'а. При proxy-pool browser каждый
+    # SERP-фетч 13-45s (camoufox relaunch), фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинит
+    # якорь в SERP-фазе. См. _cian_anchor_timeout_s + константы выше.
+    _cian_anchor_timeout = _cian_anchor_timeout_s(
+        pages_per_anchor, request_delay_sec, detail_top_n, enrich_houses
+    )
+    logger.info(
+        "cian-sweep run_id=%d: anchor_timeout=%.0fs "
+        "(base=%d, serp=%d×%d×(%.0f+%.0f)s + detail_top_n=%d×%.0fs + houses=%.0fs)",
+        run_id,
+        _cian_anchor_timeout,
+        ANCHOR_TIMEOUT_SEC,
+        _CIAN_ROOM_BUCKETS,
+        pages_per_anchor,
+        request_delay_sec,
+        _CIAN_PER_SERP_FETCH_S,
+        detail_top_n,
+        _CIAN_PER_DETAIL_S,
+        _CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0,
+    )
+
     try:
         for idx, (lat, lon, name) in enumerate(_anchors, start=1):
             # Cooperative cancel перед каждым anchor
@@ -2435,7 +2494,7 @@ async def run_cian_city_sweep(
                 )
 
             try:
-                await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC)
+                await asyncio.wait_for(_cian_anchor_phases(), timeout=_cian_anchor_timeout)
             except TimeoutError:
                 logger.warning(
                     "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
@@ -2446,7 +2505,7 @@ async def run_cian_city_sweep(
                     name,
                     lat,
                     lon,
-                    ANCHOR_TIMEOUT_SEC,
+                    _cian_anchor_timeout,
                 )
                 counters.errors_count += 1
                 consecutive_failures += 1
@@ -2467,7 +2526,7 @@ async def run_cian_city_sweep(
                         db,
                         run_id,
                         f"cian sweep aborted: {consecutive_failures} consecutive "
-                        f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)",
+                        f"anchor failures (last: anchor timeout {_cian_anchor_timeout:.0f}s)",
                         counters.to_dict(),
                     )
                     return counters