feat: add stale proxy expiry based on last_seen TTL

Evict proxies not returned by sources for >3 refresh cycles and not
currently alive. Cleans up proxies removed upstream faster than waiting
for max_fails consecutive health test failures.

src/s5p/pool.py

@@ -297,15 +297,31 @@ class ProxyPool:
                 if not skip_eviction and entry.fails >= self._cfg.max_fails:
                     evict_keys.append(key)
 
+        # stale proxy expiry: remove proxies not seen by sources recently
+        now = time.time()
+        stale_cutoff = now - self._cfg.refresh * 3
+        stale_keys = [
+            k for k, e in self._proxies.items()
+            if e.last_seen < stale_cutoff and not e.alive and k not in evict_keys
+        ]
+        evict_keys.extend(stale_keys)
+
         for key in evict_keys:
             del self._proxies[key]
 
         self._rebuild_alive()
+        parts = []
+        fail_evicted = len(evict_keys) - len(stale_keys)
+        if fail_evicted:
+            parts.append(f"evicted {fail_evicted}")
+        if stale_keys:
+            parts.append(f"stale {len(stale_keys)}")
+        suffix = f" ({', '.join(parts)})" if parts else ""
         logger.info(
             "pool: %d proxies, %d alive%s",
             len(self._proxies),
             len(self._alive_keys),
-            f" (evicted {len(evict_keys)})" if evict_keys else "",
+            suffix,
         )
 
     def _rebuild_alive(self) -> None:

tests/test_pool.py

@@ -153,6 +153,54 @@ class TestProxyPoolWeight:
         pool.report_failure(hop)  # should not raise
 
 
+class TestProxyPoolStaleExpiry:
+    """Test stale proxy eviction."""
+
+    def test_stale_dead_proxy_evicted(self):
+        import asyncio
+        from unittest.mock import AsyncMock, patch
+
+        cfg = ProxyPoolConfig(sources=[], refresh=300)
+        pool = ProxyPool(cfg, [], timeout=10.0)
+
+        now = time.time()
+        # stale + dead: should be evicted
+        stale = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
+        pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
+            hop=stale, alive=False, last_seen=now - 1200,
+        )
+        # fresh + dead: should survive (recently seen, might recover)
+        fresh_dead = ChainHop(proto="socks5", host="10.0.0.2", port=1080)
+        pool._proxies["socks5://10.0.0.2:1080"] = ProxyEntry(
+            hop=fresh_dead, alive=False, last_seen=now,
+        )
+
+        with patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=False):
+            asyncio.run(pool._run_health_tests())
+
+        assert "socks5://10.0.0.1:1080" not in pool._proxies
+        assert "socks5://10.0.0.2:1080" in pool._proxies
+
+    def test_stale_alive_proxy_kept(self):
+        import asyncio
+        from unittest.mock import AsyncMock, patch
+
+        cfg = ProxyPoolConfig(sources=[], refresh=300)
+        pool = ProxyPool(cfg, [], timeout=10.0)
+
+        now = time.time()
+        # stale but alive: should survive (still passing health tests)
+        hop = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
+        pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
+            hop=hop, alive=True, last_seen=now - 1200, last_ok=now,
+        )
+
+        with patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=True):
+            asyncio.run(pool._run_health_tests())
+
+        assert "socks5://10.0.0.1:1080" in pool._proxies
+
+
 class TestProxyPoolFetchFile:
     """Test file source parsing."""