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

20260612 优化CIR在线累计恢复性能

Browse Source
This commit is contained in:
yuyr
2026-06-12 14:26:49 +08:00
parent 4047214ddf
commit 4e6bd687db
13 changed files with 715 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/validation/run_tree_from_tal.rs
View File

@@ -96,6 +96,7 @@ pub struct RunTreeFromTalAuditOutput {
pub tree: TreeRunOutput,
pub publication_points: Vec<PublicationPointAudit>,
pub roa_cache_stats: crate::validation::objects::RoaValidationCacheStats,
pub cir_input: crate::cir::CirInputSnapshot,
pub downloads: Vec<crate::audit::AuditDownloadEvent>,
pub download_stats: crate::audit::AuditDownloadStats,
pub current_repo_objects: Vec<CurrentRepoObject>,
@@ -608,6 +609,7 @@ pub fn run_tree_from_tal_url_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
let downloads = download_log.snapshot_events();
@@ -619,6 +621,7 @@ pub fn run_tree_from_tal_url_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -669,6 +672,7 @@ pub fn run_tree_from_tal_url_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
let downloads = download_log.snapshot_events();
@@ -680,6 +684,7 @@ pub fn run_tree_from_tal_url_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -754,6 +759,7 @@ where
tree,
publication_points,
roa_cache_stats,
cir_input,
} = if phase2_enabled {
run_tree_parallel_phase2_audit(root, &runner, config)?
} else {
@@ -769,6 +775,7 @@ where
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: snapshot_current_repo_objects(
@@ -872,6 +879,7 @@ where
tree,
publication_points,
roa_cache_stats,
cir_input,
} = if phase2_enabled {
run_tree_parallel_phase2_audit_multi_root(root_handles, &runner, config)?
} else {
@@ -887,6 +895,7 @@ where
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: snapshot_current_repo_objects(
@@ -1355,6 +1364,7 @@ pub fn run_tree_from_tal_bytes_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
let downloads = download_log.snapshot_events();
@@ -1366,6 +1376,7 @@ pub fn run_tree_from_tal_bytes_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -1429,6 +1440,7 @@ pub fn run_tree_from_tal_bytes_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
drop(_tree);
@@ -1441,6 +1453,7 @@ pub fn run_tree_from_tal_bytes_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -1500,6 +1513,7 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
let downloads = download_log.snapshot_events();
@@ -1511,6 +1525,7 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -1574,6 +1589,7 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
let downloads = download_log.snapshot_events();
@@ -1585,6 +1601,7 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -1716,6 +1733,7 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
let downloads = download_log.snapshot_events();
@@ -1727,6 +1745,7 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -1800,6 +1819,7 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &runner, config)?;
let downloads = download_log.snapshot_events();
@@ -1811,6 +1831,7 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit_with_timing(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -1994,7 +2015,7 @@ fn run_payload_delta_replay_audit_inner(
.map_err(|e| RunTreeFromTalError::Replay(e.to_string()))?;
let delta_rsync_fetcher = PayloadDeltaReplayRsyncFetcher::new(base_index, delta_index.clone());
let download_log = DownloadLogHandle::new();
let (tree, publication_points, roa_cache_stats) = if let Some(t) = timing.as_ref() {
let (tree, publication_points, roa_cache_stats, cir_input) = if let Some(t) = timing.as_ref() {
let _phase = t.span_phase("payload_delta_replay_target_total");
let delta_runner = build_payload_delta_replay_runner(
store,
@@ -2011,8 +2032,9 @@ fn run_payload_delta_replay_audit_inner(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &delta_runner, config)?;
(tree, publication_points, roa_cache_stats)
(tree, publication_points, roa_cache_stats, cir_input)
} else {
let delta_runner = build_payload_delta_replay_runner(
store,
@@ -2029,8 +2051,9 @@ fn run_payload_delta_replay_audit_inner(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &delta_runner, config)?;
(tree, publication_points, roa_cache_stats)
(tree, publication_points, roa_cache_stats, cir_input)
};
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
@@ -2041,6 +2064,7 @@ fn run_payload_delta_replay_audit_inner(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),
@@ -2153,7 +2177,7 @@ fn run_payload_delta_replay_step_audit_inner(
PayloadDeltaReplayCurrentStoreRsyncFetcher::new(store, delta_index.clone());
let download_log = DownloadLogHandle::new();
let (tree, publication_points, roa_cache_stats) = if let Some(t) = timing.as_ref() {
let (tree, publication_points, roa_cache_stats, cir_input) = if let Some(t) = timing.as_ref() {
let _phase = t.span_phase("payload_delta_replay_step_total");
let delta_runner = build_payload_delta_replay_current_store_runner(
store,
@@ -2170,8 +2194,9 @@ fn run_payload_delta_replay_step_audit_inner(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &delta_runner, config)?;
(tree, publication_points, roa_cache_stats)
(tree, publication_points, roa_cache_stats, cir_input)
} else {
let delta_runner = build_payload_delta_replay_current_store_runner(
store,
@@ -2188,8 +2213,9 @@ fn run_payload_delta_replay_step_audit_inner(
tree,
publication_points,
roa_cache_stats,
cir_input,
} = run_tree_serial_audit(root, &delta_runner, config)?;
(tree, publication_points, roa_cache_stats)
(tree, publication_points, roa_cache_stats, cir_input)
};
let downloads = download_log.snapshot_events();
let download_stats = DownloadLogHandle::stats_from_events(&downloads);
@@ -2200,6 +2226,7 @@ fn run_payload_delta_replay_step_audit_inner(
tree,
publication_points,
roa_cache_stats,
cir_input,
downloads,
download_stats,
current_repo_objects: Vec::new(),

170
src/validation/tree.rs
View File

@@ -1,5 +1,5 @@
use crate::audit::DiscoveredFrom;
use crate::audit::PublicationPointAudit;
use crate::audit::{DiscoveredFrom, ObjectAuditEntry, PublicationPointAudit};
use crate::cir::{CirInputAccumulator, CirInputSnapshot};
use crate::data_model::rc::{AsResourceSet, IpResourceSet};
use crate::report::Warning;
use crate::validation::manifest::PublicationPointSource;
@@ -77,6 +77,8 @@ pub struct PublicationPointRunResult {
pub warnings: Vec<Warning>,
pub objects: ObjectsOutput,
pub audit: PublicationPointAudit,
pub cir_fresh_objects: Vec<ObjectAuditEntry>,
pub cir_cached_objects: Vec<ObjectAuditEntry>,
/// Candidate child CA instances to enqueue after this publication point completes.
///
/// - For `Fresh`, these are discovered from the current validated publication point.
@@ -126,6 +128,42 @@ pub struct TreeRunAuditOutput {
pub tree: TreeRunOutput,
pub publication_points: Vec<PublicationPointAudit>,
pub roa_cache_stats: RoaValidationCacheStats,
pub cir_input: CirInputSnapshot,
}
pub(crate) fn submit_publication_point_cir_input(
cir_input: &mut CirInputAccumulator,
source: PublicationPointSource,
audit: &PublicationPointAudit,
cir_fresh_objects: &[ObjectAuditEntry],
cir_cached_objects: &[ObjectAuditEntry],
) -> Result<(), TreeRunError> {
match source {
PublicationPointSource::Fresh => {
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Fresh, &audit.objects)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
}
PublicationPointSource::FailedFetchNoCache => {
let entries = if cir_fresh_objects.is_empty() {
&audit.objects
} else {
cir_fresh_objects
};
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Fresh, entries)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
}
PublicationPointSource::VcirCurrentInstance => {
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Fresh, cir_fresh_objects)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Cached, cir_cached_objects)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
}
}
Ok(())
}
pub fn run_tree_serial(
@@ -179,6 +217,7 @@ pub fn run_tree_serial_audit_multi_root(
let mut router_keys: Vec<RouterKeyPayload> = Vec::new();
let mut publication_points: Vec<PublicationPointAudit> = Vec::new();
let mut roa_cache_stats = RoaValidationCacheStats::default();
let mut cir_input = CirInputAccumulator::default();
while let Some(node) = queue.pop_front() {
let ca = &node.handle;
@@ -218,10 +257,20 @@ pub fn run_tree_serial_audit_multi_root(
aspas.extend(res.objects.aspas);
router_keys.extend(res.objects.router_keys);
let source = res.source;
let cir_fresh_objects = res.cir_fresh_objects;
let cir_cached_objects = res.cir_cached_objects;
let mut audit = res.audit;
audit.node_id = Some(node.id);
audit.parent_node_id = node.parent_id;
audit.discovered_from = node.discovered_from.clone();
submit_publication_point_cir_input(
&mut cir_input,
source,
&audit,
&cir_fresh_objects,
&cir_cached_objects,
)?;
if config.compact_audit {
audit.objects.clear();
audit.warnings.clear();
@@ -267,6 +316,7 @@ pub fn run_tree_serial_audit_multi_root(
},
publication_points,
roa_cache_stats,
cir_input: cir_input.finalize(),
})
}
@@ -274,7 +324,9 @@ pub fn run_tree_serial_audit_multi_root(
mod tests {
use std::sync::{Arc, Mutex};
use crate::audit::{DiscoveredFrom, PublicationPointAudit};
use crate::audit::{
AuditObjectKind, AuditObjectResult, DiscoveredFrom, ObjectAuditEntry, PublicationPointAudit,
};
use crate::validation::objects::{ObjectsOutput, ObjectsStats};
use super::{
@@ -353,6 +405,8 @@ mod tests {
roa_cache_stats: crate::validation::objects::RoaValidationCacheStats::default(),
},
audit: PublicationPointAudit::default(),
cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(),
discovered_children: children,
})
}
@@ -395,4 +449,114 @@ mod tests {
assert_eq!(out.tree.instances_processed, 2);
assert_eq!(out.publication_points.len(), 2);
}
#[test]
fn run_tree_serial_audit_accumulates_fresh_and_cached_cir_input() {
struct CirRunner;
impl PublicationPointRunner for CirRunner {
fn run_publication_point(
&self,
ca: &CaInstanceHandle,
) -> Result<PublicationPointRunResult, String> {
if ca.manifest_rsync_uri.ends_with("cached.mft") {
let fresh_reject = ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/cached/current.mft".to_string(),
sha256_hex: "11".repeat(32),
kind: AuditObjectKind::Manifest,
result: AuditObjectResult::Error,
detail: Some("fresh manifest rejected".to_string()),
};
let cached = ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/cached/old.roa".to_string(),
sha256_hex: "22".repeat(32),
kind: AuditObjectKind::Roa,
result: AuditObjectResult::Ok,
detail: None,
};
return Ok(PublicationPointRunResult {
source:
crate::validation::manifest::PublicationPointSource::VcirCurrentInstance,
snapshot: None,
warnings: Vec::new(),
objects: ObjectsOutput {
vrps: Vec::new(),
aspas: Vec::new(),
router_keys: Vec::new(),
local_outputs_cache: Vec::new(),
warnings: Vec::new(),
stats: ObjectsStats::default(),
audit: Vec::new(),
roa_cache_stats:
crate::validation::objects::RoaValidationCacheStats::default(),
},
audit: PublicationPointAudit {
objects: vec![fresh_reject.clone(), cached.clone()],
..PublicationPointAudit::default()
},
cir_fresh_objects: vec![fresh_reject],
cir_cached_objects: vec![cached],
discovered_children: Vec::new(),
});
}
let fresh = ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/fresh/a.roa".to_string(),
sha256_hex: "33".repeat(32),
kind: AuditObjectKind::Roa,
result: AuditObjectResult::Ok,
detail: None,
};
Ok(PublicationPointRunResult {
source: crate::validation::manifest::PublicationPointSource::Fresh,
snapshot: None,
warnings: Vec::new(),
objects: ObjectsOutput {
vrps: Vec::new(),
aspas: Vec::new(),
router_keys: Vec::new(),
local_outputs_cache: Vec::new(),
warnings: Vec::new(),
stats: ObjectsStats::default(),
audit: Vec::new(),
roa_cache_stats:
crate::validation::objects::RoaValidationCacheStats::default(),
},
audit: PublicationPointAudit {
objects: vec![fresh],
..PublicationPointAudit::default()
},
cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(),
discovered_children: Vec::new(),
})
}
}
let out = run_tree_serial_audit_multi_root(
vec![
sample_handle("rsync://example.test/repo/fresh/root.mft"),
sample_handle("rsync://example.test/repo/cached/cached.mft"),
],
&CirRunner,
&TreeRunConfig::default(),
)
.expect("tree run");
assert_eq!(out.cir_input.fresh_validated_objects.len(), 2);
assert_eq!(out.cir_input.cached_validated_objects.len(), 1);
assert_eq!(out.cir_input.fresh_rejected_objects.len(), 1);
assert!(
out.cir_input
.fresh_validated_objects
.iter()
.any(|item| item.rsync_uri == "rsync://example.test/repo/fresh/a.roa")
);
assert!(
out.cir_input
.cached_validated_objects
.iter()
.any(|item| item.rsync_uri == "rsync://example.test/repo/cached/old.roa")
);
}
}

36
src/validation/tree_parallel.rs
View File

@@ -3,6 +3,7 @@ use std::sync::mpsc::{self, Receiver, SyncSender, TryRecvError, TrySendError};
use std::time::{Duration, Instant};
use crate::audit::{DiscoveredFrom, PublicationPointAudit};
use crate::cir::CirInputAccumulator;
use crate::parallel::object_worker::ObjectWorkerSubmitError;
use crate::parallel::repo_runtime::{RepoSyncRequestStatus, RepoSyncRuntimeOutcome};
use crate::parallel::types::RepoIdentity;
@@ -10,6 +11,7 @@ use crate::policy::SignedObjectFailurePolicy;
use crate::report::Warning;
use crate::storage::VcirReplaceTimingBreakdown;
use crate::validation::manifest::PublicationPointData;
use crate::validation::manifest::PublicationPointSource;
use crate::validation::objects::{
ObjectsOutput, OwnedRoaTask, ParallelObjectsPrepare, ParallelObjectsStage,
RoaValidationCacheInput, prepare_publication_point_for_parallel_roa_with_cache,
@@ -88,9 +90,12 @@ impl FinishedPublicationPointNode {
#[derive(Debug)]
enum FinishedPublicationPointResult {
Ok {
source: PublicationPointSource,
warnings: Vec<Warning>,
objects: ObjectsOutput,
audit: PublicationPointAudit,
cir_fresh_objects: Vec<crate::audit::ObjectAuditEntry>,
cir_cached_objects: Vec<crate::audit::ObjectAuditEntry>,
},
Err(String),
}
@@ -410,14 +415,22 @@ fn compact_phase2_finished_result(
) -> FinishedPublicationPointResult {
result.objects.audit.clear();
result.objects.local_outputs_cache.clear();
let cir_fresh_objects = if compact_audit && result.source == PublicationPointSource::Fresh {
result.audit.objects.clone()
} else {
result.cir_fresh_objects
};
if compact_audit {
result.audit.objects.clear();
result.audit.warnings.clear();
}
FinishedPublicationPointResult::Ok {
source: result.source,
warnings: result.warnings,
objects: result.objects,
audit: result.audit,
cir_fresh_objects,
cir_cached_objects: result.cir_cached_objects,
}
}
@@ -2025,13 +2038,17 @@ fn build_tree_output(mut finished: Vec<FinishedPublicationPoint>) -> TreeRunAudi
let mut router_keys = Vec::new();
let mut publication_points = Vec::new();
let mut roa_cache_stats = crate::validation::objects::RoaValidationCacheStats::default();
let mut cir_input = CirInputAccumulator::default();
for item in finished {
match item.result {
FinishedPublicationPointResult::Ok {
source,
warnings: result_warnings,
objects,
audit,
cir_fresh_objects,
cir_cached_objects,
} => {
instances_processed += 1;
warnings.extend(result_warnings);
@@ -2045,6 +2062,14 @@ fn build_tree_output(mut finished: Vec<FinishedPublicationPoint>) -> TreeRunAudi
audit.node_id = Some(item.node.id);
audit.parent_node_id = item.node.parent_id;
audit.discovered_from = item.node.discovered_from;
crate::validation::tree::submit_publication_point_cir_input(
&mut cir_input,
source,
&audit,
&cir_fresh_objects,
&cir_cached_objects,
)
.expect("CIR input collection from validated audit must not fail");
publication_points.push(audit);
}
FinishedPublicationPointResult::Err(err) => {
@@ -2068,6 +2093,7 @@ fn build_tree_output(mut finished: Vec<FinishedPublicationPoint>) -> TreeRunAudi
},
publication_points,
roa_cache_stats,
cir_input: cir_input.finalize(),
}
}
@@ -2133,6 +2159,8 @@ mod tests {
roa_cache_stats: crate::validation::objects::RoaValidationCacheStats::default(),
},
audit: PublicationPointAudit::default(),
cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(),
discovered_children: Vec::new(),
}
}
@@ -2172,10 +2200,16 @@ mod tests {
});
let result = compact_phase2_finished_result(sample, true);
match result {
FinishedPublicationPointResult::Ok { objects, audit, .. } => {
FinishedPublicationPointResult::Ok {
objects,
audit,
cir_fresh_objects,
..
} => {
assert!(audit.objects.is_empty());
assert!(audit.warnings.is_empty());
assert!(objects.audit.is_empty());
assert_eq!(cir_fresh_objects.len(), 1);
}
FinishedPublicationPointResult::Err(err) => panic!("unexpected error: {err}"),
}

35
src/validation/tree_runner.rs
View File

@@ -440,6 +440,8 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
warnings,
objects,
audit,
cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(),
discovered_children,
},
snapshot_pack_ms,
@@ -996,12 +998,33 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
&fresh_failure_audits,
);
let audit_build_ms = audit_build_started.elapsed().as_millis() as u64;
let cir_cached_objects =
if projection.source == PublicationPointSource::VcirCurrentInstance {
audit
.objects
.iter()
.filter(|entry| {
!fresh_failure_audits.iter().any(|fresh| fresh == *entry)
})
.cloned()
.collect()
} else {
Vec::new()
};
let result = PublicationPointRunResult {
source: projection.source,
snapshot: projection.snapshot,
warnings,
objects: projection.objects,
audit,
cir_fresh_objects: if projection.source
== PublicationPointSource::VcirCurrentInstance
{
fresh_failure_audits
} else {
Vec::new()
},
cir_cached_objects,
discovered_children: projection.discovered_children,
};
let total_duration_ms =
@@ -1930,9 +1953,7 @@ fn build_publication_point_audit_from_snapshot(
next_update_rfc3339_utc: pack.next_update.rfc3339_utc.clone(),
verified_at_rfc3339_utc: pack.verified_at.rfc3339_utc.clone(),
warnings,
objects: objects_out.clone(),
cir_fresh_objects: objects_out,
cir_cached_objects: Vec::new(),
objects: objects_out,
}
}
@@ -1989,8 +2010,6 @@ fn build_publication_point_audit_from_vcir(
verified_at_rfc3339_utc: String::new(),
warnings,
objects: fresh_failure_audits.to_vec(),
cir_fresh_objects: fresh_failure_audits.to_vec(),
cir_cached_objects: Vec::new(),
};
};
@@ -2027,9 +2046,7 @@ fn build_publication_point_audit_from_vcir(
.clone(),
verified_at_rfc3339_utc: vcir.last_successful_validation_time.rfc3339_utc.clone(),
warnings,
objects: objects_out.clone(),
cir_fresh_objects: objects_out,
cir_cached_objects: Vec::new(),
objects: objects_out,
};
}
@@ -2124,8 +2141,6 @@ fn build_publication_point_audit_from_vcir(
verified_at_rfc3339_utc: vcir.last_successful_validation_time.rfc3339_utc.clone(),
warnings,
objects: audit_objects,
cir_fresh_objects: fresh_failure_audits.to_vec(),
cir_cached_objects: objects_out,
}
}