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

20260408_2 增加CIR sequence，未验证drop analysis，遇到问题是static pool保存太慢，拖慢整体录制，待解决

Browse Source
This commit is contained in:
yuyr
2026-04-09 16:08:11 +08:00
parent c9ef5aaf4c
commit e083fe4daa
26 changed files with 3233 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
src/bin/cir_drop_report.rs Normal file
View File

@@ -0,0 +1,250 @@
use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use rpki::bundle::decode_ccr_compare_views;
use rpki::ccr::decode_content_info;
use rpki::cir::{decode_cir, resolve_static_pool_file};
use rpki::data_model::roa::RoaObject;
const USAGE: &str = "Usage: cir_drop_report --cir <path> --ccr <path> --report-json <path> --static-root <path> --json-out <path> --md-out <path>";
#[derive(serde::Serialize)]
struct DroppedObjectRecord {
uri: String,
sha256: String,
kind: String,
reason_code: String,
reason_text: Option<String>,
publication_point: Option<String>,
manifest_uri: Option<String>,
derived_vrp_count: usize,
}
fn classify_reason(detail: Option<&str>, result: &str) -> String {
let text = detail.unwrap_or("").to_ascii_lowercase();
if text.contains("fetch") {
"fetch_failed".to_string()
} else if text.contains("manifest") {
"manifest_invalid".to_string()
} else if text.contains("crl") {
"crl_invalid".to_string()
} else if text.contains("policy") {
"policy_rejected".to_string()
} else if text.contains("parse") {
"object_parse_failed".to_string()
} else if text.contains("signature") || text.contains("cms") {
"cms_signature_invalid".to_string()
} else if text.contains("resource") {
"resource_invalid".to_string()
} else if text.contains("expired") || text.contains("not yet valid") {
"expired_or_not_yet_valid".to_string()
} else if result == "skipped" {
"skipped".to_string()
} else if result == "error" {
"error".to_string()
} else {
"other".to_string()
}
}
fn parse_args(argv: &[String]) -> Result<(PathBuf, PathBuf, PathBuf, PathBuf, PathBuf, PathBuf), String> {
let mut cir = None;
let mut ccr = None;
let mut report = None;
let mut static_root = None;
let mut json_out = None;
let mut md_out = None;
let mut i = 1usize;
while i < argv.len() {
match argv[i].as_str() {
"--cir" => {
i += 1;
cir = Some(PathBuf::from(argv.get(i).ok_or("--cir requires a value")?));
}
"--ccr" => {
i += 1;
ccr = Some(PathBuf::from(argv.get(i).ok_or("--ccr requires a value")?));
}
"--report-json" => {
i += 1;
report = Some(PathBuf::from(
argv.get(i).ok_or("--report-json requires a value")?,
));
}
"--static-root" => {
i += 1;
static_root = Some(PathBuf::from(
argv.get(i).ok_or("--static-root requires a value")?,
));
}
"--json-out" => {
i += 1;
json_out = Some(PathBuf::from(
argv.get(i).ok_or("--json-out requires a value")?,
));
}
"--md-out" => {
i += 1;
md_out = Some(PathBuf::from(argv.get(i).ok_or("--md-out requires a value")?));
}
"-h" | "--help" => return Err(USAGE.to_string()),
other => return Err(format!("unknown argument: {other}\n\n{USAGE}")),
}
i += 1;
}
Ok((
cir.ok_or_else(|| format!("--cir is required\n\n{USAGE}"))?,
ccr.ok_or_else(|| format!("--ccr is required\n\n{USAGE}"))?,
report.ok_or_else(|| format!("--report-json is required\n\n{USAGE}"))?,
static_root.ok_or_else(|| format!("--static-root is required\n\n{USAGE}"))?,
json_out.ok_or_else(|| format!("--json-out is required\n\n{USAGE}"))?,
md_out.ok_or_else(|| format!("--md-out is required\n\n{USAGE}"))?,
))
}
fn main() -> Result<(), String> {
let argv: Vec<String> = std::env::args().collect();
let (cir_path, ccr_path, report_path, static_root, json_out, md_out) = parse_args(&argv)?;
let cir = decode_cir(&std::fs::read(&cir_path).map_err(|e| format!("read cir failed: {e}"))?)
.map_err(|e| format!("decode cir failed: {e}"))?;
let ccr = decode_content_info(
&std::fs::read(&ccr_path).map_err(|e| format!("read ccr failed: {e}"))?,
)
.map_err(|e| format!("decode ccr failed: {e}"))?;
let (vrps, vaps) =
decode_ccr_compare_views(&ccr, "unknown").map_err(|e| format!("decode compare views failed: {e}"))?;
let report: serde_json::Value =
serde_json::from_slice(&std::fs::read(&report_path).map_err(|e| format!("read report failed: {e}"))?)
.map_err(|e| format!("parse report failed: {e}"))?;
let mut object_hash_by_uri = BTreeMap::new();
for object in &cir.objects {
object_hash_by_uri.insert(object.rsync_uri.clone(), hex::encode(&object.sha256));
}
let publication_points = report["publication_points"]
.as_array()
.ok_or("report.publication_points must be an array")?;
let mut dropped_objects = Vec::new();
let mut dropped_vrp_rows = BTreeSet::new();
let mut dropped_by_kind: BTreeMap<String, usize> = BTreeMap::new();
let mut dropped_by_reason: BTreeMap<String, usize> = BTreeMap::new();
let mut unknown_roa_objects = 0usize;
for pp in publication_points {
let publication_point = pp["publication_point_rsync_uri"].as_str().map(str::to_string);
let manifest_uri = pp["manifest_rsync_uri"].as_str().map(str::to_string);
for obj in pp["objects"].as_array().into_iter().flatten() {
let result = obj["result"].as_str().unwrap_or("unknown");
if result == "ok" {
continue;
}
let uri = obj["rsync_uri"].as_str().unwrap_or("").to_string();
let hash = obj["sha256_hex"]
.as_str()
.map(str::to_string)
.or_else(|| object_hash_by_uri.get(&uri).cloned())
.unwrap_or_default();
let kind = obj["kind"].as_str().unwrap_or("other").to_string();
let detail = obj["detail"].as_str().map(str::to_string);
let reason_code = classify_reason(detail.as_deref(), result);
*dropped_by_kind.entry(kind.clone()).or_insert(0) += 1;
*dropped_by_reason.entry(reason_code.clone()).or_insert(0) += 1;
let mut derived_vrp_count = 0usize;
if kind == "roa" && !hash.is_empty() {
match resolve_static_pool_file(&static_root, &hash) {
Ok(path) => {
if let Ok(bytes) = std::fs::read(&path) {
if let Ok(roa) = RoaObject::decode_der(&bytes) {
for family in roa.roa.ip_addr_blocks {
for addr in family.addresses {
let prefix = match addr.prefix.afi {
rpki::data_model::roa::RoaAfi::Ipv4 => format!(
"{}.{}.{}.{}/{}",
addr.prefix.addr[0],
addr.prefix.addr[1],
addr.prefix.addr[2],
addr.prefix.addr[3],
addr.prefix.prefix_len
),
rpki::data_model::roa::RoaAfi::Ipv6 => {
let bytes: [u8; 16] = addr.prefix.addr;
format!("{}/{}", std::net::Ipv6Addr::from(bytes), addr.prefix.prefix_len)
}
};
let max_len = addr.max_length.unwrap_or(addr.prefix.prefix_len);
dropped_vrp_rows.insert((roa.roa.as_id, prefix, max_len));
derived_vrp_count += 1;
}
}
} else {
unknown_roa_objects += 1;
}
} else {
unknown_roa_objects += 1;
}
}
Err(_) => unknown_roa_objects += 1,
}
}
dropped_objects.push(DroppedObjectRecord {
uri,
sha256: hash,
kind,
reason_code,
reason_text: detail,
publication_point: publication_point.clone(),
manifest_uri: manifest_uri.clone(),
derived_vrp_count,
});
}
}
let output = serde_json::json!({
"summary": {
"finalVrpCount": vrps.len(),
"finalVapCount": vaps.len(),
"droppedVrpCount": dropped_vrp_rows.len(),
"droppedObjectCount": dropped_objects.len(),
"droppedByKind": dropped_by_kind,
"droppedByReason": dropped_by_reason,
"unknownDroppedRoaObjects": unknown_roa_objects,
},
"objects": dropped_objects,
});
if let Some(parent) = json_out.parent() {
std::fs::create_dir_all(parent).map_err(|e| format!("create json parent failed: {e}"))?;
}
std::fs::write(&json_out, serde_json::to_vec_pretty(&output).unwrap())
.map_err(|e| format!("write json failed: {e}"))?;
let mut md = String::new();
md.push_str("# CIR Drop Report\n\n");
md.push_str(&format!("- `final_vrp_count`: `{}`\n", vrps.len()));
md.push_str(&format!("- `final_vap_count`: `{}`\n", vaps.len()));
md.push_str(&format!("- `dropped_vrp_count`: `{}`\n", output["summary"]["droppedVrpCount"]));
md.push_str(&format!("- `dropped_object_count`: `{}`\n", output["summary"]["droppedObjectCount"]));
md.push_str(&format!(
"- `unknown_dropped_roa_objects`: `{}`\n\n",
output["summary"]["unknownDroppedRoaObjects"]
));
md.push_str("## Dropped By Kind\n\n");
for (kind, count) in output["summary"]["droppedByKind"].as_object().into_iter().flatten() {
md.push_str(&format!("- `{kind}`: `{}`\n", count.as_u64().unwrap_or(0)));
}
md.push_str("\n## Dropped By Reason\n\n");
for (reason, count) in output["summary"]["droppedByReason"].as_object().into_iter().flatten() {
md.push_str(&format!("- `{reason}`: `{}`\n", count.as_u64().unwrap_or(0)));
}
if let Some(parent) = md_out.parent() {
std::fs::create_dir_all(parent).map_err(|e| format!("create markdown parent failed: {e}"))?;
}
std::fs::write(&md_out, md).map_err(|e| format!("write markdown failed: {e}"))?;
Ok(())
}

102
src/bin/cir_ta_only_fixture.rs Normal file
View File

@@ -0,0 +1,102 @@
use std::path::PathBuf;
use rpki::cir::{encode_cir, write_bytes_to_static_pool, CanonicalInputRepresentation, CirHashAlgorithm, CirObject, CirTal, CIR_VERSION_V1};
use sha2::Digest;
const USAGE: &str = "Usage: cir_ta_only_fixture --tal-path <path> --ta-path <path> --tal-uri <url> --validation-time <rfc3339> --cir-out <path> --static-root <path>";
fn parse_args(argv: &[String]) -> Result<(PathBuf, PathBuf, String, time::OffsetDateTime, PathBuf, PathBuf), String> {
let mut tal_path = None;
let mut ta_path = None;
let mut tal_uri = None;
let mut validation_time = None;
let mut cir_out = None;
let mut static_root = None;
let mut i = 1usize;
while i < argv.len() {
match argv[i].as_str() {
"--tal-path" => {
i += 1;
tal_path = Some(PathBuf::from(argv.get(i).ok_or("--tal-path requires a value")?));
}
"--ta-path" => {
i += 1;
ta_path = Some(PathBuf::from(argv.get(i).ok_or("--ta-path requires a value")?));
}
"--tal-uri" => {
i += 1;
tal_uri = Some(argv.get(i).ok_or("--tal-uri requires a value")?.clone());
}
"--validation-time" => {
i += 1;
let raw = argv.get(i).ok_or("--validation-time requires a value")?;
validation_time = Some(
time::OffsetDateTime::parse(
raw,
&time::format_description::well_known::Rfc3339,
)
.map_err(|e| format!("invalid validation time: {e}"))?,
);
}
"--cir-out" => {
i += 1;
cir_out = Some(PathBuf::from(argv.get(i).ok_or("--cir-out requires a value")?));
}
"--static-root" => {
i += 1;
static_root =
Some(PathBuf::from(argv.get(i).ok_or("--static-root requires a value")?));
}
"-h" | "--help" => return Err(USAGE.to_string()),
other => return Err(format!("unknown argument: {other}\n\n{USAGE}")),
}
i += 1;
}
Ok((
tal_path.ok_or_else(|| format!("--tal-path is required\n\n{USAGE}"))?,
ta_path.ok_or_else(|| format!("--ta-path is required\n\n{USAGE}"))?,
tal_uri.ok_or_else(|| format!("--tal-uri is required\n\n{USAGE}"))?,
validation_time.ok_or_else(|| format!("--validation-time is required\n\n{USAGE}"))?,
cir_out.ok_or_else(|| format!("--cir-out is required\n\n{USAGE}"))?,
static_root.ok_or_else(|| format!("--static-root is required\n\n{USAGE}"))?,
))
}
fn main() -> Result<(), String> {
let argv: Vec<String> = std::env::args().collect();
let (tal_path, ta_path, tal_uri, validation_time, cir_out, static_root) = parse_args(&argv)?;
let tal_bytes = std::fs::read(&tal_path).map_err(|e| format!("read tal failed: {e}"))?;
let ta_bytes = std::fs::read(&ta_path).map_err(|e| format!("read ta failed: {e}"))?;
let tal = rpki::data_model::tal::Tal::decode_bytes(&tal_bytes)
.map_err(|e| format!("decode tal failed: {e}"))?;
let ta_rsync_uri = tal
.ta_uris
.iter()
.find(|uri| uri.scheme() == "rsync")
.ok_or("tal must contain an rsync URI")?
.as_str()
.to_string();
let sha = sha2::Sha256::digest(&ta_bytes);
let hash_hex = hex::encode(sha);
write_bytes_to_static_pool(&static_root, validation_time.date(), &hash_hex, &ta_bytes)
.map_err(|e| format!("write static pool failed: {e}"))?;
let cir = CanonicalInputRepresentation {
version: CIR_VERSION_V1,
hash_alg: CirHashAlgorithm::Sha256,
validation_time,
objects: vec![CirObject {
rsync_uri: ta_rsync_uri,
sha256: sha.to_vec(),
}],
tals: vec![CirTal { tal_uri, tal_bytes }],
};
let der = encode_cir(&cir).map_err(|e| format!("encode cir failed: {e}"))?;
if let Some(parent) = cir_out.parent() {
std::fs::create_dir_all(parent).map_err(|e| format!("create cir parent failed: {e}"))?;
}
std::fs::write(&cir_out, der).map_err(|e| format!("write cir failed: {e}"))?;
Ok(())
}

