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

20260428 降低all5 CIR replay内存峰值

Browse Source
This commit is contained in:
yuyr
2026-04-28 09:58:43 +08:00
parent 0295fd3262
commit 3b2a160c5c
23 changed files with 965 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
src/validation/objects.rs
View File

@@ -134,6 +134,30 @@ pub fn process_publication_point_for_issuer<P: PublicationPointData>(
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
) -> ObjectsOutput {
process_publication_point_for_issuer_with_options(
publication_point,
policy,
issuer_ca_der,
issuer_ca_rsync_uri,
issuer_effective_ip,
issuer_effective_as,
validation_time,
timing,
true,
)
}
pub fn process_publication_point_for_issuer_with_options<P: PublicationPointData>(
publication_point: &P,
policy: &Policy,
issuer_ca_der: &[u8],
issuer_ca_rsync_uri: Option<&str>,
issuer_effective_ip: Option<&crate::data_model::rc::IpResourceSet>,
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
collect_vcir_local_outputs: bool,
) -> ObjectsOutput {
let manifest_rsync_uri = publication_point.manifest_rsync_uri();
let manifest_bytes = publication_point.manifest_bytes();
@@ -321,12 +345,15 @@ pub fn process_publication_point_for_issuer<P: PublicationPointData>(
issuer_effective_as,
validation_time,
timing,
collect_vcir_local_outputs,
);
match result.outcome {
Ok(mut ok) => {
stats.roa_ok += 1;
vrps.append(&mut ok.vrps);
local_outputs_cache.extend(ok.local_outputs);
if collect_vcir_local_outputs {
local_outputs_cache.extend(ok.local_outputs);
}
audit.push(ObjectAuditEntry {
rsync_uri: result.rsync_uri,
sha256_hex: result.sha256_hex,
@@ -426,11 +453,14 @@ pub fn process_publication_point_for_issuer<P: PublicationPointData>(
issuer_effective_as,
validation_time,
timing,
collect_vcir_local_outputs,
) {
Ok((att, local_output)) => {
stats.aspa_ok += 1;
aspas.push(att);
local_outputs_cache.push(local_output);
if let Some(local_output) = local_output {
local_outputs_cache.push(local_output);
}
audit.push(ObjectAuditEntry {
rsync_uri: file.rsync_uri.clone(),
sha256_hex: sha256_hex_from_32(&file.sha256),
@@ -536,11 +566,37 @@ pub fn process_publication_point_for_issuer_parallel_roa<P: PublicationPointData
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
config: &ParallelPhase2Config,
) -> ObjectsOutput {
process_publication_point_for_issuer_parallel_roa_with_options(
publication_point,
policy,
issuer_ca_der,
issuer_ca_rsync_uri,
issuer_effective_ip,
issuer_effective_as,
validation_time,
timing,
config,
true,
)
}
pub fn process_publication_point_for_issuer_parallel_roa_with_options<P: PublicationPointData>(
publication_point: &P,
policy: &Policy,
issuer_ca_der: &[u8],
issuer_ca_rsync_uri: Option<&str>,
issuer_effective_ip: Option<&crate::data_model::rc::IpResourceSet>,
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
config: &ParallelPhase2Config,
collect_vcir_local_outputs: bool,
) -> ObjectsOutput {
if config.object_workers <= 1
|| policy.signed_object_failure_policy == SignedObjectFailurePolicy::DropPublicationPoint
{
return process_publication_point_for_issuer(
return process_publication_point_for_issuer_with_options(
publication_point,
policy,
issuer_ca_der,
@@ -549,13 +605,14 @@ pub fn process_publication_point_for_issuer_parallel_roa<P: PublicationPointData
issuer_effective_as,
validation_time,
timing,
collect_vcir_local_outputs,
);
}
let pool = match ParallelRoaWorkerPool::new(config) {
Ok(pool) => pool,
Err(_) => {
return process_publication_point_for_issuer(
return process_publication_point_for_issuer_with_options(
publication_point,
policy,
issuer_ca_der,
@@ -564,11 +621,12 @@ pub fn process_publication_point_for_issuer_parallel_roa<P: PublicationPointData
issuer_effective_as,
validation_time,
timing,
collect_vcir_local_outputs,
);
}
};
process_publication_point_for_issuer_parallel_roa_with_pool(
process_publication_point_for_issuer_parallel_roa_with_pool_options(
publication_point,
policy,
issuer_ca_der,
@@ -578,6 +636,7 @@ pub fn process_publication_point_for_issuer_parallel_roa<P: PublicationPointData
validation_time,
timing,
&pool,
collect_vcir_local_outputs,
)
}
@@ -591,9 +650,37 @@ pub fn process_publication_point_for_issuer_parallel_roa_with_pool<P: Publicatio
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
pool: &ParallelRoaWorkerPool,
) -> ObjectsOutput {
process_publication_point_for_issuer_parallel_roa_with_pool_options(
publication_point,
policy,
issuer_ca_der,
issuer_ca_rsync_uri,
issuer_effective_ip,
issuer_effective_as,
validation_time,
timing,
pool,
true,
)
}
pub fn process_publication_point_for_issuer_parallel_roa_with_pool_options<
P: PublicationPointData,
>(
publication_point: &P,
policy: &Policy,
issuer_ca_der: &[u8],
issuer_ca_rsync_uri: Option<&str>,
issuer_effective_ip: Option<&crate::data_model::rc::IpResourceSet>,
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
pool: &ParallelRoaWorkerPool,
collect_vcir_local_outputs: bool,
) -> ObjectsOutput {
if policy.signed_object_failure_policy == SignedObjectFailurePolicy::DropPublicationPoint {
return process_publication_point_for_issuer(
return process_publication_point_for_issuer_with_options(
publication_point,
policy,
issuer_ca_der,
@@ -602,6 +689,7 @@ pub fn process_publication_point_for_issuer_parallel_roa_with_pool<P: Publicatio
issuer_effective_as,
validation_time,
timing,
collect_vcir_local_outputs,
);
}
@@ -615,9 +703,10 @@ pub fn process_publication_point_for_issuer_parallel_roa_with_pool<P: Publicatio
validation_time,
timing,
pool,
collect_vcir_local_outputs,
)
.unwrap_or_else(|_| {
process_publication_point_for_issuer(
process_publication_point_for_issuer_with_options(
publication_point,
policy,
issuer_ca_der,
@@ -626,6 +715,7 @@ pub fn process_publication_point_for_issuer_parallel_roa_with_pool<P: Publicatio
issuer_effective_as,
validation_time,
timing,
collect_vcir_local_outputs,
)
})
}
@@ -645,6 +735,7 @@ pub(crate) struct OwnedRoaTask {
issuer_effective_ip: Option<crate::data_model::rc::IpResourceSet>,
issuer_effective_as: Option<crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
collect_vcir_local_outputs: bool,
}
#[derive(Clone)]
@@ -736,6 +827,7 @@ fn validate_owned_roa_task(task: OwnedRoaTask) -> RoaTaskResult {
task.issuer_effective_as.as_ref(),
task.validation_time,
None,
task.collect_vcir_local_outputs,
)
.map(|(vrps, local_outputs)| RoaTaskOk {
vrps,
@@ -769,6 +861,7 @@ pub(crate) struct ParallelObjectsStage {
issuer_effective_ip: Option<crate::data_model::rc::IpResourceSet>,
issuer_effective_as: Option<crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
collect_vcir_local_outputs: bool,
warnings: Vec<Warning>,
stats: ObjectsStats,
audit: Vec<ObjectAuditEntry>,
@@ -794,6 +887,7 @@ impl ParallelObjectsStage {
issuer_effective_ip: self.issuer_effective_ip.clone(),
issuer_effective_as: self.issuer_effective_as.clone(),
validation_time: self.validation_time,
collect_vcir_local_outputs: self.collect_vcir_local_outputs,
})
.collect()
}
@@ -811,6 +905,7 @@ pub(crate) fn prepare_publication_point_for_parallel_roa<P: PublicationPointData
issuer_effective_ip: Option<&crate::data_model::rc::IpResourceSet>,
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
collect_vcir_local_outputs: bool,
) -> ParallelObjectsPrepare {
let manifest_rsync_uri = publication_point.manifest_rsync_uri();
let manifest_bytes = publication_point.manifest_bytes();
@@ -985,6 +1080,7 @@ pub(crate) fn prepare_publication_point_for_parallel_roa<P: PublicationPointData
issuer_effective_ip: issuer_effective_ip.cloned(),
issuer_effective_as: issuer_effective_as.cloned(),
validation_time,
collect_vcir_local_outputs,
warnings,
stats,
audit,
@@ -1025,7 +1121,9 @@ pub(crate) fn reduce_parallel_roa_stage(
Ok(mut ok) => {
stats.roa_ok += 1;
vrps.append(&mut ok.vrps);
local_outputs_cache.extend(ok.local_outputs);
if stage.collect_vcir_local_outputs {
local_outputs_cache.extend(ok.local_outputs);
}
audit.push(ObjectAuditEntry {
rsync_uri: result.rsync_uri,
sha256_hex: result.sha256_hex,
@@ -1066,11 +1164,14 @@ pub(crate) fn reduce_parallel_roa_stage(
stage.issuer_effective_as.as_ref(),
stage.validation_time,
timing,
stage.collect_vcir_local_outputs,
) {
Ok((att, local_output)) => {
stats.aspa_ok += 1;
aspas.push(att);
local_outputs_cache.push(local_output);
if let Some(local_output) = local_output {
local_outputs_cache.push(local_output);
}
audit.push(ObjectAuditEntry {
rsync_uri: file.rsync_uri.clone(),
sha256_hex: sha256_hex_from_32(&file.sha256),
@@ -1120,6 +1221,7 @@ fn process_publication_point_for_issuer_parallel_roa_inner<P: PublicationPointDa
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
pool: &ParallelRoaWorkerPool,
collect_vcir_local_outputs: bool,
) -> Result<ObjectsOutput, String> {
let stage = match prepare_publication_point_for_parallel_roa(
0,
@@ -1129,6 +1231,7 @@ fn process_publication_point_for_issuer_parallel_roa_inner<P: PublicationPointDa
issuer_effective_ip,
issuer_effective_as,
validation_time,
collect_vcir_local_outputs,
) {
ParallelObjectsPrepare::Complete(out) => return Ok(out),
ParallelObjectsPrepare::Staged(stage) => stage,
@@ -1279,6 +1382,7 @@ pub(crate) fn validate_roa_task_serial(
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
collect_vcir_local_outputs: bool,
) -> RoaTaskResult {
let sha256_hex = sha256_hex_from_32(&task.file.sha256);
let outcome = process_roa_with_issuer(
@@ -1294,6 +1398,7 @@ pub(crate) fn validate_roa_task_serial(
issuer_effective_as,
validation_time,
timing,
collect_vcir_local_outputs,
)
.map(|(vrps, local_outputs)| RoaTaskOk {
vrps,
@@ -1322,6 +1427,7 @@ fn process_roa_with_issuer(
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
collect_vcir_local_outputs: bool,
) -> Result<(Vec<Vrp>, Vec<VcirLocalOutput>), ObjectValidateError> {
let _decode = timing
.as_ref()
@@ -1378,6 +1484,9 @@ fn process_roa_with_issuer(
drop(_subset);
let vrps = roa_to_vrps(&roa);
if !collect_vcir_local_outputs {
return Ok((vrps, Vec::new()));
}
let source_object_hash = sha256_hex_from_32(&file.sha256);
let source_ee_cert_hash = crate::audit::sha256_hex(ee.raw_der.as_slice());
let item_effective_until =
@@ -1434,6 +1543,7 @@ fn process_roa_with_issuer_parallel_cached(
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
collect_vcir_local_outputs: bool,
) -> Result<(Vec<Vrp>, Vec<VcirLocalOutput>), ObjectValidateError> {
let _decode = timing
.as_ref()
@@ -1496,6 +1606,9 @@ fn process_roa_with_issuer_parallel_cached(
drop(_subset);
let vrps = roa_to_vrps(&roa);
if !collect_vcir_local_outputs {
return Ok((vrps, Vec::new()));
}
let source_object_hash = sha256_hex_from_32(&file.sha256);
let source_ee_cert_hash = crate::audit::sha256_hex(ee.raw_der.as_slice());
let item_effective_until =
@@ -1552,7 +1665,8 @@ fn process_aspa_with_issuer(
issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
validation_time: time::OffsetDateTime,
timing: Option<&TimingHandle>,
) -> Result<(AspaAttestation, VcirLocalOutput), ObjectValidateError> {
collect_vcir_local_outputs: bool,
) -> Result<(AspaAttestation, Option<VcirLocalOutput>), ObjectValidateError> {
let _decode = timing
.as_ref()
.map(|t| t.span_phase("objects_aspa_decode_and_validate_total"));
@@ -1611,6 +1725,9 @@ fn process_aspa_with_issuer(
customer_as_id: aspa.aspa.customer_as_id,
provider_as_ids: aspa.aspa.provider_as_ids.clone(),
};
if !collect_vcir_local_outputs {
return Ok((attestation, None));
}
let source_object_hash = sha256_hex_from_32(&file.sha256);
let source_ee_cert_hash = crate::audit::sha256_hex(ee.raw_der.as_slice());
let item_effective_until =
@@ -1649,7 +1766,7 @@ fn process_aspa_with_issuer(
],
};
Ok((attestation, local_output))
Ok((attestation, Some(local_output)))
}
fn vrp_prefix_to_string(vrp: &Vrp) -> String {

1
src/validation/run.rs
View File

@@ -78,6 +78,7 @@ pub fn run_publication_point_once(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let result = runner

55
src/validation/run_tree_from_tal.rs
View File

@@ -125,6 +125,7 @@ fn make_live_runner<'a>(
repo_sync_runtime: Option<Arc<dyn RepoSyncRuntime>>,
parallel_phase2_config: Option<ParallelPhase2Config>,
ccr_accumulator: Option<CcrAccumulator>,
persist_vcir: bool,
) -> Rpkiv1PublicationPointRunner<'a> {
let parallel_roa_worker_pool = parallel_phase2_config
.as_ref()
@@ -148,6 +149,7 @@ fn make_live_runner<'a>(
parallel_phase2_config,
parallel_roa_worker_pool,
ccr_accumulator: ccr_accumulator.map(Mutex::new),
persist_vcir,
}
}
@@ -300,6 +302,7 @@ pub fn run_tree_from_tal_url_serial(
None,
None,
None,
config.persist_vcir,
);
let root = root_handle_from_trust_anchor(
@@ -337,6 +340,7 @@ pub fn run_tree_from_tal_url_serial_audit(
None,
None,
None,
config.persist_vcir,
);
let root = root_handle_from_trust_anchor(
@@ -391,6 +395,7 @@ pub fn run_tree_from_tal_url_serial_audit_with_timing(
None,
None,
None,
config.persist_vcir,
);
let root = root_handle_from_trust_anchor(
@@ -460,7 +465,9 @@ where
Some(current_repo_index),
Some(runtime),
phase2_config,
phase2_enabled.then(|| CcrAccumulator::new(vec![discovery.trust_anchor.clone()])),
(phase2_enabled && config.build_ccr_accumulator)
.then(|| CcrAccumulator::new(vec![discovery.trust_anchor.clone()])),
config.persist_vcir,
);
let root = root_handle_from_trust_anchor(
@@ -490,9 +497,7 @@ where
Some(&current_repo_index_for_output),
collect_current_repo_objects,
),
ccr_accumulator: phase2_enabled
.then(|| runner.ccr_accumulator_snapshot())
.flatten(),
ccr_accumulator: runner.ccr_accumulator_snapshot(),
})
}
@@ -554,7 +559,7 @@ where
Some(current_repo_index),
Some(runtime),
phase2_config,
phase2_enabled.then(|| {
(phase2_enabled && config.build_ccr_accumulator).then(|| {
CcrAccumulator::new(
discoveries
.iter()
@@ -562,6 +567,7 @@ where
.collect::<Vec<_>>(),
)
}),
config.persist_vcir,
);
let TreeRunAuditOutput {
@@ -585,9 +591,7 @@ where
Some(&current_repo_index_for_output),
collect_current_repo_objects,
),
ccr_accumulator: phase2_enabled
.then(|| runner.ccr_accumulator_snapshot())
.flatten(),
ccr_accumulator: runner.ccr_accumulator_snapshot(),
})
}
@@ -841,6 +845,7 @@ pub fn run_tree_from_tal_and_ta_der_serial(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -892,6 +897,7 @@ pub fn run_tree_from_tal_bytes_serial_audit(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -960,6 +966,7 @@ pub fn run_tree_from_tal_bytes_serial_audit_with_timing(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -1023,6 +1030,7 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -1087,6 +1095,7 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit_with_timing(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -1158,6 +1167,7 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -1215,6 +1225,7 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -1289,6 +1300,7 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit_with_timing(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let root = root_handle_from_trust_anchor(
@@ -1346,6 +1358,7 @@ fn build_payload_replay_runner<'a>(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
}
}
@@ -1378,6 +1391,7 @@ fn build_payload_delta_replay_runner<'a>(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
}
}
@@ -1410,6 +1424,7 @@ fn build_payload_delta_replay_current_store_runner<'a>(
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
}
}
@@ -1888,6 +1903,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
)
.unwrap_err();
@@ -1920,6 +1938,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
)
.expect("run replay root-only audit");
@@ -1962,6 +1983,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
)
.expect("run replay root-only audit");
@@ -2005,6 +2029,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
&timing,
)
@@ -2058,6 +2085,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
)
.unwrap_err();
@@ -2087,6 +2117,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
)
.unwrap_err();
@@ -2136,6 +2169,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
)
.expect("run delta replay root-only audit");
@@ -2194,6 +2230,9 @@ mod replay_api_tests {
&TreeRunConfig {
max_depth: Some(0),
max_instances: Some(1),
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
},
&timing,
)

25
src/validation/tree.rs
View File

@@ -12,6 +12,12 @@ pub struct TreeRunConfig {
pub max_depth: Option<usize>,
/// Max number of CA instances to process.
pub max_instances: Option<usize>,
/// Drop per-object audit payload when the caller only needs validation outputs.
pub compact_audit: bool,
/// Persist per-CA VCIR state for future reuse/delta runs.
pub persist_vcir: bool,
/// Build online CCR manifest projections during phase2 validation.
pub build_ccr_accumulator: bool,
}
impl Default for TreeRunConfig {
@@ -19,6 +25,9 @@ impl Default for TreeRunConfig {
Self {
max_depth: None,
max_instances: None,
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
}
}
}
@@ -192,16 +201,20 @@ pub fn run_tree_serial_audit_multi_root(
};
instances_processed += 1;
warnings.extend(res.warnings.clone());
warnings.extend(res.objects.warnings.clone());
vrps.extend(res.objects.vrps.clone());
aspas.extend(res.objects.aspas.clone());
router_keys.extend(res.objects.router_keys.clone());
warnings.extend(res.warnings);
warnings.extend(res.objects.warnings);
vrps.extend(res.objects.vrps);
aspas.extend(res.objects.aspas);
router_keys.extend(res.objects.router_keys);
let mut audit = res.audit.clone();
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();
if config.compact_audit {
audit.objects.clear();
audit.warnings.clear();
}
publication_points.push(audit);
let mut children = res.discovered_children;

72
src/validation/tree_parallel.rs
View File

@@ -51,17 +51,25 @@ struct FinishedPublicationPoint {
fn compact_phase2_finished_result(
mut result: PublicationPointRunResult,
compact_audit: bool,
) -> PublicationPointRunResult {
// Phase2 only needs warnings, objects, audit, and traversal metadata after finalize.
// Dropping the snapshot here avoids retaining manifest/files/raw-byte caches until run end.
result.snapshot = None;
if compact_audit {
result.audit.objects.clear();
result.audit.warnings.clear();
result.objects.audit.clear();
result.discovered_children.clear();
}
result
}
fn compact_phase2_finished_result_result(
result: Result<PublicationPointRunResult, String>,
compact_audit: bool,
) -> Result<PublicationPointRunResult, String> {
result.map(compact_phase2_finished_result)
result.map(|result| compact_phase2_finished_result(result, compact_audit))
}
pub fn run_tree_parallel_phase2_audit_multi_root(
@@ -150,11 +158,17 @@ pub fn run_tree_parallel_phase2_audit_multi_root(
&mut inflight_publication_points,
&mut finished,
ready,
config.compact_audit,
);
}
flush_pending_roa_dispatch(runner, &mut pending_roa_dispatch)?;
drain_object_results(runner, &mut inflight_publication_points, &mut finished)?;
drain_object_results(
runner,
&mut inflight_publication_points,
&mut finished,
config.compact_audit,
)?;
let repo_poll_timeout = event_poll_timeout(
&ca_queue,
&ready_queue,
@@ -204,6 +218,7 @@ fn stage_ready_publication_point(
inflight_publication_points: &mut HashMap<u64, InflightPublicationPoint>,
finished: &mut Vec<FinishedPublicationPoint>,
ready: ReadyCaInstance,
compact_audit: bool,
) {
let publication_point_started = Instant::now();
let mut warnings = ready.repo_outcome.warnings.clone();
@@ -229,7 +244,7 @@ fn stage_ready_publication_point(
}
finished.push(FinishedPublicationPoint {
node: ready.node,
result: compact_phase2_finished_result_result(fallback),
result: compact_phase2_finished_result_result(fallback, compact_audit),
});
return;
}
@@ -252,6 +267,7 @@ fn stage_ready_publication_point(
ready.node.handle.effective_ip_resources.as_ref(),
ready.node.handle.effective_as_resources.as_ref(),
runner.validation_time,
runner.persist_vcir,
) {
ParallelObjectsPrepare::Complete(mut objects) => {
objects
@@ -271,6 +287,7 @@ fn stage_ready_publication_point(
objects,
repo_outcome,
finished,
compact_audit,
);
}
ParallelObjectsPrepare::Staged(objects_stage) => {
@@ -299,6 +316,7 @@ fn stage_ready_publication_point(
objects,
repo_outcome,
finished,
compact_audit,
);
}
Err(err) => finished.push(FinishedPublicationPoint {
@@ -367,6 +385,7 @@ fn finalize_ready_objects(
objects: ObjectsOutput,
repo_outcome: RepoSyncRuntimeOutcome,
finished: &mut Vec<FinishedPublicationPoint>,
compact_audit: bool,
) {
let result = runner
.finalize_fresh_publication_point_from_reducer(
@@ -384,7 +403,7 @@ fn finalize_ready_objects(
.map(|out| out.result);
finished.push(FinishedPublicationPoint {
node,
result: compact_phase2_finished_result_result(result),
result: compact_phase2_finished_result_result(result, compact_audit),
});
}
@@ -416,6 +435,7 @@ fn drain_object_results(
runner: &Rpkiv1PublicationPointRunner<'_>,
inflight_publication_points: &mut HashMap<u64, InflightPublicationPoint>,
finished: &mut Vec<FinishedPublicationPoint>,
compact_audit: bool,
) -> Result<(), TreeRunError> {
let Some(pool) = runner.parallel_roa_worker_pool.as_ref() else {
return Ok(());
@@ -479,7 +499,7 @@ fn drain_object_results(
);
finished.push(FinishedPublicationPoint {
node: state.node,
result: compact_phase2_finished_result_result(result),
result: compact_phase2_finished_result_result(result, compact_audit),
});
}
Err(err) => finished.push(FinishedPublicationPoint {
@@ -572,11 +592,11 @@ fn build_tree_output(mut finished: Vec<FinishedPublicationPoint>) -> TreeRunAudi
match item.result {
Ok(result) => {
instances_processed += 1;
warnings.extend(result.warnings.clone());
warnings.extend(result.objects.warnings.clone());
vrps.extend(result.objects.vrps.clone());
aspas.extend(result.objects.aspas.clone());
router_keys.extend(result.objects.router_keys.clone());
warnings.extend(result.warnings);
warnings.extend(result.objects.warnings);
vrps.extend(result.objects.vrps);
aspas.extend(result.objects.aspas);
router_keys.extend(result.objects.router_keys);
let mut audit: PublicationPointAudit = result.audit;
audit.node_id = Some(item.node.id);
@@ -666,15 +686,43 @@ mod tests {
#[test]
fn compact_phase2_finished_result_drops_snapshot() {
let result = compact_phase2_finished_result(sample_result());
let result = compact_phase2_finished_result(sample_result(), false);
assert!(result.snapshot.is_none());
assert_eq!(result.source, PublicationPointSource::Fresh);
assert!(result.discovered_children.is_empty());
}
#[test]
fn compact_phase2_finished_result_can_drop_audit_payload() {
let mut sample = sample_result();
sample.audit.objects.push(crate::audit::ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/a.roa".to_string(),
sha256_hex: "11".repeat(32),
kind: crate::audit::AuditObjectKind::Roa,
result: crate::audit::AuditObjectResult::Ok,
detail: None,
});
sample.audit.warnings.push(crate::audit::AuditWarning {
message: "warning".to_string(),
rfc_refs: Vec::new(),
context: None,
});
sample.objects.audit.push(crate::audit::ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/b.roa".to_string(),
sha256_hex: "22".repeat(32),
kind: crate::audit::AuditObjectKind::Roa,
result: crate::audit::AuditObjectResult::Ok,
detail: None,
});
let result = compact_phase2_finished_result(sample, true);
assert!(result.audit.objects.is_empty());
assert!(result.audit.warnings.is_empty());
assert!(result.objects.audit.is_empty());
}
#[test]
fn compact_phase2_finished_result_result_preserves_err() {
let err = compact_phase2_finished_result_result(Err("boom".to_string()))
let err = compact_phase2_finished_result_result(Err("boom".to_string()), false)
.expect_err("error should be preserved");
assert_eq!(err, "boom");
}

54
src/validation/tree_runner.rs
View File

@@ -44,8 +44,8 @@ use crate::validation::manifest::{
};
use crate::validation::objects::{
AspaAttestation, ParallelRoaWorkerPool, RouterKeyPayload, Vrp,
process_publication_point_for_issuer, process_publication_point_for_issuer_parallel_roa,
process_publication_point_for_issuer_parallel_roa_with_pool,
process_publication_point_for_issuer_parallel_roa_with_options,
process_publication_point_for_issuer_parallel_roa_with_pool_options,
};
use crate::validation::publication_point::PublicationPointSnapshot;
use crate::validation::tree::{
@@ -138,6 +138,11 @@ pub struct Rpkiv1PublicationPointRunner<'a> {
pub parallel_phase2_config: Option<ParallelPhase2Config>,
pub parallel_roa_worker_pool: Option<ParallelRoaWorkerPool>,
pub ccr_accumulator: Option<Mutex<CcrAccumulator>>,
/// When false, skip VCIR persistence and per-output VCIR projection building.
///
/// This is intended for replay/compare-only runs where the caller does not need
/// the resulting DB to be reused by a later delta run.
pub persist_vcir: bool,
}
impl<'a> Rpkiv1PublicationPointRunner<'a> {
@@ -267,17 +272,21 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
let snapshot_pack_ms = snapshot_pack_started.elapsed().as_millis() as u64;
let persist_vcir_started = std::time::Instant::now();
let persist_vcir_timing = persist_vcir_for_fresh_result_with_timing(
self.store,
ca,
&pack,
&objects,
&warnings,
&child_audits,
&discovered_children,
self.validation_time,
)
.map_err(|e| format!("persist VCIR failed: {e}"))?;
let persist_vcir_timing = if self.persist_vcir {
persist_vcir_for_fresh_result_with_timing(
self.store,
ca,
&pack,
&objects,
&warnings,
&child_audits,
&discovered_children,
self.validation_time,
)
.map_err(|e| format!("persist VCIR failed: {e}"))?
} else {
PersistVcirTimingBreakdown::default()
};
let persist_vcir_ms = persist_vcir_started.elapsed().as_millis() as u64;
// local_outputs_cache only exists to build/persist VCIR. Release it before the
@@ -607,7 +616,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
.as_ref()
.map(|t| t.span_phase("objects_processing_total"));
if let Some(phase2_pool) = self.parallel_roa_worker_pool.as_ref() {
process_publication_point_for_issuer_parallel_roa_with_pool(
process_publication_point_for_issuer_parallel_roa_with_pool_options(
&fresh_point,
self.policy,
&ca.ca_certificate_der,
@@ -617,9 +626,10 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
self.validation_time,
self.timing.as_ref(),
phase2_pool,
self.persist_vcir,
)
} else if let Some(phase2_config) = self.parallel_phase2_config.as_ref() {
process_publication_point_for_issuer_parallel_roa(
process_publication_point_for_issuer_parallel_roa_with_options(
&fresh_point,
self.policy,
&ca.ca_certificate_der,
@@ -629,9 +639,10 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
self.validation_time,
self.timing.as_ref(),
phase2_config,
self.persist_vcir,
)
} else {
process_publication_point_for_issuer(
crate::validation::objects::process_publication_point_for_issuer_with_options(
&fresh_point,
self.policy,
&ca.ca_certificate_der,
@@ -640,6 +651,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
ca.effective_as_resources.as_ref(),
self.validation_time,
self.timing.as_ref(),
self.persist_vcir,
)
}
};
@@ -3283,6 +3295,7 @@ mod tests {
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: Some(Mutex::new(CcrAccumulator::new(Vec::new()))),
persist_vcir: true,
}
}
@@ -4159,6 +4172,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let ca = CaInstanceHandle {
depth: 0,
@@ -4711,6 +4725,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
// For this fixture-driven smoke, we provide the correct issuer CA certificate (the CA for
@@ -4873,6 +4888,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let first = runner.run_publication_point(&handle).expect("first run ok");
@@ -4985,6 +5001,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let first = runner.run_publication_point(&handle).expect("first run ok");
@@ -5100,6 +5117,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let first = runner.run_publication_point(&handle).expect("first run ok");
@@ -5187,6 +5205,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let first = ok_runner
.run_publication_point(&handle)
@@ -5217,6 +5236,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let second = bad_runner
.run_publication_point(&handle)
@@ -6551,6 +6571,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let first = runner_rrdp
.run_publication_point(&handle)
@@ -6584,6 +6605,7 @@ authorityKeyIdentifier = keyid:always
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
};
let third = runner_rsync
.run_publication_point(&handle)