fix(tradein/scrapers): wire proxy_pool into Avito legacy pipeline, rotate on ban (#2202) (#2328)

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

@@ -42,7 +42,7 @@ from sqlalchemy.orm import Session
 
 from app.core.config import settings
 from app.core.shutdown import shutdown_requested
-from app.services import scrape_runs
+from app.services import proxy_pool, scrape_runs
 from app.services.scrapers.avito import AvitoScraper
 from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
 from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
@@ -250,6 +250,188 @@ def _avito_proxies() -> dict[str, str] | None:
     return {"http": url, "https": url} if url else None
 
 
+def _mask_proxy_url(url: str) -> str:
+    """Скрыть пароль в proxy-url для логов (scheme://user:***@host)."""
+    if "@" not in url or "//" not in url:
+        return url
+    scheme, rest = url.split("//", 1)
+    creds, host = rest.split("@", 1)
+    if ":" in creds:
+        user, _pwd = creds.split(":", 1)
+        creds = f"{user}:***"
+    return f"{scheme}//{creds}@{host}"
+
+
+def _pool_proxy_dict(
+    db: Session, provider: str, *, run_id: int | None = None
+) -> tuple[dict[str, str] | None, proxy_pool.ProxyLease | None]:
+    """#2202: health-filtered lease из scrape_proxies вместо статического env-прокси.
+
+    За флагом settings.use_proxy_pool_curl (ship-dark, default False — прод не
+    меняется, пока флаг не включён явно). Флаг off ИЛИ пул пуст/ошибка acquire →
+    fallback на статический _avito_proxies() (settings.scraper_proxy_url), поведение
+    как раньше — сбор никогда не падает из-за проблем пула.
+
+    run_id — id текущего scrape_runs-прогона (если он есть у вызывающего, см.
+    run_avito_city_sweep/run_avito_newbuilding_sweep). Прокидывается в
+    proxy_pool.acquire(), чтобы scrape_proxies.leased_by нёс реальный run_id для
+    диагностики банов вместо NON_RUN_LEASE_MARKER. None (default) — прежнее
+    поведение для вызывающих без run_id в scope (run_avito_pipeline standalone).
+
+    Returns (proxies-dict для curl_cffi AsyncSession, lease | None). lease нужен
+    вызывающему для release+mark_health в конце run (см. _release_pool_lease) и для
+    последующей ротации на бан (см. _rotate_pool_proxy).
+    """
+    # getattr-defensive: некоторые тестовые/adapter settings-объекты (SimpleNamespace)
+    # не несут этого поля — трактуем отсутствие как off (то же ship-dark поведение).
+    if not getattr(settings, "use_proxy_pool_curl", False):
+        return _avito_proxies(), None
+    try:
+        lease = proxy_pool.acquire(db, provider, run_id=run_id)
+    except Exception:
+        logger.warning(
+            "pipeline: proxy_pool.acquire(%s) failed — fallback to env proxy",
+            provider,
+            exc_info=True,
+        )
+        lease = None
+    if lease is None:
+        logger.warning(
+            "pipeline: proxy pool empty/unhealthy for provider=%s — fallback to env proxy",
+            provider,
+        )
+        return _avito_proxies(), None
+    logger.info(
+        "pipeline: using pool proxy id=%d provider=%s url=%s for this run",
+        lease.id,
+        provider,
+        _mask_proxy_url(lease.url),
+    )
+    return {"http": lease.url, "https": lease.url}, lease
+
+
+def _release_pool_lease(db: Session, lease: proxy_pool.ProxyLease | None, ok: bool) -> None:
+    """Best-effort release+mark_health в конце run (#2202) — проблема пула не должна
+    ронять уже собранные данные. lease=None (пул выключен/не выдал прокси) → no-op.
+    """
+    if lease is None:
+        return
+    try:
+        proxy_pool.mark_health(db, lease.id, ok)
+    except Exception:
+        logger.warning("pipeline: mark_health failed for lease id=%d", lease.id, exc_info=True)
+    try:
+        proxy_pool.release(db, lease.id)
+    except Exception:
+        logger.warning("pipeline: release failed for lease id=%d", lease.id, exc_info=True)
+
+
+async def _rotate_pool_proxy(
+    db: Session,
+    provider: str,
+    session: AsyncSession,
+    current_lease: proxy_pool.ProxyLease | None,
+) -> proxy_pool.ProxyLease | None:
+    """#2202: переключиться на ДРУГОЙ здоровый прокси из пула при бане (не changeip
+    одного и того же IP). Помечает current_lease нездоровым (mark_health ok=False),
+    затем acquire()-ит следующий свободный здоровый прокси, и ТОЛЬКО ПОСЛЕ этого
+    освобождает current_lease (release). Порядок здесь принципиален (#2202 MAJOR-фикс,
+    review-репро): release ДО acquire() убирает current_lease из leased_by-фильтра
+    немедленно, и один бан поднимает consecutive_fails только до 1 (порог исключения —
+    proxy_pool.MAX_CONSECUTIVE_FAILS=3) — только что забаненный прокси тут же снова
+    проходит health-фильтр и acquire() может вернуть его же обратно (ORDER BY
+    last_ok_at NULLS LAST, id — никак не гарантирует, что он не окажется первым).
+    Пока current_lease НЕ освобождён, acquire()'s `leased_by IS NULL` условие
+    гарантированно его исключает — никакой отдельный exclude_id-параметр не нужен.
+
+    Успешная ротация мутирует session.proxies — curl_cffi AsyncSession читает
+    self.proxies заново на каждый request (не кэширует на конструкторе), так что
+    переприсвоение атрибута действует на следующий вызов без пересоздания сессии.
+
+    Returns новый lease, или None если пул пуст/ошибка (вызывающий код в этом случае
+    падает обратно на changeip-ротацию одного IP — см. _rotate_avito_proxy).
+    current_lease освобождается в обоих случаях (успех/провал) — он подтверждённо
+    нездоров, дальше эта функция за него не отвечает.
+    """
+    if current_lease is not None:
+        try:
+            proxy_pool.mark_health(db, current_lease.id, ok=False)
+        except Exception:
+            logger.warning(
+                "pipeline: mark_health(False) failed for lease id=%d",
+                current_lease.id,
+                exc_info=True,
+            )
+
+    try:
+        new_lease = proxy_pool.acquire(db, provider)
+    except Exception:
+        logger.warning(
+            "pipeline: proxy_pool.acquire(%s) failed during rotation", provider, exc_info=True
+        )
+        new_lease = None
+
+    if current_lease is not None:
+        try:
+            proxy_pool.release(db, current_lease.id)
+        except Exception:
+            logger.warning(
+                "pipeline: release failed for lease id=%d", current_lease.id, exc_info=True
+            )
+
+    if new_lease is None:
+        logger.warning(
+            "pipeline: proxy pool exhausted for provider=%s during rotation — "
+            "no healthy proxy to rotate to",
+            provider,
+        )
+        return None
+
+    session.proxies = {"http": new_lease.url, "https": new_lease.url}
+    logger.info(
+        "pipeline: ROTATED to pool proxy id=%d provider=%s url=%s (previous lease id=%s)",
+        new_lease.id,
+        provider,
+        _mask_proxy_url(new_lease.url),
+        current_lease.id if current_lease is not None else None,
+    )
+    return new_lease
+
+
+async def _rotate_avito_proxy(
+    db: Session,
+    session: AsyncSession | None,
+    current_lease: proxy_pool.ProxyLease | None,
+    *,
+    reason: str,
+    rotations_done: int,
+) -> tuple[bool, proxy_pool.ProxyLease | None]:
+    """#2202: единая точка ротации на бан для Avito.
+
+    Сначала пробует переключить прокси ПУЛА (health-фильтр, гарантированно другой
+    узел) — только когда settings.use_proxy_pool_curl включён И есть curl-сессия
+    (session is not None, т.е. не browser-mode). Если пул выключен, пуст, или это
+    browser-mode (session=None) — падает на старый changeip (смена exit-IP ОДНОГО
+    прокси-аккаунта, как раньше).
+
+    NB: пока делит общий rotations_done/avito_proxy_max_rotations бюджет с
+    changeip-путём — раздельные бюджеты (пул своё, changeip своё) можно завести
+    отдельным follow-up, если пул-ротации на практике окажутся частым событием.
+
+    Returns (rotated, lease для последующих вызовов — обновлён при успешной
+    пул-ротации, иначе тот же current_lease).
+    """
+    if getattr(settings, "use_proxy_pool_curl", False) and session is not None:
+        new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease)
+        if new_lease is not None:
+            return True, new_lease
+        logger.warning(
+            "pipeline: pool rotation unavailable (reason=%s) — falling back to changeip", reason
+        )
+    rotated = await _rotate_proxy_ip(reason=reason, rotations_done=rotations_done, source="avito")
+    return rotated, current_lease
+
+
 # ── Result dataclasses ──────────────────────────────────────────
 
 
