diff --git a/tradein-mvp/backend/app/services/proxy_pool.py b/tradein-mvp/backend/app/services/proxy_pool.py index 8ed2176e..0e7dce9c 100644 --- a/tradein-mvp/backend/app/services/proxy_pool.py +++ b/tradein-mvp/backend/app/services/proxy_pool.py @@ -292,6 +292,13 @@ def acquire(db: Session, provider: str, *, run_id: int | None = None) -> ProxyLe остаётся полноценным кандидатом для Яндекса и остальных источников. Бан по чужому source на выдачу не влияет вообще. + ОБА запроса также отсекают узлы с истёкшей арендой порта у провайдера + (expires_at IS NOT NULL AND expires_at <= now()) — иначе после истечения аренды + площадка отвечает 407/рвёт соединение, а пул продолжает выдавать этот узел до + ручного вмешательства. expires_at IS NULL ("срок не отслеживается") выдаче не + мешает. Перед отбором отдельным запросом логируется WARNING по каждому такому + узлу — иначе сужение пула из-за истёкшей аренды прошло бы для оператора молча. + Конкурентные acquire не дерутся за одну строку: SKIP LOCKED пропускает залоченную другим вызовом строку, второй параллельный acquire берёт следующую свободную. @@ -299,6 +306,34 @@ def acquire(db: Session, provider: str, *, run_id: int | None = None) -> ProxyLe """ lease_marker = run_id if run_id is not None else NON_RUN_LEASE_MARKER + # Просроченные (expires_at в прошлом) enabled-узлы никогда не попадут ни в основной, + # ни в fallback-отбор ниже (оба фильтруют expires_at). Без явного WARNING сужение пула + # прошло бы молча — площадка начинает отдавать 407/рвать соединение по истёкшему + # порту, а pull продолжал бы его выдавать, пока человек не заметит руками. + expired_rows = ( + db.execute( + text( + """ + SELECT id + FROM scrape_proxies + WHERE enabled + AND leased_by IS NULL + AND expires_at IS NOT NULL + AND expires_at <= now() + """ + ) + ) + .mappings() + .all() + ) + for expired_row in expired_rows: + logger.warning( + "proxy_pool: proxy id=%d skipped for acquire(provider=%s) — lease expired " + "(expires_at <= now()), not eligible until renewed or disabled", + int(expired_row["id"]), + provider, + ) + row = ( db.execute( text( @@ -309,6 +344,7 @@ def acquire(db: Session, provider: str, *, run_id: int | None = None) -> ProxyLe AND consecutive_fails < CAST(:max_fails AS integer) AND provider_affinity IN (:provider, 'any') AND leased_by IS NULL + AND (expires_at IS NULL OR expires_at > now()) AND NOT EXISTS ( SELECT 1 FROM scrape_proxy_source_bans b @@ -344,6 +380,7 @@ def acquire(db: Session, provider: str, *, run_id: int | None = None) -> ProxyLe WHERE sp.enabled AND sp.consecutive_fails < CAST(:max_fails AS integer) AND sp.leased_by IS NULL + AND (sp.expires_at IS NULL OR sp.expires_at > now()) AND NOT EXISTS ( SELECT 1 FROM scrape_proxy_source_bans b diff --git a/tradein-mvp/backend/tests/services/test_proxy_pool.py b/tradein-mvp/backend/tests/services/test_proxy_pool.py index 478e394a..25c5455c 100644 --- a/tradein-mvp/backend/tests/services/test_proxy_pool.py +++ b/tradein-mvp/backend/tests/services/test_proxy_pool.py @@ -131,6 +131,20 @@ class FakeSession: sql = str(stmt) p = params or {} + if "expires_at <= now()" in sql and "FOR UPDATE SKIP LOCKED" not in sql: + # acquire() отдельный WARNING-запрос (#3287): просроченные enabled+свободные + # узлы, вне зависимости от affinity/consecutive_fails — они всё равно не + # попадут ни в основной, ни в fallback-отбор ниже. + expired = [ + r + for r in self.rows + if r["enabled"] + and r["leased_by"] is None + and r.get("expires_at") is not None + and r["expires_at"] <= datetime.now(UTC) + ] + return _FakeResult([{"id": r["id"]} for r in expired]) + if "FOR UPDATE SKIP LOCKED" in sql: # acquire SELECT (primary affinity-scoped or fallback) max_fails = p["max_fails"] provider = p["provider"] @@ -139,10 +153,19 @@ class FakeSession: # NOT EXISTS по scrape_proxy_source_bans, иначе мок реализовывал бы логику # независимо от проверяемого кода и не отличил бы старый запрос от нового. filters_bans = "scrape_proxy_source_bans" in sql + # #3287: истёкшая аренда порта не выдаётся — гейтим по подстроке фильтра, + # тот же принцип, что и у filters_bans выше. + filters_expiry = "expires_at" in sql def _not_banned(row: dict[str, Any]) -> bool: return not filters_bans or not self._has_active_ban(row["id"], provider) + def _not_expired(row: dict[str, Any]) -> bool: + if not filters_expiry: + return True + exp = row.get("expires_at") + return exp is None or exp > datetime.now(UTC) + if "provider_affinity IN" in sql: # primary: своя affinity ИЛИ 'any' cands = [ r @@ -152,6 +175,7 @@ class FakeSession: and r["provider_affinity"] in (provider, "any") and r["leased_by"] is None and _not_banned(r) + and _not_expired(r) ] else: # fallback: любая affinity, но не последний узел выделенной affinity # (domclick и т.п. — #2600 review). ВАЖНО: применяем эту фильтрацию, @@ -186,6 +210,7 @@ class FakeSession: and r["consecutive_fails"] < max_fails and r["leased_by"] is None and _not_banned(r) + and _not_expired(r) and (not protects_last_node or _has_backup(r)) ] # ORDER BY (browser_unfit_since IS NOT NULL), last_ok_at NULLS LAST, id. @@ -469,6 +494,7 @@ def _proxy( browser_unfit_since: datetime | None = None, browser_fail_streak: int = 0, browser_check_at: datetime | None = None, + expires_at: datetime | None = None, ) -> dict[str, Any]: return { "id": pid, @@ -489,6 +515,8 @@ def _proxy( "browser_unfit_since": browser_unfit_since, "browser_fail_streak": browser_fail_streak, "browser_check_at": browser_check_at, + # срок аренды порта у провайдера — #3287, None = "не отслеживается" + "expires_at": expires_at, } @@ -623,6 +651,71 @@ def test_acquire_fallback_backup_must_be_usable_for_its_own_source() -> None: assert lease is not None and lease.id == 2 +# ── acquire × expires_at — просроченная аренда порта не выдаётся (#3287) ─────── + + +def test_acquire_skips_expired_lease() -> None: + """expires_at в прошлом → узел не выдаётся, даже если enabled/здоров/свободен.""" + db = FakeSession( + [_proxy(1, affinity="avito", expires_at=datetime.now(UTC) - timedelta(minutes=1))] + ) + assert acquire(db, "avito", run_id=1) is None # type: ignore[arg-type] + assert db._by_id(1)["leased_by"] is None # узел не тронут + + +def test_acquire_issues_proxy_with_null_expires_at() -> None: + """expires_at IS NULL — "срок не отслеживается", выдаче не мешает (как раньше).""" + db = FakeSession([_proxy(1, affinity="avito", expires_at=None)]) + lease = acquire(db, "avito", run_id=1) # type: ignore[arg-type] + assert lease is not None and lease.id == 1 + + +def test_acquire_issues_proxy_with_future_expires_at() -> None: + """expires_at в будущем — аренда ещё жива, узел выдаётся.""" + db = FakeSession( + [_proxy(1, affinity="avito", expires_at=datetime.now(UTC) + timedelta(hours=1))] + ) + lease = acquire(db, "avito", run_id=1) # type: ignore[arg-type] + assert lease is not None and lease.id == 1 + + +def test_acquire_fallback_skips_expired_lease() -> None: + """Fallback-ветка (чужая affinity) тоже не выдаёт просроченный узел.""" + db = FakeSession( + [_proxy(1, affinity="cian", expires_at=datetime.now(UTC) - timedelta(minutes=1))] + ) + assert acquire(db, "avito", run_id=1) is None # type: ignore[arg-type] + + +def test_acquire_fallback_prefers_non_expired_over_expired() -> None: + """Просроченный узел пропускается, живой той же чужой affinity — выдан fallback'ом.""" + db = FakeSession( + [ + _proxy(1, affinity="cian", expires_at=datetime.now(UTC) - timedelta(minutes=1)), + _proxy(2, affinity="cian", expires_at=datetime.now(UTC) + timedelta(hours=1)), + ] + ) + lease = acquire(db, "avito", run_id=1) # type: ignore[arg-type] + assert lease is not None + assert lease.id == 2 + + +def test_acquire_warns_on_expired_proxy(caplog: pytest.LogCaptureFixture) -> None: + """Просроченный узел логируется WARNING'ом — оператору нужен явный сигнал (#3287).""" + db = FakeSession( + [ + _proxy(1, affinity="avito", expires_at=datetime.now(UTC) - timedelta(minutes=1)), + _proxy(2, affinity="avito", expires_at=datetime.now(UTC) + timedelta(hours=1)), + ] + ) + with caplog.at_level("WARNING", logger="app.services.proxy_pool"): + lease = acquire(db, "avito", run_id=1) # type: ignore[arg-type] + assert lease is not None and lease.id == 2 # живой узел всё равно выдан + assert any( + "expired" in r.message and "id=1" in r.message for r in caplog.records + ), "ожидался WARNING про просроченный proxy id=1" + + # ── release ──────────────────────────────────────────────────────────────────