From bd9c2d50bd159f1fa14286d0e50e3b1c4686c6b0 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 13 Aug 2026 20:37:42 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein/proxy):=20=D1=83=D1=87=D0=B8=D1=82?= =?UTF-8?q?=D1=8B=D0=B2=D0=B0=D1=82=D1=8C=20=D0=B8=D1=81=D1=82=D0=BE=D1=80?= =?UTF-8?q?=D0=B8=D1=8E=20=D0=B1=D0=B0=D0=BD=D0=BE=D0=B2=20=D0=BF=D1=80?= =?UTF-8?q?=D0=B8=20=D0=B2=D1=8B=D0=B1=D0=BE=D1=80=D0=B5=20egress-=D1=83?= =?UTF-8?q?=D0=B7=D0=BB=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Замер на проде 2026-08-13: резолвер выбирал asocks-residential-1, который отдаёт 403 и на Авито, и на Циане, при трёх рабочих мобильных узлах (Циан через них — 200). Причина: ранжирование шло по consecutive_fails, затем по свежести last_ok_at. У всех узлов ноль сбоев, поэтому решала свежесть healthcheck'а, а он проверяет доступность самого прокси, а не то, пускает ли через него площадка. Узел одновременно «здоров» и заблокирован. Как только истекал TTL бана, он снова становился первым кандидатом — при ban_count=17 по Циану. - В ранжирование добавлен ban_count по этому источнику: NOT EXISTS заменён на LEFT JOIN, истёкшие строки банов больше не теряются, а работают как история. Порядок: сначала узлы без истории, дальше по возрастанию ban_count, при равенстве — прежний tie-break. - Активный бан по-прежнему исключает узел полностью, семантика не менялась. - В лог выбора добавлен ban_count — чтобы было видно, что узел с историей выбран осознанно. Полный URL с credentials по-прежнему не логируется. На проде это ставит residential-1 последним для Авито (5 банов против 0 у mobile-2/3) вместо первого. --- .../backend/app/services/proxy_egress.py | 73 +++++-- .../tests/services/test_proxy_egress.py | 184 +++++++++++++++++- 2 files changed, 229 insertions(+), 28 deletions(-) diff --git a/tradein-mvp/backend/app/services/proxy_egress.py b/tradein-mvp/backend/app/services/proxy_egress.py index c28426a4..8227a40a 100644 --- a/tradein-mvp/backend/app/services/proxy_egress.py +++ b/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 diff --git a/tradein-mvp/backend/tests/services/test_proxy_egress.py b/tradein-mvp/backend/tests/services/test_proxy_egress.py index cf303b2b..b123bb9c 100644 --- a/tradein-mvp/backend/tests/services/test_proxy_egress.py +++ b/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: