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

@@ -19,10 +19,19 @@ reap_stale_leases) — она рассчитана на долгоживущие
 и без мутаций.
 
 ПРАВИЛО ВЫБОРА: enabled=true, consecutive_fails < proxy_pool.MAX_CONSECUTIVE_FAILS
-(тот же карантинный порог, что у acquire), нет активной строки в
-scrape_proxy_source_bans для ЭТОГО source. Среди кандидатов — меньший consecutive_fails,
-при равенстве — более свежий last_ok_at (NULLS LAST). Не изобретаем ротацию/балансировку:
-это резолвер «дай рабочий прокси прямо сейчас», не lease-менеджер.
+(тот же карантинный порог, что у acquire), нет АКТИВНОЙ строки (banned_until > now())
+в scrape_proxy_source_bans для ЭТОГО source — это по-прежнему жёсткий фильтр, не
+влияющий на порядок. Порядок среди прошедших фильтр (замер 2026-08-10, #2825 доп.):
+сначала узлы БЕЗ ИСТОРИИ банов по этому source, затем по возрастанию ban_count —
+даже если сама строка бана истекла (banned_until <= now()), её ban_count всё равно
+учитывается, ведь строка НЕ удаляется сразу (purge только через 7 суток чистой
+работы, см. 210-я миграция) и остаётся памятью «этот узел здесь уже банился N раз».
+Внутри равного ban_count — прежние критерии без изменений: меньший consecutive_fails,
+при равенстве — более свежий last_ok_at (NULLS LAST). Так хронически банящийся узел
+(здоров по health-check, но регулярно ловит 403 от конкретной площадки) не всплывает
+первым сразу после истечения TTL — свежий healthcheck сам по себе больше не решает.
+Не изобретаем ротацию/балансировку: это резолвер «дай рабочий прокси прямо сейчас»,
+не lease-менеджер.
 
 FAIL-CLOSED ПРОТИВ ТИХОГО ОБХОДА ПУЛА (#2616, deep-review этой правки): пул и статичный
 `SCRAPER_PROXY_URL` — РАЗНЫЕ вещи, и путать их нельзя. Два разных исхода "кандидата нет":
@@ -39,7 +48,9 @@ FAIL-CLOSED ПРОТИВ ТИХОГО ОБХОДА ПУЛА (#2616, deep-review
      env в обход учёта банов.
 
 НАБЛЮДАЕМОСТЬ: при выборе из пула логируем label/host:port (БЕЗ credentials — url
-несёт логин/пароль, в лог никогда не идёт целиком) и id узла; при legit-fallback —
+несёт логин/пароль, в лог никогда не идёт целиком), id узла и ban_count по этому
+source (0, если истории нет) — чтобы по логу было видно, что узел с историей банов
+выбран осознанно (пул исчерпан по чистым узлам), а не тихо; при legit-fallback —
 warning с текстом «пуст» (сценарий 1); при exhaustion — error с разбивкой
 banned_for_source/unhealthy_or_disabled (сценарий 2) — тексты НАМЕРЕННО разные, чтобы
 их нельзя было спутать в логах/алертах.
@@ -98,6 +109,10 @@ class _Candidate:
     id: int
     url: str
     label: str | None
+    ban_count: int
+    """ban_count по scrape_proxy_source_bans ДЛЯ ЭТОГО source (0, если строки нет —
+    узел ни разу не банился этой площадкой). Учитывает и истёкшие строки бана
+    (banned_until <= now(), но ещё не спурженные) — см. докстринг модуля."""
 
 
 def _safe_label(proxy_id: int, label: str | None, url: str) -> str:
@@ -113,23 +128,31 @@ def _safe_label(proxy_id: int, label: str | None, url: str) -> str:
 
 
 def _pick_candidate(db: Session, source: str) -> _Candidate | None:
-    """READ-ONLY выбор egress для source. Без FOR UPDATE — резолвер не арендует узел."""
+    """READ-ONLY выбор egress для source. Без FOR UPDATE — резолвер не арендует узел.
+
+    LEFT JOIN (не EXISTS) на scrape_proxy_source_bans — нужен сам ban_count для
+    ранжирования, а не только факт активного бана. Активный бан (banned_until > now())
+    по-прежнему полный фильтр в WHERE, это НЕ меняется; но истёкшая (и ещё не
+    спурженная) строка бана остаётся в ORDER BY как история — см. докстринг модуля.
+    COALESCE(b.ban_count, 0) — узел без единой строки истории по source ранжируется
+    как ban_count=0, естественно раньше любого узла с реальной историей банов.
+    """
     row = (
         db.execute(
             text(
                 """
-                SELECT id, url, label
-                FROM scrape_proxies
-                WHERE enabled
-                  AND consecutive_fails < CAST(:max_fails AS integer)
-                  AND NOT EXISTS (
-                      SELECT 1
-                      FROM scrape_proxy_source_bans b
-                      WHERE b.proxy_id = scrape_proxies.id
-                        AND b.source = CAST(:source AS text)
-                        AND b.banned_until > now()
-                  )
-                ORDER BY consecutive_fails ASC, last_ok_at DESC NULLS LAST, id
+                SELECT sp.id, sp.url, sp.label, COALESCE(b.ban_count, 0) AS ban_count
+                FROM scrape_proxies AS sp
+                LEFT JOIN scrape_proxy_source_bans AS b
+                    ON b.proxy_id = sp.id
+                   AND b.source = CAST(:source AS text)
+                WHERE sp.enabled
+                  AND sp.consecutive_fails < CAST(:max_fails AS integer)
+                  AND (b.banned_until IS NULL OR b.banned_until <= now())
+                ORDER BY COALESCE(b.ban_count, 0) ASC,
+                         sp.consecutive_fails ASC,
+                         sp.last_ok_at DESC NULLS LAST,
+                         sp.id
                 LIMIT 1
                 """
             ),
@@ -143,7 +166,16 @@ def _pick_candidate(db: Session, source: str) -> _Candidate | None:
     # середине более широкой операции).
     if row is None:
         return None
-    return _Candidate(id=int(row["id"]), url=str(row["url"]), label=row["label"])
+    return _Candidate(
+        id=int(row["id"]),
+        url=str(row["url"]),
+        label=row["label"],
+        # .get(..., 0) — не .__getitem__: production-SELECT ВСЕГДА проецирует
+        # ban_count (см. запрос выше), но нулевой default защищает от полного KeyError
+        # у сторонних fake-db в других test-модулях (напр. test_2830_pool_bypass_tails),
+        # которые мокают этот же db.execute() урезанным dict без нового столбца.
+        ban_count=int(row.get("ban_count", 0)),
+    )
 
 
 @dataclass(frozen=True)
@@ -239,10 +271,11 @@ def resolve_proxy_url(db: Session, source: str) -> str | None:
 
     if candidate is not None:
         logger.info(
-            "proxy_egress: source=%s -> pool proxy id=%d (%s)",
+            "proxy_egress: source=%s -> pool proxy id=%d (%s) ban_count=%d",
             source,
             candidate.id,
             _safe_label(candidate.id, candidate.label, candidate.url),
+            candidate.ban_count,
         )
         return candidate.url
 

tradein-mvp/backend/tests/services/test_proxy_egress.py

@@ -1,4 +1,5 @@
-"""Offline-тесты резолвера egress-прокси по источнику (#2825, fail-closed #2616).
+"""Offline-тесты резолвера egress-прокси по источнику (#2825, fail-closed #2616,
+ban-history ранжирование доп. #2825 от 2026-08-13).
 
 Покрытие БЕЗ live-сети/БД: FakeSession эмулирует ДВА запроса над scrape_proxies +
 scrape_proxy_source_bans — основной SELECT кандидата (`_pick_candidate`) и, только
@@ -6,11 +7,16 @@ scrape_proxy_source_bans — основной SELECT кандидата (`_pick_
 "пул пуст" от "пул не пуст, все отсеяны".
 
   - выбирается небанненный прокси;
-  - забаненный ДЛЯ ИСТОЧНИКА не выбирается;
+  - забаненный ДЛЯ ИСТОЧНИКА (АКТИВНО, banned_until > now()) не выбирается;
   - забаненный для ДРУГОГО источника — выбирается (суть #2600 п.2: Авито банит IP,
-    Яндекс через тот же IP ходит чисто);
-  - при нескольких кандидатах — меньший consecutive_fails выигрывает;
-  - при равном consecutive_fails — более свежий last_ok_at выигрывает;
+    Яндекс через тот же IP ходит чисто) — включая случай, когда у него накопилась
+    ИСТОРИЯ банов по другому source: на ранжирование ДЛЯ ТЕКУЩЕГО source это не влияет;
+  - узел с историей банов (даже истёкшей) по ЭТОМУ source уступает чистому узлу без
+    истории, даже когда у чистого узла хуже consecutive_fails/last_ok_at;
+  - при равной истории (ban_count) — работает прежний tie-break: меньший
+    consecutive_fails, затем более свежий last_ok_at;
+  - активный бан (banned_until > now()) по-прежнему полностью исключает узел, вне
+    зависимости от ban_count;
   - пул ПУСТ (0 строк вообще) → легитимный fallback на settings.scraper_proxy_url,
     logger.WARNING с текстом «пуст»;
   - пул пуст И SCRAPER_PROXY_URL не задан → None (прямое подключение), WARNING;
@@ -59,6 +65,14 @@ class FakeSession:
             for b in self.bans
         )
 
+    def _ban_count(self, pid: int, source: str) -> int:
+        """COALESCE(b.ban_count, 0) семантика LEFT JOIN — история учитывается ДАЖЕ
+        если сама строка бана уже истекла (banned_until <= now(), ещё не спурженная)."""
+        for b in self.bans:
+            if b["proxy_id"] == pid and b["source"] == source:
+                return int(b.get("ban_count", 1))
+        return 0
+
     def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
         sql = str(stmt)
         p = params or {}
@@ -67,7 +81,7 @@ class FakeSession:
         max_fails = p["max_fails"]
         source = p["source"]
 
-        if "pool_total" in sql:  # _diagnose_no_candidate aggregate
+        if "pool_total" in sql:  # _diagnose_no_candidate aggregate (активные баны, без истории)
             unhealthy = sum(
                 1 for r in self.rows if not r["enabled"] or r["consecutive_fails"] >= max_fails
             )
@@ -88,7 +102,9 @@ class FakeSession:
                 ]
             )
 
-        # _pick_candidate primary SELECT
+        # _pick_candidate primary SELECT — LEFT JOIN на bans по source: активный бан
+        # по-прежнему исключает узел (WHERE), а ban_count (в т.ч. от истёкшего бана)
+        # ранжирует прошедших фильтр: сначала без истории (0), затем по возрастанию.
         cands = [
             r
             for r in self.rows
@@ -98,12 +114,15 @@ class FakeSession:
         ]
         cands.sort(
             key=lambda r: (
+                self._ban_count(r["id"], source),
                 r["consecutive_fails"],
                 -(r["last_ok_at"] or datetime.min.replace(tzinfo=UTC)).timestamp(),
                 r["id"],
             )
         )
-        return _FakeResult([dict(r) for r in cands[:1]])
+        return _FakeResult(
+            [{**r, "ban_count": self._ban_count(r["id"], source)} for r in cands[:1]]
+        )
 
 
 def _proxy(
@@ -207,6 +226,155 @@ def test_tiebreak_fresher_last_ok_at_wins_on_equal_fails() -> None:
     assert result == "http://u:p@fresh.local:8080"
 
 
+def test_clean_node_beats_node_with_expired_ban_history_for_source() -> None:
+    """Замер на проде 2026-08-10: asocks-residential-1 отдавал 403 и cian, и avito, но
+    после истечения TTL всплывал первым, потому что consecutive_fails=0 у ОБОИХ узлов
+    и решал только свежий healthcheck. История (ban_count) ДОЛЖНА перевешивать даже
+    когда у банившегося узла лучше consecutive_fails/last_ok_at."""
+    now = datetime.now(UTC)
+    db = FakeSession(
+        [
+            _proxy(
+                1,
+                consecutive_fails=0,
+                last_ok_at=now,  # свежее всех — раньше выиграл бы по старому правилу
+                url="http://u:p@chronic.local:8080",
+            ),
+            _proxy(
+                2,
+                consecutive_fails=1,
+                last_ok_at=now - timedelta(hours=3),
+                url="http://u:p@clean.local:8080",
+            ),
+        ],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "avito",
+                "banned_until": now - timedelta(hours=1),  # ИСТЁК, но ban_count остаётся
+                "ban_count": 4,
+            }
+        ],
+    )
+    result = resolve_proxy_url(db, "avito")
+    assert result == "http://u:p@clean.local:8080"
+
+
+def test_no_history_node_beats_node_with_ban_count_one() -> None:
+    """Узел БЕЗ ЕДИНОЙ строки истории (ban_count трактуется как 0) выигрывает у узла с
+    ban_count=1, даже при равном consecutive_fails/last_ok_at."""
+    now = datetime.now(UTC)
+    db = FakeSession(
+        [
+            _proxy(1, consecutive_fails=0, last_ok_at=now, url="http://u:p@once-banned.local:8080"),
+            _proxy(
+                2, consecutive_fails=0, last_ok_at=now, url="http://u:p@never-banned.local:8080"
+            ),
+        ],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "cian",
+                "banned_until": now - timedelta(hours=2),
+                "ban_count": 1,
+            }
+        ],
+    )
+    result = resolve_proxy_url(db, "cian")
+    assert result == "http://u:p@never-banned.local:8080"
+
+
+def test_equal_ban_history_falls_back_to_prior_tiebreak() -> None:
+    """При РАВНОМ ban_count у обоих узлов -- прежний порядок tie-break (consecutive_fails,
+    затем last_ok_at) без изменений."""
+    now = datetime.now(UTC)
+    db = FakeSession(
+        [
+            _proxy(
+                1,
+                consecutive_fails=2,
+                last_ok_at=now,
+                url="http://u:p@flaky-history.local:8080",
+            ),
+            _proxy(
+                2,
+                consecutive_fails=0,
+                last_ok_at=now - timedelta(hours=1),
+                url="http://u:p@solid-history.local:8080",
+            ),
+        ],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "yandex",
+                "banned_until": now - timedelta(hours=5),
+                "ban_count": 2,
+            },
+            {
+                "proxy_id": 2,
+                "source": "yandex",
+                "banned_until": now - timedelta(hours=5),
+                "ban_count": 2,
+            },
+        ],
+    )
+    result = resolve_proxy_url(db, "yandex")
+    # Равный ban_count=2 у обоих -- решает consecutive_fails (0 < 2).
+    assert result == "http://u:p@solid-history.local:8080"
+
+
+def test_ban_history_on_other_source_does_not_affect_ranking() -> None:
+    """Высокий ban_count по source=cian у узла НЕ влияет на его ранжирование для
+    source=avito -- история строго per-source, ровно как активный бан (#2600 п.2)."""
+    now = datetime.now(UTC)
+    db = FakeSession(
+        [
+            _proxy(
+                1,
+                consecutive_fails=0,
+                last_ok_at=now,
+                url="http://u:p@cian-history-only.local:8080",
+            ),
+            _proxy(
+                2,
+                consecutive_fails=0,
+                last_ok_at=now - timedelta(hours=2),
+                url="http://u:p@clean-everywhere.local:8080",
+            ),
+        ],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "cian",  # ДРУГОЙ source, не avito
+                "banned_until": now - timedelta(hours=1),
+                "ban_count": 9,
+            }
+        ],
+    )
+    result = resolve_proxy_url(db, "avito")
+    # Для avito у узла 1 ban_count=0 (истории по avito нет) -- выигрывает по last_ok_at.
+    assert result == "http://u:p@cian-history-only.local:8080"
+
+
+def test_active_ban_still_excludes_regardless_of_ban_count() -> None:
+    """Активный бан по-прежнему полный фильтр -- ban_count=1 (низкий) не спасает узел
+    с АКТИВНЫМ баном от исключения."""
+    now = datetime.now(UTC)
+    db = FakeSession(
+        [_proxy(1, url="http://u:p@actively-banned.local:8080")],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "avito",
+                "banned_until": now + timedelta(hours=6),
+                "ban_count": 1,
+            }
+        ],
+    )
+    with pytest.raises(ProxyPoolExhaustedError):
+        resolve_proxy_url(db, "avito")
+
+
 def test_empty_pool_falls_back_to_env_with_warning(
     monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None: