graft batch 63: 7 lands from fluent-1

.github/workflows/runner-policy.yml

@@ -21,4 +21,4 @@ concurrency:
 
 jobs:
   runner-policy:
-    uses: TheDancingDeveloper-org/github-policy/.github/workflows/runner-policy-reusable.yml@48b48161921f236d0086231f8f28ccc8fa8c5cef # main
+    uses: TheDancingDeveloper-org/github-policy/.github/workflows/runner-policy-reusable.yml@af19e2d05a5ceeb2f7aa838b2c764a62ee335219 # main

crates/librtbit-tracker-comms/src/tracker_comms.rs

@@ -474,6 +474,7 @@ impl TrackerComms {
                 u32::try_from(response.complete).ok(),
                 u32::try_from(response.incomplete).ok(),
                 peers_returned,
+                Some(response.complete),
                 interval,
             );
         }
@@ -609,6 +610,12 @@ impl TrackerComms {
             Ok(response) => {
                 trace!(len = response.addrs.len(), "received announce response");
                 let peers_returned = response.addrs.len() as u32;
+                let num_downloaded = client
+                    .scrape(addr, vec![self.info_hash])
+                    .await
+                    .ok()
+                    .and_then(|stats| stats.into_iter().next())
+                    .map(|stats| u64::from(stats.completed));
                 for addr in response.addrs {
                     self.tx.send(addr).await.context("rx closed")?;
                 }
@@ -619,6 +626,7 @@ impl TrackerComms {
                         Some(response.seeders),
                         Some(response.leechers),
                         peers_returned,
+                        num_downloaded,
                         sleep,
                     );
                 }

crates/librtbit-tracker-comms/src/tracker_status.rs

@@ -34,12 +34,16 @@ pub enum TrackerAnnounceState {
 pub struct TrackerStatus {
     pub url: String,
     pub state: TrackerAnnounceState,
+    /// Zero-based announce-list tier containing this tracker.
+    pub tier: u32,
     /// Seeder count ("complete") reported by the last successful announce.
     pub seeders: Option<u32>,
     /// Leecher count ("incomplete") reported by the last successful announce.
     pub leechers: Option<u32>,
     /// Number of peer addresses returned by the last successful announce.
     pub peers_returned: Option<u32>,
+    /// Number of peers that have completed the torrent, as reported by the tracker.
+    pub num_downloaded: Option<u64>,
     /// Unix timestamp of the last successful announce.
     pub last_announce_unix: Option<u64>,
     /// Announce interval (seconds) the tracker asked for.
@@ -48,13 +52,15 @@ pub struct TrackerStatus {
 }
 
 impl TrackerStatus {
-    fn new(url: String) -> Self {
+    fn new(url: String, tier: u32) -> Self {
         Self {
             url,
             state: TrackerAnnounceState::NotContacted,
+            tier,
             seeders: None,
             leechers: None,
             peers_returned: None,
+            num_downloaded: None,
             last_announce_unix: None,
             interval_secs: None,
             last_error: None,
@@ -92,9 +98,14 @@ impl Default for TrackerStatusRegistry {
 }
 
 impl TrackerStatusRegistry {
-    fn insert_or_get<'a>(&self, g: &'a mut HashMap<String, Entry>, url: &str) -> &'a mut Entry {
+    fn insert_or_get<'a>(
+        &self,
+        g: &'a mut HashMap<String, Entry>,
+        url: &str,
+        tier: u32,
+    ) -> &'a mut Entry {
         g.entry(url.to_owned()).or_insert_with(|| Entry {
-            status: TrackerStatus::new(url.to_owned()),
+            status: TrackerStatus::new(url.to_owned(), tier),
             generation: self.next_generation.fetch_add(1, Ordering::Relaxed),
         })
     }
@@ -110,8 +121,13 @@ impl TrackerStatusRegistry {
     /// Register a tracker without changing its state. Returns the entry's
     /// generation, which an announce loop passes to [`Self::is_current`].
     pub fn ensure(&self, url: &str) -> u64 {
+        self.ensure_with_tier(url, 0)
+    }
+
+    /// Register a tracker with its announce-list tier.
+    pub fn ensure_with_tier(&self, url: &str, tier: u32) -> u64 {
         let mut g = self.inner.write();
-        self.insert_or_get(&mut g, url).generation
+        self.insert_or_get(&mut g, url, tier).generation
     }
 
     /// Whether a loop started at `generation` for `url` should keep announcing.
@@ -161,13 +177,15 @@ impl TrackerStatusRegistry {
         seeders: Option<u32>,
         leechers: Option<u32>,
         peers_returned: u32,
+        num_downloaded: Option<u64>,
         interval_secs: u64,
     ) {
         self.with_existing(url, |e| {
             e.state = TrackerAnnounceState::Working;
             e.seeders = seeders.or(e.seeders);
             e.leechers = leechers.or(e.leechers);
             e.peers_returned = Some(peers_returned);
+            e.num_downloaded = num_downloaded.or(e.num_downloaded);
             e.last_announce_unix = Some(now_unix());
             e.interval_secs = Some(interval_secs);
             e.last_error = None;
@@ -183,7 +201,7 @@ impl TrackerStatusRegistry {
 
     pub fn record_disabled(&self, url: &str, reason: &str) {
         let mut g = self.inner.write();
-        let e = &mut self.insert_or_get(&mut g, url).status;
+        let e = &mut self.insert_or_get(&mut g, url, 0).status;
         e.state = TrackerAnnounceState::Disabled;
         e.last_error = Some(reason.to_owned());
     }
@@ -219,7 +237,7 @@ mod tests {
         assert!(!r.remove(URL));
 
         // A stopping loop's late report must not resurrect the entry.
-        r.record_success(URL, Some(1), Some(1), 1, 60);
+        r.record_success(URL, Some(1), Some(1), 1, Some(1), 60);
         r.record_error(URL, "boom");
         assert!(r.snapshot().is_empty());
 
@@ -230,6 +248,17 @@ mod tests {
         assert!(r.is_current(URL, readded));
     }
 
+    #[test]
+    fn snapshot_retains_tier_and_completed_count() {
+        let r = TrackerStatusRegistry::default();
+        r.ensure_with_tier(URL, 2);
+        r.record_success(URL, Some(4), Some(3), 7, Some(19), 60);
+
+        let status = r.snapshot().pop().unwrap();
+        assert_eq!(status.tier, 2);
+        assert_eq!(status.num_downloaded, Some(19));
+    }
+
     #[test]
     fn reannounce_and_remove_wake_subscribers() {
         let r = TrackerStatusRegistry::default();

crates/librtbit/examples/simulate_traffic.rs

@@ -179,8 +179,8 @@ impl TestHarness {
                     download_bps: Default::default(),
                     peer_limit: None,
                     concurrent_init_limit: None,
-                    recheck_read_mibps: None,
                 },
+                recheck_read_mibps: None,
                 disable_local_service_discovery: false,
                 connect: Some(ConnectionOptions {
                     proxy_url: None,

crates/librtbit/src/http_api/handlers/qbit/tests.rs

@@ -1570,6 +1570,28 @@ fn tracker_status_maps_to_qbit_codes() {
     assert_eq!(qbit_tracker_status(Error), 4);
 }
 
+#[test]
+fn tracker_info_maps_tier_and_completed_count() {
+    use super::qbit_tracker_info;
+    use tracker_comms::{TrackerAnnounceState, TrackerStatus};
+
+    let info = qbit_tracker_info(TrackerStatus {
+        url: "https://tracker.example/announce".to_owned(),
+        state: TrackerAnnounceState::Working,
+        tier: 3,
+        seeders: Some(4),
+        leechers: Some(2),
+        peers_returned: Some(6),
+        num_downloaded: Some(17),
+        last_announce_unix: Some(1),
+        interval_secs: Some(60),
+        last_error: None,
+    });
+
+    assert_eq!(info.tier, 3);
+    assert_eq!(info.num_downloaded, 17);
+}
+
 #[test]
 fn rate_limit_conversions_round_trip_and_treat_zero_as_unlimited() {
     use std::num::NonZeroU32;
@@ -1675,6 +1697,12 @@ fn incomplete_read_only_payload_reports_missing_files() {
         live: None,
         queued_for_init: None,
     };
+    let mut checking = stats(None);
+    checking.state = TorrentStatsState::Initializing;
+    checking.queued_for_init = Some(false);
+    assert_eq!(qbit_state_for(&checking, false, false), "checkingDL");
+    checking.queued_for_init = Some(true);
+    assert_eq!(qbit_state_for(&checking, false, false), "queuedForChecking");
     assert_eq!(
         qbit_state_for(&stats(Some(READ_ONLY_INCOMPLETE_PAYLOAD)), false, false),
         "missingFiles"

crates/librtbit/src/http_api/handlers/qbit/torrents.rs

@@ -28,7 +28,7 @@ pub(super) fn map_state(
     forced: bool,
 ) -> &'static str {
     match (state, finished) {
-        (TorrentStatsState::Initializing, _) => "metaDL",
+        (TorrentStatsState::Initializing, _) => "checkingDL",
         (TorrentStatsState::Live, false) if forced => "forcedDL",
         (TorrentStatsState::Live, true) if forced => "forcedUP",
         (TorrentStatsState::Live, false) => "downloading",
@@ -56,6 +56,10 @@ pub(super) fn qbit_state_for(
     {
         return "missingFiles";
     }
+    if matches!(stats.state, TorrentStatsState::Initializing) && stats.queued_for_init == Some(true)
+    {
+        return "queuedForChecking";
+    }
     map_state(stats.state, stats.finished, queued, forced)
 }
 
@@ -81,7 +85,7 @@ pub(super) fn matches_filter(
         "all" => true,
         "downloading" => matches!(
             qbit_state,
-            "downloading" | "metaDL" | "forcedDL" | "queuedDL"
+            "downloading" | "checkingDL" | "queuedForChecking" | "forcedDL" | "queuedDL"
         ),
         "seeding" => matches!(qbit_state, "uploading" | "forcedUP" | "queuedUP"),
         "completed" => stats.finished,
@@ -710,6 +714,21 @@ pub(super) fn qbit_tracker_status(state: tracker_comms::TrackerAnnounceState) ->
     }
 }
 
+pub(super) fn qbit_tracker_info(t: tracker_comms::TrackerStatus) -> QbitTrackerInfo {
+    QbitTrackerInfo {
+        status: qbit_tracker_status(t.state),
+        tier: i32::try_from(t.tier).unwrap_or(i32::MAX),
+        num_peers: t.peers_returned.map_or(-1, i64::from),
+        num_seeds: t.seeders.map_or(-1, i64::from),
+        num_leeches: t.leechers.map_or(-1, i64::from),
+        num_downloaded: t
+            .num_downloaded
+            .map_or(-1, |n| i64::try_from(n).unwrap_or(i64::MAX)),
+        msg: t.last_error.unwrap_or_default(),
+        url: t.url,
+    }
+}
+
 pub(super) async fn h_torrents_trackers(
     State(state): State<Arc<QbitState>>,
     Query(query): Query<HashQuery>,
@@ -727,16 +746,7 @@ pub(super) async fn h_torrents_trackers(
     let trackers: Vec<QbitTrackerInfo> = response
         .trackers
         .into_iter()
-        .map(|t| QbitTrackerInfo {
-            status: qbit_tracker_status(t.state),
-            tier: 0,
-            num_peers: t.peers_returned.map_or(-1, i64::from),
-            num_seeds: t.seeders.map_or(-1, i64::from),
-            num_leeches: t.leechers.map_or(-1, i64::from),
-            num_downloaded: -1,
-            msg: t.last_error.unwrap_or_default(),
-            url: t.url,
-        })
+        .map(qbit_tracker_info)
         .collect();
     axum::Json(trackers).into_response()
 }

crates/librtbit/src/http_api/handlers/qbit_parity_spec.json

@@ -1287,7 +1287,7 @@
       "endpoint": "torrents/trackers",
       "method": "GET",
       "status": "partial",
-      "notes": "Real per-tracker announce state and peer counts. Tracker tier and num_downloaded remain unavailable because configured trackers are currently stored without tier or scrape-download metadata.",
+      "notes": "Real per-tracker announce state, announce-list tier, peer counts, and completed/downloaded count reported by HTTP announce responses or UDP scrape responses.",
       "response": {
         "content": "json",
         "schema": [

crates/librtbit/src/ip_ranges.rs

@@ -69,6 +69,9 @@ impl IpRanges {
 
         let limits = crate::safe_fetch::FetchLimits {
             max_bytes: 256 * 1024 * 1024,
+            // Large compressed blocklists can legitimately take longer than
+            // the normal per-hop timeout to download and parse.
+            timeout: std::time::Duration::from_secs(5 * 60),
             ..Default::default()
         };
         let body = crate::safe_fetch::fetch_bytes(parsed_url.as_str(), &limits)

crates/librtbit/src/rss/fetch.rs

@@ -73,9 +73,15 @@ pub fn assert_public_addresses(addresses: &[IpAddr]) -> Result<(), FeedUrlError>
     Ok(())
 }
 
-/// Keep at most [`MAX_FEED_BYTES`] of a feed body.
-pub fn cap_feed_body(body: &[u8]) -> Vec<u8> {
-    body.iter().take(MAX_FEED_BYTES).copied().collect()
+/// Reject a feed body that exceeds [`MAX_FEED_BYTES`].
+///
+/// Truncating XML can turn a valid feed into malformed input and hide the
+/// fact that the remote response exceeded our safety limit.
+pub fn cap_feed_body(body: &[u8]) -> anyhow::Result<Vec<u8>> {
+    if body.len() > MAX_FEED_BYTES {
+        anyhow::bail!("RSS feed body exceeds the {MAX_FEED_BYTES} byte cap")
+    }
+    Ok(body.to_vec())
 }
 
 #[cfg(test)]
@@ -134,6 +140,8 @@ mod tests {
             "http://240.0.0.1/feed.xml",
             "http://255.255.255.255/feed.xml",
             "http://[64:ff9b::1.2.3.4]/feed.xml",
+            "http://[64:ff9b:1::1]/feed.xml",
+            "http://[2001::1]/feed.xml",
             "http://[::192.0.2.1]/feed.xml",
             "http://[2002:c000:201::]/feed.xml",
         ] {
@@ -151,7 +159,7 @@ mod tests {
     #[test]
     fn caps_a_feed_body_at_the_limit() {
         let big = vec![b'a'; MAX_FEED_BYTES + 100];
-        assert_eq!(cap_feed_body(&big).len(), MAX_FEED_BYTES);
-        assert_eq!(cap_feed_body(b"<rss/>"), b"<rss/>");
+        assert!(cap_feed_body(&big).is_err());
+        assert_eq!(cap_feed_body(b"<rss/>").unwrap(), b"<rss/>");
     }
 }

crates/librtbit/src/rss/monitor.rs

@@ -94,7 +94,7 @@ impl RssMonitor {
             let bytes = tokio::fs::read(&path).await.map_err(|e| {
                 anyhow::anyhow!("error reading RSS feed file {}: {e}", path.display())
             })?;
-            return Ok(fetch::cap_feed_body(&bytes));
+            return fetch::cap_feed_body(&bytes);
         }
 
         let _ = client;

crates/librtbit/src/safe_fetch.rs

@@ -86,8 +86,12 @@ pub fn is_forbidden_address(ip: IpAddr) -> bool {
                 // Unique local fc00::/7 and link local fe80::/10.
                 || (seg0 & 0xfe00) == 0xfc00
                 || (seg0 & 0xffc0) == 0xfe80
-                // NAT64 well-known prefix 64:ff9b::/96.
+                // NAT64 well-known prefixes (including the RFC 6052 /96
+                // form and the RFC 6052 /48 form used by local translators).
                 || is_nat64(v6)
+                || (seg0 == 0x64 && v6.segments()[1] == 0xff9b && v6.segments()[2] == 1)
+                // Teredo tunnels can expose a client behind a private NAT.
+                || seg0 == 0x2001 && v6.segments()[1] == 0
                 // IPv4-compatible ::a.b.c.d and 6to4 2002::/16.
                 || is_ipv4_compatible(v6)
                 || seg0 == 0x2002
@@ -148,6 +152,9 @@ pub async fn get(url: &str, limits: &FetchLimits) -> anyhow::Result<Response> {
         let (host, port, addrs) = vet_host(&current).await?;
         let mut builder = Client::builder()
             .timeout(limits.timeout)
+            // Resolve-pinning only protects direct connections. Never let an
+            // environment proxy route around the vetted destination.
+            .no_proxy()
             .redirect(reqwest::redirect::Policy::none());
         for ip in addrs {
             builder = builder.resolve(&host, SocketAddr::new(ip, port));