20260420_2 完成输出report和ccr优化，snapshot耗时优化到70多秒

src/ccr/accumulator.rs

@@ -0,0 +1,415 @@
+use std::collections::BTreeMap;
+
+use crate::ccr::build::{
+    build_aspa_payload_state, build_roa_payload_state, build_router_key_state_from_runtime,
+    build_trust_anchor_state,
+};
+use crate::ccr::encode::encode_manifest_state_payload_der;
+use crate::ccr::hash::compute_state_hash;
+use crate::ccr::model::{
+    CcrDigestAlgorithm, ManifestInstance, ManifestState, RpkiCanonicalCacheRepresentation,
+};
+use crate::data_model::common::BigUnsigned;
+use crate::data_model::ta::TrustAnchor;
+use crate::storage::VcirCcrManifestProjection;
+use crate::validation::objects::{AspaAttestation, RouterKeyPayload, Vrp};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CcrManifestContribution {
+    pub manifest_rsync_uri: String,
+    pub hash: Vec<u8>,
+    pub size: u64,
+    pub aki: Vec<u8>,
+    pub manifest_number_be: Vec<u8>,
+    pub this_update: time::OffsetDateTime,
+    pub locations_der: Vec<Vec<u8>>,
+    pub subordinate_skis: Vec<Vec<u8>>,
+}
+
+impl CcrManifestContribution {
+    fn from_projection(projection: &VcirCcrManifestProjection) -> Result<Self, String> {
+        let this_update = projection
+            .manifest_this_update
+            .parse()
+            .map_err(|e| format!("parse projection manifest_this_update failed: {e}"))?;
+        Ok(Self {
+            manifest_rsync_uri: projection.manifest_rsync_uri.clone(),
+            hash: projection.manifest_sha256.clone(),
+            size: projection.manifest_size,
+            aki: projection.manifest_ee_aki.clone(),
+            manifest_number_be: projection.manifest_number_be.clone(),
+            this_update,
+            locations_der: projection.manifest_sia_locations_der.clone(),
+            subordinate_skis: projection.subordinate_skis.clone(),
+        })
+    }
+
+    fn to_manifest_instance(&self) -> ManifestInstance {
+        ManifestInstance {
+            hash: self.hash.clone(),
+            size: self.size,
+            aki: self.aki.clone(),
+            manifest_number: BigUnsigned {
+                bytes_be: self.manifest_number_be.clone(),
+            },
+            this_update: self.this_update,
+            locations: self.locations_der.clone(),
+            subordinates: self.subordinate_skis.clone(),
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CcrAccumulator {
+    trust_anchors: Vec<TrustAnchor>,
+    manifests_by_hash: BTreeMap<Vec<u8>, CcrManifestContribution>,
+    most_recent_update: time::OffsetDateTime,
+}
+
+impl CcrAccumulator {
+    pub fn new(trust_anchors: Vec<TrustAnchor>) -> Self {
+        Self {
+            trust_anchors,
+            manifests_by_hash: BTreeMap::new(),
+            most_recent_update: time::OffsetDateTime::UNIX_EPOCH,
+        }
+    }
+
+    pub fn append_manifest_projection(
+        &mut self,
+        projection: &VcirCcrManifestProjection,
+    ) -> Result<(), String> {
+        let contribution = CcrManifestContribution::from_projection(projection)?;
+        match self.manifests_by_hash.get(contribution.hash.as_slice()) {
+            Some(existing) if existing != &contribution => {
+                return Err(format!(
+                    "duplicate manifest hash with conflicting content for URI: {}",
+                    contribution.manifest_rsync_uri
+                ));
+            }
+            Some(_) => {}
+            None => {
+                self.manifests_by_hash
+                    .insert(contribution.hash.clone(), contribution.clone());
+            }
+        }
+        if contribution.this_update > self.most_recent_update {
+            self.most_recent_update = contribution.this_update;
+        }
+        Ok(())
+    }
+
+    pub fn finish(
+        &self,
+        produced_at: time::OffsetDateTime,
+        vrps: &[Vrp],
+        aspas: &[AspaAttestation],
+        router_keys: &[RouterKeyPayload],
+    ) -> Result<RpkiCanonicalCacheRepresentation, String> {
+        let manifest_instances = self
+            .manifests_by_hash
+            .values()
+            .map(CcrManifestContribution::to_manifest_instance)
+            .collect::<Vec<_>>();
+        let manifest_payload_der = encode_manifest_state_payload_der(&manifest_instances)
+            .map_err(|e| format!("manifest state encoding failed: {e}"))?;
+        let manifest_state = ManifestState {
+            mis: manifest_instances,
+            most_recent_update: self.most_recent_update,
+            hash: compute_state_hash(&manifest_payload_der),
+        };
+        let vrp_state = build_roa_payload_state(vrps).map_err(|e| e.to_string())?;
+        let aspa_state = build_aspa_payload_state(aspas).map_err(|e| e.to_string())?;
+        let ta_state = build_trust_anchor_state(&self.trust_anchors).map_err(|e| e.to_string())?;
+        let router_key_state =
+            build_router_key_state_from_runtime(router_keys).map_err(|e| e.to_string())?;
+        Ok(RpkiCanonicalCacheRepresentation {
+            version: 0,
+            hash_alg: CcrDigestAlgorithm::Sha256,
+            produced_at,
+            mfts: Some(manifest_state),
+            vrps: Some(vrp_state),
+            vaps: Some(aspa_state),
+            tas: Some(ta_state),
+            rks: Some(router_key_state),
+        })
+    }
+
+    pub fn manifest_count(&self) -> usize {
+        self.manifests_by_hash.len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ccr::export::build_ccr_from_run;
+    use crate::ccr::verify::verify_content_info;
+    use crate::data_model::manifest::ManifestObject;
+    use crate::data_model::rc::SubjectInfoAccess;
+    use crate::data_model::roa::{IpPrefix, RoaAfi};
+    use crate::data_model::ta::TrustAnchor;
+    use crate::data_model::tal::Tal;
+    use crate::storage::{
+        PackTime, RawByHashEntry, RocksStore, ValidatedCaInstanceResult, ValidatedManifestMeta,
+        VcirArtifactKind, VcirArtifactRole, VcirArtifactValidationStatus, VcirAuditSummary,
+        VcirCcrManifestProjection, VcirChildEntry, VcirInstanceGate, VcirRelatedArtifact,
+        VcirSummary,
+    };
+    use sha2::Digest;
+
+    fn sample_trust_anchor() -> TrustAnchor {
+        let base = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+        let tal_bytes = std::fs::read(base.join("tests/fixtures/tal/apnic-rfc7730-https.tal"))
+            .expect("read tal");
+        let ta_der = std::fs::read(base.join("tests/fixtures/ta/apnic-ta.cer")).expect("read ta");
+        let tal = Tal::decode_bytes(&tal_bytes).expect("decode tal");
+        TrustAnchor::bind_der(tal, &ta_der, None).expect("bind ta")
+    }
+
+    fn sample_vcir_and_manifest(
+        store: &RocksStore,
+    ) -> (
+        ValidatedCaInstanceResult,
+        Vec<Vrp>,
+        Vec<AspaAttestation>,
+        Vec<RouterKeyPayload>,
+    ) {
+        let base = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+        let manifest_der = std::fs::read(
+            base.join(
+                "tests/fixtures/repository/rpki.cernet.net/repo/cernet/0/05FC9C5B88506F7C0D3F862C8895BED67E9F8EBA.mft",
+            ),
+        )
+        .expect("read manifest");
+        let manifest = ManifestObject::decode_der(&manifest_der).expect("decode manifest");
+        let manifest_hash = hex::encode(sha2::Sha256::digest(&manifest_der));
+        let mut raw = RawByHashEntry::from_bytes(manifest_hash.clone(), manifest_der.clone());
+        raw.origin_uris
+            .push("rsync://example.test/repo/current.mft".to_string());
+        raw.object_type = Some("mft".to_string());
+        raw.encoding = Some("der".to_string());
+        store.put_raw_by_hash_entry(&raw).expect("put raw");
+
+        let projection = VcirCcrManifestProjection {
+            manifest_rsync_uri: "rsync://example.test/repo/current.mft".to_string(),
+            manifest_sha256: sha2::Sha256::digest(&manifest_der).to_vec(),
+            manifest_size: manifest_der.len() as u64,
+            manifest_ee_aki: manifest.signed_object.signed_data.certificates[0]
+                .resource_cert
+                .tbs
+                .extensions
+                .authority_key_identifier
+                .clone()
+                .expect("manifest aki"),
+            manifest_number_be: manifest.manifest.manifest_number.bytes_be.clone(),
+            manifest_this_update: PackTime::from_utc_offset_datetime(manifest.manifest.this_update),
+            manifest_sia_locations_der: match manifest.signed_object.signed_data.certificates[0]
+                .resource_cert
+                .tbs
+                .extensions
+                .subject_info_access
+                .as_ref()
+                .expect("manifest sia")
+            {
+                SubjectInfoAccess::Ee(ee_sia) => ee_sia
+                    .access_descriptions
+                    .iter()
+                    .map(|ad| {
+                        let oid = encode_oid_for_test(&ad.access_method_oid)
+                            .expect("encode access method oid");
+                        let uri = encode_tlv_for_test(0x86, ad.access_location.as_bytes().to_vec());
+                        encode_sequence_for_test(&[oid, uri])
+                    })
+                    .collect(),
+                SubjectInfoAccess::Ca(_) => panic!("manifest ee sia should not be CA variant"),
+            },
+            subordinate_skis: vec![vec![0x33; 20]],
+        };
+
+        let vcir = ValidatedCaInstanceResult {
+            manifest_rsync_uri: "rsync://example.test/repo/current.mft".to_string(),
+            parent_manifest_rsync_uri: None,
+            tal_id: "apnic".to_string(),
+            ca_subject_name: "CN=test".to_string(),
+            ca_ski: "11".repeat(20),
+            issuer_ski: "22".repeat(20),
+            last_successful_validation_time: PackTime::from_utc_offset_datetime(
+                manifest.manifest.this_update,
+            ),
+            current_manifest_rsync_uri: "rsync://example.test/repo/current.mft".to_string(),
+            current_crl_rsync_uri: "rsync://example.test/repo/current.crl".to_string(),
+            validated_manifest_meta: ValidatedManifestMeta {
+                validated_manifest_number: manifest.manifest.manifest_number.bytes_be.clone(),
+                validated_manifest_this_update: PackTime::from_utc_offset_datetime(
+                    manifest.manifest.this_update,
+                ),
+                validated_manifest_next_update: PackTime::from_utc_offset_datetime(
+                    manifest.manifest.next_update,
+                ),
+            },
+            ccr_manifest_projection: projection.clone(),
+            instance_gate: VcirInstanceGate {
+                manifest_next_update: PackTime::from_utc_offset_datetime(
+                    manifest.manifest.next_update,
+                ),
+                current_crl_next_update: PackTime::from_utc_offset_datetime(
+                    manifest.manifest.next_update,
+                ),
+                self_ca_not_after: PackTime::from_utc_offset_datetime(
+                    manifest.manifest.next_update,
+                ),
+                instance_effective_until: PackTime::from_utc_offset_datetime(
+                    manifest.manifest.next_update,
+                ),
+            },
+            child_entries: vec![VcirChildEntry {
+                child_manifest_rsync_uri: "rsync://example.test/repo/child.mft".to_string(),
+                child_cert_rsync_uri: "rsync://example.test/repo/child.cer".to_string(),
+                child_cert_hash: "aa".repeat(32),
+                child_ski: "33".repeat(20),
+                child_rsync_base_uri: "rsync://example.test/repo/".to_string(),
+                child_publication_point_rsync_uri: "rsync://example.test/repo/".to_string(),
+                child_rrdp_notification_uri: None,
+                child_effective_ip_resources: None,
+                child_effective_as_resources: None,
+                accepted_at_validation_time: PackTime::from_utc_offset_datetime(
+                    manifest.manifest.this_update,
+                ),
+            }],
+            local_outputs: Vec::new(),
+            related_artifacts: vec![VcirRelatedArtifact {
+                artifact_role: VcirArtifactRole::Manifest,
+                artifact_kind: VcirArtifactKind::Mft,
+                uri: Some("rsync://example.test/repo/current.mft".to_string()),
+                sha256: manifest_hash,
+                object_type: Some("mft".to_string()),
+                validation_status: VcirArtifactValidationStatus::Accepted,
+            }],
+            summary: VcirSummary {
+                local_vrp_count: 0,
+                local_aspa_count: 0,
+                local_router_key_count: 0,
+                child_count: 1,
+                accepted_object_count: 1,
+                rejected_object_count: 0,
+            },
+            audit_summary: VcirAuditSummary {
+                failed_fetch_eligible: true,
+                last_failed_fetch_reason: None,
+                warning_count: 0,
+                audit_flags: Vec::new(),
+            },
+        };
+
+        let vrps = vec![Vrp {
+            asn: 64496,
+            prefix: IpPrefix {
+                afi: RoaAfi::Ipv4,
+                prefix_len: 8,
+                addr: [10, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            },
+            max_length: 8,
+        }];
+        let aspas = vec![AspaAttestation {
+            customer_as_id: 64496,
+            provider_as_ids: vec![64497],
+        }];
+        let router_keys = vec![RouterKeyPayload {
+            as_id: 64496,
+            ski: vec![0x11; 20],
+            spki_der: vec![0x30, 0x00],
+            source_object_uri: "rsync://example.test/repo/router.cer".to_string(),
+            source_object_hash: hex::encode([0x11; 32]),
+            source_ee_cert_hash: hex::encode([0x11; 32]),
+            item_effective_until: PackTime::from_utc_offset_datetime(
+                time::OffsetDateTime::now_utc() + time::Duration::hours(1),
+            ),
+        }];
+        (vcir, vrps, aspas, router_keys)
+    }
+
+    fn encode_oid_for_test(oid: &str) -> Result<Vec<u8>, String> {
+        let arcs = oid
+            .split('.')
+            .map(|part| part.parse::<u64>().map_err(|_| format!("bad oid: {oid}")))
+            .collect::<Result<Vec<_>, _>>()?;
+        if arcs.len() < 2 {
+            return Err(format!("bad oid: {oid}"));
+        }
+        let mut body = Vec::new();
+        body.push((arcs[0] * 40 + arcs[1]) as u8);
+        for arc in &arcs[2..] {
+            encode_base128_for_test(*arc, &mut body);
+        }
+        Ok(encode_tlv_for_test(0x06, body))
+    }
+
+    fn encode_base128_for_test(mut value: u64, out: &mut Vec<u8>) {
+        let mut tmp = vec![(value & 0x7F) as u8];
+        value >>= 7;
+        while value > 0 {
+            tmp.push(((value & 0x7F) as u8) | 0x80);
+            value >>= 7;
+        }
+        tmp.reverse();
+        out.extend_from_slice(&tmp);
+    }
+
+    fn encode_sequence_for_test(elements: &[Vec<u8>]) -> Vec<u8> {
+        let total_len: usize = elements.iter().map(Vec::len).sum();
+        let mut buf = Vec::with_capacity(total_len);
+        for element in elements {
+            buf.extend_from_slice(element);
+        }
+        encode_tlv_for_test(0x30, buf)
+    }
+
+    fn encode_tlv_for_test(tag: u8, value: Vec<u8>) -> Vec<u8> {
+        let mut out = Vec::with_capacity(1 + 9 + value.len());
+        out.push(tag);
+        if value.len() < 0x80 {
+            out.push(value.len() as u8);
+        } else {
+            out.push(0x81);
+            out.push(value.len() as u8);
+        }
+        out.extend_from_slice(&value);
+        out
+    }
+
+    #[test]
+    fn accumulator_finish_matches_builder_on_fresh_vcir_inputs() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let store = RocksStore::open(td.path()).expect("open rocksdb");
+        let (vcir, vrps, aspas, router_keys) = sample_vcir_and_manifest(&store);
+        store.put_vcir(&vcir).expect("put vcir");
+        let trust_anchor = sample_trust_anchor();
+
+        let builder_ccr = build_ccr_from_run(
+            &store,
+            &[trust_anchor.clone()],
+            &vrps,
+            &aspas,
+            &router_keys,
+            time::OffsetDateTime::now_utc(),
+        )
+        .expect("build ccr from run");
+
+        let mut accumulator = CcrAccumulator::new(vec![trust_anchor]);
+        accumulator
+            .append_manifest_projection(&vcir.ccr_manifest_projection)
+            .expect("append manifest projection");
+        let accumulated_ccr = accumulator
+            .finish(time::OffsetDateTime::now_utc(), &vrps, &aspas, &router_keys)
+            .expect("finish accumulator");
+
+        assert_eq!(builder_ccr.mfts, accumulated_ccr.mfts);
+        assert_eq!(builder_ccr.vrps, accumulated_ccr.vrps);
+        assert_eq!(builder_ccr.vaps, accumulated_ccr.vaps);
+        assert_eq!(builder_ccr.tas, accumulated_ccr.tas);
+        assert_eq!(builder_ccr.rks, accumulated_ccr.rks);
+        let ci = crate::ccr::model::CcrContentInfo::new(accumulated_ccr);
+        verify_content_info(&ci).expect("verify accumulated ccr");
+    }
+}