20261015 实现RRDP失步检测与TA证书择优

src/cli/anchors.rs

@@ -44,6 +44,54 @@ pub(crate) fn cache_root(state: &Path) -> PathBuf {
     PathBuf::from(name)
 }
 
+struct CachedTa {
+    bytes: Vec<u8>,
+    info: Value,
+    fetched_at: Value,
+}
+
+fn load_cached_ta(
+    path: &Path,
+    tal: &Tal,
+    strict_name: bool,
+    validation_time: OffsetDateTime,
+) -> Result<CachedTa, String> {
+    let cached: Value = fs::read(path)
+        .map_err(|error| format!("read cached TA metadata failed: {error}"))
+        .and_then(|bytes| serde_json::from_slice(&bytes).map_err(|error| error.to_string()))?;
+    let encoded = cached["der"]
+        .as_str()
+        .ok_or("cached TA metadata has no DER value")?;
+    let bytes =
+        hex::decode(encoded).map_err(|error| format!("decode cached TA failed: {error}"))?;
+    let info = validate_certificate(tal, &bytes, strict_name, validation_time)?;
+    Ok(CachedTa {
+        bytes,
+        info,
+        fetched_at: cached["record"]["fetchedAt"].clone(),
+    })
+}
+
+/// draft-ietf-sidrops-rpki-ta-tiebreaker-06 §2. Both inputs must already
+/// pass the current TAL binding, certificate and validity checks. Compare DER
+/// fields, never the unauthenticated display dates in the JSON cache record.
+fn prefer_download(cached: &[u8], downloaded: &[u8]) -> Result<(bool, &'static str), String> {
+    use std::cmp::Ordering;
+    let (_, old) = X509Certificate::from_der(cached).map_err(|error| error.to_string())?;
+    let (_, new) = X509Certificate::from_der(downloaded).map_err(|error| error.to_string())?;
+    Ok(
+        match new.validity().not_before.cmp(&old.validity().not_before) {
+            Ordering::Greater => (true, "newer_not_before"),
+            Ordering::Less => (false, "cached_newer_not_before"),
+            Ordering::Equal => match new.validity().not_after.cmp(&old.validity().not_after) {
+                Ordering::Less => (true, "shorter_validity"),
+                Ordering::Greater => (false, "cached_shorter_validity"),
+                Ordering::Equal => (true, "most_recently_fetched"),
+            },
+        },
+    )
+}
+
 pub(crate) fn resolve_at(
     config: &Config,
     validation_time: OffsetDateTime,
@@ -99,7 +147,7 @@ fn resolve_with_at(
     let mut inputs = Vec::new();
     let mut records = Vec::new();
     for anchor in &config.anchors {
-        let mut record = json!({"id":anchor.id,"mode":if anchor.ta.is_some(){"fixed"}else{"auto"},"source":Value::Null,"certificateSha256":Value::Null,"fetchedAt":Value::Null,"notBefore":Value::Null,"notAfter":Value::Null,"refreshFailed":false,"errorCode":Value::Null});
+        let mut record = json!({"id":anchor.id,"mode":if anchor.ta.is_some(){"fixed"}else{"auto"},"source":Value::Null,"certificateSha256":Value::Null,"fetchedAt":Value::Null,"notBefore":Value::Null,"notAfter":Value::Null,"refreshFailed":false,"errorCode":Value::Null,"selectionReason":Value::Null});
         let attempt = (|| -> Result<PathBuf, String> {
             let tal = Tal::decode_bytes(&fs::read(&anchor.tal).map_err(|e| e.to_string())?)
                 .map_err(|e| e.to_string())?;
@@ -115,11 +163,15 @@ fn resolve_with_at(
                     .unwrap()
                     .extend(info.as_object().unwrap().clone());
                 record["source"] = json!("file");
+                record["selectionReason"] = json!("fixed_input");
                 return Ok(path.clone());
             }
             let key = hex::encode(Sha256::digest(&tal.subject_public_key_info_der));
             let path = root.join(format!("{key}.json"));
             let der_path = root.join(format!("{key}.cer"));
+            // A corrupt/expired/mismatched cache cannot veto a valid download.
+            // Keep its error for the existing failed-refresh fallback path.
+            let cached = load_cached_ta(&path, &tal, strict_name, validation_time);
             let mut last_error = "no usable TA URI".to_string();
             let mut error_code = "download_failed";
             for uri in &tal.ta_uris {
@@ -127,6 +179,29 @@ fn resolve_with_at(
                     Ok(bytes) => {
                         match validate_certificate(&tal, &bytes, strict_name, validation_time) {
                             Ok(info) => {
+                                let (download, reason) = match &cached {
+                                    Ok(old) => prefer_download(&old.bytes, &bytes)?,
+                                    Err(_) => (true, "no_valid_cache"),
+                                };
+                                record["selectionReason"] = json!(reason);
+                                if !download {
+                                    let old = cached.as_ref().map_err(|error| error.to_string())?;
+                                    record
+                                        .as_object_mut()
+                                        .unwrap()
+                                        .extend(old.info.as_object().unwrap().clone());
+                                    record["source"] = json!("cache");
+                                    record["fetchedAt"] = old.fetched_at.clone();
+                                    // Keep the authoritative cache and its actual fetch time;
+                                    // repair only the runtime DER projection if necessary.
+                                    super::publication::write_atomic(&der_path, &old.bytes)?;
+                                    crate::logging::emit(
+                                        crate::logging::Level::Info,
+                                        "ta_certificate_selected",
+                                        || json!({"tal_id":anchor.id,"source":"cache","reason":reason}),
+                                    );
+                                    return Ok(der_path);
+                                }
                                 record
                                     .as_object_mut()
                                     .unwrap()
@@ -155,40 +230,17 @@ fn resolve_with_at(
             record["refreshFailed"] = json!(true);
             record["errorCode"] = json!(error_code);
             if config.ta_refresh_policy == "allow-valid-cache" && path.exists() {
-                let cached: Value = match fs::read(&path)
-                    .map_err(|e| e.to_string())
-                    .and_then(|bytes| serde_json::from_slice(&bytes).map_err(|e| e.to_string()))
-                {
-                    Ok(value) => value,
-                    Err(error) => {
-                        record["errorCode"] = json!("cache_unusable");
-                        return Err(format!("read cached TA metadata failed: {error}"));
-                    }
-                };
-                let bytes = match cached["der"].as_str() {
-                    Some(encoded) => match hex::decode(encoded) {
-                        Ok(bytes) => bytes,
-                        Err(error) => {
-                            record["errorCode"] = json!("cache_unusable");
-                            return Err(format!("decode cached TA failed: {error}"));
-                        }
-                    },
-                    None => {
-                        record["errorCode"] = json!("cache_unusable");
-                        return Err("cached TA metadata has no DER value".into());
-                    }
-                };
-                let info = validate_certificate(&tal, &bytes, strict_name, validation_time)
-                    .inspect_err(|_| {
-                        record["errorCode"] = json!("cache_unusable");
-                    })?;
+                let cached = cached.inspect_err(|_| {
+                    record["errorCode"] = json!("cache_unusable");
+                })?;
                 record
                     .as_object_mut()
                     .unwrap()
-                    .extend(info.as_object().unwrap().clone());
+                    .extend(cached.info.as_object().unwrap().clone());
                 record["source"] = json!("cache");
-                record["fetchedAt"] = cached["record"]["fetchedAt"].clone();
-                super::publication::write_atomic(&der_path, &bytes)?;
+                record["fetchedAt"] = cached.fetched_at;
+                record["selectionReason"] = json!("refresh_failed_valid_cache");
+                super::publication::write_atomic(&der_path, &cached.bytes)?;
                 crate::logging::emit(
                     crate::logging::Level::Warn,
                     "ta_refresh_failed",
@@ -217,6 +269,10 @@ fn resolve_with_at(
     Ok((inputs, records))
 }
 
+#[cfg(test)]
+#[path = "anchors_tiebreaker_tests.rs"]
+mod tiebreaker_tests;
+
 #[cfg(test)]
 mod tests {
     use super::{cache_root, resolve_with};

src/cli/anchors_tiebreaker_tests.rs

@@ -0,0 +1,207 @@
+use super::*;
+use crate::cli::config::Anchor;
+
+fn auto_config(state: &Path) -> Config {
+    Config {
+        config_version: 1,
+        rrdp_state_dir: Some(state.to_path_buf()),
+        validation_time: Some("2026-06-01T00:00:00Z".into()),
+        anchors: vec![Anchor {
+            id: "custom".into(),
+            tal: crate::test_support::synthetic_repository().tal(),
+            ta: None,
+            constraints: None,
+        }],
+        ..Default::default()
+    }
+}
+
+fn variant(name: &str) -> Vec<u8> {
+    let fixture = crate::test_support::synthetic_repository();
+    fs::read(if name == "base" {
+        fixture.trust_anchor()
+    } else {
+        fixture.ta_variant(name)
+    })
+    .unwrap()
+}
+
+fn cache_path(config: &Config, extension: &str) -> PathBuf {
+    let tal = Tal::decode_bytes(&fs::read(&config.anchors[0].tal).unwrap()).unwrap();
+    cache_root(config.rrdp_state_dir.as_ref().unwrap()).join(format!(
+        "{}.{}",
+        hex::encode(Sha256::digest(&tal.subject_public_key_info_der)),
+        extension
+    ))
+}
+
+#[test]
+fn tiebreaker_older_download_cannot_replace_newer_cached_ta() {
+    let temp = tempfile::tempdir().unwrap();
+    let config = auto_config(&temp.path().join("state"));
+    let cached = variant("newer");
+    resolve_with(&config, |_| Ok(cached.clone())).unwrap();
+    let cache_before = fs::read(cache_path(&config, "json")).unwrap();
+    let (inputs, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap();
+    assert_eq!(inputs.len(), 1);
+    assert_eq!(fs::read(cache_path(&config, "cer")).unwrap(), cached);
+    assert_eq!(fs::read(cache_path(&config, "json")).unwrap(), cache_before);
+    assert_eq!(records[0]["source"], "cache");
+    assert_eq!(records[0]["refreshFailed"], false);
+}
+
+#[test]
+fn tiebreaker_equal_start_keeps_shorter_cached_validity() {
+    let temp = tempfile::tempdir().unwrap();
+    let config = auto_config(&temp.path().join("state"));
+    let cached = variant("shorter");
+    resolve_with(&config, |_| Ok(cached.clone())).unwrap();
+    resolve_with(&config, |_| Ok(variant("base"))).unwrap();
+    assert_eq!(fs::read(cache_path(&config, "cer")).unwrap(), cached);
+}
+
+#[test]
+fn tiebreaker_download_wins_each_precedence_level() {
+    for (old, new, reason) in [
+        ("shorter", "newer", "newer_not_before"),
+        ("longer", "base", "shorter_validity"),
+        ("base", "equal-period", "most_recently_fetched"),
+        ("base", "base", "most_recently_fetched"),
+    ] {
+        let temp = tempfile::tempdir().unwrap();
+        let config = auto_config(&temp.path().join("state"));
+        resolve_with(&config, |_| Ok(variant(old))).unwrap();
+        let selected = variant(new);
+        let (inputs, records) = resolve_with(&config, |_| Ok(selected.clone())).unwrap();
+        assert_eq!(inputs.len(), 1, "{old} -> {new}");
+        assert_eq!(records[0]["source"], "download");
+        assert_eq!(records[0]["selectionReason"], reason);
+        assert_eq!(records[0]["refreshFailed"], false);
+        assert_eq!(fs::read(cache_path(&config, "cer")).unwrap(), selected);
+        let saved: Value =
+            serde_json::from_slice(&fs::read(cache_path(&config, "json")).unwrap()).unwrap();
+        assert_eq!(saved["der"], hex::encode(selected));
+    }
+}
+
+#[test]
+fn tiebreaker_revalidates_cache_and_ignores_display_metadata() {
+    let temp = tempfile::tempdir().unwrap();
+    let config = auto_config(&temp.path().join("state"));
+    resolve_with(&config, |_| Ok(variant("newer"))).unwrap();
+    let path = cache_path(&config, "json");
+    let mut saved: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
+    saved["record"]["notBefore"] = json!("1900-01-01T00:00:00Z");
+    saved["record"]["notAfter"] = json!("1901-01-01T00:00:00Z");
+    saved["record"]["certificateSha256"] = json!("not-the-certificate-hash");
+    fs::write(&path, serde_json::to_vec(&saved).unwrap()).unwrap();
+    fs::write(cache_path(&config, "cer"), b"broken projection").unwrap();
+    let (_, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap();
+    assert_eq!(records[0]["selectionReason"], "cached_newer_not_before");
+    assert_eq!(
+        records[0]["certificateSha256"],
+        hex::encode(Sha256::digest(variant("newer")))
+    );
+    assert_ne!(records[0]["notBefore"], saved["record"]["notBefore"]);
+    assert_eq!(
+        fs::read(cache_path(&config, "cer")).unwrap(),
+        variant("newer")
+    );
+}
+
+#[test]
+fn tiebreaker_invalid_cached_candidates_cannot_veto_valid_download() {
+    let mut bad_signature = variant("newer");
+    *bad_signature.last_mut().unwrap() ^= 1;
+    let candidates = [
+        variant("expired"),
+        variant("future"),
+        variant("wrong-key"),
+        bad_signature,
+        b"corrupt DER".to_vec(),
+    ];
+    for bytes in candidates {
+        let temp = tempfile::tempdir().unwrap();
+        let config = auto_config(&temp.path().join("state"));
+        let path = cache_path(&config, "json");
+        fs::create_dir_all(path.parent().unwrap()).unwrap();
+        fs::write(
+            &path,
+            serde_json::to_vec(&json!({"der":hex::encode(bytes), "record":{}})).unwrap(),
+        )
+        .unwrap();
+        let (inputs, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap();
+        assert_eq!(inputs.len(), 1);
+        assert_eq!(records[0]["selectionReason"], "no_valid_cache");
+        assert_eq!(
+            fs::read(cache_path(&config, "cer")).unwrap(),
+            variant("base")
+        );
+    }
+}
+
+#[test]
+fn tiebreaker_invalid_downloads_obey_refresh_policy() {
+    for candidate in ["expired", "future", "wrong-key"] {
+        let temp = tempfile::tempdir().unwrap();
+        let mut config = auto_config(&temp.path().join("state"));
+        resolve_with(&config, |_| Ok(variant("base"))).unwrap();
+        let before = fs::read(cache_path(&config, "json")).unwrap();
+        let (inputs, records) = resolve_with(&config, |_| Ok(variant(candidate))).unwrap();
+        assert_eq!(inputs.len(), 1);
+        assert_eq!(records[0]["source"], "cache");
+        assert_eq!(records[0]["refreshFailed"], true);
+        assert_eq!(records[0]["errorCode"], "certificate_invalid");
+        assert_eq!(records[0]["selectionReason"], "refresh_failed_valid_cache");
+        assert_eq!(fs::read(cache_path(&config, "json")).unwrap(), before);
+        config.ta_refresh_policy = "require-fresh".into();
+        let (inputs, records) = resolve_with(&config, |_| Ok(variant(candidate))).unwrap();
+        assert!(inputs.is_empty());
+        assert_eq!(records[0]["refreshFailed"], true);
+        assert!(records[0]["source"].is_null());
+        assert_eq!(fs::read(cache_path(&config, "json")).unwrap(), before);
+    }
+}
+
+#[test]
+fn tiebreaker_healthy_cached_choice_works_with_require_fresh() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut config = auto_config(&temp.path().join("state"));
+    config.ta_refresh_policy = "require-fresh".into();
+    resolve_with(&config, |_| Ok(variant("newer"))).unwrap();
+    let (inputs, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap();
+    assert_eq!(inputs.len(), 1);
+    assert_eq!(records[0]["source"], "cache");
+    assert_eq!(records[0]["refreshFailed"], false);
+    let (inputs, records) = resolve_with(&config, |_| Err("offline".into())).unwrap();
+    assert!(inputs.is_empty());
+    assert_eq!(records[0]["errorCode"], "download_failed");
+}
+
+#[test]
+fn tiebreaker_retries_uri_and_preserves_fixed_ta_mode() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut config = auto_config(&temp.path().join("state"));
+    let mut attempts = 0;
+    let (inputs, records) = resolve_with(&config, |_| {
+        attempts += 1;
+        if attempts == 1 {
+            Err("unavailable first URI".into())
+        } else {
+            Ok(variant("newer"))
+        }
+    })
+    .unwrap();
+    assert_eq!(attempts, 2);
+    assert_eq!(inputs.len(), 1);
+    assert_eq!(records[0]["source"], "download");
+    config.anchors[0].ta = Some(crate::test_support::synthetic_repository().trust_anchor());
+    let (inputs, records) = resolve_with(&config, |_| panic!("fixed TA must not fetch")).unwrap();
+    assert_eq!(inputs.len(), 1);
+    assert_eq!(records[0]["source"], "file");
+    assert_eq!(records[0]["selectionReason"], "fixed_input");
+    assert_eq!(
+        fs::read(cache_path(&config, "cer")).unwrap(),
+        variant("newer")
+    );
+}

src/cli/validate.rs

@@ -151,7 +151,12 @@ fn execute(config: &Config, run_id: &str) -> Result<i32, String> {
         serde_json::from_slice(&fs::read(root.join("summary.json")).map_err(|e| e.to_string())?)
             .map_err(|e| e.to_string())?;
     let repository_failed = summary["repositoryFailedCount"].as_u64().unwrap_or(0);
-    let cached = anchors.iter().filter(|a| a["source"] == "cache").count();
+    // A healthy tiebreaker can prefer a cached TA after a successful download.
+    // Only a failed refresh using cache makes the result partial.
+    let cached = anchors
+        .iter()
+        .filter(|a| a["source"] == "cache" && a["refreshFailed"] == true)
+        .count();
     let status = if failed > 0 || repository_failed > 0 || cached > 0 {
         "partial"
     } else {

src/logging/text.rs

@@ -215,6 +215,11 @@ fn message(event: &str, fields: &Value) -> String {
             repository(fields),
             reason(fields)
         ),
+        "rrdp_session_desynchronized" => format!(
+            "RRDP delta {} from {} changed its hash within the same session; recovering with the latest snapshot.",
+            number(fields, "serial"),
+            repository(fields)
+        ),
         "rrdp_cross_origin_rejected" => format!(
             "Rejected the RRDP {} because its origin ({}) differs from the notification origin ({}).",
             string(fields, "resource", "resource"),
@@ -231,6 +236,11 @@ fn message(event: &str, fields: &Value) -> String {
             string(fields, "mode", "the selected"),
             repository(fields)
         ),
+        "ta_certificate_selected" => format!(
+            "Selected the cached trust anchor for {} because it is preferred to the downloaded certificate ({}).",
+            string(fields, "tal_id", "the configured anchor"),
+            string(fields, "reason", "TA tiebreaker")
+        ),
         "rrdp_download_completed" => format!(
             "Finished downloading repository data from {}.",
             repository(fields)
@@ -448,8 +458,10 @@ mod tests {
             "repo_terminal_failure",
             "http_fetch_failed",
             "rrdp_delta_fallback",
+            "rrdp_session_desynchronized",
             "rrdp_cross_origin_rejected",
             "rrdp_sync_selected",
+            "ta_certificate_selected",
             "rrdp_download_completed",
             "daemon_started",
             "daemon_state_recovered",

src/repository/storage/models_summary.rs

@@ -21,6 +21,10 @@ pub struct RrdpSourceRecord {
     pub last_snapshot_uri: Option<String>,
     pub last_snapshot_hash: Option<String>,
     pub last_error: Option<String>,
+    /// Delta references in the last successfully synchronized notification.
+    /// Older databases have no baseline until their next successful sync.
+    #[serde(default)]
+    pub last_delta_hashes: std::collections::BTreeMap<u64, [u8; 32]>,
 }
 
 impl RrdpSourceRecord {
@@ -43,6 +47,16 @@ impl RrdpSourceRecord {
         if let Some(last_error) = &self.last_error {
             validate_non_empty("rrdp_source.last_error", last_error)?;
         }
+        if !self.last_delta_hashes.is_empty()
+            && (self.last_session_id.is_none()
+                || self.last_serial.is_none()
+                || self.last_delta_hashes.keys().any(|serial| Some(*serial) > self.last_serial))
+        {
+            return Err(StorageError::InvalidData {
+                entity: "rrdp_source",
+                detail: "delta hash history requires a session and serial covering all references".into(),
+            });
+        }
         Ok(())
     }
 }

src/repository/sync/rrdp.rs

@@ -9,10 +9,10 @@ mod snapshot_apply;
 
 use crate::repository::sync::store_projection::{
     build_repository_view_present_entry, build_repository_view_withdrawn_entry,
-    current_rrdp_owner_is, ensure_rrdp_uri_can_be_owned_by, put_repository_view_present,
-    put_repository_view_withdrawn, put_rrdp_source_member_present,
+    build_rrdp_source_record_on_success, 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_repo_blob_bytes,
+    upsert_repo_blob_bytes,
 };
 use base64::Engine;
 use serde::{Deserialize, Serialize};

src/repository/sync/rrdp/models_and_parsing.rs

@@ -321,6 +321,8 @@ pub trait Fetcher: Send + Sync {
 pub struct RrdpState {
     pub session_id: String,
     pub serial: u64,
+    #[serde(default)]
+    pub delta_hashes: std::collections::BTreeMap<u64, [u8; 32]>,
 }
 
 impl RrdpState {
@@ -342,7 +344,11 @@ pub(crate) fn load_rrdp_local_state(
         .map_err(|e| e.to_string())?
         && let (Some(session_id), Some(serial)) = (record.last_session_id, record.last_serial)
     {
-        return Ok(Some(RrdpState { session_id, serial }));
+        return Ok(Some(RrdpState {
+            session_id,
+            serial,
+            delta_hashes: record.last_delta_hashes,
+        }));
     }
 
     Ok(None)
@@ -351,20 +357,28 @@ pub(crate) fn load_rrdp_local_state(
 pub(crate) fn persist_rrdp_local_state(
     store: &RocksStore,
     notification_uri: &str,
-    state: &RrdpState,
-    sync_state: RrdpSourceSyncState,
-    last_snapshot_uri: Option<&str>,
-    last_snapshot_hash_hex: Option<&str>,
+    notification: &Notification,
+    sync_state: Option<RrdpSourceSyncState>,
 ) -> Result<(), String> {
-    update_rrdp_source_record_on_success(
+    let mut record = build_rrdp_source_record_on_success(
         store,
         notification_uri,
-        state.session_id.as_str(),
-        state.serial,
+        &notification.session_id.to_string(),
+        notification.serial,
         sync_state,
-        last_snapshot_uri,
-        last_snapshot_hash_hex,
-    )
+        Some(&notification.snapshot_uri),
+        Some(&hex::encode(notification.snapshot_hash_sha256)),
+    )?;
+    // Retain only the previous successful notification window (RFC 9697 §3),
+    // rather than accumulating an unbounded history for a long-lived session.
+    record.last_delta_hashes = notification
+        .deltas
+        .iter()
+        .map(|delta| (delta.serial, delta.hash_sha256))
+        .collect();
+    store
+        .put_rrdp_source_record(&record)
+        .map_err(|error| error.to_string())
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]

src/repository/sync/rrdp/notification_sync.rs

@@ -123,7 +123,46 @@ fn sync_from_notification_inner(
         "delta_chain_unavailable"
     };
 
-    if let Some(s) = same_session_state {
+    if let Some(s) = same_session_state
+        && s.serial > notif.serial
+    {
+        return Err(RrdpError::NotificationSerialRollback {
+            old: s.serial,
+            new: notif.serial,
+        }
+        .into());
+    }
+
+    // RFC 9697 §3–5: compare references before either the noop fast path or
+    // fetching a new delta. Old delta files do not need to be downloaded.
+    let mutation = same_session_state.and_then(|previous| {
+        notif.deltas.iter().find_map(|delta| {
+            previous
+                .delta_hashes
+                .get(&delta.serial)
+                .filter(|old_hash| **old_hash != delta.hash_sha256)
+                .map(|old_hash| (delta.serial, *old_hash, delta.hash_sha256))
+        })
+    });
+    if let Some((serial, old_hash, new_hash)) = mutation {
+        snapshot_reason = "session_desynchronized";
+        crate::logging::emit(
+            crate::logging::Level::Warn,
+            "rrdp_session_desynchronized",
+            || {
+                serde_json::json!({
+                    "notify_uri": notification_uri, "session_id": notif.session_id.to_string(),
+                    "serial": serial, "old_hash": hex::encode(old_hash), "new_hash": hex::encode(new_hash),
+                    "recovery": "snapshot", "rfc": "RFC 9697 §3–5",
+                })
+            },
+        );
+        if let Some(t) = timing.as_ref() {
+            t.record_count("rrdp_session_desynchronization_total", 1);
+        }
+    }
+
+    if let Some(s) = same_session_state.filter(|_| mutation.is_none()) {
         if s.serial == notif.serial {
             log_selection("unchanged", "same_session_and_serial");
             let _hydrate_step = timing
@@ -142,18 +181,13 @@ fn sync_from_notification_inner(
             if let Some(t) = timing.as_ref() {
                 t.record_count("rrdp_current_index_hydrated_objects_total", hydrated as u64);
             }
+            persist_rrdp_local_state(store, notification_uri, &notif, None)
+                .map_err(RrdpSyncError::Storage)?;
             return Ok(0);
         }
-        if s.serial > notif.serial {
-            return Err(RrdpError::NotificationSerialRollback {
-                old: s.serial,
-                new: notif.serial,
-            }
-            .into());
-        }
     }
 
-    if let Some(s) = same_session_state {
+    if let Some(s) = same_session_state.filter(|_| mutation.is_none()) {
         // RFC 8182 §3.4.1: if session matches, MAY use deltas when a contiguous chain from the
         // last processed serial to the current serial can be processed (i.e. deltas cover the gap).
         let want_first = s.serial + 1;
@@ -273,17 +307,11 @@ fn sync_from_notification_inner(
                         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,
-                        };
                         persist_rrdp_local_state(
                             store,
                             notification_uri,
-                            &new_state,
-                            RrdpSourceSyncState::DeltaReady,
-                            Some(&notif.snapshot_uri),
-                            Some(&hex::encode(notif.snapshot_hash_sha256)),
+                            &notif,
+                            Some(RrdpSourceSyncState::DeltaReady),
                         )
                         .map_err(RrdpSyncError::Storage)?;
                         drop(_write_state_step);
@@ -394,17 +422,11 @@ fn sync_from_notification_inner(
     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,
-    };
     persist_rrdp_local_state(
         store,
         notification_uri,
-        &new_state,
-        RrdpSourceSyncState::SnapshotOnly,
-        Some(&notif.snapshot_uri),
-        Some(&hex::encode(notif.snapshot_hash_sha256)),
+        &notif,
+        Some(RrdpSourceSyncState::SnapshotOnly),
     )
     .map_err(RrdpSyncError::Storage)?;
     drop(_write_state_step);

src/repository/sync/rrdp/snapshot_sync.rs

@@ -162,17 +162,11 @@ fn sync_from_notification_snapshot_inner(
     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,
-    };
     persist_rrdp_local_state(
         store,
         notification_uri,
-        &state,
-        RrdpSourceSyncState::SnapshotOnly,
-        Some(&notif.snapshot_uri),
-        Some(&hex::encode(notif.snapshot_hash_sha256)),
+        &notif,
+        Some(RrdpSourceSyncState::SnapshotOnly),
     )
     .map_err(RrdpSyncError::Storage)?;
     drop(_write_state_step);

src/repository/sync/rrdp/tests.rs

@@ -3,3 +3,4 @@ include!("tests_parts/parsing.rs");
 include!("tests_parts/delta_apply.rs");
 include!("tests_parts/sync.rs");
 include!("tests_parts/edge_cases.rs");
+include!("tests_parts/desynchronization.rs");

src/repository/sync/rrdp/tests_parts/desynchronization.rs

@@ -0,0 +1,323 @@
+// RFC 9697: compare notification history without downloading old delta files.
+const DESYNC_NOTIFY: &str = "https://example.net/notification.xml";
+const DESYNC_SESSION: &str = "550e8400-e29b-41d4-a716-446655440000";
+const DESYNC_OBJECT: &str = "rsync://example.net/repo/a.roa";
+const DESYNC_OBSOLETE: &str = "rsync://example.net/repo/obsolete.roa";
+
+struct DesyncFetcher {
+    files: HashMap<String, Vec<u8>>,
+    requests: std::sync::Mutex<Vec<String>>,
+}
+
+impl DesyncFetcher {
+    fn snapshot(bytes: Vec<u8>) -> Self {
+        Self {
+            files: HashMap::from([("https://example.net/snapshot.xml".to_string(), bytes)]),
+            requests: Default::default(),
+        }
+    }
+
+    fn requested(&self) -> Vec<String> {
+        self.requests.lock().unwrap().clone()
+    }
+}
+
+impl Fetcher for DesyncFetcher {
+    fn fetch(&self, uri: &str) -> Result<Vec<u8>, String> {
+        self.requests.lock().unwrap().push(uri.to_string());
+        self.files
+            .get(uri)
+            .cloned()
+            .ok_or_else(|| format!("not found: {uri}"))
+    }
+}
+
+fn desync_notification(session: &str, serial: u64, snapshot: &[u8], old_hash: &str) -> Vec<u8> {
+    let snapshot_hash = hex::encode(sha2::Sha256::digest(snapshot));
+    let mut deltas = vec![("old", 100, "https://example.net/delta-100.xml", old_hash)];
+    let hash_101 = "22".repeat(32);
+    if serial == 101 {
+        deltas.push(("new", 101, "https://example.net/delta-101.xml", &hash_101));
+    }
+    notification_xml_with_deltas(
+        session,
+        serial,
+        "https://example.net/snapshot.xml",
+        &snapshot_hash,
+        &deltas,
+    )
+}
+
+fn seed_desync_repository(store: &RocksStore) {
+    let snapshot = snapshot_xml(
+        DESYNC_SESSION,
+        100,
+        &[(DESYNC_OBJECT, b"old"), (DESYNC_OBSOLETE, b"obsolete")],
+    );
+    let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"11".repeat(32));
+    sync_from_notification_snapshot(
+        store,
+        DESYNC_NOTIFY,
+        &notification,
+        &DesyncFetcher::snapshot(snapshot),
+    )
+    .unwrap();
+}
+
+#[test]
+fn desynchronization_same_serial_recovers_snapshot_before_noop() {
+    let temp = tempfile::tempdir().unwrap();
+    let store = RocksStore::open(temp.path()).unwrap();
+    seed_desync_repository(&store);
+    let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"recovered")]);
+    let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"33".repeat(32));
+    let fetcher = DesyncFetcher::snapshot(snapshot);
+    assert_eq!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &notification, &fetcher).unwrap(),
+        1
+    );
+    assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]);
+    assert_current_object(&store, DESYNC_OBJECT, b"recovered");
+    assert!(
+        store
+            .load_current_object_bytes_by_uri(DESYNC_OBSOLETE)
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[test]
+fn desynchronization_advanced_serial_survives_restart_and_skips_all_deltas() {
+    let temp = tempfile::tempdir().unwrap();
+    {
+        let store = RocksStore::open(temp.path()).unwrap();
+        seed_desync_repository(&store);
+    }
+    let store = RocksStore::open(temp.path()).unwrap();
+    let snapshot = snapshot_xml(DESYNC_SESSION, 101, &[(DESYNC_OBJECT, b"recovered")]);
+    let notification = desync_notification(DESYNC_SESSION, 101, &snapshot, &"33".repeat(32));
+    let fetcher = DesyncFetcher::snapshot(snapshot);
+    sync_from_notification(&store, DESYNC_NOTIFY, &notification, &fetcher).unwrap();
+    assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]);
+    assert_current_object(&store, DESYNC_OBJECT, b"recovered");
+    assert!(
+        store
+            .load_current_object_bytes_by_uri(DESYNC_OBSOLETE)
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[test]
+fn desynchronization_failed_recovery_keeps_baseline_for_retry() {
+    let temp = tempfile::tempdir().unwrap();
+    let store = RocksStore::open(temp.path()).unwrap();
+    seed_desync_repository(&store);
+    let before = store
+        .get_rrdp_source_record(DESYNC_NOTIFY)
+        .unwrap()
+        .unwrap();
+    let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"recovered")]);
+    let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"33".repeat(32));
+    // A failed download must not bless the new notification history.
+    let mut failing = DesyncFetcher::snapshot(snapshot.clone());
+    failing.files.clear();
+    assert!(sync_from_notification(&store, DESYNC_NOTIFY, &notification, &failing).is_err());
+    assert_eq!(
+        store
+            .get_rrdp_source_record(DESYNC_NOTIFY)
+            .unwrap()
+            .unwrap(),
+        before
+    );
+    assert_current_object(&store, DESYNC_OBJECT, b"old");
+
+    // Nor may a downloaded snapshot with a bad hash advance the baseline.
+    let corrupt = DesyncFetcher::snapshot(b"corrupt snapshot".to_vec());
+    assert!(matches!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &notification, &corrupt),
+        Err(RrdpSyncError::Rrdp(RrdpError::SnapshotHashMismatch))
+    ));
+    assert_eq!(
+        store
+            .get_rrdp_source_record(DESYNC_NOTIFY)
+            .unwrap()
+            .unwrap(),
+        before
+    );
+
+    let fetcher = DesyncFetcher::snapshot(snapshot);
+    sync_from_notification(&store, DESYNC_NOTIFY, &notification, &fetcher).unwrap();
+    assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]);
+    assert_current_object(&store, DESYNC_OBJECT, b"recovered");
+    // Once recovery succeeds the changed hash becomes the new baseline.
+    assert_eq!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &notification, &PanicOnRrdpFetch).unwrap(),
+        0
+    );
+}
+
+#[test]
+fn desynchronization_noop_learns_new_references_before_future_mutation() {
+    let temp = tempfile::tempdir().unwrap();
+    let store = RocksStore::open(temp.path()).unwrap();
+    seed_desync_repository(&store);
+    let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"recovered")]);
+    let hash = hex::encode(sha2::Sha256::digest(&snapshot));
+    let h99 = "99".repeat(32);
+    let h100 = "11".repeat(32);
+    let expanded = notification_xml_with_deltas(
+        DESYNC_SESSION,
+        100,
+        "https://example.net/snapshot.xml",
+        &hash,
+        &[
+            ("d99", 99, "https://example.net/d99.xml", &h99),
+            ("d100", 100, "https://example.net/d100.xml", &h100),
+        ],
+    );
+    assert_eq!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &expanded, &PanicOnRrdpFetch).unwrap(),
+        0
+    );
+    assert_current_object(&store, DESYNC_OBJECT, b"old");
+    let mutated = String::from_utf8(expanded)
+        .unwrap()
+        .replace(&h99, &"88".repeat(32));
+    let fetcher = DesyncFetcher::snapshot(snapshot);
+    sync_from_notification(&store, DESYNC_NOTIFY, mutated.as_bytes(), &fetcher).unwrap();
+    assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]);
+    assert_current_object(&store, DESYNC_OBJECT, b"recovered");
+}
+
+#[test]
+fn desynchronization_successful_delta_updates_and_prunes_history_window() {
+    let temp = tempfile::tempdir().unwrap();
+    let store = RocksStore::open(temp.path()).unwrap();
+    seed_desync_repository(&store);
+    let delta = delta_xml(
+        DESYNC_SESSION,
+        101,
+        &[&format!(
+            r#"<withdraw uri="{DESYNC_OBSOLETE}" hash="{}"/>"#,
+            hex::encode(sha2::Sha256::digest(b"obsolete"))
+        )],
+    );
+    let hash = hex::encode(sha2::Sha256::digest(&delta));
+    let snapshot = snapshot_xml(DESYNC_SESSION, 101, &[(DESYNC_OBJECT, b"recovered")]);
+    let snapshot_hash = hex::encode(sha2::Sha256::digest(&snapshot));
+    let notification = notification_xml_with_deltas(
+        DESYNC_SESSION,
+        101,
+        "https://example.net/snapshot.xml",
+        &snapshot_hash,
+        &[("new", 101, "https://example.net/delta-101.xml", &hash)],
+    );
+    let fetcher = DesyncFetcher {
+        files: HashMap::from([("https://example.net/delta-101.xml".into(), delta)]),
+        requests: Default::default(),
+    };
+    assert_eq!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &notification, &fetcher).unwrap(),
+        1
+    );
+    assert_eq!(fetcher.requested(), ["https://example.net/delta-101.xml"]);
+    let state = load_rrdp_local_state(&store, DESYNC_NOTIFY)
+        .unwrap()
+        .unwrap();
+    assert_eq!(state.delta_hashes.len(), 1);
+    assert_eq!(hex::encode(state.delta_hashes[&101]), hash);
+    assert_current_object(&store, DESYNC_OBJECT, b"old");
+
+    let mutated = String::from_utf8(notification)
+        .unwrap()
+        .replace(&hash, &"ff".repeat(32));
+    let recovery = DesyncFetcher::snapshot(snapshot);
+    sync_from_notification(&store, DESYNC_NOTIFY, mutated.as_bytes(), &recovery).unwrap();
+    assert_eq!(recovery.requested(), ["https://example.net/snapshot.xml"]);
+    assert_current_object(&store, DESYNC_OBJECT, b"recovered");
+}
+
+#[test]
+fn desynchronization_new_session_replaces_hash_baseline() {
+    let temp = tempfile::tempdir().unwrap();
+    let store = RocksStore::open(temp.path()).unwrap();
+    seed_desync_repository(&store);
+    let session = "550e8400-e29b-41d4-a716-446655440001";
+    let snapshot = snapshot_xml(session, 100, &[(DESYNC_OBJECT, b"new session")]);
+    let notification = desync_notification(session, 100, &snapshot, &"44".repeat(32));
+    sync_from_notification(
+        &store,
+        DESYNC_NOTIFY,
+        &notification,
+        &DesyncFetcher::snapshot(snapshot),
+    )
+    .unwrap();
+    let state = load_rrdp_local_state(&store, DESYNC_NOTIFY)
+        .unwrap()
+        .unwrap();
+    assert_eq!(state.session_id, session);
+    assert_eq!(state.delta_hashes[&100], [0x44; 32]);
+    assert_eq!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &notification, &PanicOnRrdpFetch).unwrap(),
+        0
+    );
+    assert_current_object(&store, DESYNC_OBJECT, b"new session");
+}
+
+#[test]
+fn desynchronization_legacy_source_record_defaults_to_empty_history() {
+    let temp = tempfile::tempdir().unwrap();
+    let store = RocksStore::open(temp.path()).unwrap();
+    seed_desync_repository(&store);
+    let record = store
+        .get_rrdp_source_record(DESYNC_NOTIFY)
+        .unwrap()
+        .unwrap();
+    let mut legacy = serde_cbor::value::to_value(&record).unwrap();
+    if let serde_cbor::Value::Map(map) = &mut legacy {
+        map.remove(&serde_cbor::Value::Text("last_delta_hashes".into()));
+    } else {
+        panic!("source record must encode as a CBOR map");
+    }
+    let decoded: crate::repository::storage::RrdpSourceRecord =
+        serde_cbor::from_slice(&serde_cbor::to_vec(&legacy).unwrap()).unwrap();
+    assert!(decoded.last_delta_hashes.is_empty());
+    store.put_rrdp_source_record(&decoded).unwrap();
+    let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"unused")]);
+    let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"55".repeat(32));
+    assert_eq!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &notification, &PanicOnRrdpFetch).unwrap(),
+        0
+    );
+    assert_eq!(
+        load_rrdp_local_state(&store, DESYNC_NOTIFY)
+            .unwrap()
+            .unwrap()
+            .delta_hashes[&100],
+        [0x55; 32]
+    );
+    assert_current_object(&store, DESYNC_OBJECT, b"old");
+}
+
+#[test]
+fn desynchronization_does_not_override_serial_rollback_guard() {
+    let temp = tempfile::tempdir().unwrap();
+    let store = RocksStore::open(temp.path()).unwrap();
+    seed_desync_repository(&store);
+    let before = store.get_rrdp_source_record(DESYNC_NOTIFY).unwrap();
+    let notification = notification_xml(
+        DESYNC_SESSION,
+        99,
+        "https://example.net/snapshot.xml",
+        &"00".repeat(32),
+    );
+    assert!(matches!(
+        sync_from_notification(&store, DESYNC_NOTIFY, &notification, &PanicOnRrdpFetch),
+        Err(RrdpSyncError::Rrdp(RrdpError::NotificationSerialRollback {
+            old: 100,
+            new: 99
+        }))
+    ));
+    assert_eq!(store.get_rrdp_source_record(DESYNC_NOTIFY).unwrap(), before);
+}

