20260401 live recorder 扩展到1+N个delta，完成5个RIR录制，以及三方replay，结果三方均对齐vrps和vaps

src/fetch/http.rs

@@ -1,12 +1,16 @@
 use std::time::Duration;
 
 use reqwest::blocking::Client;
+use reqwest::header::HeaderMap;
 
 use crate::sync::rrdp::Fetcher;
 
 #[derive(Clone, Debug)]
 pub struct HttpFetcherConfig {
+    /// Short timeout used for connection establishment and small metadata objects.
     pub timeout: Duration,
+    /// Larger timeout used for RRDP snapshot / delta bodies.
+    pub large_body_timeout: Duration,
     pub user_agent: String,
 }
 
@@ -14,6 +18,7 @@ impl Default for HttpFetcherConfig {
     fn default() -> Self {
         Self {
             timeout: Duration::from_secs(20),
+            large_body_timeout: Duration::from_secs(180),
             user_agent: "rpki-dev/0.1 (stage2)".to_string(),
         }
     }
@@ -26,34 +31,154 @@ impl Default for HttpFetcherConfig {
 /// - fetching RRDP notification/snapshot files (RFC 8182 §3.4)
 #[derive(Clone, Debug)]
 pub struct BlockingHttpFetcher {
-    client: Client,
+    short_client: Client,
+    large_body_client: Client,
+    short_timeout: Duration,
+    large_body_timeout: Duration,
 }
 
 impl BlockingHttpFetcher {
     pub fn new(config: HttpFetcherConfig) -> Result<Self, String> {
-        let client = Client::builder()
+        let short_timeout = config.timeout;
+        let large_body_timeout = std::cmp::max(config.large_body_timeout, config.timeout);
+        let short_client = Client::builder()
+            .connect_timeout(config.timeout)
             .timeout(config.timeout)
+            .user_agent(config.user_agent.clone())
+            .build()
+            .map_err(|e| e.to_string())?;
+        let large_body_client = Client::builder()
+            .connect_timeout(config.timeout)
+            .timeout(large_body_timeout)
             .user_agent(config.user_agent)
             .build()
             .map_err(|e| e.to_string())?;
-        Ok(Self { client })
+        Ok(Self {
+            short_client,
+            large_body_client,
+            short_timeout,
+            large_body_timeout,
+        })
     }
 
     pub fn fetch_bytes(&self, uri: &str) -> Result<Vec<u8>, String> {
-        let resp = self
-            .client
+        let started = std::time::Instant::now();
+        let (client, timeout_profile, timeout_value) = self.client_for_uri(uri);
+        let resp = client
             .get(uri)
             .send()
-            .map_err(|e| format!("http request failed: {e}"))?;
+            .map_err(|e| {
+                let msg = format!("http request failed: {e}");
+                crate::progress_log::emit(
+                    "http_fetch_failed",
+                    serde_json::json!({
+                        "uri": uri,
+                        "stage": "request",
+                        "timeout_profile": timeout_profile,
+                        "request_timeout_ms": timeout_value.as_millis() as u64,
+                        "duration_ms": started.elapsed().as_millis() as u64,
+                        "error": msg,
+                    }),
+                );
+                msg
+            })?;
 
         let status = resp.status();
+        let headers = resp.headers().clone();
         if !status.is_success() {
-            return Err(format!("http status {status}"));
+            let body_preview = resp
+                .text()
+                .ok()
+                .map(|text| text.chars().take(160).collect::<String>());
+            let body_prefix = body_preview
+                .clone()
+                .unwrap_or_else(|| "<unavailable>".to_string());
+            let msg = format!(
+                "http status {status}; content_type={}; content_encoding={}; content_length={}; transfer_encoding={}; body_prefix={}",
+                header_value(&headers, "content-type"),
+                header_value(&headers, "content-encoding"),
+                header_value(&headers, "content-length"),
+                header_value(&headers, "transfer-encoding"),
+                body_prefix,
+            );
+            crate::progress_log::emit(
+                "http_fetch_failed",
+                serde_json::json!({
+                    "uri": uri,
+                    "stage": "status",
+                    "timeout_profile": timeout_profile,
+                    "request_timeout_ms": timeout_value.as_millis() as u64,
+                    "duration_ms": started.elapsed().as_millis() as u64,
+                    "status": status.as_u16(),
+                    "content_type": header_value_opt(&headers, "content-type"),
+                    "content_encoding": header_value_opt(&headers, "content-encoding"),
+                    "content_length": header_value_opt(&headers, "content-length"),
+                    "transfer_encoding": header_value_opt(&headers, "transfer-encoding"),
+                    "body_prefix": body_preview,
+                    "error": msg,
+                }),
+            );
+            return Err(msg);
         }
 
-        resp.bytes()
-            .map(|b| b.to_vec())
-            .map_err(|e| format!("http read body failed: {e}"))
+        match resp.bytes() {
+            Ok(bytes) => {
+                let duration_ms = started.elapsed().as_millis() as u64;
+                if (duration_ms as f64) / 1000.0 >= crate::progress_log::slow_threshold_secs() {
+                    crate::progress_log::emit(
+                        "http_fetch_slow",
+                        serde_json::json!({
+                            "uri": uri,
+                            "status": status.as_u16(),
+                            "timeout_profile": timeout_profile,
+                            "request_timeout_ms": timeout_value.as_millis() as u64,
+                            "duration_ms": duration_ms,
+                            "bytes": bytes.len(),
+                            "content_type": header_value_opt(&headers, "content-type"),
+                            "content_encoding": header_value_opt(&headers, "content-encoding"),
+                            "content_length": header_value_opt(&headers, "content-length"),
+                            "transfer_encoding": header_value_opt(&headers, "transfer-encoding"),
+                        }),
+                    );
+                }
+                Ok(bytes.to_vec())
+            }
+            Err(e) => {
+                let msg = format!(
+                    "http read body failed: {e}; status={}; content_type={}; content_encoding={}; content_length={}; transfer_encoding={}",
+                    status,
+                    header_value(&headers, "content-type"),
+                    header_value(&headers, "content-encoding"),
+                    header_value(&headers, "content-length"),
+                    header_value(&headers, "transfer-encoding"),
+                );
+                crate::progress_log::emit(
+                    "http_fetch_failed",
+                    serde_json::json!({
+                        "uri": uri,
+                        "stage": "read_body",
+                        "timeout_profile": timeout_profile,
+                        "request_timeout_ms": timeout_value.as_millis() as u64,
+                        "duration_ms": started.elapsed().as_millis() as u64,
+                        "status": status.as_u16(),
+                        "content_type": header_value_opt(&headers, "content-type"),
+                        "content_encoding": header_value_opt(&headers, "content-encoding"),
+                        "content_length": header_value_opt(&headers, "content-length"),
+                        "transfer_encoding": header_value_opt(&headers, "transfer-encoding"),
+                        "error": msg,
+                    }),
+                );
+                Err(msg)
+            }
+        }
+    }
+
+    fn client_for_uri(&self, uri: &str) -> (&Client, &'static str, Duration) {
+        if uses_large_body_timeout(uri) {
+            (&self.large_body_client, "large_body", self.large_body_timeout)
+        } else {
+            (&self.short_client, "short", self.short_timeout)
+        }
     }
 }
 
@@ -63,12 +188,30 @@ impl Fetcher for BlockingHttpFetcher {
     }
 }
 
+fn header_value(headers: &HeaderMap, name: &str) -> String {
+    header_value_opt(headers, name).unwrap_or_else(|| "<none>".to_string())
+}
+
+fn header_value_opt(headers: &HeaderMap, name: &str) -> Option<String> {
+    headers
+        .get(name)
+        .and_then(|v| v.to_str().ok())
+        .map(|v| v.to_string())
+}
+
+fn uses_large_body_timeout(uri: &str) -> bool {
+    uri.starts_with("https://")
+        && uri.ends_with(".xml")
+        && !uri.ends_with("notification.xml")
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use std::io::{Read, Write};
     use std::net::TcpListener;
     use std::thread;
+    use std::time::Duration as StdDuration;
 
     fn spawn_one_shot_http_server(status_line: &'static str, body: &'static [u8]) -> String {
         let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind");
@@ -110,4 +253,40 @@ mod tests {
         let err = http.fetch_bytes(&url).unwrap_err();
         assert!(err.contains("http status"), "{err}");
     }
+
+    #[test]
+    fn fetch_bytes_times_out_on_idle_body_read() {
+        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind");
+        let addr = listener.local_addr().expect("addr");
+        thread::spawn(move || {
+            let (mut stream, _) = listener.accept().expect("accept");
+            let mut buf = [0u8; 1024];
+            let _ = stream.read(&mut buf);
+            stream
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nh")
+                .expect("write partial body");
+            std::thread::sleep(StdDuration::from_secs(2));
+            let _ = stream.write_all(b"ello");
+        });
+        let url = format!("http://{}/", addr);
+        let http = BlockingHttpFetcher::new(HttpFetcherConfig {
+            timeout: Duration::from_secs(1),
+            ..HttpFetcherConfig::default()
+        })
+        .expect("http");
+        let err = http.fetch_bytes(&url).unwrap_err();
+        assert!(err.contains("http read body failed"), "{err}");
+    }
+
+    #[test]
+    fn uses_large_body_timeout_selects_rrdp_snapshot_and_delta_not_notification() {
+        assert!(!uses_large_body_timeout("https://rrdp.example.test/notification.xml"));
+        assert!(uses_large_body_timeout(
+            "https://rrdp.example.test/session/123/snapshot.xml"
+        ));
+        assert!(uses_large_body_timeout(
+            "https://rrdp.example.test/session/123/delta-42.xml"
+        ));
+        assert!(!uses_large_body_timeout("https://tal.example.test/example.tal"));
+    }
 }

src/fetch/rsync.rs

@@ -8,6 +8,14 @@ pub enum RsyncFetchError {
 
 pub type RsyncFetchResult<T> = Result<T, RsyncFetchError>;
 
+pub fn normalize_rsync_base_uri(s: &str) -> String {
+    if s.ends_with('/') {
+        s.to_string()
+    } else {
+        format!("{s}/")
+    }
+}
+
 /// Fetch repository objects from a publication point.
 ///
 /// v1: this is intentionally abstract so unit tests can use a mock, and later we can
@@ -15,6 +23,33 @@ pub type RsyncFetchResult<T> = Result<T, RsyncFetchError>;
 pub trait RsyncFetcher {
     /// Return a list of objects as `(rsync_uri, bytes)` pairs.
     fn fetch_objects(&self, rsync_base_uri: &str) -> RsyncFetchResult<Vec<(String, Vec<u8>)>>;
+
+    /// Stream fetched objects to a visitor without requiring callers to materialize the
+    /// full result vector in memory.
+    fn visit_objects(
+        &self,
+        rsync_base_uri: &str,
+        visitor: &mut dyn FnMut(String, Vec<u8>) -> Result<(), String>,
+    ) -> RsyncFetchResult<(usize, u64)> {
+        let objects = self.fetch_objects(rsync_base_uri)?;
+        let mut count = 0usize;
+        let mut bytes_total = 0u64;
+        for (uri, bytes) in objects {
+            bytes_total += bytes.len() as u64;
+            count += 1;
+            visitor(uri, bytes).map_err(RsyncFetchError::Fetch)?;
+        }
+        Ok((count, bytes_total))
+    }
+
+    /// Return the deduplication key used by orchestration layers.
+    ///
+    /// By default this is the normalized publication point base URI. Fetchers that
+    /// intentionally widen their fetch scope (for example to a full rsync module)
+    /// should override this so callers can safely deduplicate at the same scope.
+    fn dedup_key(&self, rsync_base_uri: &str) -> String {
+        normalize_rsync_base_uri(rsync_base_uri)
+    }
 }
 
 /// A simple "rsync" implementation backed by a local directory.
@@ -73,14 +108,6 @@ fn walk_dir_collect(
     Ok(())
 }
 
-fn normalize_rsync_base_uri(s: &str) -> String {
-    if s.ends_with('/') {
-        s.to_string()
-    } else {
-        format!("{s}/")
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -115,4 +142,14 @@ mod tests {
             RsyncFetchError::Fetch(msg) => assert!(!msg.is_empty()),
         }
     }
+
+    #[test]
+    fn default_dedup_key_is_normalized_base_uri() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let fetcher = LocalDirRsyncFetcher::new(tmp.path());
+        assert_eq!(
+            fetcher.dedup_key("rsync://example.net/repo"),
+            "rsync://example.net/repo/"
+        );
+    }
 }

src/fetch/rsync_system.rs

@@ -5,7 +5,9 @@ use std::time::Duration;
 use sha2::Digest;
 use uuid::Uuid;
 
-use crate::fetch::rsync::{RsyncFetchError, RsyncFetchResult, RsyncFetcher};
+use crate::fetch::rsync::{
+    RsyncFetchError, RsyncFetchResult, RsyncFetcher, normalize_rsync_base_uri,
+};
 
 #[derive(Clone, Debug)]
 pub struct SystemRsyncConfig {
@@ -94,30 +96,55 @@ impl SystemRsyncFetcher {
         }
         Ok(())
     }
+
+    fn module_fetch_uri(&self, rsync_base_uri: &str) -> String {
+        rsync_capture_scope_uri(rsync_base_uri).unwrap_or_else(|| normalize_rsync_base_uri(rsync_base_uri))
+    }
 }
 
 impl RsyncFetcher for SystemRsyncFetcher {
     fn fetch_objects(&self, rsync_base_uri: &str) -> RsyncFetchResult<Vec<(String, Vec<u8>)>> {
-        let base = normalize_rsync_base_uri(rsync_base_uri);
+        let mut out = Vec::new();
+        self.visit_objects(rsync_base_uri, &mut |uri, bytes| {
+            out.push((uri, bytes));
+            Ok(())
+        })?;
+        Ok(out)
+    }
+
+    fn visit_objects(
+        &self,
+        rsync_base_uri: &str,
+        visitor: &mut dyn FnMut(String, Vec<u8>) -> Result<(), String>,
+    ) -> RsyncFetchResult<(usize, u64)> {
+        let base = self.module_fetch_uri(rsync_base_uri);
+        let mut count = 0usize;
+        let mut bytes_total = 0u64;
+        let mut wrapped = |uri: String, bytes: Vec<u8>| -> Result<(), String> {
+            bytes_total += bytes.len() as u64;
+            count += 1;
+            visitor(uri, bytes)
+        };
+
         if let Some(dst) = self
             .mirror_dst_dir(&base)
             .map_err(|e| RsyncFetchError::Fetch(e.to_string()))?
         {
             self.run_rsync(&base, &dst)
                 .map_err(RsyncFetchError::Fetch)?;
-            let mut out = Vec::new();
-            walk_dir_collect(&dst, &dst, &base, &mut out).map_err(RsyncFetchError::Fetch)?;
-            return Ok(out);
+            walk_dir_visit(&dst, &dst, &base, &mut wrapped).map_err(RsyncFetchError::Fetch)?;
+            return Ok((count, bytes_total));
         }
 
         let tmp = TempDir::new().map_err(|e| RsyncFetchError::Fetch(e.to_string()))?;
         self.run_rsync(&base, tmp.path())
             .map_err(RsyncFetchError::Fetch)?;
+        walk_dir_visit(tmp.path(), tmp.path(), &base, &mut wrapped).map_err(RsyncFetchError::Fetch)?;
+        Ok((count, bytes_total))
+    }
 
-        let mut out = Vec::new();
-        walk_dir_collect(tmp.path(), tmp.path(), &base, &mut out)
-            .map_err(RsyncFetchError::Fetch)?;
-        Ok(out)
+    fn dedup_key(&self, rsync_base_uri: &str) -> String {
+        self.module_fetch_uri(rsync_base_uri)
     }
 }
 
@@ -144,12 +171,20 @@ impl Drop for TempDir {
     }
 }
 
