diff --git a/Cargo.lock b/Cargo.lock index aa7fddfa3..dc6ffd938 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2539,6 +2539,7 @@ dependencies = [ "serde", "serde_json", "sha1 0.10.6", + "slatedb", "tempfile", "thiserror 2.0.18", "tokio", diff --git a/crab/docs/architecture/indexed-remote-reads.md b/crab/docs/architecture/indexed-remote-reads.md index 8a50df839..eb032baae 100644 --- a/crab/docs/architecture/indexed-remote-reads.md +++ b/crab/docs/architecture/indexed-remote-reads.md @@ -2324,3 +2324,47 @@ the report SHA-256 is The shared `crab-remote` preparation consumer, final 10-GiB qualification, history scaling, provider latency, lifecycle and full blame semantics remain open gates. + + +## Shared capsule preparation in progress + +The shared `Prepared::upload_capsule` caller in HTTP receive now uses captured +file, shard and xorb lookups instead of materializing unrelated historical +catalogs. An ordinary Git upload with no new Crab pointer performs zero catalog +reads. Selected pointer and prepared-content closures preserve size, membership +and descriptor conflict checks. Capsule-specific implementation lives in +`crates/crab-remote/src/prepare/capsule.rs`; its public API remains under +`crab_remote::prepare`. + +Four regressions failed before the change and pass afterward: one redundant +historical-catalog read for a Git-only upload; Xet-word-order formatting of a raw +body digest; acceptance of rewritten xorb/shard scratch bodies after inspection; +and no shared chunk candidate index after a successful native-content commit. +The initial Git-only fixture sent an all-history pack containing an old pointer; +its corrected incremental pack excludes the advertised tip before measuring. +No weakened assertion or production failure is inferred from that fixture issue. + +Six public-API integration tests and strict all-target Clippy pass. The tests use +native Git packs, real Xet artifacts, explicit publication leases, exact origin +and source-anchor checks, complete file reconstruction, and a noncanonical index +fault that must leave committed refs successful. The SlateDB test dependency uses +the same pinned 0.15.0 release already used by metadata; the lockfile adds only +that test dependency edge. Production dependency versions do not change. + +The dedicated RustFS public-API test passes for both newly created and occupied +xorb keys. It uses native Git preparation, real S3 publication, cold source and +origin-anchor checks, and complete file-hash reconstruction. The historical +pointer integration also accepts the correct declared size and rejects a +conflicting one. The final source passes strict all-target Clippy and the +no-default-feature check. + +Eight HTTP receive/fault tests pass. Three dedicated real-RustFS cells also pass: +the public native-content API, native Git push/fetch and browser editing over the +HTTP listener, and the six receive-fault/restart scenarios. Source hashes remain +unchanged throughout these runs. The API test-binary SHA-256 is +`fe8a75004fd4ef7ec0946b999d32a829b83b28f94b51317a9c34e9990e9ebf5d`; the HTTP test-binary SHA-256 is +`92318648278160dd57f96f35dbcffc01f44799a75a43b84442251965c07fef51`. These are small correctness fixtures, separate from the required +large-file, growth and latency cells. + +The final combined 10-GiB matrix, unrelated-history/ref/frontier scaling, broader +lifecycle/provider proof and fuller blame semantics still prevent completion. diff --git a/crates/crab-remote/Cargo.toml b/crates/crab-remote/Cargo.toml index 67659fc1a..bf30ec429 100644 --- a/crates/crab-remote/Cargo.toml +++ b/crates/crab-remote/Cargo.toml @@ -41,6 +41,7 @@ futures-util = { workspace = true, optional = true } object_store = { workspace = true, optional = true } [dev-dependencies] +slatedb = "=0.15.0" sha1 = "0.10" tracing-subscriber = "0.3" async-trait = { workspace = true } diff --git a/crates/crab-remote/README.md b/crates/crab-remote/README.md index 07fabe32e..2e6d60219 100644 --- a/crates/crab-remote/README.md +++ b/crates/crab-remote/README.md @@ -58,6 +58,19 @@ The pinned-view outcome also reports the last inventory published by that pass, or the input inventory if neither phase published. It deliberately excludes later ref-head updates instead of rescanning them for CLI statistics. +Capsule preparation resolves outgoing pointer files and only the prepared +shard/xorb identities against a captured paged catalog. Ordinary Git and LFS-only +uploads do not read historical Xet catalog pages. Selected dependencies retain +complete closure, size and immutable-descriptor conflict checks. + +Scratch ownership does not freeze artifact bytes. Bounded reads revalidate the +inspected raw Blake3 body identity before upload; Xet protocol identities retain +their distinct word-ordered representation. Capsule commitment retains exact +verified create/read identities with their source catalog, publishes rebuildable +chunk hints only after proven ref visibility, and preserves a committed outcome +if hint-index maintenance fails. Uncertain publication does not publish hints. +Caller-owned ref leases and GC fences cover this complete operation. + `objects` owns bounded tree reads, deterministic tree edits and encoding, commit encoding, and Git object identities. `prepare` owns normalized packs and hydrated content artifacts used by remote commits and recovery. File bodies and diff --git a/crates/crab-remote/src/prepare.rs b/crates/crab-remote/src/prepare.rs index d92e34e2f..7210a1e20 100644 --- a/crates/crab-remote/src/prepare.rs +++ b/crates/crab-remote/src/prepare.rs @@ -1,5 +1,5 @@ use std::{ - collections::BTreeMap, + collections::{BTreeMap, HashMap}, fs::File, io::{BufRead, BufReader}, path::PathBuf, @@ -20,7 +20,8 @@ use crab_git::{ }; use crab_metadata::{ capsule_protocol::{ - FileCatalogEntry, PointerCatalog, ShardCatalogEntry, XorbCatalogEntry, XorbChunkEntry, + FileCatalogEntry, PointerCatalog, PointerCatalogLookup, ShardCatalogEntry, + XorbCatalogEntry, XorbChunkEntry, }, git_object_locator::GitObjectKind, git_visibility::{GitCatalogVisibilityIndex, GitVisibilityEdit, GitVisibilityIndex}, @@ -28,8 +29,11 @@ use crab_metadata::{ use crab_remote_git::{OperationContext, OperationKind, RemoteGitRepository}; use gix_hash::ObjectId; use gix_object::Kind; +use tokio::io::AsyncReadExt; use tokio_util::sync::CancellationToken; +mod capsule; + /// Failure while validating and preparing incoming Git objects. #[derive(Debug, thiserror::Error)] pub enum Error { @@ -125,6 +129,41 @@ impl ContentArtifact { pub const fn size(&self) -> u64 { self.size } + + // Scratch ownership does not freeze file bytes. Revalidate the inspected + // identity with a bounded read before publishing content or its receipt. + async fn read_verified(&self, maximum: u64, cancel: &CancellationToken) -> Result { + if cancel.is_cancelled() { + return Err(Error::Cancelled); + } + if self.size > maximum { + return Err(Error::Content( + "prepared artifact exceeds its byte limit".to_owned(), + )); + } + let mut bytes = Vec::new(); + tokio::fs::File::open(&self.path) + .await? + .take(self.size + 1) + .read_to_end(&mut bytes) + .await?; + let bytes = Bytes::from(bytes); + let size = self.size; + let hash = self.body_hash; + let bytes = tokio::task::spawn_blocking(move || { + if bytes.len() as u64 != size || blake3::hash(&bytes).as_bytes() != &hash { + return Err(Error::Content( + "prepared artifact changed after inspection".to_owned(), + )); + } + Ok(bytes) + }) + .await??; + if cancel.is_cancelled() { + return Err(Error::Cancelled); + } + Ok(bytes) + } } /// One Crab pointer supplied by a prepared reconstruction shard. @@ -236,15 +275,14 @@ impl PreparedContent { { return Err(Error::Content("shard identity is invalid".to_owned())); } - let bytes = tokio::fs::read(&artifact.path).await?; - if bytes.len() as u64 != artifact.size - || blake3::hash(&bytes).as_bytes() != &artifact.body_hash - || <[u8; 32]>::from(crab_xet::hash::compute_data_hash(&bytes)) - != artifact.protocol_hash + let bytes = artifact + .read_verified(crab_xet::shard::MAX_PUSH_SHARD_SIZE_BYTES, cancel) + .await?; + if <[u8; 32]>::from(crab_xet::hash::compute_data_hash(&bytes)) != artifact.protocol_hash { return Err(Error::Content("shard binding changed".to_owned())); } - shard_bodies.insert(artifact.protocol_hash, Bytes::from(bytes)); + shard_bodies.insert(artifact.protocol_hash, bytes); } let mut seen_files = std::collections::BTreeSet::new(); let mut used_shards = std::collections::BTreeSet::new(); @@ -330,7 +368,9 @@ impl PreparedContent { crab_xet::hash::MerkleHash::from(artifact.protocol_hash).hex(), XorbCatalogEntry::new( artifact.size, - crab_xet::hash::MerkleHash::from(artifact.body_hash).hex(), + blake3::Hash::from_bytes(artifact.body_hash) + .to_hex() + .to_string(), inspected .chunks .iter() @@ -428,6 +468,8 @@ pub struct CapsuleArtifacts { edits: Vec, packs: Vec, sections: Vec, + pointer_delta: PointerCatalog, + origins: HashMap, changes_namespace: bool, } @@ -809,395 +851,6 @@ impl Prepared { shards, }) } - - /// Upload artifacts for one captured capsule-protocol repository view. - /// - /// The caller must hold ref leases and GC fences from before opening - /// `view` through commitment. This validates all historical dependencies - /// against that view, uploads same-publication Xet content before Git - /// visibility, and retains the exact root snapshot for per-ref CAS. - pub async fn upload_capsule( - &self, - view: &crab_read::capsule_protocol::CapsuleRepositoryView, - limits: crab_read::dependency_proof::DependencyProofLimits, - cancel: &CancellationToken, - ) -> Result { - let started = Instant::now(); - let repository_refs = self - .repository - .refs() - .entries - .iter() - .map(|reference| (reference.name.clone(), reference.target.to_string())) - .collect::>(); - if repository_refs != *view.refs() - || self.repository.generation() != view.root().root().generation() - { - return Err(Error::Request( - "Capsule view differs from the prepared repository version", - )); - } - let base_catalog = view.pointer_catalog().await?; - let prepared_files = self - .content - .as_ref() - .map(|content| { - content - .files - .iter() - .map(|file| file.file_hash) - .collect::>() - }) - .unwrap_or_default(); - crab_read::dependency_proof::verify_capsule_dependencies_except_crab( - &self.layout, - &base_catalog, - self.plan.pointers(), - &prepared_files, - limits, - cancel, - ) - .await - .map_err(|error| Error::Dependency(Box::new(error)))?; - - let pointer_delta = match &self.content { - Some(content) => { - upload_capsule_content(&self.layout, &base_catalog, content, cancel).await? - } - None => PointerCatalog::new(), - }; - let mut packs = Vec::new(); - if let Some(pack) = &self.pack { - if cancel.is_cancelled() { - return Err(Error::Cancelled); - } - let (pack_bytes, index, reverse_index, locator) = tokio::try_join!( - tokio::fs::read(pack.pack_path()), - tokio::fs::read(pack.index_path()), - tokio::fs::read(pack.reverse_path()), - tokio::fs::read(pack.kinds_path()), - )?; - let external_delta_bases = pack - .external_delta_bases() - .iter() - .map(ToString::to_string) - .collect(); - packs.push( - crab_metadata::capsule_protocol::CapsuleGitPack::new_with_external_delta_bases( - Bytes::from(pack_bytes), - Bytes::from(index), - Bytes::from(reverse_index), - Bytes::from(locator), - pack.git_sha1().to_string(), - u64::from(pack.object_count()), - external_delta_bases, - )?, - ); - } - let mut sections = Vec::with_capacity(2); - if !pointer_delta.is_empty() { - sections.push(crab_metadata::capsule_protocol::CapsuleSection::new( - crab_metadata::capsule_protocol::CapsuleSectionKind::CatalogDelta, - pointer_delta.encode_delta()?, - )); - } - if !self.visibility.is_empty() { - let visibility = crab_metadata::capsule_protocol::CapsuleVisibilityDelta::new( - self.visibility.clone(), - )?; - sections.push(crab_metadata::capsule_protocol::CapsuleSection::new( - crab_metadata::capsule_protocol::CapsuleSectionKind::VisibilityDelta, - visibility.encode()?, - )); - } - let edits = self - .updates - .iter() - .map(|update| { - crab_metadata::capsule_protocol::CapsuleRefEdit::new( - update.name.clone(), - update.old.map(|oid| oid.to_string()), - update.new.map(|oid| oid.to_string()), - self.plan - .peeled() - .get(&update.name) - .map(ToString::to_string), - ) - }) - .collect::>(); - let changes_namespace = edits - .iter() - .any(|edit| edit.expected_old().is_none() != edit.new_oid().is_none()); - tracing::info!( - phase = "capsule_input", - packs = packs.len(), - elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, - "Git receive phase completed" - ); - Ok(CapsuleArtifacts { - layout: self.layout.clone(), - base: view.root_snapshot().clone(), - edits, - packs, - sections, - changes_namespace, - }) - } -} - -impl CapsuleArtifacts { - /// Commit through independently mutable ref heads and one capsule authority. - /// - /// Revalidate authorization and repository policy immediately before this - /// call. The caller must retain its original ref leases and GC fences until - /// the returned outcome is known and all lease cleanup has completed. - pub async fn commit( - self, - plan_id: Option<&str>, - namespace_ttl: Duration, - cancel: &CancellationToken, - ) -> Result { - if cancel.is_cancelled() { - return Err(Error::Cancelled); - } - let transaction = match plan_id { - Some(plan_id) => crab_metadata::capsule_protocol::CapsuleTransaction::for_plan( - self.base.record().digest(), - plan_id, - self.edits, - )?, - None => crab_metadata::capsule_protocol::CapsuleTransaction::new( - self.base.record().digest(), - self.edits, - )?, - }; - let transaction_id = transaction.id()?; - let started = Instant::now(); - let capsule = crab_metadata::capsule_protocol::Capsule::build( - &transaction, - self.packs, - self.sections, - )?; - tracing::info!( - phase = "capsule_encode", - %transaction_id, - bytes = capsule.bytes().len(), - elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, - "Git receive phase completed" - ); - let publish = if self.changes_namespace { - let ref_names = transaction - .edits() - .iter() - .map(|edit| edit.ref_name().to_owned()) - .collect::>(); - let layout = self.layout.clone(); - let commit_layout = layout.clone(); - let base = self.base; - crab_write::with_ref_namespaces( - layout.store(), - &layout, - &ref_names, - namespace_ttl, - cancel, - move |scoped| async move { - if scoped.is_cancelled() { - return Err(crab_write::WriteError::Cancelled); - } - crab_write::capsule_protocol::validate_ref_namespace( - &commit_layout, - base.record().root(), - transaction.edits(), - ) - .await?; - crab_write::capsule_protocol::publish( - &commit_layout, - base, - &transaction, - &capsule, - ) - .await - }, - ) - .await - } else { - crab_write::capsule_protocol::publish(&self.layout, self.base, &transaction, &capsule) - .await - }; - match publish { - Ok(_) => Ok(CapsuleCommitOutcome::Committed { transaction_id }), - Err(error @ crab_write::WriteError::CapsuleCommitUncertain { .. }) => { - Ok(CapsuleCommitOutcome::Indeterminate { - transaction_id, - source: Box::new(error), - }) - } - Err(error) => Err(error.into()), - } - } -} - -async fn upload_capsule_content( - layout: &crab_storage::StoreLayout, - base: &PointerCatalog, - content: &PreparedContent, - cancel: &CancellationToken, -) -> Result { - let local = content.pointer_catalog(); - let mut delta = PointerCatalog::new(); - let mut needed_shards = std::collections::BTreeSet::new(); - for (file_hash, file) in local.files() { - if let Some(existing) = base.files().get(file_hash) { - if existing.size() != file.size() { - return Err(Error::Content(format!( - "file {file_hash} conflicts with the captured pointer catalog" - ))); - } - continue; - } - needed_shards.insert(file.shard_hash().to_owned()); - delta.insert_file(file_hash.clone(), file.clone())?; - } - let mut needed_xorbs = std::collections::BTreeSet::new(); - for shard_hash in &needed_shards { - let shard = local - .shards() - .get(shard_hash) - .ok_or_else(|| Error::Content("prepared shard catalog entry is missing".to_owned()))?; - if let Some(existing) = base.shards().get(shard_hash) { - if existing != shard { - return Err(Error::Content(format!( - "shard {shard_hash} conflicts with the captured pointer catalog" - ))); - } - continue; - } - needed_xorbs.extend(shard.xorb_hashes().iter().cloned()); - delta.insert_shard(shard_hash.clone(), shard.clone())?; - } - let xorb_artifacts = content - .xorbs - .iter() - .map(|artifact| { - ( - crab_xet::hash::MerkleHash::from(artifact.protocol_hash).hex(), - artifact, - ) - }) - .collect::>(); - for xorb_hash in needed_xorbs { - let local_entry = local - .xorbs() - .get(&xorb_hash) - .ok_or_else(|| Error::Content("prepared xorb catalog entry is missing".to_owned()))?; - if let Some(existing) = base.xorbs().get(&xorb_hash) { - if existing.chunks() != local_entry.chunks() { - return Err(Error::Content(format!( - "xorb {xorb_hash} conflicts with the captured pointer catalog" - ))); - } - continue; - } - if cancel.is_cancelled() { - return Err(Error::Cancelled); - } - let artifact = xorb_artifacts - .get(&xorb_hash) - .ok_or_else(|| Error::Content("prepared xorb artifact is missing".to_owned()))?; - let bytes = Bytes::from(tokio::fs::read(&artifact.path).await?); - let hash = crab_xet::hash::MerkleHash::from(artifact.protocol_hash); - let entry = match layout - .store() - .create_or_read_immutable( - &layout.xorb_path(&hash), - bytes, - crab_xet::xorb::format::MAX_XORB_SIZE as u64, - ) - .await? - { - crab_storage::ImmutableCreateOutcome::Created(_) => local_entry.clone(), - crab_storage::ImmutableCreateOutcome::Existing { - bytes: existing, .. - } => catalog_xorb_from_existing(&xorb_hash, local_entry, existing)?, - }; - delta.insert_xorb(xorb_hash, entry)?; - } - let shard_artifacts = content - .shards - .iter() - .map(|artifact| { - ( - crab_xet::hash::MerkleHash::from(artifact.protocol_hash).hex(), - artifact, - ) - }) - .collect::>(); - for shard_hash in needed_shards { - if base.shards().contains_key(&shard_hash) { - continue; - } - if cancel.is_cancelled() { - return Err(Error::Cancelled); - } - let artifact = shard_artifacts - .get(&shard_hash) - .ok_or_else(|| Error::Content("prepared shard artifact is missing".to_owned()))?; - let bytes = Bytes::from(tokio::fs::read(&artifact.path).await?); - layout - .store() - .put_if_absent_verified( - &layout.shard_path(&crab_xet::hash::MerkleHash::from(artifact.protocol_hash)), - bytes, - ) - .await?; - } - if !delta.shards().is_empty() { - crab_metadata::ref_registry::union_register_repo_shards( - layout.store(), - layout, - delta.shards().keys().cloned().collect(), - ) - .await?; - } - let mut complete = base.clone(); - complete.apply(&delta)?; - Ok(delta) -} - -fn catalog_xorb_from_existing( - xorb_hash: &str, - local: &XorbCatalogEntry, - bytes: Bytes, -) -> Result { - let parser = - crab_xet::xorb::parser::XorbParser::parse(bytes.clone()).map_err(Error::ContentFormat)?; - parser - .verify_payload_digest() - .map_err(Error::ContentFormat)?; - parser.verify_all_chunks().map_err(Error::ContentFormat)?; - if parser.hash().hex() != xorb_hash { - return Err(Error::Content( - "existing xorb has the wrong logical identity".to_owned(), - )); - } - let chunks = (0..parser.num_chunks()) - .map(|index| { - parser - .chunk_meta(index) - .map(|chunk| XorbChunkEntry::new(chunk.hash.hex(), chunk.uncompressed_len)) - }) - .collect::, _>>() - .map_err(Error::ContentFormat)?; - if chunks != local.chunks() { - return Err(Error::Content( - "existing xorb has a conflicting chunk layout".to_owned(), - )); - } - Ok(XorbCatalogEntry::new( - bytes.len() as u64, - blake3::hash(&bytes).to_hex().to_string(), - chunks, - )) } impl Artifacts<'_> { diff --git a/crates/crab-remote/src/prepare/capsule.rs b/crates/crab-remote/src/prepare/capsule.rs new file mode 100644 index 000000000..dcf26b988 --- /dev/null +++ b/crates/crab-remote/src/prepare/capsule.rs @@ -0,0 +1,495 @@ +use super::*; + +impl Prepared { + /// Upload artifacts for one captured capsule-protocol repository view. + /// + /// The caller must hold ref leases and GC fences from before opening + /// `view` through commitment. This validates all historical dependencies + /// against that view, uploads same-publication Xet content before Git + /// visibility, and retains the exact root snapshot for per-ref CAS. + pub async fn upload_capsule( + &self, + view: &crab_read::capsule_protocol::CapsuleRepositoryView, + limits: crab_read::dependency_proof::DependencyProofLimits, + cancel: &CancellationToken, + ) -> Result { + let started = Instant::now(); + let repository_refs = self + .repository + .refs() + .entries + .iter() + .map(|reference| (reference.name.clone(), reference.target.to_string())) + .collect::>(); + if repository_refs != *view.refs() + || self.repository.generation() != view.root().root().generation() + { + return Err(Error::Request( + "Capsule view differs from the prepared repository version", + )); + } + if self.plan.pointers().len() > limits.max_dependencies { + return Err(Error::Dependency(Box::new( + crab_read::dependency_proof::DependencyProofError::Limit("pointer count"), + ))); + } + let lookup = view.pointer_catalog_lookup( + &self.layout, + crab_metadata::capsule_protocol::MAX_CATALOG_BYTES, + )?; + let hashes = self + .plan + .pointers() + .iter() + .filter_map(|dependency| match &dependency.pointer { + crab_git::pointer_detect::PointerKind::Crab(pointer) => { + Some(crab_xet::hash::MerkleHash::from(pointer.file_hash).hex()) + } + _ => None, + }) + .collect::>(); + // Ordinary Git and LFS-only changes need no Xet catalog pages. Selected + // Crab pointers retain complete authenticated shard/xorb closure checks. + let base_catalog = if hashes.is_empty() { + PointerCatalog::new() + } else { + lookup.files(self.layout.store(), &hashes).await? + }; + let prepared_files = self + .content + .as_ref() + .map(|content| { + content + .files + .iter() + .map(|file| file.file_hash) + .collect::>() + }) + .unwrap_or_default(); + crab_read::dependency_proof::verify_capsule_dependencies_except_crab( + &self.layout, + &base_catalog, + self.plan.pointers(), + &prepared_files, + limits, + cancel, + ) + .await + .map_err(|error| Error::Dependency(Box::new(error)))?; + + let (pointer_delta, origins) = match &self.content { + Some(content) => { + upload_capsule_content(&self.layout, &lookup, base_catalog, content, cancel).await? + } + None => (PointerCatalog::new(), HashMap::new()), + }; + let mut packs = Vec::new(); + if let Some(pack) = &self.pack { + if cancel.is_cancelled() { + return Err(Error::Cancelled); + } + let (pack_bytes, index, reverse_index, locator) = tokio::try_join!( + tokio::fs::read(pack.pack_path()), + tokio::fs::read(pack.index_path()), + tokio::fs::read(pack.reverse_path()), + tokio::fs::read(pack.kinds_path()), + )?; + let external_delta_bases = pack + .external_delta_bases() + .iter() + .map(ToString::to_string) + .collect(); + packs.push( + crab_metadata::capsule_protocol::CapsuleGitPack::new_with_external_delta_bases( + Bytes::from(pack_bytes), + Bytes::from(index), + Bytes::from(reverse_index), + Bytes::from(locator), + pack.git_sha1().to_string(), + u64::from(pack.object_count()), + external_delta_bases, + )?, + ); + } + let mut sections = Vec::with_capacity(1); + if !self.visibility.is_empty() { + let visibility = crab_metadata::capsule_protocol::CapsuleVisibilityDelta::new( + self.visibility.clone(), + )?; + sections.push(crab_metadata::capsule_protocol::CapsuleSection::new( + crab_metadata::capsule_protocol::CapsuleSectionKind::VisibilityDelta, + visibility.encode()?, + )); + } + let edits = self + .updates + .iter() + .map(|update| { + crab_metadata::capsule_protocol::CapsuleRefEdit::new( + update.name.clone(), + update.old.map(|oid| oid.to_string()), + update.new.map(|oid| oid.to_string()), + self.plan + .peeled() + .get(&update.name) + .map(ToString::to_string), + ) + }) + .collect::>(); + let changes_namespace = edits + .iter() + .any(|edit| edit.expected_old().is_none() != edit.new_oid().is_none()); + tracing::info!( + phase = "capsule_input", + packs = packs.len(), + elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, + "Git receive phase completed" + ); + Ok(CapsuleArtifacts { + layout: self.layout.clone(), + base: view.root_snapshot().clone(), + edits, + packs, + sections, + pointer_delta, + origins, + changes_namespace, + }) + } +} + +impl CapsuleArtifacts { + /// Commit through independently mutable ref heads and one capsule authority. + /// + /// Revalidate authorization and repository policy immediately before this + /// call. The caller must retain its original ref leases and GC fences until + /// the returned outcome is known and all lease cleanup has completed. + pub async fn commit( + self, + plan_id: Option<&str>, + namespace_ttl: Duration, + cancel: &CancellationToken, + ) -> Result { + if cancel.is_cancelled() { + return Err(Error::Cancelled); + } + let transaction = match plan_id { + Some(plan_id) => crab_metadata::capsule_protocol::CapsuleTransaction::for_plan( + self.base.record().digest(), + plan_id, + self.edits, + )?, + None => crab_metadata::capsule_protocol::CapsuleTransaction::new( + self.base.record().digest(), + self.edits, + )?, + }; + let transaction_id = transaction.id()?; + let started = Instant::now(); + let mut sections = self.sections; + if !self.pointer_delta.is_empty() { + sections.insert( + 0, + crab_metadata::capsule_protocol::CapsuleSection::new( + crab_metadata::capsule_protocol::CapsuleSectionKind::CatalogDelta, + self.pointer_delta.encode_delta()?, + ), + ); + } + let capsule = + crab_metadata::capsule_protocol::Capsule::build(&transaction, self.packs, sections)?; + tracing::info!( + phase = "capsule_encode", + %transaction_id, + bytes = capsule.bytes().len(), + elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, + "Git receive phase completed" + ); + let publish = if self.changes_namespace { + let ref_names = transaction + .edits() + .iter() + .map(|edit| edit.ref_name().to_owned()) + .collect::>(); + let layout = self.layout.clone(); + let commit_layout = layout.clone(); + let base = self.base; + crab_write::with_ref_namespaces( + layout.store(), + &layout, + &ref_names, + namespace_ttl, + cancel, + move |scoped| async move { + if scoped.is_cancelled() { + return Err(crab_write::WriteError::Cancelled); + } + crab_write::capsule_protocol::validate_ref_namespace( + &commit_layout, + base.record().root(), + transaction.edits(), + ) + .await?; + crab_write::capsule_protocol::publish( + &commit_layout, + base, + &transaction, + &capsule, + ) + .await + }, + ) + .await + } else { + crab_write::capsule_protocol::publish(&self.layout, self.base, &transaction, &capsule) + .await + }; + match publish { + Ok(_) => { + // Ref visibility owns the payload before hints become discoverable. + // Failed hint maintenance cannot reject an already committed Git push. + let config = + crab_metadata::remote_index::RemoteIndexConfig::for_repo_with_global_prefix( + self.layout.repo_prefix(), + self.layout.global_prefix(), + ); + if let Err(error) = crab_metadata::remote_index::write_capsule_chunk_candidates( + Arc::clone(self.layout.store().inner()), + &config, + self.layout.repo_prefix(), + &self.pointer_delta, + &self.origins, + ) + .await + { + tracing::warn!(%error, "capsule committed; chunk candidate index needs repair"); + } + Ok(CapsuleCommitOutcome::Committed { transaction_id }) + } + Err(error @ crab_write::WriteError::CapsuleCommitUncertain { .. }) => { + Ok(CapsuleCommitOutcome::Indeterminate { + transaction_id, + source: Box::new(error), + }) + } + Err(error) => Err(error.into()), + } + } +} + +async fn upload_capsule_content( + layout: &crab_storage::StoreLayout, + lookup: &PointerCatalogLookup, + mut base: PointerCatalog, + content: &PreparedContent, + cancel: &CancellationToken, +) -> Result<( + PointerCatalog, + HashMap, +)> { + let local = content.pointer_catalog(); + let mut delta = PointerCatalog::new(); + let mut origins = HashMap::new(); + let mut needed_shards = std::collections::BTreeSet::new(); + for (file_hash, file) in local.files() { + if let Some(existing) = base.files().get(file_hash) { + if existing.size() != file.size() { + return Err(Error::Content(format!( + "file {file_hash} conflicts with the captured pointer catalog" + ))); + } + continue; + } + needed_shards.insert(file.shard_hash().to_owned()); + delta.insert_file(file_hash.clone(), file.clone())?; + } + if !needed_shards.is_empty() { + base.apply( + &lookup + .shards( + layout.store(), + &needed_shards.iter().cloned().collect::>(), + ) + .await?, + )?; + } + let mut needed_xorbs = std::collections::BTreeSet::new(); + for shard_hash in &needed_shards { + let shard = local + .shards() + .get(shard_hash) + .ok_or_else(|| Error::Content("prepared shard catalog entry is missing".to_owned()))?; + if let Some(existing) = base.shards().get(shard_hash) { + if existing != shard { + return Err(Error::Content(format!( + "shard {shard_hash} conflicts with the captured pointer catalog" + ))); + } + continue; + } + needed_xorbs.extend(shard.xorb_hashes().iter().cloned()); + delta.insert_shard(shard_hash.clone(), shard.clone())?; + } + if !needed_xorbs.is_empty() { + base.apply( + &lookup + .xorbs( + layout.store(), + &needed_xorbs.iter().cloned().collect::>(), + ) + .await?, + )?; + } + let xorb_artifacts = content + .xorbs + .iter() + .map(|artifact| { + ( + crab_xet::hash::MerkleHash::from(artifact.protocol_hash).hex(), + artifact, + ) + }) + .collect::>(); + for xorb_hash in needed_xorbs { + let local_entry = local + .xorbs() + .get(&xorb_hash) + .ok_or_else(|| Error::Content("prepared xorb catalog entry is missing".to_owned()))?; + if let Some(existing) = base.xorbs().get(&xorb_hash) { + if existing.chunks() != local_entry.chunks() { + return Err(Error::Content(format!( + "xorb {xorb_hash} conflicts with the captured pointer catalog" + ))); + } + continue; + } + if cancel.is_cancelled() { + return Err(Error::Cancelled); + } + let artifact = xorb_artifacts + .get(&xorb_hash) + .ok_or_else(|| Error::Content("prepared xorb artifact is missing".to_owned()))?; + let bytes = artifact + .read_verified(crab_xet::xorb::format::MAX_XORB_SIZE as u64, cancel) + .await?; + let hash = crab_xet::hash::MerkleHash::from(artifact.protocol_hash); + let path = layout.xorb_path(&hash); + let payload_digest = crab_xet::xorb::parser::XorbParser::parse(bytes.clone()) + .map_err(Error::ContentFormat)? + .payload_digest(); + let (entry, payload_digest, identity) = match layout + .store() + .create_or_read_immutable(&path, bytes, crab_xet::xorb::format::MAX_XORB_SIZE as u64) + .await? + { + crab_storage::ImmutableCreateOutcome::Created(identity) => { + (local_entry.clone(), payload_digest, identity) + } + crab_storage::ImmutableCreateOutcome::Existing { + bytes: existing, + identity, + } => { + let expected = local_entry.clone(); + let expected_hash = xorb_hash.clone(); + let (entry, payload_digest) = tokio::task::spawn_blocking(move || { + catalog_xorb_from_existing(&expected_hash, &expected, existing) + }) + .await??; + (entry, payload_digest, identity) + } + }; + origins.insert( + hash, + crab_metadata::receipts::OriginReceipt::new( + "canonical-origin".to_owned(), + path.to_string(), + hash.into(), + payload_digest, + entry.encoded_size(), + identity.e_tag, + identity.version, + ), + ); + delta.insert_xorb(xorb_hash, entry)?; + } + let shard_artifacts = content + .shards + .iter() + .map(|artifact| { + ( + crab_xet::hash::MerkleHash::from(artifact.protocol_hash).hex(), + artifact, + ) + }) + .collect::>(); + for shard_hash in needed_shards { + if base.shards().contains_key(&shard_hash) { + continue; + } + if cancel.is_cancelled() { + return Err(Error::Cancelled); + } + let artifact = shard_artifacts + .get(&shard_hash) + .ok_or_else(|| Error::Content("prepared shard artifact is missing".to_owned()))?; + let bytes = artifact + .read_verified(crab_xet::shard::MAX_PUSH_SHARD_SIZE_BYTES, cancel) + .await?; + layout + .store() + .put_if_absent_verified( + &layout.shard_path(&crab_xet::hash::MerkleHash::from(artifact.protocol_hash)), + bytes, + ) + .await?; + } + if !delta.shards().is_empty() { + crab_metadata::ref_registry::union_register_repo_shards( + layout.store(), + layout, + delta.shards().keys().cloned().collect(), + ) + .await?; + } + base.apply(&delta)?; + Ok((delta, origins)) +} + +fn catalog_xorb_from_existing( + xorb_hash: &str, + local: &XorbCatalogEntry, + bytes: Bytes, +) -> Result<(XorbCatalogEntry, [u8; 32])> { + let parser = + crab_xet::xorb::parser::XorbParser::parse(bytes.clone()).map_err(Error::ContentFormat)?; + parser + .verify_payload_digest() + .map_err(Error::ContentFormat)?; + parser.verify_all_chunks().map_err(Error::ContentFormat)?; + if parser.hash().hex() != xorb_hash { + return Err(Error::Content( + "existing xorb has the wrong logical identity".to_owned(), + )); + } + let chunks = (0..parser.num_chunks()) + .map(|index| { + parser + .chunk_meta(index) + .map(|chunk| XorbChunkEntry::new(chunk.hash.hex(), chunk.uncompressed_len)) + }) + .collect::, _>>() + .map_err(Error::ContentFormat)?; + if chunks != local.chunks() { + return Err(Error::Content( + "existing xorb has a conflicting chunk layout".to_owned(), + )); + } + Ok(( + XorbCatalogEntry::new( + bytes.len() as u64, + blake3::hash(&bytes).to_hex().to_string(), + chunks, + ), + parser.payload_digest(), + )) +} diff --git a/crates/crab-remote/tests/capsule_preparation.rs b/crates/crab-remote/tests/capsule_preparation.rs new file mode 100644 index 000000000..52d9db6fe --- /dev/null +++ b/crates/crab-remote/tests/capsule_preparation.rs @@ -0,0 +1,621 @@ +#![cfg(feature = "publication")] + +use std::sync::Arc; + +use bytes::Bytes; +use crab_remote::prepare::{ContentArtifact, ContentFile, PreparedContent}; +use crab_xet::hash::{MerkleHash, compute_data_hash}; +use crab_xet::shard::{PushShardSession, file_info_from_placements, xorb_info_from_placements}; +use crab_xet::xorb::builder::{RunId, XorbBuilder}; +use crab_xet::xorb::format::Chunk; +use tokio_util::sync::CancellationToken; + +const LIMIT: u64 = 8 * 1024 * 1024; + +async fn content_fixture() -> (tempfile::TempDir, PreparedContent) { + let directory = tempfile::tempdir().unwrap(); + let data = Bytes::from_static(b"prepared capsule content must remain reconstructable"); + let chunk_hash = compute_data_hash(&data); + let file_hash = MerkleHash::from(*blake3::hash(&data).as_bytes()); + let mut builder = XorbBuilder::new(); + builder + .push( + &Chunk { + hash: chunk_hash, + data: data.clone(), + }, + RunId(0), + ) + .unwrap(); + let xorb = builder.finalize().unwrap().pop().unwrap(); + let placements = xorb + .placements + .iter() + .map(|placement| (placement.chunk_hash, placement.clone())) + .collect(); + let file = file_info_from_placements(file_hash, &[chunk_hash], &placements).unwrap(); + let xorb_info = Arc::new(xorb_info_from_placements(xorb.hash, &xorb.placements).unwrap()); + let mut shards = PushShardSession::new(); + shards.add_file_bundle(file, &[xorb_info]).unwrap(); + let (shard, shard_hash) = shards.finalize().unwrap().pop().unwrap(); + let xorb_path = directory.path().join("content.xorb"); + let shard_path = directory.path().join("content.shard"); + std::fs::write(&xorb_path, &xorb.bytes).unwrap(); + std::fs::write(&shard_path, &shard).unwrap(); + let content = PreparedContent::new( + vec![ContentArtifact::new( + xorb.hash.into(), + *blake3::hash(&xorb.bytes).as_bytes(), + xorb.bytes.len() as u64, + xorb_path, + )], + vec![ContentArtifact::new( + shard_hash.into(), + *blake3::hash(&shard).as_bytes(), + shard.len() as u64, + shard_path, + )], + vec![ContentFile::new( + file_hash.into(), + data.len() as u64, + shard_hash.into(), + )], + LIMIT, + &CancellationToken::new(), + ) + .await + .unwrap(); + (directory, content) +} + +#[tokio::test] +async fn prepared_catalog_preserves_raw_blake3_body_identity() { + let (directory, content) = content_fixture().await; + let expected = blake3::hash(&std::fs::read(directory.path().join("content.xorb")).unwrap()) + .to_hex() + .to_string(); + let actual = content + .pointer_catalog() + .xorbs() + .values() + .next() + .unwrap() + .body_digest(); + assert_eq!( + actual, expected, + "body digests use raw Blake3 hex; only protocol identities use Xet word order" + ); +} + +#[derive(Default)] +struct Reads(std::sync::atomic::AtomicUsize); +impl crab_storage::StorageObserver for Reads { + fn started(&self, _: crab_storage::StorageOperation) {} + fn finished(&self, observation: crab_storage::StorageObservation) { + if matches!( + observation.operation, + crab_storage::StorageOperation::Get | crab_storage::StorageOperation::Range + ) { + self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + } +} + +fn git(directory: &std::path::Path, args: &[&str]) -> Vec { + git_input(directory, args, b"") +} + +fn git_input(directory: &std::path::Path, args: &[&str], input: &[u8]) -> Vec { + use std::io::Write; + let mut child = std::process::Command::new("git") + .args([ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.invalid", + "-c", + "commit.gpgSign=false", + "-c", + "core.hooksPath=/dev/null", + ]) + .args(args) + .current_dir(directory) + .env("GIT_CONFIG_NOSYSTEM", "1") + .env("GIT_CONFIG_GLOBAL", "/dev/null") + .env_remove("GIT_DIR") + .env_remove("GIT_WORK_TREE") + .env_remove("GIT_INDEX_FILE") + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .unwrap(); + child.stdin.take().unwrap().write_all(input).unwrap(); + let output = child.wait_with_output().unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + output.stdout +} + +fn source_commit(directory: &std::path::Path, content: &PreparedContent) { + git(directory, &["init", "-b", "main"]); + let file = &content.files()[0]; + let pointer = format!( + "version https://crab.build/spec/v1\nfile-hash {}\nsize {}\n", + blake3::Hash::from_bytes(*file.file_hash()).to_hex(), + file.size() + ); + std::fs::write(directory.join("model.bin"), pointer).unwrap(); + git(directory, &["add", "model.bin"]); + git(directory, &["commit", "-m", "publish pointer"]); +} + +async fn layout() -> (crab_storage::StoreLayout, Arc) { + let reads = Arc::new(Reads::default()); + let store = crab_storage::Store::new(Arc::new(object_store::memory::InMemory::new())) + .with_storage_observer(reads.clone()); + let layout = crab_storage::StoreLayout::new(store, "repositories/prepared".to_owned()); + crab_write::capsule_protocol::initialize(&layout, &"1".repeat(64), "refs/heads/main") + .await + .unwrap(); + (layout, reads) +} + +fn proof_limits() -> crab_read::dependency_proof::DependencyProofLimits { + crab_read::dependency_proof::DependencyProofLimits { + max_dependencies: 16, + max_total_file_bytes: LIMIT, + max_duration: std::time::Duration::from_secs(60), + lookup: crab_metadata::file_index_lookup::FileIndexLookupLimits { + max_files: 16, + max_shard_visits: 16, + max_shard_bytes: LIMIT, + max_recipe_entries: 1024, + }, + content: crab_read::pointer_proof::PointerProofLimits { + max_file_bytes: LIMIT, + max_shard_bytes: LIMIT, + max_xorb_bytes: LIMIT, + max_read_bytes: LIMIT, + max_chunks: 1024, + max_duration: std::time::Duration::from_secs(60), + }, + } +} + +async fn prepare( + layout: &crab_storage::StoreLayout, + directory: &std::path::Path, + content: Option, +) -> ( + crab_remote::prepare::Prepared, + crab_read::capsule_protocol::CapsuleRepositoryView, + Arc, + crab_remote::publication::PublicationLeases, +) { + let cancel = CancellationToken::new(); + let leases = crab_remote::publication::acquire_leases( + layout.store(), + layout, + ["refs/heads/main".to_owned()], + crab_remote::publication::LeaseOptions::renewing(std::time::Duration::from_secs(60)), + &cancel, + ) + .await + .unwrap(); + let root = crab_write::capsule_protocol::open_root(layout) + .await + .unwrap(); + let view = crab_read::capsule_protocol::open_git_view_from_root( + layout, + root, + crab_read::capsule_protocol::CapsuleReadLimits { + max_capsule_bytes: LIMIT, + max_frontier_bytes: LIMIT, + }, + ) + .await + .unwrap(); + let runtime = Arc::new(crab_remote_git::RemoteGitRuntime::default()); + let repository = view + .git_repository_from_store( + layout.clone(), + crab_remote_git::RepositoryIdentity::new("memory", "prepared", 1).unwrap(), + runtime.clone(), + crab_remote_git::RepositoryOptions::default(), + LIMIT, + &cancel, + ) + .await + .unwrap(); + let oid = gix_hash::ObjectId::from_hex( + String::from_utf8(git(directory, &["rev-parse", "HEAD"])) + .unwrap() + .trim() + .as_bytes(), + ) + .unwrap(); + let mut revisions = String::from("HEAD\n"); + if let Some(old) = view.refs().get("refs/heads/main") { + revisions.push_str(&format!("^{old}\n")); + } + let pack = git_input( + directory, + &["pack-objects", "--stdout", "--revs"], + revisions.as_bytes(), + ); + let incoming = directory.join("incoming.pack"); + std::fs::write(&incoming, pack).unwrap(); + let limits = crab_git::incoming_pack::ReceiveLimits { + max_pack_bytes: LIMIT, + max_objects: 1024, + max_object_bytes: LIMIT as usize, + max_inflated_bytes: LIMIT, + max_delta_depth: 128, + }; + let prepared = crab_remote::prepare::prepare_capsule( + repository, + view.git_visibility_index().await.unwrap(), + crab_git::incoming_pack::ScratchSpace::new(directory, limits), + Some(std::io::BufReader::new( + std::fs::File::open(incoming).unwrap(), + )), + vec![crab_git::receive_plan::RefUpdate { + name: "refs/heads/main".to_owned(), + old: view + .refs() + .get("refs/heads/main") + .map(|oid| gix_hash::ObjectId::from_hex(oid.as_bytes()).unwrap()), + new: Some(oid), + }], + std::collections::BTreeMap::new(), + &cancel, + crab_remote::prepare::Options { + layout: layout.clone(), + graph: crab_git::receive_plan::GraphLimits { + max_ref_updates: 16, + max_graph_steps: 1024, + max_object_bytes: LIMIT as usize, + max_read_bytes: LIMIT, + }, + pack: limits, + policy: |_: &str| crab_git::receive_plan::RefPolicy { + allow_delete: true, + allow_non_fast_forward: false, + }, + }, + ) + .await + .unwrap(); + let prepared = match content { + Some(content) => prepared.attach_content(content).unwrap(), + None => prepared, + }; + (prepared, view, runtime, leases) +} + +#[tokio::test(flavor = "multi_thread")] +async fn capsule_upload_rejects_changed_prepared_bytes() { + let mut rejected = Vec::new(); + for artifact in ["content.xorb", "content.shard"] { + let (directory, content) = content_fixture().await; + source_commit(directory.path(), &content); + let (layout, _) = layout().await; + let (prepared, view, runtime, leases) = + prepare(&layout, directory.path(), Some(content)).await; + std::fs::write(directory.path().join(artifact), b"changed after inspection").unwrap(); + let result = prepared + .upload_capsule(&view, proof_limits(), &CancellationToken::new()) + .await; + leases.release().await; + runtime.shutdown().await; + rejected.push((artifact, result.is_err())); + } + assert!( + rejected.iter().all(|(_, failed)| *failed), + "changed artifacts must fail before ref publication: {rejected:?}" + ); +} + +async fn verify_indexed_publication( + layout: crab_storage::StoreLayout, + occupied: bool, +) { + let (directory, content) = content_fixture().await; + source_commit(directory.path(), &content); + let chunk = content + .pointer_catalog() + .xorbs() + .values() + .next() + .unwrap() + .chunks()[0] + .hash() + .to_owned(); + if occupied { + let artifact = &content.xorbs()[0]; + layout + .store() + .put( + &layout.xorb_path(&MerkleHash::from(*artifact.protocol_hash())), + Bytes::from(std::fs::read(directory.path().join("content.xorb")).unwrap()), + ) + .await + .unwrap(); + } + let (prepared, view, runtime, leases) = prepare(&layout, directory.path(), Some(content)).await; + let artifacts = prepared + .upload_capsule(&view, proof_limits(), &CancellationToken::new()) + .await + .unwrap(); + let result = artifacts + .commit( + None, + std::time::Duration::from_secs(60), + &CancellationToken::new(), + ) + .await; + leases.release_after_commit().await; + runtime.shutdown().await; + assert!(matches!( + result.unwrap(), + crab_remote::prepare::CapsuleCommitOutcome::Committed { .. } + )); + let config = crab_metadata::remote_index::RemoteIndexConfig::for_repo_with_global_prefix( + layout.repo_prefix(), + layout.global_prefix(), + ); + let reader = + slatedb::DbReader::builder(config.chunk_index_path, Arc::clone(layout.store().inner())) + .build() + .await + .expect("shared publication must populate cold native-add candidates"); + let hash = MerkleHash::from_hex(&chunk).unwrap(); + let candidate = reader + .get(crab_metadata::key_codec::capsule_chunk_head_key( + &hash.into(), + )) + .await + .unwrap() + .expect("published payload candidate"); + let placement = crab_metadata::receipts::CommittedChunkPlacement::decode(&candidate).unwrap(); + let anchor = reader + .get(crab_metadata::key_codec::capsule_chunk_anchor_key( + &placement.source_anchor_id, + )) + .await + .unwrap() + .unwrap(); + let anchor = crab_metadata::capsule_protocol::CapsuleChunkAnchor::decode( + &placement.source_anchor_id, + &anchor, + ) + .unwrap(); + reader.close().await.unwrap(); + let path = layout.xorb_path(&MerkleHash::from(placement.xorb_hash)); + let (body, identity) = layout.store().get_with_etag(&path).await.unwrap(); + let root = crab_write::capsule_protocol::open_root(&layout) + .await + .unwrap(); + let current = crab_read::capsule_protocol::open_git_view_from_root( + &layout, + root, + crab_read::capsule_protocol::CapsuleReadLimits { + max_capsule_bytes: LIMIT, + max_frontier_bytes: LIMIT, + }, + ) + .await + .unwrap(); + let catalog = current.pointer_catalog().await.unwrap(); + anchor + .validate_placement(&placement, &catalog, path.as_ref()) + .unwrap(); + assert_eq!( + ( + anchor.origin.etag, + anchor.origin.object_version, + anchor.body_digest + ), + ( + identity.e_tag, + identity.version, + blake3::hash(&body).to_hex().to_string() + ) + ); + let crab_git::pointer_detect::PointerKind::Crab(pointer) = crab_git::pointer_detect::classify( + &std::fs::read(directory.path().join("model.bin")).unwrap(), + ) else { + panic!("fixture pointer"); + }; + crab_read::pointer_proof::verify_crab_pointer( + &layout, + &pointer, + MerkleHash::from_hex(catalog.files().values().next().unwrap().shard_hash()).unwrap(), + proof_limits().content, + &CancellationToken::new(), + ) + .await + .unwrap(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn capsule_commit_indexes_verified_native_payload() { + for occupied in [false, true] { + let (layout, _) = layout().await; + verify_indexed_publication(layout, occupied).await; + } +} + +#[tokio::test(flavor = "multi_thread")] +#[ignore = "requires CRAB_E2E_FLAT_BUCKET and explicit scratch S3 credentials"] +async fn live_capsule_preparation_publishes_and_reconstructs_native_content() { + let bucket = std::env::var("CRAB_E2E_FLAT_BUCKET").expect("explicit scratch bucket required"); + let store = + crab_storage::build_static_env_store(&bucket, crab_storage::StorageProviderKind::S3) + .unwrap(); + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + for occupied in [false, true] { + let prefix = format!("shared-preparation/{nonce}/{occupied}"); + let layout = crab_storage::StoreLayout::with_global_prefix( + store.clone(), + format!("{prefix}/repo"), + format!("{prefix}/global"), + ); + crab_write::capsule_protocol::initialize(&layout, &"1".repeat(64), "refs/heads/main") + .await + .unwrap(); + verify_indexed_publication(layout, occupied).await; + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn git_only_capsule_upload_does_not_read_historical_catalog() { + let (directory, content) = content_fixture().await; + source_commit(directory.path(), &content); + let (layout, reads) = layout().await; + let (prepared, view, runtime, leases) = prepare(&layout, directory.path(), Some(content)).await; + let artifacts = prepared + .upload_capsule(&view, proof_limits(), &CancellationToken::new()) + .await + .unwrap(); + let committed = artifacts + .commit( + None, + std::time::Duration::from_secs(60), + &CancellationToken::new(), + ) + .await + .unwrap(); + assert!(matches!( + committed, + crab_remote::prepare::CapsuleCommitOutcome::Committed { .. } + )); + leases.release_after_commit().await; + runtime.shutdown().await; + git(directory.path(), &["rm", "model.bin"]); + std::fs::write(directory.path().join("README.md"), b"ordinary Git update").unwrap(); + git(directory.path(), &["add", "README.md"]); + git( + directory.path(), + &["commit", "-m", "replace pointer with ordinary Git file"], + ); + let (prepared, view, runtime, leases) = prepare(&layout, directory.path(), None).await; + assert!(prepared.plan().pointers().is_empty()); + reads.0.store(0, std::sync::atomic::Ordering::Relaxed); + let result = prepared + .upload_capsule(&view, proof_limits(), &CancellationToken::new()) + .await; + let count = reads.0.load(std::sync::atomic::Ordering::Relaxed); + leases.release().await; + runtime.shutdown().await; + result.unwrap(); + assert_eq!( + count, 0, + "ordinary Git publication must not read unrelated Xet history" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn committed_capsule_survives_candidate_index_failure() { + let (directory, content) = content_fixture().await; + source_commit(directory.path(), &content); + let (layout, _) = layout().await; + let config = crab_metadata::remote_index::RemoteIndexConfig::for_repo_with_global_prefix( + layout.repo_prefix(), + layout.global_prefix(), + ); + let db = slatedb::Db::open(config.chunk_index_path, Arc::clone(layout.store().inner())) + .await + .unwrap(); + db.put( + crab_metadata::key_codec::encode_system_key(crab_metadata::key_codec::SYS_FORMAT_VERSION), + 2u32.to_le_bytes(), + ) + .await + .unwrap(); + db.close().await.unwrap(); + let (prepared, view, runtime, leases) = prepare(&layout, directory.path(), Some(content)).await; + let artifacts = prepared + .upload_capsule(&view, proof_limits(), &CancellationToken::new()) + .await + .unwrap(); + let result = artifacts + .commit( + None, + std::time::Duration::from_secs(60), + &CancellationToken::new(), + ) + .await; + leases.release_after_commit().await; + runtime.shutdown().await; + assert!( + matches!( + result.unwrap(), + crab_remote::prepare::CapsuleCommitOutcome::Committed { .. } + ), + "repairable hint-index failure must not reject committed refs" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn historical_pointer_lookup_preserves_declared_size_checks() { + for size_delta in [0, 1] { + let (directory, content) = content_fixture().await; + let file = content.files()[0].clone(); + source_commit(directory.path(), &content); + let (layout, _) = layout().await; + let (prepared, view, runtime, leases) = + prepare(&layout, directory.path(), Some(content)).await; + let artifacts = prepared + .upload_capsule(&view, proof_limits(), &CancellationToken::new()) + .await + .unwrap(); + let committed = artifacts + .commit( + None, + std::time::Duration::from_secs(60), + &CancellationToken::new(), + ) + .await + .unwrap(); + assert!(matches!( + committed, + crab_remote::prepare::CapsuleCommitOutcome::Committed { .. } + )); + leases.release_after_commit().await; + runtime.shutdown().await; + // A hint changes the Git blob identity while retaining the existing file. + // Native negotiation therefore sends a real historical-pointer dependency. + let pointer = format!( + "version https://crab.build/spec/v1\nfile-hash {}\nsize {}\nshard-hint {}\n", + blake3::Hash::from_bytes(*file.file_hash()).to_hex(), + file.size() + size_delta, + blake3::Hash::from_bytes(*file.shard_hash()).to_hex() + ); + std::fs::write(directory.path().join("model.bin"), pointer).unwrap(); + git(directory.path(), &["add", "model.bin"]); + git( + directory.path(), + &["commit", "-m", "update pointer binding"], + ); + let (prepared, view, runtime, leases) = prepare(&layout, directory.path(), None).await; + assert_eq!(prepared.plan().pointers().len(), 1); + let result = prepared + .upload_capsule(&view, proof_limits(), &CancellationToken::new()) + .await; + leases.release().await; + runtime.shutdown().await; + assert_eq!( + result.is_ok(), + size_delta == 0, + "historical pointer size delta {size_delta}" + ); + } +}