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

CHANGELOG.md

@@ -1,6 +1,14 @@
 # Changelog
 
-## 0.1.0 — Unreleased
+## Unreleased
+
+- Detect RRDP session desynchronization from persisted notification delta hashes
+  (RFC 9697), warn and recover using the current snapshot before applying deltas.
+- Select automatic TA certificates using `draft-ietf-sidrops-rpki-ta-tiebreaker-06`:
+  prefer later notBefore, then earlier notAfter, then the newly fetched certificate.
+  A healthy preference for cache preserves a successful validation result.
+
+## 0.1.0 — 2026-09-29
 
 - Vendor five OpenBSD TALs and constraints with pinned provenance; enable policies
   in both all5 Compose examples and add the TOML `ta-constraints-enabled` switch.

README.md

@@ -1,13 +1,22 @@
-# Panda RPKI
+# panda-rpki
 
-Panda RPKI is an RPKI relying party written in Rust. It synchronizes repositories
+panda-rpki is an open-source RPKI Relying Party (RP) implementation written in Rust.
-over RRDP, validates signed objects, and exports routing data as CSV and Canonical
+It retrieves and validates RPKI data and produces validated routing information for
-Cache Representation (CCR). Run a single cycle or use the daemon for continuous operation.
+routing systems and other applications.
+
+The current v0.1.0 implementation is under active development and testing. It may
+contain bugs, incomplete features, or unexpected behavior. Feedback is welcome:
+please open an issue or submit a pull request when you find a problem or have an
+improvement to propose.
+
+Panda RPKI synchronizes repositories over RRDP, validates signed objects, and exports
+routing data as CSV and Canonical Cache Representation (CCR). Run a single cycle or
+use the daemon for continuous operation.
 
 Configuration is provided through TOML. See the [TOML configuration reference](docs/configuration.md)
 for every field, default, constraint and complete example.
 
-This build is validated on Linux x86-64. Other platforms are not yet verified.
+The v0.1.0 build is validated on Linux x86-64. Other platforms are not yet verified.
 Review the [standards and limitations](docs/conformance-matrix.md) and [security
 policy](SECURITY.md) before evaluating it for routing use.
 
