yuyr/panda-rpki-oss
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add configurable RRDP snapshot and delta timeouts

Browse Source
This commit is contained in:
yuyr
2026-09-29 16:39:17 +08:00
parent e3a1d0a6ff
commit 9c056aac64
12 changed files with 225 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
src/cli/config.rs
View File

@@ -28,6 +28,8 @@ pub struct Config {
pub ta_constraints_enabled: bool,
pub max_ca_depth: usize,
pub http_timeout_secs: u64,
pub rrdp_snapshot_timeout_secs: u64,
pub rrdp_delta_timeout_secs: u64,
pub http_root_cert: Vec<PathBuf>,
pub parallel_max_repo_sync_workers_global: usize,
pub parallel_phase2_object_workers: usize,
@@ -117,7 +119,11 @@ impl Default for Config {
ta_refresh_policy: "allow-valid-cache".into(),
ta_constraints_enabled: true,
max_ca_depth: 64,
http_timeout_secs: 300,
http_timeout_secs: 10,
rrdp_snapshot_timeout_secs:
crate::repository::fetch::http::DEFAULT_RRDP_SNAPSHOT_TIMEOUT_SECS,
rrdp_delta_timeout_secs:
crate::repository::fetch::http::DEFAULT_RRDP_DELTA_TIMEOUT_SECS,
http_root_cert: Vec::new(),
parallel_max_repo_sync_workers_global: 8,
parallel_phase2_object_workers: 8,
@@ -356,6 +362,15 @@ impl Config {
normalizations.push(json!({"field":"http-timeout-secs","input":0,"effective":1}));
config.http_timeout_secs = 1;
}
if config.rrdp_snapshot_timeout_secs < 1 {
normalizations
.push(json!({"field":"rrdp-snapshot-timeout-secs","input":0,"effective":1}));
config.rrdp_snapshot_timeout_secs = 1;
}
if config.rrdp_delta_timeout_secs < 1 {
normalizations.push(json!({"field":"rrdp-delta-timeout-secs","input":0,"effective":1}));
config.rrdp_delta_timeout_secs = 1;
}
if config.parallel_repo_worker_queue_capacity < 1024 {
normalizations.push(json!({"field":"parallel-repo-worker-queue-capacity","input":config.parallel_repo_worker_queue_capacity,"effective":1024}));
config.parallel_repo_worker_queue_capacity = 1024;
@@ -618,6 +633,8 @@ resource-validation-mode = "rfc6487"
validation-time = "2026-09-22T08:00:00+08:00"
sync-preference = "rsync-only"
rsync-timeout-secs = 47
rrdp-snapshot-timeout-secs = 720
rrdp-delta-timeout-secs = 90
rsync-command = {:?}
rsync-mirror-root = "mirror"
rsync-scope = "publication-point"
@@ -652,6 +669,8 @@ tal = {:?}
);
assert_eq!(config.sync_preference().unwrap(), SyncPreference::RsyncOnly);
assert_eq!(config.rsync_timeout_secs, 47);
assert_eq!(config.rrdp_snapshot_timeout_secs, 720);
assert_eq!(config.rrdp_delta_timeout_secs, 90);
assert_eq!(config.rsync_scope, RsyncScopePolicy::PublicationPoint);
assert_eq!(
config.effective_validation_time().unwrap().unix_timestamp(),
@@ -662,6 +681,8 @@ tal = {:?}
assert_eq!(printed["effective"]["resource-validation-mode"], "rfc6487");
assert_eq!(printed["effective"]["sync-preference"], "rsync-only");
assert_eq!(printed["effective"]["rsync-timeout-secs"], 47);
assert_eq!(printed["effective"]["rrdp-snapshot-timeout-secs"], 720);
assert_eq!(printed["effective"]["rrdp-delta-timeout-secs"], 90);
assert_eq!(
printed["effective"]["validation-time"],
"2026-09-22T00:00:00Z"
@@ -689,6 +710,51 @@ tal = {:?}
);
}
#[test]
fn rrdp_timeout_defaults_are_distinct_and_zero_values_are_normalized() {
let fixture = crate::test_support::synthetic_repository();
let directory = tempfile::tempdir().unwrap();
let config_path = directory.path().join("validate.toml");
std::fs::write(
&config_path,
format!(
r#"
config-version = 1
mode = "validate"
out = "out"
rrdp-snapshot-timeout-secs = 0
rrdp-delta-timeout-secs = 0
[[anchors]]
id = "custom"
tal = {:?}
"#,
fixture.tal().to_string_lossy(),
),
)
.unwrap();
let (config, printed) = Config::load(&config_path, "validate").unwrap();
assert_eq!(Config::default().http_timeout_secs, 10);
assert_eq!(Config::default().rrdp_snapshot_timeout_secs, 600);
assert_eq!(Config::default().rrdp_delta_timeout_secs, 180);
assert_eq!(config.rrdp_snapshot_timeout_secs, 1);
assert_eq!(config.rrdp_delta_timeout_secs, 1);
assert_eq!(printed["effective"]["rrdp-snapshot-timeout-secs"], 1);
assert_eq!(printed["effective"]["rrdp-delta-timeout-secs"], 1);
let normalizations = printed["normalizations"].as_array().unwrap();
assert!(
normalizations
.iter()
.any(|item| item["field"] == "rrdp-snapshot-timeout-secs")
);
assert!(
normalizations
.iter()
.any(|item| item["field"] == "rrdp-delta-timeout-secs")
);
}
#[test]
fn constraints_selection_disable_override_and_reload() {
let dir = tempfile::tempdir().unwrap();

2
src/cli/validate.rs
View File

@@ -128,6 +128,8 @@ fn execute(config: &Config, run_id: &str) -> Result<i32, String> {
strict_policy: Some(config.strict),
resource_validation_mode: Some(config.resource_validation_mode),
disable_rrdp: sync_preference == SyncPreference::RsyncOnly,
rrdp_snapshot_timeout_secs: Some(config.rrdp_snapshot_timeout_secs),
rrdp_delta_timeout_secs: Some(config.rrdp_delta_timeout_secs),
rsync_command: config.rsync_command.clone(),
rsync_timeout_secs: config.rsync_timeout_secs,
rsync_mirror_root: config.rsync_mirror_root.clone(),

132
src/repository/fetch/http.rs
View File

@@ -28,6 +28,9 @@ pub fn with_scoped_http_timeout_override<R>(timeout: Duration, f: impl FnOnce()
/// Default User-Agent sent with every outgoing HTTP request (RRDP
/// notification/snapshot/delta and TAL/TA downloads).
pub const DEFAULT_HTTP_USER_AGENT: &str = "panda-rpki/0.2";
pub const DEFAULT_LARGE_BODY_TIMEOUT_SECS: u64 = 180;
pub const DEFAULT_RRDP_SNAPSHOT_TIMEOUT_SECS: u64 = 600;
pub const DEFAULT_RRDP_DELTA_TIMEOUT_SECS: u64 = 180;
/// Environment variable that overrides the HTTP User-Agent. Unset, empty, or
/// values containing characters that are illegal in a header value fall back
@@ -53,8 +56,12 @@ pub struct HttpFetcherConfig {
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.
/// Generic large-body timeout, retained for untyped HTTP callers.
pub large_body_timeout: Duration,
/// Timeout for RRDP snapshot response bodies.
pub rrdp_snapshot_timeout: Duration,
/// Timeout for RRDP delta response bodies.
pub rrdp_delta_timeout: Duration,
pub user_agent: String,
/// Extra PEM trust anchors for HTTPS transport tests or controlled endpoints.
pub extra_root_certificates_pem: Vec<Vec<u8>>,
@@ -65,7 +72,9 @@ impl Default for HttpFetcherConfig {
Self {
connect_timeout: Duration::from_secs(15),
timeout: Duration::from_secs(30),
large_body_timeout: Duration::from_secs(180),
large_body_timeout: Duration::from_secs(DEFAULT_LARGE_BODY_TIMEOUT_SECS),
rrdp_snapshot_timeout: Duration::from_secs(DEFAULT_RRDP_SNAPSHOT_TIMEOUT_SECS),
rrdp_delta_timeout: Duration::from_secs(DEFAULT_RRDP_DELTA_TIMEOUT_SECS),
user_agent: resolve_http_user_agent(std::env::var(HTTP_USER_AGENT_ENV).ok()),
extra_root_certificates_pem: Vec::new(),
}
@@ -83,15 +92,22 @@ pub struct BlockingHttpFetcher {
large_body_client: Client,
retry_short_client: Client,
rrdp_short_client: Client,
rrdp_large_body_client: Client,
rrdp_snapshot_client: Client,
rrdp_delta_client: Client,
rrdp_retry_short_client: Client,
short_timeout: Duration,
large_body_timeout: Duration,
rrdp_snapshot_timeout: Duration,
rrdp_delta_timeout: Duration,
}
impl BlockingHttpFetcher {
pub fn new(config: HttpFetcherConfig) -> Result<Self, String> {
let short_timeout = config.timeout;
let rrdp_snapshot_timeout = config.rrdp_snapshot_timeout;
let rrdp_delta_timeout = config.rrdp_delta_timeout;
// Preserve the existing timeout behavior for untyped HTTP callers.
// RRDP's typed fetch path below uses its exact snapshot/delta profile.
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 = Self::client_builder(
@@ -126,10 +142,18 @@ impl BlockingHttpFetcher {
)?
.build()
.map_err(|e| e.to_string())?;
let rrdp_large_body_client = Self::rrdp_client_builder(
let rrdp_snapshot_client = Self::rrdp_client_builder(
&config,
connect_timeout,
large_body_timeout,
rrdp_snapshot_timeout,
config.user_agent.clone(),
)?
.build()
.map_err(|e| e.to_string())?;
let rrdp_delta_client = Self::rrdp_client_builder(
&config,
connect_timeout,
rrdp_delta_timeout,
config.user_agent.clone(),
)?
.build()
@@ -147,10 +171,13 @@ impl BlockingHttpFetcher {
large_body_client,
retry_short_client,
rrdp_short_client,
rrdp_large_body_client,
rrdp_snapshot_client,
rrdp_delta_client,
rrdp_retry_short_client,
short_timeout,
large_body_timeout,
rrdp_snapshot_timeout,
rrdp_delta_timeout,
})
}
@@ -300,18 +327,27 @@ impl BlockingHttpFetcher {
}
}
fn rrdp_client_for_uri(&self, uri: &str) -> (&Client, &'static str, Duration) {
fn rrdp_client_for_resource(
&self,
resource: RrdpResourceKind,
) -> (&Client, &'static str, Duration) {
if let Some(timeout) = HTTP_TIMEOUT_OVERRIDE.with(|cell| *cell.borrow()) {
return (&self.rrdp_retry_short_client, "retry_short", timeout);
}
if uses_large_body_timeout(uri) {
(
&self.rrdp_large_body_client,
"large_body",
self.large_body_timeout,
)
} else {
(&self.rrdp_short_client, "short", self.short_timeout)
match resource {
RrdpResourceKind::Notification => {
(&self.rrdp_short_client, "short", self.short_timeout)
}
RrdpResourceKind::Snapshot => (
&self.rrdp_snapshot_client,
"rrdp_snapshot",
self.rrdp_snapshot_timeout,
),
RrdpResourceKind::Delta => (
&self.rrdp_delta_client,
"rrdp_delta",
self.rrdp_delta_timeout,
),
}
}
@@ -336,7 +372,7 @@ impl BlockingHttpFetcher {
}
for redirects in 0..=MAX_RRDP_REDIRECTS {
let (client, profile, timeout) = self.rrdp_client_for_uri(current.as_str());
let (client, profile, timeout) = self.rrdp_client_for_resource(resource);
let response = client
.get(current.clone())
.send()
@@ -729,6 +765,8 @@ mod tests {
let http = BlockingHttpFetcher::new(HttpFetcherConfig {
timeout: Duration::from_secs(3),
large_body_timeout: Duration::from_secs(9),
rrdp_snapshot_timeout: Duration::from_secs(9),
rrdp_delta_timeout: Duration::from_secs(6),
..HttpFetcherConfig::default()
})
.expect("http");
@@ -744,6 +782,68 @@ mod tests {
assert_eq!(timeout_large, Duration::from_secs(9));
}
#[test]
fn typed_rrdp_resource_selects_its_own_timeout_independent_of_uri() {
let http = BlockingHttpFetcher::new(HttpFetcherConfig {
timeout: Duration::from_secs(10),
rrdp_snapshot_timeout: Duration::from_secs(600),
rrdp_delta_timeout: Duration::from_secs(180),
..HttpFetcherConfig::default()
})
.expect("http");
let (_, notification_profile, notification_timeout) =
http.rrdp_client_for_resource(RrdpResourceKind::Notification);
let (_, snapshot_profile, snapshot_timeout) =
http.rrdp_client_for_resource(RrdpResourceKind::Snapshot);
let (_, delta_profile, delta_timeout) =
http.rrdp_client_for_resource(RrdpResourceKind::Delta);
assert_eq!(notification_profile, "short");
assert_eq!(notification_timeout, Duration::from_secs(10));
assert_eq!(snapshot_profile, "rrdp_snapshot");
assert_eq!(snapshot_timeout, Duration::from_secs(600));
assert_eq!(delta_profile, "rrdp_delta");
assert_eq!(delta_timeout, Duration::from_secs(180));
}
#[test]
fn rrdp_snapshot_and_delta_requests_enforce_independent_timeouts() {
let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind");
let addr = listener.local_addr().expect("addr");
let server = thread::spawn(move || {
for _ in 0..2 {
let (mut stream, _) = listener.accept().expect("accept");
let mut request = [0u8; 1024];
let _ = stream.read(&mut request);
thread::sleep(StdDuration::from_millis(250));
let response =
b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok";
let _ = stream.write_all(response);
}
});
let uri = format!("http://{addr}/resource.xml");
let origin = RrdpOrigin::parse(&uri, RrdpResourceKind::Notification).expect("origin");
let http = BlockingHttpFetcher::new(HttpFetcherConfig {
timeout: Duration::from_secs(3),
rrdp_snapshot_timeout: Duration::from_secs(3),
rrdp_delta_timeout: Duration::from_millis(100),
..HttpFetcherConfig::default()
})
.expect("http");
assert_eq!(
http.fetch_rrdp(RrdpResourceKind::Snapshot, &uri, &origin)
.expect("snapshot response should fit its timeout"),
b"ok"
);
assert!(
http.fetch_rrdp(RrdpResourceKind::Delta, &uri, &origin)
.is_err()
);
server.join().expect("server thread");
}
#[test]
fn resolve_http_user_agent_falls_back_to_default() {
assert_eq!(resolve_http_user_agent(None), DEFAULT_HTTP_USER_AGENT);

25
src/runtime/run.rs
View File

@@ -112,13 +112,19 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
let validation_started = std::time::Instant::now();
crate::logging::observer::phase("runtime_setup");
let collect_current_repo_objects = false;
let mut http_config = HttpFetcherConfig {
timeout: std::time::Duration::from_secs(args.http_timeout_secs.max(1)),
extra_root_certificates_pem: http_root_certificates_pem.clone(),
..HttpFetcherConfig::default()
};
if let Some(timeout) = args.rrdp_snapshot_timeout_secs {
http_config.rrdp_snapshot_timeout = std::time::Duration::from_secs(timeout.max(1));
}
if let Some(timeout) = args.rrdp_delta_timeout_secs {
http_config.rrdp_delta_timeout = std::time::Duration::from_secs(timeout.max(1));
}
let out = if let Some(dir) = args.rsync_local_dir.as_ref() {
let http = BlockingHttpFetcher::new(HttpFetcherConfig {
timeout: std::time::Duration::from_secs(args.http_timeout_secs.max(1)),
extra_root_certificates_pem: http_root_certificates_pem.clone(),
..HttpFetcherConfig::default()
})
.map_err(|e| e.to_string())?;
let http = BlockingHttpFetcher::new(http_config.clone()).map_err(|e| e.to_string())?;
let rsync = LocalDirRsyncFetcher::new(dir);
run_online_validation_with_fetchers(
Arc::clone(&store),
@@ -132,12 +138,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
timing.as_ref().map(|(_, t)| t),
)?
} else {
let http = BlockingHttpFetcher::new(HttpFetcherConfig {
timeout: std::time::Duration::from_secs(args.http_timeout_secs.max(1)),
extra_root_certificates_pem: http_root_certificates_pem.clone(),
..HttpFetcherConfig::default()
})
.map_err(|e| e.to_string())?;
let http = BlockingHttpFetcher::new(http_config).map_err(|e| e.to_string())?;
let rsync = SystemRsyncFetcher::new(SystemRsyncConfig {
rsync_bin: args
.rsync_command

3
src/runtime/types.rs
View File

@@ -79,6 +79,9 @@ pub struct RunConfig {
pub rsync_command: Option<PathBuf>,
pub http_timeout_secs: u64,
/// Per-resource RRDP response timeouts; None selects the fetcher default.
pub rrdp_snapshot_timeout_secs: Option<u64>,
pub rrdp_delta_timeout_secs: Option<u64>,
pub http_root_cert_paths: Vec<PathBuf>,
pub rsync_timeout_secs: u64,
pub rsync_mirror_root: Option<PathBuf>,