Use older valid carrier copies during recovery and re-import packs after a prune

docs/spec.md

@@ -460,6 +460,12 @@ inductive check of only the newly reachable range establishes completeness of
 the new state. The frontier advances only after pack authentication, import,
 and the incremental connectivity check all succeed.
 
+Imported pack IDs do not prove that their objects remain in the local object
+database; deleting a remote-tracking ref and pruning removes them. When the
+incremental check fails, the client MUST re-import every pack of the
+authenticated chain and run a full connectivity walk before failing. The
+re-import does not move the floor; only a successful full check advances it.
+
 A missing frontier in a v5 client state requires a full connectivity walk. A
 client state from another wire version is not accepted. The incremental rule assumes the trusted
 client's existing object database has not been corrupted outside this
@@ -482,6 +488,10 @@ publication is a single-parent fast-forward, and a merge can hide a legitimate
 state behind a non-first parent. Discovery MUST deduplicate manifest IDs and
 validate candidates against the authenticated policy and predecessor chain. The byte, outer-commit, candidate, and
 cryptographic-operation budgets MUST be fixed. Exhaustion MUST fail closed.
+Reading a historical manifest or restoring historical ciphertext MUST try each
+stored copy in that bounded history newest first and use the first whose
+content ID matches, so a corrupt copy cannot hide an older valid one. Bytes
+read from rejected copies count against the byte budget.
 Discovery MUST report authenticated descendants and replays. If authenticated
 descendants conflict, recovery MUST refuse to choose a fork. A replayed older
 manifest MUST NOT be treated as a newer candidate. Directory storage has no

src/repository.rs

