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

20260617 发布点级验证缓存复用

Browse Source
This commit is contained in:
yuyr
2026-06-17 23:49:53 +08:00
parent 68006467f7
commit ad2a25aede
18 changed files with 1841 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/validation/manifest.rs
View File

@@ -14,6 +14,7 @@ use x509_parser::prelude::FromDer;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PublicationPointSource {
Fresh,
PublicationPointCache,
VcirCurrentInstance,
FailedFetchNoCache,
}

2
src/validation/run.rs
View File

@@ -80,6 +80,8 @@ pub fn run_publication_point_once(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let result = runner

52
src/validation/run_tree_from_tal.rs
View File

@@ -136,6 +136,8 @@ fn make_live_runner<'a>(
ccr_accumulator: Option<CcrAccumulator>,
persist_vcir: bool,
enable_roa_validation_cache: bool,
publication_point_cache_observe_only: bool,
enable_publication_point_validation_cache: bool,
) -> Rpkiv1PublicationPointRunner<'a> {
let parallel_roa_worker_pool = parallel_phase2_config
.as_ref()
@@ -161,6 +163,8 @@ fn make_live_runner<'a>(
ccr_accumulator: ccr_accumulator.map(Mutex::new),
persist_vcir,
enable_roa_validation_cache,
publication_point_cache_observe_only,
enable_publication_point_validation_cache,
}
}
@@ -557,6 +561,8 @@ pub fn run_tree_from_tal_url_serial(
None,
config.persist_vcir,
config.enable_roa_validation_cache,
config.publication_point_cache_observe_only,
config.enable_publication_point_validation_cache,
);
let root = root_handle_from_trust_anchor(
@@ -597,6 +603,8 @@ pub fn run_tree_from_tal_url_serial_audit(
None,
config.persist_vcir,
config.enable_roa_validation_cache,
config.publication_point_cache_observe_only,
config.enable_publication_point_validation_cache,
);
let root = root_handle_from_trust_anchor(
@@ -659,6 +667,8 @@ pub fn run_tree_from_tal_url_serial_audit_with_timing(
None,
config.persist_vcir,
config.enable_roa_validation_cache,
config.publication_point_cache_observe_only,
config.enable_publication_point_validation_cache,
);
let root = root_handle_from_trust_anchor(
@@ -747,6 +757,8 @@ where
.then(|| CcrAccumulator::new(vec![discovery.trust_anchor.clone()])),
config.persist_vcir,
config.enable_roa_validation_cache,
config.publication_point_cache_observe_only,
config.enable_publication_point_validation_cache,
);
let root = root_handle_from_trust_anchor(
@@ -873,6 +885,8 @@ where
}),
config.persist_vcir,
config.enable_roa_validation_cache,
config.publication_point_cache_observe_only,
config.enable_publication_point_validation_cache,
);
let TreeRunAuditOutput {
@@ -1299,6 +1313,8 @@ pub fn run_tree_from_tal_and_ta_der_serial(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1352,6 +1368,8 @@ pub fn run_tree_from_tal_bytes_serial_audit(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1427,6 +1445,8 @@ pub fn run_tree_from_tal_bytes_serial_audit_with_timing(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1501,6 +1521,8 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1576,6 +1598,8 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit_with_timing(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1658,6 +1682,8 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1721,6 +1747,8 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1806,6 +1834,8 @@ pub fn run_tree_from_tal_and_ta_der_payload_replay_serial_audit_with_timing(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: config.enable_roa_validation_cache,
publication_point_cache_observe_only: config.publication_point_cache_observe_only,
enable_publication_point_validation_cache: config.enable_publication_point_validation_cache,
};
let root = root_handle_from_trust_anchor(
@@ -1871,6 +1901,8 @@ fn build_payload_replay_runner<'a>(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
}
}
@@ -1906,6 +1938,8 @@ fn build_payload_delta_replay_runner<'a>(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
}
}
@@ -1941,6 +1975,8 @@ fn build_payload_delta_replay_current_store_runner<'a>(
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
}
}
@@ -2504,6 +2540,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
@@ -2541,6 +2579,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
@@ -2588,6 +2628,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
@@ -2636,6 +2678,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
&timing,
@@ -2694,6 +2738,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
@@ -2728,6 +2774,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
@@ -2782,6 +2830,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
@@ -2845,6 +2895,8 @@ mod replay_api_tests {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
&timing,

9
src/validation/tree.rs
View File

@@ -22,6 +22,10 @@ pub struct TreeRunConfig {
pub build_ccr_accumulator: bool,
/// Reuse accepted ROA validation outputs from previous VCIR when explicitly enabled.
pub enable_roa_validation_cache: bool,
/// Evaluate publication-point cache eligibility without changing validation results.
pub publication_point_cache_observe_only: bool,
/// Reuse complete publication-point validation projections when explicitly enabled.
pub enable_publication_point_validation_cache: bool,
/// Prefetch transport requests from the previous run before normal tree traversal.
pub enable_transport_request_prefetch: bool,
}
@@ -35,6 +39,8 @@ impl Default for TreeRunConfig {
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
}
}
@@ -154,7 +160,8 @@ pub(crate) fn submit_publication_point_cir_input(
.submit_audit_entries(crate::cir::CirInputSection::Fresh, entries)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
}
PublicationPointSource::VcirCurrentInstance => {
PublicationPointSource::VcirCurrentInstance
| PublicationPointSource::PublicationPointCache => {
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Fresh, cir_fresh_objects)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;

27
src/validation/tree_parallel.rs
View File

@@ -909,6 +909,33 @@ fn stage_ready_publication_point(
};
let mut warnings = ready.repo_outcome.warnings.clone();
let repo_outcome = ready.repo_outcome.clone();
if let Some(result) = runner.observe_or_reuse_publication_point_cache(
&ready.node.handle,
repo_outcome.repo_sync_source.as_deref(),
repo_outcome.repo_sync_phase.as_deref(),
repo_outcome.repo_sync_duration_ms,
repo_outcome.repo_sync_err.as_deref(),
&warnings,
) {
metrics.complete_count = 1;
metrics.discovered_children = result.discovered_children.len();
let child_enqueue_started = Instant::now();
enqueue_discovered_children(
runner,
next_id,
ca_queue,
&ready.node,
result.discovered_children.clone(),
);
metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started);
finished.push(FinishedPublicationPoint {
node: FinishedPublicationPointNode::from_queued(ready.node),
result: compact_phase2_finished_result(result, compact_audit),
});
metrics.total_ms = elapsed_ms(publication_point_started);
return metrics;
}
let stage_fresh_started = Instant::now();
let stage = runner.stage_fresh_publication_point_after_repo_ready(
&ready.node.handle,

597
src/validation/tree_runner.rs
View File

@@ -25,10 +25,11 @@ use crate::replay::archive::ReplayArchiveIndex;
use crate::replay::delta_archive::ReplayDeltaArchiveIndex;
use crate::report::{RfcRef, Warning};
use crate::storage::{
PackFile, PackTime, RawByHashEntry, RocksStore, ValidatedCaInstanceResult, VcirArtifactKind,
VcirArtifactRole, VcirArtifactValidationStatus, VcirAuditSummary, VcirCcrManifestProjection,
VcirChildEntry, VcirInstanceGate, VcirLocalOutput, VcirLocalOutputPayload, VcirOutputType,
VcirRelatedArtifact, VcirReplaceTimingBreakdown, VcirSourceObjectType, VcirSummary,
PackFile, PackTime, PublicationPointCacheProjection, RawByHashEntry, RocksStore,
ValidatedCaInstanceResult, VcirArtifactKind, VcirArtifactRole, VcirArtifactValidationStatus,
VcirAuditSummary, VcirCcrManifestProjection, VcirChildEntry, VcirInstanceGate, VcirLocalOutput,
VcirLocalOutputPayload, VcirOutputType, VcirRelatedArtifact, VcirReplaceTimingBreakdown,
VcirSourceObjectType, VcirSummary,
};
use crate::sync::repo::{
sync_publication_point, sync_publication_point_replay, sync_publication_point_replay_delta,
@@ -154,6 +155,8 @@ pub struct Rpkiv1PublicationPointRunner<'a> {
/// the resulting DB to be reused by a later delta run.
pub persist_vcir: bool,
pub enable_roa_validation_cache: bool,
pub publication_point_cache_observe_only: bool,
pub enable_publication_point_validation_cache: bool,
}
impl<'a> Rpkiv1PublicationPointRunner<'a> {
@@ -214,6 +217,284 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
}
}
pub(crate) fn observe_or_reuse_publication_point_cache(
&self,
ca: &CaInstanceHandle,
repo_sync_source: Option<&str>,
repo_sync_phase: Option<&str>,
repo_sync_duration_ms: u64,
repo_sync_err: Option<&str>,
warnings: &[Warning],
) -> Option<PublicationPointRunResult> {
if !self.publication_point_cache_observe_only
&& !self.enable_publication_point_validation_cache
{
return None;
}
let lookup_started = std::time::Instant::now();
if let Some(timing) = self.timing.as_ref() {
timing.record_count("publication_point_cache_lookup_total", 1);
}
let projection = match self
.store
.get_publication_point_cache_projection(&ca.manifest_rsync_uri)
{
Ok(Some(projection)) => projection,
Ok(None) => {
self.record_publication_point_cache_miss("missing_projection");
return None;
}
Err(e) => {
self.record_publication_point_cache_miss("projection_load_error");
crate::progress_log::emit(
"publication_point_cache_lookup_error",
serde_json::json!({
"manifest_rsync_uri": ca.manifest_rsync_uri,
"error": e.to_string(),
}),
);
return None;
}
};
let current_identity = match self.current_publication_point_cache_identity(ca) {
Ok(identity) => identity,
Err(reason) => {
self.record_publication_point_cache_miss(reason.as_str());
return None;
}
};
if projection.ca_cert_uri != ca.ca_certificate_rsync_uri {
self.record_publication_point_cache_miss("ca_uri_mismatch");
return None;
}
if projection.ca_cert_sha256 != current_identity.ca_cert_sha256 {
self.record_publication_point_cache_miss("ca_hash_mismatch");
return None;
}
if projection.manifest_sha256 != current_identity.manifest_sha256 {
self.record_publication_point_cache_miss("manifest_hash_mismatch");
return None;
}
if projection.tal_id != ca.tal_id {
self.record_publication_point_cache_miss("tal_mismatch");
return None;
}
if projection.ta_context_digest != current_identity.ta_context_digest {
self.record_publication_point_cache_miss("ta_context_mismatch");
return None;
}
if projection.parent_context_digest != current_identity.parent_context_digest {
self.record_publication_point_cache_miss("parent_context_mismatch");
return None;
}
if projection.validation_policy_fingerprint != current_identity.policy_fingerprint {
self.record_publication_point_cache_miss("policy_mismatch");
return None;
}
let instance_not_before =
match parse_snapshot_time_value(&projection.instance_effective_not_before) {
Ok(value) => value,
Err(_) => {
self.record_publication_point_cache_miss("instance_not_before_invalid");
return None;
}
};
let instance_until = match parse_snapshot_time_value(&projection.instance_effective_until) {
Ok(value) => value,
Err(_) => {
self.record_publication_point_cache_miss("instance_until_invalid");
return None;
}
};
if self.validation_time < instance_not_before || self.validation_time >= instance_until {
self.record_publication_point_cache_miss("instance_time_gate_miss");
return None;
}
if let Err(reason) =
publication_point_cache_projection_items_valid(&projection, self.validation_time)
{
self.record_publication_point_cache_miss(reason);
return None;
}
if let Some(timing) = self.timing.as_ref() {
timing.record_count("publication_point_cache_theoretical_hits", 1);
timing.record_phase_nanos(
"publication_point_cache_lookup_total",
lookup_started
.elapsed()
.as_nanos()
.min(u128::from(u64::MAX)) as u64,
);
}
if self.publication_point_cache_observe_only {
return None;
}
match self.build_publication_point_cache_result(
ca,
projection,
repo_sync_source,
repo_sync_phase,
repo_sync_duration_ms,
repo_sync_err,
warnings,
) {
Ok(result) => {
if let Some(timing) = self.timing.as_ref() {
timing.record_count("publication_point_cache_reuse_hits", 1);
}
Some(result)
}
Err(e) => {
self.record_publication_point_cache_miss("reuse_build_error");
crate::progress_log::emit(
"publication_point_cache_reuse_error",
serde_json::json!({
"manifest_rsync_uri": ca.manifest_rsync_uri,
"error": e,
}),
);
None
}
}
}
fn record_publication_point_cache_miss(&self, reason: &str) {
if let Some(timing) = self.timing.as_ref() {
timing.record_count("publication_point_cache_miss_total", 1);
match reason {
"missing_projection" => {
timing.record_count("publication_point_cache_miss_missing_projection", 1)
}
"current_manifest_missing" => {
timing.record_count("publication_point_cache_miss_current_manifest_missing", 1)
}
"ca_hash_mismatch" => {
timing.record_count("publication_point_cache_miss_ca_hash_mismatch", 1)
}
"manifest_hash_mismatch" => {
timing.record_count("publication_point_cache_miss_manifest_hash_mismatch", 1)
}
"parent_context_mismatch" => {
timing.record_count("publication_point_cache_miss_parent_context_mismatch", 1)
}
"policy_mismatch" => {
timing.record_count("publication_point_cache_miss_policy_mismatch", 1)
}
"instance_time_gate_miss" => {
timing.record_count("publication_point_cache_miss_instance_time_gate", 1)
}
"output_time_gate_miss" => {
timing.record_count("publication_point_cache_miss_output_time_gate", 1)
}
"child_time_gate_miss" => {
timing.record_count("publication_point_cache_miss_child_time_gate", 1)
}
_ => timing.record_count("publication_point_cache_miss_other", 1),
}
}
}
fn current_publication_point_cache_identity(
&self,
ca: &CaInstanceHandle,
) -> Result<PublicationPointCacheIdentity, String> {
let ca_cert_sha256 = match ca.ca_certificate_rsync_uri.as_deref() {
Some(uri) => self
.current_hash_for_uri(uri)
.unwrap_or_else(|| sha256_digest_32(&ca.ca_certificate_der)),
None => sha256_digest_32(&ca.ca_certificate_der),
};
let manifest_sha256 = self
.current_hash_for_uri(&ca.manifest_rsync_uri)
.ok_or_else(|| "current_manifest_missing".to_string())?;
Ok(PublicationPointCacheIdentity {
ca_cert_sha256,
manifest_sha256,
ta_context_digest: ta_context_digest_for_ca(ca),
parent_context_digest: parent_context_digest_for_ca(ca),
policy_fingerprint: publication_point_cache_policy_fingerprint(self.policy),
})
}
fn current_hash_for_uri(&self, uri: &str) -> Option<[u8; 32]> {
if let Some(index) = self.current_repo_index.as_ref() {
if let Ok(index) = index.lock() {
if let Some(entry) = index.get_by_uri(uri) {
return Some(entry.current_hash);
}
}
}
self.store
.load_current_object_with_hash_by_uri(uri)
.ok()
.flatten()
.map(|entry| entry.current_hash)
}
fn build_publication_point_cache_result(
&self,
ca: &CaInstanceHandle,
projection: PublicationPointCacheProjection,
repo_sync_source: Option<&str>,
repo_sync_phase: Option<&str>,
repo_sync_duration_ms: u64,
repo_sync_err: Option<&str>,
warnings: &[Warning],
) -> Result<PublicationPointRunResult, String> {
let mut warnings = warnings.to_vec();
let output_reuse_count = projection.outputs.len() as u64;
let child_reuse_count = projection.children.len() as u64;
let mut vcir = projection.to_vcir_for_reuse(self.validation_time);
vcir.parent_manifest_rsync_uri = ca.parent_manifest_rsync_uri.clone();
let mut objects =
build_objects_output_from_vcir(&vcir, self.validation_time, &mut warnings);
let (discovered_children, child_audits) = restore_children_from_publication_point_cache(
self.store,
ca,
&projection,
self.validation_time,
&mut warnings,
);
let ccr_projection = projection.ccr_manifest_projection.clone();
self.append_ccr_manifest_projection(&ccr_projection)?;
let audit = build_publication_point_audit_from_vcir(
ca,
PublicationPointSource::PublicationPointCache,
repo_sync_source,
repo_sync_phase,
Some(repo_sync_duration_ms),
repo_sync_err,
Some(&vcir),
None,
&warnings,
&objects,
&child_audits,
&[],
);
let cir_cached_objects = audit.objects.clone();
objects.local_outputs_cache.clear();
if let Some(timing) = self.timing.as_ref() {
timing.record_count("publication_point_cache_outputs_reused", output_reuse_count);
timing.record_count("publication_point_cache_children_reused", child_reuse_count);
}
Ok(PublicationPointRunResult {
source: PublicationPointSource::PublicationPointCache,
snapshot: None,
warnings,
objects,
audit,
cir_fresh_objects: Vec::new(),
cir_cached_objects,
discovered_children,
})
}
pub(crate) fn ccr_accumulator_snapshot(&self) -> Option<CcrAccumulator> {
self.ccr_accumulator
.as_ref()
@@ -242,6 +523,11 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
"invalid reuse projection source: fresh does not belong to failed-fetch reuse"
.to_string(),
),
PublicationPointSource::PublicationPointCache => self.append_ccr_manifest_projection(
projection.ccr_manifest_projection.as_ref().ok_or_else(|| {
"publication-point cache reuse is missing CCR manifest projection".to_string()
})?,
),
PublicationPointSource::VcirCurrentInstance => self.append_ccr_manifest_projection(
projection.ccr_manifest_projection.as_ref().ok_or_else(|| {
"vcir current-instance reuse is missing CCR manifest projection".to_string()
@@ -386,6 +672,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
let persist_vcir_timing = if self.persist_vcir {
persist_vcir_for_fresh_result_with_timing(
self.store,
self.policy,
ca,
&pack,
&mut objects,
@@ -393,6 +680,8 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
&child_audits,
&discovered_children,
self.validation_time,
self.publication_point_cache_observe_only
|| self.enable_publication_point_validation_cache,
)
.map_err(|e| format!("persist VCIR failed: {e}"))?
} else {
@@ -716,6 +1005,36 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
}),
);
if let Some(result) = self.observe_or_reuse_publication_point_cache(
ca,
repo_sync_source.as_deref(),
repo_sync_phase.as_deref(),
repo_sync_duration_ms,
repo_sync_err.as_deref(),
&warnings,
) {
let total_duration_ms = publication_point_started.elapsed().as_millis() as u64;
crate::progress_log::emit(
"publication_point_finish",
serde_json::json!({
"manifest_rsync_uri": ca.manifest_rsync_uri,
"publication_point_rsync_uri": ca.publication_point_rsync_uri,
"source": source_label(result.source),
"repo_sync_source": repo_sync_source,
"repo_sync_phase": repo_sync_phase,
"repo_sync_duration_ms": repo_sync_duration_ms,
"total_duration_ms": total_duration_ms,
"post_repo_duration_ms": total_duration_ms.saturating_sub(repo_sync_duration_ms),
"warning_count": result.warnings.len(),
"vrp_count": result.objects.vrps.len(),
"vap_count": result.objects.aspas.len(),
"router_key_count": result.objects.router_keys.len(),
"child_count": result.discovered_children.len(),
}),
);
return Ok(result);
}
let fresh_stage = self.stage_fresh_publication_point_after_repo_ready(
ca,
repo_sync_ok,
@@ -1094,6 +1413,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
);
}
PublicationPointSource::Fresh => {}
PublicationPointSource::PublicationPointCache => {}
PublicationPointSource::VcirCurrentInstance => {}
}
if (total_duration_ms as f64) / 1000.0
@@ -1141,6 +1461,119 @@ enum CachedIssuerCrl {
Pending(Vec<u8>),
Ok(VerifiedIssuerCrl),
}
struct PublicationPointCacheIdentity {
ca_cert_sha256: [u8; 32],
manifest_sha256: [u8; 32],
ta_context_digest: [u8; 32],
parent_context_digest: [u8; 32],
policy_fingerprint: [u8; 32],
}
fn sha256_digest_32(bytes: impl AsRef<[u8]>) -> [u8; 32] {
let digest = sha2::Sha256::digest(bytes.as_ref());
let mut out = [0u8; 32];
out.copy_from_slice(&digest);
out
}
fn hash_serialized_parts(parts: &[(&str, Vec<u8>)]) -> [u8; 32] {
let mut hasher = sha2::Sha256::new();
for (label, value) in parts {
hasher.update((label.len() as u64).to_be_bytes());
hasher.update(label.as_bytes());
hasher.update((value.len() as u64).to_be_bytes());
hasher.update(value);
}
let digest = hasher.finalize();
let mut out = [0u8; 32];
out.copy_from_slice(&digest);
out
}
fn cbor_or_debug_bytes<T: serde::Serialize + std::fmt::Debug>(value: &T) -> Vec<u8> {
serde_cbor::to_vec(value).unwrap_or_else(|_| format!("{value:?}").into_bytes())
}
fn ta_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
hash_serialized_parts(&[
("version", b"publication-point-cache-ta-v1".to_vec()),
("tal_id", ca.tal_id.as_bytes().to_vec()),
])
}
fn parent_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
hash_serialized_parts(&[
(
"version",
b"publication-point-cache-parent-context-v1".to_vec(),
),
("tal_id", ca.tal_id.as_bytes().to_vec()),
(
"parent_manifest",
ca.parent_manifest_rsync_uri
.as_deref()
.unwrap_or("")
.as_bytes()
.to_vec(),
),
(
"effective_ip",
cbor_or_debug_bytes(&ca.effective_ip_resources),
),
(
"effective_as",
cbor_or_debug_bytes(&ca.effective_as_resources),
),
])
}
fn publication_point_cache_policy_fingerprint(policy: &Policy) -> [u8; 32] {
hash_serialized_parts(&[
("version", b"publication-point-cache-policy-v1".to_vec()),
("policy", cbor_or_debug_bytes(policy)),
])
}
fn publication_point_cache_projection_items_valid(
projection: &PublicationPointCacheProjection,
validation_time: time::OffsetDateTime,
) -> Result<(), &'static str> {
for output in &projection.outputs {
if !pack_time_window_contains(
&output.item_effective_not_before,
&output.item_effective_until,
validation_time,
) {
return Err("output_time_gate_miss");
}
}
for child in &projection.children {
if !pack_time_window_contains(
&child.child_effective_not_before,
&child.child_effective_until,
validation_time,
) {
return Err("child_time_gate_miss");
}
}
Ok(())
}
fn pack_time_window_contains(
not_before: &PackTime,
until: &PackTime,
validation_time: time::OffsetDateTime,
) -> bool {
let Ok(not_before) = parse_snapshot_time_value(not_before) else {
return false;
};
let Ok(until) = parse_snapshot_time_value(until) else {
return false;
};
validation_time >= not_before && validation_time < until
}
#[derive(Clone, Debug)]
struct VcirReuseProjection {
source: PublicationPointSource,
@@ -1759,6 +2192,7 @@ fn kind_from_rsync_uri(uri: &str) -> AuditObjectKind {
fn source_label(source: PublicationPointSource) -> String {
match source {
PublicationPointSource::Fresh => "fresh".to_string(),
PublicationPointSource::PublicationPointCache => "publication_point_cache".to_string(),
PublicationPointSource::VcirCurrentInstance => "vcir_current_instance".to_string(),
PublicationPointSource::FailedFetchNoCache => "failed_fetch_no_cache".to_string(),
}
@@ -1797,6 +2231,7 @@ fn repo_sync_failure_phase_label(
fn terminal_state_label(source: PublicationPointSource) -> &'static str {
match source {
PublicationPointSource::Fresh => "fresh",
PublicationPointSource::PublicationPointCache => "publication_point_cache",
PublicationPointSource::VcirCurrentInstance => "fallback_current_instance",
PublicationPointSource::FailedFetchNoCache => "failed_no_cache",
}
@@ -2694,8 +3129,128 @@ fn restore_children_from_vcir(
(children, audits)
}
fn restore_children_from_publication_point_cache(
store: &RocksStore,
ca: &CaInstanceHandle,
projection: &PublicationPointCacheProjection,
validation_time: time::OffsetDateTime,
warnings: &mut Vec<Warning>,
) -> (Vec<DiscoveredChildCaInstance>, Vec<ObjectAuditEntry>) {
let mut children = Vec::new();
let mut audits = Vec::new();
for child in &projection.children {
let effective_not_before =
match parse_snapshot_time_value(&child.child_effective_not_before) {
Ok(value) => value,
Err(e) => {
warnings.push(
Warning::new(format!(
"publication-point cache child has invalid effective notBefore: {e}"
))
.with_context(&child.child_cert_rsync_uri),
);
continue;
}
};
let effective_until = match parse_snapshot_time_value(&child.child_effective_until) {
Ok(value) => value,
Err(e) => {
warnings.push(
Warning::new(format!(
"publication-point cache child has invalid effective until: {e}"
))
.with_context(&child.child_cert_rsync_uri),
);
continue;
}
};
if validation_time < effective_not_before || validation_time > effective_until {
audits.push(ObjectAuditEntry {
rsync_uri: child.child_cert_rsync_uri.clone(),
sha256_hex: child.child_cert_hash.clone(),
kind: AuditObjectKind::Certificate,
result: AuditObjectResult::Skipped,
detail: Some("skipped: publication-point cache child expired".to_string()),
});
continue;
}
match store.get_blob_bytes(&child.child_cert_hash) {
Ok(Some(bytes)) => {
children.push(DiscoveredChildCaInstance {
handle: CaInstanceHandle {
depth: 0,
tal_id: ca.tal_id.clone(),
parent_manifest_rsync_uri: Some(ca.manifest_rsync_uri.clone()),
ca_certificate_der: bytes,
ca_certificate_rsync_uri: Some(child.child_cert_rsync_uri.clone()),
effective_ip_resources: child.child_effective_ip_resources.clone(),
effective_as_resources: child.child_effective_as_resources.clone(),
rsync_base_uri: child.child_rsync_base_uri.clone(),
manifest_rsync_uri: child.child_manifest_rsync_uri.clone(),
publication_point_rsync_uri: child
.child_publication_point_rsync_uri
.clone(),
rrdp_notification_uri: child.child_rrdp_notification_uri.clone(),
},
discovered_from: crate::audit::DiscoveredFrom {
parent_manifest_rsync_uri: ca.manifest_rsync_uri.clone(),
child_ca_certificate_rsync_uri: child.child_cert_rsync_uri.clone(),
child_ca_certificate_sha256_hex: child.child_cert_hash.clone(),
},
});
audits.push(ObjectAuditEntry {
rsync_uri: child.child_cert_rsync_uri.clone(),
sha256_hex: child.child_cert_hash.clone(),
kind: AuditObjectKind::Certificate,
result: AuditObjectResult::Ok,
detail: Some(
"restored child CA instance from publication-point cache".to_string(),
),
});
}
Ok(None) => {
warnings.push(
Warning::new(
"child certificate bytes missing for publication-point cache restoration",
)
.with_context(&child.child_cert_rsync_uri),
);
audits.push(ObjectAuditEntry {
rsync_uri: child.child_cert_rsync_uri.clone(),
sha256_hex: child.child_cert_hash.clone(),
kind: AuditObjectKind::Certificate,
result: AuditObjectResult::Error,
detail: Some(
"child certificate bytes missing for publication-point cache restoration"
.to_string(),
),
});
}
Err(e) => {
warnings.push(
Warning::new(format!(
"child certificate bytes load failed for publication-point cache restoration: {e}"
))
.with_context(&child.child_cert_rsync_uri),
);
audits.push(ObjectAuditEntry {
rsync_uri: child.child_cert_rsync_uri.clone(),
sha256_hex: child.child_cert_hash.clone(),
kind: AuditObjectKind::Certificate,
result: AuditObjectResult::Error,
detail: Some(format!(
"child certificate bytes load failed for publication-point cache restoration: {e}"
)),
});
}
}
}
(children, audits)
}
fn persist_vcir_for_fresh_result_with_timing(
store: &RocksStore,
policy: &Policy,
ca: &CaInstanceHandle,
pack: &PublicationPointSnapshot,
objects: &mut crate::validation::objects::ObjectsOutput,
@@ -2703,6 +3258,7 @@ fn persist_vcir_for_fresh_result_with_timing(
child_audits: &[ObjectAuditEntry],
discovered_children: &[DiscoveredChildCaInstance],
validation_time: time::OffsetDateTime,
write_publication_point_cache_projection: bool,
) -> Result<PersistVcirTimingBreakdown, String> {
let mut timing = PersistVcirTimingBreakdown::default();
@@ -2729,8 +3285,18 @@ fn persist_vcir_for_fresh_result_with_timing(
timing.build_vcir = build_vcir_timing;
let replace_vcir_started = std::time::Instant::now();
let publication_point_cache_projection = if write_publication_point_cache_projection {
Some(build_publication_point_cache_projection_from_fresh(
policy, ca, pack, &vcir,
)?)
} else {
None
};
let replace_timing = store
.replace_vcir_and_manifest_replay_meta(&vcir)
.replace_vcir_manifest_replay_meta_and_publication_point_cache_projection(
&vcir,
publication_point_cache_projection.as_ref(),
)
.map_err(|e| format!("store VCIR and manifest replay meta failed: {e}"))?;
timing.replace_vcir_ms = replace_vcir_started.elapsed().as_millis() as u64;
timing.replace_vcir = replace_timing;
@@ -2738,6 +3304,27 @@ fn persist_vcir_for_fresh_result_with_timing(
Ok(timing)
}
fn build_publication_point_cache_projection_from_fresh(
policy: &Policy,
ca: &CaInstanceHandle,
pack: &PublicationPointSnapshot,
vcir: &ValidatedCaInstanceResult,
) -> Result<PublicationPointCacheProjection, String> {
let ca_cert_sha256 = sha256_digest_32(&ca.ca_certificate_der);
let manifest_sha256 = sha256_digest_32(&pack.manifest_bytes);
PublicationPointCacheProjection::from_vcir_with_context(
vcir,
pack.publication_point_rsync_uri.clone(),
ca.ca_certificate_rsync_uri.clone(),
ca_cert_sha256,
manifest_sha256,
ta_context_digest_for_ca(ca),
parent_context_digest_for_ca(ca),
publication_point_cache_policy_fingerprint(policy),
)
.map_err(|e| e.to_string())
}
fn build_vcir_from_fresh_result_with_timing(
ca: &CaInstanceHandle,
pack: &PublicationPointSnapshot,

487
src/validation/tree_runner/tests.rs
View File

@@ -4,10 +4,11 @@ use crate::data_model::roa::RoaAfi;
use crate::fetch::rsync::LocalDirRsyncFetcher;
use crate::fetch::rsync::{RsyncFetchError, RsyncFetcher};
use crate::storage::{
PackFile, PackTime, RawByHashEntry, RocksStore, ValidatedCaInstanceResult,
ValidatedManifestMeta, VcirArtifactKind, VcirArtifactRole, VcirArtifactValidationStatus,
VcirAuditSummary, VcirChildEntry, VcirInstanceGate, VcirLocalOutput, VcirLocalOutputPayload,
VcirOutputType, VcirRelatedArtifact, VcirSourceObjectType, VcirSummary,
PackFile, PackTime, PublicationPointCacheProjection, RawByHashEntry, RepositoryViewEntry,
RepositoryViewState, RocksStore, ValidatedCaInstanceResult, ValidatedManifestMeta,
VcirArtifactKind, VcirArtifactRole, VcirArtifactValidationStatus, VcirAuditSummary,
VcirChildEntry, VcirInstanceGate, VcirLocalOutput, VcirLocalOutputPayload, VcirOutputType,
VcirRelatedArtifact, VcirSourceObjectType, VcirSummary,
};
use crate::sync::rrdp::Fetcher;
use crate::validation::publication_point::PublicationPointSnapshot;
@@ -69,6 +70,8 @@ fn sample_runner_with_ccr_accumulator<'a>(
ccr_accumulator: Some(Mutex::new(CcrAccumulator::new(Vec::new()))),
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
}
}
@@ -960,6 +963,8 @@ fn finalize_fresh_publication_point_releases_local_outputs_cache_after_persist()
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let ca = CaInstanceHandle {
depth: 0,
@@ -1062,6 +1067,7 @@ fn persist_vcir_for_fresh_result_stores_vcir_and_replay_meta_for_real_snapshot()
let mut objects = objects;
persist_vcir_for_fresh_result_with_timing(
&store,
&Policy::default(),
&ca,
&pack,
&mut objects,
@@ -1069,6 +1075,7 @@ fn persist_vcir_for_fresh_result_stores_vcir_and_replay_meta_for_real_snapshot()
&[],
&[],
validation_time,
false,
)
.map(|_timing| ())
.expect("persist vcir for fresh result");
@@ -1522,6 +1529,8 @@ fn runner_offline_rsync_fixture_produces_pack_and_warnings() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
// For this fixture-driven smoke, we provide the correct issuer CA certificate (the CA for
@@ -1653,6 +1662,8 @@ fn runner_roa_validation_cache_reuses_vcir_outputs_on_second_fixture_run() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let first = first_runner
.run_publication_point(&handle)
@@ -1681,6 +1692,8 @@ fn runner_roa_validation_cache_reuses_vcir_outputs_on_second_fixture_run() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: true,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let second = second_runner
.run_publication_point(&handle)
@@ -1702,6 +1715,456 @@ fn runner_roa_validation_cache_reuses_vcir_outputs_on_second_fixture_run() {
assert_eq!(second.objects.roa_cache_stats.fresh_roas, 0);
}
#[test]
fn runner_publication_point_cache_observe_and_reuse_path() {
let fixture_dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("tests/fixtures/repository/rpki.cernet.net/repo/cernet/0");
assert!(fixture_dir.is_dir(), "fixture directory must exist");
let rsync_base_uri = "rsync://rpki.cernet.net/repo/cernet/0/".to_string();
let manifest_file = "05FC9C5B88506F7C0D3F862C8895BED67E9F8EBA.mft";
let manifest_rsync_uri = format!("{rsync_base_uri}{manifest_file}");
let fixture_manifest_bytes =
std::fs::read(fixture_dir.join(manifest_file)).expect("read manifest fixture");
let fixture_manifest =
crate::data_model::manifest::ManifestObject::decode_der(&fixture_manifest_bytes)
.expect("decode manifest fixture");
let validation_time = fixture_manifest.manifest.this_update + time::Duration::seconds(60);
let store_dir = tempfile::tempdir().expect("store dir");
let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
let policy = Policy {
sync_preference: crate::policy::SyncPreference::RsyncOnly,
..Policy::default()
};
let issuer_ca_der = std::fs::read(
std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(
"tests/fixtures/repository/rpki.apnic.net/repository/B527EF581D6611E2BB468F7C72FD1FF2/BfycW4hQb3wNP4YsiJW-1n6fjro.cer",
),
)
.expect("read issuer ca fixture");
let issuer_ca = ResourceCertificate::decode_der(&issuer_ca_der).expect("decode issuer ca");
let handle = CaInstanceHandle {
depth: 0,
tal_id: "test-tal".to_string(),
parent_manifest_rsync_uri: None,
ca_certificate_der: issuer_ca_der,
ca_certificate_rsync_uri: Some("rsync://rpki.apnic.net/repository/B527EF581D6611E2BB468F7C72FD1FF2/BfycW4hQb3wNP4YsiJW-1n6fjro.cer".to_string()),
effective_ip_resources: issuer_ca.tbs.extensions.ip_resources.clone(),
effective_as_resources: issuer_ca.tbs.extensions.as_resources.clone(),
rsync_base_uri: rsync_base_uri.clone(),
manifest_rsync_uri: manifest_rsync_uri.clone(),
publication_point_rsync_uri: rsync_base_uri,
rrdp_notification_uri: None,
};
let first_runner = Rpkiv1PublicationPointRunner {
store: &store,
policy: &policy,
http_fetcher: &NeverHttpFetcher,
rsync_fetcher: &LocalDirRsyncFetcher::new(&fixture_dir),
validation_time,
timing: None,
download_log: None,
replay_archive_index: None,
replay_delta_index: None,
rrdp_dedup: false,
rrdp_repo_cache: Mutex::new(HashMap::new()),
rsync_dedup: false,
rsync_repo_cache: Mutex::new(HashMap::new()),
current_repo_index: None,
repo_sync_runtime: None,
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: true,
enable_publication_point_validation_cache: false,
};
let first = first_runner
.run_publication_point(&handle)
.expect("first fresh run");
assert_eq!(first.source, PublicationPointSource::Fresh);
assert!(
store
.get_publication_point_cache_projection(&manifest_rsync_uri)
.expect("load publication-point projection")
.is_some()
);
let timing = crate::analysis::timing::TimingHandle::new(crate::analysis::timing::TimingMeta {
recorded_at_utc_rfc3339: "2026-01-01T00:00:00Z".to_string(),
validation_time_utc_rfc3339: "2026-01-01T00:00:00Z".to_string(),
tal_url: None,
db_path: None,
});
let observe_runner = Rpkiv1PublicationPointRunner {
store: &store,
policy: &policy,
http_fetcher: &NeverHttpFetcher,
rsync_fetcher: &LocalDirRsyncFetcher::new(&fixture_dir),
validation_time,
timing: Some(timing.clone()),
download_log: None,
replay_archive_index: None,
replay_delta_index: None,
rrdp_dedup: false,
rrdp_repo_cache: Mutex::new(HashMap::new()),
rsync_dedup: false,
rsync_repo_cache: Mutex::new(HashMap::new()),
current_repo_index: None,
repo_sync_runtime: None,
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: true,
enable_publication_point_validation_cache: false,
};
let observed = observe_runner
.run_publication_point(&handle)
.expect("observe-only run");
assert_eq!(observed.source, PublicationPointSource::Fresh);
assert_eq!(observed.objects.vrps, first.objects.vrps);
assert_eq!(
timing
.counts_snapshot()
.get("publication_point_cache_theoretical_hits")
.copied(),
Some(1)
);
let cache_runner = Rpkiv1PublicationPointRunner {
store: &store,
policy: &policy,
http_fetcher: &NeverHttpFetcher,
rsync_fetcher: &LocalDirRsyncFetcher::new(&fixture_dir),
validation_time,
timing: None,
download_log: None,
replay_archive_index: None,
replay_delta_index: None,
rrdp_dedup: false,
rrdp_repo_cache: Mutex::new(HashMap::new()),
rsync_dedup: false,
rsync_repo_cache: Mutex::new(HashMap::new()),
current_repo_index: None,
repo_sync_runtime: None,
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: true,
};
let cached = cache_runner
.run_publication_point(&handle)
.expect("publication-point cache run");
assert_eq!(cached.source, PublicationPointSource::PublicationPointCache);
assert_eq!(cached.objects.vrps, first.objects.vrps);
assert_eq!(cached.objects.aspas, first.objects.aspas);
assert_eq!(
cached.discovered_children.len(),
first.discovered_children.len()
);
assert!(!cached.cir_cached_objects.is_empty());
}
fn seed_publication_point_cache_projection(
store: &RocksStore,
policy: &Policy,
ca: &CaInstanceHandle,
validation_time: time::OffsetDateTime,
) -> ValidatedCaInstanceResult {
let child_bytes = b"child-cert".to_vec();
let child_hash = sha256_hex(&child_bytes);
let vcir = sample_vcir_for_projection(validation_time, &child_hash);
store
.put_blob_bytes_batch(&[
(child_hash, child_bytes),
(sha256_hex(b"manifest-bytes"), b"manifest-bytes".to_vec()),
])
.expect("put cache bytes");
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: ca.manifest_rsync_uri.clone(),
current_hash: Some(sha256_hex(b"manifest-bytes")),
repository_source: Some(ca.publication_point_rsync_uri.clone()),
object_type: Some("mft".to_string()),
state: RepositoryViewState::Present,
})
.expect("put manifest current view");
let projection = PublicationPointCacheProjection::from_vcir_with_context(
&vcir,
ca.publication_point_rsync_uri.clone(),
ca.ca_certificate_rsync_uri.clone(),
sha256_32(&ca.ca_certificate_der),
sha256_32(b"manifest-bytes"),
ta_context_digest_for_ca(ca),
parent_context_digest_for_ca(ca),
publication_point_cache_policy_fingerprint(policy),
)
.expect("build publication point projection");
store
.put_vcir_with_publication_point_cache_projection(&vcir, Some(&projection))
.expect("put publication point projection");
vcir
}
fn publication_point_cache_fixture_ca() -> CaInstanceHandle {
CaInstanceHandle {
depth: 1,
tal_id: "test-tal".to_string(),
parent_manifest_rsync_uri: None,
ca_certificate_der: b"ca-cert".to_vec(),
ca_certificate_rsync_uri: Some("rsync://example.test/repo/issuer/issuer.cer".to_string()),
effective_ip_resources: None,
effective_as_resources: None,
rsync_base_uri: "rsync://example.test/repo/issuer/".to_string(),
manifest_rsync_uri: "rsync://example.test/repo/issuer/issuer.mft".to_string(),
publication_point_rsync_uri: "rsync://example.test/repo/issuer/".to_string(),
rrdp_notification_uri: Some("https://example.test/notification.xml".to_string()),
}
}
#[test]
fn runner_publication_point_cache_reuses_projection_outputs_children_and_ccr() {
let store_dir = tempfile::tempdir().expect("store dir");
let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
let policy = Policy::default();
let validation_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::minutes(1);
let ca = publication_point_cache_fixture_ca();
let vcir = seed_publication_point_cache_projection(&store, &policy, &ca, validation_time);
let timing = crate::analysis::timing::TimingHandle::new(crate::analysis::timing::TimingMeta {
recorded_at_utc_rfc3339: "2026-01-01T00:00:00Z".to_string(),
validation_time_utc_rfc3339: "2026-01-01T00:00:00Z".to_string(),
tal_url: None,
db_path: None,
});
let runner = Rpkiv1PublicationPointRunner {
store: &store,
policy: &policy,
http_fetcher: &NeverHttpFetcher,
rsync_fetcher: &FailingRsyncFetcher,
validation_time,
timing: Some(timing.clone()),
download_log: None,
replay_archive_index: None,
replay_delta_index: None,
rrdp_dedup: false,
rrdp_repo_cache: Mutex::new(HashMap::new()),
rsync_dedup: false,
rsync_repo_cache: Mutex::new(HashMap::new()),
current_repo_index: None,
repo_sync_runtime: None,
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: Some(Mutex::new(CcrAccumulator::new(Vec::new()))),
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: true,
};
let result = runner
.observe_or_reuse_publication_point_cache(&ca, Some("rrdp"), Some("delta"), 7, None, &[])
.expect("cache result");
assert_eq!(result.source, PublicationPointSource::PublicationPointCache);
assert_eq!(result.objects.vrps.len(), 1);
assert_eq!(result.objects.aspas.len(), 1);
assert_eq!(result.objects.router_keys.len(), 1);
assert_eq!(result.discovered_children.len(), 1);
assert_eq!(
result.discovered_children[0].handle.manifest_rsync_uri,
vcir.child_entries[0].child_manifest_rsync_uri
);
assert!(result.cir_fresh_objects.is_empty());
assert!(!result.cir_cached_objects.is_empty());
assert_eq!(
runner
.ccr_accumulator_snapshot()
.expect("ccr accumulator")
.manifest_count(),
1
);
let counts = timing.counts_snapshot();
assert_eq!(
counts.get("publication_point_cache_reuse_hits").copied(),
Some(1)
);
assert_eq!(
counts
.get("publication_point_cache_outputs_reused")
.copied(),
Some(3)
);
assert_eq!(
counts
.get("publication_point_cache_children_reused")
.copied(),
Some(1)
);
}
#[test]
fn runner_publication_point_cache_blocks_parent_policy_and_output_time_mismatch() {
let store_dir = tempfile::tempdir().expect("store dir");
let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
let policy = Policy::default();
let validation_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::minutes(1);
let ca = publication_point_cache_fixture_ca();
seed_publication_point_cache_projection(&store, &policy, &ca, validation_time);
let timing = crate::analysis::timing::TimingHandle::new(crate::analysis::timing::TimingMeta {
recorded_at_utc_rfc3339: "2026-01-01T00:00:00Z".to_string(),
validation_time_utc_rfc3339: "2026-01-01T00:00:00Z".to_string(),
tal_url: None,
db_path: None,
});
let mut parent_changed_ca = ca.clone();
parent_changed_ca.parent_manifest_rsync_uri =
Some("rsync://example.test/repo/other-parent.mft".to_string());
let parent_changed_runner = Rpkiv1PublicationPointRunner {
store: &store,
policy: &policy,
http_fetcher: &NeverHttpFetcher,
rsync_fetcher: &FailingRsyncFetcher,
validation_time,
timing: Some(timing.clone()),
download_log: None,
replay_archive_index: None,
replay_delta_index: None,
rrdp_dedup: false,
rrdp_repo_cache: Mutex::new(HashMap::new()),
rsync_dedup: false,
rsync_repo_cache: Mutex::new(HashMap::new()),
current_repo_index: None,
repo_sync_runtime: None,
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: true,
};
assert!(
parent_changed_runner
.observe_or_reuse_publication_point_cache(
&parent_changed_ca,
Some("rrdp"),
Some("delta"),
7,
None,
&[]
)
.is_none()
);
assert_eq!(
timing
.counts_snapshot()
.get("publication_point_cache_miss_parent_context_mismatch")
.copied(),
Some(1)
);
let strict_policy = Policy {
strict: crate::policy::StrictPolicy::all(),
..Policy::default()
};
let policy_changed_runner = Rpkiv1PublicationPointRunner {
store: &store,
policy: &strict_policy,
http_fetcher: &NeverHttpFetcher,
rsync_fetcher: &FailingRsyncFetcher,
validation_time,
timing: Some(timing.clone()),
download_log: None,
replay_archive_index: None,
replay_delta_index: None,
rrdp_dedup: false,
rrdp_repo_cache: Mutex::new(HashMap::new()),
rsync_dedup: false,
rsync_repo_cache: Mutex::new(HashMap::new()),
current_repo_index: None,
repo_sync_runtime: None,
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: true,
};
assert!(
policy_changed_runner
.observe_or_reuse_publication_point_cache(
&ca,
Some("rrdp"),
Some("delta"),
7,
None,
&[]
)
.is_none()
);
assert_eq!(
timing
.counts_snapshot()
.get("publication_point_cache_miss_policy_mismatch")
.copied(),
Some(1)
);
let output_expired_runner = Rpkiv1PublicationPointRunner {
store: &store,
policy: &policy,
http_fetcher: &NeverHttpFetcher,
rsync_fetcher: &FailingRsyncFetcher,
validation_time: validation_time + time::Duration::minutes(40),
timing: Some(timing.clone()),
download_log: None,
replay_archive_index: None,
replay_delta_index: None,
rrdp_dedup: false,
rrdp_repo_cache: Mutex::new(HashMap::new()),
rsync_dedup: false,
rsync_repo_cache: Mutex::new(HashMap::new()),
current_repo_index: None,
repo_sync_runtime: None,
parallel_phase2_config: None,
parallel_roa_worker_pool: None,
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: true,
};
assert!(
output_expired_runner
.observe_or_reuse_publication_point_cache(
&ca,
Some("rrdp"),
Some("delta"),
7,
None,
&[]
)
.is_none()
);
assert_eq!(
timing
.counts_snapshot()
.get("publication_point_cache_miss_output_time_gate")
.copied(),
Some(1)
);
}
#[test]
fn runner_rsync_dedup_skips_second_sync_for_same_base() {
let fixture_dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
@@ -1789,6 +2252,8 @@ fn runner_rsync_dedup_skips_second_sync_for_same_base() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let first = runner.run_publication_point(&handle).expect("first run ok");
@@ -1903,6 +2368,8 @@ fn runner_rsync_dedup_skips_second_sync_for_same_module_scope() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let first = runner.run_publication_point(&handle).expect("first run ok");
@@ -2020,6 +2487,8 @@ fn runner_rsync_dedup_works_in_rsync_only_mode_even_when_rrdp_notify_exists() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let first = runner.run_publication_point(&handle).expect("first run ok");
@@ -2108,6 +2577,8 @@ fn runner_when_repo_sync_fails_uses_current_instance_vcir_and_keeps_children_emp
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let first = ok_runner
.run_publication_point(&handle)
@@ -2140,6 +2611,8 @@ fn runner_when_repo_sync_fails_uses_current_instance_vcir_and_keeps_children_emp
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let second = bad_runner
.run_publication_point(&handle)
@@ -3089,6 +3562,8 @@ fn runner_roa_validation_cache_uses_projection_not_full_vcir_fallback() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: true,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
assert!(
@@ -3821,6 +4296,8 @@ fn runner_dedup_paths_execute_with_timing_enabled() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let first = runner_rrdp
.run_publication_point(&handle)
@@ -3856,6 +4333,8 @@ fn runner_dedup_paths_execute_with_timing_enabled() {
ccr_accumulator: None,
persist_vcir: true,
enable_roa_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
};
let third = runner_rsync
.run_publication_point(&handle)