fix(tradein/yandex): инкрементальный per-combo save + bounded gate-timeout (анти-зомби city_sweep)

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

@@ -1167,8 +1167,9 @@ async def run_avito_newbuilding_sweep(
 class YandexCitySweepCounters:
     """Aggregate counters для Yandex.Недвижимость city sweep run.
 
-    Фазы: SERP (fetch_around_multi_room) → save_listings → address-enrich.
+    Фазы: SERP (fetch_around_multi_room) → save_listings инкрементально → address-enrich.
     address_* — результат T10-обогащения (detail <title> → полный адрес с номером дома).
+    combos_skipped — сколько (room×price) combo пропущено из-за таймаутов gate-API.
     """
 
     anchors_total: int = 0
@@ -1177,6 +1178,7 @@ class YandexCitySweepCounters:
     lots_inserted: int = 0
     lots_updated: int = 0
     price_history_rows: int = 0
+    combos_skipped: int = 0
     address_attempted: int = 0
     address_enriched: int = 0
     address_failed: int = 0
@@ -1327,17 +1329,68 @@ async def run_yandex_city_sweep(
             # по ссылке → неправильные значения при asyncio scheduling.
             _lat, _lon, _name = lat, lon, name
 
+            # _combo_lots_accumulator: per-anchor мутируемый контейнер для on_combo.
+            # Захватывается по ссылке и в _yandex_anchor_phases и в _on_combo —
+            # избегаем B023 (loop-variable capture) через явный bind в default-arg.
+            _combo_lots_accumulator: list[ScrapedLot] = anchor_lots
+
             async def _yandex_anchor_phases(
                 _a_lat: float = _lat,
                 _a_lon: float = _lon,
                 _a_name: str = _name,
+                _al: list[ScrapedLot] = _combo_lots_accumulator,
             ) -> None:
-                """Все фазы одного anchor'а (SERP + address-enrich)."""
-                nonlocal anchor_lots, consecutive_failures
+                """Все фазы одного anchor'а (SERP + address-enrich).
+
+                FIX: инкрементальный save per-combo — on_combo вызывается scraper'ом после
+                каждого (room×price) combo, сохраняет порцию сразу в БД + обновляет heartbeat.
+                anchor_lots накапливается (через _al) для address-enrich и price-history.
+                """
+                nonlocal consecutive_failures
+
+                # ── Phase 1+2: SERP + инкрементальный save per-combo ──────
+                def _on_combo(
+                    combo_label: str,
+                    new_lots: list[ScrapedLot],
+                    _accumulator: list[ScrapedLot] = _al,
+                ) -> None:
+                    """Callback: вызывается scraper'ом после каждого combo с новыми лотами.
+
+                    Сохраняет порцию новых лотов немедленно (анти-зомби: данные в БД
+                    до конца anchor'а) и обновляет heartbeat для мониторинга прогресса.
+                    """
+                    _accumulator.extend(new_lots)
+                    counters.lots_fetched += len(new_lots)
+                    try:
+                        ins, upd = save_listings(db, new_lots, run_id=run_id)
+                        counters.lots_inserted += ins
+                        counters.lots_updated += upd
+                    except Exception as save_exc:
+                        counters.errors_count += 1
+                        logger.warning(
+                            "yandex-sweep run_id=%d combo %s: save_listings failed: %s",
+                            run_id,
+                            combo_label,
+                            save_exc,
+                        )
+                        try:
+                            db.rollback()
+                        except Exception:
+                            pass
+                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+                    logger.debug(
+                        "yandex-sweep run_id=%d combo %s: saved %d lots "
+                        "(ins=%d upd=%d total_fetched=%d)",
+                        run_id,
+                        combo_label,
+                        len(new_lots),
+                        counters.lots_inserted,
+                        counters.lots_updated,
+                        counters.lots_fetched,
+                    )
 
-                # ── Phase 1+2: SERP + save ─────────────────────────────────
                 async with YandexRealtyScraper() as scraper:
-                    anchor_lots = await scraper.fetch_around_multi_room(
+                    await scraper.fetch_around_multi_room(
                         _a_lat,
                         _a_lon,
                         radius_m,
@@ -1345,16 +1398,15 @@ async def run_yandex_city_sweep(
                         rooms_list=_rooms_list,
                         price_ranges=_price_ranges,
                         segments=_segments,
+                        on_combo=_on_combo,
                     )
-                counters.lots_fetched += len(anchor_lots)
-                if anchor_lots:
-                    inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
-                    counters.lots_inserted += inserted
-                    counters.lots_updated += updated
-                    # Price-history из gate price.previous/trend (graceful — провал
-                    # истории НЕ должен ронять сейв листингов).
+                # _al накоплен on_combo; save уже вызван per-combo.
+
+                # Price-history из gate price.previous/trend (graceful — провал
+                # истории НЕ должен ронять сейв листингов).
+                if _al:
                     try:
-                        counters.price_history_rows += record_yandex_price_history(db, anchor_lots)
+                        counters.price_history_rows += record_yandex_price_history(db, _al)
                     except Exception as ph_exc:
                         logger.warning(
                             "yandex-sweep run_id=%d: price-history failed: %s",
@@ -1365,20 +1417,21 @@ async def run_yandex_city_sweep(
                             db.rollback()
                         except Exception:
                             pass
+
                 consecutive_failures = 0
                 logger.info(
                     "yandex-sweep run_id=%d center-combos %s: fetched=%d ins=%d upd=%d",
                     run_id,
                     _a_name,
-                    len(anchor_lots),
+                    counters.lots_fetched,
                     counters.lots_inserted,
                     counters.lots_updated,
                 )
 
                 # ── Phase 3: address-enrich ────────────────────────────────
-                if not (enrich_address and anchor_lots):
+                if not (enrich_address and _al):
                     return
-                source_ids = [lot.source_id for lot in anchor_lots if lot.source_id]
+                source_ids = [lot.source_id for lot in _al if lot.source_id]
                 if not source_ids:
                     return
                 id_rows = (

tradein-mvp/backend/app/services/scrapers/browser_fetcher.py

@@ -40,12 +40,15 @@ class BrowserFetcher:
             html = await fetcher.fetch("https://example.com")
     """
 
-    def __init__(self, source: str = "avito") -> None:
+    def __init__(self, source: str = "avito", fetch_timeout_s: float = _HTTP_TIMEOUT_S) -> None:
         # source — логический источник ("avito"/"cian"/"yandex"/"domclick"). Сервер
         # роутит /fetch по нему на отдельный браузер+прокси, когда включён
         # FEATURE_BROWSER_POOL_ENABLED (Phase 1). При выключенном флаге source
         # игнорируется — поведение не меняется.
+        # fetch_timeout_s — таймаут httpx-клиента для POST /fetch. Yandex-путь передаёт
+        # 30s чтобы один проблемный combo занимал ≤30s×retries вместо 120s×retries.
         self._source = source
+        self._fetch_timeout_s = fetch_timeout_s
         self._client: httpx.AsyncClient | None = None
         self._endpoint: str | None = None
 
@@ -55,7 +58,7 @@ class BrowserFetcher:
         from app.core.config import settings
 
         self._endpoint = settings.browser_http_endpoint
-        self._client = httpx.AsyncClient(timeout=_HTTP_TIMEOUT_S)
+        self._client = httpx.AsyncClient(timeout=self._fetch_timeout_s)
         logger.info("BrowserFetcher: клиент создан, endpoint=%s", self._endpoint)
         return self
 

tradein-mvp/backend/app/services/scrapers/yandex_realty.py

@@ -83,6 +83,10 @@ _YANDEX_MIN_BRACKET = 50_000
 _GATE_PAGE_SIZE = 20
 _YANDEX_TARPIT_MAX_RETRIES: int = 2
 
+# Таймаут httpx-клиента BrowserFetcher для yandex-пути: 30s вместо глобального 120s.
+# Один проблемный combo занимает ≤30s×(1+_YANDEX_TARPIT_MAX_RETRIES)=90s вместо 360s.
+_YANDEX_BROWSER_FETCH_TIMEOUT_S: float = 30.0
+
 
 @dataclass
 class _CurlResponse:
@@ -431,8 +435,13 @@ class YandexRealtyScraper(BaseScraper):
         camoufox's real-browser fingerprint avoids the mobile-proxy IP tarpit
         escalation that raw system-curl over SOCKS5 triggers. Opening per-page is
         expensive and unnecessary.
+
+        fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S (30s) ограничивает потери
+        на один тайм-аутовый запрос: worst-case 30s×(1+retries)=90s на combo вместо 360s.
         """
-        self._browser = BrowserFetcher(source="yandex")
+        self._browser = BrowserFetcher(
+            source="yandex", fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S
+        )
         await self._browser.__aenter__()
         return self
 
@@ -662,6 +671,7 @@ class YandexRealtyScraper(BaseScraper):
         rooms_list: list[str] | None = None,
         price_ranges: list[tuple[int | None, int | None]] | None = None,
         segments: list[str] | None = None,
+        on_combo: Any = None,
         **_legacy_kwargs: Any,
     ) -> list[ScrapedLot]:
         """Fetch via segment x rooms x price-range combos; paginate each combo to totalPages.
@@ -672,9 +682,15 @@ class YandexRealtyScraper(BaseScraper):
         Legacy mode (rooms_list=None, price_ranges=None): single citywide sweep.
         segments: list of newFlat values, default ["NO"] (vtorichka only). Pass
             ["NO", "YES"] to sweep both vtorichka and novostroyki in one call.
+
+        on_combo: опциональный callback(combo_label: str, new_lots: list[ScrapedLot]) -> None.
+            Вызывается после каждого combo с новыми (не повторными) лотами из него.
+            Позволяет инкрементальный save: вызывающий код сохраняет new_lots сразу,
+            не дожидаясь конца sweep'а. Дубли между combo НЕ передаются повторно.
         """
         seen: dict[str, ScrapedLot] = {}
         _segments = segments or ["NO"]
+        combos_skipped = 0
 
         if rooms_list is None and price_ranges is None:
             combos: list[tuple[str | None, int | None, int | None]] = [(None, None, None)]
@@ -688,6 +704,8 @@ class YandexRealtyScraper(BaseScraper):
             for rooms, price_min, price_max in combos:
                 combo_label = f"{_seg}/{_combo_label(rooms, price_min, price_max)}"
                 total_pages: int | None = None
+                combo_new_lots: list[ScrapedLot] = []
+                combo_skipped = False
 
                 for page in range(1, max_pages + 1):
                     if page == 1:
@@ -709,7 +727,9 @@ class YandexRealtyScraper(BaseScraper):
                                 break
                             if resp.status_code == 0:  # type: ignore[union-attr]
                                 logger.warning(
-                                    "yandex gate: tarpit combo=%s page=1 -- rotating", combo_label
+                                    "yandex gate: tarpit combo=%s page=1 attempt=%d -- rotating",
+                                    combo_label,
+                                    _attempt + 1,
                                 )
                                 await self._rotate_ip()
                                 await asyncio.sleep(2)
@@ -740,13 +760,18 @@ class YandexRealtyScraper(BaseScraper):
                             break
 
                         if payload_p1 is None:
-                            logger.debug(
-                                "yandex gate combo [%s] page=1: failed -- skipping", combo_label
+                            logger.warning(
+                                "yandex gate combo [%s] page=1: all retries exhausted -- skipping",
+                                combo_label,
                             )
+                            combo_skipped = True
+                            combos_skipped += 1
                             break
 
                         result = _extract_gate_data(payload_p1)
                         if result is None:
+                            combo_skipped = True
+                            combos_skipped += 1
                             break
                         _entities_p1, pager_p1 = result
                         total_pages = min(
@@ -764,6 +789,7 @@ class YandexRealtyScraper(BaseScraper):
                             key = lot.source_id or lot.source_url
                             if key and key not in seen:
                                 seen[key] = lot
+                                combo_new_lots.append(lot)
                         await self.sleep_between_requests()
                         if total_pages <= 1:
                             break
@@ -793,13 +819,25 @@ class YandexRealtyScraper(BaseScraper):
                         key = lot.source_id or lot.source_url
                         if key and key not in seen:
                             seen[key] = lot
+                            combo_new_lots.append(lot)
+
+                # Инкрементальный save: вызываем on_combo если есть новые лоты.
+                # Скипнутые combo (combo_skipped=True) не вызывают on_combo.
+                if on_combo is not None and combo_new_lots and not combo_skipped:
+                    try:
+                        on_combo(combo_label, combo_new_lots)
+                    except Exception:
+                        logger.exception(
+                            "yandex gate: on_combo callback failed combo=%s", combo_label
+                        )
 
         logger.info(
-            "yandex gate aggregate: %d unique lots (%d combos x %d segments=%s)",
+            "yandex gate aggregate: %d unique lots (%d combos x %d segments=%s, skipped=%d)",
             len(seen),
             len(combos),
             len(_segments),
             _segments,
+            combos_skipped,
         )
         return list(seen.values())
 

tradein-mvp/backend/tests/test_yandex_city_sweep.py

@@ -153,6 +153,7 @@ def test_counters_defaults() -> None:
     c = YandexCitySweepCounters()
     assert c.anchors_total == 0
     assert c.lots_fetched == 0
+    assert c.combos_skipped == 0
     assert c.address_attempted == 0
     assert c.address_enriched == 0
     assert c.address_failed == 0
@@ -176,7 +177,12 @@ def test_counters_to_dict_all_keys() -> None:
 async def test_sweep_iterates_all_anchors_and_aggregates(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """fetch_around_multi_room вызывается для каждого anchor, counters агрегируются."""
+    """fetch_around_multi_room вызывается для каждого anchor, counters агрегируются.
+
+    FIX: fake_fetch_multi вызывает on_combo (инкрементальный save per-combo) —
+    имитирует реальное поведение scraper'а. save_listings должен вызываться
+    on_combo'м, а не одним батчем в конце anchor'а.
+    """
     serp_calls: list[str] = []
     save_calls: list[int] = []
 
@@ -186,10 +192,15 @@ async def test_sweep_iterates_all_anchors_and_aggregates(
         lon: float,
         radius_m: int = 1000,
         max_pages: int = 2,
+        on_combo: Any = None,
         **_: Any,
     ) -> list[ScrapedLot]:
         serp_calls.append(f"{lat:.4f}")
-        return [_fake_lot(f"{lat}-{i}") for i in range(3)]
+        lots = [_fake_lot(f"{lat}-{i}") for i in range(3)]
+        # Имитируем один combo — вызываем on_combo если предоставлен.
+        if on_combo is not None:
+            on_combo(f"vtorichka/combo_{lat:.4f}", lots)
+        return lots
 
     def fake_save(
         _db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
@@ -212,6 +223,8 @@ async def test_sweep_iterates_all_anchors_and_aggregates(
     )
 
     assert len(serp_calls) == len(TEST_ANCHORS)
+    # save вызывается инкрементально через on_combo — по одному разу на anchor
+    # (fake имитирует один combo per anchor).
     assert len(save_calls) == len(TEST_ANCHORS)
     assert counters.anchors_total == len(TEST_ANCHORS)
     assert counters.anchors_done == len(TEST_ANCHORS)
@@ -284,8 +297,12 @@ async def test_address_enrich_fills_counters(
         lon: float,
         radius_m: int = 1000,
         max_pages: int = 2,
+        on_combo: Any = None,
         **_: Any,
     ) -> list[ScrapedLot]:
+        # Вызываем on_combo чтобы anchor_lots накопился (инкрементальный save).
+        if on_combo is not None:
+            on_combo("vtorichka/combo_enrich_test", lots)
         return lots
 
     monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
@@ -343,9 +360,13 @@ async def test_address_enrich_http_error_does_not_abort(
         lon: float,
         radius_m: int = 1000,
         max_pages: int = 2,
+        on_combo: Any = None,
         **_: Any,
     ) -> list[ScrapedLot]:
-        return [_fake_lot("id-err")]
+        lots = [_fake_lot("id-err")]
+        if on_combo is not None:
+            on_combo("vtorichka/combo_http_err", lots)
+        return lots
 
     monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
     monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
@@ -498,11 +519,16 @@ async def test_non_consecutive_failures_do_not_abort(
         lon: float,
         radius_m: int = 1000,
         max_pages: int = 2,
+        on_combo: Any = None,
         **_: Any,
     ) -> list[ScrapedLot]:
         if _name_of(lat, anchors) in failing:
             raise RuntimeError("transient")
-        return [_fake_lot(f"{lat}-0")]
+        lots = [_fake_lot(f"{lat}-0")]
+        # Вызываем on_combo для инкрементального save.
+        if on_combo is not None:
+            on_combo(f"vtorichka/combo_{lat:.4f}", lots)
+        return lots
 
     monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch)
     monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
@@ -689,8 +715,10 @@ async def test_prod_mode_uses_center_anchor_with_combos(
         max_pages: int = 2,
         rooms_list: list | None = None,
         price_ranges: list | None = None,
+        on_combo: Any = None,
         **_: Any,
     ) -> list[ScrapedLot]:
+        lots = [_fake_lot("center-1"), _fake_lot("center-2")]
         calls.append(
             {
                 "lat": lat,
@@ -701,7 +729,10 @@ async def test_prod_mode_uses_center_anchor_with_combos(
                 "max_pages": max_pages,
             }
         )
-        return [_fake_lot("center-1"), _fake_lot("center-2")]
+        # Имитируем on_combo (инкрементальный save) — один combo.
+        if on_combo is not None:
+            on_combo("vtorichka/center_combo", lots)
+        return lots
 
     monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
     monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
@@ -745,7 +776,7 @@ async def test_prod_mode_uses_center_anchor_with_combos(
     # max_pages forwarded from pages_per_anchor
     assert call["max_pages"] == 3
 
-    # lots aggregated correctly
+    # lots aggregated correctly via on_combo
     assert counters.lots_fetched == 2
     assert counters.lots_inserted == 2
     assert fake.done is not None
@@ -895,3 +926,133 @@ async def test_explicit_anchor_mode_uses_anchor_timeout_sec(
         assert (
             t == ANCHOR_TIMEOUT_SEC
         ), f"Expected ANCHOR_TIMEOUT_SEC={ANCHOR_TIMEOUT_SEC}s, got {t}s"
+
+
+# ── FIX A: инкрементальный save per-combo ────────────────────────────────────
+
+
+async def test_incremental_save_called_per_combo(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """save_listings вызывается on_combo per-combo, не одним батчем в конце anchor'а.
+
+    Имитируем 3 combo через on_combo — save_listings должен быть вызван 3 раза
+    (по одному на combo) с порциями лотов. Проверяем, что при краше anchor'а
+    уже сохранённые combo не теряются (данные в БД до finish anchor'а).
+    """
+    save_batches: list[list[str]] = []
+
+    async def fake_fetch_multi(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        on_combo: Any = None,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        all_lots = []
+        for combo_idx in range(3):
+            combo_lots = [_fake_lot(f"{lat}-combo{combo_idx}-lot0")]
+            all_lots.extend(combo_lots)
+            if on_combo is not None:
+                on_combo(f"vtorichka/combo_{combo_idx}", combo_lots)
+        return all_lots
+
+    def fake_save(
+        _db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
+    ) -> tuple[int, int]:
+        save_batches.append([lot.source_id for lot in lots if lot.source_id])
+        return len(lots), 0
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=MagicMock(),
+        run_id=30,
+        anchors=TEST_ANCHORS[:1],
+        pages_per_anchor=2,
+        request_delay_sec=0.0,
+        enrich_address=False,
+    )
+
+    # save_listings вызван ровно 3 раза (по одному на combo), не одним батчем.
+    assert (
+        len(save_batches) == 3
+    ), f"Ожидалось 3 save вызова (per-combo), получено {len(save_batches)}: {save_batches}"
+    # Каждый batch содержит ровно 1 лот (размер combo).
+    for batch in save_batches:
+        assert len(batch) == 1, f"Каждый combo-batch должен иметь 1 лот, получено {batch}"
+
+    # Суммарные counters корректны.
+    assert counters.lots_fetched == 3
+    assert counters.lots_inserted == 3
+    assert fake.done is not None
+
+    # heartbeat вызывался после каждого combo (через on_combo).
+    assert (
+        len(fake.heartbeats) >= 3
+    ), f"Ожидалось ≥3 heartbeat обновлений (per-combo), получено {len(fake.heartbeats)}"
+
+
+# ── FIX B: bounded retry — скипает combo после exhaustion ────────────────────
+
+
+async def test_bounded_retry_combo_skipped_logged(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """on_combo НЕ вызывается если combo пропущен (все retry page-1 исчерпаны).
+
+    Имитируем sweep с двумя combos: первый скипается (on_combo не вызван),
+    второй успешен (on_combo вызван). Проверяем, что save_listings вызван
+    только для успешного combo, а не скипнутого.
+    """
+    save_batches: list[list[str]] = []
+    on_combo_calls: list[str] = []
+
+    async def fake_fetch_multi(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        on_combo: Any = None,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        # Combo 0: скипается — on_combo НЕ вызывается (симулирует exhausted retries).
+        # Combo 1: успешен — on_combo вызывается с лотом.
+        success_lot = _fake_lot(f"{lat}-success")
+        if on_combo is not None:
+            on_combo_calls.append("combo_1_success")
+            on_combo("vtorichka/combo_1", [success_lot])
+        return [success_lot]
+
+    def fake_save(
+        _db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
+    ) -> tuple[int, int]:
+        save_batches.append([lot.source_id for lot in lots if lot.source_id])
+        return len(lots), 0
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=MagicMock(),
+        run_id=31,
+        anchors=TEST_ANCHORS[:1],
+        pages_per_anchor=2,
+        request_delay_sec=0.0,
+        enrich_address=False,
+    )
+
+    # Только успешный combo вызвал on_combo и save.
+    assert len(on_combo_calls) == 1
+    assert len(save_batches) == 1
+    assert counters.lots_fetched == 1
+    assert counters.lots_inserted == 1
+    assert fake.done is not None

tradein-mvp/backend/tests/test_yandex_realty_serp.py

@@ -1176,3 +1176,119 @@ async def test_walk_price_range_open_bracket_no_split():
     assert all(
         pm is None for pm in seen_prices
     ), f"Открытый брекет: priceMax всегда None, got {seen_prices}"
+
+
+# ---------------------------------------------------------------------------
+# PART J: fetch_around_multi_room on_combo callback (FIX A — incremental save)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_fetch_around_multi_room_on_combo_called_per_combo():
+    """on_combo вызывается после каждого combo с новыми (деduped) лотами.
+
+    2 combo (rooms_list=['1'], 2 price_ranges): каждый имеет уникальный лот.
+    on_combo должен быть вызван дважды — по одному разу на combo.
+    """
+    s = YandexRealtyScraper()
+
+    entity_0 = dict(_ENTITY_FULL, offerId="combo0_lot", url="//realty.yandex.ru/offer/combo0_lot")
+    entity_1 = dict(_ENTITY_FULL, offerId="combo1_lot", url="//realty.yandex.ru/offer/combo1_lot")
+    pager = {"page": 0, "pageSize": 20, "totalItems": 1, "totalPages": 1}
+
+    call_idx = 0
+    payloads = [
+        _make_gate_payload([entity_0], pager),
+        _make_gate_payload([entity_1], pager),
+    ]
+
+    async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse:
+        nonlocal call_idx
+        idx = call_idx
+        call_idx += 1
+        if idx < len(payloads):
+            return _CurlResponse(status_code=200, text=json.dumps(payloads[idx]))
+        return _CurlResponse(status_code=200, text=json.dumps(_make_gate_payload([])))
+
+    on_combo_calls: list[tuple[str, list]] = []
+
+    def on_combo(label: str, new_lots: list) -> None:
+        on_combo_calls.append((label, list(new_lots)))
+
+    with patch.object(s, "_http_get", side_effect=fake_http_get):
+        with patch.object(s, "sleep_between_requests", new_callable=AsyncMock):
+            lots = await s.fetch_around_multi_room(
+                lat=0.0,
+                lon=0.0,
+                rooms_list=["1"],
+                price_ranges=[(None, 5_000_000), (5_000_000, 10_000_000)],
+                max_pages=1,
+                on_combo=on_combo,
+            )
+
+    # Два уникальных лота, два combo.
+    assert len(lots) == 2
+    # on_combo вызван ровно дважды — по одному на combo.
+    assert len(on_combo_calls) == 2, f"Ожидалось 2 on_combo вызова, получено {len(on_combo_calls)}"
+    # Каждый on_combo получил ровно один новый лот.
+    for _label, combo_lots in on_combo_calls:
+        assert len(combo_lots) == 1
+    # Конкретные source_id в правильных batch'ах.
+    all_ids = {lot.source_id for _, combo_lots in on_combo_calls for lot in combo_lots}
+    assert all_ids == {"combo0_lot", "combo1_lot"}
+
+
+@pytest.mark.asyncio
+async def test_fetch_around_multi_room_on_combo_not_called_for_failed_combo():
+    """on_combo НЕ вызывается для combo, у которого все page-1 retries исчерпаны.
+
+    FIX B: bounded retry — если status_code=0 на всех попытках page=1,
+    combo пропускается. on_combo не должен быть вызван для пропущенного combo.
+    Успешный combo (второй) — on_combo должен сработать.
+    """
+    s = YandexRealtyScraper()
+
+    entity_ok = dict(_ENTITY_FULL, offerId="ok_lot", url="//realty.yandex.ru/offer/ok_lot")
+    pager = {"page": 0, "pageSize": 20, "totalItems": 1, "totalPages": 1}
+
+    call_idx = 0
+
+    async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse:
+        nonlocal call_idx
+        call_idx += 1
+        # Первые (1 + _YANDEX_TARPIT_MAX_RETRIES) запросы — это retries для combo[0] → fail.
+        from app.services.scrapers.yandex_realty import _YANDEX_TARPIT_MAX_RETRIES
+
+        if call_idx <= 1 + _YANDEX_TARPIT_MAX_RETRIES:
+            return _CurlResponse(status_code=0, text="")
+        # Следующий запрос — combo[1] page=1 → success.
+        return _CurlResponse(
+            status_code=200, text=json.dumps(_make_gate_payload([entity_ok], pager))
+        )
+
+    on_combo_calls: list[str] = []
+
+    def on_combo(label: str, new_lots: list) -> None:
+        on_combo_calls.append(label)
+
+    with patch.object(s, "_http_get", side_effect=fake_http_get):
+        with patch.object(s, "_rotate_ip", new_callable=AsyncMock):
+            with patch("asyncio.sleep", new_callable=AsyncMock):
+                with patch.object(s, "sleep_between_requests", new_callable=AsyncMock):
+                    lots = await s.fetch_around_multi_room(
+                        lat=0.0,
+                        lon=0.0,
+                        rooms_list=["1"],
+                        price_ranges=[(None, 5_000_000), (5_000_000, 10_000_000)],
+                        max_pages=1,
+                        on_combo=on_combo,
+                    )
+
+    # combo[0] скипнут, combo[1] успешен → 1 лот в итоге.
+    assert len(lots) == 1
+    assert lots[0].source_id == "ok_lot"
+    # on_combo вызван только для успешного combo, не для скипнутого.
+    assert len(on_combo_calls) == 1, (
+        f"on_combo должен быть вызван только 1 раз (успешный combo), "
+        f"получено {len(on_combo_calls)}: {on_combo_calls}"
+    )