NASP/rpki
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260410 完成五个rir 基于cir的三方回放，raw by hash 独立db，发现内存占用大，连续大rir 录制发生oom

Browse Source
This commit is contained in:
yuyr
2026-04-11 11:24:32 +08:00
parent e083fe4daa
commit 77fc2f1a41
96 changed files with 5159 additions and 1720 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
src/fetch/http.rs
View File

@@ -7,6 +7,8 @@ use crate::sync::rrdp::Fetcher;
#[derive(Clone, Debug)]
pub struct HttpFetcherConfig {
/// Connection-establishment timeout for HTTP requests.
pub connect_timeout: Duration,
/// Short timeout used for connection establishment and small metadata objects.
pub timeout: Duration,
/// Larger timeout used for RRDP snapshot / delta bodies.
@@ -17,7 +19,8 @@ pub struct HttpFetcherConfig {
impl Default for HttpFetcherConfig {
fn default() -> Self {
Self {
timeout: Duration::from_secs(20),
connect_timeout: Duration::from_secs(15),
timeout: Duration::from_secs(30),
large_body_timeout: Duration::from_secs(180),
user_agent: "rpki-dev/0.1 (stage2)".to_string(),
}
@@ -41,14 +44,15 @@ impl BlockingHttpFetcher {
pub fn new(config: HttpFetcherConfig) -> Result<Self, String> {
let short_timeout = config.timeout;
let large_body_timeout = std::cmp::max(config.large_body_timeout, config.timeout);
let connect_timeout = std::cmp::min(config.connect_timeout, config.timeout);
let short_client = Client::builder()
.connect_timeout(config.timeout)
.connect_timeout(connect_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)
.connect_timeout(connect_timeout)
.timeout(large_body_timeout)
.user_agent(config.user_agent)
.build()
@@ -64,24 +68,21 @@ impl BlockingHttpFetcher {
pub fn fetch_bytes(&self, uri: &str) -> Result<Vec<u8>, String> {
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| {
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 resp = client.get(uri).send().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();
@@ -175,7 +176,11 @@ impl BlockingHttpFetcher {
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)
(
&self.large_body_client,
"large_body",
self.large_body_timeout,
)
} else {
(&self.short_client, "short", self.short_timeout)
}
@@ -200,9 +205,7 @@ fn header_value_opt(headers: &HeaderMap, name: &str) -> Option<String> {
}
fn uses_large_body_timeout(uri: &str) -> bool {
uri.starts_with("https://")
&& uri.ends_with(".xml")
&& !uri.ends_with("notification.xml")
uri.starts_with("https://") && uri.ends_with(".xml") && !uri.ends_with("notification.xml")
}
#[cfg(test)]
@@ -280,13 +283,17 @@ mod tests {
#[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/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"));
assert!(!uses_large_body_timeout(
"https://tal.example.test/example.tal"
));
}
}

417
src/fetch/rsync_system.rs
View File

@@ -1,6 +1,10 @@
use std::cell::RefCell;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::process::Stdio;
use std::thread;
use std::time::Duration;
use std::time::Instant;
use sha2::Digest;
use uuid::Uuid;
@@ -12,6 +16,7 @@ use crate::fetch::rsync::{
#[derive(Clone, Debug)]
pub struct SystemRsyncConfig {
pub rsync_bin: PathBuf,
pub connect_timeout: Duration,
pub timeout: Duration,
pub extra_args: Vec<String>,
/// Optional root directory for persistent rsync mirrors.
@@ -28,7 +33,8 @@ impl Default for SystemRsyncConfig {
fn default() -> Self {
Self {
rsync_bin: PathBuf::from("rsync"),
timeout: Duration::from_secs(60),
connect_timeout: Duration::from_secs(15),
timeout: Duration::from_secs(30),
extra_args: Vec::new(),
mirror_root: None,
}
@@ -44,6 +50,39 @@ pub struct SystemRsyncFetcher {
config: SystemRsyncConfig,
}
thread_local! {
static RSYNC_TIMEOUT_OVERRIDE: RefCell<Option<Duration>> = const { RefCell::new(None) };
static RSYNC_FAIL_FAST_PROFILE: RefCell<Option<RsyncFailFastProfile>> = const { RefCell::new(None) };
}
pub fn with_scoped_rsync_timeout_override<R>(timeout: Duration, f: impl FnOnce() -> R) -> R {
RSYNC_TIMEOUT_OVERRIDE.with(|cell| {
let previous = cell.replace(Some(timeout));
let result = f();
let _ = cell.replace(previous);
result
})
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RsyncFailFastProfile {
pub initial_wall_clock_timeout: Duration,
pub max_wall_clock_timeout: Duration,
pub max_attempts: usize,
}
pub fn with_scoped_rsync_fail_fast_profile<R>(
profile: RsyncFailFastProfile,
f: impl FnOnce() -> R,
) -> R {
RSYNC_FAIL_FAST_PROFILE.with(|cell| {
let previous = cell.replace(Some(profile));
let result = f();
let _ = cell.replace(previous);
result
})
}
impl SystemRsyncFetcher {
pub fn new(config: SystemRsyncConfig) -> Self {
Self { config }
@@ -69,32 +108,164 @@ impl SystemRsyncFetcher {
}
fn run_rsync(&self, src: &str, dst: &Path) -> Result<(), String> {
let fail_fast = RSYNC_FAIL_FAST_PROFILE.with(|cell| *cell.borrow());
if let Some(profile) = fail_fast {
return self.run_rsync_fail_fast(src, dst, profile);
}
self.run_rsync_once(src, dst, None, false)
}
fn run_rsync_once(
&self,
src: &str,
dst: &Path,
wall_clock_timeout: Option<Duration>,
keep_partial: bool,
) -> Result<(), String> {
// `--timeout` is I/O timeout in seconds (applies to network reads/writes).
let timeout_secs = self.config.timeout.as_secs().max(1).to_string();
let timeout =
RSYNC_TIMEOUT_OVERRIDE.with(|cell| cell.borrow().unwrap_or(self.config.timeout));
let connect_timeout_secs = self.config.connect_timeout.as_secs().max(1).to_string();
let timeout_secs = timeout.as_secs().max(1).to_string();
let is_remote_rsync = src.starts_with("rsync://");
let mut cmd = Command::new(&self.config.rsync_bin);
cmd.arg("-rt")
.arg("--delete")
.arg("--timeout")
.arg(timeout_secs)
.args(&self.config.extra_args)
.arg(src)
.arg(dst);
let out = cmd
.output()
.map_err(|e| format!("rsync spawn failed: {e}"))?;
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr);
let stdout = String::from_utf8_lossy(&out.stdout);
return Err(format!(
"rsync failed: status={} stdout={} stderr={}",
out.status,
stdout.trim(),
stderr.trim()
));
.args(&self.config.extra_args);
if is_remote_rsync {
cmd.arg("--contimeout").arg(connect_timeout_secs);
}
if keep_partial {
cmd.arg("--partial");
}
cmd.arg(src)
.arg(dst)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let mut child = cmd
.spawn()
.map_err(|e| format!("rsync spawn failed: {e}"))?;
if let Some(limit) = wall_clock_timeout {
let started = Instant::now();
loop {
match child
.try_wait()
.map_err(|e| format!("rsync wait failed: {e}"))?
{
Some(_status) => {
let out = child
.wait_with_output()
.map_err(|e| format!("rsync wait_with_output failed: {e}"))?;
if out.status.success() {
return Ok(());
}
let stderr = String::from_utf8_lossy(&out.stderr);
let stdout = String::from_utf8_lossy(&out.stdout);
return Err(format!(
"rsync failed: status={} stdout={} stderr={}",
out.status,
stdout.trim(),
stderr.trim()
));
}
None => {
if started.elapsed() >= limit {
let _ = child.kill();
let out = child
.wait_with_output()
.map_err(|e| format!("rsync wait_with_output failed: {e}"))?;
let stderr = String::from_utf8_lossy(&out.stderr);
let stdout = String::from_utf8_lossy(&out.stdout);
return Err(format!(
"rsync wall-clock timeout after {}s: stdout={} stderr={}",
limit.as_secs(),
stdout.trim(),
stderr.trim()
));
}
thread::sleep(Duration::from_millis(100));
}
}
}
}
let out = child
.wait_with_output()
.map_err(|e| format!("rsync wait_with_output failed: {e}"))?;
if out.status.success() {
return Ok(());
}
let stderr = String::from_utf8_lossy(&out.stderr);
let stdout = String::from_utf8_lossy(&out.stdout);
Err(format!(
"rsync failed: status={} stdout={} stderr={}",
out.status,
stdout.trim(),
stderr.trim()
))
}
fn run_rsync_fail_fast(
&self,
src: &str,
dst: &Path,
profile: RsyncFailFastProfile,
) -> Result<(), String> {
let mut attempt = 0usize;
let mut timeout = profile.initial_wall_clock_timeout;
let mut previous_progress = (0usize, 0u64);
let mut zero_progress_attempts = 0usize;
let max_timeout = std::cmp::max(
profile.max_wall_clock_timeout,
profile.initial_wall_clock_timeout,
);
loop {
attempt += 1;
match self.run_rsync_once(src, dst, Some(timeout), true) {
Ok(()) => return Ok(()),
Err(err) => {
if is_hard_fail_rsync_error(&err) {
return Err(format!(
"rsync fail-fast hard-fail on attempt {}: {}",
attempt, err
));
}
if !err.contains("wall-clock timeout") {
return Err(err);
}
let progress = dir_progress(dst)
.map_err(|e| format!("rsync fail-fast progress stat failed: {e}"))?;
if progress == (0, 0) {
zero_progress_attempts += 1;
if zero_progress_attempts >= 2 || attempt >= profile.max_attempts {
return Err(format!(
"rsync fail-fast gave up after {} attempts with no progress: {}",
attempt, err
));
}
} else if progress == previous_progress {
return Err(format!(
"rsync fail-fast gave up after {} attempts with no additional progress: {}",
attempt, err
));
} else {
previous_progress = progress;
}
if attempt >= profile.max_attempts {
return Err(format!(
"rsync fail-fast exhausted {} attempts: {}",
profile.max_attempts, err
));
}
timeout = std::cmp::min(timeout.saturating_mul(2), max_timeout);
}
}
}
Ok(())
}
fn module_fetch_uri(&self, rsync_base_uri: &str) -> String {
@@ -140,7 +311,8 @@ impl RsyncFetcher for SystemRsyncFetcher {
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)?;
walk_dir_visit(tmp.path(), tmp.path(), &base, &mut wrapped)
.map_err(RsyncFetchError::Fetch)?;
Ok((count, bytes_total))
}
@@ -178,7 +350,10 @@ fn rsync_module_root_uri(s: &str) -> Option<String> {
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();
let mut segments: Vec<&str> = path
.split('/')
.filter(|segment| !segment.is_empty())
.collect();
if segments.is_empty() {
return None;
}
@@ -216,6 +391,38 @@ fn walk_dir_collect(
Ok(())
}
fn dir_progress(root: &Path) -> Result<(usize, u64), String> {
if !root.exists() {
return Ok((0, 0));
}
let mut files = 0usize;
let mut bytes = 0u64;
let mut stack = vec![root.to_path_buf()];
while let Some(path) = stack.pop() {
let rd = std::fs::read_dir(&path).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() {
stack.push(path);
} else if meta.is_file() {
files += 1;
bytes += meta.len();
}
}
}
Ok((files, bytes))
}
fn is_hard_fail_rsync_error(msg: &str) -> bool {
let lower = msg.to_ascii_lowercase();
lower.contains("no route to host")
|| lower.contains("network is unreachable")
|| lower.contains("connection refused")
|| lower.contains("name or service not known")
}
fn walk_dir_visit(
root: &Path,
current: &Path,
@@ -314,6 +521,7 @@ mod tests {
// 1) Spawn error.
let f = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: PathBuf::from("/this/does/not/exist/rsync"),
connect_timeout: Duration::from_secs(1),
timeout: Duration::from_secs(1),
extra_args: Vec::new(),
mirror_root: None,
@@ -326,6 +534,7 @@ mod tests {
// 2) Non-zero exit status.
let f = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: PathBuf::from("false"),
connect_timeout: Duration::from_secs(1),
timeout: Duration::from_secs(1),
extra_args: Vec::new(),
mirror_root: None,
@@ -345,6 +554,7 @@ mod tests {
let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: PathBuf::from("rsync"),
connect_timeout: Duration::from_secs(1),
timeout: Duration::from_secs(1),
extra_args: Vec::new(),
mirror_root: Some(root_file.clone()),
@@ -371,6 +581,7 @@ mod tests {
let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: PathBuf::from("rsync"),
connect_timeout: Duration::from_secs(1),
timeout: Duration::from_secs(1),
extra_args: Vec::new(),
mirror_root: Some(root.clone()),
@@ -406,4 +617,168 @@ mod tests {
assert_eq!(out.len(), 1);
assert_eq!(out[0].0, "rsync://example.net/repo/a.cer");
}
#[cfg(unix)]
#[test]
fn rsync_fail_fast_retries_when_progress_is_made() {
use std::os::unix::fs::PermissionsExt;
let temp = tempfile::tempdir().expect("tempdir");
let script = temp.path().join("fake-rsync.sh");
let state = temp.path().join("state.txt");
std::fs::write(
&script,
format!(
"#!/usr/bin/env bash\nset -euo pipefail\nSTATE=\"{}\"\nDST=\"${{@: -1}}\"\nCOUNT=0\nif [[ -f \"$STATE\" ]]; then COUNT=$(cat \"$STATE\"); fi\nCOUNT=$((COUNT+1))\necho \"$COUNT\" > \"$STATE\"\nmkdir -p \"$DST\"\nif [[ \"$COUNT\" -eq 1 ]]; then\n echo first > \"$DST/part1\"\n sleep 2\nelse\n echo second > \"$DST/part2\"\nfi\n",
state.display()
),
)
.expect("write script");
let mut perms = std::fs::metadata(&script).unwrap().permissions();
perms.set_mode(0o755);
std::fs::set_permissions(&script, perms).unwrap();
let dst = temp.path().join("dst");
let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: script,
connect_timeout: Duration::from_secs(15),
timeout: Duration::from_secs(60),
extra_args: Vec::new(),
mirror_root: None,
});
fetcher
.run_rsync_fail_fast(
"rsync://example.net/repo/",
&dst,
RsyncFailFastProfile {
initial_wall_clock_timeout: Duration::from_secs(1),
max_wall_clock_timeout: Duration::from_secs(4),
max_attempts: 3,
},
)
.expect("eventual success");
assert!(dst.join("part1").exists());
assert!(dst.join("part2").exists());
}
#[cfg(unix)]
#[test]
fn rsync_fail_fast_gives_up_after_two_zero_progress_timeouts() {
use std::os::unix::fs::PermissionsExt;
let temp = tempfile::tempdir().expect("tempdir");
let script = temp.path().join("fake-rsync.sh");
std::fs::write(&script, "#!/usr/bin/env bash\nset -euo pipefail\nsleep 5\n")
.expect("write script");
let mut perms = std::fs::metadata(&script).unwrap().permissions();
perms.set_mode(0o755);
std::fs::set_permissions(&script, perms).unwrap();
let dst = temp.path().join("dst");
let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: script,
connect_timeout: Duration::from_secs(15),
timeout: Duration::from_secs(60),
extra_args: Vec::new(),
mirror_root: None,
});
let err = fetcher
.run_rsync_fail_fast(
"rsync://example.net/repo/",
&dst,
RsyncFailFastProfile {
initial_wall_clock_timeout: Duration::from_secs(1),
max_wall_clock_timeout: Duration::from_secs(2),
max_attempts: 4,
},
)
.expect_err("must fail");
assert!(err.contains("no progress"), "{err}");
}
#[cfg(unix)]
#[test]
fn rsync_fail_fast_hard_fail_stops_after_first_attempt() {
use std::os::unix::fs::PermissionsExt;
let temp = tempfile::tempdir().expect("tempdir");
let script = temp.path().join("fake-rsync.sh");
let state = temp.path().join("state.txt");
std::fs::write(
&script,
format!(
"#!/usr/bin/env bash\nset -euo pipefail\nSTATE=\"{}\"\nCOUNT=0\nif [[ -f \"$STATE\" ]]; then COUNT=$(cat \"$STATE\"); fi\nCOUNT=$((COUNT+1))\necho \"$COUNT\" > \"$STATE\"\necho 'rsync: [Receiver] failed to connect to host (1.2.3.4): Connection refused (111)' >&2\nexit 10\n",
state.display()
),
)
.expect("write script");
let mut perms = std::fs::metadata(&script).unwrap().permissions();
perms.set_mode(0o755);
std::fs::set_permissions(&script, perms).unwrap();
let dst = temp.path().join("dst");
let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: script,
connect_timeout: Duration::from_secs(15),
timeout: Duration::from_secs(60),
extra_args: Vec::new(),
mirror_root: None,
});
let err = fetcher
.run_rsync_fail_fast(
"rsync://example.net/repo/",
&dst,
RsyncFailFastProfile {
initial_wall_clock_timeout: Duration::from_secs(10),
max_wall_clock_timeout: Duration::from_secs(80),
max_attempts: 4,
},
)
.expect_err("must hard fail");
assert!(err.contains("hard-fail"), "{err}");
let count = std::fs::read_to_string(&state).unwrap();
assert_eq!(count.trim(), "1");
}
#[cfg(unix)]
#[test]
fn run_rsync_once_passes_contimeout_and_timeout_args() {
use std::os::unix::fs::PermissionsExt;
let temp = tempfile::tempdir().expect("tempdir");
let script = temp.path().join("capture-rsync.sh");
let args_file = temp.path().join("args.txt");
std::fs::write(
&script,
format!(
"#!/usr/bin/env bash\nset -euo pipefail\nprintf '%s\\n' \"$@\" > \"{}\"\nDST=\"${{@: -1}}\"\nmkdir -p \"$DST\"\n",
args_file.display()
),
)
.expect("write script");
let mut perms = std::fs::metadata(&script).unwrap().permissions();
perms.set_mode(0o755);
std::fs::set_permissions(&script, perms).unwrap();
let dst = temp.path().join("dst");
let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: script,
connect_timeout: Duration::from_secs(15),
timeout: Duration::from_secs(30),
extra_args: Vec::new(),
mirror_root: None,
});
fetcher
.run_rsync_once("rsync://example.net/repo/", &dst, None, false)
.expect("rsync");
let args = std::fs::read_to_string(&args_file).expect("read args");
assert!(args.contains("--contimeout\n15\n"), "{args}");
assert!(args.contains("--timeout\n30\n"), "{args}");
}
}