将fetch pp cache改成使用vcir结构，跑通apnic全量同步

src/audit_downloads.rs

@@ -39,14 +39,17 @@ impl DownloadLogHandle {
             }
             .to_string();
 
-            let st = out.by_kind.entry(kind_key).or_insert_with(|| AuditDownloadKindStats {
-                ok_total: 0,
-                fail_total: 0,
-                duration_ms_total: 0,
-                bytes_total: None,
-                objects_count_total: None,
-                objects_bytes_total: None,
-            });
+            let st = out
+                .by_kind
+                .entry(kind_key)
+                .or_insert_with(|| AuditDownloadKindStats {
+                    ok_total: 0,
+                    fail_total: 0,
+                    duration_ms_total: 0,
+                    bytes_total: None,
+                    objects_count_total: None,
+                    objects_bytes_total: None,
+                });
             if e.success {
                 st.ok_total = st.ok_total.saturating_add(1);
             } else {
@@ -165,4 +168,3 @@ fn duration_to_ms(d: Duration) -> u64 {
     let ms = d.as_millis();
     ms.min(u128::from(u64::MAX)) as u64
 }
-

src/audit_trace.rs

@@ -0,0 +1,581 @@
+use crate::storage::{
+    AuditRuleIndexEntry, AuditRuleKind, RawByHashEntry, RocksStore, ValidatedCaInstanceResult,
+    VcirArtifactKind, VcirArtifactRole, VcirArtifactValidationStatus, VcirLocalOutput,
+    VcirOutputType,
+};
+use serde::Serialize;
+use std::collections::HashSet;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AuditTraceError {
+    #[error("storage error: {0}")]
+    Storage(#[from] crate::storage::StorageError),
+
+    #[error("audit rule index points to missing VCIR: {manifest_rsync_uri}")]
+    MissingVcir { manifest_rsync_uri: String },
+
+    #[error(
+        "audit rule index points to missing local output: rule_hash={rule_hash}, output_id={output_id}, manifest={manifest_rsync_uri}"
+    )]
+    MissingLocalOutput {
+        rule_hash: String,
+        output_id: String,
+        manifest_rsync_uri: String,
+    },
+
+    #[error("detected VCIR parent cycle at {manifest_rsync_uri}")]
+    ParentCycle { manifest_rsync_uri: String },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+pub struct AuditTraceRawRef {
+    pub sha256_hex: String,
+    pub raw_present: bool,
+    #[serde(skip_serializing_if = "Vec::is_empty", default)]
+    pub origin_uris: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub object_type: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub byte_len: Option<usize>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+pub struct AuditTraceArtifact {
+    pub artifact_role: VcirArtifactRole,
+    pub artifact_kind: VcirArtifactKind,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub uri: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub object_type: Option<String>,
+    pub validation_status: VcirArtifactValidationStatus,
+    pub raw: AuditTraceRawRef,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+pub struct AuditTraceChainNode {
+    pub manifest_rsync_uri: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub parent_manifest_rsync_uri: Option<String>,
+    pub tal_id: String,
+    pub ca_subject_name: String,
+    pub ca_ski: String,
+    pub issuer_ski: String,
+    pub current_manifest_rsync_uri: String,
+    pub current_crl_rsync_uri: String,
+    pub last_successful_validation_time_rfc3339_utc: String,
+    pub local_output_count: usize,
+    pub child_count: usize,
+    pub related_artifacts: Vec<AuditTraceArtifact>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+pub struct AuditTraceResolvedOutput {
+    pub output_id: String,
+    pub output_type: VcirOutputType,
+    pub rule_hash: String,
+    pub source_object_uri: String,
+    pub source_object_type: String,
+    pub source_object_hash: String,
+    pub source_ee_cert_hash: String,
+    pub item_effective_until_rfc3339_utc: String,
+    pub payload_json: String,
+    pub validation_path_hint: Vec<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+pub struct AuditRuleTrace {
+    pub rule: AuditRuleIndexEntry,
+    pub resolved_output: AuditTraceResolvedOutput,
+    pub source_object_raw: AuditTraceRawRef,
+    pub source_ee_cert_raw: AuditTraceRawRef,
+    pub chain_leaf_to_root: Vec<AuditTraceChainNode>,
+}
+
+pub fn trace_rule_to_root(
+    store: &RocksStore,
+    kind: AuditRuleKind,
+    rule_hash: &str,
+) -> Result<Option<AuditRuleTrace>, AuditTraceError> {
+    let Some(rule) = store.get_audit_rule_index_entry(kind, rule_hash)? else {
+        return Ok(None);
+    };
+
+    let Some(leaf_vcir) = store.get_vcir(&rule.manifest_rsync_uri)? else {
+        return Err(AuditTraceError::MissingVcir {
+            manifest_rsync_uri: rule.manifest_rsync_uri.clone(),
+        });
+    };
+
+    let Some(local_output) = leaf_vcir
+        .local_outputs
+        .iter()
+        .find(|output| output.output_id == rule.output_id && output.rule_hash == rule.rule_hash)
+        .or_else(|| {
+            leaf_vcir
+                .local_outputs
+                .iter()
+                .find(|output| output.rule_hash == rule.rule_hash)
+        })
+        .cloned()
+    else {
+        return Err(AuditTraceError::MissingLocalOutput {
+            rule_hash: rule.rule_hash.clone(),
+            output_id: rule.output_id.clone(),
+            manifest_rsync_uri: rule.manifest_rsync_uri.clone(),
+        });
+    };
+
+    let chain = trace_vcir_chain_to_root(store, &leaf_vcir.manifest_rsync_uri)?
+        .expect("leaf VCIR already loaded must exist");
+
+    Ok(Some(AuditRuleTrace {
+        rule,
+        resolved_output: resolved_output_from_local(&local_output),
+        source_object_raw: resolve_raw_ref(store, &local_output.source_object_hash)?,
+        source_ee_cert_raw: resolve_raw_ref(store, &local_output.source_ee_cert_hash)?,
+        chain_leaf_to_root: chain,
+    }))
+}
+
+pub fn trace_vcir_chain_to_root(
+    store: &RocksStore,
+    manifest_rsync_uri: &str,
+) -> Result<Option<Vec<AuditTraceChainNode>>, AuditTraceError> {
+    let Some(mut current) = store.get_vcir(manifest_rsync_uri)? else {
+        return Ok(None);
+    };
+
+    let mut seen = HashSet::new();
+    let mut chain = Vec::new();
+    loop {
+        if !seen.insert(current.manifest_rsync_uri.clone()) {
+            return Err(AuditTraceError::ParentCycle {
+                manifest_rsync_uri: current.manifest_rsync_uri,
+            });
+        }
+        let parent = current.parent_manifest_rsync_uri.clone();
+        chain.push(trace_chain_node(store, &current)?);
+        let Some(parent_manifest_rsync_uri) = parent else {
+            break;
+        };
+        let Some(parent_vcir) = store.get_vcir(&parent_manifest_rsync_uri)? else {
+            return Err(AuditTraceError::MissingVcir {
+                manifest_rsync_uri: parent_manifest_rsync_uri,
+            });
+        };
+        current = parent_vcir;
+    }
+
+    Ok(Some(chain))
+}
+
+fn trace_chain_node(
+    store: &RocksStore,
+    vcir: &ValidatedCaInstanceResult,
+) -> Result<AuditTraceChainNode, AuditTraceError> {
+    let mut related_artifacts = Vec::with_capacity(vcir.related_artifacts.len());
+    for artifact in &vcir.related_artifacts {
+        related_artifacts.push(AuditTraceArtifact {
+            artifact_role: artifact.artifact_role,
+            artifact_kind: artifact.artifact_kind,
+            uri: artifact.uri.clone(),
+            object_type: artifact.object_type.clone(),
+            validation_status: artifact.validation_status,
+            raw: resolve_raw_ref(store, &artifact.sha256)?,
+        });
+    }
+
+    Ok(AuditTraceChainNode {
+        manifest_rsync_uri: vcir.manifest_rsync_uri.clone(),
+        parent_manifest_rsync_uri: vcir.parent_manifest_rsync_uri.clone(),
+        tal_id: vcir.tal_id.clone(),
+        ca_subject_name: vcir.ca_subject_name.clone(),
+        ca_ski: vcir.ca_ski.clone(),
+        issuer_ski: vcir.issuer_ski.clone(),
+        current_manifest_rsync_uri: vcir.current_manifest_rsync_uri.clone(),
+        current_crl_rsync_uri: vcir.current_crl_rsync_uri.clone(),
+        last_successful_validation_time_rfc3339_utc: vcir
+            .last_successful_validation_time
+            .rfc3339_utc
+            .clone(),
+        local_output_count: vcir.local_outputs.len(),
+        child_count: vcir.child_entries.len(),
+        related_artifacts,
+    })
+}
+
+fn resolved_output_from_local(local: &VcirLocalOutput) -> AuditTraceResolvedOutput {
+    AuditTraceResolvedOutput {
+        output_id: local.output_id.clone(),
+        output_type: local.output_type,
+        rule_hash: local.rule_hash.clone(),
+        source_object_uri: local.source_object_uri.clone(),
+        source_object_type: local.source_object_type.clone(),
+        source_object_hash: local.source_object_hash.clone(),
+        source_ee_cert_hash: local.source_ee_cert_hash.clone(),
+        item_effective_until_rfc3339_utc: local.item_effective_until.rfc3339_utc.clone(),
+        payload_json: local.payload_json.clone(),
+        validation_path_hint: local.validation_path_hint.clone(),
+    }
+}
+
+fn resolve_raw_ref(
+    store: &RocksStore,
+    sha256_hex: &str,
+) -> Result<AuditTraceRawRef, AuditTraceError> {
+    let raw = store.get_raw_by_hash_entry(sha256_hex)?;
+    Ok(raw_ref_from_entry(sha256_hex, raw.as_ref()))
+}
+
+fn raw_ref_from_entry(sha256_hex: &str, entry: Option<&RawByHashEntry>) -> AuditTraceRawRef {
+    match entry {
+        Some(entry) => AuditTraceRawRef {
+            sha256_hex: sha256_hex.to_string(),
+            raw_present: true,
+            origin_uris: entry.origin_uris.clone(),
+            object_type: entry.object_type.clone(),
+            byte_len: Some(entry.bytes.len()),
+        },
+        None => AuditTraceRawRef {
+            sha256_hex: sha256_hex.to_string(),
+            raw_present: false,
+            origin_uris: Vec::new(),
+            object_type: None,
+            byte_len: None,
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::audit::sha256_hex;
+    use crate::storage::{
+        PackTime, ValidatedManifestMeta, VcirAuditSummary, VcirChildEntry, VcirInstanceGate,
+        VcirRelatedArtifact, VcirSummary,
+    };
+
+    fn sample_vcir(
+        manifest_rsync_uri: &str,
+        parent_manifest_rsync_uri: Option<&str>,
+        tal_id: &str,
+        local_output: Option<VcirLocalOutput>,
+        related_artifacts: Vec<VcirRelatedArtifact>,
+    ) -> ValidatedCaInstanceResult {
+        let now = time::OffsetDateTime::now_utc();
+        let next = PackTime::from_utc_offset_datetime(now + time::Duration::hours(1));
+        let local_outputs: Vec<VcirLocalOutput> = local_output.into_iter().collect();
+        ValidatedCaInstanceResult {
+            manifest_rsync_uri: manifest_rsync_uri.to_string(),
+            parent_manifest_rsync_uri: parent_manifest_rsync_uri.map(str::to_string),
+            tal_id: tal_id.to_string(),
+            ca_subject_name: format!("CN={manifest_rsync_uri}"),
+            ca_ski: "11".repeat(20),
+            issuer_ski: "22".repeat(20),
+            last_successful_validation_time: PackTime::from_utc_offset_datetime(now),
+            current_manifest_rsync_uri: manifest_rsync_uri.to_string(),
+            current_crl_rsync_uri: manifest_rsync_uri.replace(".mft", ".crl"),
+            validated_manifest_meta: ValidatedManifestMeta {
+                validated_manifest_number: vec![1],
+                validated_manifest_this_update: PackTime::from_utc_offset_datetime(now),
+                validated_manifest_next_update: next.clone(),
+            },
+            instance_gate: VcirInstanceGate {
+                manifest_next_update: next.clone(),
+                current_crl_next_update: next.clone(),
+                self_ca_not_after: PackTime::from_utc_offset_datetime(
+                    now + time::Duration::hours(2),
+                ),
+                instance_effective_until: next,
+            },
+            child_entries: vec![VcirChildEntry {
+                child_manifest_rsync_uri: "rsync://example.test/child/child.mft".to_string(),
+                child_cert_rsync_uri: "rsync://example.test/parent/child.cer".to_string(),
+                child_cert_hash: sha256_hex(b"child-cert"),
+                child_ski: "33".repeat(20),
+                child_rsync_base_uri: "rsync://example.test/child/".to_string(),
+                child_publication_point_rsync_uri: "rsync://example.test/child/".to_string(),
+                child_rrdp_notification_uri: Some(
+                    "https://example.test/child/notify.xml".to_string(),
+                ),
+                child_effective_ip_resources: None,
+                child_effective_as_resources: None,
+                accepted_at_validation_time: PackTime::from_utc_offset_datetime(now),
+            }],
+            summary: VcirSummary {
+                local_vrp_count: local_outputs
+                    .iter()
+                    .filter(|output| output.output_type == VcirOutputType::Vrp)
+                    .count() as u32,
+                local_aspa_count: local_outputs
+                    .iter()
+                    .filter(|output| output.output_type == VcirOutputType::Aspa)
+                    .count() as u32,
+                child_count: 1,
+                accepted_object_count: related_artifacts.len() as u32,
+                rejected_object_count: 0,
+            },
+            local_outputs,
+            related_artifacts,
+            audit_summary: VcirAuditSummary {
+                failed_fetch_eligible: true,
+                last_failed_fetch_reason: None,
+                warning_count: 0,
+                audit_flags: Vec::new(),
+            },
+        }
+    }
+
+    fn sample_local_output(manifest_rsync_uri: &str) -> VcirLocalOutput {
+        let now = time::OffsetDateTime::now_utc();
+        VcirLocalOutput {
+            output_id: sha256_hex(b"vrp-output"),
+            output_type: VcirOutputType::Vrp,
+            item_effective_until: PackTime::from_utc_offset_datetime(
+                now + time::Duration::minutes(30),
+            ),
+            source_object_uri: "rsync://example.test/leaf/a.roa".to_string(),
+            source_object_type: "roa".to_string(),
+            source_object_hash: sha256_hex(b"roa-raw"),
+            source_ee_cert_hash: sha256_hex(b"roa-ee"),
+            payload_json:
+                serde_json::json!({"asn": 64496, "prefix": "203.0.113.0/24", "max_length": 24})
+                    .to_string(),
+            rule_hash: sha256_hex(b"roa-rule"),
+            validation_path_hint: vec![
+                manifest_rsync_uri.to_string(),
+                "rsync://example.test/leaf/a.roa".to_string(),
+                sha256_hex(b"roa-raw"),
+            ],
+        }
+    }
+
+    fn sample_artifacts(manifest_rsync_uri: &str, roa_hash: &str) -> Vec<VcirRelatedArtifact> {
+        vec![
+            VcirRelatedArtifact {
+                artifact_role: VcirArtifactRole::Manifest,
+                artifact_kind: VcirArtifactKind::Mft,
+                uri: Some(manifest_rsync_uri.to_string()),
+                sha256: sha256_hex(manifest_rsync_uri.as_bytes()),
+                object_type: Some("mft".to_string()),
+                validation_status: VcirArtifactValidationStatus::Accepted,
+            },
+            VcirRelatedArtifact {
+                artifact_role: VcirArtifactRole::CurrentCrl,
+                artifact_kind: VcirArtifactKind::Crl,
+                uri: Some(manifest_rsync_uri.replace(".mft", ".crl")),
+                sha256: sha256_hex(format!("{}-crl", manifest_rsync_uri).as_bytes()),
+                object_type: Some("crl".to_string()),
+                validation_status: VcirArtifactValidationStatus::Accepted,
+            },
+            VcirRelatedArtifact {
+                artifact_role: VcirArtifactRole::SignedObject,
+                artifact_kind: VcirArtifactKind::Roa,
+                uri: Some("rsync://example.test/leaf/a.roa".to_string()),
+                sha256: roa_hash.to_string(),
+                object_type: Some("roa".to_string()),
+                validation_status: VcirArtifactValidationStatus::Accepted,
+            },
+        ]
+    }
+
+    fn put_raw(store: &RocksStore, bytes: &[u8], uri: &str, object_type: &str) {
+        let mut entry = RawByHashEntry::from_bytes(sha256_hex(bytes), bytes.to_vec());
+        entry.origin_uris.push(uri.to_string());
+        entry.object_type = Some(object_type.to_string());
+        entry.encoding = Some("der".to_string());
+        store
+            .put_raw_by_hash_entry(&entry)
+            .expect("put raw evidence");
+    }
+
+    #[test]
+    fn trace_rule_to_root_returns_leaf_to_root_chain_and_evidence_refs() {
+        let store_dir = tempfile::tempdir().expect("store dir");
+        let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
+
+        let root_manifest = "rsync://example.test/root/root.mft";
+        let leaf_manifest = "rsync://example.test/leaf/leaf.mft";
+        let local = sample_local_output(leaf_manifest);
+        let leaf_vcir = sample_vcir(
+            leaf_manifest,
+            Some(root_manifest),
+            "test-tal",
+            Some(local.clone()),
+            sample_artifacts(leaf_manifest, &local.source_object_hash),
+        );
+        let root_vcir = sample_vcir(
+            root_manifest,
+            None,
+            "test-tal",
+            None,
+            sample_artifacts(root_manifest, &sha256_hex(b"root-object")),
+        );
+        store.put_vcir(&leaf_vcir).expect("put leaf vcir");
+        store.put_vcir(&root_vcir).expect("put root vcir");
+
+        let rule_entry = AuditRuleIndexEntry {
+            kind: AuditRuleKind::Roa,
+            rule_hash: local.rule_hash.clone(),
+            manifest_rsync_uri: leaf_manifest.to_string(),
+            source_object_uri: local.source_object_uri.clone(),
+            source_object_hash: local.source_object_hash.clone(),
+            output_id: local.output_id.clone(),
+            item_effective_until: local.item_effective_until.clone(),
+        };
+        store
+            .put_audit_rule_index_entry(&rule_entry)
+            .expect("put rule index");
+
+        put_raw(&store, leaf_manifest.as_bytes(), leaf_manifest, "mft");
+        put_raw(
+            &store,
+            format!("{}-crl", leaf_manifest).as_bytes(),
+            &leaf_manifest.replace(".mft", ".crl"),
+            "crl",
+        );
+        put_raw(&store, b"roa-raw", &local.source_object_uri, "roa");
+        put_raw(&store, b"roa-ee", "rsync://example.test/leaf/a.ee", "cer");
+        put_raw(&store, root_manifest.as_bytes(), root_manifest, "mft");
+        put_raw(
+            &store,
+            format!("{}-crl", root_manifest).as_bytes(),
+            &root_manifest.replace(".mft", ".crl"),
+            "crl",
+        );
+
+        let trace = trace_rule_to_root(&store, AuditRuleKind::Roa, &local.rule_hash)
+            .expect("trace rule")
+            .expect("trace exists");
+
+        assert_eq!(trace.rule, rule_entry);
+        assert_eq!(trace.resolved_output.output_id, local.output_id);
+        assert_eq!(trace.chain_leaf_to_root.len(), 2);
+        assert_eq!(
+            trace.chain_leaf_to_root[0].manifest_rsync_uri,
+            leaf_manifest
+        );
+        assert_eq!(
+            trace.chain_leaf_to_root[1].manifest_rsync_uri,
+            root_manifest
+        );
+        assert_eq!(
+            trace.chain_leaf_to_root[0]
+                .parent_manifest_rsync_uri
+                .as_deref(),
+            Some(root_manifest)
+        );
+        assert!(trace.source_object_raw.raw_present);
+        assert!(trace.source_ee_cert_raw.raw_present);
+        assert!(
+            trace.chain_leaf_to_root[0]
+                .related_artifacts
+                .iter()
+                .any(|artifact| {
+                    artifact.uri.as_deref() == Some(leaf_manifest) && artifact.raw.raw_present
+                })
+        );
+    }
+
+    #[test]
+    fn trace_rule_to_root_returns_none_for_missing_rule_index() {
+        let store_dir = tempfile::tempdir().expect("store dir");
+        let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
+
+        assert!(
+            trace_rule_to_root(&store, AuditRuleKind::Roa, &sha256_hex(b"missing"))
+                .expect("missing trace ok")
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn trace_rule_to_root_errors_when_index_points_to_missing_vcir() {
+        let store_dir = tempfile::tempdir().expect("store dir");
+        let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
+        let rule_hash = sha256_hex(b"missing-vcir-rule");
+        store
+            .put_audit_rule_index_entry(&AuditRuleIndexEntry {
+                kind: AuditRuleKind::Roa,
+                rule_hash: rule_hash.clone(),
+                manifest_rsync_uri: "rsync://example.test/missing.mft".to_string(),
+                source_object_uri: "rsync://example.test/missing.roa".to_string(),
+                source_object_hash: sha256_hex(b"missing-source"),
+                output_id: sha256_hex(b"missing-output"),
+                item_effective_until: PackTime::from_utc_offset_datetime(
+                    time::OffsetDateTime::now_utc() + time::Duration::minutes(1),
+                ),
+            })
+            .expect("put rule index");
+
+        let err = trace_rule_to_root(&store, AuditRuleKind::Roa, &rule_hash).unwrap_err();
+        assert!(matches!(
+            err,
+            AuditTraceError::MissingVcir { manifest_rsync_uri }
+            if manifest_rsync_uri == "rsync://example.test/missing.mft"
+        ));
+    }
+
+    #[test]
+    fn trace_rule_to_root_errors_when_vcir_local_output_is_missing() {
+        let store_dir = tempfile::tempdir().expect("store dir");
+        let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
+        let manifest = "rsync://example.test/leaf/leaf.mft";
+        let vcir = sample_vcir(
+            manifest,
+            None,
+            "test-tal",
+            None,
+            sample_artifacts(manifest, &sha256_hex(b"leaf-object")),
+        );
+        store.put_vcir(&vcir).expect("put vcir");
+        let rule_hash = sha256_hex(b"missing-output-rule");
+        store
+            .put_audit_rule_index_entry(&AuditRuleIndexEntry {
+                kind: AuditRuleKind::Roa,
+                rule_hash: rule_hash.clone(),
+                manifest_rsync_uri: manifest.to_string(),
+                source_object_uri: "rsync://example.test/leaf/a.roa".to_string(),
+                source_object_hash: sha256_hex(b"leaf-object"),
+                output_id: sha256_hex(b"missing-output"),
+                item_effective_until: PackTime::from_utc_offset_datetime(
+                    time::OffsetDateTime::now_utc() + time::Duration::minutes(1),
+                ),
+            })
+            .expect("put rule index");
+
+        let err = trace_rule_to_root(&store, AuditRuleKind::Roa, &rule_hash).unwrap_err();
+        assert!(matches!(err, AuditTraceError::MissingLocalOutput { .. }));
+    }
+
+    #[test]
+    fn trace_vcir_chain_to_root_detects_parent_cycle() {
+        let store_dir = tempfile::tempdir().expect("store dir");
+        let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
+        let a_manifest = "rsync://example.test/a.mft";
+        let b_manifest = "rsync://example.test/b.mft";
+        let a_vcir = sample_vcir(
+            a_manifest,
+            Some(b_manifest),
+            "test-tal",
+            None,
+            sample_artifacts(a_manifest, &sha256_hex(b"a-object")),
+        );
+        let b_vcir = sample_vcir(
+            b_manifest,
+            Some(a_manifest),
+            "test-tal",
+            None,
+            sample_artifacts(b_manifest, &sha256_hex(b"b-object")),
+        );
+        store.put_vcir(&a_vcir).expect("put a");
+        store.put_vcir(&b_vcir).expect("put b");
+
+        let err = trace_vcir_chain_to_root(&store, a_manifest).unwrap_err();
+        assert!(matches!(
+            err,
+            AuditTraceError::ParentCycle { manifest_rsync_uri }
+            if manifest_rsync_uri == a_manifest
+        ));
+    }
+}

src/bin/db_stats.rs

@@ -1,21 +1,36 @@
+use std::collections::BTreeMap;
 use std::path::PathBuf;
 
-use rocksdb::{ColumnFamilyDescriptor, DB, IteratorMode, Options};
+use rocksdb::{DB, IteratorMode, Options};
+use rpki::storage::{
+    ALL_COLUMN_FAMILY_NAMES, CF_AUDIT_RULE_INDEX, CF_RAW_BY_HASH, CF_RAW_OBJECTS,
+    CF_REPOSITORY_VIEW, CF_RRDP_OBJECT_INDEX, CF_RRDP_SOURCE, CF_RRDP_SOURCE_MEMBER, CF_RRDP_STATE,
+    CF_RRDP_URI_OWNER, CF_VCIR, column_family_descriptors,
+};
 
-const CF_RAW_OBJECTS: &str = "raw_objects";
-const CF_FETCH_CACHE_PP: &str = "fetch_cache_pp";
-const CF_RRDP_STATE: &str = "rrdp_state";
-const CF_RRDP_OBJECT_INDEX: &str = "rrdp_object_index";
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum DbStatsMode {
+    Estimate,
+    Exact,
+}
 
-fn enable_blobdb_if_supported(opts: &mut Options) {
-    // Keep this in sync with `rpki::storage`:
-    // blob files are CF-level options; readers should open CFs with blob enabled too.
-    #[allow(dead_code)]
-    fn _set(opts: &mut Options) {
-        opts.set_enable_blob_files(true);
-        opts.set_min_blob_size(1024);
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+enum CfGroup {
+    CurrentRepositoryView,
+    CurrentValidationState,
+    CurrentRrdpState,
+    LegacyCompatibility,
+}
+
+impl CfGroup {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::CurrentRepositoryView => "current_repository_view",
+            Self::CurrentValidationState => "current_validation_state",
+            Self::CurrentRrdpState => "current_rrdp_state",
+            Self::LegacyCompatibility => "legacy_compatibility",
+        }
     }
-    _set(opts);
 }
 
 fn usage() -> String {
@@ -29,21 +44,16 @@ Options:
   --db <path>     RocksDB directory
   --exact         Iterate to count keys (slower; default uses RocksDB estimates)
   --help          Show this help
+
+Output groups:
+  - current_repository_view: repository_view + raw_by_hash
+  - current_validation_state: vcir + audit_rule_index
+  - current_rrdp_state: rrdp_state + rrdp_source + rrdp_source_member + rrdp_uri_owner
+  - legacy_compatibility: raw_objects + rrdp_object_index
 "
     )
 }
 
-fn cf_descriptors() -> Vec<ColumnFamilyDescriptor> {
-    let mut cf_opts = Options::default();
-    enable_blobdb_if_supported(&mut cf_opts);
-    vec![
-        ColumnFamilyDescriptor::new(CF_RAW_OBJECTS, cf_opts.clone()),
-        ColumnFamilyDescriptor::new(CF_FETCH_CACHE_PP, cf_opts.clone()),
-        ColumnFamilyDescriptor::new(CF_RRDP_STATE, cf_opts.clone()),
-        ColumnFamilyDescriptor::new(CF_RRDP_OBJECT_INDEX, cf_opts),
-    ]
-}
-
 fn estimate_keys(db: &DB, cf_name: &str) -> Result<Option<u64>, Box<dyn std::error::Error>> {
     let cf = db
         .cf_handle(cf_name)
@@ -64,6 +74,35 @@ fn exact_keys(db: &DB, cf_name: &str) -> Result<u64, Box<dyn std::error::Error>>
     Ok(count)
 }
 
+fn cf_group(cf_name: &str) -> CfGroup {
+    match cf_name {
+        CF_REPOSITORY_VIEW | CF_RAW_BY_HASH => CfGroup::CurrentRepositoryView,
+        CF_VCIR | CF_AUDIT_RULE_INDEX => CfGroup::CurrentValidationState,
+        CF_RRDP_STATE | CF_RRDP_SOURCE | CF_RRDP_SOURCE_MEMBER | CF_RRDP_URI_OWNER => {
+            CfGroup::CurrentRrdpState
+        }
+        CF_RAW_OBJECTS | CF_RRDP_OBJECT_INDEX => CfGroup::LegacyCompatibility,
+        _ => CfGroup::LegacyCompatibility,
+    }
+}
+
+fn summarize_counts<'a>(
+    counts: impl IntoIterator<Item = (&'a str, u64)>,
+) -> BTreeMap<CfGroup, u64> {
+    let mut grouped = BTreeMap::new();
+    for (cf_name, count) in counts {
+        *grouped.entry(cf_group(cf_name)).or_insert(0) += count;
+    }
+    grouped
+}
+
+fn mode_label(mode: DbStatsMode) -> &'static str {
+    match mode {
+        DbStatsMode::Estimate => "estimate",
+        DbStatsMode::Exact => "exact",
+    }
+}
+
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let argv: Vec<String> = std::env::args().collect();
     if argv.iter().any(|a| a == "--help" || a == "-h") {
@@ -72,7 +111,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     }
 
     let mut db_path: Option<PathBuf> = None;
-    let mut exact = false;
+    let mut mode = DbStatsMode::Estimate;
     let mut i = 1usize;
     while i < argv.len() {
         match argv[i].as_str() {
@@ -81,7 +120,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                 let v = argv.get(i).ok_or("--db requires a value")?;
                 db_path = Some(PathBuf::from(v));
             }
-            "--exact" => exact = true,
+            "--exact" => mode = DbStatsMode::Exact,
             other => return Err(format!("unknown argument: {other}\n\n{}", usage()).into()),
         }
         i += 1;
@@ -93,33 +132,75 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     opts.create_if_missing(false);
     opts.create_missing_column_families(false);
 
-    let db = DB::open_cf_descriptors(&opts, &db_path, cf_descriptors())?;
-
-    let cf_names = [
-        CF_RAW_OBJECTS,
-        CF_FETCH_CACHE_PP,
-        CF_RRDP_STATE,
-        CF_RRDP_OBJECT_INDEX,
-    ];
+    let db = DB::open_cf_descriptors(&opts, &db_path, column_family_descriptors())?;
 
     println!("db={}", db_path.display());
-    println!("mode={}", if exact { "exact" } else { "estimate" });
+    println!("mode={}", mode_label(mode));
 
+    let mut per_cf = Vec::with_capacity(ALL_COLUMN_FAMILY_NAMES.len());
     let mut total: u64 = 0;
-    for name in cf_names {
-        let n = if exact {
-            exact_keys(&db, name)?
-        } else {
-            estimate_keys(&db, name)?.unwrap_or(0)
+    for &name in ALL_COLUMN_FAMILY_NAMES {
+        let n = match mode {
+            DbStatsMode::Exact => exact_keys(&db, name)?,
+            DbStatsMode::Estimate => estimate_keys(&db, name)?.unwrap_or(0),
         };
         total = total.saturating_add(n);
+        per_cf.push((name, n));
         println!("{name}={n}");
     }
     println!("total={total}");
 
-    // Also print # of SST files (useful sanity signal).
+    for (group, count) in summarize_counts(per_cf.iter().copied()) {
+        println!("group_{}={count}", group.as_str());
+    }
+
     let live = db.live_files()?;
     println!("sst_files={}", live.len());
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cf_group_classifies_current_and_legacy_keyspaces() {
+        assert_eq!(cf_group(CF_REPOSITORY_VIEW), CfGroup::CurrentRepositoryView);
+        assert_eq!(cf_group(CF_RAW_BY_HASH), CfGroup::CurrentRepositoryView);
+        assert_eq!(cf_group(CF_VCIR), CfGroup::CurrentValidationState);
+        assert_eq!(
+            cf_group(CF_AUDIT_RULE_INDEX),
+            CfGroup::CurrentValidationState
+        );
+        assert_eq!(cf_group(CF_RRDP_SOURCE), CfGroup::CurrentRrdpState);
+        assert_eq!(cf_group(CF_RRDP_URI_OWNER), CfGroup::CurrentRrdpState);
+        assert_eq!(cf_group(CF_RAW_OBJECTS), CfGroup::LegacyCompatibility);
+    }
+
+    #[test]
+    fn summarize_counts_accumulates_by_group() {
+        let grouped = summarize_counts([
+            (CF_REPOSITORY_VIEW, 5),
+            (CF_RAW_BY_HASH, 7),
+            (CF_VCIR, 11),
+            (CF_AUDIT_RULE_INDEX, 13),
+            (CF_RRDP_STATE, 17),
+            (CF_RRDP_SOURCE_MEMBER, 19),
+            (CF_RRDP_OBJECT_INDEX, 29),
+        ]);
+
+        assert_eq!(grouped.get(&CfGroup::CurrentRepositoryView), Some(&12));
+        assert_eq!(grouped.get(&CfGroup::CurrentValidationState), Some(&24));
+        assert_eq!(grouped.get(&CfGroup::CurrentRrdpState), Some(&36));
+        assert_eq!(grouped.get(&CfGroup::LegacyCompatibility), Some(&29));
+    }
+
+    #[test]
+    fn usage_mentions_grouped_output_and_exact_mode() {
+        let text = usage();
+        assert!(text.contains("--exact"), "{text}");
+        assert!(text.contains("current_validation_state"), "{text}");
+        assert!(text.contains("legacy_compatibility"), "{text}");
+    }
+}

src/bin/rrdp_state_dump.rs

@@ -1,16 +1,40 @@
 use std::path::PathBuf;
 
-use rocksdb::{ColumnFamilyDescriptor, DB, IteratorMode, Options};
+use rocksdb::{DB, IteratorMode, Options};
+use rpki::storage::{
+    CF_RRDP_SOURCE, CF_RRDP_SOURCE_MEMBER, CF_RRDP_STATE, CF_RRDP_URI_OWNER,
+    RrdpSourceMemberRecord, RrdpSourceRecord, RrdpUriOwnerRecord, column_family_descriptors,
+};
+use rpki::sync::rrdp::RrdpState;
 
-fn enable_blobdb_if_supported(opts: &mut Options) {
-    // Keep this in sync with `rpki::storage`:
-    // blob files are CF-level options; readers should open CFs with blob enabled too.
-    #[allow(dead_code)]
-    fn _set(opts: &mut Options) {
-        opts.set_enable_blob_files(true);
-        opts.set_min_blob_size(1024);
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum DumpView {
+    LegacyState,
+    Source,
+    Members,
+    Owners,
+    All,
+}
+
+impl DumpView {
+    fn parse(value: &str) -> Result<Self, String> {
+        match value {
+            "legacy-state" => Ok(Self::LegacyState),
+            "source" => Ok(Self::Source),
+            "members" => Ok(Self::Members),
+            "owners" => Ok(Self::Owners),
+            "all" => Ok(Self::All),
+            other => Err(format!(
+                "invalid --view: {other} (expected one of: legacy-state, source, members, owners, all)"
+            )),
+        }
     }
-    _set(opts);
+}
+
+#[derive(Debug)]
+struct DumpArgs {
+    db_path: PathBuf,
+    view: DumpView,
 }
 
 fn usage() -> String {
@@ -18,23 +42,23 @@ fn usage() -> String {
     format!(
         "\
 Usage:
-  {bin} --db <path>
+  {bin} --db <path> [--view <legacy-state|source|members|owners|all>]
 
 Options:
   --db <path>     RocksDB directory
+  --view <name>   Dump one RRDP view; default is all
   --help          Show this help
 "
     )
 }
 
-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let argv: Vec<String> = std::env::args().collect();
+fn parse_args(argv: &[String]) -> Result<DumpArgs, String> {
     if argv.iter().any(|a| a == "--help" || a == "-h") {
-        print!("{}", usage());
-        return Ok(());
+        return Err(usage());
     }
 
     let mut db_path: Option<PathBuf> = None;
+    let mut view = DumpView::All;
     let mut i = 1usize;
     while i < argv.len() {
         match argv[i].as_str() {
@@ -43,43 +67,280 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                 let v = argv.get(i).ok_or("--db requires a value")?;
                 db_path = Some(PathBuf::from(v));
             }
-            other => return Err(format!("unknown argument: {other}\n\n{}", usage()).into()),
+            "--view" => {
+                i += 1;
+                let v = argv.get(i).ok_or("--view requires a value")?;
+                view = DumpView::parse(v)?;
+            }
+            other => return Err(format!("unknown argument: {other}\n\n{}", usage())),
         }
         i += 1;
     }
 
-    let db_path = db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?;
+    Ok(DumpArgs {
+        db_path: db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?,
+        view,
+    })
+}
 
+fn open_db(path: &std::path::Path) -> Result<DB, Box<dyn std::error::Error>> {
     let mut opts = Options::default();
     opts.create_if_missing(false);
     opts.create_missing_column_families(false);
+    Ok(DB::open_cf_descriptors(
+        &opts,
+        path,
+        column_family_descriptors(),
+    )?)
+}
 
-    // Open only the column families we need.
-    let mut cf_opts = Options::default();
-    enable_blobdb_if_supported(&mut cf_opts);
-    let cfs = vec![
-        ColumnFamilyDescriptor::new("raw_objects", cf_opts.clone()),
-        ColumnFamilyDescriptor::new("fetch_cache_pp", cf_opts.clone()),
-        ColumnFamilyDescriptor::new("rrdp_state", cf_opts.clone()),
-        ColumnFamilyDescriptor::new("rrdp_object_index", cf_opts),
-    ];
-    let db = DB::open_cf_descriptors(&opts, &db_path, cfs)?;
+fn collect_legacy_state(db: &DB) -> Result<Vec<(String, RrdpState)>, Box<dyn std::error::Error>> {
     let cf = db
-        .cf_handle("rrdp_state")
+        .cf_handle(CF_RRDP_STATE)
         .ok_or("missing column family: rrdp_state")?;
-
-    let mut out: Vec<(String, u64, String)> = Vec::new();
+    let mut out = Vec::new();
     for res in db.iterator_cf(cf, IteratorMode::Start) {
         let (k, v) = res?;
-        let k = String::from_utf8_lossy(&k).to_string();
-        let st = rpki::sync::rrdp::RrdpState::decode(&v)
-            .map_err(|e| format!("decode rrdp_state failed for {k}: {e}"))?;
-        out.push((k, st.serial, st.session_id));
+        let notify_uri = String::from_utf8_lossy(&k).to_string();
+        let state = RrdpState::decode(&v)
+            .map_err(|e| format!("decode rrdp_state failed for {notify_uri}: {e}"))?;
+        out.push((notify_uri, state));
     }
-
     out.sort_by(|a, b| a.0.cmp(&b.0));
-    for (k, serial, session) in out {
-        println!("{k}\t{serial}\t{session}");
+    Ok(out)
+}
+
+fn collect_source_records(db: &DB) -> Result<Vec<RrdpSourceRecord>, Box<dyn std::error::Error>> {
+    let cf = db
+        .cf_handle(CF_RRDP_SOURCE)
+        .ok_or("missing column family: rrdp_source")?;
+    let mut out = Vec::new();
+    for res in db.iterator_cf(cf, IteratorMode::Start) {
+        let (_k, v) = res?;
+        let record: RrdpSourceRecord = serde_cbor::from_slice(&v)?;
+        record.validate_internal()?;
+        out.push(record);
+    }
+    out.sort_by(|a, b| a.notify_uri.cmp(&b.notify_uri));
+    Ok(out)
+}
+
+fn collect_source_member_records(
+    db: &DB,
+) -> Result<Vec<RrdpSourceMemberRecord>, Box<dyn std::error::Error>> {
+    let cf = db
+        .cf_handle(CF_RRDP_SOURCE_MEMBER)
+        .ok_or("missing column family: rrdp_source_member")?;
+    let mut out = Vec::new();
+    for res in db.iterator_cf(cf, IteratorMode::Start) {
+        let (_k, v) = res?;
+        let record: RrdpSourceMemberRecord = serde_cbor::from_slice(&v)?;
+        record.validate_internal()?;
+        out.push(record);
+    }
+    out.sort_by(|a, b| {
+        a.notify_uri
+            .cmp(&b.notify_uri)
+            .then(a.rsync_uri.cmp(&b.rsync_uri))
+    });
+    Ok(out)
+}
+
+fn collect_uri_owner_records(
+    db: &DB,
+) -> Result<Vec<RrdpUriOwnerRecord>, Box<dyn std::error::Error>> {
+    let cf = db
+        .cf_handle(CF_RRDP_URI_OWNER)
+        .ok_or("missing column family: rrdp_uri_owner")?;
+    let mut out = Vec::new();
+    for res in db.iterator_cf(cf, IteratorMode::Start) {
+        let (_k, v) = res?;
+        let record: RrdpUriOwnerRecord = serde_cbor::from_slice(&v)?;
+        record.validate_internal()?;
+        out.push(record);
+    }
+    out.sort_by(|a, b| a.rsync_uri.cmp(&b.rsync_uri));
+    Ok(out)
+}
+
+fn print_legacy_state(entries: &[(String, RrdpState)]) {
+    println!("[legacy-state]");
+    println!("notify_uri\tserial\tsession_id");
+    for (notify_uri, state) in entries {
+        println!("{notify_uri}\t{}\t{}", state.serial, state.session_id);
+    }
+}
+
+fn print_source_records(entries: &[RrdpSourceRecord]) {
+    println!("[source]");
+    println!("notify_uri\tlast_serial\tlast_session_id\tsync_state\tlast_snapshot_uri\tlast_error");
+    for record in entries {
+        println!(
+            "{}\t{}\t{}\t{:?}\t{}\t{}",
+            record.notify_uri,
+            record
+                .last_serial
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "-".to_string()),
+            record.last_session_id.as_deref().unwrap_or("-"),
+            record.sync_state,
+            record.last_snapshot_uri.as_deref().unwrap_or("-"),
+            record.last_error.as_deref().unwrap_or("-"),
+        );
+    }
+}
+
+fn print_source_member_records(entries: &[RrdpSourceMemberRecord]) {
+    println!("[members]");
+    println!("notify_uri\trsync_uri\tpresent\thash\tsession_id\tserial");
+    for record in entries {
+        println!(
+            "{}\t{}\t{}\t{}\t{}\t{}",
+            record.notify_uri,
+            record.rsync_uri,
+            record.present,
+            record.current_hash.as_deref().unwrap_or("-"),
+            record.last_confirmed_session_id,
+            record.last_confirmed_serial,
+        );
+    }
+}
+
+fn print_uri_owner_records(entries: &[RrdpUriOwnerRecord]) {
+    println!("[owners]");
+    println!("rsync_uri\tnotify_uri\towner_state\thash\tsession_id\tserial");
+    for record in entries {
+        println!(
+            "{}\t{}\t{:?}\t{}\t{}\t{}",
+            record.rsync_uri,
+            record.notify_uri,
+            record.owner_state,
+            record.current_hash.as_deref().unwrap_or("-"),
+            record.last_confirmed_session_id,
+            record.last_confirmed_serial,
+        );
+    }
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let argv: Vec<String> = std::env::args().collect();
+    let args = parse_args(&argv).map_err(|e| -> Box<dyn std::error::Error> { e.into() })?;
+    let db = open_db(&args.db_path)?;
+
+    match args.view {
+        DumpView::LegacyState => print_legacy_state(&collect_legacy_state(&db)?),
+        DumpView::Source => print_source_records(&collect_source_records(&db)?),
+        DumpView::Members => print_source_member_records(&collect_source_member_records(&db)?),
+        DumpView::Owners => print_uri_owner_records(&collect_uri_owner_records(&db)?),
+        DumpView::All => {
+            print_legacy_state(&collect_legacy_state(&db)?);
+            print_source_records(&collect_source_records(&db)?);
+            print_source_member_records(&collect_source_member_records(&db)?);
+            print_uri_owner_records(&collect_uri_owner_records(&db)?);
+        }
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rpki::storage::{PackTime, RocksStore, RrdpSourceSyncState, RrdpUriOwnerState};
+
+    #[test]
+    fn parse_args_accepts_view_and_db() {
+        let args = parse_args(&[
+            "rrdp_state_dump".to_string(),
+            "--db".to_string(),
+            "db".to_string(),
+            "--view".to_string(),
+            "owners".to_string(),
+        ])
+        .expect("parse args");
+        assert_eq!(args.db_path, PathBuf::from("db"));
+        assert_eq!(args.view, DumpView::Owners);
+    }
+
+    #[test]
+    fn parse_args_rejects_invalid_view() {
+        let err = parse_args(&[
+            "rrdp_state_dump".to_string(),
+            "--db".to_string(),
+            "db".to_string(),
+            "--view".to_string(),
+            "nope".to_string(),
+        ])
+        .unwrap_err();
+        assert!(err.contains("invalid --view"), "{err}");
+    }
+
+    #[test]
+    fn collect_rrdp_views_reads_legacy_and_current_records() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = RocksStore::open(dir.path()).expect("open store");
+        let legacy_state = RrdpState {
+            session_id: "session-1".to_string(),
+            serial: 42,
+        };
+        store
+            .put_rrdp_state(
+                "https://example.test/notify.xml",
+                &legacy_state.encode().expect("encode legacy state"),
+            )
+            .expect("put legacy state");
+        store
+            .put_rrdp_source_record(&RrdpSourceRecord {
+                notify_uri: "https://example.test/notify.xml".to_string(),
+                last_session_id: Some("session-1".to_string()),
+                last_serial: Some(42),
+                first_seen_at: PackTime::from_utc_offset_datetime(time::OffsetDateTime::now_utc()),
+                last_seen_at: PackTime::from_utc_offset_datetime(time::OffsetDateTime::now_utc()),
+                last_sync_at: None,
+                sync_state: RrdpSourceSyncState::DeltaReady,
+                last_snapshot_uri: Some("https://example.test/snapshot.xml".to_string()),
+                last_snapshot_hash: None,
+                last_error: None,
+            })
+            .expect("put source record");
+        store
+            .put_rrdp_source_member_record(&RrdpSourceMemberRecord {
+                notify_uri: "https://example.test/notify.xml".to_string(),
+                rsync_uri: "rsync://example.test/repo/a.roa".to_string(),
+                current_hash: Some("11".repeat(32)),
+                object_type: Some("roa".to_string()),
+                present: true,
+                last_confirmed_session_id: "session-1".to_string(),
+                last_confirmed_serial: 42,
+                last_changed_at: PackTime::from_utc_offset_datetime(time::OffsetDateTime::now_utc()),
+            })
+            .expect("put member record");
+        store
+            .put_rrdp_uri_owner_record(&RrdpUriOwnerRecord {
+                rsync_uri: "rsync://example.test/repo/a.roa".to_string(),
+                notify_uri: "https://example.test/notify.xml".to_string(),
+                current_hash: Some("11".repeat(32)),
+                last_confirmed_session_id: "session-1".to_string(),
+                last_confirmed_serial: 42,
+                last_changed_at: PackTime::from_utc_offset_datetime(time::OffsetDateTime::now_utc()),
+                owner_state: RrdpUriOwnerState::Active,
+            })
+            .expect("put owner record");
+        drop(store);
+
+        let db = open_db(dir.path()).expect("open db");
+        let legacy = collect_legacy_state(&db).expect("legacy dump");
+        let sources = collect_source_records(&db).expect("source dump");
+        let members = collect_source_member_records(&db).expect("members dump");
+        let owners = collect_uri_owner_records(&db).expect("owners dump");
+
+        assert_eq!(legacy.len(), 1);
+        assert_eq!(legacy[0].0, "https://example.test/notify.xml");
+        assert_eq!(legacy[0].1.serial, 42);
+        assert_eq!(sources.len(), 1);
+        assert_eq!(sources[0].sync_state, RrdpSourceSyncState::DeltaReady);
+        assert_eq!(members.len(), 1);
+        assert_eq!(members[0].rsync_uri, "rsync://example.test/repo/a.roa");
+        assert_eq!(owners.len(), 1);
+        assert_eq!(owners[0].owner_state, RrdpUriOwnerState::Active);
+    }
+}

src/cli.rs

@@ -37,7 +37,6 @@ pub struct CliArgs {
     pub max_instances: Option<usize>,
     pub validation_time: Option<time::OffsetDateTime>,
 
-    pub revalidate_only: bool,
     pub analyze: bool,
     pub profile_cpu: bool,
 }
@@ -66,7 +65,6 @@ Options:
   --max-depth <n>            Max CA instance depth (0 = root only)
   --max-instances <n>        Max number of CA instances to process
   --validation-time <rfc3339> Validation time in RFC3339 (default: now UTC)
-  --revalidate-only          Skip RRDP/rsync sync; re-validate from existing DB cache
   --analyze                  Write timing analysis JSON under target/live/analyze/<timestamp>/
   --profile-cpu              (Requires build feature 'profile') Write CPU flamegraph under analyze dir
 
@@ -91,7 +89,6 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
     let mut max_depth: Option<usize> = None;
     let mut max_instances: Option<usize> = None;
     let mut validation_time: Option<time::OffsetDateTime> = None;
-    let mut revalidate_only: bool = false;
     let mut analyze: bool = false;
     let mut profile_cpu: bool = false;
 
@@ -178,9 +175,6 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
                     .map_err(|e| format!("invalid --validation-time (RFC3339 expected): {e}"))?;
                 validation_time = Some(t);
             }
-            "--revalidate-only" => {
-                revalidate_only = true;
-            }
             "--analyze" => {
                 analyze = true;
             }
@@ -222,7 +216,6 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
         max_depth,
         max_instances,
         validation_time,
-        revalidate_only,
         analyze,
         profile_cpu,
     })
@@ -356,7 +349,6 @@ pub fn run(argv: &[String]) -> Result<(), String> {
     let config = TreeRunConfig {
         max_depth: args.max_depth,
         max_instances: args.max_instances,
-        revalidate_only: args.revalidate_only,
     };
 
     use time::format_description::well_known::Rfc3339;
@@ -804,6 +796,20 @@ mod tests {
         assert!(args.validation_time.is_some());
     }
 
+    #[test]
+    fn parse_rejects_removed_revalidate_only_flag() {
+        let argv = vec![
+            "rpki".to_string(),
+            "--db".to_string(),
+            "db".to_string(),
+            "--tal-url".to_string(),
+            "https://example.test/x.tal".to_string(),
+            "--revalidate-only".to_string(),
+        ];
+        let err = parse_args(&argv).unwrap_err();
+        assert!(err.contains("unknown argument: --revalidate-only"), "{err}");
+    }
+
     #[test]
     fn read_policy_accepts_valid_toml() {
         let dir = tempfile::tempdir().expect("tmpdir");

src/data_model/rc.rs

@@ -5,6 +5,8 @@ use x509_parser::asn1_rs::{Class as Asn1Class, Tag as Asn1Tag};
 use x509_parser::extensions::ParsedExtension;
 use x509_parser::prelude::{FromDer, X509Certificate, X509Extension, X509Version};
 
+use serde::{Deserialize, Serialize};
+
 use crate::data_model::common::{
     Asn1TimeUtc, InvalidTimeEncodingError, UtcTime, X509NameDer, asn1_time_to_model,
 };
@@ -183,7 +185,7 @@ pub struct AccessDescription {
     pub access_location: String,
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
 pub enum Afi {
     Ipv4,
     Ipv6,
@@ -205,7 +207,7 @@ impl Afi {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct IpResourceSet {
     pub families: Vec<IpAddressFamily>,
 }
@@ -240,7 +242,7 @@ pub enum IpResourceSetDecodeError {
     InvalidEncoding,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct IpAddressFamily {
     pub afi: Afi,
     pub choice: IpAddressChoice,
@@ -261,32 +263,32 @@ impl IpAddressFamily {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum IpAddressChoice {
     Inherit,
     AddressesOrRanges(Vec<IpAddressOrRange>),
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum IpAddressOrRange {
     Prefix(IpPrefix),
     Range(IpAddressRange),
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct IpAddressRange {
     pub min: Vec<u8>,
     pub max: Vec<u8>,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub struct IpPrefix {
     pub afi: Afi,
     pub prefix_len: u16,
     pub addr: Vec<u8>,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct AsResourceSet {
     pub asnum: Option<AsIdentifierChoice>,
     pub rdi: Option<AsIdentifierChoice>,
@@ -330,7 +332,7 @@ pub enum AsResourceSetDecodeError {
     InvalidEncoding,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum AsIdentifierChoice {
     Inherit,
     AsIdsOrRanges(Vec<AsIdOrRange>),
@@ -347,7 +349,7 @@ impl AsIdentifierChoice {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum AsIdOrRange {
     Id(u32),
     Range { min: u32, max: u32 },

src/fetch/rsync_system.rs

@@ -52,12 +52,8 @@ impl SystemRsyncFetcher {
             return Ok(None);
         };
 
-        std::fs::create_dir_all(root).map_err(|e| {
-            format!(
-                "create rsync mirror root failed: {}: {e}",
-                root.display()
-            )
-        })?;
+        std::fs::create_dir_all(root)
+            .map_err(|e| format!("create rsync mirror root failed: {}: {e}", root.display()))?;
 
         let hash = hex::encode(sha2::Sha256::digest(normalized_rsync_base_uri.as_bytes()));
         let dir = root.join(hash);
@@ -107,7 +103,8 @@ impl RsyncFetcher for SystemRsyncFetcher {
             .mirror_dst_dir(&base)
             .map_err(|e| RsyncFetchError::Fetch(e.to_string()))?
         {
-            self.run_rsync(&base, &dst).map_err(RsyncFetchError::Fetch)?;
+            self.run_rsync(&base, &dst)
+                .map_err(RsyncFetchError::Fetch)?;
             let mut out = Vec::new();
             walk_dir_collect(&dst, &dst, &base, &mut out).map_err(RsyncFetchError::Fetch)?;
             return Ok(out);
@@ -249,4 +246,74 @@ mod tests {
         assert!(e.contains("rsync failed:"), "{e}");
         assert!(e.contains("status="), "{e}");
     }
+
+    #[test]
+    fn mirror_dst_dir_reports_root_creation_error() {
+        let temp = tempfile::tempdir().expect("tempdir");
+        let root_file = temp.path().join("mirror-root-file");
+        std::fs::write(&root_file, b"not a directory").expect("write root file");
+
+        let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
+            rsync_bin: PathBuf::from("rsync"),
+            timeout: Duration::from_secs(1),
+            extra_args: Vec::new(),
+            mirror_root: Some(root_file.clone()),
+        });
+
+        let err = fetcher
+            .mirror_dst_dir("rsync://example.net/repo/")
+            .expect_err("file mirror root must fail");
+        assert!(err.contains("create rsync mirror root failed"), "{err}");
+        assert!(err.contains(&root_file.display().to_string()), "{err}");
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn mirror_dst_dir_reports_directory_creation_error_inside_root() {
+        use std::os::unix::fs::PermissionsExt;
+
+        let temp = tempfile::tempdir().expect("tempdir");
+        let root = temp.path().join("mirror");
+        std::fs::create_dir_all(&root).expect("mkdir root");
+        let mut perms = std::fs::metadata(&root).expect("metadata").permissions();
+        perms.set_mode(0o555);
+        std::fs::set_permissions(&root, perms).expect("chmod root readonly");
+
+        let fetcher = SystemRsyncFetcher::new(SystemRsyncConfig {
+            rsync_bin: PathBuf::from("rsync"),
+            timeout: Duration::from_secs(1),
+            extra_args: Vec::new(),
+            mirror_root: Some(root.clone()),
+        });
+
+        let err = fetcher
+            .mirror_dst_dir("rsync://example.net/repo/")
+            .expect_err("readonly mirror root must fail");
+        assert!(
+            err.contains("create rsync mirror directory failed"),
+            "{err}"
+        );
+
+        let mut perms = std::fs::metadata(&root).expect("metadata").permissions();
+        perms.set_mode(0o755);
+        std::fs::set_permissions(&root, perms).expect("restore perms");
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn walk_dir_collect_ignores_non_file_entries() {
+        use std::os::unix::net::UnixListener;
+
+        let temp = tempfile::tempdir().expect("tempdir");
+        let root = temp.path();
+        std::fs::write(root.join("a.cer"), b"x").expect("write file");
+        let socket_path = root.join("skip.sock");
+        let _listener = UnixListener::bind(&socket_path).expect("bind socket");
+
+        let mut out: Vec<(String, Vec<u8>)> = Vec::new();
+        walk_dir_collect(root, root, "rsync://example.net/repo/", &mut out).expect("walk");
+
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].0, "rsync://example.net/repo/a.cer");
+    }
 }

src/lib.rs

@@ -7,6 +7,8 @@ pub mod audit;
 #[cfg(feature = "full")]
 pub mod audit_downloads;
 #[cfg(feature = "full")]
+pub mod audit_trace;
+#[cfg(feature = "full")]
 pub mod cli;
 #[cfg(feature = "full")]
 pub mod fetch;

src/policy.rs

@@ -16,13 +16,13 @@ impl Default for SyncPreference {
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum CaFailedFetchPolicy {
-    UseFetchCachePp,
+    ReuseCurrentInstanceVcir,
     StopAllOutput,
 }
 
 impl Default for CaFailedFetchPolicy {
     fn default() -> Self {
-        Self::UseFetchCachePp
+        Self::ReuseCurrentInstanceVcir
     }
 }
 

src/sync/mod.rs

@@ -1,2 +1,3 @@
 pub mod repo;
 pub mod rrdp;
+pub(crate) mod store_projection;

src/sync/repo.rs

@@ -1,12 +1,16 @@
 use crate::analysis::timing::TimingHandle;
+use crate::audit::AuditDownloadKind;
 use crate::audit_downloads::DownloadLogHandle;
-use crate::audit::{AuditDownloadKind};
 use crate::fetch::rsync::{RsyncFetchError, RsyncFetcher};
 use crate::policy::{Policy, SyncPreference};
 use crate::report::{RfcRef, Warning};
 use crate::storage::RocksStore;
 use crate::sync::rrdp::sync_from_notification_with_timing_and_download_log;
 use crate::sync::rrdp::{Fetcher as HttpFetcher, RrdpSyncError};
+use crate::sync::store_projection::{
+    put_repository_view_present, put_repository_view_withdrawn, upsert_raw_by_hash_evidence,
+};
+use std::collections::HashSet;
 use std::thread;
 use std::time::Duration;
 
@@ -143,9 +147,8 @@ fn try_rrdp_sync(
         let _total = timing
             .as_ref()
             .map(|t| t.span_phase("rrdp_fetch_notification_total"));
-        let mut dl_span = download_log.map(|dl| {
-            dl.span_download(AuditDownloadKind::RrdpNotification, notification_uri)
-        });
+        let mut dl_span = download_log
+            .map(|dl| dl.span_download(AuditDownloadKind::RrdpNotification, notification_uri));
         match http_fetcher.fetch(notification_uri) {
             Ok(v) => {
                 if let Some(t) = timing.as_ref() {
@@ -307,12 +310,43 @@ fn rsync_sync_into_raw_objects(
         t.record_count("rsync_objects_bytes_total", bytes_total);
     }
     drop(_p);
+
+    let existing_view = store
+        .list_repository_view_entries_with_prefix(rsync_base_uri)
+        .map_err(|e| RepoSyncError::Storage(e.to_string()))?;
+    let new_set: HashSet<&str> = objects.iter().map(|(uri, _)| uri.as_str()).collect();
+
     let _w = timing
         .as_ref()
         .map(|t| t.span_phase("rsync_write_raw_objects_total"));
     store
-        .put_raw_batch(objects)
-        .map_err(|e| RepoSyncError::Storage(e.to_string()))
+        .put_raw_batch(objects.clone())
+        .map_err(|e| RepoSyncError::Storage(e.to_string()))?;
+    drop(_w);
+
+    let _proj = timing
+        .as_ref()
+        .map(|t| t.span_phase("rsync_write_repository_view_total"));
+    for entry in existing_view {
+        if !new_set.contains(entry.rsync_uri.as_str()) {
+            put_repository_view_withdrawn(
+                store,
+                rsync_base_uri,
+                &entry.rsync_uri,
+                entry.current_hash,
+            )
+            .map_err(RepoSyncError::Storage)?;
+        }
+    }
+
+    for (uri, bytes) in &objects {
+        let current_hash =
+            upsert_raw_by_hash_evidence(store, uri, bytes).map_err(RepoSyncError::Storage)?;
+        put_repository_view_present(store, rsync_base_uri, uri, &current_hash)
+            .map_err(RepoSyncError::Storage)?;
+    }
+
+    Ok(objects.len())
 }
 
 #[cfg(test)]
@@ -458,6 +492,29 @@ mod tests {
             Some(b"cer".to_vec())
         );
 
+        let view = store
+            .get_repository_view_entry("rsync://example.test/repo/a.mft")
+            .expect("get repository view")
+            .expect("repository view entry present");
+        assert_eq!(
+            view.current_hash.as_deref(),
+            Some(hex::encode(sha2::Sha256::digest(b"mft")).as_str())
+        );
+        assert_eq!(
+            view.repository_source.as_deref(),
+            Some("rsync://example.test/repo/")
+        );
+
+        let raw = store
+            .get_raw_by_hash_entry(hex::encode(sha2::Sha256::digest(b"roa")).as_str())
+            .expect("get raw_by_hash")
+            .expect("raw_by_hash entry present");
+        assert!(
+            raw.origin_uris
+                .iter()
+                .any(|uri| uri == "rsync://example.test/repo/sub/b.roa")
+        );
+
         let timing_path = temp.path().join("timing.json");
         timing.write_json(&timing_path, 5).expect("write json");
         let v: serde_json::Value =
@@ -478,6 +535,71 @@ mod tests {
         );
     }
 
+    #[test]
+    fn rsync_second_sync_marks_missing_repository_view_entries_withdrawn() {
+        let temp = tempfile::tempdir().expect("tempdir");
+
+        let repo_dir = temp.path().join("repo");
+        std::fs::create_dir_all(repo_dir.join("sub")).expect("mkdir");
+        std::fs::write(repo_dir.join("a.mft"), b"mft-v1").expect("write a");
+        std::fs::write(repo_dir.join("sub").join("b.roa"), b"roa-v1").expect("write b");
+
+        let store_dir = temp.path().join("db");
+        let store = RocksStore::open(&store_dir).expect("open rocksdb");
+        let policy = Policy {
+            sync_preference: SyncPreference::RsyncOnly,
+            ..Policy::default()
+        };
+        let http = DummyHttpFetcher;
+        let rsync = LocalDirRsyncFetcher::new(&repo_dir);
+
+        sync_publication_point(
+            &store,
+            &policy,
+            None,
+            "rsync://example.test/repo/",
+            &http,
+            &rsync,
+            None,
+            None,
+        )
+        .expect("first sync ok");
+
+        std::fs::remove_file(repo_dir.join("sub").join("b.roa")).expect("remove b");
+        std::fs::write(repo_dir.join("c.crl"), b"crl-v2").expect("write c");
+
+        sync_publication_point(
+            &store,
+            &policy,
+            None,
+            "rsync://example.test/repo/",
+            &http,
+            &rsync,
+            None,
+            None,
+        )
+        .expect("second sync ok");
+
+        let withdrawn = store
+            .get_repository_view_entry("rsync://example.test/repo/sub/b.roa")
+            .expect("get withdrawn repo view")
+            .expect("withdrawn entry exists");
+        assert_eq!(
+            withdrawn.state,
+            crate::storage::RepositoryViewState::Withdrawn
+        );
+        assert_eq!(
+            withdrawn.repository_source.as_deref(),
+            Some("rsync://example.test/repo/")
+        );
+
+        let added = store
+            .get_repository_view_entry("rsync://example.test/repo/c.crl")
+            .expect("get added repo view")
+            .expect("added entry exists");
+        assert_eq!(added.state, crate::storage::RepositoryViewState::Present);
+    }
+
     #[test]
     fn rrdp_retry_succeeds_without_rsync_when_notification_fetch_is_transient() {
         let temp = tempfile::tempdir().expect("tempdir");
@@ -602,7 +724,9 @@ mod tests {
             Some(1)
         );
         assert_eq!(
-            counts.get("repo_sync_rrdp_ok_total").and_then(|v| v.as_u64()),
+            counts
+                .get("repo_sync_rrdp_ok_total")
+                .and_then(|v| v.as_u64()),
             Some(1)
         );
     }
@@ -681,7 +805,11 @@ mod tests {
         );
 
         let events = download_log.snapshot_events();
-        assert_eq!(events.len(), 3, "expected notification + snapshot + rsync fallback");
+        assert_eq!(
+            events.len(),
+            3,
+            "expected notification + snapshot + rsync fallback"
+        );
         assert_eq!(events[0].kind, AuditDownloadKind::RrdpNotification);
         assert!(events[0].success);
         assert_eq!(events[1].kind, AuditDownloadKind::RrdpSnapshot);

src/sync/rrdp.rs

@@ -1,8 +1,13 @@
 use crate::analysis::timing::TimingHandle;
 use crate::audit::AuditDownloadKind;
 use crate::audit_downloads::DownloadLogHandle;
-use crate::storage::RocksStore;
-use crate::storage::RrdpDeltaOp;
+use crate::storage::{RocksStore, RrdpDeltaOp, RrdpSourceSyncState};
+use crate::sync::store_projection::{
+    compute_sha256_hex, current_rrdp_owner_is, ensure_rrdp_uri_can_be_owned_by,
+    put_repository_view_present, put_repository_view_withdrawn, put_rrdp_source_member_present,
+    put_rrdp_source_member_withdrawn, put_rrdp_uri_owner_active, put_rrdp_uri_owner_withdrawn,
+    update_rrdp_source_record_on_success, upsert_raw_by_hash_evidence,
+};
 use base64::Engine;
 use serde::{Deserialize, Serialize};
 use sha2::Digest;
@@ -490,7 +495,9 @@ fn sync_from_notification_snapshot_inner(
     let _parse_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "parse_notification_snapshot"));
-    let _parse_total = timing.as_ref().map(|t| t.span_phase("rrdp_parse_notification_total"));
+    let _parse_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_parse_notification_total"));
     let notif = parse_notification_snapshot(notification_xml)?;
     drop(_parse_step);
     drop(_parse_total);
@@ -498,7 +505,9 @@ fn sync_from_notification_snapshot_inner(
     let _fetch_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "fetch_snapshot"));
-    let _fetch_total = timing.as_ref().map(|t| t.span_phase("rrdp_fetch_snapshot_total"));
+    let _fetch_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_fetch_snapshot_total"));
     let mut dl_span = download_log
         .map(|dl| dl.span_download(AuditDownloadKind::RrdpSnapshot, &notif.snapshot_uri));
     let snapshot_xml = match fetcher.fetch(&notif.snapshot_uri) {
@@ -529,7 +538,9 @@ fn sync_from_notification_snapshot_inner(
     let _hash_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "hash_snapshot"));
-    let _hash_total = timing.as_ref().map(|t| t.span_phase("rrdp_hash_snapshot_total"));
+    let _hash_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_hash_snapshot_total"));
     let computed = sha2::Sha256::digest(&snapshot_xml);
     if computed.as_slice() != notif.snapshot_hash_sha256.as_slice() {
         return Err(RrdpError::SnapshotHashMismatch.into());
@@ -540,7 +551,9 @@ fn sync_from_notification_snapshot_inner(
     let _apply_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "apply_snapshot"));
-    let _apply_total = timing.as_ref().map(|t| t.span_phase("rrdp_apply_snapshot_total"));
+    let _apply_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_apply_snapshot_total"));
     let published = apply_snapshot(
         store,
         notification_uri,
@@ -557,7 +570,9 @@ fn sync_from_notification_snapshot_inner(
     let _write_state_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "write_state"));
-    let _write_state_total = timing.as_ref().map(|t| t.span_phase("rrdp_write_state_total"));
+    let _write_state_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_write_state_total"));
     let state = RrdpState {
         session_id: notif.session_id.to_string(),
         serial: notif.serial,
@@ -566,6 +581,16 @@ fn sync_from_notification_snapshot_inner(
     store
         .put_rrdp_state(notification_uri, &bytes)
         .map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
+    update_rrdp_source_record_on_success(
+        store,
+        notification_uri,
+        notif.session_id.to_string().as_str(),
+        notif.serial,
+        RrdpSourceSyncState::SnapshotOnly,
+        Some(&notif.snapshot_uri),
+        Some(&hex::encode(notif.snapshot_hash_sha256)),
+    )
+    .map_err(RrdpSyncError::Storage)?;
     drop(_write_state_step);
     drop(_write_state_total);
 
@@ -578,7 +603,14 @@ pub fn sync_from_notification(
     notification_xml: &[u8],
     fetcher: &dyn Fetcher,
 ) -> RrdpSyncResult<usize> {
-    sync_from_notification_inner(store, notification_uri, notification_xml, fetcher, None, None)
+    sync_from_notification_inner(
+        store,
+        notification_uri,
+        notification_xml,
+        fetcher,
+        None,
+        None,
+    )
 }
 
 pub fn sync_from_notification_with_timing(
@@ -588,7 +620,14 @@ pub fn sync_from_notification_with_timing(
     fetcher: &dyn Fetcher,
     timing: Option<&TimingHandle>,
 ) -> RrdpSyncResult<usize> {
-    sync_from_notification_inner(store, notification_uri, notification_xml, fetcher, timing, None)
+    sync_from_notification_inner(
+        store,
+        notification_uri,
+        notification_xml,
+        fetcher,
+        timing,
+        None,
+    )
 }
 
 pub fn sync_from_notification_with_timing_and_download_log(
@@ -620,7 +659,9 @@ fn sync_from_notification_inner(
     let _parse_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "parse_notification"));
-    let _parse_total = timing.as_ref().map(|t| t.span_phase("rrdp_parse_notification_total"));
+    let _parse_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_parse_notification_total"));
     let notif = parse_notification(notification_xml)?;
     drop(_parse_step);
     drop(_parse_total);
@@ -634,7 +675,9 @@ fn sync_from_notification_inner(
     let _read_state_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "read_state"));
-    let _read_state_total = timing.as_ref().map(|t| t.span_phase("rrdp_read_state_total"));
+    let _read_state_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_read_state_total"));
     let state = store
         .get_rrdp_state(notification_uri)
         .map_err(|e| RrdpSyncError::Storage(e.to_string()))?
@@ -676,8 +719,9 @@ fn sync_from_notification_inner(
                 let _fetch_d_step = timing
                     .as_ref()
                     .map(|t| t.span_rrdp_repo_step(notification_uri, "fetch_deltas"));
-                let _fetch_d_total =
-                    timing.as_ref().map(|t| t.span_phase("rrdp_fetch_deltas_total"));
+                let _fetch_d_total = timing
+                    .as_ref()
+                    .map(|t| t.span_phase("rrdp_fetch_deltas_total"));
                 let mut fetched: Vec<(u64, [u8; 32], Vec<u8>)> =
                     Vec::with_capacity((want_last - want_first + 1) as usize);
                 let mut fetch_ok = true;
@@ -727,8 +771,9 @@ fn sync_from_notification_inner(
                     let _apply_d_step = timing
                         .as_ref()
                         .map(|t| t.span_rrdp_repo_step(notification_uri, "apply_deltas"));
-                    let _apply_d_total =
-                        timing.as_ref().map(|t| t.span_phase("rrdp_apply_deltas_total"));
+                    let _apply_d_total = timing
+                        .as_ref()
+                        .map(|t| t.span_phase("rrdp_apply_deltas_total"));
                     let mut applied_total = 0usize;
                     let mut ok = true;
                     for (serial, expected_hash, bytes) in &fetched {
@@ -754,8 +799,9 @@ fn sync_from_notification_inner(
                         let _write_state_step = timing
                             .as_ref()
                             .map(|t| t.span_rrdp_repo_step(notification_uri, "write_state"));
-                        let _write_state_total =
-                            timing.as_ref().map(|t| t.span_phase("rrdp_write_state_total"));
+                        let _write_state_total = timing
+                            .as_ref()
+                            .map(|t| t.span_phase("rrdp_write_state_total"));
                         let new_state = RrdpState {
                             session_id: notif.session_id.to_string(),
                             serial: notif.serial,
@@ -764,6 +810,16 @@ fn sync_from_notification_inner(
                         store
                             .put_rrdp_state(notification_uri, &bytes)
                             .map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
+                        update_rrdp_source_record_on_success(
+                            store,
+                            notification_uri,
+                            notif.session_id.to_string().as_str(),
+                            notif.serial,
+                            RrdpSourceSyncState::DeltaReady,
+                            Some(&notif.snapshot_uri),
+                            Some(&hex::encode(notif.snapshot_hash_sha256)),
+                        )
+                        .map_err(RrdpSyncError::Storage)?;
                         drop(_write_state_step);
                         drop(_write_state_total);
                         if let Some(t) = timing.as_ref() {
@@ -780,7 +836,9 @@ fn sync_from_notification_inner(
     let _fetch_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "fetch_snapshot"));
-    let _fetch_total = timing.as_ref().map(|t| t.span_phase("rrdp_fetch_snapshot_total"));
+    let _fetch_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_fetch_snapshot_total"));
     let mut dl_span = download_log
         .map(|dl| dl.span_download(AuditDownloadKind::RrdpSnapshot, &notif.snapshot_uri));
     let snapshot_xml = match fetcher.fetch(&notif.snapshot_uri) {
@@ -811,7 +869,9 @@ fn sync_from_notification_inner(
     let _hash_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "hash_snapshot"));
-    let _hash_total = timing.as_ref().map(|t| t.span_phase("rrdp_hash_snapshot_total"));
+    let _hash_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_hash_snapshot_total"));
     let computed = sha2::Sha256::digest(&snapshot_xml);
     if computed.as_slice() != notif.snapshot_hash_sha256.as_slice() {
         return Err(RrdpError::SnapshotHashMismatch.into());
@@ -822,7 +882,9 @@ fn sync_from_notification_inner(
     let _apply_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "apply_snapshot"));
-    let _apply_total = timing.as_ref().map(|t| t.span_phase("rrdp_apply_snapshot_total"));
+    let _apply_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_apply_snapshot_total"));
     let published = apply_snapshot(
         store,
         notification_uri,
@@ -839,7 +901,9 @@ fn sync_from_notification_inner(
     let _write_state_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "write_state"));
-    let _write_state_total = timing.as_ref().map(|t| t.span_phase("rrdp_write_state_total"));
+    let _write_state_total = timing
+        .as_ref()
+        .map(|t| t.span_phase("rrdp_write_state_total"));
     let new_state = RrdpState {
         session_id: notif.session_id.to_string(),
         serial: notif.serial,
@@ -848,6 +912,16 @@ fn sync_from_notification_inner(
     store
         .put_rrdp_state(notification_uri, &bytes)
         .map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
+    update_rrdp_source_record_on_success(
+        store,
+        notification_uri,
+        notif.session_id.to_string().as_str(),
+        notif.serial,
+        RrdpSourceSyncState::SnapshotOnly,
+        Some(&notif.snapshot_uri),
+        Some(&hex::encode(notif.snapshot_hash_sha256)),
+    )
+    .map_err(RrdpSyncError::Storage)?;
     drop(_write_state_step);
     drop(_write_state_total);
 
@@ -883,7 +957,20 @@ fn apply_delta(
         .into());
     }
 
+    enum DeltaProjectionEffect {
+        Upsert {
+            rsync_uri: String,
+            bytes: Vec<u8>,
+        },
+        Delete {
+            rsync_uri: String,
+            previous_hash: String,
+        },
+    }
+
+    let session_id = expected_session_id.to_string();
     let mut ops: Vec<RrdpDeltaOp> = Vec::with_capacity(delta.elements.len());
+    let mut projection: Vec<DeltaProjectionEffect> = Vec::with_capacity(delta.elements.len());
     for e in delta.elements {
         match e {
             DeltaElement::Publish {
@@ -897,6 +984,8 @@ fn apply_delta(
                 if !is_member {
                     return Err(RrdpError::DeltaTargetNotFromRepository { rsync_uri: uri }.into());
                 }
+                ensure_rrdp_uri_can_be_owned_by(store, notification_uri, uri.as_str())
+                    .map_err(RrdpSyncError::Storage)?;
                 let old_bytes = store
                     .get_raw(uri.as_str())
                     .map_err(|e| RrdpSyncError::Storage(e.to_string()))?
@@ -909,6 +998,10 @@ fn apply_delta(
                 }
 
                 ops.push(RrdpDeltaOp::Upsert {
+                    rsync_uri: uri.clone(),
+                    bytes: bytes.clone(),
+                });
+                projection.push(DeltaProjectionEffect::Upsert {
                     rsync_uri: uri,
                     bytes,
                 });
@@ -926,7 +1019,13 @@ fn apply_delta(
                         RrdpError::DeltaPublishWithoutHashForExisting { rsync_uri: uri }.into(),
                     );
                 }
+                ensure_rrdp_uri_can_be_owned_by(store, notification_uri, uri.as_str())
+                    .map_err(RrdpSyncError::Storage)?;
                 ops.push(RrdpDeltaOp::Upsert {
+                    rsync_uri: uri.clone(),
+                    bytes: bytes.clone(),
+                });
+                projection.push(DeltaProjectionEffect::Upsert {
                     rsync_uri: uri,
                     bytes,
                 });
@@ -938,6 +1037,8 @@ fn apply_delta(
                 if !is_member {
                     return Err(RrdpError::DeltaTargetNotFromRepository { rsync_uri: uri }.into());
                 }
+                ensure_rrdp_uri_can_be_owned_by(store, notification_uri, uri.as_str())
+                    .map_err(RrdpSyncError::Storage)?;
                 let old_bytes = store
                     .get_raw(uri.as_str())
                     .map_err(|e| RrdpSyncError::Storage(e.to_string()))?
@@ -948,14 +1049,86 @@ fn apply_delta(
                 if old_computed.as_slice() != hash_sha256.as_slice() {
                     return Err(RrdpError::DeltaTargetHashMismatch { rsync_uri: uri }.into());
                 }
-                ops.push(RrdpDeltaOp::Delete { rsync_uri: uri });
+                let previous_hash = hex::encode(old_computed);
+                ops.push(RrdpDeltaOp::Delete {
+                    rsync_uri: uri.clone(),
+                });
+                projection.push(DeltaProjectionEffect::Delete {
+                    rsync_uri: uri,
+                    previous_hash,
+                });
             }
         }
     }
 
     store
         .apply_rrdp_delta(notification_uri, ops.as_slice())
-        .map_err(|e| RrdpSyncError::Storage(e.to_string()))
+        .map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
+
+    for effect in projection {
+        match effect {
+            DeltaProjectionEffect::Upsert { rsync_uri, bytes } => {
+                let current_hash = upsert_raw_by_hash_evidence(store, &rsync_uri, &bytes)
+                    .map_err(RrdpSyncError::Storage)?;
+                put_repository_view_present(store, notification_uri, &rsync_uri, &current_hash)
+                    .map_err(RrdpSyncError::Storage)?;
+                put_rrdp_source_member_present(
+                    store,
+                    notification_uri,
+                    &session_id,
+                    expected_serial,
+                    &rsync_uri,
+                    &current_hash,
+                )
+                .map_err(RrdpSyncError::Storage)?;
+                put_rrdp_uri_owner_active(
+                    store,
+                    notification_uri,
+                    &session_id,
+                    expected_serial,
+                    &rsync_uri,
+                    &current_hash,
+                )
+                .map_err(RrdpSyncError::Storage)?;
+            }
+            DeltaProjectionEffect::Delete {
+                rsync_uri,
+                previous_hash,
+            } => {
+                put_rrdp_source_member_withdrawn(
+                    store,
+                    notification_uri,
+                    &session_id,
+                    expected_serial,
+                    &rsync_uri,
+                    Some(previous_hash.clone()),
+                )
+                .map_err(RrdpSyncError::Storage)?;
+                if current_rrdp_owner_is(store, notification_uri, &rsync_uri)
+                    .map_err(RrdpSyncError::Storage)?
+                {
+                    put_repository_view_withdrawn(
+                        store,
+                        notification_uri,
+                        &rsync_uri,
+                        Some(previous_hash.clone()),
+                    )
+                    .map_err(RrdpSyncError::Storage)?;
+                    put_rrdp_uri_owner_withdrawn(
+                        store,
+                        notification_uri,
+                        &session_id,
+                        expected_serial,
+                        &rsync_uri,
+                        Some(previous_hash),
+                    )
+                    .map_err(RrdpSyncError::Storage)?;
+                }
+            }
+        }
+    }
+
+    Ok(ops.len())
 }
 
 fn apply_snapshot(
@@ -1004,12 +1177,92 @@ fn apply_snapshot(
             .decode(content_b64.as_bytes())
             .map_err(|e| RrdpError::PublishBase64(e.to_string()))?;
 
+        ensure_rrdp_uri_can_be_owned_by(store, notification_uri, uri)
+            .map_err(RrdpSyncError::Storage)?;
         published.push((uri.to_string(), bytes));
     }
 
+    let previous_members: Vec<String> = store
+        .rrdp_object_index_iter(notification_uri)
+        .map_err(|e| RrdpSyncError::Storage(e.to_string()))?
+        .collect();
+    let new_set: std::collections::HashSet<&str> =
+        published.iter().map(|(uri, _)| uri.as_str()).collect();
+    let mut withdrawn: Vec<(String, Option<String>)> = Vec::new();
+    for old_uri in &previous_members {
+        if new_set.contains(old_uri.as_str()) {
+            continue;
+        }
+        let previous_hash = store
+            .get_repository_view_entry(old_uri)
+            .map_err(|e| RrdpSyncError::Storage(e.to_string()))?
+            .and_then(|entry| entry.current_hash)
+            .or_else(|| {
+                store
+                    .get_raw(old_uri)
+                    .ok()
+                    .flatten()
+                    .map(|bytes| compute_sha256_hex(&bytes))
+            });
+        withdrawn.push((old_uri.clone(), previous_hash));
+    }
+
     store
         .apply_rrdp_snapshot(notification_uri, published.as_slice())
-        .map_err(|e| RrdpSyncError::Storage(e.to_string()))
+        .map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
+
+    let session_id = expected_session_id.to_string();
+    for (uri, bytes) in &published {
+        let current_hash =
+            upsert_raw_by_hash_evidence(store, uri, bytes).map_err(RrdpSyncError::Storage)?;
+        put_repository_view_present(store, notification_uri, uri, &current_hash)
+            .map_err(RrdpSyncError::Storage)?;
+        put_rrdp_source_member_present(
+            store,
+            notification_uri,
+            &session_id,
+            expected_serial,
+            uri,
+            &current_hash,
+        )
+        .map_err(RrdpSyncError::Storage)?;
+        put_rrdp_uri_owner_active(
+            store,
+            notification_uri,
+            &session_id,
+            expected_serial,
+            uri,
+            &current_hash,
+        )
+        .map_err(RrdpSyncError::Storage)?;
+    }
+
+    for (uri, previous_hash) in withdrawn {
+        put_rrdp_source_member_withdrawn(
+            store,
+            notification_uri,
+            &session_id,
+            expected_serial,
+            &uri,
+            previous_hash.clone(),
+        )
+        .map_err(RrdpSyncError::Storage)?;
+        if current_rrdp_owner_is(store, notification_uri, &uri).map_err(RrdpSyncError::Storage)? {
+            put_repository_view_withdrawn(store, notification_uri, &uri, previous_hash.clone())
+                .map_err(RrdpSyncError::Storage)?;
+            put_rrdp_uri_owner_withdrawn(
+                store,
+                notification_uri,
+                &session_id,
+                expected_serial,
+                &uri,
+                previous_hash,
+            )
+            .map_err(RrdpSyncError::Storage)?;
+        }
+    }
+
+    Ok(published.len())
 }
 
 fn parse_rrdp_xml(xml: &[u8]) -> Result<roxmltree::Document<'_>, RrdpError> {
@@ -1477,6 +1730,34 @@ mod tests {
                 .expect("contains"),
             "c added to rrdp repo index"
         );
+
+        let a_view = store
+            .get_repository_view_entry("rsync://example.net/repo/a.mft")
+            .expect("get a view")
+            .expect("a view exists");
+        assert_eq!(a_view.state, crate::storage::RepositoryViewState::Withdrawn);
+        let b_view = store
+            .get_repository_view_entry("rsync://example.net/repo/b.roa")
+            .expect("get b view")
+            .expect("b view exists");
+        assert_eq!(b_view.state, crate::storage::RepositoryViewState::Present);
+        assert_eq!(
+            b_view.current_hash.as_deref(),
+            Some(hex::encode(sha2::Sha256::digest(b"b2")).as_str())
+        );
+        let c_owner = store
+            .get_rrdp_uri_owner_record("rsync://example.net/repo/c.crl")
+            .expect("get c owner")
+            .expect("c owner exists");
+        assert_eq!(
+            c_owner.owner_state,
+            crate::storage::RrdpUriOwnerState::Active
+        );
+        let a_member = store
+            .get_rrdp_source_member_record(notif_uri, "rsync://example.net/repo/a.mft")
+            .expect("get a member")
+            .expect("a member exists");
+        assert!(!a_member.present);
     }
 
     #[test]
@@ -1695,6 +1976,41 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn sync_from_notification_snapshot_rejects_cross_source_owner_conflict() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let store = RocksStore::open(tmp.path()).expect("open rocksdb");
+
+        let sid_a = "550e8400-e29b-41d4-a716-446655440000";
+        let sid_b = "550e8400-e29b-41d4-a716-446655440001";
+        let uri = "rsync://example.net/repo/a.mft";
+
+        let notif_a_uri = "https://example.net/a/notification.xml";
+        let snapshot_a_uri = "https://example.net/a/snapshot.xml";
+        let snapshot_a = snapshot_xml(sid_a, 1, &[(uri, b"a1")]);
+        let snapshot_a_hash = hex::encode(sha2::Sha256::digest(&snapshot_a));
+        let notif_a = notification_xml(sid_a, 1, snapshot_a_uri, &snapshot_a_hash);
+        let fetcher_a = MapFetcher {
+            map: HashMap::from([(snapshot_a_uri.to_string(), snapshot_a)]),
+        };
+        sync_from_notification_snapshot(&store, notif_a_uri, &notif_a, &fetcher_a)
+            .expect("seed source a");
+
+        let notif_b_uri = "https://example.net/b/notification.xml";
+        let snapshot_b_uri = "https://example.net/b/snapshot.xml";
+        let snapshot_b = snapshot_xml(sid_b, 1, &[(uri, b"b1")]);
+        let snapshot_b_hash = hex::encode(sha2::Sha256::digest(&snapshot_b));
+        let notif_b = notification_xml(sid_b, 1, snapshot_b_uri, &snapshot_b_hash);
+        let fetcher_b = MapFetcher {
+            map: HashMap::from([(snapshot_b_uri.to_string(), snapshot_b)]),
+        };
+
+        let err = sync_from_notification_snapshot(&store, notif_b_uri, &notif_b, &fetcher_b)
+            .expect_err("cross-source overwrite must fail");
+        assert!(matches!(err, RrdpSyncError::Storage(_)));
+        assert!(err.to_string().contains("owner conflict"), "{err}");
+    }
+
     #[test]
     fn sync_from_notification_snapshot_applies_snapshot_and_stores_state() {
         let tmp = tempfile::tempdir().expect("tempdir");
@@ -1743,6 +2059,46 @@ mod tests {
         let state = RrdpState::decode(&state_bytes).expect("decode state");
         assert_eq!(state.session_id, sid);
         assert_eq!(state.serial, serial);
+
+        let source = store
+            .get_rrdp_source_record(notif_uri)
+            .expect("get rrdp source")
+            .expect("rrdp source exists");
+        assert_eq!(source.last_session_id.as_deref(), Some(sid));
+        assert_eq!(source.last_serial, Some(serial));
+        assert_eq!(
+            source.sync_state,
+            crate::storage::RrdpSourceSyncState::SnapshotOnly
+        );
+
+        let view = store
+            .get_repository_view_entry("rsync://example.net/repo/a.mft")
+            .expect("get repository view")
+            .expect("repository view exists");
+        assert_eq!(view.state, crate::storage::RepositoryViewState::Present);
+        assert_eq!(view.repository_source.as_deref(), Some(notif_uri));
+
+        let raw = store
+            .get_raw_by_hash_entry(hex::encode(sha2::Sha256::digest(b"mft-bytes")).as_str())
+            .expect("get raw_by_hash")
+            .expect("raw_by_hash exists");
+        assert!(
+            raw.origin_uris
+                .iter()
+                .any(|uri| uri == "rsync://example.net/repo/a.mft")
+        );
+
+        let member = store
+            .get_rrdp_source_member_record(notif_uri, "rsync://example.net/repo/a.mft")
+            .expect("get member")
+            .expect("member exists");
+        assert!(member.present);
+        let owner = store
+            .get_rrdp_uri_owner_record("rsync://example.net/repo/a.mft")
+            .expect("get owner")
+            .expect("owner exists");
+        assert_eq!(owner.notify_uri, notif_uri);
+        assert_eq!(owner.owner_state, crate::storage::RrdpUriOwnerState::Active);
     }
 
     #[test]

src/sync/store_projection.rs

@@ -0,0 +1,262 @@
+use std::collections::BTreeSet;
+
+use crate::storage::{
+    PackTime, RawByHashEntry, RepositoryViewEntry, RepositoryViewState, RocksStore,
+    RrdpSourceMemberRecord, RrdpSourceRecord, RrdpSourceSyncState, RrdpUriOwnerRecord,
+    RrdpUriOwnerState,
+};
+use sha2::Digest;
+
+pub fn infer_object_type_from_uri(uri: &str) -> Option<String> {
+    let ext = uri.rsplit('.').next()?;
+    let ext = ext.to_ascii_lowercase();
+    match ext.as_str() {
+        "cer" | "crl" | "mft" | "roa" | "asa" | "gbr" | "tal" | "xml" => Some(ext),
+        _ => None,
+    }
+}
+
+pub fn upsert_raw_by_hash_evidence(
+    store: &RocksStore,
+    rsync_uri: &str,
+    bytes: &[u8],
+) -> Result<String, String> {
+    let sha256_hex = compute_sha256_hex(bytes);
+    let mut entry = match store
+        .get_raw_by_hash_entry(&sha256_hex)
+        .map_err(|e| e.to_string())?
+    {
+        Some(existing) => existing,
+        None => RawByHashEntry::from_bytes(sha256_hex.clone(), bytes.to_vec()),
+    };
+
+    if entry.bytes != bytes {
+        return Err(format!(
+            "raw_by_hash collision for {rsync_uri}: same sha256 maps to different bytes"
+        ));
+    }
+
+    let mut origins: BTreeSet<String> = entry.origin_uris.into_iter().collect();
+    origins.insert(rsync_uri.to_string());
+    entry.origin_uris = origins.into_iter().collect();
+    if entry.object_type.is_none() {
+        entry.object_type = infer_object_type_from_uri(rsync_uri);
+    }
+
+    store
+        .put_raw_by_hash_entry(&entry)
+        .map_err(|e| e.to_string())?;
+    Ok(sha256_hex)
+}
+
+pub fn put_repository_view_present(
+    store: &RocksStore,
+    repository_source: &str,
+    rsync_uri: &str,
+    current_hash: &str,
+) -> Result<(), String> {
+    let entry = RepositoryViewEntry {
+        rsync_uri: rsync_uri.to_string(),
+        current_hash: Some(current_hash.to_string()),
+        repository_source: Some(repository_source.to_string()),
+        object_type: infer_object_type_from_uri(rsync_uri),
+        state: RepositoryViewState::Present,
+    };
+    store
+        .put_repository_view_entry(&entry)
+        .map_err(|e| e.to_string())
+}
+
+pub fn put_repository_view_withdrawn(
+    store: &RocksStore,
+    repository_source: &str,
+    rsync_uri: &str,
+    current_hash: Option<String>,
+) -> Result<(), String> {
+    let entry = RepositoryViewEntry {
+        rsync_uri: rsync_uri.to_string(),
+        current_hash,
+        repository_source: Some(repository_source.to_string()),
+        object_type: infer_object_type_from_uri(rsync_uri),
+        state: RepositoryViewState::Withdrawn,
+    };
+    store
+        .put_repository_view_entry(&entry)
+        .map_err(|e| e.to_string())
+}
+
+pub fn ensure_rrdp_uri_can_be_owned_by(
+    store: &RocksStore,
+    notification_uri: &str,
+    rsync_uri: &str,
+) -> Result<(), String> {
+    let Some(owner) = store
+        .get_rrdp_uri_owner_record(rsync_uri)
+        .map_err(|e| e.to_string())?
+    else {
+        return Ok(());
+    };
+
+    if owner.notify_uri != notification_uri && owner.owner_state == RrdpUriOwnerState::Active {
+        return Err(format!(
+            "RRDP source owner conflict for {rsync_uri}: current owner {} but incoming source {}",
+            owner.notify_uri, notification_uri
+        ));
+    }
+    Ok(())
+}
+
+pub fn current_rrdp_owner_is(
+    store: &RocksStore,
+    notification_uri: &str,
+    rsync_uri: &str,
+) -> Result<bool, String> {
+    Ok(matches!(
+        store
+            .get_rrdp_uri_owner_record(rsync_uri)
+            .map_err(|e| e.to_string())?,
+        Some(owner)
+            if owner.notify_uri == notification_uri && owner.owner_state == RrdpUriOwnerState::Active
+    ))
+}
+
+pub fn put_rrdp_source_member_present(
+    store: &RocksStore,
+    notification_uri: &str,
+    session_id: &str,
+    serial: u64,
+    rsync_uri: &str,
+    current_hash: &str,
+) -> Result<(), String> {
+    let record = RrdpSourceMemberRecord {
+        notify_uri: notification_uri.to_string(),
+        rsync_uri: rsync_uri.to_string(),
+        current_hash: Some(current_hash.to_string()),
+        object_type: infer_object_type_from_uri(rsync_uri),
+        present: true,
+        last_confirmed_session_id: session_id.to_string(),
+        last_confirmed_serial: serial,
+        last_changed_at: now_pack_time(),
+    };
+    store
+        .put_rrdp_source_member_record(&record)
+        .map_err(|e| e.to_string())
+}
+
+pub fn put_rrdp_source_member_withdrawn(
+    store: &RocksStore,
+    notification_uri: &str,
+    session_id: &str,
+    serial: u64,
+    rsync_uri: &str,
+    current_hash: Option<String>,
+) -> Result<(), String> {
+    let record = RrdpSourceMemberRecord {
+        notify_uri: notification_uri.to_string(),
+        rsync_uri: rsync_uri.to_string(),
+        current_hash,
+        object_type: infer_object_type_from_uri(rsync_uri),
+        present: false,
+        last_confirmed_session_id: session_id.to_string(),
+        last_confirmed_serial: serial,
+        last_changed_at: now_pack_time(),
+    };
+    store
+        .put_rrdp_source_member_record(&record)
+        .map_err(|e| e.to_string())
+}
+
+pub fn put_rrdp_uri_owner_active(
+    store: &RocksStore,
+    notification_uri: &str,
+    session_id: &str,
+    serial: u64,
+    rsync_uri: &str,
+    current_hash: &str,
+) -> Result<(), String> {
+    let record = RrdpUriOwnerRecord {
+        rsync_uri: rsync_uri.to_string(),
+        notify_uri: notification_uri.to_string(),
+        current_hash: Some(current_hash.to_string()),
+        last_confirmed_session_id: session_id.to_string(),
+        last_confirmed_serial: serial,
+        last_changed_at: now_pack_time(),
+        owner_state: RrdpUriOwnerState::Active,
+    };
+    store
+        .put_rrdp_uri_owner_record(&record)
+        .map_err(|e| e.to_string())
+}
+
+pub fn put_rrdp_uri_owner_withdrawn(
+    store: &RocksStore,
+    notification_uri: &str,
+    session_id: &str,
+    serial: u64,
+    rsync_uri: &str,
+    current_hash: Option<String>,
+) -> Result<(), String> {
+    let record = RrdpUriOwnerRecord {
+        rsync_uri: rsync_uri.to_string(),
+        notify_uri: notification_uri.to_string(),
+        current_hash,
+        last_confirmed_session_id: session_id.to_string(),
+        last_confirmed_serial: serial,
+        last_changed_at: now_pack_time(),
+        owner_state: RrdpUriOwnerState::Withdrawn,
+    };
+    store
+        .put_rrdp_uri_owner_record(&record)
+        .map_err(|e| e.to_string())
+}
+
+pub fn update_rrdp_source_record_on_success(
+    store: &RocksStore,
+    notification_uri: &str,
+    session_id: &str,
+    serial: u64,
+    sync_state: RrdpSourceSyncState,
+    last_snapshot_uri: Option<&str>,
+    last_snapshot_hash_hex: Option<&str>,
+) -> Result<(), String> {
+    let now = now_pack_time();
+    let mut record = match store
+        .get_rrdp_source_record(notification_uri)
+        .map_err(|e| e.to_string())?
+    {
+        Some(existing) => existing,
+        None => RrdpSourceRecord {
+            notify_uri: notification_uri.to_string(),
+            last_session_id: None,
+            last_serial: None,
+            first_seen_at: now.clone(),
+            last_seen_at: now.clone(),
+            last_sync_at: None,
+            sync_state,
+            last_snapshot_uri: None,
+            last_snapshot_hash: None,
+            last_error: None,
+        },
+    };
+
+    record.last_session_id = Some(session_id.to_string());
+    record.last_serial = Some(serial);
+    record.last_seen_at = now.clone();
+    record.last_sync_at = Some(now);
+    record.sync_state = sync_state;
+    record.last_snapshot_uri = last_snapshot_uri.map(str::to_string);
+    record.last_snapshot_hash = last_snapshot_hash_hex.map(str::to_string);
+    record.last_error = None;
+
+    store
+        .put_rrdp_source_record(&record)
+        .map_err(|e| e.to_string())
+}
+
+pub fn compute_sha256_hex(bytes: &[u8]) -> String {
+    hex::encode(sha2::Sha256::digest(bytes))
+}
+
+pub fn now_pack_time() -> PackTime {
+    PackTime::from_utc_offset_datetime(time::OffsetDateTime::now_utc())
+}

src/validation/mod.rs

@@ -4,6 +4,7 @@ pub mod cert_path;
 pub mod from_tal;
 pub mod manifest;
 pub mod objects;
+pub mod publication_point;
 pub mod run;
 pub mod run_tree_from_tal;
 pub mod tree;

src/validation/objects.rs

@@ -10,8 +10,10 @@ use crate::data_model::roa::{IpPrefix, RoaAfi, RoaDecodeError, RoaObject, RoaVal
 use crate::data_model::signed_object::SignedObjectVerifyError;
 use crate::policy::{Policy, SignedObjectFailurePolicy};
 use crate::report::{RfcRef, Warning};
-use crate::storage::{FetchCachePpPack, PackFile};
+use crate::storage::PackFile;
 use crate::validation::cert_path::{CertPathError, validate_ee_cert_path_with_prevalidated_issuer};
+use crate::validation::manifest::PublicationPointData;
+use crate::validation::publication_point::PublicationPointSnapshot;
 use x509_parser::prelude::FromDer;
 use x509_parser::x509::SubjectPublicKeyInfo;
 
@@ -81,10 +83,10 @@ pub struct ObjectsStats {
     pub publication_point_dropped: bool,
 }
 
-/// Process objects from a fetch_cache_pp publication point pack using a known issuer CA certificate
+/// Process objects from a publication point snapshot using a known issuer CA certificate
 /// and its effective resources (resolved via the resource-path, RFC 6487 §7.2).
-pub fn process_fetch_cache_pp_pack_for_issuer(
-    pack: &FetchCachePpPack,
+pub fn process_publication_point_for_issuer<P: PublicationPointData>(
+    publication_point: &P,
     policy: &Policy,
     issuer_ca_der: &[u8],
     issuer_ca_rsync_uri: Option<&str>,
@@ -93,23 +95,24 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
     validation_time: time::OffsetDateTime,
     timing: Option<&TimingHandle>,
 ) -> ObjectsOutput {
+    let manifest_rsync_uri = publication_point.manifest_rsync_uri();
+    let manifest_bytes = publication_point.manifest_bytes();
+    let locked_files = publication_point.files();
     let mut warnings: Vec<Warning> = Vec::new();
     let mut stats = ObjectsStats::default();
-    stats.roa_total = pack
-        .files
+    stats.roa_total = locked_files
         .iter()
         .filter(|f| f.rsync_uri.ends_with(".roa"))
         .count();
-    stats.aspa_total = pack
-        .files
+    stats.aspa_total = locked_files
         .iter()
         .filter(|f| f.rsync_uri.ends_with(".asa"))
         .count();
     let mut audit: Vec<ObjectAuditEntry> = Vec::new();
 
     // Enforce that `manifest_bytes` is actually a manifest object.
-    let _manifest =
-        ManifestObject::decode_der(&pack.manifest_bytes).expect("fetch_cache_pp manifest decodes");
+    let _manifest = ManifestObject::decode_der(manifest_bytes)
+        .expect("publication point snapshot manifest decodes");
 
     // Decode issuer CA once; if it fails we cannot validate ROA/ASPA EE certificates.
     let issuer_ca = match ResourceCertificate::decode_der(issuer_ca_der) {
@@ -121,9 +124,9 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                     "dropping publication point: issuer CA decode failed: {e}"
                 ))
                 .with_rfc_refs(&[RfcRef("RFC 6487 §7.2"), RfcRef("RFC 5280 §6.1")])
-                .with_context(&pack.manifest_rsync_uri),
+                .with_context(manifest_rsync_uri),
             );
-            for f in &pack.files {
+            for f in locked_files {
                 if f.rsync_uri.ends_with(".roa") {
                     audit.push(ObjectAuditEntry {
                         rsync_uri: f.rsync_uri.clone(),
@@ -163,7 +166,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                     rem.len()
                 ))
                 .with_rfc_refs(&[RfcRef("RFC 5280 §4.1.2.7")])
-                .with_context(&pack.manifest_rsync_uri),
+                .with_context(manifest_rsync_uri),
             );
             return ObjectsOutput {
                 vrps: Vec::new(),
@@ -180,7 +183,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                     "dropping publication point: issuer SPKI parse failed: {e}"
                 ))
                 .with_rfc_refs(&[RfcRef("RFC 5280 §4.1.2.7")])
-                .with_context(&pack.manifest_rsync_uri),
+                .with_context(manifest_rsync_uri),
             );
             return ObjectsOutput {
                 vrps: Vec::new(),
@@ -192,8 +195,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
         }
     };
 
-    let mut crl_cache: std::collections::HashMap<String, CachedIssuerCrl> = pack
-        .files
+    let mut crl_cache: std::collections::HashMap<String, CachedIssuerCrl> = locked_files
         .iter()
         .filter(|f| f.rsync_uri.ends_with(".crl"))
         .map(|f| {
@@ -207,23 +209,26 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
     let issuer_resources_index =
         build_issuer_resources_index(issuer_effective_ip, issuer_effective_as);
 
-    // If the pack has signed objects but no CRLs at all, we cannot validate any embedded EE
-    // certificate paths deterministically (EE CRLDP must reference an rsync URI in the pack).
+    // If the snapshot has signed objects but no CRLs at all, we cannot validate any embedded EE
+    // certificate paths deterministically (EE CRLDP must reference an rsync URI in the snapshot).
     if crl_cache.is_empty() && (stats.roa_total > 0 || stats.aspa_total > 0) {
         stats.publication_point_dropped = true;
         warnings.push(
-            Warning::new("dropping publication point: no CRL files in fetch_cache_pp")
+            Warning::new("dropping publication point: no CRL files in validated publication point")
                 .with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")])
-                .with_context(&pack.manifest_rsync_uri),
+                .with_context(manifest_rsync_uri),
         );
-        for f in &pack.files {
+        for f in locked_files {
             if f.rsync_uri.ends_with(".roa") {
                 audit.push(ObjectAuditEntry {
                     rsync_uri: f.rsync_uri.clone(),
                     sha256_hex: sha256_hex_from_32(&f.sha256),
                     kind: AuditObjectKind::Roa,
                     result: AuditObjectResult::Skipped,
-                    detail: Some("skipped due to missing CRL files in fetch_cache_pp".to_string()),
+                    detail: Some(
+                        "skipped due to missing CRL files in validated publication point"
+                            .to_string(),
+                    ),
                 });
             } else if f.rsync_uri.ends_with(".asa") {
                 audit.push(ObjectAuditEntry {
@@ -231,7 +236,10 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                     sha256_hex: sha256_hex_from_32(&f.sha256),
                     kind: AuditObjectKind::Aspa,
                     result: AuditObjectResult::Skipped,
-                    detail: Some("skipped due to missing CRL files in fetch_cache_pp".to_string()),
+                    detail: Some(
+                        "skipped due to missing CRL files in validated publication point"
+                            .to_string(),
+                    ),
                 });
             }
         }
@@ -247,7 +255,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
     let mut vrps: Vec<Vrp> = Vec::new();
     let mut aspas: Vec<AspaAttestation> = Vec::new();
 
-    for (idx, file) in pack.files.iter().enumerate() {
+    for (idx, file) in locked_files.iter().enumerate() {
         if file.rsync_uri.ends_with(".roa") {
             let _t = timing.as_ref().map(|t| t.span_phase("objects_roa_total"));
             match process_roa_with_issuer(
@@ -300,7 +308,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                             result: AuditObjectResult::Error,
                             detail: Some(e.to_string()),
                         });
-                        for f in pack.files.iter().skip(idx + 1) {
+                        for f in locked_files.iter().skip(idx + 1) {
                             if f.rsync_uri.ends_with(".roa") {
                                 audit.push(ObjectAuditEntry {
                                     rsync_uri: f.rsync_uri.clone(),
@@ -333,7 +341,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                                 file.rsync_uri
                             ))
                             .with_rfc_refs(&refs)
-                            .with_context(&pack.manifest_rsync_uri),
+                            .with_context(manifest_rsync_uri),
                         );
                         return ObjectsOutput {
                             vrps: Vec::new(),
@@ -397,7 +405,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                             result: AuditObjectResult::Error,
                             detail: Some(e.to_string()),
                         });
-                        for f in pack.files.iter().skip(idx + 1) {
+                        for f in locked_files.iter().skip(idx + 1) {
                             if f.rsync_uri.ends_with(".roa") {
                                 audit.push(ObjectAuditEntry {
                                     rsync_uri: f.rsync_uri.clone(),
@@ -430,7 +438,7 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
                                 file.rsync_uri
                             ))
                             .with_rfc_refs(&refs)
-                            .with_context(&pack.manifest_rsync_uri),
+                            .with_context(manifest_rsync_uri),
                         );
                         return ObjectsOutput {
                             vrps: Vec::new(),
@@ -453,6 +461,28 @@ pub fn process_fetch_cache_pp_pack_for_issuer(
         audit,
     }
 }
+/// Compatibility wrapper that processes a publication point snapshot.
+pub fn process_publication_point_snapshot_for_issuer(
+    pack: &PublicationPointSnapshot,
+    policy: &Policy,
+    issuer_ca_der: &[u8],
+    issuer_ca_rsync_uri: Option<&str>,
+    issuer_effective_ip: Option<&crate::data_model::rc::IpResourceSet>,
+    issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
+    validation_time: time::OffsetDateTime,
+    timing: Option<&TimingHandle>,
+) -> ObjectsOutput {
+    process_publication_point_for_issuer(
+        pack,
+        policy,
+        issuer_ca_der,
+        issuer_ca_rsync_uri,
+        issuer_effective_ip,
+        issuer_effective_as,
+        validation_time,
+        timing,
+    )
+}
 
 #[derive(Debug, thiserror::Error)]
 enum ObjectValidateError {
@@ -480,12 +510,12 @@ enum ObjectValidateError {
     MissingCrlDpUris,
 
     #[error(
-        "no CRL available in fetch_cache_pp (cannot validate certificates) (RFC 9286 §7; RFC 6487 §4.8.6)"
+        "no CRL available in publication point snapshot (cannot validate certificates) (RFC 9286 §7; RFC 6487 §4.8.6)"
     )]
     MissingCrlInPack,
 
     #[error(
-        "CRL referenced by CRLDistributionPoints not found in fetch_cache_pp: {0} (RFC 6487 §4.8.6; RFC 9286 §4.2.1)"
+        "CRL referenced by CRLDistributionPoints not found in publication point snapshot: {0} (RFC 6487 §4.8.6; RFC 9286 §4.2.1)"
     )]
     CrlNotFound(String),
 
@@ -1283,7 +1313,7 @@ mod tests {
     }
 
     #[test]
-    fn choose_crl_for_certificate_reports_missing_crl_in_pack() {
+    fn choose_crl_for_certificate_reports_missing_crl_in_snapshot() {
         let roa_der =
             fixture_bytes("tests/fixtures/repository/rpki.cernet.net/repo/cernet/0/AS4538.roa");
         let roa = RoaObject::decode_der(&roa_der).expect("decode roa");
@@ -1335,7 +1365,7 @@ mod tests {
     }
 
     #[test]
-    fn choose_crl_for_certificate_reports_not_found_when_crldp_does_not_match_pack() {
+    fn choose_crl_for_certificate_reports_not_found_when_crldp_does_not_match_snapshot() {
         let roa_der =
             fixture_bytes("tests/fixtures/repository/rpki.cernet.net/repo/cernet/0/AS4538.roa");
         let roa = RoaObject::decode_der(&roa_der).expect("decode roa");
@@ -1421,4 +1451,225 @@ mod tests {
         let err = validate_ee_resources_subset(ee, Some(&issuer_ip), None, &idx).unwrap_err();
         assert!(matches!(err, ObjectValidateError::EeResourcesNotSubset));
     }
+
+    #[test]
+    fn extra_rfc_refs_for_crl_selection_distinguishes_crl_errors() {
+        assert_eq!(
+            extra_rfc_refs_for_crl_selection(&ObjectValidateError::MissingCrlDpUris),
+            RFC_CRLDP
+        );
+        assert_eq!(
+            extra_rfc_refs_for_crl_selection(&ObjectValidateError::CrlNotFound(
+                "rsync://example.test/x.crl".to_string(),
+            )),
+            RFC_CRLDP_AND_LOCKED_PACK
+        );
+        assert!(
+            extra_rfc_refs_for_crl_selection(&ObjectValidateError::MissingCrlInPack).is_empty()
+        );
+    }
+
+    #[test]
+    fn as_subset_helpers_cover_success_and_failure_paths() {
+        let child = AsIdentifierChoice::AsIdsOrRanges(vec![
+            AsIdOrRange::Id(5),
+            AsIdOrRange::Range { min: 7, max: 9 },
+        ]);
+        let parent =
+            AsIdentifierChoice::AsIdsOrRanges(vec![AsIdOrRange::Range { min: 1, max: 10 }]);
+        let parent_intervals = [(1, 10)];
+
+        assert!(as_choice_subset(None, Some(&parent)));
+        assert!(!as_choice_subset(Some(&child), None));
+        assert!(!as_choice_subset(
+            Some(&AsIdentifierChoice::Inherit),
+            Some(&parent)
+        ));
+        assert!(!as_choice_subset(
+            Some(&child),
+            Some(&AsIdentifierChoice::Inherit)
+        ));
+        assert!(as_choice_subset(Some(&child), Some(&parent)));
+
+        assert!(as_choice_subset_indexed(None, Some(&parent_intervals)));
+        assert!(!as_choice_subset_indexed(Some(&child), None));
+        assert!(!as_choice_subset_indexed(
+            Some(&AsIdentifierChoice::Inherit),
+            Some(&parent_intervals),
+        ));
+        assert!(as_choice_subset_indexed(
+            Some(&child),
+            Some(&parent_intervals)
+        ));
+        assert!(!as_choice_subset_indexed(
+            Some(&AsIdentifierChoice::AsIdsOrRanges(vec![
+                AsIdOrRange::Range { min: 11, max: 12 }
+            ])),
+            Some(&parent_intervals),
+        ));
+
+        let child_set = AsResourceSet {
+            asnum: Some(child.clone()),
+            rdi: Some(AsIdentifierChoice::AsIdsOrRanges(vec![AsIdOrRange::Id(42)])),
+        };
+        let parent_set = AsResourceSet {
+            asnum: Some(parent.clone()),
+            rdi: Some(AsIdentifierChoice::AsIdsOrRanges(vec![
+                AsIdOrRange::Range { min: 40, max: 50 },
+            ])),
+        };
+        assert!(as_resources_is_subset(&child_set, &parent_set));
+        assert!(as_resources_is_subset_indexed(
+            &child_set,
+            &parent_set,
+            &IssuerResourcesIndex {
+                asnum: Some(vec![(1, 10)]),
+                rdi: Some(vec![(40, 50)]),
+                ..IssuerResourcesIndex::default()
+            },
+        ));
+    }
+
+    #[test]
+    fn ip_subset_helpers_cover_strict_and_indexed_paths() {
+        let parent = IpResourceSet {
+            families: vec![
+                IpAddressFamily {
+                    afi: Afi::Ipv4,
+                    choice: IpAddressChoice::AddressesOrRanges(vec![
+                        IpAddressOrRange::Prefix(IpPrefix {
+                            afi: Afi::Ipv4,
+                            prefix_len: 8,
+                            addr: vec![10, 0, 0, 0],
+                        }),
+                        IpAddressOrRange::Range(IpAddressRange {
+                            min: vec![192, 0, 2, 0],
+                            max: vec![192, 0, 2, 255],
+                        }),
+                    ]),
+                },
+                IpAddressFamily {
+                    afi: Afi::Ipv6,
+                    choice: IpAddressChoice::Inherit,
+                },
+            ],
+        };
+        let child = IpResourceSet {
+            families: vec![IpAddressFamily {
+                afi: Afi::Ipv4,
+                choice: IpAddressChoice::AddressesOrRanges(vec![
+                    IpAddressOrRange::Prefix(IpPrefix {
+                        afi: Afi::Ipv4,
+                        prefix_len: 16,
+                        addr: vec![10, 1, 0, 0],
+                    }),
+                    IpAddressOrRange::Range(IpAddressRange {
+                        min: vec![192, 0, 2, 10],
+                        max: vec![192, 0, 2, 20],
+                    }),
+                ]),
+            }],
+        };
+        let strict_bad = IpResourceSet {
+            families: vec![IpAddressFamily {
+                afi: Afi::Ipv4,
+                choice: IpAddressChoice::Inherit,
+            }],
+        };
+
+        assert!(ip_resources_is_subset(&child, &parent));
+        assert!(ip_resources_to_merged_intervals(&parent).contains_key(&AfiKey::V4));
+        assert!(ip_resources_to_merged_intervals_strict(&child).is_ok());
+        assert!(ip_resources_to_merged_intervals_strict(&strict_bad).is_err());
+
+        let idx = build_issuer_resources_index(
+            Some(&parent),
+            Some(&AsResourceSet {
+                asnum: Some(AsIdentifierChoice::AsIdsOrRanges(vec![
+                    AsIdOrRange::Range {
+                        min: 64500,
+                        max: 64510,
+                    },
+                ])),
+                rdi: Some(AsIdentifierChoice::Inherit),
+            }),
+        );
+        assert!(idx.ip_v4.is_some());
+        assert!(idx.ip_v6.is_none());
+        assert!(idx.asnum.is_some());
+        assert!(idx.rdi.is_none());
+        assert!(ip_resources_is_subset_indexed(&child, &parent, &idx));
+        assert!(!ip_resources_is_subset_indexed(&strict_bad, &parent, &idx));
+    }
+
+    #[test]
+    fn interval_and_byte_helpers_cover_edge_cases() {
+        let parent = vec![(vec![0, 0, 0, 0], vec![0, 0, 0, 10])];
+        assert!(interval_is_covered(&parent, &[0, 0, 0, 1], &[0, 0, 0, 2]));
+        assert!(!interval_is_covered(
+            &parent,
+            &[0, 0, 0, 11],
+            &[0, 0, 0, 12]
+        ));
+        assert!(intervals_are_covered(
+            &parent,
+            &[(vec![0, 0, 0, 1], vec![0, 0, 0, 2])]
+        ));
+        assert!(!intervals_are_covered(
+            &parent,
+            &[(vec![0, 0, 0, 9], vec![0, 0, 0, 11])],
+        ));
+
+        let prefix = RcIpPrefix {
+            afi: Afi::Ipv4,
+            prefix_len: 24,
+            addr: vec![203, 0, 113, 7],
+        };
+        assert_eq!(
+            prefix_to_range(&prefix),
+            (vec![203, 0, 113, 0], vec![203, 0, 113, 255])
+        );
+        assert_eq!(increment_bytes(&[0, 0, 0, 255]), vec![0, 0, 1, 0]);
+        assert!(bytes_is_next(&[0, 0, 1, 0], &[0, 0, 0, 255]));
+        assert!(!bytes_is_next(&[1, 2], &[1]));
+    }
+
+    #[test]
+    fn merged_interval_helpers_cover_empty_and_break_paths() {
+        let mut empty: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
+        merge_ip_intervals_in_place(&mut empty);
+        assert!(empty.is_empty());
+
+        let mut v = vec![
+            (vec![0, 0, 0, 20], vec![0, 0, 0, 30]),
+            (vec![0, 0, 0, 0], vec![0, 0, 0, 10]),
+            (vec![0, 0, 0, 11], vec![0, 0, 0, 19]),
+        ];
+        v.sort_by(|(a, _), (b, _)| a.cmp(b));
+        merge_ip_intervals_in_place(&mut v);
+        assert_eq!(v, vec![(vec![0, 0, 0, 0], vec![0, 0, 0, 30])]);
+
+        assert_eq!(
+            as_choice_to_merged_intervals(&AsIdentifierChoice::AsIdsOrRanges(vec![
+                AsIdOrRange::Id(1),
+                AsIdOrRange::Range { min: 2, max: 3 },
+                AsIdOrRange::Range { min: 7, max: 9 },
+            ])),
+            vec![(1, 3), (7, 9)]
+        );
+        assert!(as_interval_is_covered(&[(1, 3), (7, 9)], 2, 3));
+        assert!(!as_interval_is_covered(&[(7, 9)], 2, 3));
+    }
+
+    #[test]
+    fn roa_output_helpers_cover_vrps_and_afi_strings() {
+        let roa_der =
+            fixture_bytes("tests/fixtures/repository/rpki.cernet.net/repo/cernet/0/AS4538.roa");
+        let roa = RoaObject::decode_der(&roa_der).expect("decode roa");
+        let vrps = roa_to_vrps(&roa);
+        assert!(!vrps.is_empty());
+        assert!(vrps.iter().all(|vrp| vrp.asn == roa.roa.as_id));
+        assert_eq!(roa_afi_to_string(RoaAfi::Ipv4), "ipv4");
+        assert_eq!(roa_afi_to_string(RoaAfi::Ipv6), "ipv6");
+    }
 }

src/validation/publication_point.rs

@@ -0,0 +1,18 @@
+use crate::storage::{PackFile, PackTime};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PublicationPointSnapshot {
+    pub format_version: u32,
+    pub manifest_rsync_uri: String,
+    pub publication_point_rsync_uri: String,
+    pub manifest_number_be: Vec<u8>,
+    pub this_update: PackTime,
+    pub next_update: PackTime,
+    pub verified_at: PackTime,
+    pub manifest_bytes: Vec<u8>,
+    pub files: Vec<PackFile>,
+}
+
+impl PublicationPointSnapshot {
+    pub const FORMAT_VERSION_V1: u32 = 1;
+}

src/validation/run.rs

@@ -1,34 +1,35 @@
 use crate::data_model::rc::{AsResourceSet, IpResourceSet};
 use crate::fetch::rsync::RsyncFetcher;
 use crate::policy::Policy;
-use crate::storage::{FetchCachePpKey, RocksStore};
-use crate::sync::repo::{RepoSyncResult, sync_publication_point};
+use crate::report::Warning;
+use crate::storage::RocksStore;
 use crate::sync::rrdp::Fetcher as HttpFetcher;
-use crate::validation::manifest::{PublicationPointResult, process_manifest_publication_point};
-use crate::validation::objects::{ObjectsOutput, process_fetch_cache_pp_pack_for_issuer};
+use crate::validation::manifest::PublicationPointSource;
+use crate::validation::objects::ObjectsOutput;
+use crate::validation::tree::{CaInstanceHandle, PublicationPointRunner};
+use crate::validation::tree_runner::Rpkiv1PublicationPointRunner;
+use std::collections::HashMap;
+use std::sync::Mutex;
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct RunOutput {
-    pub repo_sync: RepoSyncResult,
-    pub publication_point: PublicationPointResult,
+    pub publication_point_source: PublicationPointSource,
+    pub publication_point_warnings: Vec<Warning>,
     pub objects: ObjectsOutput,
 }
 
 #[derive(Debug, thiserror::Error)]
 pub enum RunError {
-    #[error("repo sync failed: {0}")]
-    RepoSync(#[from] crate::sync::repo::RepoSyncError),
-
-    #[error("manifest processing failed: {0}")]
-    Manifest(#[from] crate::validation::manifest::ManifestProcessError),
+    #[error("publication point runner failed: {0}")]
+    Runner(String),
 }
 
-/// v1 serial offline-friendly end-to-end execution for a single publication point.
+/// Offline-friendly end-to-end execution for a single publication point.
 ///
-/// This orchestrates:
-/// 1) repo sync (RRDP or rsync fallback) into `raw_objects`
-/// 2) manifest RP processing into a fetch_cache_pp pack (`fetch_cache_pp:<manifest-rsync-uri>`)
-/// 3) signed object processing (ROA/ASPA) from the fetch_cache_pp pack
+/// This reuses the same fresh-first runtime path as the tree runner:
+/// 1) repository sync (RRDP or rsync fallback) into the current repository view
+/// 2) manifest processing into a validated publication point result
+/// 3) signed object processing (ROA/ASPA) and latest-VCIR persistence
 pub fn run_publication_point_once(
     store: &RocksStore,
     policy: &Policy,
@@ -44,49 +45,41 @@ pub fn run_publication_point_once(
     issuer_effective_as: Option<&AsResourceSet>,
     validation_time: time::OffsetDateTime,
 ) -> Result<RunOutput, RunError> {
-    let repo_sync = sync_publication_point(
+    let handle = CaInstanceHandle {
+        depth: 0,
+        tal_id: "single-publication-point".to_string(),
+        parent_manifest_rsync_uri: None,
+        ca_certificate_der: issuer_ca_der.to_vec(),
+        ca_certificate_rsync_uri: issuer_ca_rsync_uri.map(str::to_string),
+        effective_ip_resources: issuer_effective_ip.cloned(),
+        effective_as_resources: issuer_effective_as.cloned(),
+        rsync_base_uri: rsync_base_uri.to_string(),
+        manifest_rsync_uri: manifest_rsync_uri.to_string(),
+        publication_point_rsync_uri: publication_point_rsync_uri.to_string(),
+        rrdp_notification_uri: rrdp_notification_uri.map(str::to_string),
+    };
+
+    let runner = Rpkiv1PublicationPointRunner {
         store,
         policy,
-        rrdp_notification_uri,
-        rsync_base_uri,
         http_fetcher,
         rsync_fetcher,
-        None,
-        None,
-    )?;
-
-    let publication_point = process_manifest_publication_point(
-        store,
-        policy,
-        manifest_rsync_uri,
-        publication_point_rsync_uri,
-        issuer_ca_der,
-        issuer_ca_rsync_uri,
         validation_time,
-    )?;
+        timing: None,
+        download_log: None,
+        rrdp_dedup: false,
+        rrdp_repo_cache: Mutex::new(HashMap::new()),
+        rsync_dedup: false,
+        rsync_repo_cache: Mutex::new(HashMap::new()),
+    };
 
-    let objects = process_fetch_cache_pp_pack_for_issuer(
-        &publication_point.pack,
-        policy,
-        issuer_ca_der,
-        issuer_ca_rsync_uri,
-        issuer_effective_ip,
-        issuer_effective_as,
-        validation_time,
-        None,
-    );
+    let result = runner
+        .run_publication_point(&handle)
+        .map_err(RunError::Runner)?;
 
     Ok(RunOutput {
-        repo_sync,
-        publication_point,
-        objects,
+        publication_point_source: result.source,
+        publication_point_warnings: result.warnings,
+        objects: result.objects,
     })
 }
-
-pub fn fetch_cache_pp_exists(store: &RocksStore, manifest_rsync_uri: &str) -> Result<bool, String> {
-    let key = FetchCachePpKey::from_manifest_rsync_uri(manifest_rsync_uri);
-    store
-        .get_fetch_cache_pp(&key)
-        .map(|v| v.is_some())
-        .map_err(|e| e.to_string())
-}

src/validation/run_tree_from_tal.rs

@@ -17,6 +17,44 @@ use crate::validation::tree_runner::Rpkiv1PublicationPointRunner;
 use std::collections::HashMap;
 use std::sync::Mutex;
 
+fn tal_id_from_url_like(s: &str) -> Option<String> {
+    let url = Url::parse(s).ok()?;
+    if let Some(last) = url
+        .path_segments()
+        .and_then(|segments| segments.filter(|seg| !seg.is_empty()).next_back())
+    {
+        let stem = last.rsplit_once('.').map(|(stem, _)| stem).unwrap_or(last);
+        let trimmed = stem.trim();
+        if !trimmed.is_empty() {
+            return Some(trimmed.to_string());
+        }
+    }
+    url.host_str().map(|host| host.to_string())
+}
+
+fn derive_tal_id(discovery: &DiscoveredRootCaInstance) -> String {
+    discovery
+        .tal_url
+        .as_deref()
+        .and_then(tal_id_from_url_like)
+        .or_else(|| {
+            discovery
+                .trust_anchor
+                .resolved_ta_uri
+                .as_ref()
+                .and_then(|uri| tal_id_from_url_like(uri.as_str()))
+        })
+        .or_else(|| {
+            discovery
+                .trust_anchor
+                .tal
+                .ta_uris
+                .first()
+                .and_then(|uri| tal_id_from_url_like(uri.as_str()))
+        })
+        .unwrap_or_else(|| "unknown-tal".to_string())
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct RunTreeFromTalOutput {
     pub discovery: DiscoveredRootCaInstance,
@@ -43,12 +81,15 @@ pub enum RunTreeFromTalError {
 
 pub fn root_handle_from_trust_anchor(
     trust_anchor: &TrustAnchor,
+    tal_id: String,
     ca_certificate_rsync_uri: Option<String>,
     ca_instance: &crate::validation::ca_instance::CaInstanceUris,
 ) -> CaInstanceHandle {
     let ta_rc = trust_anchor.ta_certificate.rc_ca.clone();
     CaInstanceHandle {
         depth: 0,
+        tal_id,
+        parent_manifest_rsync_uri: None,
         ca_certificate_der: trust_anchor.ta_certificate.raw_der.clone(),
         ca_certificate_rsync_uri,
         effective_ip_resources: ta_rc.tbs.extensions.ip_resources.clone(),
@@ -79,14 +120,18 @@ pub fn run_tree_from_tal_url_serial(
         validation_time,
         timing: None,
         download_log: None,
-        revalidate_only: config.revalidate_only,
         rrdp_dedup: true,
         rrdp_repo_cache: Mutex::new(HashMap::new()),
         rsync_dedup: true,
         rsync_repo_cache: Mutex::new(HashMap::new()),
     };
 
-    let root = root_handle_from_trust_anchor(&discovery.trust_anchor, None, &discovery.ca_instance);
+    let root = root_handle_from_trust_anchor(
+        &discovery.trust_anchor,
+        derive_tal_id(&discovery),
+        None,
+        &discovery.ca_instance,
+    );
     let tree = run_tree_serial(root, &runner, config)?;
 
     Ok(RunTreeFromTalOutput { discovery, tree })
@@ -112,14 +157,18 @@ pub fn run_tree_from_tal_url_serial_audit(
         validation_time,
         timing: None,
         download_log: Some(download_log.clone()),
-        revalidate_only: config.revalidate_only,
         rrdp_dedup: true,
         rrdp_repo_cache: Mutex::new(HashMap::new()),
         rsync_dedup: true,
         rsync_repo_cache: Mutex::new(HashMap::new()),
     };
 
-    let root = root_handle_from_trust_anchor(&discovery.trust_anchor, None, &discovery.ca_instance);
+    let root = root_handle_from_trust_anchor(
+        &discovery.trust_anchor,
+        derive_tal_id(&discovery),
+        None,
+        &discovery.ca_instance,
+    );
     let TreeRunAuditOutput {
         tree,
         publication_points,
@@ -159,14 +208,18 @@ pub fn run_tree_from_tal_url_serial_audit_with_timing(
         validation_time,
         timing: Some(timing.clone()),
         download_log: Some(download_log.clone()),
-        revalidate_only: config.revalidate_only,
         rrdp_dedup: true,
         rrdp_repo_cache: Mutex::new(HashMap::new()),
         rsync_dedup: true,
         rsync_repo_cache: Mutex::new(HashMap::new()),
     };
 
-    let root = root_handle_from_trust_anchor(&discovery.trust_anchor, None, &discovery.ca_instance);
+    let root = root_handle_from_trust_anchor(
+        &discovery.trust_anchor,
+        derive_tal_id(&discovery),
+        None,
+        &discovery.ca_instance,
+    );
     let _tree = timing.span_phase("tree_run_total");
     let TreeRunAuditOutput {
         tree,
@@ -206,14 +259,18 @@ pub fn run_tree_from_tal_and_ta_der_serial(
         validation_time,
         timing: None,
         download_log: None,
-        revalidate_only: config.revalidate_only,
         rrdp_dedup: true,
         rrdp_repo_cache: Mutex::new(HashMap::new()),
         rsync_dedup: true,
         rsync_repo_cache: Mutex::new(HashMap::new()),
     };
 
-    let root = root_handle_from_trust_anchor(&discovery.trust_anchor, None, &discovery.ca_instance);
+    let root = root_handle_from_trust_anchor(
+        &discovery.trust_anchor,
+        derive_tal_id(&discovery),
+        None,
+        &discovery.ca_instance,
+    );
     let tree = run_tree_serial(root, &runner, config)?;
 
     Ok(RunTreeFromTalOutput { discovery, tree })
@@ -242,14 +299,18 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit(
         validation_time,
         timing: None,
         download_log: Some(download_log.clone()),
-        revalidate_only: config.revalidate_only,
         rrdp_dedup: true,
         rrdp_repo_cache: Mutex::new(HashMap::new()),
         rsync_dedup: true,
         rsync_repo_cache: Mutex::new(HashMap::new()),
     };
 
-    let root = root_handle_from_trust_anchor(&discovery.trust_anchor, None, &discovery.ca_instance);
+    let root = root_handle_from_trust_anchor(
+        &discovery.trust_anchor,
+        derive_tal_id(&discovery),
+        None,
+        &discovery.ca_instance,
+    );
     let TreeRunAuditOutput {
         tree,
         publication_points,
@@ -292,14 +353,18 @@ pub fn run_tree_from_tal_and_ta_der_serial_audit_with_timing(
         validation_time,
         timing: Some(timing.clone()),
         download_log: Some(download_log.clone()),
-        revalidate_only: config.revalidate_only,
         rrdp_dedup: true,
         rrdp_repo_cache: Mutex::new(HashMap::new()),
         rsync_dedup: true,
         rsync_repo_cache: Mutex::new(HashMap::new()),
     };
 
-    let root = root_handle_from_trust_anchor(&discovery.trust_anchor, None, &discovery.ca_instance);
+    let root = root_handle_from_trust_anchor(
+        &discovery.trust_anchor,
+        derive_tal_id(&discovery),
+        None,
+        &discovery.ca_instance,
+    );
     let _tree = timing.span_phase("tree_run_total");
     let TreeRunAuditOutput {
         tree,

src/validation/tree.rs

@@ -1,10 +1,10 @@
 use crate::audit::DiscoveredFrom;
 use crate::audit::PublicationPointAudit;
 use crate::data_model::rc::{AsResourceSet, IpResourceSet};
-use crate::report::{RfcRef, Warning};
-use crate::storage::FetchCachePpPack;
+use crate::report::Warning;
 use crate::validation::manifest::PublicationPointSource;
 use crate::validation::objects::{AspaAttestation, ObjectsOutput, Vrp};
+use crate::validation::publication_point::PublicationPointSnapshot;
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct TreeRunConfig {
@@ -12,10 +12,6 @@ pub struct TreeRunConfig {
     pub max_depth: Option<usize>,
     /// Max number of CA instances to process.
     pub max_instances: Option<usize>,
-    /// Skip RRDP/rsync fetch and re-validate from existing `fetch_cache_pp` packs in the DB.
-    ///
-    /// This is primarily intended for profiling/analysis runs to remove network noise.
-    pub revalidate_only: bool,
 }
 
 impl Default for TreeRunConfig {
@@ -23,7 +19,6 @@ impl Default for TreeRunConfig {
         Self {
             max_depth: None,
             max_instances: None,
-            revalidate_only: false,
         }
     }
 }
@@ -31,6 +26,8 @@ impl Default for TreeRunConfig {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CaInstanceHandle {
     pub depth: usize,
+    pub tal_id: String,
+    pub parent_manifest_rsync_uri: Option<String>,
     /// DER bytes of the CA certificate for this CA instance.
     pub ca_certificate_der: Vec<u8>,
     /// rsync URI of this CA certificate object (where it is published).
@@ -59,15 +56,15 @@ impl CaInstanceHandle {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct PublicationPointRunResult {
     pub source: PublicationPointSource,
-    pub pack: FetchCachePpPack,
+    pub snapshot: Option<PublicationPointSnapshot>,
     pub warnings: Vec<Warning>,
     pub objects: ObjectsOutput,
     pub audit: PublicationPointAudit,
-    /// Candidate child CA instances discovered from this publication point.
+    /// Candidate child CA instances to enqueue after this publication point completes.
     ///
-    /// RFC 9286 §6.6 restriction is enforced by the tree engine: if this
-    /// publication point used fetch_cache_pp due to failed fetch, children MUST NOT
-    /// be enqueued/processed in this run.
+    /// - For `Fresh`, these are discovered from the current validated publication point.
+    /// - For `VcirCurrentInstance`, these are restored from the current instance VCIR and then
+    ///   processed fresh-first when their own turn arrives.
     pub discovered_children: Vec<DiscoveredChildCaInstance>,
 }
 
@@ -187,37 +184,25 @@ pub fn run_tree_serial_audit(
         audit.discovered_from = node.discovered_from.clone();
         publication_points.push(audit);
 
-        let enqueue_children =
-            res.source == PublicationPointSource::Fresh || config.revalidate_only;
-        if !enqueue_children && !res.discovered_children.is_empty() {
-            warnings.push(
-                Warning::new("skipping child CA discovery due to failed fetch cache use")
-                    .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")])
-                    .with_context(&ca.manifest_rsync_uri),
-            );
-        }
-
-        if enqueue_children {
-            let mut children = res.discovered_children;
-            children.sort_by(|a, b| {
-                a.handle
-                    .manifest_rsync_uri
-                    .cmp(&b.handle.manifest_rsync_uri)
-                    .then_with(|| {
-                        a.discovered_from
-                            .child_ca_certificate_rsync_uri
-                            .cmp(&b.discovered_from.child_ca_certificate_rsync_uri)
-                    })
+        let mut children = res.discovered_children;
+        children.sort_by(|a, b| {
+            a.handle
+                .manifest_rsync_uri
+                .cmp(&b.handle.manifest_rsync_uri)
+                .then_with(|| {
+                    a.discovered_from
+                        .child_ca_certificate_rsync_uri
+                        .cmp(&b.discovered_from.child_ca_certificate_rsync_uri)
+                })
+        });
+        for child in children {
+            queue.push_back(QueuedCaInstance {
+                id: next_id,
+                handle: child.handle.with_depth(ca.depth + 1),
+                parent_id: Some(node.id),
+                discovered_from: Some(child.discovered_from),
             });
-            for child in children {
-                queue.push_back(QueuedCaInstance {
-                    id: next_id,
-                    handle: child.handle.with_depth(ca.depth + 1),
-                    parent_id: Some(node.id),
-                    discovered_from: Some(child.discovered_from),
-                });
-                next_id += 1;
-            }
+            next_id += 1;
         }
     }