add delta sync and fail fetch process

src/bin/db_stats.rs

@@ -0,0 +1,125 @@
+use std::path::PathBuf;
+
+use rocksdb::{ColumnFamilyDescriptor, DB, IteratorMode, Options};
+
+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";
+
+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);
+    }
+    _set(opts);
+}
+
+fn usage() -> String {
+    let bin = "db_stats";
+    format!(
+        "\
+Usage:
+  {bin} --db <path> [--exact]
+
+Options:
+  --db <path>     RocksDB directory
+  --exact         Iterate to count keys (slower; default uses RocksDB estimates)
+  --help          Show this help
+"
+    )
+}
+
+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)
+        .ok_or_else(|| format!("missing column family: {cf_name}"))?;
+    Ok(db.property_int_value_cf(cf, "rocksdb.estimate-num-keys")?)
+}
+
+fn exact_keys(db: &DB, cf_name: &str) -> Result<u64, Box<dyn std::error::Error>> {
+    let cf = db
+        .cf_handle(cf_name)
+        .ok_or_else(|| format!("missing column family: {cf_name}"))?;
+    let mode = IteratorMode::Start;
+    let mut count = 0u64;
+    for res in db.iterator_cf(cf, mode) {
+        res?;
+        count += 1;
+    }
+    Ok(count)
+}
+
+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") {
+        print!("{}", usage());
+        return Ok(());
+    }
+
+    let mut db_path: Option<PathBuf> = None;
+    let mut exact = false;
+    let mut i = 1usize;
+    while i < argv.len() {
+        match argv[i].as_str() {
+            "--db" => {
+                i += 1;
+                let v = argv.get(i).ok_or("--db requires a value")?;
+                db_path = Some(PathBuf::from(v));
+            }
+            "--exact" => exact = true,
+            other => return Err(format!("unknown argument: {other}\n\n{}", usage()).into()),
+        }
+        i += 1;
+    }
+
+    let db_path = db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?;
+
+    let mut opts = Options::default();
+    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,
+    ];
+
+    println!("db={}", db_path.display());
+    println!("mode={}", if exact { "exact" } else { "estimate" });
+
+    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)
+        };
+        total = total.saturating_add(n);
+        println!("{name}={n}");
+    }
+    println!("total={total}");
+
+    // Also print # of SST files (useful sanity signal).
+    let live = db.live_files()?;
+    println!("sst_files={}", live.len());
+
+    Ok(())
+}

src/bin/rrdp_state_dump.rs

@@ -0,0 +1,85 @@
+use std::path::PathBuf;
+
+use rocksdb::{ColumnFamilyDescriptor, DB, IteratorMode, Options};
+
+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);
+    }
+    _set(opts);
+}
+
+fn usage() -> String {
+    let bin = "rrdp_state_dump";
+    format!(
+        "\
+Usage:
+  {bin} --db <path>
+
+Options:
+  --db <path>     RocksDB directory
+  --help          Show this help
+"
+    )
+}
+
+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") {
+        print!("{}", usage());
+        return Ok(());
+    }
+
+    let mut db_path: Option<PathBuf> = None;
+    let mut i = 1usize;
+    while i < argv.len() {
+        match argv[i].as_str() {
+            "--db" => {
+                i += 1;
+                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()),
+        }
+        i += 1;
+    }
+
+    let db_path = db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?;
+
+    let mut opts = Options::default();
+    opts.create_if_missing(false);
+    opts.create_missing_column_families(false);
+
+    // 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)?;
+    let cf = db
+        .cf_handle("rrdp_state")
+        .ok_or("missing column family: rrdp_state")?;
+
+    let mut out: Vec<(String, u64, String)> = 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));
+    }
+
+    out.sort_by(|a, b| a.0.cmp(&b.0));
+    for (k, serial, session) in out {
+        println!("{k}\t{serial}\t{session}");
+    }
+    Ok(())
+}

src/cli.rs

@@ -27,6 +27,9 @@ pub struct CliArgs {
 
     pub rsync_local_dir: Option<PathBuf>,
 
+    pub http_timeout_secs: u64,
+    pub rsync_timeout_secs: u64,
+
     pub max_depth: Option<usize>,
     pub max_instances: Option<usize>,
     pub validation_time: Option<time::OffsetDateTime>,
@@ -50,6 +53,8 @@ Options:
   --ta-path <path>           TA certificate DER file path (offline-friendly)
 
   --rsync-local-dir <path>   Use LocalDirRsyncFetcher rooted at this directory (offline tests)
+  --http-timeout-secs <n>    HTTP fetch timeout seconds (default: 20)
+  --rsync-timeout-secs <n>   rsync I/O timeout seconds (default: 60)
   --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)
@@ -69,6 +74,8 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
     let mut report_json_path: Option<PathBuf> = None;
 
     let mut rsync_local_dir: Option<PathBuf> = None;
+    let mut http_timeout_secs: u64 = 20;
+    let mut rsync_timeout_secs: u64 = 60;
     let mut max_depth: Option<usize> = None;
     let mut max_instances: Option<usize> = None;
     let mut validation_time: Option<time::OffsetDateTime> = None;
@@ -113,6 +120,20 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
                 let v = argv.get(i).ok_or("--rsync-local-dir requires a value")?;
                 rsync_local_dir = Some(PathBuf::from(v));
             }