@@ -59,6 +68,8 @@ For continuous operation see [Usage](docs/usage.md#continuous-operation).
 - Multiple named anchors, fixed or automatically refreshed trust anchors, and
   optional per-anchor resource constraints.
 - RRDP snapshot, delta, replace/withdraw and protocol fallback.
+- RRDP session desynchronization detection (RFC 9697) and automatic TA
+  certificate selection (TA tiebreaker draft -06; unreleased).
 - HTTPS same-origin checks for RRDP references and redirects.
 - Independent bounded repository and object worker pools.
 - CA, CRL, manifest, ROA, ASPA and BGPsec router-certificate processing.
@@ -91,5 +102,5 @@ target/release/panda-rpki --help
 
 ## License
 
-Panda RPKI uses the [BSD-3-Clause license](LICENSE).
+panda-rpki uses the [BSD-3-Clause license](LICENSE).
 See [third-party notices](THIRD_PARTY_NOTICES.md) for dependency attribution.

docker/examples/validate-all5.toml

@@ -8,6 +8,11 @@ ccr-out = "/output/result.ccr"
 run-timeout-secs = 1800
 require-complete = true
 ta-refresh-policy = "allow-valid-cache"
+# Notification/TA requests use http-timeout; RRDP data bodies have independent
+# per-resource limits and are selected automatically within the same run.
+http-timeout-secs = 10
+rrdp-snapshot-timeout-secs = 600
+rrdp-delta-timeout-secs = 180
 # Set false to ignore all constraints files, including adjacent auto-discovery.
 ta-constraints-enabled = true
 log-level = "info"

docs/command-line-reference.md

@@ -64,7 +64,9 @@ The remaining common keys and defaults are:
 | Key | Default |
 | --- | ---: |
 | `max-ca-depth` | 64 |
-| `http-timeout-secs` | 300 (0 is normalized to 1) |
+| `http-timeout-secs` | 10 (0 is normalized to 1) |
+| `rrdp-snapshot-timeout-secs` | 600 (0 is normalized to 1) |
+| `rrdp-delta-timeout-secs` | 180 (0 is normalized to 1) |
 | `http-root-cert` | `[]` |
 | `parallel-max-repo-sync-workers-global` | 8 |
 | `parallel-phase2-object-workers` | 8 |

docs/configuration.md

@@ -55,7 +55,9 @@ those choices do not change the built-in defaults.
 | `ta-constraints-enabled` | Boolean | true | Apply explicit or adjacent TA constraints files. False skips all constraints loading and enforcement, including adjacent discovery; unused missing or malformed files are ignored. |
 | `ta-refresh-policy` | String | `allow-valid-cache` | On automatic TA download failure, permit a revalidated, usable cached certificate. `require-fresh` prohibits fallback. Does not replace fixed TA files. |
 | `max-ca-depth` | Nonnegative integer | 64 | Maximum CA validation-tree depth. Zero is accepted by configuration parsing, but is a depth limit, not unlimited traversal. Usually keep 64. |
-| `http-timeout-secs` | Nonnegative integer, seconds | 300 | HTTP fetch timeout. Zero is normalized to one second, not unlimited. Does not set the whole-run or rsync timeout. |
+| `http-timeout-secs` | Nonnegative integer, seconds | 10 | Timeout for RRDP notification, TA/TAL and other small HTTP requests. Snapshot/delta bodies use their resource-specific settings below. Zero is normalized to one second, not unlimited. Does not set the whole-run or rsync timeout. |
+| `rrdp-snapshot-timeout-secs` | Nonnegative integer, seconds | 600 | Timeout for downloading an RRDP snapshot response body. Zero is normalized to one second. Selected by the RRDP resource type, including after same-origin redirects. |
+| `rrdp-delta-timeout-secs` | Nonnegative integer, seconds | 180 | Timeout for downloading an RRDP delta response body. Zero is normalized to one second. Each delta resource uses this value; it is independent of the validation run's overall timeout. |
 | `http-root-cert` | Array of path strings | `[]` | Additional HTTP TLS root certificates in PEM format. Every file must be readable and parseable. Empty means no custom additions; system TLS trust remains. These are not RPKI TA certificates. |
 | `parallel-max-repo-sync-workers-global` | Positive integer | 8 | Global repository-sync worker count. Zero is rejected. |
 | `parallel-phase2-object-workers` | Positive integer | 8 | Phase-2 object-processing worker count. Zero is rejected. |
@@ -154,10 +156,19 @@ no separate background refresh timer, and no invented HTTPS alternative for an
 rsync-only TAL. Candidates must pass certificate and TAL-binding validation at
 the configured logical validation time.
 
+When a fetched candidate and a revalidated cached certificate are both usable,
+selection follows `draft-ietf-sidrops-rpki-ta-tiebreaker-06` §2: prefer the later
+`notBefore`; if equal, prefer the earlier `notAfter`; if both are equal, use the
+fetched candidate. Comparisons use the signed certificate fields. Selecting a
+preferred cached certificate after a successful fetch is healthy, including
+under `require-fresh`: `trustAnchors[].source` is `cache`, `refreshFailed` is false,
+and `selectionReason` explains the choice. It does not increment
+`taCacheFallbackCount` or make the run partial.
+
 The cache is `<rrdp-state-dir>.ta-cache`, keyed by the TAL public-key SHA-256.
 Download failure permits fallback only under allow-valid-cache with a usable,
 revalidated cached certificate; that makes the result at least partial.
-Require-fresh prohibits fallback. Existing fixed `.cer` files do not seed this
+Require-fresh prohibits fallback after a failed refresh. Existing fixed `.cer` files do not seed this
 cache automatically; switching to automatic mode initially needs reachable URIs.
 One failed TA with others completing yields partial; all TAs failing yields
 failed. A fixed-certificate error found during preflight exits 2 immediately.
@@ -279,7 +290,9 @@ TA refresh policy does not make an anchor with an explicit ta automatic.
 log-format = "json"
 log-level = "debug"
 log-progress-interval-secs = 0
-http-timeout-secs = 600
+http-timeout-secs = 10 # notification, TA/TAL and other small HTTP requests
+rrdp-snapshot-timeout-secs = 600
+rrdp-delta-timeout-secs = 180
 http-root-cert = ["tls/internal-root.pem"]
 parallel-phase2-object-workers = 4
 parallel-max-repo-sync-workers-global = 4
@@ -290,6 +303,10 @@ parallel-repo-worker-queue-capacity = 1024
 Replace existing top-level settings; do not add duplicate keys. Prepare the
 custom TLS PEM file first, or keep [] when no extra roots are needed. This does
 not change the TAL public key, relax TA validation or disable system TLS checks.
+The snapshot and delta timeouts are selected per RRDP resource, not per
+validation invocation: an `auto` run may use the snapshot value for one
+repository and the delta value for another. RRDP notification and TA/TAL HTTP
+requests continue to use `http-timeout-secs`.
 
 ### 8.4 Validation policy and rsync transport: top-level fragment
 

docs/conformance-matrix.md

@@ -1,5 +1,22 @@
 # v0.1.0 conformance matrix
 
+## Unreleased additions
+
+| Behaviour | Standard reference | Implementation | Evidence |
+| --- | --- | --- | --- |
+| Detect changed hashes for overlapping delta serials within a session, warn and recover with a snapshot | [RFC 9697 §§3–5](https://www.rfc-editor.org/rfc/rfc9697.html) | `src/repository/sync/rrdp/`, `src/repository/storage/models_summary.rs` | `tests_parts/desynchronization.rs`; `tests/security_recovery_e2e.rs` |
+| Choose between valid fetched and cached TA certificates by later notBefore, earlier notAfter, then freshly fetched | [TA tiebreaker draft -06 §2](https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-rpki-ta-tiebreaker-06) | `src/cli/anchors.rs` | `anchors_tiebreaker_tests.rs`; `tests/security_recovery_e2e.rs` |
+
+The RRDP hash window is the last successfully processed notification's delta
+references, retained across restarts. An old database without this window learns
+it on its next successful sync. Failed recovery keeps the previous baseline.
+No historical delta download is required for comparison. TA selection compares
+revalidated DER certificates; fixed TA inputs retain their explicit selection.
+The tiebreaker is a draft implementation. Failed-refresh fallback still follows
+`ta-refresh-policy`; see [configuration](configuration.md#5-anchor-fields).
+
+## Released v0.1.0 baseline
+
 | Behaviour | Standard reference | Implementation | Evidence |
 | --- | --- | --- | --- |
 | TAL syntax and TA public-key binding | RFC 8630 §2 | `src/validation/from_tal.rs`, `src/model/ta.rs` | parser and synthetic integration tests |
@@ -29,6 +46,6 @@ revalidation of the entire trust chain by another relying party.
 
 The worker pool and interval index are implementation mechanisms, not new wire
 protocols; their tests must also prove deterministic output and unchanged
-validation decisions. Only the table entries above are in v0.1.0 scope.
+validation decisions. Only the released baseline table is in v0.1.0 scope.
 Adding a new source file or protocol feature requires a new row with a document
 revision, section, test, and reviewer decision.

docs/usage.md

@@ -36,6 +36,14 @@ TAL public-key hash. Refresh failures may use that cache under
 `allow-valid-cache` and produce `partial`; use `require-fresh` to reject the
 fallback.
 
+Automatic TA refresh compares usable fetched and cached certificates using the
+[TA tiebreaker rules](configuration.md#5-anchor-fields). A preferred cached
+certificate after a successful fetch keeps the run successful. RRDP also checks
+the previous notification's delta hashes (RFC 9697); a changed hash within the
+same session emits `rrdp_session_desynchronized` and triggers a fresh snapshot,
+including when the notification serial has not advanced. This check uses stored
+hashes and does not fetch old deltas.
+
 ## Multiple anchors and constraints
 
 Use one named `[[anchors]]` table per RIR. There is no positional TAL/TA pairing:

examples/config/daemon-all5.toml

@@ -7,7 +7,7 @@ shutdown-grace-secs = 30
 # Common settings: these choices are explicit, not all built-in defaults.
 ta-refresh-policy = "allow-valid-cache"
 max-ca-depth = 64
-http-timeout-secs = 300
+http-timeout-secs = 10
 http-root-cert = [] # optional: ["tls/internal-root.pem"]
 parallel-max-repo-sync-workers-global = 8
 parallel-phase2-object-workers = 8

examples/config/daemon-three.toml

@@ -8,7 +8,7 @@ shutdown-grace-secs = 30
 # Common settings: these choices are explicit, not all built-in defaults.
 ta-refresh-policy = "allow-valid-cache"
 max-ca-depth = 64
-http-timeout-secs = 300
+http-timeout-secs = 10
 http-root-cert = [] # optional: ["tls/internal-root.pem"]
 parallel-max-repo-sync-workers-global = 8
 parallel-phase2-object-workers = 8

examples/config/validate-full.toml

@@ -8,7 +8,9 @@ publish-dir = "publish/full"
 # Common settings: these choices are explicit, not all built-in defaults.
 ta-refresh-policy = "allow-valid-cache"
 max-ca-depth = 64
-http-timeout-secs = 300
+http-timeout-secs = 10
+rrdp-snapshot-timeout-secs = 600
+rrdp-delta-timeout-secs = 180
 http-root-cert = [] # optional: ["tls/internal-root.pem"]
 parallel-max-repo-sync-workers-global = 8
 parallel-phase2-object-workers = 8

examples/config/validate-policy.toml

@@ -21,7 +21,7 @@ rsync-scope = "module-root" # "host" or "publication-point"
 
 ta-refresh-policy = "allow-valid-cache"
 max-ca-depth = 64
-http-timeout-secs = 300
+http-timeout-secs = 10
 http-root-cert = []
 parallel-max-repo-sync-workers-global = 8
 parallel-phase2-object-workers = 8

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)
+                let cached = cached.inspect_err(|_| {
-                    .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");
                 })?;
                 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();
+                record["fetchedAt"] = cached.fetched_at;
-                super::publication::write_atomic(&der_path, &bytes)?;
+                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/config.rs

@@ -28,6 +28,8 @@ pub struct Config {
     pub ta_constraints_enabled: bool,
     pub max_ca_depth: usize,
     pub http_timeout_secs: u64,
+    pub rrdp_snapshot_timeout_secs: u64,
+    pub rrdp_delta_timeout_secs: u64,
     pub http_root_cert: Vec<PathBuf>,
     pub parallel_max_repo_sync_workers_global: usize,
     pub parallel_phase2_object_workers: usize,
@@ -117,7 +119,11 @@ impl Default for Config {
             ta_refresh_policy: "allow-valid-cache".into(),
             ta_constraints_enabled: true,
             max_ca_depth: 64,
-            http_timeout_secs: 300,
+            http_timeout_secs: 10,
+            rrdp_snapshot_timeout_secs:
+                crate::repository::fetch::http::DEFAULT_RRDP_SNAPSHOT_TIMEOUT_SECS,
+            rrdp_delta_timeout_secs:
+                crate::repository::fetch::http::DEFAULT_RRDP_DELTA_TIMEOUT_SECS,
             http_root_cert: Vec::new(),
             parallel_max_repo_sync_workers_global: 8,
             parallel_phase2_object_workers: 8,
@@ -356,6 +362,15 @@ impl Config {
             normalizations.push(json!({"field":"http-timeout-secs","input":0,"effective":1}));
             config.http_timeout_secs = 1;
         }
+        if config.rrdp_snapshot_timeout_secs < 1 {
+            normalizations
+                .push(json!({"field":"rrdp-snapshot-timeout-secs","input":0,"effective":1}));
+            config.rrdp_snapshot_timeout_secs = 1;
+        }
+        if config.rrdp_delta_timeout_secs < 1 {
+            normalizations.push(json!({"field":"rrdp-delta-timeout-secs","input":0,"effective":1}));
+            config.rrdp_delta_timeout_secs = 1;
+        }
         if config.parallel_repo_worker_queue_capacity < 1024 {
             normalizations.push(json!({"field":"parallel-repo-worker-queue-capacity","input":config.parallel_repo_worker_queue_capacity,"effective":1024}));
             config.parallel_repo_worker_queue_capacity = 1024;
@@ -618,6 +633,8 @@ resource-validation-mode = "rfc6487"
 validation-time = "2026-09-22T08:00:00+08:00"
 sync-preference = "rsync-only"
 rsync-timeout-secs = 47
+rrdp-snapshot-timeout-secs = 720
+rrdp-delta-timeout-secs = 90
 rsync-command = {:?}
 rsync-mirror-root = "mirror"
 rsync-scope = "publication-point"
@@ -652,6 +669,8 @@ tal = {:?}
         );
         assert_eq!(config.sync_preference().unwrap(), SyncPreference::RsyncOnly);
         assert_eq!(config.rsync_timeout_secs, 47);
+        assert_eq!(config.rrdp_snapshot_timeout_secs, 720);
+        assert_eq!(config.rrdp_delta_timeout_secs, 90);
         assert_eq!(config.rsync_scope, RsyncScopePolicy::PublicationPoint);
         assert_eq!(
             config.effective_validation_time().unwrap().unix_timestamp(),
@@ -662,6 +681,8 @@ tal = {:?}
         assert_eq!(printed["effective"]["resource-validation-mode"], "rfc6487");
         assert_eq!(printed["effective"]["sync-preference"], "rsync-only");
         assert_eq!(printed["effective"]["rsync-timeout-secs"], 47);
+        assert_eq!(printed["effective"]["rrdp-snapshot-timeout-secs"], 720);
+        assert_eq!(printed["effective"]["rrdp-delta-timeout-secs"], 90);
         assert_eq!(
             printed["effective"]["validation-time"],
             "2026-09-22T00:00:00Z"
@@ -689,6 +710,51 @@ tal = {:?}
         );
     }
 
+    #[test]
+    fn rrdp_timeout_defaults_are_distinct_and_zero_values_are_normalized() {
+        let fixture = crate::test_support::synthetic_repository();
+        let directory = tempfile::tempdir().unwrap();
+        let config_path = directory.path().join("validate.toml");
+        std::fs::write(
+            &config_path,
+            format!(
+                r#"
+config-version = 1
+mode = "validate"
+out = "out"
+rrdp-snapshot-timeout-secs = 0
+rrdp-delta-timeout-secs = 0
+
+[[anchors]]
+id = "custom"
+tal = {:?}
+"#,
+                fixture.tal().to_string_lossy(),
+            ),
+        )
+        .unwrap();
+
+        let (config, printed) = Config::load(&config_path, "validate").unwrap();
+        assert_eq!(Config::default().http_timeout_secs, 10);
+        assert_eq!(Config::default().rrdp_snapshot_timeout_secs, 600);
+        assert_eq!(Config::default().rrdp_delta_timeout_secs, 180);
+        assert_eq!(config.rrdp_snapshot_timeout_secs, 1);
+        assert_eq!(config.rrdp_delta_timeout_secs, 1);
+        assert_eq!(printed["effective"]["rrdp-snapshot-timeout-secs"], 1);
+        assert_eq!(printed["effective"]["rrdp-delta-timeout-secs"], 1);
+        let normalizations = printed["normalizations"].as_array().unwrap();
+        assert!(
+            normalizations
+                .iter()
+                .any(|item| item["field"] == "rrdp-snapshot-timeout-secs")
+        );
+        assert!(
+            normalizations
+                .iter()
+                .any(|item| item["field"] == "rrdp-delta-timeout-secs")
+        );
+    }
+
     #[test]
     fn constraints_selection_disable_override_and_reload() {
         let dir = tempfile::tempdir().unwrap();

src/cli/validate.rs

@@ -128,6 +128,8 @@ fn execute(config: &Config, run_id: &str) -> Result<i32, String> {
         strict_policy: Some(config.strict),
         resource_validation_mode: Some(config.resource_validation_mode),
         disable_rrdp: sync_preference == SyncPreference::RsyncOnly,
+        rrdp_snapshot_timeout_secs: Some(config.rrdp_snapshot_timeout_secs),
+        rrdp_delta_timeout_secs: Some(config.rrdp_delta_timeout_secs),
         rsync_command: config.rsync_command.clone(),
         rsync_timeout_secs: config.rsync_timeout_secs,
         rsync_mirror_root: config.rsync_mirror_root.clone(),
@@ -149,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/fetch/http.rs

@@ -28,6 +28,9 @@ pub fn with_scoped_http_timeout_override<R>(timeout: Duration, f: impl FnOnce()
 /// Default User-Agent sent with every outgoing HTTP request (RRDP
 /// notification/snapshot/delta and TAL/TA downloads).
 pub const DEFAULT_HTTP_USER_AGENT: &str = "panda-rpki/0.2";
+pub const DEFAULT_LARGE_BODY_TIMEOUT_SECS: u64 = 180;
+pub const DEFAULT_RRDP_SNAPSHOT_TIMEOUT_SECS: u64 = 600;
+pub const DEFAULT_RRDP_DELTA_TIMEOUT_SECS: u64 = 180;
 
 /// Environment variable that overrides the HTTP User-Agent. Unset, empty, or
 /// values containing characters that are illegal in a header value fall back
@@ -53,8 +56,12 @@ pub struct HttpFetcherConfig {
     pub connect_timeout: Duration,
     /// Short timeout used for connection establishment and small metadata objects.
     pub timeout: Duration,
-    /// Larger timeout used for RRDP snapshot / delta bodies.
+    /// Generic large-body timeout, retained for untyped HTTP callers.
     pub large_body_timeout: Duration,
+    /// Timeout for RRDP snapshot response bodies.
+    pub rrdp_snapshot_timeout: Duration,
+    /// Timeout for RRDP delta response bodies.
+    pub rrdp_delta_timeout: Duration,
     pub user_agent: String,
     /// Extra PEM trust anchors for HTTPS transport tests or controlled endpoints.
     pub extra_root_certificates_pem: Vec<Vec<u8>>,
@@ -65,7 +72,9 @@ impl Default for HttpFetcherConfig {
         Self {
             connect_timeout: Duration::from_secs(15),
             timeout: Duration::from_secs(30),
-            large_body_timeout: Duration::from_secs(180),
+            large_body_timeout: Duration::from_secs(DEFAULT_LARGE_BODY_TIMEOUT_SECS),
+            rrdp_snapshot_timeout: Duration::from_secs(DEFAULT_RRDP_SNAPSHOT_TIMEOUT_SECS),
+            rrdp_delta_timeout: Duration::from_secs(DEFAULT_RRDP_DELTA_TIMEOUT_SECS),
             user_agent: resolve_http_user_agent(std::env::var(HTTP_USER_AGENT_ENV).ok()),
             extra_root_certificates_pem: Vec::new(),
         }
@@ -83,15 +92,22 @@ pub struct BlockingHttpFetcher {
     large_body_client: Client,
     retry_short_client: Client,
     rrdp_short_client: Client,
-    rrdp_large_body_client: Client,
+    rrdp_snapshot_client: Client,
+    rrdp_delta_client: Client,
     rrdp_retry_short_client: Client,
     short_timeout: Duration,
     large_body_timeout: Duration,
+    rrdp_snapshot_timeout: Duration,
+    rrdp_delta_timeout: Duration,
 }
 
 impl BlockingHttpFetcher {
     pub fn new(config: HttpFetcherConfig) -> Result<Self, String> {
         let short_timeout = config.timeout;
+        let rrdp_snapshot_timeout = config.rrdp_snapshot_timeout;
+        let rrdp_delta_timeout = config.rrdp_delta_timeout;
+        // Preserve the existing timeout behavior for untyped HTTP callers.
+        // RRDP's typed fetch path below uses its exact snapshot/delta profile.
         let large_body_timeout = std::cmp::max(config.large_body_timeout, config.timeout);
         let connect_timeout = std::cmp::min(config.connect_timeout, config.timeout);
         let short_client = Self::client_builder(
@@ -126,10 +142,18 @@ impl BlockingHttpFetcher {
         )?
         .build()
         .map_err(|e| e.to_string())?;
-        let rrdp_large_body_client = Self::rrdp_client_builder(
+        let rrdp_snapshot_client = Self::rrdp_client_builder(
             &config,
             connect_timeout,
-            large_body_timeout,
+            rrdp_snapshot_timeout,
+            config.user_agent.clone(),
+        )?
+        .build()
+        .map_err(|e| e.to_string())?;
+        let rrdp_delta_client = Self::rrdp_client_builder(
+            &config,
+            connect_timeout,
+            rrdp_delta_timeout,
             config.user_agent.clone(),
         )?
         .build()
@@ -147,10 +171,13 @@ impl BlockingHttpFetcher {
             large_body_client,
             retry_short_client,
             rrdp_short_client,
-            rrdp_large_body_client,
+            rrdp_snapshot_client,
+            rrdp_delta_client,
             rrdp_retry_short_client,
             short_timeout,
             large_body_timeout,
+            rrdp_snapshot_timeout,
+            rrdp_delta_timeout,
         })
     }
 
@@ -300,19 +327,28 @@ impl BlockingHttpFetcher {
         }
     }
 
-    fn rrdp_client_for_uri(&self, uri: &str) -> (&Client, &'static str, Duration) {
+    fn rrdp_client_for_resource(
+        &self,
+        resource: RrdpResourceKind,
+    ) -> (&Client, &'static str, Duration) {
         if let Some(timeout) = HTTP_TIMEOUT_OVERRIDE.with(|cell| *cell.borrow()) {
             return (&self.rrdp_retry_short_client, "retry_short", timeout);
         }
-        if uses_large_body_timeout(uri) {
+        match resource {
-            (
+            RrdpResourceKind::Notification => {
-                &self.rrdp_large_body_client,
-                "large_body",
-                self.large_body_timeout,
-            )
-        } else {
                 (&self.rrdp_short_client, "short", self.short_timeout)
             }
+            RrdpResourceKind::Snapshot => (
+                &self.rrdp_snapshot_client,
+                "rrdp_snapshot",
+                self.rrdp_snapshot_timeout,
+            ),
+            RrdpResourceKind::Delta => (
+                &self.rrdp_delta_client,
+                "rrdp_delta",
+                self.rrdp_delta_timeout,
+            ),
+        }
     }
 
     fn rrdp_response(
@@ -336,7 +372,7 @@ impl BlockingHttpFetcher {
         }
 
         for redirects in 0..=MAX_RRDP_REDIRECTS {
-            let (client, profile, timeout) = self.rrdp_client_for_uri(current.as_str());
+            let (client, profile, timeout) = self.rrdp_client_for_resource(resource);
             let response = client
                 .get(current.clone())
                 .send()
@@ -729,6 +765,8 @@ mod tests {
         let http = BlockingHttpFetcher::new(HttpFetcherConfig {
             timeout: Duration::from_secs(3),
             large_body_timeout: Duration::from_secs(9),
+            rrdp_snapshot_timeout: Duration::from_secs(9),
+            rrdp_delta_timeout: Duration::from_secs(6),
             ..HttpFetcherConfig::default()
         })
         .expect("http");
@@ -744,6 +782,68 @@ mod tests {
         assert_eq!(timeout_large, Duration::from_secs(9));
     }
 
+    #[test]
+    fn typed_rrdp_resource_selects_its_own_timeout_independent_of_uri() {
+        let http = BlockingHttpFetcher::new(HttpFetcherConfig {
+            timeout: Duration::from_secs(10),
+            rrdp_snapshot_timeout: Duration::from_secs(600),
+            rrdp_delta_timeout: Duration::from_secs(180),
+            ..HttpFetcherConfig::default()
+        })
+        .expect("http");
+
+        let (_, notification_profile, notification_timeout) =
+            http.rrdp_client_for_resource(RrdpResourceKind::Notification);
+        let (_, snapshot_profile, snapshot_timeout) =
+            http.rrdp_client_for_resource(RrdpResourceKind::Snapshot);
+        let (_, delta_profile, delta_timeout) =
+            http.rrdp_client_for_resource(RrdpResourceKind::Delta);
+
+        assert_eq!(notification_profile, "short");
+        assert_eq!(notification_timeout, Duration::from_secs(10));
+        assert_eq!(snapshot_profile, "rrdp_snapshot");
+        assert_eq!(snapshot_timeout, Duration::from_secs(600));
+        assert_eq!(delta_profile, "rrdp_delta");
+        assert_eq!(delta_timeout, Duration::from_secs(180));
+    }
+
+    #[test]
+    fn rrdp_snapshot_and_delta_requests_enforce_independent_timeouts() {
+        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind");
+        let addr = listener.local_addr().expect("addr");
+        let server = thread::spawn(move || {
+            for _ in 0..2 {
+                let (mut stream, _) = listener.accept().expect("accept");
+                let mut request = [0u8; 1024];
+                let _ = stream.read(&mut request);
+                thread::sleep(StdDuration::from_millis(250));
+                let response =
+                    b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok";
+                let _ = stream.write_all(response);
+            }
+        });
+        let uri = format!("http://{addr}/resource.xml");
+        let origin = RrdpOrigin::parse(&uri, RrdpResourceKind::Notification).expect("origin");
+        let http = BlockingHttpFetcher::new(HttpFetcherConfig {
+            timeout: Duration::from_secs(3),
+            rrdp_snapshot_timeout: Duration::from_secs(3),
+            rrdp_delta_timeout: Duration::from_millis(100),
+            ..HttpFetcherConfig::default()
+        })
+        .expect("http");
+
+        assert_eq!(
+            http.fetch_rrdp(RrdpResourceKind::Snapshot, &uri, &origin)
+                .expect("snapshot response should fit its timeout"),
+            b"ok"
+        );
+        assert!(
+            http.fetch_rrdp(RrdpResourceKind::Delta, &uri, &origin)
+                .is_err()
+        );
+        server.join().expect("server thread");
+    }
+
     #[test]
     fn resolve_http_user_agent_falls_back_to_default() {
         assert_eq!(resolve_http_user_agent(None), DEFAULT_HTTP_USER_AGENT);

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,
+    build_rrdp_source_record_on_success, current_rrdp_owner_is, ensure_rrdp_uri_can_be_owned_by,
-    put_repository_view_withdrawn, put_rrdp_source_member_present,
+    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,
+    notification: &Notification,
-    sync_state: RrdpSourceSyncState,
+    sync_state: Option<RrdpSourceSyncState>,
-    last_snapshot_uri: Option<&str>,
-    last_snapshot_hash_hex: Option<&str>,
 ) -> 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(),
+        &notification.session_id.to_string(),
-        state.serial,
+        notification.serial,
         sync_state,
-        last_snapshot_uri,
+        Some(&notification.snapshot_uri),
-        last_snapshot_hash_hex,
+        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,
+                            &notif,
-                            RrdpSourceSyncState::DeltaReady,
+                            Some(RrdpSourceSyncState::DeltaReady),
-                            Some(&notif.snapshot_uri),
-                            Some(&hex::encode(notif.snapshot_hash_sha256)),
                         )
                         .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,
+        &notif,
-        RrdpSourceSyncState::SnapshotOnly,
+        Some(RrdpSourceSyncState::SnapshotOnly),
-        Some(&notif.snapshot_uri),
-        Some(&hex::encode(notif.snapshot_hash_sha256)),
     )
     .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,
+        &notif,
-        RrdpSourceSyncState::SnapshotOnly,
+        Some(RrdpSourceSyncState::SnapshotOnly),
-        Some(&notif.snapshot_uri),
-        Some(&hex::encode(notif.snapshot_hash_sha256)),
     )
     .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
+    // The caller adds the notification's delta history before the single write,
-        .put_rrdp_source_record(&record)
+    // so session/serial and the history cannot be committed independently.
-        .map_err(|e| e.to_string())
+    Ok(record)
 }
 
 pub fn compute_sha256_hex(bytes: &[u8]) -> String {

src/runtime/run.rs

@@ -112,13 +112,19 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
     let validation_started = std::time::Instant::now();
     crate::logging::observer::phase("runtime_setup");
     let collect_current_repo_objects = false;
-    let out = if let Some(dir) = args.rsync_local_dir.as_ref() {
+    let mut http_config = HttpFetcherConfig {
-        let http = BlockingHttpFetcher::new(HttpFetcherConfig {
         timeout: std::time::Duration::from_secs(args.http_timeout_secs.max(1)),
         extra_root_certificates_pem: http_root_certificates_pem.clone(),
         ..HttpFetcherConfig::default()
-        })
+    };
-        .map_err(|e| e.to_string())?;
+    if let Some(timeout) = args.rrdp_snapshot_timeout_secs {
+        http_config.rrdp_snapshot_timeout = std::time::Duration::from_secs(timeout.max(1));
+    }
+    if let Some(timeout) = args.rrdp_delta_timeout_secs {
+        http_config.rrdp_delta_timeout = std::time::Duration::from_secs(timeout.max(1));
+    }
+    let out = if let Some(dir) = args.rsync_local_dir.as_ref() {
+        let http = BlockingHttpFetcher::new(http_config.clone()).map_err(|e| e.to_string())?;
         let rsync = LocalDirRsyncFetcher::new(dir);
         run_online_validation_with_fetchers(
             Arc::clone(&store),
@@ -132,12 +138,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
             timing.as_ref().map(|(_, t)| t),
         )?
     } else {
-        let http = BlockingHttpFetcher::new(HttpFetcherConfig {
+        let http = BlockingHttpFetcher::new(http_config).map_err(|e| e.to_string())?;
-            timeout: std::time::Duration::from_secs(args.http_timeout_secs.max(1)),
-            extra_root_certificates_pem: http_root_certificates_pem.clone(),
-            ..HttpFetcherConfig::default()
-        })
-        .map_err(|e| e.to_string())?;
         let rsync = SystemRsyncFetcher::new(SystemRsyncConfig {
             rsync_bin: args
                 .rsync_command

src/runtime/types.rs

@@ -79,6 +79,9 @@ pub struct RunConfig {
     pub rsync_command: Option<PathBuf>,
 
     pub http_timeout_secs: u64,
+    /// Per-resource RRDP response timeouts; None selects the fetcher default.
+    pub rrdp_snapshot_timeout_secs: Option<u64>,
+    pub rrdp_delta_timeout_secs: Option<u64>,
     pub http_root_cert_paths: Vec<PathBuf>,
     pub rsync_timeout_secs: u64,
     pub rsync_mirror_root: Option<PathBuf>,

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 {

tests/security_recovery_e2e.rs

@@ -0,0 +1,9 @@
+#[test]
+fn persisted_rrdp_desynchronization_and_ta_tiebreaker_are_end_to_end() {
+    let status = std::process::Command::new("python3")
+        .arg("tests/support/test_security_recovery.py")
+        .arg(env!("CARGO_BIN_EXE_panda-rpki"))
+        .status()
+        .expect("start RRDP recovery and TA selection integration driver");
+    assert!(status.success(), "security recovery E2E failed: {status}");
+}

tests/support/README.md

@@ -11,3 +11,17 @@ network request is included in this directory. The generator requires Python
 `cryptography`, OpenSSL's `cms` command, and Python 3; CI installs those test
 tools explicitly. Test output is created under a temporary directory and is
 not part of a release artifact.
+# Security recovery and TA selection
+
+`cargo test --offline --test security_recovery_e2e` runs the real CLI against a
+local HTTPS repository with persisted state. Across three processes it checks
+initial snapshot, recovery after a delta-hash mutation at unchanged session/serial,
+and a subsequent noop. Request assertions prove that no historical delta is
+downloaded. A newer cached TA must survive a valid older download under
+`require-fresh`, with a successful result and no cache-fallback count.
+
+To preserve the per-step request and summary evidence:
+
+```sh
+python3 tests/support/test_security_recovery.py target/debug/panda-rpki --proof /tmp/security-recovery.json
+```

tests/support/generate_repository.py

@@ -545,6 +545,29 @@ def make_https_certificates(output: Path) -> None:
     write_key(output / "certs/rrdp-server.key", server_key)
 
 
+def make_ta_variants(output: Path, name: x509.Name, key, foreign_key) -> None:
+    """Real self-signed versions of one TAL-bound key, for tiebreaker tests."""
+    variants = {
+        "older": (datetime(2025, 1, 1, tzinfo=UTC), VALID_TO),
+        "newer": (datetime(2026, 5, 1, tzinfo=UTC), VALID_TO),
+        "shorter": (VALID_FROM, datetime(2026, 12, 31, tzinfo=UTC)),
+        "longer": (VALID_FROM, datetime(2030, 1, 1, tzinfo=UTC)),
+        "equal-period": (VALID_FROM, VALID_TO),
+        "expired": (datetime(2025, 1, 1, tzinfo=UTC), datetime(2025, 12, 31, tzinfo=UTC)),
+        "future": (datetime(2027, 1, 1, tzinfo=UTC), datetime(2028, 1, 1, tzinfo=UTC)),
+        "wrong-key": (VALID_FROM, VALID_TO),
+    }
+    for serial, (variant, (start, end)) in enumerate(variants.items(), 0x14700100):
+        selected_key = foreign_key if variant == "wrong-key" else key
+        certificate = sign_cert(
+            name, name, selected_key.public_key(), selected_key, serial, True, selected_key,
+            ip_ext=ip_resources([CHILD_PREFIX]), as_ext=as_resources([TEST_ASN]),
+            ca_repository=ROOT_REPO_URI, manifest_uri=ROOT_MFT_URI, notify_uri=RRDP_URI,
+            not_before=start, not_after=end,
+        )
+        write(output / f"ta-variants/{variant}.cer", certificate.public_bytes(Encoding.DER))
+
+
 def main() -> None:
     global RRDP_HOST, FOREIGN_RRDP_HOST, RRDP_PORT, RSYNC_HOST, RSYNC_PORT
     global TAL_URI, TAL_HTTPS_URI, RRDP_URI, STRICT_NAMES
@@ -638,6 +661,7 @@ def main() -> None:
     ).encode()
     write(output / "tal/custom.tal", tal)
     write(output / "ta/custom-ta.cer", root_cert.public_bytes(Encoding.DER))
+    make_ta_variants(output, root_name, root_key, child_key)
     with tempfile.TemporaryDirectory(prefix="custom-rpki-cms-") as temp:
         work = Path(temp)
         common = output / ".common"

tests/support/test_security_recovery.py

@@ -0,0 +1,151 @@
+"""Real CLI, HTTPS and persisted state for RFC 9697 and TA tiebreaker -06."""
+
+from __future__ import annotations
+
+import argparse
+import functools
+import http.server
+import json
+from pathlib import Path
+import shutil
+import ssl
+import sys
+import tempfile
+import threading
+
+from test_public_config import (
+    assert_successful_output,
+    free_port,
+    make_rsync_wrapper,
+    read_json,
+    run_checked,
+    write_config,
+)
+
+
+def run(binary: str) -> dict:
+    support = Path(__file__).resolve().parent
+    with tempfile.TemporaryDirectory(prefix="panda-rpki-security-") as tmp:
+        base = Path(tmp)
+        fixture = base / "fixture"
+        port = free_port()
+        run_checked([
+            sys.executable, str(support / "generate_repository.py"),
+            "--rrdp-host", "127.0.0.1", "--rrdp-port", str(port),
+            "--rsync-host", "127.0.0.1", "--strict-names", "--output", str(fixture),
+        ])
+        active = base / "active"
+        shutil.copytree(fixture / "cases/baseline-v1/http", active)
+        # Prefer real HTTPS for automatic TA retrieval. Retain the rsync URI:
+        # the validation core uses it as the key for its offline TA projection.
+        tal = fixture / "tal/custom.tal"
+        lines = tal.read_text().splitlines()
+        tal.write_text("\n".join(
+            [line for line in lines if line.startswith("https://")]
+            + [line for line in lines if line.startswith("rsync://")]
+            + [line for line in lines if not line.startswith(("https://", "rsync://"))]
+        ) + "\n", encoding="utf-8")
+        served_ta = active / "ta/custom-ta.cer"
+        served_ta.write_bytes((fixture / "ta-variants/newer.cer").read_bytes())
+        notification = active / "rrdp/notification.xml"
+        original = notification.read_text()
+
+        def advertise_hash(value: str) -> None:
+            reference = (
+                f'<delta serial="1" uri="https://127.0.0.1:{port}/rrdp/delta-1.xml" '
+                f'hash="{value * 32}"/>'
+            )
+            notification.write_text(original.replace("</notification>", reference + "</notification>"))
+
+        advertise_hash("11")
+        requests: list[str] = []
+
+        class Handler(http.server.SimpleHTTPRequestHandler):
+            def do_GET(self) -> None:
+                requests.append(self.path)
+                super().do_GET()
+
+            def log_message(self, format: str, *args) -> None:
+                pass
+
+        server = http.server.ThreadingHTTPServer(
+            ("127.0.0.1", port), functools.partial(Handler, directory=str(active))
+        )
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        context.load_cert_chain(fixture / "certs/rrdp-server.pem", fixture / "certs/rrdp-server.key")
+        server.socket = context.wrap_socket(server.socket, server_side=True)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            output, state = base / "output", base / "state"
+            wrapper, wrapper_log = base / "rsync.py", base / "rsync.log"
+            make_rsync_wrapper(wrapper, fixture, wrapper_log)
+            config = base / "validate.toml"
+            write_config(
+                config, fixture=fixture, output=output, state=state,
+                ccr=output / "result.ccr", wrapper=wrapper, mirror=base / "mirror",
+                sync_preference="rrdp-then-rsync", rrdp_mode="auto",
+                validation_time="2026-06-01T00:00:00Z",
+                resource_mode="rfc6487", scope="publication-point",
+            )
+            text = "\n".join(line for line in config.read_text().splitlines() if not line.startswith("ta = "))
+            text = text.replace("[strict]", 'ta-refresh-policy = "require-fresh"\n\n[strict]')
+            config.write_text(text + "\n", encoding="utf-8")
+            proof = {"steps": []}
+
+            def validate(name: str) -> tuple[dict, str, list[str]]:
+                start = len(requests)
+                result = run_checked([binary, "validate", "--config", str(config)])
+                assert_successful_output(output, "2026-06-01T00:00:00Z")
+                summary = read_json(output / "summary.json")
+                seen = requests[start:]
+                assert summary["taCacheFallbackCount"] == 0, summary
+                assert not any("delta-" in uri for uri in seen), seen
+                assert not wrapper_log.exists(), "unexpected rsync fallback"
+                proof["steps"].append({
+                    "name": name, "requests": seen, "vrps": summary["vrps"],
+                    "resultStatus": summary["resultStatus"],
+                    "taCacheFallbackCount": summary["taCacheFallbackCount"],
+                    "trustAnchors": summary["trustAnchors"],
+                })
+                return summary, result.stderr, seen
+
+            summary, _, seen = validate("initial_snapshot_newer_ta")
+            assert "/rrdp/snapshot.xml" in seen, seen
+            assert summary["trustAnchors"][0]["source"] == "download", summary
+            cache = next(base.joinpath("state.ta-cache").glob("*.json"))
+            cache_before = cache.read_bytes()
+
+            # A new process sees a mutated historical delta hash at the same
+            # session/serial, and a valid but older TA with the identical key.
+            advertise_hash("33")
+            served_ta.write_bytes((fixture / "ta/custom-ta.cer").read_bytes())
+            summary, stderr, seen = validate("restarted_hash_mutation_and_older_ta")
+            assert "/rrdp/snapshot.xml" in seen, seen
+            assert "rrdp_session_desynchronized" in stderr, stderr
+            anchor = summary["trustAnchors"][0]
+            assert anchor["source"] == "cache" and not anchor["refreshFailed"], anchor
+            assert anchor["selectionReason"] == "cached_newer_not_before", anchor
+            assert cache.read_bytes() == cache_before, "older TA replaced the preferred cache"
+
+            _, stderr, seen = validate("restarted_noop_after_recovery")
+            assert "/rrdp/snapshot.xml" not in seen, seen
+            assert "rrdp_session_desynchronized" not in stderr, stderr
+            assert cache.read_bytes() == cache_before
+            return proof
+        finally:
+            server.shutdown()
+            thread.join(timeout=5)
+            server.server_close()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("binary")
+    parser.add_argument("--proof", type=Path)
+    args = parser.parse_args()
+    proof = run(args.binary)
+    if args.proof:
+        args.proof.parent.mkdir(parents=True, exist_ok=True)
+        args.proof.write_text(json.dumps(proof, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps(proof, indent=2))