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

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 ──────────────────────────────────────────────────────────────────