+            "--http-timeout-secs" => {
+                i += 1;
+                let v = argv.get(i).ok_or("--http-timeout-secs requires a value")?;
+                http_timeout_secs = v
+                    .parse::<u64>()
+                    .map_err(|_| format!("invalid --http-timeout-secs: {v}"))?;
+            }
+            "--rsync-timeout-secs" => {
+                i += 1;
+                let v = argv.get(i).ok_or("--rsync-timeout-secs requires a value")?;
+                rsync_timeout_secs = v
+                    .parse::<u64>()
+                    .map_err(|_| format!("invalid --rsync-timeout-secs: {v}"))?;
+            }
             "--max-depth" => {
                 i += 1;
                 let v = argv.get(i).ok_or("--max-depth requires a value")?;
@@ -166,6 +187,8 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
         policy_path,
         report_json_path,
         rsync_local_dir,
+        http_timeout_secs,
+        rsync_timeout_secs,
         max_depth,
         max_instances,
         validation_time,
@@ -289,7 +312,11 @@ pub fn run(argv: &[String]) -> Result<(), String> {
         .unwrap_or_else(time::OffsetDateTime::now_utc);
 
     let store = RocksStore::open(&args.db_path).map_err(|e| e.to_string())?;
-    let http = BlockingHttpFetcher::new(HttpFetcherConfig::default()).map_err(|e| e.to_string())?;
+    let http = BlockingHttpFetcher::new(HttpFetcherConfig {
+        timeout: std::time::Duration::from_secs(args.http_timeout_secs.max(1)),
+        ..HttpFetcherConfig::default()
+    })
+    .map_err(|e| e.to_string())?;
 
     let config = TreeRunConfig {
         max_depth: args.max_depth,
@@ -334,7 +361,10 @@ pub fn run(argv: &[String]) -> Result<(), String> {
             _ => unreachable!("validated by parse_args"),
         }
     } else {
-        let rsync = SystemRsyncFetcher::new(SystemRsyncConfig::default());
+        let rsync = SystemRsyncFetcher::new(SystemRsyncConfig {
+            timeout: std::time::Duration::from_secs(args.rsync_timeout_secs.max(1)),
+            ..SystemRsyncConfig::default()
+        });
         match (
             args.tal_url.as_ref(),
             args.tal_path.as_ref(),

src/storage.rs

@@ -11,6 +11,9 @@ use std::collections::HashSet;
 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";
+
+const RRDP_OBJECT_INDEX_PREFIX: &[u8] = b"rrdp_obj:";
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct FetchCachePpKey(String);
@@ -47,6 +50,12 @@ pub mod pack {
     pub use super::{FetchCachePpPack, PackDecodeError, PackFile, PackTime};
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum RrdpDeltaOp {
+    Upsert { rsync_uri: String, bytes: Vec<u8> },
+    Delete { rsync_uri: String },
+}
+
 impl RocksStore {
     pub fn open(path: &Path) -> StorageResult<Self> {
         let mut base_opts = Options::default();
@@ -56,13 +65,24 @@ impl RocksStore {
         // Prefer conservative compression; may be overridden later.
         base_opts.set_compression_type(DBCompressionType::Lz4);
 
-        // Best-effort BlobDB enablement (ignored if bindings don't support it).
+        // Blob files / BlobDB enablement.
+        //
+        // IMPORTANT: `enable_blob_files` is a *column family option* in RocksDB. Setting it only
+        // on the DB's base options is not sufficient; every CF that stores values must enable it.
         enable_blobdb_if_supported(&mut base_opts);
 
+        fn cf_opts() -> Options {
+            let mut opts = Options::default();
+            opts.set_compression_type(DBCompressionType::Lz4);
+            enable_blobdb_if_supported(&mut opts);
+            opts
+        }
+
         let cfs = vec![
-            ColumnFamilyDescriptor::new(CF_RAW_OBJECTS, Options::default()),
-            ColumnFamilyDescriptor::new(CF_FETCH_CACHE_PP, Options::default()),
-            ColumnFamilyDescriptor::new(CF_RRDP_STATE, Options::default()),
+            ColumnFamilyDescriptor::new(CF_RAW_OBJECTS, cf_opts()),
+            ColumnFamilyDescriptor::new(CF_FETCH_CACHE_PP, cf_opts()),
+            ColumnFamilyDescriptor::new(CF_RRDP_STATE, cf_opts()),
+            ColumnFamilyDescriptor::new(CF_RRDP_OBJECT_INDEX, cf_opts()),
         ];
 
         let db = DB::open_cf_descriptors(&base_opts, path, cfs)
@@ -148,6 +168,167 @@ impl RocksStore {
         Ok(())
     }
 
+    fn rrdp_object_index_key(notification_uri: &str, rsync_uri: &str) -> Vec<u8> {
+        let mut out = Vec::with_capacity(
+            RRDP_OBJECT_INDEX_PREFIX.len() + notification_uri.len() + 1 + rsync_uri.len(),
+        );
+        out.extend_from_slice(RRDP_OBJECT_INDEX_PREFIX);
+        out.extend_from_slice(notification_uri.as_bytes());
+        out.push(0);
+        out.extend_from_slice(rsync_uri.as_bytes());
+        out
+    }
+
+    fn rrdp_object_index_prefix(notification_uri: &str) -> Vec<u8> {
+        let mut out =
+            Vec::with_capacity(RRDP_OBJECT_INDEX_PREFIX.len() + notification_uri.len() + 1);
+        out.extend_from_slice(RRDP_OBJECT_INDEX_PREFIX);
+        out.extend_from_slice(notification_uri.as_bytes());
+        out.push(0);
+        out
+    }
+
+    #[allow(dead_code)]
+    pub fn rrdp_object_index_contains(
+        &self,
+        notification_uri: &str,
+        rsync_uri: &str,
+    ) -> StorageResult<bool> {
+        let cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
+        let k = Self::rrdp_object_index_key(notification_uri, rsync_uri);
+        Ok(self
+            .db
+            .get_cf(cf, k)
+            .map_err(|e| StorageError::RocksDb(e.to_string()))?
+            .is_some())
+    }
+
+    #[allow(dead_code)]
+    pub fn rrdp_object_index_iter(
+        &self,
+        notification_uri: &str,
+    ) -> StorageResult<impl Iterator<Item = String> + '_> {
+        let cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
+        let prefix = Self::rrdp_object_index_prefix(notification_uri);
+        let prefix_len = prefix.len();
+        let mode = IteratorMode::From(prefix.as_slice(), Direction::Forward);
+        Ok(self
+            .db
+            .iterator_cf(cf, mode)
+            .take_while(move |res| match res {
+                Ok((k, _v)) => k.starts_with(prefix.as_slice()),
+                Err(_) => false,
+            })
+            .filter_map(move |res| {
+                let (k, _v) = res.ok()?;
+                let rsync_part = k.get(prefix_len..)?;
+                let s = std::str::from_utf8(rsync_part).ok()?;
+                Some(s.to_string())
+            }))
+    }
+
+    #[allow(dead_code)]
+    pub fn rrdp_object_index_clear(&self, notification_uri: &str) -> StorageResult<usize> {
+        let cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
+        let prefix = Self::rrdp_object_index_prefix(notification_uri);
+        let mode = IteratorMode::From(prefix.as_slice(), Direction::Forward);
+        let keys: Vec<Box<[u8]>> = self
+            .db
+            .iterator_cf(cf, mode)
+            .take_while(|res| match res {
+                Ok((k, _v)) => k.starts_with(prefix.as_slice()),
+                Err(_) => false,
+            })
+            .filter_map(|res| res.ok().map(|(k, _v)| k))
+            .collect();
+
+        if keys.is_empty() {
+            return Ok(0);
+        }
+
+        let mut batch = WriteBatch::default();
+        for k in &keys {
+            batch.delete_cf(cf, k);
+        }
+        self.write_batch(batch)?;
+        Ok(keys.len())
+    }
+
+    /// Apply an RRDP snapshot as a complete repository state for this `notification_uri`.
+    ///
+    /// This updates:
+    /// - `raw_objects` (publish all objects, delete objects that were present in the previous
+    ///   snapshot state but absent from this snapshot)
+    /// - `rrdp_object_index` membership (used to scope deletes to this RRDP repository)
+    ///
+    /// RFC 8182 §3.5.2.1: snapshots reflect the complete and current repository contents.
+    pub fn apply_rrdp_snapshot(
+        &self,
+        notification_uri: &str,
+        published: &[(String, Vec<u8>)],
+    ) -> StorageResult<usize> {
+        let raw_cf = self.cf(CF_RAW_OBJECTS)?;
+        let idx_cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
+
+        let mut new_set: HashSet<&str> = HashSet::with_capacity(published.len());
+        for (u, _b) in published {
+            new_set.insert(u.as_str());
+        }
+
+        let old_uris: Vec<String> = self.rrdp_object_index_iter(notification_uri)?.collect();
+
+        let mut batch = WriteBatch::default();
+
+        for old in &old_uris {
+            if !new_set.contains(old.as_str()) {
+                batch.delete_cf(raw_cf, old.as_bytes());
+                let k = Self::rrdp_object_index_key(notification_uri, old.as_str());
+                batch.delete_cf(idx_cf, k);
+            }
+        }
+
+        for (uri, bytes) in published {
+            batch.put_cf(raw_cf, uri.as_bytes(), bytes.as_slice());
+            let k = Self::rrdp_object_index_key(notification_uri, uri.as_str());
+            batch.put_cf(idx_cf, k, b"");
+        }
+
+        self.write_batch(batch)?;
+        Ok(published.len())
+    }
+
+    pub fn apply_rrdp_delta(
+        &self,
+        notification_uri: &str,
+        ops: &[RrdpDeltaOp],
+    ) -> StorageResult<usize> {
+        if ops.is_empty() {
+            return Ok(0);
+        }
+
+        let raw_cf = self.cf(CF_RAW_OBJECTS)?;
+        let idx_cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
+
+        let mut batch = WriteBatch::default();
+        for op in ops {
+            match op {
+                RrdpDeltaOp::Upsert { rsync_uri, bytes } => {
+                    batch.put_cf(raw_cf, rsync_uri.as_bytes(), bytes.as_slice());
+                    let k = Self::rrdp_object_index_key(notification_uri, rsync_uri.as_str());
+                    batch.put_cf(idx_cf, k, b"");
+                }
+                RrdpDeltaOp::Delete { rsync_uri } => {
+                    batch.delete_cf(raw_cf, rsync_uri.as_bytes());
+                    let k = Self::rrdp_object_index_key(notification_uri, rsync_uri.as_str());
+                    batch.delete_cf(idx_cf, k);
+                }
+            }
+        }
+
+        self.write_batch(batch)?;
+        Ok(ops.len())
+    }
+
     #[allow(dead_code)]
     pub fn raw_iter_prefix<'a>(
         &'a self,
@@ -419,3 +600,189 @@ fn compute_sha256_32(bytes: &[u8]) -> [u8; 32] {
     out.copy_from_slice(&digest);
     out
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn rrdp_object_index_and_snapshot_delta_helpers_work_end_to_end() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let store = RocksStore::open(td.path()).expect("open rocksdb");
+
+        let notification_uri = "https://rrdp.example.test/notification.xml";
+        let u1 = "rsync://rpki.example.test/repo/obj1.cer".to_string();
+        let u2 = "rsync://rpki.example.test/repo/obj2.mft".to_string();
+
+        // Empty clear is a fast no-op.
+        assert_eq!(
+            store
+                .rrdp_object_index_clear(notification_uri)
+                .expect("clear empty"),
+            0
+        );
+
+        // Snapshot publishes two objects.
+        let published_v1 = vec![
+            (u1.clone(), vec![1u8, 2, 3]),
+            (u2.clone(), vec![9u8, 8, 7]),
+        ];
+        let n = store
+            .apply_rrdp_snapshot(notification_uri, &published_v1)
+            .expect("apply snapshot v1");
+        assert_eq!(n, 2);
+
+        assert_eq!(
+            store.get_raw(&u1).expect("get_raw u1"),
+            Some(vec![1u8, 2, 3])
+        );
+        assert_eq!(
+            store.get_raw(&u2).expect("get_raw u2"),
+            Some(vec![9u8, 8, 7])
+        );
+        assert!(
+            store
+                .rrdp_object_index_contains(notification_uri, &u1)
+                .expect("contains u1")
+        );
+        assert!(
+            store
+                .rrdp_object_index_contains(notification_uri, &u2)
+                .expect("contains u2")
+        );
+
+        let mut listed: Vec<String> = store
+            .rrdp_object_index_iter(notification_uri)
+            .expect("iter")
+            .collect();
+        listed.sort();
+        assert_eq!(listed, vec![u1.clone(), u2.clone()]);
+
+        // Snapshot v2 removes u1 and updates u2.
+        let published_v2 = vec![(u2.clone(), vec![0u8, 1, 2, 3])];
+        store
+            .apply_rrdp_snapshot(notification_uri, &published_v2)
+            .expect("apply snapshot v2");
+        assert_eq!(store.get_raw(&u1).expect("get_raw removed"), None);
+        assert_eq!(
+            store.get_raw(&u2).expect("get_raw updated"),
+            Some(vec![0u8, 1, 2, 3])
+        );
+
+        // Delta can upsert and delete, and uses the index to scope membership.
+        let u3 = "rsync://rpki.example.test/repo/obj3.crl".to_string();
+        let ops = vec![
+            RrdpDeltaOp::Upsert {
+                rsync_uri: u3.clone(),
+                bytes: vec![4u8, 5, 6],
+            },
+            RrdpDeltaOp::Delete { rsync_uri: u2.clone() },
+        ];
+        let applied = store
+            .apply_rrdp_delta(notification_uri, &ops)
+            .expect("apply delta");
+        assert_eq!(applied, 2);
+        assert_eq!(store.get_raw(&u2).expect("get_raw deleted"), None);
+        assert_eq!(
+            store.get_raw(&u3).expect("get_raw u3"),
+            Some(vec![4u8, 5, 6])
+        );
+
+        // Prefix iterators yield only matching keys.
+        let prefix = b"rsync://rpki.example.test/repo/";
+        let mut got: Vec<String> = store
+            .raw_iter_prefix(prefix)
+            .expect("raw_iter_prefix")
+            .map(|(k, _v)| String::from_utf8(k.to_vec()).expect("utf8 key"))
+            .collect();
+        got.sort();
+        assert_eq!(got, vec![u3.clone()]);
+
+        let all: Vec<String> = store
+            .raw_iter_all()
+            .expect("raw_iter_all")
+            .map(|(k, _v)| String::from_utf8(k.to_vec()).expect("utf8 key"))
+            .collect();
+        assert!(all.contains(&u3));
+
+        // Clearing removes all index entries for this RRDP repository.
+        let cleared = store
+            .rrdp_object_index_clear(notification_uri)
+            .expect("clear");
+        assert!(cleared >= 1);
+        assert!(
+            !store
+                .rrdp_object_index_contains(notification_uri, &u3)
+                .expect("contains after clear")
+        );
+    }
+
+    fn minimal_valid_pack() -> FetchCachePpPack {
+        let now = time::OffsetDateTime::now_utc();
+        FetchCachePpPack {
+            format_version: FetchCachePpPack::FORMAT_VERSION_V1,
+            manifest_rsync_uri: "rsync://example.test/repo/pp/manifest.mft".to_string(),
+            publication_point_rsync_uri: "rsync://example.test/repo/pp/".to_string(),
+            manifest_number_be: vec![1],
+            this_update: PackTime::from_utc_offset_datetime(now),
+            next_update: PackTime::from_utc_offset_datetime(now + time::Duration::hours(1)),
+            verified_at: PackTime::from_utc_offset_datetime(now),
+            manifest_bytes: vec![0x01],
+            files: vec![PackFile::from_bytes_compute_sha256(
+                "rsync://example.test/repo/pp/a.roa",
+                vec![1u8, 2, 3],
+            )],
+        }
+    }
+
+    #[test]
+    fn apply_rrdp_delta_empty_ops_is_noop() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let store = RocksStore::open(td.path()).expect("open rocksdb");
+        let n = store
+            .apply_rrdp_delta("https://rrdp.example.test/notification.xml", &[])
+            .expect("apply empty delta");
+        assert_eq!(n, 0);
+    }
+
+    #[test]
+    fn fetch_cache_pp_pack_rejects_invalid_time_fields_and_duplicates() {
+        // Invalid `next_update`.
+        let mut p = minimal_valid_pack();
+        p.next_update.rfc3339_utc = "not-a-time".to_string();
+        let e = p.validate_internal().unwrap_err().to_string();
+        assert!(e.contains("invalid time field next_update"), "{e}");
+
+        // Invalid `verified_at`.
+        let mut p = minimal_valid_pack();
+        p.verified_at.rfc3339_utc = "also-not-a-time".to_string();
+        let e = p.validate_internal().unwrap_err().to_string();
+        assert!(e.contains("invalid time field verified_at"), "{e}");
+
+        // Duplicate file rsync URI.
+        let mut p = minimal_valid_pack();
+        let f = p.files[0].clone();
+        p.files.push(f);
+        let e = p.validate_internal().unwrap_err().to_string();
+        assert!(e.contains("duplicate file rsync uri"), "{e}");
+    }
+
+    #[test]
+    fn fetch_cache_pp_iter_all_lists_keys() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let store = RocksStore::open(td.path()).expect("open rocksdb");
+
+        let key = FetchCachePpKey::from_manifest_rsync_uri("rsync://example.test/repo/pp/manifest.mft");
+        let bytes = minimal_valid_pack().encode().expect("encode pack");
+        store
+            .put_fetch_cache_pp(&key, &bytes)
+            .expect("put fetch_cache_pp");
+
+        let keys: Vec<String> = store
+            .fetch_cache_pp_iter_all()
+            .expect("iter all")
+            .map(|(k, _v)| String::from_utf8(k.to_vec()).expect("utf8 key"))
+            .collect();
+        assert!(keys.iter().any(|k| k == key.as_str()), "missing key in iterator");
+    }
+}

src/sync/repo.rs

@@ -3,8 +3,7 @@ use crate::policy::{Policy, SyncPreference};
 use crate::report::{RfcRef, Warning};
 use crate::storage::RocksStore;
 use crate::sync::rrdp::{
-    Fetcher as HttpFetcher, RrdpState, RrdpSyncError, parse_notification_snapshot,
-    sync_from_notification_snapshot,
+    Fetcher as HttpFetcher, RrdpSyncError, sync_from_notification,
 };
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -91,24 +90,7 @@ fn try_rrdp_sync(
         .fetch(notification_uri)
         .map_err(RrdpSyncError::Fetch)?;
 
-    // Stage2 snapshot-only optimization: if the stored RRDP state matches the current notification's
-    // session_id+serial, skip snapshot fetch/apply. This avoids repeatedly downloading/applying the
-    // same snapshot when traversing many CA instances sharing an RRDP endpoint.
-    //
-    // RFC 8182 §3.4.1-§3.4.3: clients use notification to discover snapshot and can avoid fetching
-    // snapshot when serial hasn't advanced.
-    if let Ok(notif) = parse_notification_snapshot(&notification_xml) {
-        if let Ok(Some(state_bytes)) = store.get_rrdp_state(notification_uri) {
-            if let Ok(state) = RrdpState::decode(&state_bytes) {
-                if state.session_id == notif.session_id.to_string() && state.serial == notif.serial
-                {
-                    return Ok(0);
-                }
-            }
-        }
-    }
-
-    sync_from_notification_snapshot(store, notification_uri, &notification_xml, http_fetcher)
+    sync_from_notification(store, notification_uri, &notification_xml, http_fetcher)
 }
 
 fn rsync_sync_into_raw_objects(

src/validation/ca_path.rs

@@ -7,6 +7,8 @@ use crate::data_model::rc::{
 };
 use x509_parser::prelude::{FromDer, X509Certificate};
 
+use crate::validation::x509_name::x509_names_equivalent;
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ValidatedSubordinateCa {
     pub child_ca: ResourceCertificate,
@@ -135,7 +137,7 @@ pub fn validate_subordinate_ca_cert(
         return Err(CaPathError::IssuerNotCa);
     }
 
-    if child_ca.tbs.issuer_name != issuer_ca.tbs.subject_name {
+    if !x509_names_equivalent(&child_ca.tbs.issuer_name, &issuer_ca.tbs.subject_name) {
         return Err(CaPathError::IssuerSubjectMismatch {
             child_issuer_dn: child_ca.tbs.issuer_name.to_string(),
             issuer_subject_dn: issuer_ca.tbs.subject_name.to_string(),
@@ -697,6 +699,7 @@ mod tests {
     };
     use crate::data_model::common::X509NameDer;
     use der_parser::num_bigint::BigUint;
+    use std::process::Command;
     fn dummy_cert(
         kind: ResourceCertKind,
         subject_dn: &str,
@@ -736,6 +739,158 @@ mod tests {
         }
     }
 
+    fn openssl_available() -> bool {
+        Command::new("openssl")
+            .arg("version")
+            .output()
+            .map(|o| o.status.success())
+            .unwrap_or(false)
+    }
+
+    fn write_cert_der_with_addext(dir: &std::path::Path, addext: Option<&str>) -> Vec<u8> {
+        assert!(openssl_available(), "openssl is required for this test");
+        let key = dir.join("k.pem");
+        let cert = dir.join("c.pem");
+        let der = dir.join("c.der");
+
+        let mut cmd = Command::new("openssl");
+        cmd.arg("req")
+            .arg("-x509")
+            .arg("-newkey")
+            .arg("rsa:2048")
+            .arg("-nodes")
+            .arg("-keyout")
+            .arg(&key)
+            .arg("-subj")
+            .arg("/CN=ku")
+            .arg("-days")
+            .arg("1")
+            .arg("-out")
+            .arg(&cert);
+        if let Some(ext) = addext {
+            cmd.arg("-addext").arg(ext);
+        }
+        let out = cmd.output().expect("openssl req");
+        assert!(
+            out.status.success(),
+            "openssl req failed: {}",
+            String::from_utf8_lossy(&out.stderr)
+        );
+
+        let out = Command::new("openssl")
+            .arg("x509")
+            .arg("-in")
+            .arg(&cert)
+            .arg("-outform")
+            .arg("DER")
+            .arg("-out")
+            .arg(&der)
+            .output()
+            .expect("openssl x509");
+        assert!(
+            out.status.success(),
+            "openssl x509 failed: {}",
+            String::from_utf8_lossy(&out.stderr)
+        );
+        std::fs::read(&der).expect("read der")
+    }
+
+    fn gen_issuer_and_child_der(dir: &std::path::Path) -> (Vec<u8>, Vec<u8>, Vec<u8>) {
+        assert!(openssl_available(), "openssl is required for this test");
+        let issuer_key = dir.join("issuer.key");
+        let issuer_csr = dir.join("issuer.csr");
+        let issuer_pem = dir.join("issuer.pem");
+        let issuer_der = dir.join("issuer.der");
+
+        let child_key = dir.join("child.key");
+        let child_csr = dir.join("child.csr");
+        let child_pem = dir.join("child.pem");
+        let child_der = dir.join("child.der");
+
+        let other_key = dir.join("other.key");
+        let other_csr = dir.join("other.csr");
+        let other_pem = dir.join("other.pem");
+        let other_der = dir.join("other.der");
+
+        let run = |cmd: &mut Command| {
+            let out = cmd.output().expect("run openssl");
+            assert!(
+                out.status.success(),
+                "command failed: {:?}\nstderr={}",
+                cmd,
+                String::from_utf8_lossy(&out.stderr)
+            );
+        };
+
+        // Issuer self-signed.
+        run(Command::new("openssl").args(["genrsa", "-out"]).arg(&issuer_key).arg("2048"));
+        run(Command::new("openssl")
+            .args(["req", "-new", "-key"])
+            .arg(&issuer_key)
+            .args(["-subj", "/CN=issuer", "-out"])
+            .arg(&issuer_csr));
+        run(Command::new("openssl")
+            .args(["x509", "-req", "-in"])
+            .arg(&issuer_csr)
+            .args(["-signkey"])
+            .arg(&issuer_key)
+            .args(["-days", "1", "-out"])
+            .arg(&issuer_pem));
+        run(Command::new("openssl")
+            .args(["x509", "-in"])
+            .arg(&issuer_pem)
+            .args(["-outform", "DER", "-out"])
+            .arg(&issuer_der));
+
+        // Child signed by issuer.
+        run(Command::new("openssl").args(["genrsa", "-out"]).arg(&child_key).arg("2048"));
+        run(Command::new("openssl")
+            .args(["req", "-new", "-key"])
+            .arg(&child_key)
+            .args(["-subj", "/CN=child", "-out"])
+            .arg(&child_csr));
+        run(Command::new("openssl")
+            .args(["x509", "-req", "-in"])
+            .arg(&child_csr)
+            .args(["-CA"])
+            .arg(&issuer_pem)
+            .args(["-CAkey"])
+            .arg(&issuer_key)
+            .args(["-CAcreateserial", "-days", "1", "-out"])
+            .arg(&child_pem));
+        run(Command::new("openssl")
+            .args(["x509", "-in"])
+            .arg(&child_pem)
+            .args(["-outform", "DER", "-out"])
+            .arg(&child_der));
+
+        // Other self-signed issuer.
+        run(Command::new("openssl").args(["genrsa", "-out"]).arg(&other_key).arg("2048"));
+        run(Command::new("openssl")
+            .args(["req", "-new", "-key"])
+            .arg(&other_key)
+            .args(["-subj", "/CN=other", "-out"])
+            .arg(&other_csr));
+        run(Command::new("openssl")
+            .args(["x509", "-req", "-in"])
+            .arg(&other_csr)
+            .args(["-signkey"])
+            .arg(&other_key)
+            .args(["-days", "1", "-out"])
+            .arg(&other_pem));
+        run(Command::new("openssl")
+            .args(["x509", "-in"])
+            .arg(&other_pem)
+            .args(["-outform", "DER", "-out"])
+            .arg(&other_der));
+
+        (
+            std::fs::read(&issuer_der).expect("read issuer der"),
+            std::fs::read(&child_der).expect("read child der"),
+            std::fs::read(&other_der).expect("read other der"),
+        )
+    }
+
     #[test]
     fn resolve_child_ip_resources_rejects_inherit_without_parent_effective_resources() {
         let child = IpResourceSet {
@@ -953,6 +1108,142 @@ mod tests {
         validate_child_crldp_contains_issuer_crl_uri(&child, "rsync://example.test/issuer.crl")
             .expect("crldp ok");
     }
+
+    #[test]
+    fn validate_child_ca_key_usage_accepts_only_keycertsign_and_crlsign_critical() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let der = write_cert_der_with_addext(
+            td.path(),
+            Some("keyUsage = critical, keyCertSign, cRLSign"),
+        );
+        validate_child_ca_key_usage(&der).expect("key usage ok");
+    }
+
+    #[test]
+    fn validate_child_ca_key_usage_rejects_missing_noncritical_and_invalid_bits() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let missing = write_cert_der_with_addext(td.path(), None);
+        let err = validate_child_ca_key_usage(&missing).unwrap_err();
+        assert!(matches!(err, CaPathError::KeyUsageMissing), "{err}");
+
+        let td = tempfile::tempdir().expect("tempdir");
+        let noncritical =
+            write_cert_der_with_addext(td.path(), Some("keyUsage = keyCertSign, cRLSign"));
+        let err = validate_child_ca_key_usage(&noncritical).unwrap_err();
+        assert!(matches!(err, CaPathError::KeyUsageNotCritical), "{err}");
+
+        let td = tempfile::tempdir().expect("tempdir");
+        let invalid = write_cert_der_with_addext(
+            td.path(),
+            Some("keyUsage = critical, keyCertSign, cRLSign, digitalSignature"),
+        );
+        let err = validate_child_ca_key_usage(&invalid).unwrap_err();
+        assert!(matches!(err, CaPathError::KeyUsageInvalidBits), "{err}");
+    }
+
+    #[test]
+    fn verify_cert_signature_with_issuer_accepts_valid_chain_and_rejects_wrong_issuer() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let (issuer, child, other) = gen_issuer_and_child_der(td.path());
+        verify_cert_signature_with_issuer(&child, &issuer).expect("signature ok");
+        let err = verify_cert_signature_with_issuer(&child, &other).unwrap_err();
+        assert!(matches!(err, CaPathError::ChildSignatureInvalid(_)), "{err}");
+    }
+
+    #[test]
+    fn resolve_child_ip_and_as_resources_success_paths() {
+        use crate::data_model::rc::{AsIdOrRange, IpAddressOrRange, IpPrefix};
+
+        let parent_ip = 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],
+                    },
+                )]),
+            }],
+        };
+
+        let child_ip_inherit = IpResourceSet {
+            families: vec![IpAddressFamily {
+                afi: Afi::Ipv4,
+                choice: IpAddressChoice::Inherit,
+            }],
+        };
+        let eff = resolve_child_ip_resources(Some(&child_ip_inherit), Some(&parent_ip))
+            .expect("inherit resolves")
+            .expect("some ip");
+        assert_eq!(eff.families.len(), 1);
+        assert!(matches!(
+            eff.families[0].choice,
+            IpAddressChoice::AddressesOrRanges(_)
+        ));
+
+        let child_ip_subset = 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],
+                    },
+                )]),
+            }],
+        };
+        resolve_child_ip_resources(Some(&child_ip_subset), Some(&parent_ip))
+            .expect("subset ok")
+            .expect("some");
+
+        let child_ip_bad = IpResourceSet {
+            families: vec![IpAddressFamily {
+                afi: Afi::Ipv4,
+                choice: IpAddressChoice::AddressesOrRanges(vec![IpAddressOrRange::Prefix(
+                    IpPrefix {
+                        afi: Afi::Ipv4,
+                        prefix_len: 16,
+                        addr: vec![11, 0, 0, 0],
+                    },
+                )]),
+            }],
+        };
+        let err = resolve_child_ip_resources(Some(&child_ip_bad), Some(&parent_ip)).unwrap_err();
+        assert!(matches!(err, CaPathError::ResourcesNotSubset), "{err}");
+
+        let parent_as = AsResourceSet {
+            asnum: Some(AsIdentifierChoice::AsIdsOrRanges(vec![AsIdOrRange::Range {
+                min: 1,
+                max: 100,
+            }])),
+            rdi: None,
+        };
+        let child_as_inherit = AsResourceSet {
+            asnum: Some(AsIdentifierChoice::Inherit),
+            rdi: None,
+        };
+        let eff_as = resolve_child_as_resources(Some(&child_as_inherit), Some(&parent_as))
+            .expect("inherit as")
+            .expect("some");
+        assert_eq!(eff_as.asnum, parent_as.asnum);
+
+        let child_as_subset = AsResourceSet {
+            asnum: Some(AsIdentifierChoice::AsIdsOrRanges(vec![AsIdOrRange::Id(50)])),
+            rdi: None,
+        };
+        resolve_child_as_resources(Some(&child_as_subset), Some(&parent_as))
+            .expect("subset as")
+            .expect("some");
+
+        let child_as_bad = AsResourceSet {
+            asnum: Some(AsIdentifierChoice::AsIdsOrRanges(vec![AsIdOrRange::Id(200)])),
+            rdi: None,
+        };
+        let err = resolve_child_as_resources(Some(&child_as_bad), Some(&parent_as)).unwrap_err();
+        assert!(matches!(err, CaPathError::ResourcesNotSubset), "{err}");
+    }
 }
 
 fn increment_bytes(v: &[u8]) -> Vec<u8> {

src/validation/cert_path.rs

@@ -5,6 +5,8 @@ use crate::data_model::rc::{
 };
 use x509_parser::prelude::{FromDer, X509Certificate};
 
+use crate::validation::x509_name::x509_names_equivalent;
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ValidatedEeCertPath {
     pub ee: ResourceCertificate,
@@ -112,7 +114,7 @@ pub fn validate_ee_cert_path(
         return Err(CertPathError::IssuerNotCa);
     }
 
-    if ee.tbs.issuer_name != issuer_ca.tbs.subject_name {
+    if !x509_names_equivalent(&ee.tbs.issuer_name, &issuer_ca.tbs.subject_name) {
         return Err(CertPathError::IssuerSubjectMismatch {
             ee_issuer_dn: ee.tbs.issuer_name.to_string(),
             issuer_subject_dn: issuer_ca.tbs.subject_name.to_string(),

src/validation/manifest.rs

@@ -23,6 +23,9 @@ pub struct PublicationPointResult {
 
 #[derive(Debug, thiserror::Error)]
 pub enum ManifestFreshError {
+    #[error("repo sync failed: {detail} (RFC 8182 §3.4.5; RFC 9286 §6.6)")]
+    RepoSyncFailed { detail: String },
+
     #[error(
         "manifest not found in raw_objects: {manifest_rsync_uri} (RFC 9286 §6.2; RFC 9286 §6.6)"
     )]
@@ -148,14 +151,44 @@ pub fn process_manifest_publication_point(
     issuer_ca_rsync_uri: Option<&str>,
     validation_time: time::OffsetDateTime,
 ) -> Result<PublicationPointResult, ManifestProcessError> {
-    let fresh = try_build_fresh_pack(
+    process_manifest_publication_point_after_repo_sync(
         store,
+        policy,
         manifest_rsync_uri,
         publication_point_rsync_uri,
         issuer_ca_der,
         issuer_ca_rsync_uri,
         validation_time,
-    );
+        true,
+        None,
+    )
+}
+
+pub fn process_manifest_publication_point_after_repo_sync(
+    store: &RocksStore,
+    policy: &Policy,
+    manifest_rsync_uri: &str,
+    publication_point_rsync_uri: &str,
+    issuer_ca_der: &[u8],
+    issuer_ca_rsync_uri: Option<&str>,
+    validation_time: time::OffsetDateTime,
+    repo_sync_ok: bool,
+    repo_sync_error: Option<&str>,
+) -> Result<PublicationPointResult, ManifestProcessError> {
+    let fresh = if repo_sync_ok {
+        try_build_fresh_pack(
+            store,
+            manifest_rsync_uri,
+            publication_point_rsync_uri,
+            issuer_ca_der,
+            issuer_ca_rsync_uri,
+            validation_time,
+        )
+    } else {
+        Err(ManifestFreshError::RepoSyncFailed {
+            detail: repo_sync_error.unwrap_or("repo sync failed").to_string(),
+        })
+    };
 
     match fresh {
         Ok(pack) => {
@@ -354,9 +387,16 @@ fn try_build_fresh_pack(
 
     // RFC 9286 §4.2.1: replay/rollback detection for manifestNumber and thisUpdate.
     //
-    // If a purported "new" manifest contains a manifestNumber equal to or lower than previously
-    // validated manifests, or a thisUpdate less recent than previously validated manifests,
-    // this is treated as a failed fetch and processing continues via the cached objects path (§6.6).
+    // Important nuance for revalidation across runs:
+    // - If the manifestNumber is equal to the previously validated manifestNumber *and* the
+    //   manifest bytes are identical, then this is the same manifest being revalidated and MUST
+    //   be accepted (otherwise, RPs would incorrectly treat stable repositories as "failed fetch"
+    //   and fall back to fetch_cache_pp).
+    // - If manifestNumber is equal but the manifest bytes differ, treat this as invalid (a
+    //   repository is not allowed to change the manifest while keeping the manifestNumber).
+    // - If manifestNumber is lower, treat as rollback and reject.
+    // - If manifestNumber is higher, require thisUpdate to be more recent than the previously
+    //   validated thisUpdate.
     let key = FetchCachePpKey::from_manifest_rsync_uri(manifest_rsync_uri);
     if let Some(old_bytes) = store.get_fetch_cache_pp(&key).ok().flatten() {
         if let Ok(old_pack) = FetchCachePpPack::decode(&old_bytes) {
@@ -365,28 +405,39 @@ fn try_build_fresh_pack(
             {
                 let new_num = manifest.manifest.manifest_number.bytes_be.as_slice();
                 let old_num = old_pack.manifest_number_be.as_slice();
-                if cmp_minimal_be_unsigned(new_num, old_num) != Ordering::Greater {
-                    return Err(ManifestFreshError::ManifestNumberNotIncreasing {
-                        old_hex: hex::encode_upper(old_num),
-                        new_hex: hex::encode_upper(new_num),
-                    });
-                }
-
-                let old_this_update = old_pack
-                    .this_update
-                    .parse()
-                    .expect("pack internal validation ensures this_update parses");
-                if this_update <= old_this_update {
-                    use time::format_description::well_known::Rfc3339;
-                    return Err(ManifestFreshError::ThisUpdateNotIncreasing {
-                        old_rfc3339_utc: old_this_update
-                            .to_offset(time::UtcOffset::UTC)
-                            .format(&Rfc3339)
-                            .expect("format old thisUpdate"),
-                        new_rfc3339_utc: this_update
-                            .format(&Rfc3339)
-                            .expect("format new thisUpdate"),
-                    });
+                match cmp_minimal_be_unsigned(new_num, old_num) {
+                    Ordering::Greater => {
+                        let old_this_update = old_pack
+                            .this_update
+                            .parse()
+                            .expect("pack internal validation ensures this_update parses");
+                        if this_update <= old_this_update {
+                            use time::format_description::well_known::Rfc3339;
+                            return Err(ManifestFreshError::ThisUpdateNotIncreasing {
+                                old_rfc3339_utc: old_this_update
+                                    .to_offset(time::UtcOffset::UTC)
+                                    .format(&Rfc3339)
+                                    .expect("format old thisUpdate"),
+                                new_rfc3339_utc: this_update
+                                    .format(&Rfc3339)
+                                    .expect("format new thisUpdate"),
+                            });
+                        }
+                    }
+                    Ordering::Equal => {
+                        if old_pack.manifest_bytes != manifest_bytes {
+                            return Err(ManifestFreshError::ManifestNumberNotIncreasing {
+                                old_hex: hex::encode_upper(old_num),
+                                new_hex: hex::encode_upper(new_num),
+                            });
+                        }
+                    }
+                    Ordering::Less => {
+                        return Err(ManifestFreshError::ManifestNumberNotIncreasing {
+                            old_hex: hex::encode_upper(old_num),
+                            new_hex: hex::encode_upper(new_num),
+                        });
+                    }
                 }
             }
         }

src/validation/mod.rs

@@ -8,3 +8,4 @@ pub mod run;
 pub mod run_tree_from_tal;
 pub mod tree;
 pub mod tree_runner;
+pub mod x509_name;

src/validation/tree_runner.rs

@@ -10,7 +10,9 @@ use crate::sync::repo::sync_publication_point;
 use crate::sync::rrdp::Fetcher;
 use crate::validation::ca_instance::ca_instance_uris_from_ca_certificate;
 use crate::validation::ca_path::{CaPathError, validate_subordinate_ca_cert};
-use crate::validation::manifest::{PublicationPointSource, process_manifest_publication_point};
+use crate::validation::manifest::{
+    PublicationPointSource, process_manifest_publication_point_after_repo_sync,
+};
 use crate::validation::objects::process_fetch_cache_pp_pack_for_issuer;
 use crate::validation::tree::{
     CaInstanceHandle, DiscoveredChildCaInstance, PublicationPointRunResult, PublicationPointRunner,
@@ -31,7 +33,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
     ) -> Result<PublicationPointRunResult, String> {
         let mut warnings: Vec<Warning> = Vec::new();
 
-        if let Err(e) = sync_publication_point(
+        let (repo_sync_ok, repo_sync_err): (bool, Option<String>) = match sync_publication_point(
             self.store,
             self.policy,
             ca.rrdp_notification_uri.as_deref(),
@@ -39,16 +41,23 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
             self.http_fetcher,
             self.rsync_fetcher,
         ) {
-            warnings.push(
-                Warning::new(format!(
-                    "repo sync failed (continuing with cached/raw data): {e}"
-                ))
-                .with_rfc_refs(&[RfcRef("RFC 8182 §3.4.5"), RfcRef("RFC 9286 §6.6")])
-                .with_context(&ca.rsync_base_uri),
-            );
-        }
+            Ok(res) => {
+                warnings.extend(res.warnings);
+                (true, None)
+            }
+            Err(e) => {
+                warnings.push(
+                    Warning::new(format!(
+                        "repo sync failed (continuing with fetch_cache_pp only): {e}"
+                    ))
+                    .with_rfc_refs(&[RfcRef("RFC 8182 §3.4.5"), RfcRef("RFC 9286 §6.6")])
+                    .with_context(&ca.rsync_base_uri),
+                );
+                (false, Some(e.to_string()))
+            }
+        };
 
-        let pp = match process_manifest_publication_point(
+        let pp = match process_manifest_publication_point_after_repo_sync(
             self.store,
             self.policy,
             &ca.manifest_rsync_uri,
@@ -56,6 +65,8 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
             &ca.ca_certificate_der,
             ca.ca_certificate_rsync_uri.as_deref(),
             self.validation_time,
+            repo_sync_ok,
+            repo_sync_err.as_deref(),
         ) {
             Ok(v) => v,
             Err(e) => return Err(format!("{e}")),
@@ -376,6 +387,7 @@ mod tests {
     use super::*;
     use crate::data_model::rc::ResourceCertificate;
     use crate::fetch::rsync::LocalDirRsyncFetcher;
+    use crate::fetch::rsync::{RsyncFetchError, RsyncFetcher};
     use crate::storage::{FetchCachePpPack, PackFile, PackTime};
     use crate::sync::rrdp::Fetcher;
     use crate::validation::tree::PublicationPointRunner;
@@ -389,6 +401,13 @@ mod tests {
         }
     }
 
+    struct FailingRsyncFetcher;
+    impl RsyncFetcher for FailingRsyncFetcher {
+        fn fetch_objects(&self, _rsync_base_uri: &str) -> Result<Vec<(String, Vec<u8>)>, RsyncFetchError> {
+            Err(RsyncFetchError::Fetch("rsync disabled in test".to_string()))
+        }
+    }
+
     fn openssl_available() -> bool {
         Command::new("openssl")
             .arg("version")
@@ -590,6 +609,34 @@ authorityKeyIdentifier = keyid:always
         }
     }
 
+    #[test]
+    fn never_http_fetcher_returns_error() {
+        let f = NeverHttpFetcher;
+        let err = f.fetch("https://example.test/").unwrap_err();
+        assert!(err.contains("disabled"), "{err}");
+    }
+
+    #[test]
+    fn kind_from_rsync_uri_classifies_known_extensions() {
+        assert_eq!(kind_from_rsync_uri("rsync://example.test/x.crl"), AuditObjectKind::Crl);
+        assert_eq!(kind_from_rsync_uri("rsync://example.test/x.cer"), AuditObjectKind::Certificate);
+        assert_eq!(kind_from_rsync_uri("rsync://example.test/x.roa"), AuditObjectKind::Roa);
+        assert_eq!(kind_from_rsync_uri("rsync://example.test/x.asa"), AuditObjectKind::Aspa);
+        assert_eq!(kind_from_rsync_uri("rsync://example.test/x.bin"), AuditObjectKind::Other);
+    }
+
+    #[test]
+    fn select_issuer_crl_from_pack_reports_missing_crldp_for_self_signed_cert() {
+        let ta_der = std::fs::read(std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(
+            "tests/fixtures/ta/apnic-ta.cer",
+        ))
+        .expect("read TA fixture");
+
+        let pack = dummy_pack_with_files(vec![]);
+        let err = select_issuer_crl_from_pack(&ta_der, &pack).unwrap_err();
+        assert!(err.contains("CRLDistributionPoints missing"), "{err}");
+    }
+
     #[test]
     fn select_issuer_crl_from_pack_finds_matching_crl() {
         // Use real fixtures to ensure child cert has CRLDP rsync URI and CRL exists.
@@ -801,6 +848,127 @@ authorityKeyIdentifier = keyid:always
         );
     }
 
+    #[test]
+    fn runner_when_repo_sync_fails_uses_fetch_cache_pp_and_skips_child_discovery() {
+        let fixture_dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .join("tests/fixtures/repository/rpki.cernet.net/repo/cernet/0");
+        assert!(fixture_dir.is_dir(), "fixture directory must exist");
+
+        let rsync_base_uri = "rsync://rpki.cernet.net/repo/cernet/0/".to_string();
+        let manifest_file = "05FC9C5B88506F7C0D3F862C8895BED67E9F8EBA.mft";
+        let manifest_rsync_uri = format!("{rsync_base_uri}{manifest_file}");
+
+        let fixture_manifest_bytes =
+            std::fs::read(fixture_dir.join(manifest_file)).expect("read manifest fixture");
+        let fixture_manifest =
+            crate::data_model::manifest::ManifestObject::decode_der(&fixture_manifest_bytes)
+                .expect("decode manifest fixture");
+        let validation_time = fixture_manifest.manifest.this_update + time::Duration::seconds(60);
+
+        let store_dir = tempfile::tempdir().expect("store dir");
+        let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
+        let policy = Policy {
+            sync_preference: crate::policy::SyncPreference::RsyncOnly,
+            ..Policy::default()
+        };
+
+        let issuer_ca_der = std::fs::read(
+            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(
+                "tests/fixtures/repository/rpki.apnic.net/repository/B527EF581D6611E2BB468F7C72FD1FF2/BfycW4hQb3wNP4YsiJW-1n6fjro.cer",
+            ),
+        )
+        .expect("read issuer ca fixture");
+        let issuer_ca = ResourceCertificate::decode_der(&issuer_ca_der).expect("decode issuer ca");
+
+        let handle = CaInstanceHandle {
+            depth: 0,
+            ca_certificate_der: issuer_ca_der,
+            ca_certificate_rsync_uri: Some("rsync://rpki.apnic.net/repository/B527EF581D6611E2BB468F7C72FD1FF2/BfycW4hQb3wNP4YsiJW-1n6fjro.cer".to_string()),
+            effective_ip_resources: issuer_ca.tbs.extensions.ip_resources.clone(),
+            effective_as_resources: issuer_ca.tbs.extensions.as_resources.clone(),
+            rsync_base_uri: rsync_base_uri.clone(),
+            manifest_rsync_uri: manifest_rsync_uri.clone(),
+            publication_point_rsync_uri: rsync_base_uri.clone(),
+            rrdp_notification_uri: None,
+        };
+
+        // First: successful repo sync to populate fetch_cache_pp.
+        let ok_runner = Rpkiv1PublicationPointRunner {
+            store: &store,
+            policy: &policy,
+            http_fetcher: &NeverHttpFetcher,
+            rsync_fetcher: &LocalDirRsyncFetcher::new(&fixture_dir),
+            validation_time,
+        };
+        let first = ok_runner
+            .run_publication_point(&handle)
+            .expect("first run ok");
+        assert_eq!(first.source, PublicationPointSource::Fresh);
+        assert!(first.discovered_children.is_empty(), "fixture has no child .cer");
+
+        // Second: repo sync fails, but we can still use fetch_cache_pp.
+        let bad_runner = Rpkiv1PublicationPointRunner {
+            store: &store,
+            policy: &policy,
+            http_fetcher: &NeverHttpFetcher,
+            rsync_fetcher: &FailingRsyncFetcher,
+            validation_time,
+        };
+        let second = bad_runner
+            .run_publication_point(&handle)
+            .expect("should fall back to fetch_cache_pp");
+        assert_eq!(second.source, PublicationPointSource::FetchCachePp);
+        assert!(second.discovered_children.is_empty());
+        assert!(
+            second
+                .warnings
+                .iter()
+                .any(|w| w.message.contains("repo sync failed")),
+            "expected warning about repo sync failure"
+        );
+    }
+
+    #[test]
+    fn build_publication_point_audit_emits_no_audit_entry_for_duplicate_pack_uri() {
+        let pack = dummy_pack_with_files(vec![
+            PackFile::from_bytes_compute_sha256("rsync://example.test/repo/dup.roa", vec![1u8]),
+            PackFile::from_bytes_compute_sha256("rsync://example.test/repo/dup.roa", vec![2u8]),
+        ]);
+        let pp = crate::validation::manifest::PublicationPointResult {
+            source: crate::validation::manifest::PublicationPointSource::FetchCachePp,
+            pack: pack.clone(),
+            warnings: Vec::new(),
+        };
+        let ca = CaInstanceHandle {
+            depth: 0,
+            ca_certificate_der: vec![1],
+            ca_certificate_rsync_uri: None,
+            effective_ip_resources: None,
+            effective_as_resources: None,
+            rsync_base_uri: pack.publication_point_rsync_uri.clone(),
+            manifest_rsync_uri: pack.manifest_rsync_uri.clone(),
+            publication_point_rsync_uri: pack.publication_point_rsync_uri.clone(),
+            rrdp_notification_uri: None,
+        };
+        let objects = crate::validation::objects::ObjectsOutput {
+            vrps: Vec::new(),
+            aspas: Vec::new(),
+            warnings: Vec::new(),
+            stats: crate::validation::objects::ObjectsStats::default(),
+            audit: Vec::new(),
+        };
+
+        let audit = build_publication_point_audit(&ca, &pp, &[], &objects, &[]);
+        assert_eq!(audit.source, "fetch_cache_pp");
+        assert!(
+            audit
+                .objects
+                .iter()
+                .any(|e| e.detail.as_deref() == Some("skipped: no audit entry")),
+            "expected a duplicate key to produce a 'no audit entry' placeholder"
+        );
+    }
+
     #[test]
     fn build_publication_point_audit_marks_invalid_crl_as_error_and_overlays_roa_audit() {
         let now = time::OffsetDateTime::now_utc();

src/validation/x509_name.rs

@@ -0,0 +1,85 @@
+use crate::data_model::common::X509NameDer;
+use x509_parser::prelude::FromDer;
+
+fn canonicalize(name: &X509NameDer) -> Option<String> {
+    let (rem, parsed) = x509_parser::x509::X509Name::from_der(name.as_raw()).ok()?;
+    if !rem.is_empty() {
+        return None;
+    }
+    Some(parsed.to_string())
+}
+
+/// Compare two X.509 distinguished names using a tolerant semantic comparison.
+///
+/// RPKI repositories in the wild sometimes encode the same name using different
+/// ASN.1 string types (e.g., PrintableString vs UTF8String) while remaining
+/// semantically equivalent. RFC 5280 path validation requires name matching, but
+/// DER byte equality is too strict for interoperability.
+pub fn x509_names_equivalent(a: &X509NameDer, b: &X509NameDer) -> bool {
+    let Some(ca) = canonicalize(a) else {
+        return a == b;
+    };
+    let Some(cb) = canonicalize(b) else {
+        return a == b;
+    };
+    ca == cb
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn x509_names_equivalent_falls_back_to_der_equality_when_parse_fails() {
+        // Invalid tag (not a SEQUENCE) makes x509-parser fail and forces DER equality fallback.
+        let a = X509NameDer(vec![0x01, 0x00]);
+        let b = X509NameDer(vec![0x01, 0x00]);
+        assert!(x509_names_equivalent(&a, &b));
+    }
+
+    #[test]
+    fn x509_names_equivalent_compares_semantic_names_when_parse_succeeds() {
+        let cert_der = std::fs::read(
+            "tests/fixtures/repository/rpki.apnic.net/repository/B527EF581D6611E2BB468F7C72FD1FF2/BfycW4hQb3wNP4YsiJW-1n6fjro.cer",
+        )
+        .expect("read certificate fixture");
+
+        let (_rem, cert) =
+            x509_parser::parse_x509_certificate(&cert_der).expect("parse certificate fixture");
+
+        let subject = X509NameDer(cert.tbs_certificate.subject.as_raw().to_vec());
+        let issuer = X509NameDer(cert.tbs_certificate.issuer.as_raw().to_vec());
+
+        assert!(x509_names_equivalent(&subject, &subject));
+        assert!(!x509_names_equivalent(&subject, &issuer));
+    }
+
+    #[test]
+    fn x509_names_equivalent_falls_back_when_name_has_trailing_bytes() {
+        // Use a real name DER and append a trailing byte so parsing yields leftover `rem`.
+        let cert_der = std::fs::read(
+            "tests/fixtures/repository/rpki.apnic.net/repository/B527EF581D6611E2BB468F7C72FD1FF2/BfycW4hQb3wNP4YsiJW-1n6fjro.cer",
+        )
+        .expect("read certificate fixture");
+        let (_rem, cert) =
+            x509_parser::parse_x509_certificate(&cert_der).expect("parse certificate fixture");
+        let mut name = cert.tbs_certificate.subject.as_raw().to_vec();
+        name.push(0x00);
+        let a = X509NameDer(name.clone());
+        let b = X509NameDer(name);
+        assert!(x509_names_equivalent(&a, &b));
+    }
+
+    #[test]
+    fn x509_names_equivalent_falls_back_when_one_side_fails_to_parse() {
+        let cert_der = std::fs::read(
+            "tests/fixtures/repository/rpki.apnic.net/repository/B527EF581D6611E2BB468F7C72FD1FF2/BfycW4hQb3wNP4YsiJW-1n6fjro.cer",
+        )
+        .expect("read certificate fixture");
+        let (_rem, cert) =
+            x509_parser::parse_x509_certificate(&cert_der).expect("parse certificate fixture");
+        let good = X509NameDer(cert.tbs_certificate.subject.as_raw().to_vec());
+        let bad = X509NameDer(vec![0x01, 0x00]);
+        assert!(!x509_names_equivalent(&good, &bad));
+    }
+}