@@ -744,10 +744,42 @@ impl<S: Storage> EncryptedRepository<S> {
         let state = read_client_state(state_path)?;
         self.validate_pinned_head(&chain, &state)?;
 
-        let stop_generation = state.imported_generation;
         let mut imported: HashSet<String> = state.packs.into_iter().collect();
+        self.import_chain_packs(repo, &chain, state.imported_generation, &mut imported)?;
+        let connectivity_timer = trace::Span::new("inner_ref_connectivity_check");
+        if git::ensure_refs_connected_since(repo, &current.manifest.refs, &state.verified_refs)
+            .is_err()
+        {
+            // Imported pack IDs are cache state, not proof that their objects
+            // are still local: deleting a remote-tracking ref and pruning
+            // removes them. Re-import the whole authenticated chain and check
+            // the full graph; the pinned floor is unchanged until that passes.
+            imported.clear();
+            self.import_chain_packs(repo, &chain, None, &mut imported)?;
+            git::ensure_refs_connected_since(repo, &current.manifest.refs, &BTreeMap::new())?;
+        }
+        drop(connectivity_timer);
+        write_client_state_locked(
+            state_path,
+            &imported,
+            &current.id,
+            &current.manifest,
+            &current.manifest.refs,
+        )?;
+        Ok(current.manifest)
+    }
+
+    /// Imports, oldest first, the packs of every chain entry newer than
+    /// `after_generation`, recording each pack ID in `imported`.
+    fn import_chain_packs(
+        &self,
+        repo: &Path,
+        chain: &[OpenedState],
+        after_generation: Option<u64>,
+        imported: &mut HashSet<String>,
+    ) -> Result<()> {
         for entry in chain.iter().rev() {
-            if stop_generation.is_some_and(|generation| entry.manifest.generation <= generation) {
+            if after_generation.is_some_and(|generation| entry.manifest.generation <= generation) {
                 continue;
             }
             for descriptor in &entry.manifest.new_packs {
@@ -784,17 +816,7 @@ impl<S: Storage> EncryptedRepository<S> {
                 imported.insert(descriptor.id.clone());
             }
         }
-        let connectivity_timer = trace::Span::new("inner_ref_connectivity_check");
-        git::ensure_refs_connected_since(repo, &current.manifest.refs, &state.verified_refs)?;
-        drop(connectivity_timer);
-        write_client_state_locked(
-            state_path,
-            &imported,
-            &current.id,
-            &current.manifest,
-            &current.manifest.refs,
-        )?;
-        Ok(current.manifest)
+        Ok(())
     }
 
     fn validate_remote_continuity(&self, state_path: &Path, chain: &[OpenedState]) -> Result<()> {

src/storage.rs

@@ -12,7 +12,6 @@ use rand::rngs::OsRng;
 use sha2::{Digest, Sha256};
 use thiserror::Error;
 
-use crate::crypto::object_id;
 use crate::error::{BudgetExhausted, ObjectUnavailable};
 use crate::trace;
 
@@ -1143,89 +1142,68 @@ impl GitStorage {
         carrier_chunk_entries(self.root(), commit, kind, id)
     }
 
-    fn object_at_commit(
+    /// Commits in `window` whose tree changed the stored copy of `id`, newest
+    /// first. Carrier history is linear, so each one starts a distinct copy.
+    fn copy_commits(&self, window: &[String], kind: ObjectKind, id: &str) -> Result<Vec<String>> {
+        validate_id(id)?;
+        let Some(newest) = window.first() else {
+            return Ok(Vec::new());
+        };
+        let directory = format!("{}/{}/{id}", kind.directory(), &id[..2]);
+        let output = carrier_git_command()
+            .arg("-C")
+            .arg(self.root())
+            .args([
+                "rev-list",
+                &format!("--max-count={}", window.len()),
+                newest,
+                "--",
+                &directory,
+            ])
+            .output()?;
+        ensure_git_success(&output, "git rev-list carrier object copies")?;
+        let window: HashSet<&str> = window.iter().map(String::as_str).collect();
+        Ok(String::from_utf8(output.stdout)?
+            .lines()
+            .filter(|commit| window.contains(commit))
+            .map(ToOwned::to_owned)
+            .collect())
+    }
+
+    /// Streams the copy of `id` stored at `commit` into `sink` chunk by chunk.
+    /// Returns false when the commit holds no copy or the copy is malformed or
+    /// does not hash to `id`; `sink` errors and process failures propagate.
+    fn read_copy_at(
         &self,
         commit: &str,
         kind: ObjectKind,
         id: &str,
-        max_bytes: u64,
-    ) -> Result<Option<Vec<u8>>> {
-        let Some(chunks) = self.chunk_entries_at(commit, kind, id)? else {
-            return Ok(None);
-        };
-        let mut bytes = Vec::new();
-        for (index, (path, _)) in chunks.iter().enumerate() {
-            let output = carrier_git_command()
-                .arg("-C")
-                .arg(self.root())
-                .args(["show", &format!("{commit}:{path}")])
-                .output()?;
-            ensure_git_success(&output, "git show historical carrier object")?;
-            let length = output.stdout.len();
-            if (index + 1 < chunks.len() && length != CARRIER_CHUNK_SIZE)
-                || (index + 1 == chunks.len() && (length == 0 || length > CARRIER_CHUNK_SIZE))
-            {
-                bail!("historical carrier object chunk has invalid size")
-            }
-            let next_length = (bytes.len() as u64)
-                .checked_add(length as u64)
-                .context("historical carrier object size overflow")?;
-            if next_length > max_bytes {
-                bail!(BudgetExhausted(
-                    "carrier recovery manifest byte budget exhausted".to_owned()
-                ))
-            }
-            bytes.extend_from_slice(&output.stdout);
-        }
-        if object_id(&bytes) != id {
-            bail!("historical carrier object content id mismatch")
-        }
-        Ok(Some(bytes))
-    }
-
-    fn restore_object_at(&self, commit: &str, kind: ObjectKind, id: &str) -> Result<bool> {
-        let Some(chunks) = self.chunk_entries_at(commit, kind, id)? else {
-            return Ok(false);
+        mut sink: impl FnMut(usize, &[u8]) -> Result<()>,
+    ) -> Result<bool> {
+        let chunks = match self.chunk_entries_at(commit, kind, id) {
+            Ok(Some(chunks)) => chunks,
+            Ok(None) => return Ok(false),
+            Err(error) if error.is::<std::io::Error>() => return Err(error),
+            Err(_) => return Ok(false),
         };
-        let staging = tempfile::Builder::new()
-            .prefix("git-remote-e2ee-restore-")
-            .tempdir_in(self.root())?;
-        let object_dir = staging.path().join("object");
-        fs::create_dir(&object_dir)?;
         let mut hasher = Sha256::new();
         for (index, (path, _)) in chunks.iter().enumerate() {
             let output = carrier_git_command()
                 .arg("-C")
                 .arg(self.root())
                 .args(["show", &format!("{commit}:{path}")])
                 .output()?;
-            ensure_git_success(&output, "git show historical carrier object")?;
             let length = output.stdout.len();
-            if (index + 1 < chunks.len() && length != CARRIER_CHUNK_SIZE)
+            if !output.status.success()
+                || (index + 1 < chunks.len() && length != CARRIER_CHUNK_SIZE)
                 || (index + 1 == chunks.len() && (length == 0 || length > CARRIER_CHUNK_SIZE))
             {
-                bail!("historical carrier object chunk has invalid size")
+                return Ok(false);
             }
             hasher.update(&output.stdout);
-            let chunk_path = object_dir.join(format!("{index:08}"));
-            let mut file = File::create(chunk_path)?;
-            file.write_all(&output.stdout)?;
-            file.sync_all()?;
+            sink(index, &output.stdout)?;
         }
-        if hex::encode(hasher.finalize()) != id {
-            bail!("historical carrier object content id mismatch")
-        }
-        let target = self.object_directory(kind, id)?;
-        if target.exists() {
-            fs::remove_dir_all(&target)?;
-        }
-        fs::create_dir_all(
-            target
-                .parent()
-                .context("carrier object path has no parent")?,
-        )?;
-        fs::rename(object_dir, target)?;
-        Ok(true)
+        Ok(hex::encode(hasher.finalize()) == id)
     }
 
     /// Publishes `next` on top of `base_commit`. On failure the checkout is
@@ -1517,10 +1495,11 @@ impl Storage for GitStorage {
             ))
         }
         let mut history = RecoveryHistory::default();
+        // Every byte read counts, including corrupt copies that are skipped.
         let mut total_bytes = 0_u64;
-        for commit_id in commits {
+        for commit_id in &commits {
             let head_id = self
-                .head_bytes_at_commit(&commit_id)?
+                .head_bytes_at_commit(commit_id)?
                 .as_deref()
                 .and_then(|bytes| {
                     std::str::from_utf8(bytes)
@@ -1532,22 +1511,31 @@ impl Storage for GitStorage {
             if let Some(id) = &head_id
                 && !history.manifests.contains_key(id)
             {
-                let remaining = max_manifest_bytes.saturating_sub(total_bytes);
-                if let Some(bytes) =
-                    self.object_at_commit(&commit_id, ObjectKind::Manifest, id, remaining)?
-                {
-                    total_bytes = total_bytes
-                        .checked_add(bytes.len() as u64)
-                        .context("carrier recovery manifest byte count overflow")?;
-                    if total_bytes > max_manifest_bytes {
-                        bail!(BudgetExhausted(
-                            "carrier recovery manifest byte budget exhausted".to_owned()
-                        ))
+                for copy in self.copy_commits(&commits, ObjectKind::Manifest, id)? {
+                    let mut bytes = Vec::new();
+                    let valid =
+                        self.read_copy_at(&copy, ObjectKind::Manifest, id, |_, chunk| {
+                            total_bytes = total_bytes
+                                .checked_add(chunk.len() as u64)
+                                .context("carrier recovery manifest byte count overflow")?;
+                            if total_bytes > max_manifest_bytes {
+                                bail!(BudgetExhausted(
+                                    "carrier recovery manifest byte budget exhausted".to_owned()
+                                ))
+                            }
+                            bytes.extend_from_slice(chunk);
+                            Ok(())
+                        })?;
+                    if valid {
+                        history.manifests.insert(id.clone(), bytes);
+                        break;
                     }
-                    history.manifests.insert(id.clone(), bytes);
                 }
             }
-            history.commits.push(RecoveryCommit { commit_id, head_id });
+            history.commits.push(RecoveryCommit {
+                commit_id: commit_id.clone(),
+                head_id,
+            });
         }
         Ok(Some(history))
     }
@@ -1559,10 +1547,32 @@ impl Storage for GitStorage {
                 "carrier recovery outer-commit budget exhausted".to_owned()
             ))
         }
-        for commit in commits {
-            if self.restore_object_at(&commit, kind, id)? {
-                return Ok(true);
+        for copy in self.copy_commits(&commits, kind, id)? {
+            let staging = tempfile::Builder::new()
+                .prefix("git-remote-e2ee-restore-")
+                .tempdir_in(self.root())?;
+            let object_dir = staging.path().join("object");
+            fs::create_dir(&object_dir)?;
+            let valid = self.read_copy_at(&copy, kind, id, |index, chunk| {
+                let mut file = File::create(object_dir.join(format!("{index:08}")))?;
+                file.write_all(chunk)?;
+                file.sync_all()?;
+                Ok(())
+            })?;
+            if !valid {
+                continue;
             }
+            let target = self.object_directory(kind, id)?;
+            if target.exists() {
+                fs::remove_dir_all(&target)?;
+            }
+            fs::create_dir_all(
+                target
+                    .parent()
+                    .context("carrier object path has no parent")?,
+            )?;
+            fs::rename(object_dir, target)?;
+            return Ok(true);
         }
         Ok(false)
     }

tests/carrier_git.rs

@@ -2161,3 +2161,72 @@ fn carrier_recovery_refuses_to_skip_a_historical_head_with_missing_objects() {
     let carrier_head = git(&carrier, &["show", "refs/heads/main:HEAD"], false);
     assert_eq!(carrier_head, junk_id);
 }
+
+#[test]
+fn carrier_history_reads_skip_corrupt_copies_for_an_older_valid_one() {
+    let temporary = tempfile::tempdir().unwrap();
+    let _cache = isolated_cache(&temporary);
+    let carrier = temporary.path().join("carrier.git");
+    let source = temporary.path().join("source");
+    init_source(&source, "content");
+
+    let owner = KeyFile::generate();
+    init_carrier(temporary.path(), &carrier, &owner);
+    let repository = EncryptedRepository::new(
+        GitStorage::open(carrier.to_str().unwrap()).unwrap(),
+        owner.clone(),
+    );
+    repository
+        .push_ref(&source, "refs/heads/main", false)
+        .unwrap();
+    let (head_id, head) = repository.current_manifest().unwrap();
+    let pack_id = head.new_packs[0].id.clone();
+    let original = GitStorage::open(carrier.to_str().unwrap()).unwrap();
+    let manifest_bytes = original.get_object(ObjectKind::Manifest, &head_id).unwrap();
+    let pack_bytes = original.get_object(ObjectKind::Pack, &pack_id).unwrap();
+
+    edit_carrier(
+        &carrier,
+        &temporary.path().join("carrier-edit"),
+        |workdir| {
+            for (directory, id) in [("manifests", &head_id), ("packs", &pack_id)] {
+                let chunk = workdir
+                    .join(directory)
+                    .join(&id[..2])
+                    .join(id)
+                    .join("00000000");
+                fs::write(chunk, b"corrupt copy").unwrap();
+            }
+        },
+    );
+
+    let storage = GitStorage::open(carrier.to_str().unwrap()).unwrap();
+    assert_ne!(
+        storage.get_object(ObjectKind::Pack, &pack_id).unwrap(),
+        pack_bytes
+    );
+    let history = storage
+        .recovery_history(2048, 64 * 1024 * 1024)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        history.commits[0].head_id.as_deref(),
+        Some(head_id.as_str())
+    );
+    assert_eq!(history.manifests[&head_id], manifest_bytes);
+
+    assert!(
+        storage
+            .restore_historical_object(ObjectKind::Pack, &pack_id)
+            .unwrap()
+    );
+    assert_eq!(
+        storage.get_object(ObjectKind::Pack, &pack_id).unwrap(),
+        pack_bytes
+    );
+    assert!(
+        !storage
+            .restore_historical_object(ObjectKind::Pack, &"0".repeat(64))
+            .unwrap()
+    );
+}