-fn normalize_rsync_base_uri(s: &str) -> String {
-    if s.ends_with('/') {
-        s.to_string()
-    } else {
-        format!("{s}/")
+fn rsync_capture_scope_uri(s: &str) -> Option<String> {
+    let normalized = normalize_rsync_base_uri(s);
+    let rest = normalized.strip_prefix("rsync://")?;
+    let mut host_and_path = rest.splitn(2, '/');
+    let authority = host_and_path.next()?;
+    let path = host_and_path.next()?;
+    let mut segments: Vec<&str> = path.split('/').filter(|segment| !segment.is_empty()).collect();
+    if segments.is_empty() {
+        return None;
     }
+    if segments.len() >= 4 {
+        segments.pop();
+    }
+    Some(format!("rsync://{authority}/{}/", segments.join("/")))
 }
 
 fn walk_dir_collect(
@@ -182,6 +217,36 @@ fn walk_dir_collect(
     Ok(())
 }
 
+fn walk_dir_visit(
+    root: &Path,
+    current: &Path,
+    rsync_base_uri: &str,
+    visitor: &mut dyn FnMut(String, Vec<u8>) -> Result<(), String>,
+) -> Result<(), String> {
+    let rd = std::fs::read_dir(current).map_err(|e| e.to_string())?;
+    for entry in rd {
+        let entry = entry.map_err(|e| e.to_string())?;
+        let path = entry.path();
+        let meta = entry.metadata().map_err(|e| e.to_string())?;
+        if meta.is_dir() {
+            walk_dir_visit(root, &path, rsync_base_uri, visitor)?;
+            continue;
+        }
+        if !meta.is_file() {
+            continue;
+        }
+        let rel = path
+            .strip_prefix(root)
+            .map_err(|e| e.to_string())?
+            .to_string_lossy()
+            .replace('\\', "/");
+        let uri = format!("{rsync_base_uri}{rel}");
+        let bytes = std::fs::read(&path).map_err(|e| e.to_string())?;
+        visitor(uri, bytes)?;
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -217,6 +282,32 @@ mod tests {
         assert_eq!(out[1].1, b"x");
     }
 
+    #[test]
+    fn rsync_capture_scope_uri_widens_only_deep_publication_points() {
+        assert_eq!(
+            rsync_capture_scope_uri("rsync://example.net/repo/ta/ca/publication-point/"),
+            Some("rsync://example.net/repo/ta/ca/".to_string())
+        );
+        assert_eq!(
+            rsync_capture_scope_uri("rsync://example.net/repo/ta/"),
+            Some("rsync://example.net/repo/ta/".to_string())
+        );
+        assert_eq!(
+            rsync_capture_scope_uri("rsync://example.net/repo/"),
+            Some("rsync://example.net/repo/".to_string())
+        );
+        assert_eq!(rsync_capture_scope_uri("https://example.net/repo"), None);
+    }
+
+    #[test]
+    fn system_rsync_dedup_key_uses_capture_scope() {
+        let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig::default());
+        assert_eq!(
+            fetcher.dedup_key("rsync://example.net/repo/ta/ca/publication-point/"),
+            "rsync://example.net/repo/ta/ca/"
+        );
+    }
+
     #[test]
     fn system_rsync_fetcher_reports_spawn_and_exit_errors() {
         let dst = tempfile::tempdir().expect("tempdir");