src/repository/sync/rrdp/tests_parts/edge_cases.rs

@@ -10,16 +10,17 @@ fn load_rrdp_local_state_uses_source_record_only() {
         None
     );
 
-    update_rrdp_source_record_on_success(
+    let record = build_rrdp_source_record_on_success(
         &store,
         notif_uri,
         "source-session",
         9,
-        crate::repository::storage::RrdpSourceSyncState::DeltaReady,
+        Some(crate::repository::storage::RrdpSourceSyncState::DeltaReady),
         Some("https://example.net/snapshot.xml"),
         Some(&hex::encode([0x11; 32])),
     )
-    .expect("write source record");
+    .expect("build source record");
+    store.put_rrdp_source_record(&record).expect("write source record");
 
     let got = load_rrdp_local_state(&store, notif_uri)
         .expect("load source preferred")
@@ -29,6 +30,7 @@ fn load_rrdp_local_state_uses_source_record_only() {
         RrdpState {
             session_id: "source-session".to_string(),
             serial: 9,
+            delta_hashes: Default::default(),
         }
     );
 }

src/repository/sync/store_projection.rs

@@ -408,15 +408,15 @@ pub fn put_rrdp_uri_owner_withdrawn(
         .map_err(|e| e.to_string())
 }
 
-pub fn update_rrdp_source_record_on_success(
+pub fn build_rrdp_source_record_on_success(
     store: &RocksStore,
     notification_uri: &str,
     session_id: &str,
     serial: u64,
-    sync_state: RrdpSourceSyncState,
+    sync_state: Option<RrdpSourceSyncState>,
     last_snapshot_uri: Option<&str>,
     last_snapshot_hash_hex: Option<&str>,
-) -> Result<(), String> {
+) -> Result<RrdpSourceRecord, String> {
     let now = now_pack_time();
     let mut record = match store
         .get_rrdp_source_record(notification_uri)
@@ -430,10 +430,11 @@ pub fn update_rrdp_source_record_on_success(
             first_seen_at: now.clone(),
             last_seen_at: now.clone(),
             last_sync_at: None,
-            sync_state,
+            sync_state: sync_state.unwrap_or(RrdpSourceSyncState::Empty),
             last_snapshot_uri: None,
             last_snapshot_hash: None,
             last_error: None,
+            last_delta_hashes: Default::default(),
         },
     };
 
@@ -441,14 +442,14 @@ pub fn update_rrdp_source_record_on_success(
     record.last_serial = Some(serial);
     record.last_seen_at = now.clone();
     record.last_sync_at = Some(now);
-    record.sync_state = sync_state;
+    record.sync_state = sync_state.unwrap_or(record.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())
+    // The caller adds the notification's delta history before the single write,
+    // so session/serial and the history cannot be committed independently.
+    Ok(record)
 }
 
 pub fn compute_sha256_hex(bytes: &[u8]) -> String {

src/test_support.rs

@@ -26,6 +26,10 @@ impl SyntheticRepositoryFixture {
     pub(crate) fn trust_anchor(&self) -> PathBuf {
         self.root.join("ta/custom-ta.cer")
     }
+
+    pub(crate) fn ta_variant(&self, name: &str) -> PathBuf {
+        self.root.join("ta-variants").join(format!("{name}.cer"))
+    }
 }
 
 pub(crate) fn synthetic_repository() -> &'static SyntheticRepositoryFixture {