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

20260415_2 支持多个RIR并行混合执行，APNIC+ARIN运行1 snapshot + 1 delta，对比rpki-client，snapshot比rpki-client慢4分钟(398vs137)，输出未收敛，delta时输出收敛(348vs181)，评估应该是正确性没有问题，下一步进一步优化性能

Browse Source
This commit is contained in:
yuyr
2026-04-16 11:33:52 +08:00
parent 585c41b83b
commit 38421b1ae7
5 changed files with 594 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
src/validation/run_tree_from_tal.rs
View File

@@ -26,7 +26,7 @@ use crate::validation::from_tal::{
};
use crate::validation::tree::{
CaInstanceHandle, TreeRunAuditOutput, TreeRunConfig, TreeRunError, TreeRunOutput,
run_tree_serial, run_tree_serial_audit,
run_tree_serial, run_tree_serial_audit, run_tree_serial_audit_multi_root,
};
use crate::validation::tree_runner::Rpkiv1PublicationPointRunner;
use std::collections::HashMap;
@@ -79,12 +79,20 @@ pub struct RunTreeFromTalOutput {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RunTreeFromTalAuditOutput {
pub discovery: DiscoveredRootCaInstance,
pub discoveries: Vec<DiscoveredRootCaInstance>,
pub tree: TreeRunOutput,
pub publication_points: Vec<PublicationPointAudit>,
pub downloads: Vec<crate::audit::AuditDownloadEvent>,
pub download_stats: crate::audit::AuditDownloadStats,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TalRootDiscovery {
pub tal_input: TalInputSpec,
pub discovery: DiscoveredRootCaInstance,
pub root_handle: CaInstanceHandle,
}
fn make_live_runner<'a>(
store: &'a crate::storage::RocksStore,
policy: &'a crate::policy::Policy,
@@ -149,6 +157,64 @@ where
)))
}
fn root_discovery_from_tal_input(
tal_input: &TalInputSpec,
http_fetcher: &dyn Fetcher,
rsync_fetcher: &dyn crate::fetch::rsync::RsyncFetcher,
) -> Result<DiscoveredRootCaInstance, FromTalError> {
match &tal_input.source {
TalSource::Url(url) => discover_root_ca_instance_from_tal_url(http_fetcher, url),
TalSource::DerBytes {
tal_bytes, ta_der, ..
} => discover_root_ca_instance_from_tal_and_ta_der(tal_bytes, ta_der, None),
TalSource::FilePath(path) => {
let tal_bytes = std::fs::read(path).map_err(|e| {
FromTalError::TalFetch(format!("read TAL file failed: {}: {e}", path.display()))
})?;
let tal = crate::data_model::tal::Tal::decode_bytes(&tal_bytes).map_err(FromTalError::from)?;
discover_root_ca_instance_from_tal_with_fetchers(http_fetcher, rsync_fetcher, tal, None)
}
TalSource::FilePathWithTa { tal_path, ta_path } => {
let tal_bytes = std::fs::read(tal_path).map_err(|e| {
FromTalError::TalFetch(format!(
"read TAL file failed: {}: {e}",
tal_path.display()
))
})?;
let ta_der = std::fs::read(ta_path).map_err(|e| {
FromTalError::TaFetch(format!(
"read TA file failed: {}: {e}",
ta_path.display()
))
})?;
discover_root_ca_instance_from_tal_and_ta_der(&tal_bytes, &ta_der, None)
}
}
}
fn discover_multiple_roots_from_tal_inputs(
tal_inputs: &[TalInputSpec],
http_fetcher: &dyn Fetcher,
rsync_fetcher: &dyn crate::fetch::rsync::RsyncFetcher,
) -> Result<Vec<TalRootDiscovery>, RunTreeFromTalError> {
let mut roots = Vec::with_capacity(tal_inputs.len());
for tal_input in tal_inputs {
let discovery = root_discovery_from_tal_input(tal_input, http_fetcher, rsync_fetcher)?;
let root_handle = root_handle_from_trust_anchor(
&discovery.trust_anchor,
tal_input.tal_id.clone(),
None,
&discovery.ca_instance,
);
roots.push(TalRootDiscovery {
tal_input: tal_input.clone(),
discovery,
root_handle,
});
}
Ok(roots)
}
#[derive(Debug, thiserror::Error)]
pub enum RunTreeFromTalError {
#[error("{0}")]
@@ -253,7 +319,8 @@ pub fn run_tree_from_tal_url_serial_audit(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -302,7 +369,8 @@ pub fn run_tree_from_tal_url_serial_audit_with_timing(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -360,7 +428,8 @@ where
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -395,6 +464,7 @@ where
.tal_url
.clone()
.unwrap_or_else(|| "embedded-tal".to_string()),
tal_bytes: tal_bytes.to_vec(),
ta_der: ta_der.to_vec(),
},
}];
@@ -432,7 +502,74 @@ where
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
download_stats,
})
}
pub fn run_tree_from_multiple_tals_parallel_phase1_audit<H, R>(
store: Arc<crate::storage::RocksStore>,
policy: &crate::policy::Policy,
tal_inputs: Vec<TalInputSpec>,
http_fetcher: &H,
rsync_fetcher: &R,
validation_time: time::OffsetDateTime,
config: &TreeRunConfig,
parallel_config: ParallelPhase1Config,
) -> Result<RunTreeFromTalAuditOutput, RunTreeFromTalError>
where
H: Fetcher + Clone + 'static,
R: crate::fetch::rsync::RsyncFetcher + Clone + 'static,
{
if tal_inputs.is_empty() {
return Err(RunTreeFromTalError::Replay("multi-TAL run requires at least one TAL input".to_string()));
}
let roots = discover_multiple_roots_from_tal_inputs(&tal_inputs, http_fetcher, rsync_fetcher)?;
let primary = roots
.first()
.cloned()
.ok_or_else(|| RunTreeFromTalError::Replay("multi-TAL root discovery returned no roots".to_string()))?;
let discoveries = roots.iter().map(|item| item.discovery.clone()).collect::<Vec<_>>();
let root_handles = roots
.into_iter()
.map(|item| item.root_handle)
.collect::<Vec<_>>();
let download_log = DownloadLogHandle::new();
let runtime = build_phase1_repo_sync_runtime(
Arc::clone(&store),
policy,
http_fetcher,
rsync_fetcher,
parallel_config,
None,
Some(download_log.clone()),
tal_inputs,
)?;
let runner = make_live_runner(
store.as_ref(),
policy,
http_fetcher,
rsync_fetcher,
validation_time,
None,
Some(download_log.clone()),
Some(runtime),
);
let TreeRunAuditOutput {
tree,
publication_points,
} = run_tree_serial_audit_multi_root(root_handles, &runner, config)?;
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery: primary.discovery.clone(),
discoveries,
tree,
publication_points,
downloads,
@@ -532,7 +669,8 @@ pub fn run_tree_from_tal_bytes_serial_audit(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -595,7 +733,8 @@ pub fn run_tree_from_tal_bytes_serial_audit_with_timing(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -649,7 +788,8 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -707,7 +847,8 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit_with_timing(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -823,7 +964,8 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -891,7 +1033,8 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit_with_timing(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -1089,7 +1232,8 @@ fn run_payload_delta_replay_audit_inner(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -1231,7 +1375,8 @@ fn run_payload_delta_replay_step_audit_inner(
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
Ok(RunTreeFromTalAuditOutput {
discovery,
discovery: discovery.clone(),
discoveries: vec![discovery],
tree,
publication_points,
downloads,
@@ -1266,6 +1411,75 @@ pub fn run_tree_from_tal_and_ta_der_payload_delta_replay_step_serial_audit(
)
}
#[cfg(test)]
mod multi_tal_tests {
use super::*;
struct RejectingHttpFetcher;
impl Fetcher for RejectingHttpFetcher {
fn fetch(&self, uri: &str) -> Result<Vec<u8>, String> {
Err(format!("unexpected http fetch: {uri}"))
}
}
struct RejectingRsyncFetcher;
impl crate::fetch::rsync::RsyncFetcher for RejectingRsyncFetcher {
fn fetch_objects(
&self,
_rsync_base_uri: &str,
) -> crate::fetch::rsync::RsyncFetchResult<Vec<(String, Vec<u8>)>> {
Err(crate::fetch::rsync::RsyncFetchError::Fetch(
"unexpected rsync fetch".to_string(),
))
}
fn dedup_key(&self, base_uri: &str) -> String {
base_uri.to_string()
}
}
#[test]
fn discover_multiple_roots_from_tal_inputs_builds_multiple_root_handles() {
let apnic_tal = std::fs::read("tests/fixtures/tal/apnic-rfc7730-https.tal")
.expect("read apnic tal");
let apnic_ta = std::fs::read("tests/fixtures/ta/apnic-ta.cer").expect("read apnic ta");
let arin_tal = std::fs::read("tests/fixtures/tal/arin.tal").expect("read arin tal");
let arin_ta = std::fs::read("tests/fixtures/ta/arin-ta.cer").expect("read arin ta");
let tal_inputs = vec![
TalInputSpec::from_ta_der(
"https://example.test/apnic.tal",
apnic_tal,
apnic_ta,
),
TalInputSpec::from_ta_der(
"https://example.test/arin.tal",
arin_tal,
arin_ta,
),
];
let roots = discover_multiple_roots_from_tal_inputs(
&tal_inputs,
&RejectingHttpFetcher,
&RejectingRsyncFetcher,
)
.expect("discover roots");
assert_eq!(roots.len(), 2);
assert_eq!(roots[0].tal_input.tal_id, "apnic");
assert_eq!(roots[1].tal_input.tal_id, "arin");
assert_eq!(roots[0].root_handle.tal_id, "apnic");
assert_eq!(roots[1].root_handle.tal_id, "arin");
assert_ne!(
roots[0].root_handle.manifest_rsync_uri,
roots[1].root_handle.manifest_rsync_uri
);
}
}
#[cfg(test)]
mod replay_api_tests {
use super::*;

44
src/validation/tree.rs
View File

@@ -122,6 +122,14 @@ pub fn run_tree_serial_audit(
root: CaInstanceHandle,
runner: &dyn PublicationPointRunner,
config: &TreeRunConfig,
) -> Result<TreeRunAuditOutput, TreeRunError> {
run_tree_serial_audit_multi_root(vec![root], runner, config)
}
pub fn run_tree_serial_audit_multi_root(
roots: Vec<CaInstanceHandle>,
runner: &dyn PublicationPointRunner,
config: &TreeRunConfig,
) -> Result<TreeRunAuditOutput, TreeRunError> {
#[derive(Clone, Debug)]
struct QueuedCaInstance {
@@ -133,13 +141,15 @@ pub fn run_tree_serial_audit(
let mut next_id: u64 = 0;
let mut queue: std::collections::VecDeque<QueuedCaInstance> = std::collections::VecDeque::new();
queue.push_back(QueuedCaInstance {
id: next_id,
handle: root,
parent_id: None,
discovered_from: None,
});
next_id += 1;
for root in roots {
queue.push_back(QueuedCaInstance {
id: next_id,
handle: root,
parent_id: None,
discovered_from: None,
});
next_id += 1;
}
let mut visited_manifest_uris: std::collections::HashSet<String> =
std::collections::HashSet::new();
@@ -245,6 +255,7 @@ mod tests {
use super::{
CaInstanceHandle, DiscoveredChildCaInstance, PublicationPointRunResult,
PublicationPointRunner, TreeRunConfig, run_tree_serial_audit,
run_tree_serial_audit_multi_root,
};
fn sample_handle(manifest: &str) -> CaInstanceHandle {
@@ -339,4 +350,23 @@ mod tests {
["rsync://example.test/repo/child.mft"]
);
}
#[test]
fn run_tree_serial_audit_multi_root_processes_all_roots() {
let seen = Arc::new(Mutex::new(Vec::new()));
let runner = PrefetchRecordingRunner {
seen_prefetch_children: Arc::clone(&seen),
};
let out = run_tree_serial_audit_multi_root(
vec![
sample_handle("rsync://example.test/repo/root-a.mft"),
sample_handle("rsync://example.test/repo/root-b.mft"),
],
&runner,
&TreeRunConfig::default(),
)
.expect("tree run");
assert_eq!(out.tree.instances_processed, 2);
assert_eq!(out.publication_points.len(), 2);
}
}