110
src/cir/export.rs
View File

@@ -2,6 +2,7 @@ use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::path::Path;
use crate::audit::{AuditObjectResult, PublicationPointAudit};
use crate::cir::encode::{CirEncodeError, encode_cir};
use crate::cir::model::{
CIR_VERSION_V1, CanonicalInputRepresentation, CirHashAlgorithm, CirObject, CirTal,
@@ -13,6 +14,14 @@ use crate::cir::static_pool::{
use crate::data_model::ta::TrustAnchor;
use crate::storage::{RepositoryViewState, RocksStore};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CirExportTiming {
pub build_cir_ms: u64,
pub static_pool_ms: u64,
pub write_cir_ms: u64,
pub total_ms: u64,
}
#[derive(Debug, thiserror::Error)]
pub enum CirExportError {
#[error("list repository_view entries failed: {0}")]
@@ -42,6 +51,7 @@ pub struct CirExportSummary {
pub object_count: usize,
pub tal_count: usize,
pub static_pool: CirStaticPoolExportSummary,
pub timing: CirExportTiming,
}
pub fn build_cir_from_run(
@@ -49,6 +59,7 @@ pub fn build_cir_from_run(
trust_anchor: &TrustAnchor,
tal_uri: &str,
validation_time: time::OffsetDateTime,
publication_points: &[PublicationPointAudit],
) -> Result<CanonicalInputRepresentation, CirExportError> {
if !(tal_uri.starts_with("https://") || tal_uri.starts_with("http://")) {
return Err(CirExportError::InvalidTalUri(tal_uri.to_string()));
@@ -69,6 +80,26 @@ pub fn build_cir_from_run(
}
}
// CIR must describe the actual input world used by validation. When a
// publication point falls back to the latest validated current instance,
// repository_view may not contain the reused manifest/object set. Pull
// those object hashes from the audit so replay can reconstruct the same
// world state.
for pp in publication_points {
if pp.source != "vcir_current_instance" {
continue;
}
for obj in &pp.objects {
if obj.result != AuditObjectResult::Ok {
continue;
}
if !obj.rsync_uri.starts_with("rsync://") {
continue;
}
objects.insert(obj.rsync_uri.clone(), obj.sha256_hex.to_ascii_lowercase());
}
}
let ta_hash = ta_sha256_hex(&trust_anchor.ta_certificate.raw_der);
let mut saw_rsync_uri = false;
for uri in &trust_anchor.tal.ta_uris {
@@ -149,17 +180,41 @@ pub fn export_cir_from_run(
trust_anchor: &TrustAnchor,
tal_uri: &str,
validation_time: time::OffsetDateTime,
publication_points: &[PublicationPointAudit],
cir_out: &Path,
static_root: &Path,
capture_date_utc: time::Date,
) -> Result<CirExportSummary, CirExportError> {
let cir = build_cir_from_run(store, trust_anchor, tal_uri, validation_time)?;
let total_started = std::time::Instant::now();
let started = std::time::Instant::now();
let cir = build_cir_from_run(
store,
trust_anchor,
tal_uri,
validation_time,
publication_points,
)?;
let build_cir_ms = started.elapsed().as_millis() as u64;
let started = std::time::Instant::now();
let static_pool = export_cir_static_pool(store, static_root, capture_date_utc, &cir, trust_anchor)?;
let static_pool_ms = started.elapsed().as_millis() as u64;
let started = std::time::Instant::now();
write_cir_file(cir_out, &cir)?;
let write_cir_ms = started.elapsed().as_millis() as u64;
Ok(CirExportSummary {
object_count: cir.objects.len(),
tal_count: cir.tals.len(),
static_pool,
timing: CirExportTiming {
build_cir_ms,
static_pool_ms,
write_cir_ms,
total_ms: total_started.elapsed().as_millis() as u64,
},
})
}
@@ -222,7 +277,13 @@ mod tests {
.unwrap();
let ta = sample_trust_anchor();
let cir = build_cir_from_run(&store, &ta, "https://example.test/root.tal", sample_time())
let cir = build_cir_from_run(
&store,
&ta,
"https://example.test/root.tal",
sample_time(),
&[],
)
.expect("build cir");
assert_eq!(cir.version, CIR_VERSION_V1);
assert_eq!(cir.tals.len(), 1);
@@ -267,6 +328,7 @@ mod tests {
&ta,
"https://example.test/root.tal",
sample_time(),
&[],
&cir_path,
&static_root,
sample_date(),
@@ -282,4 +344,48 @@ mod tests {
let object_path = static_pool_path(&static_root, sample_date(), &hash).unwrap();
assert_eq!(std::fs::read(object_path).unwrap(), bytes);
}
#[test]
fn build_cir_from_run_includes_vcir_current_instance_objects_from_audit() {
let td = tempfile::tempdir().unwrap();
let store = RocksStore::open(td.path()).unwrap();
let ta = sample_trust_anchor();
let mut pp = PublicationPointAudit {
source: "vcir_current_instance".to_string(),
..PublicationPointAudit::default()
};
pp.objects.push(crate::audit::ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/fallback.mft".to_string(),
sha256_hex: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(),
kind: crate::audit::AuditObjectKind::Manifest,
result: crate::audit::AuditObjectResult::Ok,
detail: None,
});
pp.objects.push(crate::audit::ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/fallback.roa".to_string(),
sha256_hex: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
kind: crate::audit::AuditObjectKind::Roa,
result: crate::audit::AuditObjectResult::Ok,
detail: None,
});
let cir = build_cir_from_run(
&store,
&ta,
"https://example.test/root.tal",
sample_time(),
&[pp],
)
.expect("build cir");
assert!(cir
.objects
.iter()
.any(|item| item.rsync_uri == "rsync://example.test/repo/fallback.mft"));
assert!(cir
.objects
.iter()
.any(|item| item.rsync_uri == "rsync://example.test/repo/fallback.roa"));
}
}

2
src/cir/mod.rs
View File

@@ -2,6 +2,7 @@ pub mod decode;
pub mod encode;
pub mod materialize;
pub mod model;
pub mod sequence;
#[cfg(feature = "full")]
pub mod export;
#[cfg(feature = "full")]
@@ -16,6 +17,7 @@ pub use materialize::{
pub use model::{
CIR_VERSION_V1, CirHashAlgorithm, CirObject, CirTal, CanonicalInputRepresentation,
};
pub use sequence::{CirSequenceManifest, CirSequenceStep, CirSequenceStepKind};
#[cfg(feature = "full")]
pub use export::{CirExportError, CirExportSummary, build_cir_from_run, export_cir_from_run, write_cir_file};
#[cfg(feature = "full")]

147
src/cir/sequence.rs Normal file
View File

@@ -0,0 +1,147 @@
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CirSequenceStepKind {
Full,
Delta,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CirSequenceStep {
pub step_id: String,
pub kind: CirSequenceStepKind,
pub validation_time: String,
pub cir_path: String,
pub ccr_path: String,
pub report_path: String,
pub previous_step_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CirSequenceManifest {
pub version: u32,
pub static_root: String,
pub steps: Vec<CirSequenceStep>,
}
impl CirSequenceManifest {
pub fn validate(&self) -> Result<(), String> {
if self.version == 0 {
return Err("sequence.version must be positive".to_string());
}
if self.static_root.trim().is_empty() {
return Err("sequence.static_root must not be empty".to_string());
}
if self.steps.is_empty() {
return Err("sequence.steps must not be empty".to_string());
}
let mut previous_ids = std::collections::BTreeSet::new();
for (idx, step) in self.steps.iter().enumerate() {
if step.step_id.trim().is_empty() {
return Err(format!("sequence.steps[{idx}].step_id must not be empty"));
}
if !previous_ids.insert(step.step_id.clone()) {
return Err(format!("sequence.steps[{idx}].step_id must be unique"));
}
if step.validation_time.trim().is_empty() {
return Err(format!(
"sequence.steps[{idx}].validation_time must not be empty"
));
}
if step.cir_path.trim().is_empty()
|| step.ccr_path.trim().is_empty()
|| step.report_path.trim().is_empty()
{
return Err(format!(
"sequence.steps[{idx}] output paths must not be empty"
));
}
match step.kind {
CirSequenceStepKind::Full => {
if idx != 0 {
return Err("full step must be the first step".to_string());
}
if step.previous_step_id.is_some() {
return Err("full step must not reference previous_step_id".to_string());
}
}
CirSequenceStepKind::Delta => {
if idx == 0 {
return Err("delta step cannot be the first step".to_string());
}
let previous = step.previous_step_id.as_ref().ok_or_else(|| {
format!("sequence.steps[{idx}] delta step must set previous_step_id")
})?;
if !previous_ids.contains(previous) {
return Err(format!(
"sequence.steps[{idx}] previous_step_id must reference an earlier step"
));
}
}
}
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::{CirSequenceManifest, CirSequenceStep, CirSequenceStepKind};
fn sample_manifest() -> CirSequenceManifest {
CirSequenceManifest {
version: 1,
static_root: "static".to_string(),
steps: vec![
CirSequenceStep {
step_id: "full".to_string(),
kind: CirSequenceStepKind::Full,
validation_time: "2026-04-09T00:00:00Z".to_string(),
cir_path: "full/input.cir".to_string(),
ccr_path: "full/result.ccr".to_string(),
report_path: "full/report.json".to_string(),
previous_step_id: None,
},
CirSequenceStep {
step_id: "delta-001".to_string(),
kind: CirSequenceStepKind::Delta,
validation_time: "2026-04-09T00:10:00Z".to_string(),
cir_path: "delta-001/input.cir".to_string(),
ccr_path: "delta-001/result.ccr".to_string(),
report_path: "delta-001/report.json".to_string(),
previous_step_id: Some("full".to_string()),
},
],
}
}
#[test]
fn sequence_manifest_validate_accepts_minimal_chain() {
sample_manifest().validate().expect("valid sequence");
}
#[test]
fn sequence_manifest_validate_rejects_bad_order_and_duplicates() {
let mut bad = sample_manifest();
bad.steps.swap(0, 1);
let err = bad.validate().expect_err("full must be first");
assert!(
err.contains("delta step cannot be the first step")
|| err.contains("full step must be the first step")
);
let mut dup = sample_manifest();
dup.steps[1].step_id = "full".to_string();
let err = dup.validate().expect_err("duplicate id must fail");
assert!(err.contains("must be unique"));
}
#[test]
fn sequence_manifest_validate_rejects_missing_previous_reference() {
let mut bad = sample_manifest();
bad.steps[1].previous_step_id = Some("missing".to_string());
let err = bad.validate().expect_err("missing previous must fail");
assert!(err.contains("previous_step_id"));
}
}

110
src/cli.rs
View File

@@ -22,6 +22,27 @@ use crate::validation::run_tree_from_tal::{
run_tree_from_tal_url_serial_audit_with_timing,
};
use crate::validation::tree::TreeRunConfig;
use serde::Serialize;
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
struct RunStageTiming {
validation_ms: u64,
report_build_ms: u64,
report_write_ms: Option<u64>,
ccr_build_ms: Option<u64>,
ccr_write_ms: Option<u64>,
cir_build_cir_ms: Option<u64>,
cir_static_pool_ms: Option<u64>,
cir_write_cir_ms: Option<u64>,
cir_total_ms: Option<u64>,
total_ms: u64,
publication_points: usize,
repo_sync_ms_total: u64,
download_event_count: u64,
rrdp_download_ms_total: u64,
rsync_download_ms_total: u64,
download_bytes_total: u64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CliArgs {
@@ -655,6 +676,8 @@ pub fn run(argv: &[String]) -> Result<(), String> {
None
};
let total_started = std::time::Instant::now();
let validation_started = std::time::Instant::now();
let out = if delta_replay_mode {
let tal_path = args
.tal_path
@@ -988,11 +1011,43 @@ pub fn run(argv: &[String]) -> Result<(), String> {
}
};
let validation_ms = validation_started.elapsed().as_millis() as u64;
if let Some((_out_dir, t)) = timing.as_ref() {
t.record_count("instances_processed", out.tree.instances_processed as u64);
t.record_count("instances_failed", out.tree.instances_failed as u64);
}
let publication_points = out.publication_points.len();
let repo_sync_ms_total: u64 = out
.publication_points
.iter()
.map(|pp| pp.repo_sync_duration_ms.unwrap_or(0))
.sum();
let download_event_count = out.download_stats.events_total;
let rrdp_download_ms_total: u64 = ["rrdp_notification", "rrdp_snapshot", "rrdp_delta"]
.iter()
.map(|key| {
out.download_stats
.by_kind
.get(*key)
.map(|item| item.duration_ms_total)
.unwrap_or(0)
})
.sum();
let rsync_download_ms_total = out
.download_stats
.by_kind
.get("rsync")
.map(|item| item.duration_ms_total)
.unwrap_or(0);
let download_bytes_total: u64 = out
.download_stats
.by_kind
.values()
.map(|item| item.bytes_total.unwrap_or(0))
.sum();
#[cfg(feature = "profile")]
let profiler_report = if let Some(guard) = profiler_guard.take() {
Some(
@@ -1005,7 +1060,10 @@ pub fn run(argv: &[String]) -> Result<(), String> {
None
};
let mut ccr_build_ms = None;
let mut ccr_write_ms = None;
if let Some(path) = args.ccr_out_path.as_deref() {
let started = std::time::Instant::now();
let ccr = build_ccr_from_run(
&store,
&[out.discovery.trust_anchor.clone()],
@@ -1015,10 +1073,17 @@ pub fn run(argv: &[String]) -> Result<(), String> {
time::OffsetDateTime::now_utc(),
)
.map_err(|e| e.to_string())?;
ccr_build_ms = Some(started.elapsed().as_millis() as u64);
let started = std::time::Instant::now();
write_ccr_file(path, &ccr).map_err(|e| e.to_string())?;
ccr_write_ms = Some(started.elapsed().as_millis() as u64);
eprintln!("wrote CCR: {}", path.display());
}
let mut cir_build_cir_ms = None;
let mut cir_static_pool_ms = None;
let mut cir_write_cir_ms = None;
let mut cir_total_ms = None;
if args.cir_enabled {
let cir_tal_uri = args
.tal_url
@@ -1038,25 +1103,66 @@ pub fn run(argv: &[String]) -> Result<(), String> {
&out.discovery.trust_anchor,
&cir_tal_uri,
validation_time,
&out.publication_points,
cir_out_path,
cir_static_root,
time::OffsetDateTime::now_utc().date(),
)
.map_err(|e| e.to_string())?;
cir_build_cir_ms = Some(summary.timing.build_cir_ms);
cir_static_pool_ms = Some(summary.timing.static_pool_ms);
cir_write_cir_ms = Some(summary.timing.write_cir_ms);
cir_total_ms = Some(summary.timing.total_ms);
eprintln!(
"wrote CIR: {} (objects={}, tals={}, static_written={}, static_reused={})",
"wrote CIR: {} (objects={}, tals={}, static_written={}, static_reused={}, build_cir_ms={}, static_pool_ms={}, write_cir_ms={}, total_ms={})",
cir_out_path.display(),
summary.object_count,
summary.tal_count,
summary.static_pool.written_files,
summary.static_pool.reused_files
summary.static_pool.reused_files,
summary.timing.build_cir_ms,
summary.timing.static_pool_ms,
summary.timing.write_cir_ms,
summary.timing.total_ms
);
}
let report_started = std::time::Instant::now();
let report = build_report(&policy, validation_time, out);
let report_build_ms = report_started.elapsed().as_millis() as u64;
let mut report_write_ms = None;
if let Some(p) = args.report_json_path.as_deref() {
let started = std::time::Instant::now();
write_json(p, &report)?;
report_write_ms = Some(started.elapsed().as_millis() as u64);
if let Some(parent) = p.parent() {
let stage_timing = RunStageTiming {
validation_ms,
report_build_ms,
report_write_ms,
ccr_build_ms,
ccr_write_ms,
cir_build_cir_ms,
cir_static_pool_ms,
cir_write_cir_ms,
cir_total_ms,
total_ms: total_started.elapsed().as_millis() as u64,
publication_points,
repo_sync_ms_total,
download_event_count,
rrdp_download_ms_total,
rsync_download_ms_total,
download_bytes_total,
};
let stage_timing_path = parent.join("stage-timing.json");
std::fs::write(
&stage_timing_path,
serde_json::to_vec_pretty(&stage_timing).map_err(|e| e.to_string())?,
)
.map_err(|e| format!("write stage timing failed: {}: {e}", stage_timing_path.display()))?;
eprintln!("analysis: wrote {}", stage_timing_path.display());
}
}
if let Some((out_dir, t)) = timing.as_ref() {