proxy pool: spread rule34video enrichment over checked public proxies

Rework the unused ALL_PROXIES list into util/proxy_pool.rs:
- hourly sweep of the public lists, TLS-verified cloudflare trace checks,
  >=2 successes with a median under 1.5s, one proxy per exit IP; proxies
  that show our own IP are dropped, two failures in a row evict
- per-site sets checked against a site URL (rule34video: robots.txt),
  round-robin, 10-min per-site cooldown on 429/challenge pages, live
  request outcomes fed back
- started from main (PROXY_POOL=0 disables; PROXY_POOL_LISTS,
  PROXY_POOL_CONCURRENCY), /api/proxies shows its status without credentials
- wreq "socks" feature enabled: SOCKS proxies silently failed before

rule34video enrichment is back, through the pool: full detail pages
(1080p..360p formats, tags, artist), 8 concurrent, one 4s deadline per
listing; late pages keep running and land in the 2h detail cache, which
cached listings are re-merged with on every serve. No ready proxies means
no enrichment (no direct fallback). Requester::get_via sends a pooled
request with certificate checks, sharing the session's cookie jar.

Also adds an ignored live probe showing /embed/ shares the 429 limit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AEp724zDpCWNaSeCrMSvjx

Cargo.toml

@@ -20,7 +20,7 @@ ntex-files = "3"
 serde = "1.0.228"
 serde_json = "1.0.145"
 tokio = { version = "1.49", features = ["full"] }
-wreq =  { version = "6.0.0-rc.31", features = ["cookies", "multipart", "json", "gzip", "brotli", "deflate", "zstd"] }
+wreq =  { version = "6.0.0-rc.31", features = ["cookies", "multipart", "json", "gzip", "brotli", "deflate", "zstd", "socks"] }
 wreq-util = "3.0.0-rc.14"
 percent-encoding = "2.3.2"
 capitalize = "0.3.4"

SKILL.md

@@ -64,6 +64,9 @@ Runtime env vars:
 - `PROXY` optional. Any value other than `"0"` enables proxy mode in the shared requester.
 - `BURP_URL` optional. Outbound HTTP proxy used when `PROXY` is enabled.
 - `FLARE_URL` optional but strongly recommended for provider work. Used for FlareSolverr fallback and some providers that call it directly.
+- `PROXY_POOL` optional. `0` disables the public-proxy pool used by rule34video enrichment (default on).
+- `PROXY_POOL_LISTS` optional. Comma-separated proxy list URLs replacing the default TheSpeedX lists.
+- `PROXY_POOL_CONCURRENCY` optional. Cap on concurrent proxy checks (default 500).
 - `DOMAIN` optional. Used for the `/` redirect target.
 - `DISCORD_WEBHOOK` optional. Enables `/api/test` and provider error reporting to Discord.
 

docs/architecture.md

@@ -23,9 +23,10 @@ Hottub is a Rust server that exposes Hot Tub compatible endpoints for channel di
 1. `main` loads `.env` and ensures `RUST_LOG` is set.
 2. It creates the Diesel SQLite pool from `DATABASE_URL`.
 3. It creates a shared `Requester`, enables Burp proxying when `PROXY != 0`, and builds the LRU video cache.
-4. It configures provider runtime validation in `providers::configure_runtime_validation`.
+4. It starts the public-proxy pool (`util::proxy_pool::start`) unless `PROXY_POOL=0`.
-5. It spawns a background thread that forces provider initialization via `providers::init_providers_now()`.
+5. It configures provider runtime validation in `providers::configure_runtime_validation`.
-6. It starts an `ntex` HTTP server on `0.0.0.0:18080`.
+6. It spawns a background thread that forces provider initialization via `providers::init_providers_now()`.
+7. It starts an `ntex` HTTP server on `0.0.0.0:18080`.
 
 ## Runtime Environment
 
@@ -36,6 +37,9 @@ Important environment variables:
 - `PROXY`: enables Burp proxying when not equal to `0`.
 - `BURP_URL`: outbound proxy URL used when `PROXY` is enabled.
 - `FLARE_URL`: FlareSolverr endpoint used as the last HTML-fetch fallback.
+- `PROXY_POOL`: `0` disables the public-proxy pool (default on).
+- `PROXY_POOL_LISTS`: comma-separated proxy list URLs replacing the default TheSpeedX lists.
+- `PROXY_POOL_CONCURRENCY`: cap on concurrent proxy checks (default 500).
 - `DOMAIN`: used by the `/` redirect target.
 - `DISCORD_WEBHOOK`: enables `/api/test` and provider error reporting.
 
@@ -120,7 +124,9 @@ Sends a Discord error test if `DISCORD_WEBHOOK` is configured.
 
 ### `/api/proxies`
 
-Returns the background-fetched outbound proxy snapshot from `src/util/proxy.rs`.
+Status of the public-proxy pool in `src/util/proxy_pool.rs`: the last list sweep, the fastest exit IPs, and per site (for example `rule34video.com`) the proxies ready for it with their request counters (`ok`/`failed`/`throttled`) and any 429 cooldown. Proxy credentials are never shown.
+
+The pool downloads public proxy lists hourly and keeps only proxies that pass a TLS-verified check against `cdn-cgi/trace` twice with a median under 1.5s, one per exit IP. Providers register a site (`proxy_pool::register_site`) with a cheap check URL; the pool checks its candidates against that URL and rotates requests over the ready ones (`next_proxy` / `report`). rule34video spreads its detail-page enrichment over it to stay under the site's per-IP 429 limit.
 
 ## Core Data Structures
 

src/api.rs

@@ -6,7 +6,6 @@ use crate::providers::{
 use crate::uploaders::{UploaderProfile, UploadersRequest};
 use crate::util::cache::VideoCache;
 use crate::util::discord::send_discord_error_report;
-use crate::util::proxy::{Proxy, all_proxies_snapshot};
 use crate::util::requester::Requester;
 use crate::{DbPool, db, status::*, videos::*};
 use ntex::http::header;
@@ -992,26 +991,7 @@ pub async fn test() -> Result<impl web::Responder, web::Error> {
 }
 
 pub async fn proxies() -> Result<impl web::Responder, web::Error> {
-    let proxies = all_proxies_snapshot().await.unwrap_or_default();
+    Ok(web::HttpResponse::Ok().json(&crate::util::proxy_pool::status()))
-    crate::flow_debug!("proxies endpoint snapshot_count={}", proxies.len());
-    let mut by_protocol: std::collections::BTreeMap<String, Vec<Proxy>> =
-        std::collections::BTreeMap::new();
-    for proxy in proxies {
-        by_protocol
-            .entry(proxy.protocol.clone())
-            .or_default()
-            .push(proxy);
-    }
-    for proxies in by_protocol.values_mut() {
-        proxies.sort_by(|a, b| {
-            a.host
-                .cmp(&b.host)
-                .then(a.port.cmp(&b.port))
-                .then(a.username.cmp(&b.username))
-                .then(a.password.cmp(&b.password))
-        });
-    }
-    Ok(web::HttpResponse::Ok().json(&by_protocol))
 }
 
 #[cfg(test)]

src/main.rs

@@ -61,6 +61,7 @@ async fn main() -> std::io::Result<()> {
     let proxy_enabled = env::var("PROXY").unwrap_or("0".to_string()) != "0".to_string();
     requester.set_proxy(proxy_enabled);
     crate::flow_debug!("requester initialized proxy_enabled={}", proxy_enabled);
+    util::proxy_pool::start();
 
     let cache: util::cache::VideoCache = crate::util::cache::VideoCache::new()
         .max_size(100_000)

src/providers/rule34video.rs

@@ -5,8 +5,11 @@ use crate::status::*;
 use crate::util::cache::VideoCache;
 use crate::util::discord::send_discord_error_report;
 use crate::util::parse_abbreviated_number;
+use crate::util::proxy_pool::{self, Outcome};
+use crate::util::requester::Requester;
+use crate::util::rule34video_resolve;
 use crate::util::time::parse_time_to_seconds;
-use crate::videos::{ServerOptions, VideoItem};
+use crate::videos::{ServerOptions, VideoFormat, VideoItem};
 use async_trait::async_trait;
 use error_chain::error_chain;
 use htmlentity::entity::{ICodedDataTrait, decode};
@@ -20,6 +23,18 @@ use std::vec;
 /// Hands out `/proxy/rule34video/` URLs instead of the page URLs. Switched off;
 /// the proxy and resolver stay in place (`proxies/rule34video.rs`).
 const USE_MEDIA_PROXY: bool = false;
+/// The signed `v-acctoken` format URLs last ~12h; keep details well inside that.
+const DETAIL_TTL: Duration = Duration::from_secs(2 * 60 * 60);
+const DETAIL_CACHE_SIZE: usize = 5000;
+/// Proxy-pool site key; its proxies are checked against `robots.txt`.
+const POOL_HOST: &str = "rule34video.com";
+const POOL_CHECK_URL: &str = "https://rule34video.com/robots.txt";
+const ENRICH_CONCURRENCY: usize = 8;
+/// A listing waits this long for its detail pages, then goes out regardless.
+const ENRICH_DEADLINE: Duration = Duration::from_secs(4);
+const DETAIL_TIMEOUT: Duration = Duration::from_secs(3);
+/// A failed detail fetch is retried on another proxy only with this much time left.
+const RETRY_MIN_LEFT: Duration = Duration::from_secs(1);
 const TARGET_TTL: Duration = Duration::from_secs(6 * 60 * 60);
 const TARGET_CACHE_SIZE: usize = 2000;
 /// A probed archive smaller than this loses to keyword search: "elf girl" is a
@@ -71,14 +86,23 @@ impl Target {
 #[derive(Clone)]
 pub struct Rule34videoProvider {
     url: String,
+    /// Enriched items (formats, tags, artist) by video URL. Every listing costs
+    /// ~23 detail pages, again for the prefetched next page.
+    details: VideoCache,
     /// Bare query -> resolved target, so later pages skip the archive probes.
     targets: Arc<Mutex<HashMap<String, (Instant, Target)>>>,
 }
 
 impl Rule34videoProvider {
     pub fn new() -> Self {
+        let mut details = VideoCache::new();
+        details.max_size(DETAIL_CACHE_SIZE);
+        // Registered at construction so the pool checks proxies against the
+        // site as soon as it has candidates, before the first listing.
+        proxy_pool::register_site(POOL_HOST, POOL_CHECK_URL);
         Rule34videoProvider {
             url: "https://rule34video.com".to_string(),
+            details,
             targets: Arc::new(Mutex::new(HashMap::new())),
         }
     }
@@ -314,6 +338,7 @@ impl Rule34videoProvider {
             }
         };
         let video_items = self.get_video_items_from_html(text);
+        let video_items = self.enrich_items(video_items, requester.clone()).await;
         let video_items = Self::proxy_items(video_items, &options);
         if video_items.is_empty() {
             return Ok(stale());
@@ -395,6 +420,7 @@ impl Rule34videoProvider {
             "".to_string()
         });
         let video_items = self.get_video_items_from_html(text);
+        let video_items = self.enrich_items(video_items, requester.clone()).await;
         let video_items = Self::proxy_items(video_items, &options);
 
         if !video_items.is_empty() {
@@ -468,6 +494,7 @@ impl Rule34videoProvider {
             "".to_string()
         });
         let video_items = self.get_video_items_from_html(text);
+        let video_items = self.enrich_items(video_items, requester.clone()).await;
         let video_items = Self::proxy_items(video_items, &options);
 
         if !video_items.is_empty() {
@@ -481,8 +508,191 @@ impl Rule34videoProvider {
         }
     }
 
-    /// Playback resolves the video page on demand through `/proxy/rule34video/`,
+    /// Fills formats, tags and the artist from a video page; `None` when the
-    /// instead of fetching every listed page up front (which ran into 429s).
+    /// page carries no `flashvars` (challenge, error page).
+    fn apply_detail(&self, mut item: VideoItem, html: &str) -> Option<VideoItem> {
+        let formats: Vec<VideoFormat> = rule34video_resolve::parse_formats(html)
+            .into_iter()
+            .map(|(height, url)| {
+                let quality = match height {
+                    0 => "auto".to_string(),
+                    h => format!("{h}p"),
+                };
+                let format = VideoFormat::new(url, quality.clone(), "mp4".to_string())
+                    .format_id(format!("mp4-{quality}"));
+                match height {
+                    0 => format,
+                    h => format.height(h),
+                }
+            })
+            .collect();
+        if formats.is_empty() {
+            return None;
+        }
+        item.formats = Some(formats);
+
+        let mut tags = rule34video_resolve::parse_tags(html);
+        let artists = Self::parse_artists(html);
+        if let Some((name, slug)) = artists.first() {
+            item.uploader = Some(name.clone());
+            item.uploaderUrl = Some(format!("{}/models/{slug}/", self.url));
+            item.uploaderId = Some(format!("rule34video:{slug}"));
+        }
+        for (name, _) in &artists {
+            if !tags.iter().any(|t| t.eq_ignore_ascii_case(name)) {
+                tags.push(name.clone());
+            }
+        }
+        if !tags.is_empty() {
+            item.tags = Some(tags);
+        }
+        Some(item)
+    }
+
+    /// Artists credited on a video page: `(name, slug)` from the "Artist" group.
+    fn parse_artists(html: &str) -> Vec<(String, String)> {
+        let document = Html::parse_document(html);
+        let Ok(group_sel) = Selector::parse("[data-suggest-type=\"model\"] a[href*=\"/models/\"]")
+        else {
+            return vec![];
+        };
+        let name_sel = Selector::parse(".name").ok();
+        let mut artists: Vec<(String, String)> = Vec::new();
+        for link in document.select(&group_sel) {
+            let href = link.value().attr("href").unwrap_or_default();
+            let slug = href
+                .split("/models/")
+                .nth(1)
+                .unwrap_or_default()
+                .trim_matches('/')
+                .to_string();
+            let name = name_sel
+                .as_ref()
+                .and_then(|sel| link.select(sel).next())
+                .map(|n| n.text().collect::<String>())
+                .unwrap_or_else(|| link.text().collect::<String>());
+            let name = name.split_whitespace().collect::<Vec<_>>().join(" ");
+            if slug.is_empty() || name.is_empty() || artists.iter().any(|(_, s)| *s == slug) {
+                continue;
+            }
+            artists.push((name, slug));
+        }
+        artists
+    }
+
+
+    /// Fetches the detail pages of a listing through the proxy pool, spread over
+    /// its proxies so the site's per-IP 429 limit is never reached. Returns at
+    /// the deadline with whatever arrived; the rest keep running in the
+    /// background and land in the detail cache for the next time the listing
+    /// is served. Without ready proxies the listing data is returned as is.
+    async fn enrich_items(&self, items: Vec<VideoItem>, requester: Requester) -> Vec<VideoItem> {
+        let mut items = self.with_details(items);
+        let pending: Vec<usize> = items
+            .iter()
+            .enumerate()
+            .filter(|(_, item)| item.formats.is_none() && !item.url.is_empty())
+            .map(|(index, _)| index)
+            .collect();
+        if pending.is_empty() || !proxy_pool::is_ready(POOL_HOST) {
+            return items;
+        }
+
+        let deadline = tokio::time::Instant::now() + ENRICH_DEADLINE;
+        let permits = Arc::new(tokio::sync::Semaphore::new(ENRICH_CONCURRENCY));
+        let handles: Vec<(usize, tokio::task::JoinHandle<Option<VideoItem>>)> = pending
+            .into_iter()
+            .map(|index| {
+                let provider = self.clone();
+                let item = items[index].clone();
+                let requester = requester.clone();
+                let permits = permits.clone();
+                let task = tokio::spawn(async move {
+                    let _permit = permits.acquire_owned().await.ok()?;
+                    let html = Self::fetch_detail(&requester, &item.url, deadline).await?;
+                    let enriched = provider.apply_detail(item, &html)?;
+                    provider.store_detail(&enriched);
+                    Some(enriched)
+                });
+                (index, task)
+            })
+            .collect();
+        // A timed-out wait drops only the handle; the task itself runs on.
+        for (index, handle) in handles {
+            if let Ok(Ok(Some(enriched))) = tokio::time::timeout_at(deadline, handle).await {
+                items[index] = enriched;
+            }
+        }
+        items
+    }
+
+    /// One video page through the next pool proxy, retried once on another
+    /// proxy while at least `RETRY_MIN_LEFT` of the deadline remains.
+    async fn fetch_detail(
+        requester: &Requester,
+        url: &str,
+        deadline: tokio::time::Instant,
+    ) -> Option<String> {
+        for attempt in 0..2 {
+            let left = deadline.saturating_duration_since(tokio::time::Instant::now());
+            if attempt > 0 && left < RETRY_MIN_LEFT {
+                return None;
+            }
+            let proxy = proxy_pool::next_proxy(POOL_HOST)?;
+            let headers = vec![("Referer".to_string(), "https://rule34video.com/".to_string())];
+            let (outcome, page) = match requester.get_via(&proxy, url, headers, DETAIL_TIMEOUT).await {
+                Err(_) => (Outcome::Failed, None),
+                Ok(response) => {
+                    let status = response.status().as_u16();
+                    match response.text().await {
+                        Err(_) => (Outcome::Failed, None),
+                        Ok(body) if status == 429 || Requester::looks_like_cloudflare_challenge(&body) => {
+                            (Outcome::Throttled, None)
+                        }
+                        Ok(body) if body.contains("var flashvars") => (Outcome::Ok, Some(Some(body))),
+                        // The video is gone; the proxy did its job.
+                        Ok(_) if status == 404 || status == 410 => (Outcome::Ok, Some(None)),
+                        Ok(_) => (Outcome::Failed, None),
+                    }
+                }
+            };
+            proxy_pool::report(POOL_HOST, &proxy, outcome);
+            if let Some(page) = page {
+                return page;
+            }
+        }
+        None
+    }
+
+    fn store_detail(&self, item: &VideoItem) {
+        self.details.insert(item.url.clone(), vec![item.clone()]);
+    }
+
+    /// Copies cached details (formats, tags, artist) onto listing items. Runs on
+    /// every served listing, so details that arrived after a listing's deadline
+    /// show up the next time it is served from cache.
+    fn with_details(&self, mut items: Vec<VideoItem>) -> Vec<VideoItem> {
+        for item in &mut items {
+            let Some((time, cached)) = self.details.get(&item.url) else {
+                continue;
+            };
+            if time.elapsed().unwrap_or(DETAIL_TTL) >= DETAIL_TTL {
+                continue;
+            }
+            let Some(cached) = cached.into_iter().next() else {
+                continue;
+            };
+            item.formats = cached.formats;
+            item.tags = cached.tags;
+            item.uploader = cached.uploader;
+            item.uploaderUrl = cached.uploaderUrl;
+            item.uploaderId = cached.uploaderId;
+        }
+        items
+    }
+
+    /// With `USE_MEDIA_PROXY`, playback resolves the video page on demand through
+    /// `/proxy/rule34video/` instead of using the enriched formats.
     fn proxy_items(mut items: Vec<VideoItem>, options: &ServerOptions) -> Vec<VideoItem> {
         if !USE_MEDIA_PROXY {
             return items;
@@ -613,7 +823,8 @@ impl Provider for Rule34videoProvider {
         };
 
         match result {
-            Ok(v) => v,
+            // Cached listings pick up details that arrived after their deadline.
+            Ok(v) => self.with_details(v),
             Err(e) => {
                 eprintln!("Error fetching videos: {}", e);
                 vec![]
@@ -660,6 +871,80 @@ mod tests {
         assert_eq!(items[0].duration, 88);
     }
 
+    #[test]
+    fn served_listings_pick_up_details_that_arrived_after_the_deadline() {
+        let provider = Rule34videoProvider::new();
+        let bare = VideoItem::new(
+            "42".to_string(),
+            "Some Title".to_string(),
+            "https://rule34video.com/video/42/some-title/".to_string(),
+            "Rule34video".to_string(),
+            String::new(),
+            88,
+        );
+        let mut late = bare.clone();
+        late.tags = Some(vec!["3D".to_string()]);
+        late.formats = Some(vec![VideoFormat::new(
+            "https://rule34video.com/get_file/a_720p.mp4/".to_string(),
+            "720p".to_string(),
+            "mp4".to_string(),
+        )]);
+        provider.store_detail(&late);
+
+        let served = provider.with_details(vec![bare]);
+        assert_eq!(served[0].tags, Some(vec!["3D".to_string()]));
+        assert_eq!(served[0].formats.as_ref().map(Vec::len), Some(1));
+    }
+
+    #[test]
+    fn a_detail_page_fills_formats_tags_and_artist() {
+        let html = r#"<div class="col vp-group" data-suggest-type="model"><div class="label">Artist</div>
+            <span class="chip"><a class="item btn_link" href="https://rule34video.com/models/joigang-va/">
+            <span class="name">JOIGang (VA)</span></a></span></div>
+            <script>var flashvars = { video_id: '1', video_categories: '3D, Cyberpunk 2077', video_tags: 'bondage, 3d', video_url: 'https://rule34video.com/get_file/a_360.mp4/?v-acctoken=1', video_url_text: '360p', video_alt_url2: 'https://rule34video.com/get_file/a_720p.mp4/?v-acctoken=3', video_alt_url2_text: '720p' };</script>"#;
+        let item = VideoItem::new(
+            "1".to_string(),
+            "t".to_string(),
+            "https://rule34video.com/video/1/t/".to_string(),
+            "Rule34video".to_string(),
+            String::new(),
+            0,
+        );
+        let item = Rule34videoProvider::new().apply_detail(item, html).expect("flashvars present");
+        let formats = item.formats.expect("formats");
+        assert_eq!(formats.len(), 2);
+        assert!(formats[0].url.contains("a_720p"), "best quality first");
+        assert_eq!(
+            item.tags,
+            Some(vec!["3D".into(), "Cyberpunk 2077".into(), "bondage".into(), "JOIGang (VA)".into()])
+        );
+        assert_eq!(item.uploader.as_deref(), Some("JOIGang (VA)"));
+        assert_eq!(item.uploaderUrl.as_deref(), Some("https://rule34video.com/models/joigang-va/"));
+        assert!(Rule34videoProvider::new().apply_detail(VideoItem::new(
+            "1".into(), "t".into(), "u".into(), "Rule34video".into(), String::new(), 0,
+        ), "<html>DDoS-Guard</html>").is_none());
+    }
+
+    #[tokio::test]
+    async fn without_ready_proxies_enrichment_is_skipped_not_done_directly() {
+        let provider = Rule34videoProvider::new();
+        let item = VideoItem::new(
+            "1".to_string(),
+            "t".to_string(),
+            "https://rule34video.com/video/1/t/".to_string(),
+            "Rule34video".to_string(),
+            String::new(),
+            0,
+        );
+        let started = std::time::Instant::now();
+        let items = provider
+            .enrich_items(vec![item], crate::util::requester::Requester::new())
+            .await;
+        assert!(started.elapsed() < Duration::from_millis(500), "no request was made");
+        assert_eq!(items.len(), 1);
+        assert!(items[0].formats.is_none());
+    }
+
     #[test]
     fn parses_prefixed_queries() {
         assert_eq!(

src/util/mod.rs

@@ -9,6 +9,7 @@ pub mod flow_debug;
 pub mod hanime_resolve;
 pub mod hoster_proxy;
 pub mod proxy;
+pub mod proxy_pool;
 pub mod requester;
 pub mod rule34video_resolve;
 pub mod time;

src/util/proxy.rs

@@ -1,26 +1,9 @@
 use std::fmt;
 use std::str::FromStr;
-use std::sync::Arc;
 
 use serde::Serialize;
-use tokio::sync::{OnceCell, RwLock};
-use tokio::time::{self, Duration};
 use url::Url;
 use wreq::Proxy as WreqProxy;
-use wreq::{Client, Version};
-
-use crate::util::requester::Requester;
-
-pub static ALL_PROXIES: OnceCell<Arc<RwLock<Vec<Proxy>>>> = OnceCell::const_new();
-static ALL_PROXY_KEYS: OnceCell<Arc<RwLock<std::collections::HashSet<String>>>> =
-    OnceCell::const_new();
-
-const PROXY_LIST: [&str; 3] = [
-    "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt",
-    "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt",
-    "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks4.txt",
-];
-const IFCONFIG_URL: &str = "https://ifconfig.co";
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct Proxy {
@@ -44,12 +27,6 @@ impl ProxyParseError {
     }
 }
 
-pub async fn all_proxies_snapshot() -> Option<Vec<Proxy>> {
-    let handle = ALL_PROXIES.get()?.clone();
-    let proxies = handle.read().await;
-    Some(proxies.clone())
-}
-
 impl fmt::Display for ProxyParseError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.message)
@@ -78,6 +55,11 @@ impl fmt::Display for Proxy {
 }
 
 impl Proxy {
+    /// `protocol://host:port`, without credentials: safe to show or log.
+    pub fn redacted(&self) -> String {
+        format!("{}://{}:{}", self.protocol, self.host, self.port)
+    }
+
     pub fn parse(input: &str) -> Result<Self, ProxyParseError> {
         input.parse()
     }
@@ -99,147 +81,6 @@ impl Proxy {
     }
 }
 
-pub async fn fetch_proxies(
-    requester: &mut Requester,
-    proxy_list_url: &str,
-    proxy_sink: Arc<RwLock<Vec<Proxy>>>,
-) -> Result<usize, ProxyParseError> {
-    let default_protocol = protocol_from_list_url(proxy_list_url)?;
-    let body = requester
-        .get(proxy_list_url, None)
-        .await
-        .map_err(|e| ProxyParseError::new(format!("failed to fetch proxy list: {e}")))?;
-    let mut tasks = tokio::task::JoinSet::new();
-    for (line_index, line) in body.lines().enumerate() {
-        let trimmed = line.trim();
-        if trimmed.is_empty() {
-            continue;
-        }
-
-        let proxy_input = if trimmed.contains("://") {
-            trimmed.to_string()
-        } else {
-            format!("{default_protocol}://{trimmed}")
-        };
-
-        let proxy = Proxy::parse(&proxy_input).map_err(|e| {
-            ProxyParseError::new(format!("invalid proxy on line {}: {e}", line_index + 1))
-        })?;
-
-        tasks.spawn(async move { verify_proxy(proxy).await });
-    }
-
-    let mut added = 0usize;
-    while let Some(result) = tasks.join_next().await {
-        match result {
-            Ok(Ok(proxy)) => {
-                if let Some(proxy_keys) = ALL_PROXY_KEYS.get() {
-                    let key = proxy_key(&proxy);
-                    let mut keys = proxy_keys.write().await;
-                    if !keys.insert(key) {
-                        continue;
-                    }
-                    let mut proxies = proxy_sink.write().await;
-                    proxies.push(proxy);
-                    added += 1;
-                }
-            }
-            Ok(Err(_err)) => {
-                // eprintln!("Proxy verification failed: {err}");
-            }
-            Err(err) => {
-                eprintln!("Proxy verification task failed: {err}");
-            }
-        }
-    }
-
-    Ok(added)
-}
-
-fn protocol_from_list_url(list_url: &str) -> Result<&'static str, ProxyParseError> {
-    if list_url.contains("http.txt") {
-        Ok("http")
-    } else if list_url.contains("socks4.txt") {
-        Ok("socks4")
-    } else if list_url.contains("socks5.txt") {
-        Ok("socks5")
-    } else {
-        Err(ProxyParseError::new(format!(
-            "unknown proxy list protocol for url: {list_url}"
-        )))
-    }
-}
-
-async fn verify_proxy(proxy: Proxy) -> Result<Proxy, ProxyParseError> {
-    let wreq_proxy = proxy.to_wreq_proxy()?;
-    let client = Client::builder()
-        .tls_cert_verification(false)
-        .build()
-        .map_err(|e| ProxyParseError::new(format!("failed to build http client: {e}")))?;
-
-    let response = client
-        .get(IFCONFIG_URL)
-        .version(Version::HTTP_11)
-        .proxy(wreq_proxy)
-        .timeout(Duration::from_secs(5))
-        .send()
-        .await
-        .map_err(|e| ProxyParseError::new(format!("proxy request failed: {e}")))?;
-
-    if response.status().is_success() {
-        Ok(proxy)
-    } else {
-        Err(ProxyParseError::new(format!(
-            "proxy returned status {}",
-            response.status()
-        )))
-    }
-}
-
-pub fn init_all_proxies_background(requester: Requester) {
-    if ALL_PROXIES.get().is_some() {
-        return;
-    }
-
-    if tokio::runtime::Handle::try_current().is_err() {
-        eprintln!("Skipping proxy list init: no Tokio runtime available");
-        return;
-    }
-
-    let proxy_cache = Arc::new(RwLock::new(Vec::new()));
-    let proxy_keys = Arc::new(RwLock::new(std::collections::HashSet::new()));
-    if ALL_PROXIES.set(proxy_cache.clone()).is_err() {
-        return;
-    }
-    let _ = ALL_PROXY_KEYS.set(proxy_keys);
-
-    tokio::spawn(async move {
-        let mut interval = time::interval(Duration::from_secs(60 * 60));
-        loop {
-            let mut tasks = tokio::task::JoinSet::new();
-            for list in PROXY_LIST {
-                let proxy_cache = proxy_cache.clone();
-                let mut requester = requester.clone();
-                tasks.spawn(async move { fetch_proxies(&mut requester, list, proxy_cache).await });
-            }
-
-            while let Some(result) = tasks.join_next().await {
-                match result {
-                    Ok(Ok(_added)) => {}
-                    Ok(Err(err)) => {
-                        eprintln!("Failed to fetch proxy list: {err}");
-                    }
-                    Err(err) => {
-                        eprintln!("Proxy list task failed: {err}");
-                    }
-                }
-            }
-
-            interval.tick().await;
-        }
-    });
-}
-
 impl FromStr for Proxy {
     type Err = ProxyParseError;
 
@@ -261,7 +102,6 @@ impl FromStr for Proxy {
             || url.query().is_some()
             || url.fragment().is_some()
         {
-            println!("Parsed proxy URL: {:?}", url);
             return Err(ProxyParseError::new(format!(
                 "proxy url must not include path, query, or fragment: {:?}",
                 input
@@ -287,14 +127,3 @@ impl FromStr for Proxy {
         })
     }
 }
-
-fn proxy_key(proxy: &Proxy) -> String {
-    format!(
-        "{}://{}:{}@{}:{}",
-        proxy.protocol,
-        proxy.username.as_deref().unwrap_or(""),
-        proxy.password.as_deref().unwrap_or(""),
-        proxy.host,
-        proxy.port
-    )
-}

src/util/proxy_pool.rs

@@ -0,0 +1,757 @@
+//! Pool of fast public proxies that rate-limited providers spread requests over.
+
+use std::collections::{HashMap, VecDeque};
+use std::time::{Duration, Instant};
+
+use crate::util::proxy::Proxy;
+
+/// A proxy only counts once it has answered this many checks: many listed
+/// proxies answer once and are gone a minute later.
+const MIN_SUCCESSES: u32 = 2;
+/// Proxies slower than this (median) are not worth waiting for.
+const LATENCY_CEILING_SECS: f64 = 1.5;
+/// Latency samples kept per proxy for the median.
+const SAMPLE_WINDOW: usize = 5;
+/// Consecutive failed checks or requests after which a proxy is out.
+const MAX_CONSECUTIVE_FAILURES: u32 = 2;
+
+struct Entry {
+    proxy: Proxy,
+    successes: u32,
+    samples: VecDeque<f64>,
+    exit_ip: Option<String>,
+    consecutive_failures: u32,
+    benched_until: Option<Instant>,
+    /// Real requests sent through this proxy, for the status page.
+    requests: RequestCounts,
+}
+
+#[derive(Default, Clone, Copy, serde::Serialize)]
+pub struct RequestCounts {
+    pub ok: u64,
+    pub failed: u64,
+    pub throttled: u64,
+}
+
+impl Entry {
+    fn median(&self) -> Option<f64> {
+        let mut sorted: Vec<f64> = self.samples.iter().copied().collect();
+        sorted.sort_by(f64::total_cmp);
+        sorted.get(sorted.len() / 2).copied()
+    }
+}
+
+/// Health record of every proxy that has passed at least one check.
+#[derive(Default)]
+pub(crate) struct Ranking {
+    entries: HashMap<String, Entry>,
+}
+
+impl Ranking {
+    pub(crate) fn record_success(&mut self, proxy: &Proxy, secs: f64, exit_ip: Option<&str>) {
+        let entry = self.entries.entry(proxy.to_string()).or_insert_with(|| Entry {
+            proxy: proxy.clone(),
+            successes: 0,
+            samples: VecDeque::new(),
+            exit_ip: None,
+            consecutive_failures: 0,
+            benched_until: None,
+            requests: RequestCounts::default(),
+        });
+        entry.successes += 1;
+        entry.consecutive_failures = 0;
+        if exit_ip.is_some() {
+            entry.exit_ip = exit_ip.map(str::to_string);
+        }
+        if entry.samples.len() == SAMPLE_WINDOW {
+            entry.samples.pop_front();
+        }
+        entry.samples.push_back(secs);
+    }
+
+    pub(crate) fn record_failure(&mut self, proxy: &Proxy) {
+        if let Some(entry) = self.entries.get_mut(&proxy.to_string()) {
+            entry.consecutive_failures += 1;
+        }
+    }
+
+    /// Keeps a working proxy out of rotation until `until` without counting it as broken.
+    pub(crate) fn bench(&mut self, proxy: &Proxy, until: Instant) {
+        if let Some(entry) = self.entries.get_mut(&proxy.to_string()) {
+            entry.benched_until = Some(until);
+        }
+    }
+
+    fn count(&mut self, proxy: &Proxy, outcome: Outcome) {
+        if let Some(entry) = self.entries.get_mut(&proxy.to_string()) {
+            match outcome {
+                Outcome::Ok => entry.requests.ok += 1,
+                Outcome::Failed => entry.requests.failed += 1,
+                Outcome::Throttled => entry.requests.throttled += 1,
+            }
+        }
+    }
+
+    /// Status rows for the usable proxies, fastest first. Benched proxies are
+    /// listed too, with the time they still sit out.
+    pub(crate) fn status(&self, now: Instant, limit: usize) -> Vec<ProxyStatus> {
+        self.usable(now, limit, true)
+            .into_iter()
+            .map(|(median, entry)| ProxyStatus {
+                proxy: entry.proxy.redacted(),
+                exit_ip: entry.exit_ip.clone(),
+                median_ms: (median * 1000.0).round() as u64,
+                consecutive_failures: entry.consecutive_failures,
+                requests: entry.requests,
+                benched_for_secs: entry
+                    .benched_until
+                    .filter(|until| *until > now)
+                    .map(|until| (until - now).as_secs()),
+            })
+            .collect()
+    }
+
+    /// Usable proxies, fastest first.
+    pub(crate) fn ranked(&self, now: Instant, limit: usize) -> Vec<Proxy> {
+        self.usable(now, limit, false)
+            .into_iter()
+            .map(|(_, entry)| entry.proxy.clone())
+            .collect()
+    }
+
+    fn usable(&self, now: Instant, limit: usize, include_benched: bool) -> Vec<(f64, &Entry)> {
+        let mut usable: Vec<(f64, &Entry)> = self
+            .entries
+            .values()
+            .filter(|entry| entry.successes >= MIN_SUCCESSES)
+            .filter(|entry| entry.consecutive_failures < MAX_CONSECUTIVE_FAILURES)
+            .filter(|entry| {
+                include_benched || entry.benched_until.is_none_or(|until| until <= now)
+            })
+            .filter_map(|entry| entry.median().map(|median| (median, entry)))
+            .filter(|(median, _)| *median < LATENCY_CEILING_SECS)
+            .collect();
+        usable.sort_by(|a, b| a.0.total_cmp(&b.0));
+        // Many list entries are ports of one host: spreading requests over them
+        // would still come from one IP, so keep only the fastest per exit IP.
+        let mut seen_ips = std::collections::HashSet::new();
+        usable
+            .into_iter()
+            .filter(|(_, entry)| entry.exit_ip.as_ref().is_none_or(|ip| seen_ips.insert(ip.clone())))
+            .take(limit)
+            .collect()
+    }
+
+    /// Forgets proxies that are out (repeated failures), so the record stays small.
+    pub(crate) fn prune(&mut self) {
+        self.entries
+            .retain(|_, entry| entry.consecutive_failures < MAX_CONSECUTIVE_FAILURES);
+    }
+
+    fn exit_ip(&self, proxy: &Proxy) -> Option<String> {
+        self.entries.get(&proxy.to_string())?.exit_ip.clone()
+    }
+}
+
+#[derive(serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProxyStatus {
+    /// `protocol://host:port`, credentials left out.
+    pub proxy: String,
+    pub exit_ip: Option<String>,
+    pub median_ms: u64,
+    pub consecutive_failures: u32,
+    pub requests: RequestCounts,
+    pub benched_for_secs: Option<u64>,
+}
+
+/// Proxies in a list file; bare `host:port` lines take the protocol from the
+/// file name (`socks5.txt`, ...), as the TheSpeedX lists do.
+pub(crate) fn parse_list(list_url: &str, body: &str) -> Vec<Proxy> {
+    let name = list_url.rsplit('/').next().unwrap_or_default();
+    let protocol = ["socks5", "socks4", "http"]
+        .into_iter()
+        .find(|p| name.contains(p))
+        .unwrap_or("http");
+    body.lines()
+        .map(str::trim)
+        .filter(|line| !line.is_empty())
+        .filter_map(|line| match line.contains("://") {
+            true => Proxy::parse(line).ok(),
+            false => Proxy::parse(&format!("{protocol}://{line}")).ok(),
+        })
+        .collect()
+}
+
+/// The `ip=` line of `cdn-cgi/trace`: the address the target saw.
+pub(crate) fn trace_exit_ip(body: &str) -> Option<String> {
+    body.lines()
+        .find_map(|line| line.strip_prefix("ip="))
+        .map(str::trim)
+        .filter(|ip| !ip.is_empty())
+        .map(str::to_string)
+}
+
+/// What happened to a request sent through a proxy.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Outcome {
+    Ok,
+    /// Connection error, timeout or garbage: counts like a failed check.
+    Failed,
+    /// 429 or a challenge page: the proxy works, its IP is throttled there.
+    Throttled,
+}
+
+/// Proxies a site rotates over.
+pub(crate) const SITE_SET_SIZE: usize = 20;
+pub(crate) const THROTTLE_COOLDOWN: Duration = Duration::from_secs(10 * 60);
+
+/// A site's own view of the pool: proxies checked against the site itself.
+pub(crate) struct Site {
+    pub(crate) check_url: String,
+    pub(crate) ranking: Ranking,
+    cursor: usize,
+}
+
+impl Site {
+    pub(crate) fn new(check_url: &str) -> Self {
+        Site {
+            check_url: check_url.to_string(),
+            ranking: Ranking::default(),
+            cursor: 0,
+        }
+    }
+
+    /// The next proxy in round-robin order, so requests spread evenly.
+    pub(crate) fn next(&mut self, now: Instant) -> Option<Proxy> {
+        let ready = self.ranking.ranked(now, SITE_SET_SIZE);
+        if ready.is_empty() {
+            return None;
+        }
+        let proxy = ready[self.cursor % ready.len()].clone();
+        self.cursor = self.cursor.wrapping_add(1);
+        Some(proxy)
+    }
+
+    pub(crate) fn report(&mut self, proxy: &Proxy, outcome: Outcome, now: Instant) {
+        self.ranking.count(proxy, outcome);
+        match outcome {
+            Outcome::Ok => {}
+            Outcome::Failed => self.ranking.record_failure(proxy),
+            Outcome::Throttled => self.ranking.bench(proxy, now + THROTTLE_COOLDOWN),
+        }
+    }
+}
+
+
+// ---------------------------------------------------------------------------
+// The live pool: one process-wide instance, fed by a background thread.
+// ---------------------------------------------------------------------------
+
+const DEFAULT_LISTS: [&str; 3] = [
+    "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt",
+    "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt",
+    "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks4.txt",
+];
+/// Tiny, served from Cloudflare's edge everywhere, and it echoes the exit IP.
+const TRACE_URL: &str = "https://www.cloudflare.com/cdn-cgi/trace";
+const CHECK_TIMEOUT: Duration = Duration::from_secs(3);
+const LIST_TIMEOUT: Duration = Duration::from_secs(30);
+const SWEEP_INTERVAL: Duration = Duration::from_secs(60 * 60);
+const RECHECK_INTERVAL: Duration = Duration::from_secs(5 * 60);
+const DEFAULT_CONCURRENCY: usize = 500;
+/// Fastest exit IPs listed on the status page. All proxies under the latency
+/// ceiling stay in the pool and are rechecked; sites draw from all of them.
+const FAST_SET_SIZE: usize = 20;
+
+#[derive(Default)]
+struct PoolState {
+    enabled: bool,
+    generic: Ranking,
+    sites: HashMap<String, Site>,
+    last_sweep: Option<SweepStats>,
+}
+
+#[derive(Clone, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SweepStats {
+    pub finished_unix: u64,
+    pub candidates: usize,
+    pub passed: usize,
+    pub fast: usize,
+    pub duration_secs: u64,
+}
+
+#[derive(serde::Serialize)]
+pub struct SiteStatus {
+    pub ready: usize,
+    pub proxies: Vec<ProxyStatus>,
+}
+
+#[derive(serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PoolStatus {
+    pub enabled: bool,
+    pub last_sweep: Option<SweepStats>,
+    pub fast: Vec<ProxyStatus>,
+    pub sites: std::collections::BTreeMap<String, SiteStatus>,
+}
+
+fn state() -> &'static std::sync::Mutex<PoolState> {
+    static STATE: std::sync::OnceLock<std::sync::Mutex<PoolState>> = std::sync::OnceLock::new();
+    STATE.get_or_init(Default::default)
+}
+
+fn with_state<T>(f: impl FnOnce(&mut PoolState) -> T) -> T {
+    let mut guard = state().lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+    f(&mut guard)
+}
+
+/// Lets a provider draw on the pool for `host`. Records the site only (no
+/// network): its proxies are checked against `check_url` once the pool has
+/// candidates, so registering from a provider's constructor warms it up.
+pub fn register_site(host: &str, check_url: &str) {
+    with_state(|state| {
+        state
+            .sites
+            .entry(host.to_string())
+            .or_insert_with(|| Site::new(check_url));
+    });
+}
+
+/// The next proxy for `host` in round-robin order; `None` while the site has none ready.
+pub fn next_proxy(host: &str) -> Option<Proxy> {
+    with_state(|state| state.sites.get_mut(host)?.next(Instant::now()))
+}
+
+/// Whether `host` has at least one proxy to send requests through.
+pub fn is_ready(host: &str) -> bool {
+    with_state(|state| {
+        state
+            .sites
+            .get(host)
+            .is_some_and(|site| !site.ranking.ranked(Instant::now(), 1).is_empty())
+    })
+}
+
+/// Feeds a real request's outcome back, so a dead or throttled proxy leaves
+/// the rotation before its next scheduled check.
+pub fn report(host: &str, proxy: &Proxy, outcome: Outcome) {
+    with_state(|state| {
+        // Eviction and cooldown are per site; the generic record only counts.
+        state.generic.count(proxy, outcome);
+        if let Some(site) = state.sites.get_mut(host) {
+            site.report(proxy, outcome, Instant::now());
+        }
+    });
+}
+
+pub fn status() -> PoolStatus {
+    with_state(|state| {
+        let now = Instant::now();
+        PoolStatus {
+            enabled: state.enabled,
+            last_sweep: state.last_sweep.clone(),
+            fast: state.generic.status(now, FAST_SET_SIZE),
+            sites: state
+                .sites
+                .iter()
+                .map(|(host, site)| {
+                    let ready = site.ranking.ranked(now, SITE_SET_SIZE).len();
+                    let proxies = site.ranking.status(now, SITE_SET_SIZE);
+                    (host.clone(), SiteStatus { ready, proxies })
+                })
+                .collect(),
+        }
+    })
+}
+
+/// Starts the background sweeps unless `PROXY_POOL=0`. Called once from `main`.
+pub fn start() {
+    if std::env::var("PROXY_POOL").is_ok_and(|v| v == "0") {
+        println!("proxy pool disabled (PROXY_POOL=0)");
+        return;
+    }
+    with_state(|state| state.enabled = true);
+    let lists: Vec<String> = match std::env::var("PROXY_POOL_LISTS") {
+        Ok(value) if !value.trim().is_empty() => value
+            .split(',')
+            .map(|url| url.trim().to_string())
+            .filter(|url| !url.is_empty())
+            .collect(),
+        _ => DEFAULT_LISTS.iter().map(|url| url.to_string()).collect(),
+    };
+    let concurrency = std::env::var("PROXY_POOL_CONCURRENCY")
+        .ok()
+        .and_then(|v| v.parse::<usize>().ok())
+        .filter(|n| *n > 0)
+        .unwrap_or(DEFAULT_CONCURRENCY);
+
+    std::thread::spawn(move || {
+        let runtime = match tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .build()
+        {
+            Ok(runtime) => runtime,
+            Err(e) => {
+                eprintln!("proxy pool: no runtime: {e}");
+                return;
+            }
+        };
+        runtime.block_on(run(lists, concurrency));
+    });
+}
+
+/// Our own public address, as the trace endpoint sees it without a proxy.
+async fn own_exit_ip() -> String {
+    loop {
+        let direct = check(None, TRACE_URL, CHECK_TIMEOUT).await.ok();
+        if let Some(ip) = direct.and_then(|(_, body)| trace_exit_ip(&body)) {
+            return ip;
+        }
+        tokio::time::sleep(Duration::from_secs(30)).await;
+    }
+}
+
+async fn run(lists: Vec<String>, concurrency: usize) {
+    loop {
+        // Re-read every sweep: the server's own address can change.
+        let own_ip = own_exit_ip().await;
+        sweep(&lists, concurrency, &own_ip).await;
+        check_sites(concurrency).await;
+        let rounds = (SWEEP_INTERVAL.as_secs() / RECHECK_INTERVAL.as_secs()).saturating_sub(1);
+        for _ in 0..rounds {
+            tokio::time::sleep(RECHECK_INTERVAL).await;
+            let fast = with_state(|state| state.generic.ranked(Instant::now(), usize::MAX));
+            check_generic(fast, concurrency, &own_ip).await;
+            with_state(|state| state.generic.prune());
+            check_sites(concurrency).await;
+        }
+        tokio::time::sleep(RECHECK_INTERVAL).await;
+    }
+}
+
+/// Downloads the lists and checks every candidate, then the survivors again:
+/// a proxy needs two successful checks to count.
+async fn sweep(lists: &[String], concurrency: usize, own_ip: &str) {
+    let started = Instant::now();
+    let mut candidates: Vec<Proxy> = Vec::new();
+    for list in lists {
+        match check(None, list, LIST_TIMEOUT).await.ok() {
+            Some((_, body)) => candidates.extend(parse_list(list, &body)),
+            None => eprintln!("proxy pool: could not fetch list {list}"),
+        }
+    }
+    let mut seen = std::collections::HashSet::new();
+    candidates.retain(|proxy| seen.insert(proxy.to_string()));
+    let total = candidates.len();
+
+    let passed = check_generic(candidates, concurrency, own_ip).await;
+    let passed_count = passed.len();
+    check_generic(passed, concurrency, own_ip).await;
+    let fast = with_state(|state| {
+        state.generic.prune();
+        let fast = state.generic.ranked(Instant::now(), usize::MAX).len();
+        state.last_sweep = Some(SweepStats {
+            finished_unix: std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .map(|d| d.as_secs())
+                .unwrap_or(0),
+            candidates: total,
+            passed: passed_count,
+            fast,
+            duration_secs: started.elapsed().as_secs(),
+        });
+        fast
+    });
+    println!(
+        "proxy pool: sweep of {total} candidates: {passed_count} passed, {fast} fast exit IPs ({}s)",
+        started.elapsed().as_secs()
+    );
+}
+
+/// Trace-checks `proxies` and records the results; returns the ones that passed.
+async fn check_generic(proxies: Vec<Proxy>, concurrency: usize, own_ip: &str) -> Vec<Proxy> {
+    use futures::stream::{self, StreamExt};
+    let results: Vec<(Proxy, Option<(f64, String)>)> = stream::iter(proxies)
+        .map(|proxy| async move {
+            let result = check(Some(&proxy), TRACE_URL, CHECK_TIMEOUT)
+                .await
+                .ok()
+                .and_then(|(secs, body)| trace_exit_ip(&body).map(|ip| (secs, ip)))
+                // A proxy that shows our own address hides nothing.
+                .filter(|(_, ip)| ip != own_ip);
+            (proxy, result)
+        })
+        .buffer_unordered(concurrency)
+        .collect()
+        .await;
+    with_state(|state| {
+        let mut passed = Vec::new();
+        for (proxy, result) in results {
+            match result {
+                Some((secs, ip)) => {
+                    state.generic.record_success(&proxy, secs, Some(&ip));
+                    passed.push(proxy);
+                }
+                None => state.generic.record_failure(&proxy),
+            }
+        }
+        passed
+    })
+}
+
+/// Checks every registered site against all fast exit IPs (not only the top
+/// 20) plus its own current set: many proxies that pass the trace check cannot
+/// reach a given site (DDoS-Guard drops whole datacenter ranges), so the site's
+/// set is filled from the wider list. A proxy evicted from a site is replaced
+/// by the next site-checked one at the latest on the following round.
+async fn check_sites(concurrency: usize) {
+    use futures::stream::{self, StreamExt};
+    let jobs: Vec<(String, String, Vec<Proxy>)> = with_state(|state| {
+        let now = Instant::now();
+        let fast = state.generic.ranked(now, usize::MAX);
+        state
+            .sites
+            .iter()
+            .map(|(host, site)| {
+                let mut proxies = fast.clone();
+                for proxy in site.ranking.ranked(now, usize::MAX) {
+                    if !proxies.contains(&proxy) {
+                        proxies.push(proxy);
+                    }
+                }
+                (host.clone(), site.check_url.clone(), proxies)
+            })
+            .collect()
+    });
+    for (host, check_url, proxies) in jobs {
+        // Two rounds: a site proxy also needs two successful checks.
+        for _ in 0..2 {
+            let results: Vec<(Proxy, Result<f64, CheckError>)> = stream::iter(proxies.clone())
+                .map(|proxy| {
+                    let check_url = check_url.clone();
+                    async move {
+                        let secs = check(Some(&proxy), &check_url, CHECK_TIMEOUT)
+                            .await
+                            .map(|(secs, _)| secs);
+                        (proxy, secs)
+                    }
+                })
+                .buffer_unordered(concurrency)
+                .collect()
+                .await;
+            with_state(|state| {
+                let exit_ips: HashMap<String, Option<String>> = results
+                    .iter()
+                    .map(|(proxy, _)| (proxy.to_string(), state.generic.exit_ip(proxy)))
+                    .collect();
+                let Some(site) = state.sites.get_mut(&host) else {
+                    return;
+                };
+                let now = Instant::now();
+                for (proxy, result) in &results {
+                    match result {
+                        Ok(secs) => {
+                            let ip = exit_ips.get(&proxy.to_string()).cloned().flatten();
+                            site.ranking.record_success(proxy, *secs, ip.as_deref());
+                        }
+                        // The proxy works; its IP is just throttled there.
+                        Err(CheckError::Throttled) => {
+                            site.ranking.bench(proxy, now + THROTTLE_COOLDOWN)
+                        }
+                        Err(CheckError::Failed) => site.ranking.record_failure(proxy),
+                    }
+                }
+                site.ranking.prune();
+            });
+        }
+        let ready = with_state(|state| {
+            state
+                .sites
+                .get(&host)
+                .map(|site| site.ranking.ranked(Instant::now(), SITE_SET_SIZE).len())
+                .unwrap_or(0)
+        });
+        println!("proxy pool: {host}: {ready} proxies ready");
+    }
+}
+
+fn check_client() -> &'static wreq::Client {
+    static CLIENT: std::sync::OnceLock<wreq::Client> = std::sync::OnceLock::new();
+    // Certificates are verified: public proxies are untrusted, and one that
+    // tampers with TLS must fail its check rather than feed us altered pages.
+    CLIENT.get_or_init(|| {
+        wreq::Client::builder()
+            .emulation(wreq_util::Emulation::Firefox151)
+            .build()
+            .expect("proxy pool check client")
+    })
+}
+
+enum CheckError {
+    /// 429 from the target: the proxy works, its IP is rate-limited there.
+    Throttled,
+    Failed,
+}
+
+/// GET `url` (through `proxy`, if any) within `timeout`.
+/// Returns the wall time and the body of a 2xx answer.
+async fn check(
+    proxy: Option<&Proxy>,
+    url: &str,
+    timeout: Duration,
+) -> Result<(f64, String), CheckError> {
+    let mut request = check_client().get(url).timeout(timeout);
+    if let Some(proxy) = proxy {
+        request = request.proxy(proxy.to_wreq_proxy().map_err(|_| CheckError::Failed)?);
+    }
+    let started = Instant::now();
+    let response = request.send().await.map_err(|_| CheckError::Failed)?;
+    match response.status().as_u16() {
+        429 => return Err(CheckError::Throttled),
+        status if !(200..300).contains(&status) => return Err(CheckError::Failed),
+        _ => {}
+    }
+    let body = response.text().await.map_err(|_| CheckError::Failed)?;
+    Ok((started.elapsed().as_secs_f64(), body))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn proxy(s: &str) -> Proxy {
+        Proxy::parse(s).unwrap()
+    }
+
+    #[test]
+    fn a_proxy_needs_two_successful_checks_to_rank() {
+        let mut ranking = Ranking::default();
+        let p = proxy("socks5://10.0.0.1:1080");
+        ranking.record_success(&p, 0.4, Some("1.1.1.1"));
+        assert!(ranking.ranked(Instant::now(), 20).is_empty());
+        ranking.record_success(&p, 0.5, Some("1.1.1.1"));
+        assert_eq!(ranking.ranked(Instant::now(), 20), vec![p]);
+    }
+
+    #[test]
+    fn ranks_by_median_latency_and_drops_slow_proxies() {
+        let mut ranking = Ranking::default();
+        let (fast, medium, slow) = (
+            proxy("http://10.0.0.1:80"),
+            proxy("http://10.0.0.2:80"),
+            proxy("http://10.0.0.3:80"),
+        );
+        // medium has one fast outlier, but its median is 0.9s.
+        for (p, samples, ip) in [
+            (&medium, [0.1, 0.9, 1.0], "2.2.2.2"),
+            (&slow, [1.4, 1.6, 2.0], "3.3.3.3"),
+            (&fast, [0.3, 0.4, 0.2], "1.1.1.1"),
+        ] {
+            for secs in samples {
+                ranking.record_success(p, secs, Some(ip));
+            }
+        }
+        assert_eq!(ranking.ranked(Instant::now(), 20), vec![fast.clone(), medium]);
+        assert_eq!(ranking.ranked(Instant::now(), 1), vec![fast]);
+    }
+
+    #[test]
+    fn keeps_only_the_fastest_proxy_per_exit_ip() {
+        let mut ranking = Ranking::default();
+        let port_a = proxy("http://147.161.246.39:10878");
+        let port_b = proxy("http://147.161.246.39:11997");
+        let other = proxy("socks5://5.5.5.5:1080");
+        for (p, secs, ip) in [
+            (&port_a, 0.6, "147.161.247.7"),
+            (&port_b, 0.4, "147.161.247.7"),
+            (&other, 0.9, "5.5.5.5"),
+        ] {
+            ranking.record_success(p, secs, Some(ip));
+            ranking.record_success(p, secs, Some(ip));
+        }
+        assert_eq!(ranking.ranked(Instant::now(), 20), vec![port_b, other]);
+    }
+
+    #[test]
+    fn two_failures_in_a_row_evict_until_the_next_success() {
+        let mut ranking = Ranking::default();
+        let p = proxy("http://10.0.0.1:80");
+        ranking.record_success(&p, 0.3, Some("1.1.1.1"));
+        ranking.record_success(&p, 0.3, Some("1.1.1.1"));
+        ranking.record_failure(&p);
+        assert_eq!(ranking.ranked(Instant::now(), 20), vec![p.clone()], "one failure is tolerated");
+        ranking.record_failure(&p);
+        assert!(ranking.ranked(Instant::now(), 20).is_empty());
+        ranking.record_success(&p, 0.3, Some("1.1.1.1"));
+        assert_eq!(ranking.ranked(Instant::now(), 20), vec![p]);
+    }
+
+    fn site_with(proxies: &[&Proxy]) -> Site {
+        let mut site = Site::new("https://example.com/robots.txt");
+        for (i, p) in proxies.iter().enumerate() {
+            let ip = format!("9.9.9.{i}");
+            site.ranking.record_success(p, 0.2 + i as f64 / 10.0, Some(&ip));
+            site.ranking.record_success(p, 0.2 + i as f64 / 10.0, Some(&ip));
+        }
+        site
+    }
+
+    #[test]
+    fn site_requests_rotate_evenly_over_its_proxies() {
+        let (a, b, c) = (
+            proxy("http://10.0.0.1:80"),
+            proxy("http://10.0.0.2:80"),
+            proxy("http://10.0.0.3:80"),
+        );
+        let mut site = site_with(&[&a, &b, &c]);
+        let now = Instant::now();
+        let picks: Vec<Proxy> = (0..6).filter_map(|_| site.next(now)).collect();
+        assert_eq!(picks, vec![a.clone(), b.clone(), c.clone(), a, b, c]);
+    }
+
+    #[test]
+    fn a_throttled_proxy_sits_out_its_cooldown_for_that_site() {
+        let (a, b) = (proxy("http://10.0.0.1:80"), proxy("http://10.0.0.2:80"));
+        let mut site = site_with(&[&a, &b]);
+        let now = Instant::now();
+        site.report(&a, Outcome::Throttled, now);
+        let picks: Vec<Proxy> = (0..3).filter_map(|_| site.next(now)).collect();
+        assert_eq!(picks, vec![b.clone(), b.clone(), b.clone()]);
+        let later = now + THROTTLE_COOLDOWN + std::time::Duration::from_secs(1);
+        let picks: Vec<Proxy> = (0..2).filter_map(|_| site.next(later)).collect();
+        assert!(picks.contains(&a) && picks.contains(&b));
+    }
+
+    #[test]
+    fn list_lines_take_their_protocol_from_the_list_name_unless_they_carry_one() {
+        let body = "1.2.3.4:1080\n\n socks4://5.6.7.8:4145 \nnot a proxy line:::\n";
+        assert_eq!(
+            parse_list("https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt", body),
+            vec![proxy("socks5://1.2.3.4:1080"), proxy("socks4://5.6.7.8:4145")]
+        );
+    }
+
+    #[test]
+    fn reads_the_exit_ip_from_a_cloudflare_trace() {
+        let body = "fl=12f1\nh=www.cloudflare.com\nip=147.161.247.7\nts=1730790000.1\n";
+        assert_eq!(trace_exit_ip(body).as_deref(), Some("147.161.247.7"));
+        assert_eq!(trace_exit_ip("<html>hijacked</html>"), None);
+    }
+
+    #[test]
+    fn status_rows_never_show_proxy_credentials() {
+        let mut ranking = Ranking::default();
+        let p = proxy("http://user:secret@10.0.0.1:8080");
+        ranking.record_success(&p, 0.25, Some("1.1.1.1"));
+        ranking.record_success(&p, 0.35, Some("1.1.1.1"));
+        let rows = ranking.status(Instant::now(), 20);
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].proxy, "http://10.0.0.1:8080");
+        assert_eq!(rows[0].exit_ip.as_deref(), Some("1.1.1.1"));
+        assert_eq!(rows[0].median_ms, 350);
+        assert!(!serde_json::to_string(&rows).unwrap().contains("secret"));
+    }
+}

src/util/requester.rs

@@ -311,18 +311,45 @@ impl Requester {
         let cookie_jar = Self::shared_cookie_jar();
         let client = Self::build_client(cookie_jar.clone(), None);
 
-        let requester = Requester {
+        Requester {
             client,
             cookie_jar,
             debug_trace_id: None,
             proxy: false,
             flaresolverr_session: None,
             user_agent: None,
-        };
+        }
+    }
 
-        proxy::init_all_proxies_background(requester.clone());
+    /// GET through one pool proxy, as part of this requester's session (same
-
+    /// cookie jar and user agent). Unlike the direct paths this verifies
-        requester
+    /// certificates: a public proxy must not be able to alter what we read.
+    /// It deliberately builds its own client and skips the Jina/FlareSolverr
+    /// fallbacks: a pooled request must leave from the proxy's IP or fail.
+    pub async fn get_via(
+        &self,
+        via: &proxy::Proxy,
+        url: &str,
+        headers: Vec<(String, String)>,
+        timeout: Duration,
+    ) -> Result<Response, AnyErr> {
+        let mut builder = Client::builder()
+            .emulation(Emulation::Firefox151)
+            .cookie_provider(self.cookie_jar.clone())
+            .redirect(Policy::default());
+        if let Some(user_agent) = self.user_agent.as_deref() {
+            let mut default_headers = HeaderMap::new();
+            if let Ok(value) = HeaderValue::from_str(user_agent) {
+                default_headers.insert(USER_AGENT, value);
+                builder = builder.default_headers(default_headers);
+            }
+        }
+        let client = builder.build()?;
+        let mut request = client.get(url).timeout(timeout).proxy(via.to_wreq_proxy()?);
+        for (key, value) in headers {
+            request = request.header(key, value);
+        }
+        Ok(request.send().await?)
     }
 
     pub fn set_proxy(&mut self, proxy: bool) {

src/util/rule34video_resolve.rs

@@ -56,6 +56,25 @@ pub fn parse_formats(html: &str) -> Vec<(u32, String)> {
     formats
 }
 
+/// Categories then tags from the page's `flashvars`, deduplicated.
+pub fn parse_tags(html: &str) -> Vec<String> {
+    let Some(start) = html.find("var flashvars") else {
+        return vec![];
+    };
+    let block = &html[start..];
+    let block = &block[..block.len().min(6000)];
+    let mut tags: Vec<String> = Vec::new();
+    for key in ["video_categories", "video_tags"] {
+        for tag in flashvar(block, key).unwrap_or_default().split(',') {
+            let tag = tag.trim();
+            if !tag.is_empty() && !tags.iter().any(|t| t.eq_ignore_ascii_case(tag)) {
+                tags.push(tag.to_string());
+            }
+        }
+    }
+    tags
+}
+
 /// `{id}/{slug}[/]` -> the video page URL. Only digits, letters, `-` and `_`
 /// are accepted so the endpoint cannot steer the request elsewhere.
 pub fn page_url(endpoint: &str) -> Option<String> {
@@ -114,4 +133,89 @@ mod tests {
         assert_eq!(page_url("abc/slug"), None);
         assert_eq!(page_url("123/slug/extra"), None);
     }
+
+    fn env_usize(name: &str, default: usize) -> usize {
+        std::env::var(name)
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(default)
+    }
+
+    /// Bursts `/embed/{id}` requests to see whether they share the detail pages'
+    /// rate limit (~500 requests in ~100s from one IP -> 429). Walks down from the
+    /// newest id in the RSS feed; gaps (deleted videos) still count as requests.
+    ///
+    /// `R34_EMBED_REQUESTS` (default 600) and `R34_EMBED_CONCURRENCY` (default 8)
+    /// tune the burst. Fails on the first 429 and reports when it happened.
+    ///
+    /// cargo test embed_burst -- --ignored --nocapture
+    #[tokio::test]
+    #[ignore = "live rate-limit probe against rule34video.com"]
+    async fn embed_burst_does_not_trigger_429() {
+        use futures::stream::{self, StreamExt};
+        use std::collections::BTreeMap;
+        use std::time::Instant;
+
+        let total = env_usize("R34_EMBED_REQUESTS", 600);
+        let concurrency = env_usize("R34_EMBED_CONCURRENCY", 8);
+        let requester = Requester::new();
+
+        let rss = requester
+            .clone()
+            .get_raw(&format!("{BASE_URL}/rss/"))
+            .await
+            .expect("rss request")
+            .text()
+            .await
+            .expect("rss body");
+        let newest: u64 = rss
+            .split("/video/")
+            .nth(1)
+            .and_then(|rest| rest.split('/').next())
+            .and_then(|id| id.parse().ok())
+            .expect("newest video id from rss");
+
+        let started = Instant::now();
+        let results: Vec<(usize, u16, f64)> = stream::iter(0..total)
+            .map(|i| {
+                let mut requester = requester.clone();
+                let url = format!("{BASE_URL}/embed/{}", newest - i as u64);
+                async move {
+                    let status = match requester
+                        .get_raw_with_headers(&url, vec![("Referer".to_string(), format!("{BASE_URL}/"))])
+                        .await
+                    {
+                        Ok(response) => response.status().as_u16(),
+                        Err(_) => 0,
+                    };
+                    (i, status, started.elapsed().as_secs_f64())
+                }
+            })
+            .buffer_unordered(concurrency)
+            .collect()
+            .await;
+        let elapsed = started.elapsed().as_secs_f64();
+
+        let mut by_status: BTreeMap<u16, usize> = BTreeMap::new();
+        for (_, status, _) in &results {
+            *by_status.entry(*status).or_default() += 1;
+        }
+        let first_429 = results
+            .iter()
+            .filter(|(_, status, _)| *status == 429)
+            .min_by(|a, b| a.2.total_cmp(&b.2));
+        let completed_before_429 = first_429.map(|(_, _, at)| {
+            results.iter().filter(|(_, s, t)| *s != 429 && t < at).count()
+        });
+
+        println!(
+            "{total} embed requests from id {newest} down, concurrency {concurrency}: \
+             {elapsed:.1}s ({:.1} req/s); statuses {by_status:?}",
+            total as f64 / elapsed
+        );
+        if let (Some((i, _, at)), Some(before)) = (first_429, completed_before_429) {
+            println!("first 429 at {at:.1}s (request #{i}), after {before} successful requests");
+        }
+        assert!(first_429.is_none(), "embed burst was rate-limited: {by_status:?}");
+    }
 }