@@ -332,6 +514,9 @@ async def run_avito_pipeline(
     browser_fetcher: BrowserFetcher | None = None
     own_session = False
     own_browser = False
+    # #2202: pool-lease текущей curl-сессии (None если browser-mode/pool off/shared
+    # session переданный извне — тогда владелец проксей не мы, не трогаем).
+    _lease: proxy_pool.ProxyLease | None = None
 
     scraper = AvitoScraper()
     if browser_mode:
@@ -343,12 +528,16 @@ async def run_avito_pipeline(
         scraper._browser = browser_fetcher
     else:
         own_session = shared_session is None
-        session = shared_session or AsyncSession(
-            impersonate="chrome120",
-            timeout=25,
-            headers=_CHROME_HEADERS,
-            proxies=_avito_proxies(),
-        )
+        if shared_session is not None:
+            session = shared_session
+        else:
+            _proxies, _lease = _pool_proxy_dict(db, "avito")
+            session = AsyncSession(
+                impersonate="chrome120",
+                timeout=25,
+                headers=_CHROME_HEADERS,
+                proxies=_proxies,
+            )
         scraper._cffi = session
 
     try:
@@ -429,7 +618,10 @@ async def run_avito_pipeline(
                         # После ротации сбрасываем счётчик и продолжаем — НЕ abort.
                         # Если ротации исчерпаны — abort enrichment (listings уже сохранены).
                         if enrich_rotations_done < settings.avito_proxy_max_rotations:
-                            rotated = await _rotate_proxy_ip(
+                            rotated, _lease = await _rotate_avito_proxy(
+                                db,
+                                session,
+                                _lease,
                                 reason=f"house consecutive={enrich_consecutive_blocks}",
                                 rotations_done=enrich_rotations_done,
                             )
@@ -437,7 +629,7 @@ async def run_avito_pipeline(
                             if rotated:
                                 enrich_consecutive_blocks = 0
                                 logger.info(
-                                    "pipeline:enrich ROTATE — IP changed, "
+                                    "pipeline:enrich ROTATE — proxy changed, "
                                     "reset consecutive blocks (rotation #%d/%d)",
                                     enrich_rotations_done,
                                     settings.avito_proxy_max_rotations,
@@ -545,7 +737,10 @@ async def run_avito_pipeline(
                         # После ротации сбрасываем счётчик — продолжаем detail-обход.
                         # Если ротации исчерпаны — abort detail-фазы (listings уже сохранены).
                         if enrich_rotations_done < settings.avito_proxy_max_rotations:
-                            rotated = await _rotate_proxy_ip(
+                            rotated, _lease = await _rotate_avito_proxy(
+                                db,
+                                session,
+                                _lease,
                                 reason=f"detail consecutive={enrich_consecutive_blocks}",
                                 rotations_done=enrich_rotations_done,
                             )
@@ -553,7 +748,7 @@ async def run_avito_pipeline(
                             if rotated:
                                 enrich_consecutive_blocks = 0
                                 logger.info(
-                                    "pipeline:enrich ROTATE — IP changed, "
+                                    "pipeline:enrich ROTATE — proxy changed, "
                                     "reset consecutive blocks (rotation #%d/%d)",
                                     enrich_rotations_done,
                                     settings.avito_proxy_max_rotations,
@@ -625,7 +820,10 @@ async def run_avito_pipeline(
                         if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
                             # #1790: попытка ротации перед abort'ом (общий бюджет run).
                             if enrich_rotations_done < settings.avito_proxy_max_rotations:
-                                rotated = await _rotate_proxy_ip(
+                                rotated, _lease = await _rotate_avito_proxy(
+                                    db,
+                                    session,
+                                    _lease,
                                     reason=f"house-detail consecutive={enrich_consecutive_blocks}",
                                     rotations_done=enrich_rotations_done,
                                 )
@@ -633,7 +831,7 @@ async def run_avito_pipeline(
                                 if rotated:
                                     enrich_consecutive_blocks = 0
                                     logger.info(
-                                        "pipeline:enrich ROTATE — IP changed, "
+                                        "pipeline:enrich ROTATE — proxy changed, "
                                         "reset consecutive blocks (rotation #%d/%d)",
                                         enrich_rotations_done,
                                         settings.avito_proxy_max_rotations,
@@ -696,6 +894,8 @@ async def run_avito_pipeline(
             await session.close()
         if own_browser and browser_fetcher is not None:
             await browser_fetcher.__aexit__(None, None, None)
+        if own_session:
+            _release_pool_lease(db, _lease, ok=not enrichment_aborted)
 
 
 @dataclass
@@ -776,6 +976,8 @@ async def run_avito_city_sweep(
     browser_mode = settings.scraper_fetch_mode == "browser"
     # #1790: общий бюджет IP-ротаций на весь sweep (разделяется по всем anchor'ам).
     sweep_rotations_done = 0
+    # #2202: pool-lease текущей curl-сессии (None в browser-mode/pool off).
+    _lease: proxy_pool.ProxyLease | None = None
 
     async with AsyncExitStack() as stack:
         session: AsyncSession | None = None
@@ -783,12 +985,13 @@ async def run_avito_city_sweep(
         if browser_mode:
             shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito"))
         else:
+            _proxies, _lease = _pool_proxy_dict(db, "avito", run_id=run_id)
             session = await stack.enter_async_context(
                 AsyncSession(
                     impersonate="chrome120",
                     timeout=25,
                     headers=_CHROME_HEADERS,
-                    proxies=_avito_proxies(),
+                    proxies=_proxies,
                 )
             )
         try:
@@ -844,7 +1047,7 @@ async def run_avito_city_sweep(
                     НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что
                     TimeoutError из wait_for теряет только detail-счётчики, но не SERP.
                     """
-                    nonlocal all_touched_house_ids, sweep_rotations_done
+                    nonlocal all_touched_house_ids, sweep_rotations_done, _lease
 
                     # ── Phase 1+2: SERP + save ─────────────────────────────────
                     scraper = AvitoScraper()
@@ -954,7 +1157,10 @@ async def run_avito_city_sweep(
                                     >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
                                     and sweep_rotations_done < settings.avito_proxy_max_rotations
                                 ):
-                                    rotated = await _rotate_proxy_ip(
+                                    rotated, _lease = await _rotate_avito_proxy(
+                                        db,
+                                        session,
+                                        _lease,
                                         reason=(
                                             f"city-sweep house "
                                             f"consecutive={sweep_consecutive_house_blocks}"
@@ -966,7 +1172,7 @@ async def run_avito_city_sweep(
                                         sweep_consecutive_house_blocks = 0
                                         logger.info(
                                             "city-sweep run_id=%d: house ROTATE — "
-                                            "IP changed, reset consecutive "
+                                            "proxy changed, reset consecutive "
                                             "(rotation #%d/%d)",
                                             run_id,
                                             sweep_rotations_done,
@@ -1077,7 +1283,10 @@ async def run_avito_city_sweep(
                                 if consecutive_blocks >= 3:
                                     # #1790: попытка ротации перед abort'ом.
                                     if sweep_rotations_done < settings.avito_proxy_max_rotations:
-                                        rotated = await _rotate_proxy_ip(
+                                        rotated, _lease = await _rotate_avito_proxy(
+                                            db,
+                                            session,
+                                            _lease,
                                             reason=(
                                                 f"city-sweep detail "
                                                 f"consecutive={consecutive_blocks}"
@@ -1089,7 +1298,7 @@ async def run_avito_city_sweep(
                                             consecutive_blocks = 0
                                             logger.info(
                                                 "city-sweep run_id=%d: detail ROTATE — "
-                                                "IP changed, reset consecutive "
+                                                "proxy changed, reset consecutive "
                                                 "(rotation #%d/%d)",
                                                 run_id,
                                                 sweep_rotations_done,
@@ -1177,7 +1386,10 @@ async def run_avito_city_sweep(
                                         and sweep_rotations_done
                                         < settings.avito_proxy_max_rotations
                                     ):
-                                        rotated = await _rotate_proxy_ip(
+                                        rotated, _lease = await _rotate_avito_proxy(
+                                            db,
+                                            session,
+                                            _lease,
                                             reason=(
                                                 "city-sweep house(detail) "
                                                 f"consecutive="
@@ -1190,7 +1402,7 @@ async def run_avito_city_sweep(
                                             sweep_consecutive_detail_house_blocks = 0
                                             logger.info(
                                                 "city-sweep run_id=%d: house(detail) "
-                                                "ROTATE — IP changed (rotation #%d/%d)",
+                                                "ROTATE — proxy changed (rotation #%d/%d)",
                                                 run_id,
                                                 sweep_rotations_done,
                                                 settings.avito_proxy_max_rotations,
@@ -1390,6 +1602,10 @@ async def run_avito_city_sweep(
             logger.exception("city-sweep run_id=%d: fatal error", run_id)
             scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
             raise
+        finally:
+            # #2202: best-effort release+mark_health — не должен ронять уже
+            # собранный sweep. ok=False только если весь run завершился с ошибками.
+            _release_pool_lease(db, _lease, ok=(counters.errors_count == 0))
 
 
 # ── Avito newbuilding (novostroyka) citywide sweep ────────────────
@@ -1430,18 +1646,21 @@ async def run_avito_newbuilding_sweep(
     counters = NewbuildingSweepCounters()
 
     browser_mode = settings.scraper_fetch_mode == "browser"
+    # #2202: pool-lease текущей curl-сессии (None в browser-mode/pool off).
+    _lease: proxy_pool.ProxyLease | None = None
     async with AsyncExitStack() as stack:
         session: AsyncSession | None = None
         shared_bf: BrowserFetcher | None = None
         if browser_mode:
             shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito"))
         else:
+            _proxies, _lease = _pool_proxy_dict(db, "avito", run_id=run_id)
             session = await stack.enter_async_context(
                 AsyncSession(
                     impersonate="chrome120",
                     timeout=25,
                     headers=_CHROME_HEADERS,
-                    proxies=_avito_proxies(),
+                    proxies=_proxies,
                 )
             )
         try:
@@ -1508,6 +1727,8 @@ async def run_avito_newbuilding_sweep(
             logger.exception("nb-sweep run_id=%d: fatal error", run_id)
             scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
             raise
+        finally:
+            _release_pool_lease(db, _lease, ok=(counters.errors_count == 0))
 
 
 # ── Yandex city sweep ───────────────────────────────────────────

tradein-mvp/backend/tests/test_ban_rotation_1790.py

@@ -151,6 +151,10 @@ async def test_rotation_bounded_abort_when_budget_exhausted() -> None:
         mock_settings.scraper_fetch_mode = "curl_cffi"
         mock_settings.scraper_proxy_url = None
         mock_settings.scraper_skip_seen_today = False
+        # #2202: bare MagicMock settings auto-truthy on any attribute access — pin
+        # explicitly off so _rotate_avito_proxy takes the changeip path this test
+        # asserts on (pool path is covered separately in test_scrape_pipeline_proxy_pool.py).
+        mock_settings.use_proxy_pool_curl = False
 
         result = await run_avito_pipeline(
             MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0

tradein-mvp/backend/tests/test_scrape_pipeline_proxy_pool.py

@@ -0,0 +1,408 @@
+"""Tests for scrape_pipeline.py proxy-pool wiring (#2202).
+
+Issue #2202: боевые скрейперы (scrape_pipeline.py Avito orchestrator) читали только
+статический settings.scraper_proxy_url — один mobile-IP на все прогоны, SPOF. Этот
+файл покрывает новые pool-aware helpers:
+
+  - _pool_proxy_dict: acquire() health-filtered lease за флагом USE_PROXY_POOL_CURL,
+    fallback на статический env при флаге off / пустом пуле / ошибке.
+  - _rotate_pool_proxy: переключение на ДРУГОЙ здоровый прокси из пула при бане
+    (mark_health(False)+release текущего, acquire() следующего, мутация session.proxies).
+  - _rotate_avito_proxy: единая точка ротации — пул first, changeip fallback.
+
+Acceptance-тест (issue AC4): пул с 1 «забаненным» прокси — ротация выбирает
+СЛЕДУЮЩИЙ здоровый, не зависший на забаненном и не переиспользующий только что
+провалившийся lease раньше более свежего здорового кандидата.
+
+Offline: FakeSession эмулирует scrape_proxies-таблицу (тот же паттерн, что
+tests/services/test_proxy_pool.py), без live БД/сети.
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from app.services import proxy_pool
+from app.services.proxy_pool import MAX_CONSECUTIVE_FAILS
+from app.services.scrape_pipeline import (
+    _pool_proxy_dict,
+    _release_pool_lease,
+    _rotate_avito_proxy,
+    _rotate_pool_proxy,
+)
+
+# ── stateful fake session (mirrors tests/services/test_proxy_pool.py) ─────────
+
+
+class _FakeResult:
+    def __init__(self, rows: list[dict[str, Any]]):
+        self._rows = rows
+
+    def mappings(self) -> _FakeResult:
+        return self
+
+    def fetchone(self) -> dict[str, Any] | None:
+        return self._rows[0] if self._rows else None
+
+    def all(self) -> list[dict[str, Any]]:
+        return list(self._rows)
+
+
+class FakeSession:
+    """Эмуляция Session поверх in-memory списка scrape_proxies-строк."""
+
+    def __init__(self, rows: list[dict[str, Any]]):
+        self.rows = rows
+
+    def _by_id(self, pid: int) -> dict[str, Any] | None:
+        return next((r for r in self.rows if r["id"] == pid), None)
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        sql = str(stmt)
+        p = params or {}
+
+        if "FOR UPDATE SKIP LOCKED" in sql:  # acquire SELECT
+            provider = p["provider"]
+            max_fails = p["max_fails"]
+            cands = [
+                r
+                for r in self.rows
+                if r["enabled"]
+                and r["consecutive_fails"] < max_fails
+                and r["provider_affinity"] in (provider, "any")
+                and r["leased_by"] is None
+            ]
+            cands.sort(
+                key=lambda r: (
+                    r["last_ok_at"] is None,
+                    r["last_ok_at"] or datetime.min.replace(tzinfo=UTC),
+                    r["id"],
+                )
+            )
+            return _FakeResult(cands[:1])
+
+        if "SET leased_by = CAST(:run_id" in sql:  # acquire lease UPDATE
+            row = self._by_id(p["id"])
+            if row is not None:
+                row["leased_by"] = p["run_id"]
+                row["leased_at"] = datetime.now(UTC)
+            return _FakeResult([])
+
+        if "SET leased_by = NULL" in sql:  # release
+            row = self._by_id(p["id"])
+            if row is not None:
+                row["leased_by"] = None
+                row["leased_at"] = None
+            return _FakeResult([])
+
+        if "SET consecutive_fails = 0" in sql:  # mark_health ok
+            row = self._by_id(p["id"])
+            if row is not None:
+                row["consecutive_fails"] = 0
+                row["exit_ip"] = p["exit_ip"]
+                row["latency_ms"] = p["latency_ms"]
+                row["last_ok_at"] = datetime.now(UTC)
+            return _FakeResult([])
+
+        if "consecutive_fails = consecutive_fails + 1" in sql:  # mark_health fail
+            row = self._by_id(p["id"])
+            if row is not None:
+                row["consecutive_fails"] += 1
+                if row["consecutive_fails"] >= p["disable_threshold"]:
+                    row["enabled"] = False
+            return _FakeResult([])
+
+        raise AssertionError(f"unhandled SQL: {sql}")
+
+    def commit(self) -> None:
+        pass
+
+    def rollback(self) -> None:
+        pass
+
+
+def _proxy(
+    pid: int,
+    *,
+    affinity: str = "avito",
+    enabled: bool = True,
+    fails: int = 0,
+    leased_by: int | None = None,
+    last_ok_at: datetime | None = None,
+) -> dict[str, Any]:
+    return {
+        "id": pid,
+        "url": f"http://u:p@h{pid}:8080",
+        "kind": "http",
+        "rotate_url": None,
+        "provider_affinity": affinity,
+        "enabled": enabled,
+        "consecutive_fails": fails,
+        "leased_by": leased_by,
+        "leased_at": None,
+        "last_ok_at": last_ok_at,
+        "exit_ip": None,
+        "latency_ms": None,
+    }
+
+
+def _mock_settings(*, use_proxy_pool_curl: bool, scraper_proxy_url: str | None = None):
+    return SimpleNamespace(
+        use_proxy_pool_curl=use_proxy_pool_curl,
+        scraper_proxy_url=scraper_proxy_url,
+    )
+
+
+# ── _pool_proxy_dict ────────────────────────────────────────────────────────
+
+
+def test_pool_proxy_dict_falls_back_when_flag_off():
+    db = FakeSession([_proxy(1)])
+    with patch(
+        "app.services.scrape_pipeline.settings",
+        _mock_settings(use_proxy_pool_curl=False, scraper_proxy_url="http://env:3128"),
+    ):
+        proxies, lease = _pool_proxy_dict(db, "avito")  # type: ignore[arg-type]
+
+    assert proxies == {"http": "http://env:3128", "https": "http://env:3128"}
+    assert lease is None
+
+
+def test_pool_proxy_dict_returns_lease_when_flag_on_and_healthy():
+    db = FakeSession([_proxy(1)])
+    with patch(
+        "app.services.scrape_pipeline.settings",
+        _mock_settings(use_proxy_pool_curl=True),
+    ):
+        proxies, lease = _pool_proxy_dict(db, "avito")  # type: ignore[arg-type]
+
+    assert lease is not None and lease.id == 1
+    assert proxies == {"http": "http://u:p@h1:8080", "https": "http://u:p@h1:8080"}
+    assert db._by_id(1)["leased_by"] is not None  # lease проставлен
+
+
+def test_pool_proxy_dict_falls_back_when_pool_empty():
+    db = FakeSession([])  # пустой пул
+    with patch(
+        "app.services.scrape_pipeline.settings",
+        _mock_settings(use_proxy_pool_curl=True, scraper_proxy_url="http://env:3128"),
+    ):
+        proxies, lease = _pool_proxy_dict(db, "avito")  # type: ignore[arg-type]
+
+    assert lease is None
+    assert proxies == {"http": "http://env:3128", "https": "http://env:3128"}
+
+
+# ── _rotate_pool_proxy — acceptance-тест AC4 ───────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_pool_proxy_skips_banned_picks_next_healthy():
+    """#2202 AC4: пул с 1 забаненным прокси — ротация выбирает следующий здоровый.
+
+    BAD (id=1) — уже забанен (fails>=MAX_CONSECUTIVE_FAILS), исключён acquire() на
+    уровне SQL-фильтра. CURRENT (id=3) — прокси активного прогона, только что словивший
+    бан (403/429) — вызывающий код зовёт rotate. GOOD (id=2) — свободный здоровый,
+    единственный валидный кандидат после того как CURRENT будет помечен/освобождён.
+
+    NB (#2202 MAJOR-fix): гарантия исключения CURRENT из acquire() держится на том,
+    что release(current_lease) вызывается ПОСЛЕ acquire(), не до — пока current_lease
+    leased_by-занят, SQL `leased_by IS NULL` исключает его вне зависимости от id/
+    last_ok_at сортировки. См. зеркальный тест ниже
+    (test_rotate_pool_proxy_skips_banned_picks_next_healthy_current_id_lower) —
+    та же гарантия при CURRENT id НИЖЕ GOOD id (иначе тай-брейк по id в ORDER BY
+    "last_ok_at NULLS LAST, id" мог случайно фейворить эту фикстуру).
+    """
+    db = FakeSession(
+        [
+            _proxy(1, fails=MAX_CONSECUTIVE_FAILS),  # BAD — уже забанен
+            _proxy(2, fails=0),  # GOOD — ожидаемый выбор
+            _proxy(3, fails=0, leased_by=999),  # CURRENT — в работе, банится сейчас
+        ]
+    )
+    current_lease = proxy_pool.ProxyLease(
+        id=3, url="http://u:p@h3:8080", kind="http", rotate_url=None
+    )
+    session = SimpleNamespace(proxies=None)
+
+    new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease)  # type: ignore[arg-type]
+
+    assert new_lease is not None
+    assert new_lease.id == 2  # GOOD, не BAD (excluded) и не CURRENT (только что провален)
+    assert session.proxies == {"http": "http://u:p@h2:8080", "https": "http://u:p@h2:8080"}
+
+    # CURRENT (id=3) помечен нездоровым и освобождён
+    assert db._by_id(3)["consecutive_fails"] == 1
+    assert db._by_id(3)["leased_by"] is None
+    # BAD (id=1) не тронут этой операцией
+    assert db._by_id(1)["consecutive_fails"] == MAX_CONSECUTIVE_FAILS
+    # GOOD (id=2) теперь зализан под новую ротацию
+    assert db._by_id(2)["leased_by"] is not None
+
+
+@pytest.mark.asyncio
+async def test_rotate_pool_proxy_skips_banned_picks_next_healthy_current_id_lower():
+    """#2202 AC4 regression guard (MAJOR-fix): гарантия исключения current_lease НЕ
+    должна зависеть от того, чей id меньше в фикстуре.
+
+    Зеркало test_rotate_pool_proxy_skips_banned_picks_next_healthy с CURRENT/GOOD id
+    в обратном порядке: CURRENT=1 < GOOD=2 (оригинальный тест имел CURRENT=3 > GOOD=2,
+    поэтому "тай-брейк по id" в ORDER BY last_ok_at NULLS LAST, id случайно совпадал с
+    правильным ответом и не доказывал реальную гарантию — см. review-репро на #2202).
+
+    До фикса (release(current_lease) ДО acquire()): CURRENT освобождается, попадает в
+    leased_by IS NULL кандидаты с fails=1 (< MAX_CONSECUTIVE_FAILS=3, т.е. "здоров"),
+    и при равном last_ok_at (оба NULL) выигрывает тай-брейк по id (1 < 2) — acquire()
+    возвращает ТОТ ЖЕ прокси, который только что забанили. Этот тест ловит именно этот
+    регресс: если release() снова передвинуть перед acquire(), new_lease.id будет 1
+    (CURRENT), а не 2 (GOOD), и assert упадёт.
+    """
+    db = FakeSession(
+        [
+            _proxy(1, fails=0, leased_by=999),  # CURRENT — id НИЖЕ GOOD, банится сейчас
+            _proxy(2, fails=0),  # GOOD — единственный здоровый свободный кандидат
+        ]
+    )
+    current_lease = proxy_pool.ProxyLease(
+        id=1, url="http://u:p@h1:8080", kind="http", rotate_url=None
+    )
+    session = SimpleNamespace(proxies=None)
+
+    new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease)  # type: ignore[arg-type]
+
+    assert new_lease is not None
+    assert new_lease.id == 2  # GOOD — НЕ CURRENT, несмотря на id=1 < id=2
+    assert session.proxies == {"http": "http://u:p@h2:8080", "https": "http://u:p@h2:8080"}
+
+    # CURRENT (id=1) помечен нездоровым и освобождён (после успешной ротации)
+    assert db._by_id(1)["consecutive_fails"] == 1
+    assert db._by_id(1)["leased_by"] is None
+    # GOOD (id=2) теперь зализан под новую ротацию
+    assert db._by_id(2)["leased_by"] is not None
+
+
+@pytest.mark.asyncio
+async def test_rotate_pool_proxy_exhausted_returns_none():
+    """Все кандидаты нездоровы после ротации → None (caller падает на changeip).
+
+    BAD (id=1) уже забанен. CURRENT (id=3) на MAX_CONSECUTIVE_FAILS-1 fails — ЭТА
+    ротация (mark_health ok=False) толкает его ровно до порога → тоже исключается
+    из acquire(). Ни одного здорового кандидата не остаётся.
+    """
+    db = FakeSession(
+        [
+            _proxy(1, fails=MAX_CONSECUTIVE_FAILS),  # BAD — уже забанен
+            _proxy(3, fails=MAX_CONSECUTIVE_FAILS - 1, leased_by=999),  # CURRENT — на грани
+        ]
+    )
+    current_lease = proxy_pool.ProxyLease(
+        id=3, url="http://u:p@h3:8080", kind="http", rotate_url=None
+    )
+    session = SimpleNamespace(proxies=None)
+
+    new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease)  # type: ignore[arg-type]
+
+    assert new_lease is None
+    assert session.proxies is None  # не тронут — caller остаётся на старом/changeip
+    assert db._by_id(3)["leased_by"] is None  # текущий всё равно освобождён
+
+
+# ── _rotate_avito_proxy — pool-first, changeip fallback ────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_avito_proxy_uses_pool_when_flag_on():
+    db = FakeSession(
+        [
+            _proxy(1, fails=0),  # свободный здоровый — цель ротации
+            _proxy(3, fails=0, leased_by=999),  # CURRENT
+        ]
+    )
+    current_lease = proxy_pool.ProxyLease(
+        id=3, url="http://u:p@h3:8080", kind="http", rotate_url=None
+    )
+    session = SimpleNamespace(proxies=None)
+
+    with patch("app.services.scrape_pipeline.settings", _mock_settings(use_proxy_pool_curl=True)):
+        with patch("app.services.scrape_pipeline._rotate_proxy_ip", AsyncMock()) as changeip_mock:
+            rotated, new_lease = await _rotate_avito_proxy(
+                db,  # type: ignore[arg-type]
+                session,  # type: ignore[arg-type]
+                current_lease,
+                reason="test",
+                rotations_done=0,
+            )
+
+    assert rotated is True
+    assert new_lease is not None and new_lease.id == 1
+    changeip_mock.assert_not_awaited()  # пул сработал — changeip не нужен
+
+
+@pytest.mark.asyncio
+async def test_rotate_avito_proxy_falls_back_to_changeip_when_flag_off():
+    session = SimpleNamespace(proxies=None)
+
+    with patch("app.services.scrape_pipeline.settings", _mock_settings(use_proxy_pool_curl=False)):
+        with patch(
+            "app.services.scrape_pipeline._rotate_proxy_ip", AsyncMock(return_value=True)
+        ) as changeip_mock:
+            rotated, new_lease = await _rotate_avito_proxy(
+                None,  # type: ignore[arg-type]
+                session,  # type: ignore[arg-type]
+                None,
+                reason="test",
+                rotations_done=0,
+            )
+
+    assert rotated is True
+    assert new_lease is None
+    changeip_mock.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_rotate_avito_proxy_falls_back_to_changeip_when_pool_exhausted():
+    """Пул on, но acquire() пуст (например все забанены) — падаем на changeip, не молчим."""
+    db = FakeSession([_proxy(1, fails=MAX_CONSECUTIVE_FAILS)])  # единственный — забанен
+    session = SimpleNamespace(proxies=None)
+
+    with patch("app.services.scrape_pipeline.settings", _mock_settings(use_proxy_pool_curl=True)):
+        with patch(
+            "app.services.scrape_pipeline._rotate_proxy_ip", AsyncMock(return_value=True)
+        ) as changeip_mock:
+            rotated, new_lease = await _rotate_avito_proxy(
+                db,  # type: ignore[arg-type]
+                session,  # type: ignore[arg-type]
+                None,
+                reason="test",
+                rotations_done=0,
+            )
+
+    assert rotated is True
+    assert new_lease is None
+    changeip_mock.assert_awaited_once()  # pool exhausted → fallback path taken
+
+
+# ── _release_pool_lease ──────────────────────────────────────────────────────
+
+
+def test_release_pool_lease_noop_when_none():
+    _release_pool_lease(None, None, ok=True)  # type: ignore[arg-type]  # must not raise
+
+
+def test_release_pool_lease_marks_health_and_releases():
+    db = FakeSession([_proxy(1, leased_by=100)])
+    lease = proxy_pool.ProxyLease(id=1, url="http://u:p@h1:8080", kind="http", rotate_url=None)
+
+    _release_pool_lease(db, lease, ok=True)  # type: ignore[arg-type]
+
+    assert db._by_id(1)["leased_by"] is None
+    assert db._by_id(1)["consecutive_fails"] == 0