diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs index 1fd7b36eec..5d804253bf 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs @@ -13,9 +13,18 @@ use std::{ time::{Duration, SystemTime}, }; +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; #[cfg(test)] use std::sync::atomic::AtomicUsize; +#[cfg(all(test, unix))] +thread_local! { + static PREPARED_SEGMENT_DIGEST_OPENS: std::cell::Cell = + const { std::cell::Cell::new(0) }; +} + +#[cfg(not(unix))] use same_file::Handle; use sha2::{Digest, Sha256}; use tracedecay_application::code_index::DaemonCodeIndexControlV1; @@ -57,6 +66,10 @@ use tracedecay_code_index::{ use super::{CodeIndexSchedulerErrorV1, PendingHintsV1}; use crate::code_graph_seat::CodeGraphBuildAdmissionV1; +#[cfg(all(test, unix))] +#[path = "tests/staged_segment_reads.rs"] +mod staged_segment_reads; + const MAX_DURABLE_PUBLICATION_POINTER_BYTES: u64 = 512 * 1024; const DURABLE_GENERATION_IO_CHUNK_BYTES_V1: usize = 64 * 1024; @@ -145,7 +158,7 @@ impl SharedCodeIndexBytePoolV1 { pub(super) fn release_dead_entries(&self) { self.bytes .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) + .unwrap_or_else(PoisonError::into_inner) .retain(|_, entry| entry.strong_count() > 0); self.physical_artifacts.release_dead_entries(); } @@ -153,10 +166,7 @@ impl SharedCodeIndexBytePoolV1 { #[cfg(test)] pub(super) fn stats(&self) -> CodeIndexBytePoolStatsV1 { let physical_artifacts = self.physical_artifacts.stats(); - let bytes = self - .bytes - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let bytes = self.bytes.lock().unwrap_or_else(PoisonError::into_inner); CodeIndexBytePoolStatsV1 { parse_chunk_inserted: physical_artifacts.inserted, parse_chunk_reused: physical_artifacts.reused, @@ -214,7 +224,7 @@ struct DecodedGenerationStateV1 { /// The pinned active generation. /// /// Held outside the LRU deque: every unpinned query serves from it, so LRU - /// pressure from pinned or cursor-paged reads of older generations must + /// pressure from pinned or cursor-paged readers of older generations must /// never be able to drop it and force a re-decode on the request path. active: Option>, /// Bumped by every successful publication. A decode that started before a @@ -382,7 +392,7 @@ impl Drop for HeldActiveDecodeV1 { .unwrap_or_else(PoisonError::into_inner); state .in_flight - .retain(|pending| *pending != DecodeSubjectV1::Active); + .retain(|subject| *subject != DecodeSubjectV1::Active); if state.active.is_none() { state.active = self.restore.take(); } @@ -538,6 +548,33 @@ pub(super) struct UnpublishedCodeIndexPublicationV1 { pub(super) publication: CodeIndexSealedPublicationV1, } +// On Unix, directory dev/inode is the same identity used by the handle +// library, without an endpoint-authorized open for each prepare/commit check. +#[cfg(unix)] +#[derive(Clone, Copy, PartialEq, Eq)] +struct UnixDirectoryIdentityV1 { + device: u64, + inode: u64, +} + +#[cfg(unix)] +impl UnixDirectoryIdentityV1 { + fn at(path: &Path) -> Result { + let metadata = path + .metadata() + .map_err(DaemonCodeIndexPublicationStoreV1::unavailable)?; + if !metadata.is_dir() { + return Err(DaemonCodeIndexPublicationStoreV1::unavailable( + "code-generation identity path is not a directory", + )); + } + Ok(Self { + device: metadata.dev(), + inode: metadata.ino(), + }) + } +} + /// An unpublished store transaction. No full store handle is retained here: /// the current handle supplies the source fence when commit is retried. pub struct DaemonCodeIndexPreparationV1 { @@ -545,8 +582,17 @@ pub struct DaemonCodeIndexPreparationV1 { expected_active_generation: Option, undecoded_expectation: Option, incumbent_bytes: Option>, + #[cfg(unix)] + store_identity: UnixDirectoryIdentityV1, + #[cfg(not(unix))] store_identity: Handle, + #[cfg(unix)] + segments_identity: UnixDirectoryIdentityV1, + #[cfg(not(unix))] segments_identity: Handle, + #[cfg(unix)] + generations_identity: UnixDirectoryIdentityV1, + #[cfg(not(unix))] generations_identity: Handle, cache: Arc, active_path: PathBuf, @@ -574,6 +620,20 @@ struct DaemonCodeIndexPreparationFilesV1 { // prohibit removing an open file. Retention outlives all staging cleanup. segments: StagedGenerationSegmentsV1, evidence: TemporaryEvidencePackV1, + // Only the open preparation owns these handles. Its retention lease + // prevents an authorized sweep from unlinking their sealed paths. + #[cfg(unix)] + pinned_segment_reads: VecDeque, + // Eviction limits descriptors but must not discard the identity recorded + // on the first open of any sealed pathname in this preparation. + #[cfg(unix)] + observed_segment_identities: BTreeMap, + #[cfg(unix)] + segment_read_open_attempts: u64, + // One changed sealed pathname invalidates the entire transaction. Retrying + // must not silently accept substituted bytes under the same digest name. + #[cfg(unix)] + segment_reads_poisoned: bool, directory: tempfile::TempDir, _retention: CodeGenerationStoreLockV1, referenced_segment_bytes: u64, @@ -602,6 +662,15 @@ impl DaemonCodeIndexPreparationV1 { }) } + #[cfg(all(test, unix))] + pub(super) fn staged_segment_read_open_attempts_for_test(&self) -> u64 { + let state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + match &*state { + DaemonCodeIndexPreparationStateV1::Open(files) => files.segment_read_open_attempts, + _ => 0, + } + } + fn invalidate_uncommitted(&self) { let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); if matches!(&*state, DaemonCodeIndexPreparationStateV1::Open(_)) { @@ -627,10 +696,16 @@ impl DaemonCodeIndexPreparationV1 { segments, mut evidence, _retention, + #[cfg(unix)] + pinned_segment_reads, .. } = *files; evidence.rollback_unattached(directory.path())?; drop(evidence); + // Release pinned descriptors before staged cleanup so a failed + // preparation cannot retain unlinked sealed content. + #[cfg(unix)] + drop(pinned_segment_reads); drop(segments); let result = directory .close() @@ -664,6 +739,12 @@ impl CodeIndexPreparationPort for DaemonCodeIndexPreparationV1 { let DaemonCodeIndexPreparationStateV1::Open(files) = &mut *state else { return Err(Self::closed()); }; + #[cfg(unix)] + if files.segment_reads_poisoned { + return Err(DaemonCodeIndexPublicationStoreV1::unavailable( + "staged sealed segment identity was invalidated", + )); + } if files.manifest_identity.is_some() { return Err(DaemonCodeIndexPublicationStoreV1::unavailable( "code-index preparation already entered commit", @@ -774,10 +855,12 @@ impl CodeIndexPreparationPort for DaemonCodeIndexPreparationV1 { let state = Arc::clone(&self.state); let segments_root = self.segments_root.clone(); Arc::new(move |request, buffer| { - let state = state + // Refusals before a handle is acquired must not expose a prior read. + buffer.clear(); + let mut state = state .lock() .map_err(|_| CodeIndexProductionErrorV1::Publication(Self::closed()))?; - match &*state { + match &mut *state { DaemonCodeIndexPreparationStateV1::Aborted => { Err(CodeIndexProductionErrorV1::Publication(Self::closed())) } @@ -789,18 +872,180 @@ impl CodeIndexPreparationPort for DaemonCodeIndexPreparationV1 { ) } DaemonCodeIndexPreparationStateV1::Open(files) => { - let digest = match request { - SealedGenerationSegmentReadV1::Whole { digest, .. } - | SealedGenerationSegmentReadV1::Range { digest, .. } => digest, + #[cfg(unix)] + if files.segment_reads_poisoned { + return Err(CodeIndexProductionErrorV1::Contract( + "staged sealed segment identity was invalidated".to_owned(), + )); + } + let (digest, size_bytes, offset, length) = match request { + SealedGenerationSegmentReadV1::Whole { digest, size_bytes } => { + (digest, size_bytes, 0, size_bytes) + } + SealedGenerationSegmentReadV1::Range { + digest, + size_bytes, + offset, + length, + } => (digest, size_bytes, offset, length), }; + // Rejected requests are not cache accesses: they must not + // evict a good handle or change the next eviction victim. + if offset + .checked_add(length) + .is_none_or(|end| end > size_bytes) + { + return Err(CodeIndexProductionErrorV1::Contract( + "sealed generation segment range exceeds its manifest identity" + .to_owned(), + )); + } + usize::try_from(length).map_err(|_| { + CodeIndexProductionErrorV1::Contract( + "sealed generation segment range exceeds addressable memory".to_owned(), + ) + })?; let suffix = sha256_hex_suffix(digest.as_str()).ok_or_else(|| { CodeIndexProductionErrorV1::Contract( "sealed segment digest is not sha256".to_owned(), ) })?; let canonical = segments_root.join(format!("segment-{suffix}.json")); - let path = files.segments.pending.get(&canonical).unwrap_or(&canonical); - DaemonCodeIndexPublicationStoreV1::read_segment_at(path, request, buffer) + let path = files + .segments + .pending + .get(&canonical) + .cloned() + .unwrap_or(canonical); + #[cfg(not(unix))] + { + DaemonCodeIndexPublicationStoreV1::read_segment_at(&path, request, buffer) + } + #[cfg(unix)] + { + // Range reads from the same staged segment can be + // interleaved by the sparse successor. Keep at most + // eight handles, in least-recently-used order. + if let Some(position) = files + .pinned_segment_reads + .iter() + .position(|entry| entry.path == path) + { + let cached = &files.pinned_segment_reads[position]; + if cached.segment.size_bytes != size_bytes { + return Err(CodeIndexProductionErrorV1::Contract( + "sealed generation segment identity does not match its manifest" + .to_owned(), + )); + } + let unchanged = cached.unchanged(); + let entry = + files.pinned_segment_reads.remove(position).ok_or_else(|| { + CodeIndexProductionErrorV1::Contract( + "staged segment read cache lost a handle".to_owned(), + ) + })?; + if unchanged { + files.pinned_segment_reads.push_back(entry); + } else { + // A sealed pathname and its inode must still + // match the snapshot captured at the first open. + files.segment_reads_poisoned = true; + files.pinned_segment_reads.clear(); + return Err(CodeIndexProductionErrorV1::Contract( + "staged sealed segment identity changed while pinned" + .to_owned(), + )); + } + } + if files + .pinned_segment_reads + .back() + .is_none_or(|entry| entry.path != path) + { + // An evicted pathname cannot acquire a new + // identity merely because its handle was closed. + let observed = files.observed_segment_identities.get(&path).cloned(); + let named = path.symlink_metadata(); + if let Some(original) = observed.as_ref() + && !named.as_ref().is_ok_and(|current| { + DaemonCodeIndexPublicationStoreV1::same_unix_segment_identity( + original, current, + ) + }) + { + files.segment_reads_poisoned = true; + files.pinned_segment_reads.clear(); + return Err(CodeIndexProductionErrorV1::Contract( + "staged sealed segment identity changed after eviction" + .to_owned(), + )); + } + let named = named.map_err(|error| { + CodeIndexProductionErrorV1::Contract(format!( + "staged sealed segment is unavailable: {error}" + )) + })?; + if !named.file_type().is_file() || named.len() != size_bytes { + return Err(CodeIndexProductionErrorV1::Contract( + "staged sealed segment identity does not match its manifest" + .to_owned(), + )); + } + // Admit the request and path before eviction, then + // evict before opening to retain the descriptor bound. + if files.pinned_segment_reads.len() == MAX_STAGED_SEGMENT_READ_HANDLES { + files.pinned_segment_reads.pop_front(); + } + let pinned = + DaemonCodeIndexPublicationStoreV1::open_pinned_staged_segment( + &path, + request, + &named, + &mut files.segment_read_open_attempts, + )?; + if let Some(original) = observed.as_ref() { + if !DaemonCodeIndexPublicationStoreV1::same_unix_segment_identity( + original, + &pinned.identity, + ) { + files.segment_reads_poisoned = true; + files.pinned_segment_reads.clear(); + return Err(CodeIndexProductionErrorV1::Contract( + "staged sealed segment identity changed while reopening" + .to_owned(), + )); + } + } else { + files + .observed_segment_identities + .insert(path.clone(), pinned.identity.clone()); + } + files.pinned_segment_reads.push_back(pinned); + } + let pinned = files.pinned_segment_reads.back_mut().ok_or_else(|| { + CodeIndexProductionErrorV1::Contract( + "staged segment read cache has no handle".to_owned(), + ) + })?; + let result = + DaemonCodeIndexPublicationStoreV1::read_pinned_partitioned_segment( + &mut pinned.segment, + request, + buffer, + ); + // The read itself may race an in-place write or path + // replacement. Never expose bytes after that change. + if !pinned.unchanged() { + buffer.clear(); + files.segment_reads_poisoned = true; + files.pinned_segment_reads.clear(); + return Err(CodeIndexProductionErrorV1::Contract( + "staged sealed segment identity changed during read".to_owned(), + )); + } + result + } } } }) @@ -828,6 +1073,11 @@ struct StagedGenerationSegmentsV1 { durable: DurableFileBatch, /// Temporary path of every written segment, keyed by its final name. pending: BTreeMap, + // Canonical files and newly written temporaries are authenticated by + // their original opened handles. Commit rechecks inode identity rather + // than reopening and hashing already authenticated segment pathnames. + #[cfg(unix)] + authenticated_segments: BTreeMap, flushed: bool, } @@ -836,6 +1086,8 @@ impl StagedGenerationSegmentsV1 { Self { durable: DurableFileBatch::new(), pending: BTreeMap::new(), + #[cfg(unix)] + authenticated_segments: BTreeMap::new(), flushed: false, } } @@ -844,6 +1096,32 @@ impl StagedGenerationSegmentsV1 { self.pending.contains_key(final_path) } + // A concurrent publisher can create the final name after our missing-path + // probe. Never replace an immutable content-addressed file in that gap. + fn publish_missing_immutable_file( + temporary: &Path, + final_path: &Path, + ) -> Result<(), CodeIndexPublicationStoreErrorV1> { + match tracedecay_private_fs::framed_log::rename_noreplace(temporary, final_path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + let metadata = final_path + .symlink_metadata() + .map_err(DaemonCodeIndexPublicationStoreV1::unavailable)?; + if !metadata.file_type().is_file() + || !DaemonCodeIndexPublicationStoreV1::files_equal(final_path, temporary)? + { + return Err(DaemonCodeIndexPublicationStoreV1::corruption( + "concurrently published immutable code-index file does not match its content address", + )); + } + std::fs::remove_file(temporary) + .map_err(DaemonCodeIndexPublicationStoreV1::unavailable) + } + Err(error) => Err(DaemonCodeIndexPublicationStoreV1::unavailable(error)), + } + } + /// Flush every staged segment once, then name each. A retry resumes the /// remaining names without trying to sync paths already renamed. fn publish(&mut self) -> Result<(), CodeIndexPublicationStoreErrorV1> { @@ -871,8 +1149,7 @@ impl StagedGenerationSegmentsV1 { .map_err(DaemonCodeIndexPublicationStoreV1::unavailable)?; } Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - std::fs::rename(&temporary, &final_path) - .map_err(DaemonCodeIndexPublicationStoreV1::unavailable)?; + Self::publish_missing_immutable_file(&temporary, &final_path)?; } Err(error) => return Err(DaemonCodeIndexPublicationStoreV1::unavailable(error)), } @@ -896,6 +1173,38 @@ struct PinnedGenerationSegmentV1 { file: File, } +// The file count is bounded independently of the number of staged segments. +#[cfg(unix)] +const MAX_STAGED_SEGMENT_READ_HANDLES: usize = 8; + +#[cfg(unix)] +struct PinnedPreparationSegmentV1 { + path: PathBuf, + segment: PinnedGenerationSegmentV1, + // Both stat and fstat reflect later writes. Keep the original metadata, + // otherwise comparing them to each other accepts same-inode mutations. + identity: std::fs::Metadata, +} + +#[cfg(unix)] +impl PinnedPreparationSegmentV1 { + fn unchanged(&self) -> bool { + self.path + .symlink_metadata() + .ok() + .zip(self.segment.file.metadata().ok()) + .is_some_and(|(named, pinned)| { + DaemonCodeIndexPublicationStoreV1::same_unix_segment_identity( + &self.identity, + &named, + ) && DaemonCodeIndexPublicationStoreV1::same_unix_segment_identity( + &self.identity, + &pinned, + ) + }) + } +} + impl TemporaryEvidencePackV1 { pub(super) fn create(path: PathBuf) -> Result { let file = std::fs::OpenOptions::new() @@ -1345,45 +1654,184 @@ impl DaemonCodeIndexPublicationStoreV1 { )); } let final_path = segments_root.join(format!("segment-{digest_hex}.json")); + // A pending file remains this preparation's verification authority even + // when another publisher installs its canonical name in the meantime. + // Verify its bytes at commit; publish() authenticates the competing + // canonical bytes before deduplicating without replacing them. + if staged.contains(&final_path) { + return Ok(()); + } + // A repeated write of an already authenticated canonical segment + // should not pay another endpoint-scanned open and SHA-256 read. + // The original opened-file witness is authoritative only while its + // exact pathname inode and metadata remain unchanged. + #[cfg(unix)] + if let Some(original) = staged.authenticated_segments.get(&final_path) { + let current = final_path.symlink_metadata().map_err(Self::unavailable)?; + if !Self::same_unix_segment_identity(original, ¤t) { + return Err(Self::corruption( + "previously authenticated sealed segment changed before repeated write", + )); + } + return Ok(()); + } match final_path.symlink_metadata() { Ok(metadata) => { existing_segment_bytes_read.fetch_add(metadata.len(), Ordering::Relaxed); if !metadata.file_type().is_file() || metadata.len() != u64::try_from(bytes.len()).map_err(Self::unavailable)? - || Self::state_digest_file(&final_path)? != expected_digest { return Err(Self::corruption( "existing sealed segment does not match its content address", )); } + #[cfg(unix)] + { + if Self::state_digest_existing_segment(&final_path, &metadata)? + != expected_digest + { + return Err(Self::corruption( + "existing sealed segment does not match its content address", + )); + } + staged.authenticated_segments.insert(final_path, metadata); + } + #[cfg(not(unix))] + if Self::state_digest_file(&final_path)? != expected_digest { + return Err(Self::corruption( + "existing sealed segment does not match its content address", + )); + } return Ok(()); } Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} Err(error) => return Err(Self::unavailable(error)), } - if staged.contains(&final_path) { - return Ok(()); - } let temporary_path = staging_root.join(format!("segment-{digest_hex}.json")); - let mut file = std::fs::OpenOptions::new() - .create_new(true) - .write(true) - .open(&temporary_path) - .map_err(Self::unavailable)?; - staged.pending.insert(final_path, temporary_path.clone()); - file.write_all(bytes).map_err(Self::unavailable)?; - staged - .durable - .written(&temporary_path, &file) - .map_err(Self::unavailable) + let mut options = std::fs::OpenOptions::new(); + options.create_new(true).write(true); + #[cfg(unix)] + options.read(true); + let mut file = options.open(&temporary_path).map_err(Self::unavailable)?; + // Read the actual staged bytes through their writer descriptor. + // A read and an inode/metadata witness prove the bytes at write time; + // no second endpoint-authorized pathname open is required at commit. + // Non-Unix durability and verification remain unchanged. + let admitted = + (|| -> Result, CodeIndexPublicationStoreErrorV1> { + file.write_all(bytes).map_err(Self::unavailable)?; + #[cfg(unix)] + let original = { + let before = file.metadata().map_err(Self::unavailable)?; + file.seek(SeekFrom::Start(0)).map_err(Self::unavailable)?; + if Self::state_digest_reader(&mut file)? != expected_digest { + return Err(Self::corruption( + "newly staged segment differs from its content address", + )); + } + let after = file.metadata().map_err(Self::unavailable)?; + if !Self::same_unix_segment_identity(&before, &after) { + return Err(Self::corruption( + "newly staged segment changed during authentication", + )); + } + after + }; + #[cfg(not(unix))] + let original = file.metadata().map_err(Self::unavailable)?; + staged + .durable + .written_with_identity(&temporary_path, &file, &original) + .map_err(Self::unavailable)?; + // On Linux the batch performed its final identity checks + // before enrolling the member. No fallible post-admission + // check may remove its pathname while retaining its handle. + #[cfg(target_os = "linux")] + { + Ok(Some(original)) + } + #[cfg(all(unix, not(target_os = "linux")))] + { + let after = file.metadata().map_err(Self::unavailable)?; + if !Self::same_unix_segment_identity(&original, &after) { + return Err(Self::corruption( + "newly staged segment changed during durability admission", + )); + } + Ok(Some(after)) + } + #[cfg(not(unix))] + { + Ok(None) + } + })(); + drop(file); + // The pending index is publication authority. Never enroll a file + // whose write, content authentication or durability admission failed. + let witness = match admitted { + Ok(witness) => witness, + Err(error) => { + std::fs::remove_file(&temporary_path).map_err(|cleanup| { + Self::unavailable(format!( + "staged segment admission failed: {error}; temporary cleanup failed: {cleanup}" + )) + })?; + return Err(error); + } + }; + if let Some(witness) = witness { + #[cfg(unix)] + staged + .authenticated_segments + .insert(temporary_path.clone(), witness); + #[cfg(not(unix))] + let _ = witness; + } + staged.pending.insert(final_path, temporary_path); + Ok(()) + } + + #[cfg(unix)] + fn state_digest_existing_segment( + path: &Path, + original: &std::fs::Metadata, + ) -> Result { + #[cfg(test)] + PREPARED_SEGMENT_DIGEST_OPENS.with(|count| count.set(count.get().saturating_add(1))); + let mut file = File::open(path).map_err(Self::unavailable)?; + let opened = file.metadata().map_err(Self::unavailable)?; + if !Self::same_unix_segment_identity(original, &opened) { + return Err(Self::corruption( + "existing sealed segment changed before authentication", + )); + } + let digest = Self::state_digest_reader(&mut file)?; + let after = file.metadata().map_err(Self::unavailable)?; + let named = path.symlink_metadata().map_err(Self::unavailable)?; + if !Self::same_unix_segment_identity(original, &after) + || !Self::same_unix_segment_identity(original, &named) + { + return Err(Self::corruption( + "existing sealed segment changed during authentication", + )); + } + Ok(digest) } fn state_digest_file(path: &Path) -> Result { + #[cfg(all(test, unix))] + PREPARED_SEGMENT_DIGEST_OPENS.with(|count| count.set(count.get().saturating_add(1))); let mut file = File::open(path).map_err(Self::unavailable)?; + Self::state_digest_reader(&mut file) + } + + fn state_digest_reader( + reader: &mut impl Read, + ) -> Result { let mut hasher = Sha256::new(); let mut buffer = vec![0_u8; DURABLE_GENERATION_IO_CHUNK_BYTES_V1]; loop { - let read = file.read(&mut buffer).map_err(Self::unavailable)?; + let read = reader.read(&mut buffer).map_err(Self::unavailable)?; if read == 0 { break; } @@ -1392,6 +1840,17 @@ impl DaemonCodeIndexPublicationStoreV1 { Ok(encode_tagged_lowercase_hex("sha256:", &hasher.finalize())) } + // A canonical generation is immutable. Close the staging descriptor and + // use the same atomic no-replace authority as sealed segments: tempfile's + // no-clobber persist may leave two hard links after an interrupted publish. + fn publish_missing_generation_manifest( + temporary: tempfile::NamedTempFile, + final_path: &Path, + ) -> Result<(), CodeIndexPublicationStoreErrorV1> { + let temporary = temporary.into_temp_path(); + StagedGenerationSegmentsV1::publish_missing_immutable_file(temporary.as_ref(), final_path) + } + fn files_equal(left: &Path, right: &Path) -> Result { let left_metadata = left.symlink_metadata().map_err(Self::unavailable)?; let right_metadata = right.symlink_metadata().map_err(Self::unavailable)?; @@ -1503,6 +1962,27 @@ impl DaemonCodeIndexPublicationStoreV1 { // a fixed-width pointer through another path, and a 1-second mtime // filesystem can leave both unchanged while the bytes move. The memo // is reused only when the file digest matches. + // A cold pointer validation must not authorize a second pathname + // open merely to populate the memo. Keep the original descriptor + // through parsing and authenticate its bytes again before memoizing. + #[cfg(unix)] + let (bytes, mut opened_pointer) = { + let mut file = File::open(&self.active_path).map_err(Self::map_pointer_io)?; + let mut bytes = Vec::new(); + (&mut file) + .take(MAX_DURABLE_PUBLICATION_POINTER_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(Self::map_pointer_io)?; + if u64::try_from(bytes.len()).map_err(Self::unavailable)? + > MAX_DURABLE_PUBLICATION_POINTER_BYTES + { + return Err(Self::corruption( + "durable code-generation index exceeds its byte bound", + )); + } + (bytes, file) + }; + #[cfg(not(unix))] let bytes = std::fs::read(&self.active_path).map_err(Self::map_pointer_io)?; let digest = Self::state_digest(&bytes); { @@ -1633,7 +2113,29 @@ impl DaemonCodeIndexPublicationStoreV1 { .unwrap_or_else(PoisonError::into_inner); // Install only when the file is still the bytes just parsed. A rename // that landed during validation owns the memo. - if std::fs::read(&self.active_path).ok().as_deref() == Some(bytes.as_slice()) { + // On Unix, re-read through the authenticated descriptor rather than + // opening the same endpoint-scanned pathname again. The two inode + // witnesses reject a concurrent rename; the second byte comparison + // rejects an in-place change during parsing, even on coarse clocks. + #[cfg(unix)] + let unchanged = (|| -> std::io::Result { + let opened_before = opened_pointer.metadata()?; + opened_pointer.seek(SeekFrom::Start(0))?; + let mut current = Vec::new(); + (&mut opened_pointer) + .take(MAX_DURABLE_PUBLICATION_POINTER_BYTES + 1) + .read_to_end(&mut current)?; + let opened_after = opened_pointer.metadata()?; + let named_after = std::fs::metadata(&self.active_path)?; + Ok(current == bytes + && Self::same_unix_segment_identity(&metadata, &opened_before) + && Self::same_unix_segment_identity(&opened_before, &opened_after) + && Self::same_unix_segment_identity(&opened_before, &named_after)) + })() + .unwrap_or(false); + #[cfg(not(unix))] + let unchanged = std::fs::read(&self.active_path).ok().as_deref() == Some(bytes.as_slice()); + if unchanged { *memo = Some(PublicationPointerMemoV1 { mtime, size, @@ -1644,27 +2146,57 @@ impl DaemonCodeIndexPublicationStoreV1 { Ok(Some(pointer)) } - fn remember_publication_pointer(&self, pointer: &DurablePublicationPointerV1, bytes: &[u8]) { + fn remember_publication_pointer( + &self, + pointer: &DurablePublicationPointerV1, + bytes: &[u8], + persisted: &mut File, + ) { let mut memo = self .pointer_memo .lock() .unwrap_or_else(PoisonError::into_inner); - // The memo and the file it names are one critical section. A publisher - // that observed older bytes must not install them over a newer file. - match std::fs::read(&self.active_path) { - Ok(current) if current == bytes => { - let metadata = std::fs::metadata(&self.active_path).ok(); - *memo = Some(PublicationPointerMemoV1 { - mtime: metadata - .as_ref() - .and_then(|metadata| metadata.modified().ok()), - size: metadata.map_or(0, |metadata| metadata.len()), - digest: Self::state_digest(bytes), - pointer: pointer.clone(), - }); + // The tempfile descriptor survives its atomic rename. Reopen-free + // readback authenticates that exact inode, while pathname metadata + // prevents memoizing a pointer replaced by another writer. + #[cfg(unix)] + let metadata = (|| -> std::io::Result> { + let opened = persisted.metadata()?; + let named_before = self.active_path.symlink_metadata()?; + if !Self::same_unix_segment_identity(&opened, &named_before) { + return Ok(None); } - Ok(_) => {} - Err(_) => *memo = None, + persisted.seek(SeekFrom::Start(0))?; + let mut current = Vec::new(); + (&mut *persisted) + .take(MAX_DURABLE_PUBLICATION_POINTER_BYTES + 1) + .read_to_end(&mut current)?; + let after = persisted.metadata()?; + let named_after = self.active_path.symlink_metadata()?; + Ok((current == bytes + && Self::same_unix_segment_identity(&opened, &after) + && Self::same_unix_segment_identity(&opened, &named_after)) + .then_some(named_after)) + })() + .ok() + .flatten(); + #[cfg(not(unix))] + let metadata = match std::fs::read(&self.active_path) { + Ok(current) if current == bytes => std::fs::metadata(&self.active_path).ok(), + _ => None, + }; + #[cfg(not(unix))] + let _ = persisted; + if let Some(metadata) = metadata { + *memo = Some(PublicationPointerMemoV1 { + mtime: metadata.modified().ok(), + size: metadata.len(), + digest: Self::state_digest(bytes), + pointer: pointer.clone(), + }); + } else { + // Never preserve an older memo after an unverified replacement. + *memo = None; } } @@ -1711,7 +2243,7 @@ impl DaemonCodeIndexPublicationStoreV1 { .map_err(Self::unavailable)?; temporary.write_all(bytes).map_err(Self::unavailable)?; temporary.as_file().sync_all().map_err(Self::unavailable)?; - temporary + let mut persisted = temporary .persist(&self.active_path) .map_err(|error| Self::map_pointer_io(error.error))?; Self::sync_directory( @@ -1719,7 +2251,7 @@ impl DaemonCodeIndexPublicationStoreV1 { .parent() .ok_or_else(|| Self::unavailable("active pointer has no parent directory"))?, )?; - self.remember_publication_pointer(pointer, bytes); + self.remember_publication_pointer(pointer, bytes, &mut persisted); Ok(()) } @@ -1730,6 +2262,9 @@ impl DaemonCodeIndexPublicationStoreV1 { buffer: &mut Vec, control: &dyn CodeIndexExecutionControlV1, ) -> Result<(), CodeIndexProductionErrorV1> { + // A rejected or interrupted request must not return bytes left by + // an earlier successful read, even when it never reaches the file. + buffer.clear(); let root = self .active_path .parent() @@ -1764,7 +2299,22 @@ impl DaemonCodeIndexPublicationStoreV1 { CodeIndexInterruptionV1::Cancelled, )); } - self.read_partitioned_segment(request, buffer) + let result = self.read_partitioned_segment(request, buffer); + // Endpoint authorization may block inside open(). A request that was + // live before I/O must not return data after its deadline or cancel. + if control.is_cancelled() { + buffer.clear(); + return Err(CodeIndexProductionErrorV1::Interrupted( + CodeIndexInterruptionV1::Cancelled, + )); + } + if control.is_deadline_exceeded() { + buffer.clear(); + return Err(CodeIndexProductionErrorV1::Interrupted( + CodeIndexInterruptionV1::DeadlineExceeded, + )); + } + result } fn read_partitioned_segment( @@ -1794,6 +2344,7 @@ impl DaemonCodeIndexPublicationStoreV1 { request: SealedGenerationSegmentReadV1<'_>, buffer: &mut Vec, ) -> Result<(), CodeIndexProductionErrorV1> { + buffer.clear(); let digest = match request { SealedGenerationSegmentReadV1::Whole { digest, .. } | SealedGenerationSegmentReadV1::Range { digest, .. } => digest, @@ -1810,16 +2361,17 @@ impl DaemonCodeIndexPublicationStoreV1 { request: SealedGenerationSegmentReadV1<'_>, buffer: &mut Vec, ) -> Result<(), CodeIndexProductionErrorV1> { - let (_digest, expected_size, offset, length) = match request { + buffer.clear(); + let (digest, expected_size, offset, length, whole) = match request { SealedGenerationSegmentReadV1::Whole { digest, size_bytes } => { - (digest, size_bytes, 0, size_bytes) + (digest, size_bytes, 0, size_bytes, true) } SealedGenerationSegmentReadV1::Range { digest, size_bytes, offset, length, - } => (digest, size_bytes, offset, length), + } => (digest, size_bytes, offset, length, false), }; if offset .checked_add(length) @@ -1839,23 +2391,55 @@ impl DaemonCodeIndexPublicationStoreV1 { "sealed generation segment identity does not match its manifest".to_owned(), )); } - buffer.clear(); let length = usize::try_from(length).map_err(|_| { CodeIndexProductionErrorV1::Contract( "sealed generation segment range exceeds addressable memory".to_owned(), ) })?; buffer.resize(length, 0); - File::open(path) - .and_then(|mut file| { - file.seek(SeekFrom::Start(offset))?; - file.read_exact(buffer) - }) - .map_err(|error| { - CodeIndexProductionErrorV1::Contract(format!( - "sealed generation segment read failed: {error}" - )) - }) + let read = (|| -> std::io::Result<()> { + let mut file = File::open(path)?; + #[cfg(unix)] + { + let opened = file.metadata()?; + if !Self::same_unix_segment_identity(&metadata, &opened) { + return Err(std::io::Error::other( + "sealed segment pathname changed while opening", + )); + } + } + file.seek(SeekFrom::Start(offset))?; + file.read_exact(buffer)?; + #[cfg(unix)] + { + let opened = file.metadata()?; + let named = path.symlink_metadata()?; + if !Self::same_unix_segment_identity(&metadata, &opened) + || !Self::same_unix_segment_identity(&metadata, &named) + { + return Err(std::io::Error::other( + "sealed segment pathname or bytes changed during read", + )); + } + } + Ok(()) + })(); + if let Err(error) = read { + buffer.clear(); + return Err(CodeIndexProductionErrorV1::Contract(format!( + "sealed generation segment read failed: {error}" + ))); + } + // A full segment has its content address available without another + // pathname open. Never return same-length corrupt bytes to a caller. + // Range callers retain their existing per-page digest authority. + if whole && Self::state_digest(buffer) != digest.as_str() { + buffer.clear(); + return Err(CodeIndexProductionErrorV1::Contract( + "sealed generation segment does not match its content address".to_owned(), + )); + } + Ok(()) } fn open_pinned_partitioned_segment( @@ -1907,33 +2491,117 @@ impl DaemonCodeIndexPublicationStoreV1 { "sealed generation segment metadata cannot be read: {error}" )) })?; - let file_identity = Handle::from_file(file.try_clone().map_err(|error| { + // On Unix the pathname/opened inode and change-time identity can be + // compared without Handle::from_path (another endpoint-scanned open). + #[cfg(unix)] + let same_identity = { + // Also re-stat the pathname after opening. Comparing just the + // pre-open path and fd would miss a replacement after open, while + // Handle::from_path would require another endpoint-scanned open. + let path_after_open = path.symlink_metadata().map_err(|error| { + CodeIndexProductionErrorV1::Contract(format!( + "sealed generation segment pathname cannot be rechecked: {error}" + )) + })?; + Self::same_unix_segment_identity(&path_metadata, &file_metadata) + && Self::same_unix_segment_identity(&path_after_open, &file_metadata) + }; + #[cfg(not(unix))] + let same_identity = { + let file_identity = Handle::from_file(file.try_clone().map_err(|error| { + CodeIndexProductionErrorV1::Contract(format!( + "sealed generation segment handle cannot be cloned: {error}" + )) + })?) + .map_err(|error| { + CodeIndexProductionErrorV1::Contract(format!( + "sealed generation segment handle identity cannot be read: {error}" + )) + })?; + let path_identity = Handle::from_path(&path).map_err(|error| { + CodeIndexProductionErrorV1::Contract(format!( + "sealed generation segment path identity cannot be read: {error}" + )) + })?; + file_identity == path_identity + }; + if !file_metadata.is_file() || file_metadata.len() != expected_size || !same_identity { + return Err(CodeIndexProductionErrorV1::Contract( + "sealed generation segment identity changed while it was opened".to_owned(), + )); + } + Ok(PinnedGenerationSegmentV1 { + digest: digest.to_owned(), + size_bytes: expected_size, + file, + }) + } + + #[cfg(unix)] + fn same_unix_segment_identity(named: &std::fs::Metadata, pinned: &std::fs::Metadata) -> bool { + named.file_type().is_file() + && pinned.is_file() + && named.dev() == pinned.dev() + && named.ino() == pinned.ino() + && named.len() == pinned.len() + && named.mtime() == pinned.mtime() + && named.mtime_nsec() == pinned.mtime_nsec() + && named.ctime() == pinned.ctime() + && named.ctime_nsec() == pinned.ctime_nsec() + } + + /// Open an admitted staged pathname without another pre-open stat or + /// `Handle::from_path` open. The caller validates the request and metadata + /// before evicting a cached handle; this checks identity across the open. + #[cfg(unix)] + fn open_pinned_staged_segment( + path: &Path, + request: SealedGenerationSegmentReadV1<'_>, + named: &std::fs::Metadata, + open_attempts: &mut u64, + ) -> Result { + let (digest, size_bytes) = match request { + SealedGenerationSegmentReadV1::Whole { digest, size_bytes } + | SealedGenerationSegmentReadV1::Range { + digest, size_bytes, .. + } => (digest.as_str(), size_bytes), + }; + *open_attempts = (*open_attempts).saturating_add(1); + tracing::trace!( + target: "tracedecay::code_index", + staged_segment_file_open_count = *open_attempts, + "attempted staged segment pathname open after an LRU miss" + ); + let file = File::open(path).map_err(|error| { CodeIndexProductionErrorV1::Contract(format!( - "sealed generation segment handle cannot be cloned: {error}" + "staged sealed segment cannot be opened: {error}" )) - })?) - .map_err(|error| { + })?; + let opened = file.metadata().map_err(|error| { CodeIndexProductionErrorV1::Contract(format!( - "sealed generation segment handle identity cannot be read: {error}" + "staged sealed segment metadata cannot be read: {error}" )) })?; - let path_identity = Handle::from_path(&path).map_err(|error| { + let named_after_open = path.symlink_metadata().map_err(|error| { CodeIndexProductionErrorV1::Contract(format!( - "sealed generation segment path identity cannot be read: {error}" + "staged sealed segment pathname cannot be rechecked: {error}" )) })?; - if !file_metadata.is_file() - || file_metadata.len() != expected_size - || file_identity != path_identity + if !Self::same_unix_segment_identity(named, &opened) + || !Self::same_unix_segment_identity(&named_after_open, &opened) { return Err(CodeIndexProductionErrorV1::Contract( - "sealed generation segment identity changed while it was opened".to_owned(), + "staged sealed segment was replaced while opening".to_owned(), )); } - Ok(PinnedGenerationSegmentV1 { - digest: digest.to_owned(), - size_bytes: expected_size, - file, + Ok(PinnedPreparationSegmentV1 { + path: path.to_path_buf(), + identity: opened, + segment: PinnedGenerationSegmentV1 { + digest: digest.to_owned(), + size_bytes, + file, + }, }) } @@ -1942,6 +2610,8 @@ impl DaemonCodeIndexPublicationStoreV1 { request: SealedGenerationSegmentReadV1<'_>, buffer: &mut Vec, ) -> Result<(), CodeIndexProductionErrorV1> { + // Errors must not leave bytes from an earlier successful range visible. + buffer.clear(); let (digest, expected_size, offset, length) = match request { SealedGenerationSegmentReadV1::Whole { digest, size_bytes } => { (digest.as_str(), size_bytes, 0, size_bytes) @@ -1968,13 +2638,13 @@ impl DaemonCodeIndexPublicationStoreV1 { "sealed generation segment range exceeds addressable memory".to_owned(), ) })?; - buffer.clear(); buffer.resize(length, 0); pinned .file .seek(SeekFrom::Start(offset)) .and_then(|_| pinned.file.read_exact(buffer)) .map_err(|error| { + buffer.clear(); CodeIndexProductionErrorV1::Contract(format!( "sealed generation segment read failed: {error}" )) @@ -3018,9 +3688,18 @@ impl CodeIndexAtomicPublicationPort for DaemonCodeIndexPublicationStoreV1 { { return Err(CodeIndexPublicationStoreErrorV1::CompareAndSwap); } + #[cfg(unix)] + let store_identity = UnixDirectoryIdentityV1::at(store_root)?; + #[cfg(not(unix))] let store_identity = Handle::from_path(store_root).map_err(Self::unavailable)?; + #[cfg(unix)] + let segments_identity = UnixDirectoryIdentityV1::at(&self.segments_root)?; + #[cfg(not(unix))] let segments_identity = Handle::from_path(&self.segments_root).map_err(Self::unavailable)?; + #[cfg(unix)] + let generations_identity = UnixDirectoryIdentityV1::at(&self.generations_root)?; + #[cfg(not(unix))] let generations_identity = Handle::from_path(&self.generations_root).map_err(Self::unavailable)?; self.seal_checkpoint()?; @@ -3053,6 +3732,14 @@ impl CodeIndexAtomicPublicationPort for DaemonCodeIndexPublicationStoreV1 { directory, segments: StagedGenerationSegmentsV1::new(), evidence, + #[cfg(unix)] + pinned_segment_reads: VecDeque::new(), + #[cfg(unix)] + observed_segment_identities: BTreeMap::new(), + #[cfg(unix)] + segment_read_open_attempts: 0, + #[cfg(unix)] + segment_reads_poisoned: false, _retention: retention, referenced_segment_bytes: 0, manifest_identity: None, @@ -3112,6 +3799,12 @@ impl DaemonCodeIndexPublicationStoreV1 { let DaemonCodeIndexPreparationStateV1::Open(files) = &mut *session else { return Err(DaemonCodeIndexPreparationV1::closed()); }; + #[cfg(unix)] + if files.segment_reads_poisoned { + return Err(Self::unavailable( + "staged sealed segment identity was invalidated", + )); + } let manifest_bytes = generation.manifest_bytes(); let generation_size = u64::try_from(manifest_bytes.len()).map_err(Self::unavailable)?; if generation_size > MAX_DURABLE_GENERATION_INDEX_BYTES_V1 { @@ -3263,7 +3956,19 @@ impl DaemonCodeIndexPublicationStoreV1 { let state_digest = { let _span = tracing::trace_span!("code_index.generation.publish.state_digest").entered(); - Self::state_digest_file(temporary.path())? + // Read back the fsynced bytes through the writer's existing handle. + // A second pathname open is unnecessary and expensive under EDR. + temporary + .as_file_mut() + .seek(SeekFrom::Start(0)) + .map_err(Self::unavailable)?; + let digest = Self::state_digest_reader(temporary.as_file_mut())?; + if digest != Self::state_digest(manifest_bytes) { + return Err(Self::corruption( + "staged generation bytes changed after serialization", + )); + } + digest }; let suffix = sha256_hex_suffix(&state_digest) .ok_or_else(|| Self::unavailable("sealed generation digest is not sha256"))?; @@ -3283,11 +3988,9 @@ impl DaemonCodeIndexPublicationStoreV1 { temporary.close().map_err(Self::unavailable)?; } Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - // The manifest is immutable once named. Failure cleanup never - // deletes a canonical name another scope may have attached. - temporary - .persist(&generation_path) - .map_err(|error| Self::unavailable(error.error))?; + // A concurrent publisher can win after the missing-path probe. + // Never replace its immutable generation, even transiently. + Self::publish_missing_generation_manifest(temporary, &generation_path)?; } Err(error) => return Err(Self::unavailable(error)), } @@ -3406,17 +4109,27 @@ impl DaemonCodeIndexPublicationStoreV1 { .ok_or_else(|| Self::unavailable("active code-generation pointer has no store root"))?; if self.active_path != preparation.active_path || !Arc::ptr_eq(&self.cache, &preparation.cache) - || Handle::from_path(store_root).map_err(Self::unavailable)? - != preparation.store_identity - || Handle::from_path(&self.segments_root).map_err(Self::unavailable)? - != preparation.segments_identity - || Handle::from_path(&self.generations_root).map_err(Self::unavailable)? - != preparation.generations_identity || self.disposition != preparation.disposition || CodeIndexGenerationScopeV1::for_snapshot(generation.snapshot()) != preparation.scope { return Err(CodeIndexPublicationStoreErrorV1::CompareAndSwap); } + #[cfg(unix)] + let same_directories = UnixDirectoryIdentityV1::at(store_root)? + == preparation.store_identity + && UnixDirectoryIdentityV1::at(&self.segments_root)? == preparation.segments_identity + && UnixDirectoryIdentityV1::at(&self.generations_root)? + == preparation.generations_identity; + #[cfg(not(unix))] + let same_directories = Handle::from_path(store_root).map_err(Self::unavailable)? + == preparation.store_identity + && Handle::from_path(&self.segments_root).map_err(Self::unavailable)? + == preparation.segments_identity + && Handle::from_path(&self.generations_root).map_err(Self::unavailable)? + == preparation.generations_identity; + if !same_directories { + return Err(CodeIndexPublicationStoreErrorV1::CompareAndSwap); + } Ok(()) } @@ -3453,10 +4166,22 @@ impl DaemonCodeIndexPublicationStoreV1 { Self::unavailable(error) } })?; - if !metadata.file_type().is_file() - || metadata.len() != identity.size_bytes - || Self::state_digest_file(path)? != identity.digest.as_str() - { + if !metadata.file_type().is_file() || metadata.len() != identity.size_bytes { + return Err(Self::corruption( + "prepared generation segment does not match its manifest identity", + )); + } + #[cfg(unix)] + if let Some(original) = files.segments.authenticated_segments.get(path) { + if !Self::same_unix_segment_identity(original, &metadata) { + return Err(Self::corruption( + "previously verified sealed segment changed before commit", + )); + } + verified.insert(identity.digest, identity.size_bytes); + continue; + } + if Self::state_digest_file(path)? != identity.digest.as_str() { return Err(Self::corruption( "prepared generation segment does not match its manifest identity", )); diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/staged_segment_reads.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/staged_segment_reads.rs new file mode 100644 index 0000000000..a890952550 --- /dev/null +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/staged_segment_reads.rs @@ -0,0 +1,1704 @@ +//! Exercise the production preparation, filesystem writer, and reader together. +use std::{ + fs::File, + io::Write, + os::unix::fs::MetadataExt, + path::{Path, PathBuf}, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use sha2::{Digest, Sha256}; +use tempfile::TempDir; +use tracedecay_code_index::production::{ + CodeIndexAtomicPublicationPort, CodeIndexExecutionControlV1, CodeIndexInterruptionV1, + CodeIndexPreparationPort, CodeIndexProductionErrorV1, CodeIndexPublicationStoreErrorV1, + CodeIndexPublishedGenerationV1, SealedGenerationSegmentPublicationV1, + SealedGenerationSegmentReadV1, +}; +use tracedecay_code_index_retention::code_index_generations::DurableSealedCodeGenerationIdentityV1; +use tracedecay_domain::{ManifestDigest, ProjectId, SanitizerRevision, sha256_hex_suffix}; + +use super::{ + DaemonCodeIndexPreparationStateV1, DaemonCodeIndexPreparationV1, + DaemonCodeIndexPublicationStoreV1, MAX_STAGED_SEGMENT_READ_HANDLES, + PREPARED_SEGMENT_DIGEST_OPENS, SharedCodeIndexBytePoolV1, +}; +use crate::code_index_scheduler::tests::GitFixture; +use crate::code_index_scheduler::{CodeIndexReconcileOutcomeV1, CodeIndexWorktreeSchedulerV1}; + +fn source_generation( + fixture: &GitFixture, + store_root: &Path, +) -> Arc { + let mut source = CodeIndexWorktreeSchedulerV1::open( + ProjectId::new("project.code-index-tests").expect("project identity"), + fixture.path(), + store_root.to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), + ) + .expect("source scheduler"); + assert!(matches!( + source.reconcile_now().expect("source publication"), + CodeIndexReconcileOutcomeV1::Published(_) + )); + Arc::clone(&source.latest_complete().expect("decode source").generation) +} + +fn target_store(root: &Path, fixture: &GitFixture) -> DaemonCodeIndexPublicationStoreV1 { + DaemonCodeIndexPublicationStoreV1::new( + root, + fixture.path(), + SanitizerRevision::new(tracedecay_privacy::CODE_SOURCE_SANITIZER_VERSION_V1) + .expect("sanitizer revision"), + ) + .expect("target publication store") +} + +fn segment_digest(bytes: &[u8]) -> ManifestDigest { + ManifestDigest::from_sha256_bytes(&Sha256::digest(bytes)).expect("canonical digest") +} + +fn staged_path(preparation: &DaemonCodeIndexPreparationV1, digest: &ManifestDigest) -> PathBuf { + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must still be open"); + }; + let suffix = sha256_hex_suffix(digest.as_str()).expect("digest suffix"); + let canonical = preparation + .segments_root + .join(format!("segment-{suffix}.json")); + files + .segments + .pending + .get(&canonical) + .expect("segment was staged, not published") + .clone() +} + +fn pinned_handle_count(preparation: &DaemonCodeIndexPreparationV1) -> usize { + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must still be open"); + }; + files.pinned_segment_reads.len() +} + +#[test] +fn staged_segment_range_reads_open_once_per_cached_file_and_revalidate_replacements() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn fixture() -> u32 { 7 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let contents = (0..9) + .map(|index| format!("sealed-staged-segment-{index}-bytes").into_bytes()) + .collect::>(); + let digests = contents + .iter() + .map(|bytes| segment_digest(bytes)) + .collect::>(); + for (digest, bytes) in digests.iter().zip(&contents) { + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { digest, bytes }) + .expect("write exact staged segment"); + } + let reader = preparation.segment_reader(); + let mut read = Vec::new(); + for offset in 0..16 { + reader( + SealedGenerationSegmentReadV1::Range { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + offset, + length: 4, + }, + &mut read, + ) + .expect("read staged segment range"); + assert_eq!(&read, &contents[0][offset as usize..offset as usize + 4]); + } + assert_eq!( + preparation.staged_segment_read_open_attempts_for_test(), + 1, + "16 overlapping production reads must open the first file once", + ); + let invalid_range = reader( + SealedGenerationSegmentReadV1::Range { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + offset: contents[0].len() as u64 - 1, + length: 2, + }, + &mut read, + ); + assert!(matches!( + invalid_range, + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert!(read.is_empty(), "a bad range must not expose stale bytes"); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 1); + + // Alternating ranges exercise LRU promotion, not just consecutive reads. + for offset in 0..16 { + for index in [0, 1] { + reader( + SealedGenerationSegmentReadV1::Range { + digest: &digests[index], + size_bytes: contents[index].len() as u64, + offset, + length: 4, + }, + &mut read, + ) + .expect("read interleaved staged ranges"); + assert_eq!( + &read, + &contents[index][offset as usize..offset as usize + 4] + ); + } + } + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 2); + for (digest, bytes) in digests.iter().zip(&contents).skip(1) { + reader( + SealedGenerationSegmentReadV1::Whole { + digest, + size_bytes: bytes.len() as u64, + }, + &mut read, + ) + .expect("read distinct staged segment"); + assert_eq!(&read, bytes); + } + assert_eq!( + preparation.staged_segment_read_open_attempts_for_test(), + 9, + "nine distinct segments must each need a real file open", + ); + assert_eq!( + pinned_handle_count(&preparation), + MAX_STAGED_SEGMENT_READ_HANDLES + ); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + }, + &mut read, + ) + .expect("reopen the evicted first segment"); + assert_eq!(&read, &contents[0]); + assert_eq!( + preparation.staged_segment_read_open_attempts_for_test(), + 10, + "an eight-handle bound must evict and reopen the ninth distinct file", + ); + + let path = staged_path(&preparation, &digests[0]); + let replacement_path = path.with_extension("replacement.tmp"); + std::fs::write(&replacement_path, vec![b'Z'; contents[0].len()]) + .expect("write replacement bytes"); + std::fs::rename(&replacement_path, &path).expect("replace the pinned path"); + for _ in 0..2 { + read.extend_from_slice(b"stale buffer sentinel"); + let result = reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + }, + &mut read, + ); + assert!(matches!( + result, + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert!(read.is_empty(), "replacement and retry must clear output"); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 10); + } + assert_eq!(pinned_handle_count(&preparation), 0); + assert!(matches!( + preparation.write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &digests[1], + bytes: &contents[1], + }), + Err(CodeIndexPublicationStoreErrorV1::Unavailable(_)) + )); + preparation.abort().expect("abort and clean staged files"); + assert!(!path.exists(), "abort removes replaced staged files"); + // An escaped reader must not keep the preparation or its staged bytes live. + read.extend_from_slice(b"stale after abort"); + assert!(matches!( + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + }, + &mut read, + ), + Err(CodeIndexProductionErrorV1::Publication(_)) + )); + assert_eq!(read, [] as [u8; 0]); +} + +#[test] +fn staged_segment_in_place_mutation_poison_survives_restoration_and_refuses_commit() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage a real valid publication"); + + // This auxiliary segment is deliberately outside the valid manifest. + // Commit must refuse the poisoned transaction, not just a digest mismatch + // discovered while verifying a segment the manifest happens to reference. + let bytes = b"original-staged-segment"; + let digest = segment_digest(bytes); + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &digest, + bytes, + }) + .expect("write auxiliary staged segment"); + let opens_before = preparation.staged_segment_read_open_attempts_for_test(); + let reader = preparation.segment_reader(); + let mut read = Vec::new(); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digest, + size_bytes: bytes.len() as u64, + }, + &mut read, + ) + .expect("pin original staged bytes"); + assert_eq!(read.as_slice(), bytes); + assert_eq!( + preparation.staged_segment_read_open_attempts_for_test(), + opens_before + 1 + ); + let path = staged_path(&preparation, &digest); + let before = path.symlink_metadata().expect("original metadata"); + let mut writer = File::options() + .write(true) + .open(&path) + .expect("same inode writer"); + writer + .write_all(&vec![b'X'; bytes.len()]) + .expect("same-length overwrite"); + // Force a visible timestamp change even on coarse-resolution filesystems. + writer + .set_modified(before.modified().expect("mtime") + Duration::from_secs(2)) + .expect("change mtime"); + drop(writer); + let after = path.symlink_metadata().expect("mutated metadata"); + assert_eq!( + (before.dev(), before.ino(), before.len()), + (after.dev(), after.ino(), after.len()) + ); + assert_ne!( + before.modified().expect("old mtime"), + after.modified().expect("new mtime") + ); + let result = reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digest, + size_bytes: bytes.len() as u64, + }, + &mut read, + ); + assert!(matches!( + result, + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert!(read.is_empty(), "mutated bytes must never escape"); + assert_eq!(pinned_handle_count(&preparation), 0); + + // Restoration cannot unpoison an already-invalidated transaction. + std::fs::write(&path, bytes).expect("restore original bytes"); + read.extend_from_slice(b"stale after mutation"); + assert!(matches!( + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digest, + size_bytes: bytes.len() as u64, + }, + &mut read, + ), + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert_eq!(read, [] as [u8; 0]); + assert_eq!( + preparation.staged_segment_read_open_attempts_for_test(), + opens_before + 1 + ); + assert!(matches!( + preparation.write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &digest, + bytes, + }), + Err(CodeIndexPublicationStoreErrorV1::Unavailable(detail)) + if detail.contains("identity was invalidated") + )); + assert!(matches!( + target.commit(&mut preparation, &publication), + Err(CodeIndexPublicationStoreErrorV1::Unavailable(detail)) + if detail.contains("identity was invalidated") + )); + assert!( + target + .load_active_manifest() + .expect("active manifest probe") + .is_none() + ); + preparation.abort().expect("abort poisoned preparation"); + assert!(!path.exists()); +} + +#[test] +fn staged_segment_cold_refusals_clear_output_without_poisoning_valid_reads() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let first = b"first staged bytes"; + let cold = b"cold staged bytes"; + let first_digest = segment_digest(first); + let cold_digest = segment_digest(cold); + for (digest, bytes) in [ + (&first_digest, first.as_slice()), + (&cold_digest, cold.as_slice()), + ] { + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { digest, bytes }) + .expect("stage segment"); + } + let missing_digest = segment_digest(b"never staged"); + let reader = preparation.segment_reader(); + let mut read = Vec::new(); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &first_digest, + size_bytes: first.len() as u64, + }, + &mut read, + ) + .expect("prime buffer and first handle"); + assert_eq!(read.as_slice(), first); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 1); + for request in [ + SealedGenerationSegmentReadV1::Range { + digest: &cold_digest, + size_bytes: cold.len() as u64, + offset: cold.len() as u64 - 1, + length: 2, + }, + SealedGenerationSegmentReadV1::Whole { + digest: &cold_digest, + size_bytes: cold.len() as u64 + 1, + }, + SealedGenerationSegmentReadV1::Whole { + digest: &missing_digest, + size_bytes: 12, + }, + ] { + read.extend_from_slice(b"stale cold-refusal sentinel"); + assert!(matches!( + reader(request, &mut read), + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert!( + read.is_empty(), + "even a refusal before File::open must clear output" + ); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 1); + } + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &first_digest, + size_bytes: first.len() as u64, + }, + &mut read, + ) + .expect("invalid requests must not poison an unchanged preparation"); + assert_eq!(read.as_slice(), first); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 1); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &cold_digest, + size_bytes: cold.len() as u64, + }, + &mut read, + ) + .expect("valid cold request still opens its file"); + assert_eq!(read.as_slice(), cold); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 2); + preparation.abort().expect("clean up preparation"); +} + +#[test] +fn staged_segment_evicted_path_replacement_poison_without_reopen() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let contents = (0..9) + .map(|index| format!("eviction-identity-segment-{index}").into_bytes()) + .collect::>(); + let digests = contents + .iter() + .map(|bytes| segment_digest(bytes)) + .collect::>(); + for (digest, bytes) in digests.iter().zip(&contents) { + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { digest, bytes }) + .expect("stage sealed segment"); + } + let first_path = staged_path(&preparation, &digests[0]); + let reader = preparation.segment_reader(); + let mut read = Vec::new(); + for (digest, bytes) in digests.iter().zip(&contents) { + reader( + SealedGenerationSegmentReadV1::Whole { + digest, + size_bytes: bytes.len() as u64, + }, + &mut read, + ) + .expect("pin distinct segment"); + assert_eq!(&read, bytes); + } + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 9); + assert_eq!( + pinned_handle_count(&preparation), + MAX_STAGED_SEGMENT_READ_HANDLES + ); + + let replacement_path = first_path.with_extension("eviction-replacement.tmp"); + std::fs::write(&replacement_path, vec![b'Z'; contents[0].len()]) + .expect("same-length replacement"); + std::fs::rename(&replacement_path, &first_path).expect("replace evicted segment pathname"); + read.extend_from_slice(b"old data"); + assert!(matches!( + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + }, + &mut read, + ), + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert!( + read.is_empty(), + "replacement cannot expose stale or new bytes" + ); + assert_eq!( + preparation.staged_segment_read_open_attempts_for_test(), + 9, + "replaced evicted path must be rejected before another File::open" + ); + assert_eq!(pinned_handle_count(&preparation), 0); + assert!(matches!( + preparation.write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &digests[1], + bytes: &contents[1], + }), + Err(CodeIndexPublicationStoreErrorV1::Unavailable(_)) + )); + preparation.abort().expect("remove poisoned staging"); + assert!(!first_path.exists()); +} + +#[test] +fn staged_preparation_exact_segment_bytes_commit_and_reload() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage real generation"); + let identities = CodeIndexPublishedGenerationV1::partitioned_segment_identities_from_reader( + publication.manifest_bytes().as_ref(), + ) + .expect("parse staged segment identities") + .expect("current partitioned generation revision"); + let identity = identities + .into_iter() + .find(|identity| identity.size_bytes > 4) + .expect("real generation must have a nonempty segment"); + + let reader = preparation.segment_reader(); + let initial_opens = preparation.staged_segment_read_open_attempts_for_test(); + let mut exact = Vec::new(); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &identity.digest, + size_bytes: identity.size_bytes, + }, + &mut exact, + ) + .expect("read staged sealed whole segment"); + assert_eq!( + segment_digest(&exact), + identity.digest, + "production reader must return the exact addressed segment" + ); + let mut range = Vec::new(); + for _ in 0..16 { + reader( + SealedGenerationSegmentReadV1::Range { + digest: &identity.digest, + size_bytes: identity.size_bytes, + offset: 1, + length: 3, + }, + &mut range, + ) + .expect("read repeated byte-exact range"); + assert_eq!(range.as_slice(), &exact[1..4]); + } + assert!( + preparation.staged_segment_read_open_attempts_for_test() - initial_opens <= 1, + "whole and repeated range reads may open the segment only once" + ); + let committed = target + .commit(&mut preparation, &publication) + .expect("publish the real staged generation"); + assert_eq!( + committed.as_ref(), + publication.manifest_bytes().as_ref(), + "commit must preserve the staged manifest bytes" + ); + drop(reader); + drop(preparation); + // A new store has no decoded-generation or active-manifest memo from the + // publisher; this must really reconstruct the snapshot from durable bytes. + drop(target); + let cold_store = target_store(target_root.path(), &fixture); + let reloaded = cold_store + .load_active_shared() + .expect("decode committed generation") + .expect("active durable generation"); + assert_eq!( + reloaded.manifest().generation_id, + generation.manifest().generation_id + ); + assert_eq!( + reloaded.snapshot().content_identity, + generation.snapshot().content_identity, + "a cold durable read must recover the exact published snapshot identity" + ); +} + +#[test] +fn staged_writer_authentication_avoids_commit_reopens_and_rejects_replaced_inode() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage real generation"); + let identities = CodeIndexPublishedGenerationV1::partitioned_segment_identities_from_reader( + publication.manifest_bytes().as_ref(), + ) + .expect("parse staged segment identities") + .expect("partitioned generation"); + let identity = identities + .into_iter() + .find(|identity| identity.size_bytes > 0) + .expect("nonempty generation segment"); + let path = staged_path(&preparation, &identity.digest); + let bytes = std::fs::read(&path).expect("real staged segment"); + assert!( + !bytes.is_empty(), + "the test must authenticate real segment bytes" + ); + assert_eq!(segment_digest(&bytes), identity.digest); + let before = PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get); + { + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must remain open"); + }; + assert!( + !files.segments.pending.is_empty(), + "real files must be staged" + ); + assert!( + files.segments.authenticated_segments.contains_key(&path), + "new staged bytes must be verified through their original writer" + ); + target + .verify_prepared_segments(files, publication.manifest_bytes().as_ref()) + .expect("authenticate exact pending generation without reopening"); + } + assert_eq!( + PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get), + before, + "commit verification must not reopen authenticated staged segments" + ); + + // Digest equality alone cannot authorize a replacement of the original + // writer's inode, even when the replacement has byte-identical content. + let replacement = path.with_extension("replacement.tmp"); + std::fs::write(&replacement, &bytes).expect("make equal-byte replacement"); + std::fs::rename(&replacement, &path).expect("replace staged inode"); + let before_refusal = PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get); + { + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must remain open"); + }; + assert!( + target + .verify_prepared_segments(files, publication.manifest_bytes().as_ref()) + .is_err(), + "replacing a staged inode must fail even if its bytes are identical" + ); + } + assert_eq!( + PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get), + before_refusal, + "identity rejection must not reopen a substituted staged inode" + ); + preparation + .abort() + .expect("abort replaced staged generation"); +} + +#[test] +fn committed_whole_segment_read_refuses_same_length_corruption() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage real generation"); + target + .commit(&mut preparation, &publication) + .expect("commit real generation"); + drop(preparation); + drop(target); + + let cold = target_store(target_root.path(), &fixture); + let identity = CodeIndexPublishedGenerationV1::partitioned_segment_identities_from_reader( + publication.manifest_bytes().as_ref(), + ) + .expect("parse real manifest") + .expect("partitioned generation") + .into_iter() + .find(|segment| segment.size_bytes > 4) + .expect("nonempty addressed segment"); + let request = SealedGenerationSegmentReadV1::Whole { + digest: &identity.digest, + size_bytes: identity.size_bytes, + }; + let mut bytes = Vec::new(); + cold.read_partitioned_segment(request, &mut bytes) + .expect("cold valid whole-segment read"); + assert_eq!(segment_digest(&bytes), identity.digest); + let mut range = Vec::new(); + cold.read_partitioned_segment( + SealedGenerationSegmentReadV1::Range { + digest: &identity.digest, + size_bytes: identity.size_bytes, + offset: 1, + length: 3, + }, + &mut range, + ) + .expect("valid range still returns exact bytes"); + assert_eq!(range.as_slice(), &bytes[1..4]); + + // Keep the inode and extent: size/path checks alone accept this corruption. + let segment_path = cold.segments_root.join(format!( + "segment-{}.json", + sha256_hex_suffix(identity.digest.as_str()).expect("digest suffix") + )); + let original = segment_path.symlink_metadata().expect("original segment"); + let mut corrupt = vec![b'X'; bytes.len()]; + if corrupt == bytes { + corrupt[0] ^= 1; + } + std::fs::write(&segment_path, &corrupt).expect("mutate same inode"); + let changed = segment_path.symlink_metadata().expect("mutated segment"); + assert_eq!( + (original.dev(), original.ino(), original.len()), + (changed.dev(), changed.ino(), changed.len()) + ); + let mut reused = b"stale output".to_vec(); + assert!(matches!( + cold.read_partitioned_segment(request, &mut reused), + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert_eq!(reused, [] as [u8; 0], "corrupt bytes must not escape"); +} + +#[test] +fn staged_segment_rejected_reads_preserve_full_cache_and_recency() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let contents = (0..9) + .map(|index| format!("admission-staged-segment-{index}-bytes").into_bytes()) + .collect::>(); + let digests = contents + .iter() + .map(|bytes| segment_digest(bytes)) + .collect::>(); + for (digest, bytes) in digests.iter().zip(&contents) { + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { digest, bytes }) + .expect("stage real segment"); + } + let reader = preparation.segment_reader(); + let mut read = Vec::new(); + for (digest, bytes) in digests.iter().zip(&contents).take(8) { + reader( + SealedGenerationSegmentReadV1::Whole { + digest, + size_bytes: bytes.len() as u64, + }, + &mut read, + ) + .expect("fill the handle cache"); + assert_eq!(&read, bytes); + } + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 8); + assert_eq!(pinned_handle_count(&preparation), 8); + let missing_digest = segment_digest(b"never staged admission"); + for request in [ + SealedGenerationSegmentReadV1::Range { + digest: &digests[8], + size_bytes: contents[8].len() as u64, + offset: contents[8].len() as u64 - 1, + length: 2, + }, + SealedGenerationSegmentReadV1::Range { + digest: &digests[8], + size_bytes: contents[8].len() as u64, + offset: u64::MAX, + length: 2, + }, + SealedGenerationSegmentReadV1::Whole { + digest: &digests[8], + size_bytes: contents[8].len() as u64 + 1, + }, + SealedGenerationSegmentReadV1::Whole { + digest: &missing_digest, + size_bytes: b"never staged admission".len() as u64, + }, + SealedGenerationSegmentReadV1::Range { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + offset: contents[0].len() as u64 - 1, + length: 2, + }, + SealedGenerationSegmentReadV1::Whole { + digest: &digests[0], + size_bytes: contents[0].len() as u64 + 1, + }, + ] { + read.extend_from_slice(b"stale rejected-read sentinel"); + assert!(matches!( + reader(request, &mut read), + Err(CodeIndexProductionErrorV1::Contract(_)) + )); + assert_eq!(read, [] as [u8; 0]); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 8); + assert_eq!( + pinned_handle_count(&preparation), + 8, + "a rejected request must not evict a valid handle" + ); + } + + // The next valid cold read must evict zero, not one: invalid accesses + // to zero above must not make it more recent than the other seven files. + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digests[8], + size_bytes: contents[8].len() as u64, + }, + &mut read, + ) + .expect("admit valid cold read after refusals"); + assert_eq!(read, contents[8]); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 9); + assert_eq!(pinned_handle_count(&preparation), 8); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digests[1], + size_bytes: contents[1].len() as u64, + }, + &mut read, + ) + .expect("second-oldest handle must still be cached"); + assert_eq!(read, contents[1]); + assert_eq!( + preparation.staged_segment_read_open_attempts_for_test(), + 9, + "invalid cache hits must not change the eviction victim" + ); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &digests[0], + size_bytes: contents[0].len() as u64, + }, + &mut read, + ) + .expect("reopen the actual least-recently-used segment"); + assert_eq!(read, contents[0]); + assert_eq!(preparation.staged_segment_read_open_attempts_for_test(), 10); + preparation.abort().expect("clean up staged cache fixture"); +} + +#[test] +fn staged_segment_publish_never_overwrites_a_concurrent_canonical_name() { + let root = TempDir::new().expect("sealed segment race fixture"); + let canonical = root.path().join("segment-content-address.json"); + let temporary = root.path().join("staged-segment.json"); + let original = b"sealed-original"; + let changed = b"sealed-modified"; + assert_eq!(original.len(), changed.len()); + + // Exercise the actual production helper after the missing-path probe, + // with a conflicting canonical name created in the race window. + std::fs::write(&canonical, original).expect("competing canonical bytes"); + std::fs::write(&temporary, changed).expect("staged bytes"); + assert!( + super::StagedGenerationSegmentsV1::publish_missing_immutable_file(&temporary, &canonical) + .is_err(), + "a race must refuse, not replace a different immutable segment" + ); + assert_eq!( + std::fs::read(&canonical).expect("preserved competing segment"), + original + ); + assert_eq!( + std::fs::read(&temporary).expect("staged bytes retained on refusal"), + changed + ); + + // When the competing publisher installed identical bytes, deduplication + // may finish without another publication or destroying canonical bytes. + std::fs::write(&temporary, original).expect("identical staged bytes"); + super::StagedGenerationSegmentsV1::publish_missing_immutable_file(&temporary, &canonical) + .expect("identical concurrent publication is safe to deduplicate"); + assert!(!temporary.exists()); + assert_eq!( + std::fs::read(&canonical).expect("canonical name retained"), + original + ); + + std::fs::remove_file(&canonical).expect("vacate canonical name"); + std::fs::write(&temporary, changed).expect("new staged segment"); + super::StagedGenerationSegmentsV1::publish_missing_immutable_file(&temporary, &canonical) + .expect("missing canonical name accepts one atomic publication"); + assert!(!temporary.exists()); + assert_eq!( + std::fs::read(&canonical).expect("new published bytes"), + changed + ); +} + +#[test] +fn repeated_pending_segment_accepts_identical_concurrent_canonical_publication() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn fixture() -> u32 { 7 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage real generation"); + + let (canonical, temporary) = { + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must remain open"); + }; + files + .segments + .pending + .iter() + .next() + .map(|(name, temporary)| (name.clone(), temporary.clone())) + .expect("a real generation must stage at least one segment") + }; + let bytes = std::fs::read(&temporary).expect("staged segment bytes"); + let digest = segment_digest(&bytes); + assert!( + !canonical.exists(), + "the first stage has not named its segment" + ); + + // A separate publisher wins the canonical name after this preparation + // writes its own temporary file. It has the same authenticated bytes but + // a different inode. Repeating the write must not confuse their identities. + let mut foreign = std::fs::OpenOptions::new() + .create_new(true) + .write(true) + .open(&canonical) + .expect("concurrent canonical publication"); + foreign.write_all(&bytes).expect("canonical bytes"); + foreign.sync_all().expect("durable competing segment"); + let foreign_identity = foreign.metadata().expect("canonical inode"); + drop(foreign); + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &digest, + bytes: &bytes, + }) + .expect("repeat an already-pending segment after concurrent naming"); + { + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must remain open"); + }; + assert_eq!( + files.segments.pending.get(&canonical), + Some(&temporary), + "this preparation still owns its original temporary bytes" + ); + assert!( + !files + .segments + .authenticated_segments + .contains_key(&canonical), + "the concurrently named inode must not masquerade as our pending inode" + ); + target + .verify_prepared_segments(files, publication.manifest_bytes().as_ref()) + .expect("verification authenticates pending bytes instead of another inode"); + } + + target + .commit(&mut preparation, &publication) + .expect("identical canonical bytes deduplicate at commit"); + assert!(!temporary.exists(), "deduplicated temporary is discharged"); + assert_eq!( + std::fs::read(&canonical).expect("canonical bytes after commit"), + bytes + ); + let after = canonical.metadata().expect("canonical inode after commit"); + assert_eq!(after.dev(), foreign_identity.dev()); + assert_eq!(after.ino(), foreign_identity.ino()); +} + +#[cfg(unix)] +#[test] +fn prepared_generation_rechecks_directory_identity_without_reopening_paths() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage generation"); + + target + .validate_preparation_store(&preparation, &publication) + .expect("unchanged directories must retain preparation identity"); + let canonical = target.segments_root.clone(); + let moved = target_root.path().join("detached-segments"); + std::fs::rename(&canonical, &moved).expect("move original directory inode"); + std::fs::create_dir(&canonical).expect("substitute another directory"); + assert!(matches!( + target.validate_preparation_store(&preparation, &publication), + Err(CodeIndexPublicationStoreErrorV1::CompareAndSwap) + )); + std::fs::remove_dir(&canonical).expect("remove empty substituted directory"); + std::fs::rename(&moved, &canonical).expect("restore owned original inode"); + target + .validate_preparation_store(&preparation, &publication) + .expect("restoring original inode restores the exact identity"); + preparation.abort().expect("release owned preparation"); +} + +#[test] +fn generation_manifest_publication_preserves_concurrent_canonical_bytes() { + let root = TempDir::new().expect("generation publication fixture"); + let good = b"generation-one"; + let other = b"generation-two"; + assert_eq!(good.len(), other.len()); + let digest = segment_digest(good); + let suffix = sha256_hex_suffix(digest.as_str()).expect("generation suffix"); + let canonical = root.path().join(format!("generation-{suffix}.json")); + let stage = |bytes: &[u8]| { + let mut temporary = tempfile::Builder::new() + .prefix(".generation-publication-") + .tempfile_in(root.path()) + .expect("stage generation"); + temporary.write_all(bytes).expect("write staged manifest"); + temporary + .as_file() + .sync_all() + .expect("durable staged manifest"); + temporary + }; + + // The competing pathname appears after the publisher's missing-path + // observation. A same-length conflict must not replace it. + std::fs::write(&canonical, good).expect("competing generation"); + assert!( + DaemonCodeIndexPublicationStoreV1::publish_missing_generation_manifest( + stage(other), + &canonical, + ) + .is_err(), + "conflicting concurrent generation must refuse publication" + ); + assert_eq!( + std::fs::read(&canonical).expect("competing generation retained"), + good + ); + assert_eq!( + std::fs::read_dir(root.path()) + .expect("generation directory") + .count(), + 1, + "refused publication must clean its staging file" + ); + + // A second publisher with identical content may safely deduplicate. + DaemonCodeIndexPublicationStoreV1::publish_missing_generation_manifest(stage(good), &canonical) + .expect("identical immutable generation deduplicates"); + assert_eq!(std::fs::read(&canonical).expect("original content"), good); + assert_eq!( + std::fs::read_dir(root.path()) + .expect("generation directory") + .count(), + 1, + "deduplicated publication must not retain staging" + ); + + std::fs::remove_file(&canonical).expect("vacate immutable name"); + DaemonCodeIndexPublicationStoreV1::publish_missing_generation_manifest(stage(good), &canonical) + .expect("publish new immutable name"); + assert_eq!( + std::fs::read(&canonical).expect("new generation bytes"), + good + ); + assert_eq!( + std::fs::read_dir(root.path()) + .expect("generation directory") + .count(), + 1, + "successful publication must consume staging" + ); +} + +struct InterruptOnCheckpoint { + interrupt_at: usize, + deadline: bool, + checks: AtomicUsize, +} + +impl InterruptOnCheckpoint { + fn new(interrupt_at: usize, deadline: bool) -> Self { + Self { + interrupt_at, + deadline, + checks: AtomicUsize::new(0), + } + } +} + +impl CodeIndexExecutionControlV1 for InterruptOnCheckpoint { + fn is_cancelled(&self) -> bool { + let check = self.checks.fetch_add(1, Ordering::SeqCst) + 1; + !self.deadline && check >= self.interrupt_at + } + + fn is_deadline_exceeded(&self) -> bool { + let check = self.checks.fetch_add(1, Ordering::SeqCst) + 1; + self.deadline && check >= self.interrupt_at + } +} + +#[test] +fn retained_generation_read_refuses_pre_and_post_io_interruptions_without_leaking_bytes() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage real generation"); + target + .commit(&mut preparation, &publication) + .expect("publish real generation"); + drop(preparation); + + let pointer = target + .read_publication_pointer() + .expect("read durable pointer") + .expect("active generation pointer"); + let active = pointer + .generation_index + .iter() + .find(|entry| entry.generation_id == pointer.generation_id) + .expect("active generation entry"); + let sealed = DurableSealedCodeGenerationIdentityV1 { + locator: active.generation_file.clone(), + digest: ManifestDigest::new(active.state_digest.clone()).expect("canonical digest"), + size_bytes: active.size_bytes, + }; + let segment = CodeIndexPublishedGenerationV1::partitioned_segment_identities_from_reader( + publication.manifest_bytes().as_ref(), + ) + .expect("parse real manifest") + .expect("current generation revision") + .into_iter() + .find(|segment| segment.size_bytes > 0) + .expect("nonempty retained segment"); + let request = SealedGenerationSegmentReadV1::Whole { + digest: &segment.digest, + size_bytes: segment.size_bytes, + }; + let mut expected = Vec::new(); + target + .read_partitioned_segment(request, &mut expected) + .expect("read exact committed segment"); + assert_eq!(segment_digest(&expected), segment.digest); + + // A control refuses after the initial two checks, i.e. after the real + // pointer and segment reads. A stale successful result is not acceptable. + for (checkpoint, deadline) in [(1, false), (2, true), (3, false), (4, true)] { + let control = InterruptOnCheckpoint::new(checkpoint, deadline); + let mut bytes = b"previous successful read must not survive".to_vec(); + let error = target + .read_retained_partitioned_segment(&sealed, request, &mut bytes, &control) + .expect_err("interrupted retained read must refuse"); + if deadline { + assert!(matches!( + error, + CodeIndexProductionErrorV1::Interrupted(CodeIndexInterruptionV1::DeadlineExceeded) + )); + } else { + assert!(matches!( + error, + CodeIndexProductionErrorV1::Interrupted(CodeIndexInterruptionV1::Cancelled) + )); + } + assert_eq!(bytes, [] as [u8; 0], "no earlier or late bytes may escape"); + assert_eq!( + control.checks.load(Ordering::SeqCst), + checkpoint, + "interrupt must occur at the intended pre/post-I/O checkpoint" + ); + } + + let active = InterruptOnCheckpoint::new(usize::MAX, false); + let mut bytes = b"stale".to_vec(); + target + .read_retained_partitioned_segment(&sealed, request, &mut bytes, &active) + .expect("unchanged generation remains readable"); + assert_eq!(bytes, expected); +} + +#[test] +fn staged_existing_segments_reuse_authenticated_inodes_without_reopening() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut first = target + .prepare(&generation.sealed_scope(), None) + .expect("first preparation"); + let publication = generation + .stage_publication(&mut first) + .expect("stage real generation"); + target + .commit(&mut first, &publication) + .expect("publish canonical sealed segments"); + drop(first); + + // A manifest supplied without stage ownership must authenticate every + // canonical file, not trust its digest-shaped filename. + let unproven = target + .prepare( + &generation.sealed_scope(), + Some(&generation.manifest().generation_id), + ) + .expect("unproven preparation"); + let full_before = PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get); + { + let state = unproven.state.lock().expect("unproven state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("unproven preparation must remain open"); + }; + target + .verify_prepared_segments(files, publication.manifest_bytes().as_ref()) + .expect("authenticate independently supplied manifest"); + } + let full_opens = PREPARED_SEGMENT_DIGEST_OPENS + .with(std::cell::Cell::get) + .saturating_sub(full_before); + assert!( + full_opens > 0, + "the counter must observe canonical file opens" + ); + unproven.abort().expect("abort unproven preparation"); + + let mut staged = target + .prepare( + &generation.sealed_scope(), + Some(&generation.manifest().generation_id), + ) + .expect("new preparation"); + let staged_publication = generation + .stage_publication(&mut staged) + .expect("restage existing canonical bytes"); + let before = PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get); + let existing_path = { + let state = staged.state.lock().expect("staged state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("staged preparation must remain open"); + }; + let path = files + .segments + .authenticated_segments + .keys() + .next() + .expect("previously verified canonical segment") + .clone(); + target + .verify_prepared_segments(files, staged_publication.manifest_bytes().as_ref()) + .expect("reuse sealed inode authentication"); + path + }; + let warm_opens = PREPARED_SEGMENT_DIGEST_OPENS + .with(std::cell::Cell::get) + .saturating_sub(before); + assert!( + warm_opens < full_opens, + "staged existing segments must not be opened again during verification" + ); + + // Equal bytes do not authorize silently replacing an already witnessed + // immutable inode between stage and commit. + let replacement = existing_path.with_extension("witness-replacement.tmp"); + std::fs::write( + &replacement, + std::fs::read(&existing_path).expect("canonical bytes"), + ) + .expect("same-content replacement"); + std::fs::rename(&replacement, &existing_path).expect("replace canonical inode"); + let opens_before_refusal = PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get); + { + let state = staged.state.lock().expect("staged state after replacement"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("staged preparation must remain open"); + }; + assert!( + target + .verify_prepared_segments(files, staged_publication.manifest_bytes().as_ref()) + .is_err(), + "a substituted inode must be refused even with identical bytes" + ); + } + assert_eq!( + PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get), + opens_before_refusal, + "identity refusal must not reopen the replaced pathname" + ); + staged.abort().expect("abort changed preparation"); +} + +#[test] +fn repeated_canonical_segment_writes_reuse_witness_and_refuse_replacement() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut first = target + .prepare(&generation.sealed_scope(), None) + .expect("first preparation"); + let publication = generation + .stage_publication(&mut first) + .expect("stage canonical generation"); + target + .commit(&mut first, &publication) + .expect("publish canonical segments"); + drop(first); + + let identity = CodeIndexPublishedGenerationV1::partitioned_segment_identities_from_reader( + publication.manifest_bytes().as_ref(), + ) + .expect("read real segment identities") + .expect("current partitioned generation") + .into_iter() + .find(|identity| identity.size_bytes > 0) + .expect("nonempty canonical segment"); + let suffix = sha256_hex_suffix(identity.digest.as_str()).expect("canonical digest suffix"); + let canonical = target.segments_root.join(format!("segment-{suffix}.json")); + let bytes = std::fs::read(&canonical).expect("canonical segment"); + assert_eq!(segment_digest(&bytes), identity.digest); + + let mut preparation = target + .prepare( + &generation.sealed_scope(), + Some(&generation.manifest().generation_id), + ) + .expect("repeated-write preparation"); + let before = PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get); + for _ in 0..16 { + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &identity.digest, + bytes: &bytes, + }) + .expect("authenticate or reuse exact existing segment"); + assert_eq!( + PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get) - before, + 1, + "sixteen repeated writes must open and hash the canonical file only once" + ); + } + let reader = preparation.segment_reader(); + let mut read = Vec::new(); + reader( + SealedGenerationSegmentReadV1::Whole { + digest: &identity.digest, + size_bytes: identity.size_bytes, + }, + &mut read, + ) + .expect("read the exact staged canonical segment"); + assert_eq!(read, bytes); + + // A byte-identical inode replacement is not the file that was hashed. + // A subsequent write must reject it without another File::open. + let replacement = canonical.with_extension("repeat-witness.tmp"); + std::fs::write(&replacement, &bytes).expect("same-byte replacement"); + std::fs::rename(&replacement, &canonical).expect("replace canonical inode"); + assert!( + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &identity.digest, + bytes: &bytes, + }) + .is_err(), + "a replaced immutable inode must not inherit the prior witness" + ); + assert_eq!( + PREPARED_SEGMENT_DIGEST_OPENS.with(std::cell::Cell::get) - before, + 1, + "identity refusal should not reopen a substituted inode" + ); + preparation + .abort() + .expect("release repeated-write preparation"); +} + +#[cfg(target_os = "linux")] +#[test] +fn failed_durability_admission_cannot_leave_a_publishable_segment() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn fixture() -> u32 { 7 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut target = target_store(target_root.path(), &fixture); + let mut preparation = target + .prepare(&generation.sealed_scope(), None) + .expect("prepare target"); + + // The 33rd admission flushes the first 32 real writer handles on Linux. + // Mutating a member makes that barrier fail after the 33rd temporary was + // created, exposing the old bug that nevertheless enrolled the temporary. + let contents = (0..33) + .map(|index| format!("durability-admission-{index}-bytes").into_bytes()) + .collect::>(); + let digests = contents + .iter() + .map(|bytes| segment_digest(bytes)) + .collect::>(); + for (digest, bytes) in digests.iter().zip(&contents).take(32) { + preparation + .write_segment(SealedGenerationSegmentPublicationV1::File { digest, bytes }) + .expect("stage first durability window"); + } + + let first_path = staged_path(&preparation, &digests[0]); + let prior = first_path + .symlink_metadata() + .expect("first member metadata"); + let mut writer = File::options() + .write(true) + .open(&first_path) + .expect("mutate same inode"); + writer + .write_all(&vec![b'X'; contents[0].len()]) + .expect("same-length mutation"); + writer + .set_modified(prior.modified().expect("mtime") + Duration::from_secs(2)) + .expect("force distinct mtime"); + drop(writer); + + let pending_path = { + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must be open"); + }; + let suffix = sha256_hex_suffix(digests[32].as_str()).expect("digest suffix"); + files + .directory + .path() + .join(format!("segment-{suffix}.json")) + }; + for _ in 0..2 { + let result = preparation.write_segment(SealedGenerationSegmentPublicationV1::File { + digest: &digests[32], + bytes: &contents[32], + }); + assert!(matches!( + result, + Err(CodeIndexPublicationStoreErrorV1::Unavailable(_)) + )); + assert!( + !pending_path.exists(), + "a failed writer must clean its temporary, including on retry" + ); + let state = preparation.state.lock().expect("preparation state"); + let DaemonCodeIndexPreparationStateV1::Open(files) = &*state else { + panic!("preparation must remain abortable"); + }; + assert!( + !files + .segments + .pending + .values() + .any(|path| path == &pending_path), + "failed durable admission must not be treated as staged" + ); + assert_eq!( + files.segments.pending.len(), + 32, + "only the initially admitted members remain owned" + ); + } + preparation.abort().expect("abort failed durability window"); + assert!(!pending_path.exists()); +} + +#[cfg(target_os = "linux")] +fn observed_pointer_opens( + events: &std::sync::mpsc::Receiver>, +) -> usize { + let mut opens = 0; + // Inotify delivers access events asynchronously through notify. Drain to + // quiescence rather than observing an arbitrary instant after the call. + while let Ok(event) = events.recv_timeout(Duration::from_millis(250)) { + let event = event.expect("kernel file watch event"); + if matches!( + event.kind, + notify::EventKind::Access(notify::event::AccessKind::Open(_)) + ) { + opens += 1; + } + } + opens +} + +#[cfg(target_os = "linux")] +#[test] +fn cold_publication_pointer_validation_uses_one_kernel_open_and_recovers_snapshot() { + use notify::Watcher; + + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut publisher = target_store(target_root.path(), &fixture); + let mut preparation = publisher + .prepare(&generation.sealed_scope(), None) + .expect("prepare published generation"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage source snapshot"); + publisher + .commit(&mut preparation, &publication) + .expect("commit real generation"); + drop(preparation); + drop(publisher); + + let cold = target_store(target_root.path(), &fixture); + let (sender, receiver) = std::sync::mpsc::channel(); + let mut watcher = notify::recommended_watcher(move |event: notify::Result| { + let _ = sender.send(event); + }) + .expect("native file observer"); + watcher + .watch(&cold.active_path, notify::RecursiveMode::NonRecursive) + .expect("watch the exact publication pointer"); + drop(File::open(&cold.active_path).expect("nonzero control open")); + assert_eq!( + observed_pointer_opens(&receiver), + 1, + "the observer must report a real kernel pathname open" + ); + + let pointer = cold + .read_publication_pointer() + .expect("cold pointer validation") + .expect("published pointer"); + assert_eq!( + pointer.generation_id, + generation.manifest().generation_id.as_str() + ); + assert_eq!( + observed_pointer_opens(&receiver), + 1, + "parse and post-parse byte revalidation must share one file open" + ); + + let reloaded = cold + .load_active_shared() + .expect("read committed snapshot") + .expect("published generation"); + assert_eq!( + reloaded.snapshot().content_identity, + generation.snapshot().content_identity, + "one-open validation must not alter the durable snapshot" + ); +} + +#[cfg(target_os = "linux")] +fn observed_named_opens( + events: &std::sync::mpsc::Receiver>, + path: &Path, +) -> usize { + let mut opens = 0; + while let Ok(event) = events.recv_timeout(Duration::from_millis(250)) { + let event = event.expect("kernel file watch event"); + if event.paths.iter().any(|candidate| candidate == path) + && matches!( + event.kind, + notify::EventKind::Access(notify::event::AccessKind::Open(_)) + ) + { + opens += 1; + } + } + opens +} + +#[cfg(target_os = "linux")] +#[test] +fn initial_pointer_publication_reuses_persisted_descriptor_without_pathname_open() { + use notify::Watcher; + + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 2 }\n")]); + let source_root = TempDir::new().expect("source store"); + let generation = source_generation(&fixture, source_root.path()); + let target_root = TempDir::new().expect("target store"); + let mut publisher = target_store(target_root.path(), &fixture); + let active_path = publisher.active_path.clone(); + + // Watch the parent: the final pathname does not exist until the first + // publication rename, so it cannot be watched directly beforehand. + let (sender, receiver) = std::sync::mpsc::channel(); + let mut watcher = notify::recommended_watcher(move |event: notify::Result| { + let _ = sender.send(event); + }) + .expect("native file observer"); + watcher + .watch( + active_path.parent().expect("pointer parent"), + notify::RecursiveMode::NonRecursive, + ) + .expect("watch publication directory"); + + let mut preparation = publisher + .prepare(&generation.sealed_scope(), None) + .expect("prepare first generation"); + let publication = generation + .stage_publication(&mut preparation) + .expect("stage source snapshot"); + publisher + .commit(&mut preparation, &publication) + .expect("publish first generation"); + + assert!(active_path.is_file(), "the committed pointer must exist"); + assert_eq!( + observed_named_opens(&receiver, &active_path), + 0, + "memo authentication must reuse the persisted writer, not reopen the pointer" + ); + drop(preparation); + drop(publisher); + + // Non-vacuous kernel-observer control: the same directory watch must + // record a real pathname open after the publication. + drop(File::open(&active_path).expect("control open of persisted pointer")); + assert_eq!( + observed_named_opens(&receiver, &active_path), + 1, + "watcher must observe a real open of the committed pointer" + ); + + let cold = target_store(target_root.path(), &fixture); + let reloaded = cold + .load_active_shared() + .expect("load persisted generation") + .expect("published generation"); + assert_eq!( + reloaded.snapshot().content_identity, + generation.snapshot().content_identity, + "open-free memo population must preserve cold snapshot identity" + ); +} diff --git a/crates/tracedecay-private-fs/src/framed_log.rs b/crates/tracedecay-private-fs/src/framed_log.rs index 005e20779c..b9703c5fd6 100644 --- a/crates/tracedecay-private-fs/src/framed_log.rs +++ b/crates/tracedecay-private-fs/src/framed_log.rs @@ -4,6 +4,8 @@ use std::fs::{self, File, OpenOptions}; use std::io::{self, Read, Seek, SeekFrom, Write}; #[cfg(target_os = "linux")] use std::os::fd::AsRawFd; +#[cfg(target_os = "linux")] +use std::os::unix::fs::MetadataExt; #[cfg(unix)] use std::os::unix::fs::OpenOptionsExt; #[cfg(windows)] @@ -16,6 +18,10 @@ use windows_sys::Win32::Storage::FileSystem::{ FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, }; +#[cfg(all(test, target_os = "linux"))] +#[path = "framed_log_batch_tests.rs"] +mod batch_tests; + #[cfg(test)] thread_local! { static DIRECTORY_SYNC_CALLS: std::cell::Cell = const { std::cell::Cell::new(0) }; @@ -67,23 +73,74 @@ pub fn sync_parent_directory(path: &Path, policy: DirectorySyncPolicy) -> io::Re } } -/// New files whose contents become durable together, before any is renamed -/// into place. +/// New, fully written files whose contents become durable before any is renamed. +/// Members must remain immutable until publication. /// -/// On Linux each member's writeback starts as soon as it is written, and -/// [`Self::sync`] then fsyncs the members concurrently. Concurrent fsyncs -/// share a commit on a journaling filesystem and on ZFS's intent log alike, -/// but on ZFS one fsync after another pays one intent-log commit each, so a -/// seal of a hundred segments waited out a hundred commits. The cost stays -/// proportional to the batch's own bytes. `syncfs(2)` also paid one -/// commit, but waited for every dirty inode on the filesystem, so a one-file -/// seal stalled for seconds behind unrelated writers. Other platforms fsync -/// each file as it is written. +/// Linux retains at most one concurrency window of duplicated writer handles. +/// When that window fills, its members are fsynced concurrently before their +/// handles are released. This preserves grouped log commits without reopening +/// every pathname through endpoint security or holding a descriptor per segment. +/// [`Self::sync`] flushes the final window and checks that earlier members still +/// name the same immutable files. It never flushes unrelated filesystem data. +/// Other platforms fsync each file as it is written. #[derive(Default)] pub struct DurableFileBatch { - /// Paths rather than open handles, so a large seal holds no descriptors. #[cfg(target_os = "linux")] - members: Vec, + pending: Vec, + #[cfg(target_os = "linux")] + synced: Vec, +} + +#[cfg(target_os = "linux")] +struct PendingDurableFile { + identity: DurableFileIdentity, + file: File, +} + +#[cfg(target_os = "linux")] +struct DurableFileIdentity { + path: PathBuf, + metadata: fs::Metadata, +} + +#[cfg(target_os = "linux")] +impl DurableFileIdentity { + fn validate_metadata(&self, current: &fs::Metadata) -> io::Result<()> { + let original = &self.metadata; + if !original.file_type().is_file() + || !current.file_type().is_file() + || original.dev() != current.dev() + || original.ino() != current.ino() + || original.len() != current.len() + || original.mtime() != current.mtime() + || original.mtime_nsec() != current.mtime_nsec() + || original.ctime() != current.ctime() + || original.ctime_nsec() != current.ctime_nsec() + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "durable batch member changed after it was written", + )); + } + Ok(()) + } + + fn validate_path(&self) -> io::Result<()> { + self.validate_metadata(&fs::symlink_metadata(&self.path)?) + } +} + +#[cfg(target_os = "linux")] +impl PendingDurableFile { + fn sync(&self) -> io::Result<()> { + self.identity.validate_path()?; + self.identity.validate_metadata(&self.file.metadata()?)?; + sync_owned_file(&self.identity.path, &self.file)?; + // A replacement or in-place write during the barrier must not be + // reported as a successful flush of the originally admitted bytes. + self.identity.validate_metadata(&self.file.metadata()?)?; + self.identity.validate_path() + } } impl DurableFileBatch { @@ -93,25 +150,68 @@ impl DurableFileBatch { /// Record one fully written member file, created at `path`. pub fn written(&mut self, path: &Path, file: &File) -> io::Result<()> { + let observed = file.metadata()?; + self.written_with_identity(path, file, &observed) + } + + /// Record the exact inode and metadata that the caller already authenticated + /// through its writer. A changed member must fail before admission, not + /// leave a stale handle in a later durability window. + pub fn written_with_identity( + &mut self, + path: &Path, + file: &File, + authenticated: &fs::Metadata, + ) -> io::Result<()> { #[cfg(target_os = "linux")] { - // SAFETY: `file` is an open descriptor for the whole call. + let identity = DurableFileIdentity { + path: path.to_path_buf(), + metadata: authenticated.clone(), + }; + identity.validate_metadata(&file.metadata()?)?; + // An invalid incoming member must not force a full, otherwise + // healthy window to fsync before its admission is rejected. + identity.validate_path()?; + if self.pending.len() == DURABLE_BATCH_CONCURRENT_SYNCS { + // Keep ownership on failure: no unproven member may move to + // the synced set, and no additional descriptor is admitted. + self.sync_pending()?; + self.synced + .extend(self.pending.drain(..).map(|member| member.identity)); + // A slow filesystem barrier leaves time for the incoming + // path or descriptor to change. Do not admit that change. + identity.validate_metadata(&file.metadata()?)?; + identity.validate_path()?; + } + // Duplication retains the writer's inode without a pathname open. + let retained = file.try_clone()?; + // SAFETY: `retained` is an open descriptor for the whole call. let started = unsafe { - libc::sync_file_range(file.as_raw_fd(), 0, 0, libc::SYNC_FILE_RANGE_WRITE) + libc::sync_file_range(retained.as_raw_fd(), 0, 0, libc::SYNC_FILE_RANGE_WRITE) }; if started != 0 { return Err(io::Error::last_os_error()); } - self.members.push(path.to_path_buf()); + // This is the last fallible check before admission. There must + // be no caller-side validation that could delete a registered + // member after a successful return. + identity.validate_metadata(&retained.metadata()?)?; + identity.validate_path()?; + self.pending.push(PendingDurableFile { + identity, + file: retained, + }); Ok(()) } #[cfg(not(target_os = "linux"))] { + let _ = authenticated; sync_owned_file(path, file) } } - /// Make every member file's contents durable. + /// Make every member file's contents durable, refusing changed members. #[tracing::instrument( name = "private_fs.framed_log.sync_file_batch", level = "trace", @@ -120,33 +220,44 @@ impl DurableFileBatch { pub fn sync(&self) -> io::Result<()> { #[cfg(target_os = "linux")] { - let next = std::sync::atomic::AtomicUsize::new(0); - let syncer = || -> io::Result<()> { - while let Some(member) = self.members.get(next.fetch_add(1, Ordering::Relaxed)) { - sync_file_at(member)?; - } - Ok(()) - }; - let syncers = self.members.len().min(DURABLE_BATCH_CONCURRENT_SYNCS); - std::thread::scope(|scope| { - let helpers = (1..syncers) - .map(|_| scope.spawn(syncer)) - .collect::>(); - let mut outcome = syncer(); - for helper in helpers { - let helped = helper - .join() - .unwrap_or_else(|panic| std::panic::resume_unwind(panic)); - outcome = outcome.and(helped); - } - outcome - }) + // Already-flushed windows need no further opens or barriers, but + // missing or substituted names must still refuse publication. + for identity in &self.synced { + identity.validate_path()?; + } + self.sync_pending() } #[cfg(not(target_os = "linux"))] { Ok(()) } } + + #[cfg(target_os = "linux")] + fn sync_pending(&self) -> io::Result<()> { + let next = std::sync::atomic::AtomicUsize::new(0); + let syncer = || -> io::Result<()> { + while let Some(member) = self.pending.get(next.fetch_add(1, Ordering::Relaxed)) { + member.sync()?; + } + Ok(()) + }; + let syncers = self.pending.len().min(DURABLE_BATCH_CONCURRENT_SYNCS); + std::thread::scope(|scope| { + let helpers = (1..syncers) + .map(|_| scope.spawn(syncer)) + .collect::>(); + let mut outcome = syncer(); + for helper in helpers { + let helped = match helper.join() { + Ok(helped) => helped, + Err(panic) => std::panic::resume_unwind(panic), + }; + outcome = outcome.and(helped); + } + outcome + }) + } } /// Barriers one batch keeps in flight together: enough for the filesystem to diff --git a/crates/tracedecay-private-fs/src/framed_log_batch_tests.rs b/crates/tracedecay-private-fs/src/framed_log_batch_tests.rs new file mode 100644 index 0000000000..db6c058d73 --- /dev/null +++ b/crates/tracedecay-private-fs/src/framed_log_batch_tests.rs @@ -0,0 +1,380 @@ +//! Real filesystem evidence for bounded, descriptor-owned durability batches. +use std::ffi::CString; +use std::fs::{self, File}; +use std::io::{ErrorKind, Read, Write}; +use std::mem::size_of; +use std::os::fd::{AsRawFd, FromRawFd}; +use std::os::unix::ffi::OsStrExt; +use std::os::unix::fs::MetadataExt; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +#[cfg(feature = "test-helpers")] +use super::sync_latency; +use super::{DURABLE_BATCH_CONCURRENT_SYNCS, DurableFileBatch}; + +/// Count kernel open events, not calls to an observer unrelated to the opener. +struct OpenEvents(File); + +impl OpenEvents { + fn watch(root: &Path) -> Self { + // SAFETY: inotify_init1 has no pointer arguments. + let descriptor = unsafe { libc::inotify_init1(libc::IN_NONBLOCK | libc::IN_CLOEXEC) }; + assert!( + descriptor >= 0, + "inotify: {}", + std::io::Error::last_os_error() + ); + // SAFETY: the successful descriptor is owned exactly once by this File. + let file = unsafe { File::from_raw_fd(descriptor) }; + let path = CString::new(root.as_os_str().as_bytes()).expect("fixture path"); + // SAFETY: file is live and path is NUL-terminated for the whole call. + let watch = + unsafe { libc::inotify_add_watch(file.as_raw_fd(), path.as_ptr(), libc::IN_OPEN) }; + assert!(watch >= 0, "watch: {}", std::io::Error::last_os_error()); + Self(file) + } + + fn drain(&mut self) -> usize { + let mut opens = 0; + let mut buffer = [0_u8; 8192]; + loop { + let length = match self.0.read(&mut buffer) { + Ok(length) => length, + Err(error) if error.kind() == ErrorKind::WouldBlock => return opens, + Err(error) if error.kind() == ErrorKind::Interrupted => continue, + Err(error) => panic!("read inotify events: {error}"), + }; + assert!(length > 0, "inotify unexpectedly reached EOF"); + let mut offset = 0; + while offset < length { + assert!(offset + size_of::() <= length); + // SAFETY: the complete header is in bounds; read_unaligned + // handles the byte buffer's alignment. No borrowed name escapes. + let event = unsafe { + buffer + .as_ptr() + .add(offset) + .cast::() + .read_unaligned() + }; + assert_eq!(event.mask & libc::IN_Q_OVERFLOW, 0, "lost open events"); + if event.mask & libc::IN_OPEN != 0 { + opens += 1; + } + offset += size_of::() + + usize::try_from(event.len).expect("event length"); + assert!(offset <= length, "truncated inotify name"); + } + } + } +} + +fn write_member(batch: &mut DurableFileBatch, root: &Path, index: usize) -> PathBuf { + let path = root.join(format!("segment-{index}.tmp")); + let mut file = File::create(&path).expect("member writer"); + file.write_all(format!("sealed segment {index}").as_bytes()) + .expect("member bytes"); + batch.written(&path, &file).expect("record member"); + path +} + +#[test] +fn durable_batch_flushes_multiple_windows_without_pathname_reopens() { + let root = tempfile::tempdir().expect("batch root"); + let mut events = OpenEvents::watch(root.path()); + let probe = root.path().join("open-probe"); + drop(File::create(&probe).expect("real probe open")); + assert_eq!( + events.drain(), + 1, + "the kernel observer must actually see opens" + ); + fs::remove_file(&probe).expect("remove probe"); + + let mut batch = DurableFileBatch::new(); + let members = DURABLE_BATCH_CONCURRENT_SYNCS * 2 + 3; + for index in 0..members { + write_member(&mut batch, root.path(), index); + assert!(batch.pending.len() <= DURABLE_BATCH_CONCURRENT_SYNCS); + } + assert_eq!(batch.synced.len(), DURABLE_BATCH_CONCURRENT_SYNCS * 2); + assert_eq!(batch.pending.len(), 3); + batch.sync().expect("flush final window"); + assert_eq!( + events.drain(), + members, + "only the original writer opens are allowed; flushing must not reopen members" + ); + batch.sync().expect("repeat final barrier"); + assert_eq!(events.drain(), 0, "a repeated sync must not reopen paths"); + drop(batch); + for index in 0..members { + assert_eq!( + fs::read(root.path().join(format!("segment-{index}.tmp"))) + .expect("exact durable readback"), + format!("sealed segment {index}").as_bytes() + ); + } +} + +#[cfg(feature = "test-helpers")] +#[test] +fn durable_batch_flushes_each_window_before_releasing_its_handles() { + let root = tempfile::tempdir().expect("batch root"); + let barriers = sync_latency::inject(root.path(), Duration::ZERO); + let mut batch = DurableFileBatch::new(); + let members = DURABLE_BATCH_CONCURRENT_SYNCS * 2 + 3; + for index in 0..members { + write_member(&mut batch, root.path(), index); + } + assert_eq!( + barriers.syncs(), + u64::try_from(DURABLE_BATCH_CONCURRENT_SYNCS * 2).expect("flushed count"), + "both completed windows must have issued real file barriers" + ); + batch.sync().expect("flush remaining members"); + assert_eq!( + barriers.syncs(), + u64::try_from(members).expect("member count"), + "no member may be omitted or flushed as unrelated filesystem data" + ); +} + +#[test] +fn durable_batch_rejects_a_path_not_owned_by_the_writer() { + let root = tempfile::tempdir().expect("batch root"); + let path = root.path().join("member.tmp"); + let other = root.path().join("other.tmp"); + let mut writer = File::create(&path).expect("writer"); + writer.write_all(b"original").expect("member bytes"); + fs::write(&other, b"original").expect("same bytes in another inode"); + let mut batch = DurableFileBatch::new(); + assert_eq!( + batch + .written(&other, &writer) + .expect_err("wrong inode") + .kind(), + ErrorKind::InvalidData + ); + assert_eq!(batch.pending.len(), 0); + batch.written(&path, &writer).expect("correct writer path"); + drop(writer); + batch.sync().expect("flush caller-closed writer"); + assert_eq!(fs::read(path).expect("read member"), b"original"); +} + +#[test] +fn durable_batch_refuses_pending_and_already_flushed_replacements() { + for members in [1, DURABLE_BATCH_CONCURRENT_SYNCS + 1] { + let root = tempfile::tempdir().expect("batch root"); + let mut batch = DurableFileBatch::new(); + for index in 0..members { + write_member(&mut batch, root.path(), index); + } + let first = root.path().join("segment-0.tmp"); + let replacement = root.path().join("replacement.tmp"); + // Same bytes rule out content mismatch as the reason for refusing. + fs::write(&replacement, b"sealed segment 0").expect("replacement"); + fs::rename(&replacement, &first).expect("replace recorded inode"); + for _ in 0..2 { + assert_eq!( + batch.sync().expect_err("replaced member").kind(), + ErrorKind::InvalidData + ); + } + assert_eq!( + fs::read(first).expect("replacement preserved"), + b"sealed segment 0" + ); + } +} + +#[test] +fn durable_batch_refuses_a_missing_member_from_an_earlier_window() { + let root = tempfile::tempdir().expect("batch root"); + let mut batch = DurableFileBatch::new(); + for index in 0..=DURABLE_BATCH_CONCURRENT_SYNCS { + write_member(&mut batch, root.path(), index); + } + assert_eq!(batch.synced.len(), DURABLE_BATCH_CONCURRENT_SYNCS); + fs::remove_file(root.path().join("segment-0.tmp")).expect("remove flushed member"); + assert_eq!( + batch.sync().expect_err("missing flushed member").kind(), + ErrorKind::NotFound + ); +} + +#[test] +fn durable_batch_refuses_same_inode_mutation_after_early_flush() { + let root = tempfile::tempdir().expect("batch root"); + let mut batch = DurableFileBatch::new(); + for index in 0..=DURABLE_BATCH_CONCURRENT_SYNCS { + write_member(&mut batch, root.path(), index); + } + let path = root.path().join("segment-0.tmp"); + let before = fs::metadata(&path).expect("original metadata"); + let mut writer = File::options().write(true).open(&path).expect("same inode"); + writer + .write_all(b"changed segment!") + .expect("same-length mutation"); + writer + .set_modified(before.modified().expect("mtime") + Duration::from_secs(2)) + .expect("observable mutation on coarse timestamp filesystems"); + drop(writer); + let after = fs::metadata(&path).expect("mutated metadata"); + assert_eq!( + (before.dev(), before.ino(), before.len()), + (after.dev(), after.ino(), after.len()) + ); + assert_ne!( + before.modified().expect("original mtime"), + after.modified().expect("changed mtime") + ); + assert_eq!( + batch.sync().expect_err("mutated early member").kind(), + ErrorKind::InvalidData + ); +} + +#[test] +fn durable_batch_refuses_a_changed_authenticated_writer_before_admission_and_recovers() { + let root = tempfile::tempdir().expect("batch root"); + let mut batch = DurableFileBatch::new(); + for index in 0..DURABLE_BATCH_CONCURRENT_SYNCS { + write_member(&mut batch, root.path(), index); + } + + let next = root.path().join("next.tmp"); + let mut writer = File::create(&next).expect("candidate writer"); + writer.write_all(b"next segment").expect("candidate bytes"); + let authenticated = writer.metadata().expect("authenticated inode"); + writer + .set_modified(authenticated.modified().expect("mtime") + Duration::from_secs(2)) + .expect("change metadata after authentication"); + #[cfg(feature = "test-helpers")] + let barriers = sync_latency::inject(root.path(), Duration::ZERO); + + assert_eq!( + batch + .written_with_identity(&next, &writer, &authenticated) + .expect_err("stale authentication must refuse admission") + .kind(), + ErrorKind::InvalidData + ); + #[cfg(feature = "test-helpers")] + assert_eq!( + barriers.syncs(), + 0, + "invalid next member must not flush the healthy 32-member window" + ); + assert_eq!(batch.pending.len(), DURABLE_BATCH_CONCURRENT_SYNCS); + assert_eq!(batch.synced.len(), 0); + drop(writer); + fs::remove_file(&next).expect("discard rejected temporary"); + + // A new writer for the same name can be admitted because the rejected + // inode was never registered in the batch. + let mut retry = File::create(&next).expect("retry writer"); + retry.write_all(b"next segment").expect("retry bytes"); + let retry_identity = retry.metadata().expect("retry identity"); + batch + .written_with_identity(&next, &retry, &retry_identity) + .expect("admit the replacement writer"); + #[cfg(feature = "test-helpers")] + assert_eq!( + barriers.syncs(), + u64::try_from(DURABLE_BATCH_CONCURRENT_SYNCS).expect("window size"), + "the 32 accepted members need real barriers before their handles release" + ); + assert_eq!(batch.pending.len(), 1); + assert_eq!(batch.synced.len(), DURABLE_BATCH_CONCURRENT_SYNCS); + drop(retry); + batch.sync().expect("flush the replacement member"); + #[cfg(feature = "test-helpers")] + assert_eq!( + barriers.syncs(), + u64::try_from(DURABLE_BATCH_CONCURRENT_SYNCS + 1).expect("total count"), + "retry must not skip or duplicate durability barriers" + ); + assert_eq!(fs::read(&next).expect("read back retry"), b"next segment"); +} + +#[cfg(feature = "test-helpers")] +#[test] +fn durable_batch_invalid_next_member_does_not_flush_a_full_window() { + let root = tempfile::tempdir().expect("batch root"); + let mut batch = DurableFileBatch::new(); + for index in 0..DURABLE_BATCH_CONCURRENT_SYNCS { + write_member(&mut batch, root.path(), index); + } + + let next = root.path().join("next.tmp"); + let other = root.path().join("unrelated.tmp"); + let mut writer = File::create(&next).expect("next writer"); + writer.write_all(b"next segment").expect("next bytes"); + fs::write(&other, b"next segment").expect("same bytes, different inode"); + let barriers = sync_latency::inject(root.path(), Duration::ZERO); + + assert_eq!( + batch + .written(&other, &writer) + .expect_err("a foreign pathname must be rejected before flushing") + .kind(), + ErrorKind::InvalidData + ); + assert_eq!( + barriers.syncs(), + 0, + "an invalid new member must not trigger durability work on a full valid window" + ); + assert_eq!(batch.pending.len(), DURABLE_BATCH_CONCURRENT_SYNCS); + assert_eq!(batch.synced.len(), 0); + + batch.written(&next, &writer).expect("admit correct member"); + assert_eq!( + barriers.syncs(), + u64::try_from(DURABLE_BATCH_CONCURRENT_SYNCS).expect("window size"), + "admitting the real next member must flush the previous window" + ); + assert_eq!(batch.pending.len(), 1); + assert_eq!(batch.synced.len(), DURABLE_BATCH_CONCURRENT_SYNCS); + drop(writer); + batch.sync().expect("flush admitted member"); + assert_eq!( + barriers.syncs(), + u64::try_from(DURABLE_BATCH_CONCURRENT_SYNCS + 1).expect("member count"), + "the optimization must not skip a durability barrier" + ); + assert_eq!(fs::read(&next).expect("durable member"), b"next segment"); +} + +#[test] +fn durable_batch_failed_window_keeps_handles_and_does_not_admit_another_member() { + let root = tempfile::tempdir().expect("batch root"); + let mut batch = DurableFileBatch::new(); + for index in 0..DURABLE_BATCH_CONCURRENT_SYNCS { + write_member(&mut batch, root.path(), index); + } + fs::remove_file(root.path().join("segment-0.tmp")).expect("remove pending member"); + let next = root.path().join("next.tmp"); + let mut writer = File::create(&next).expect("next writer"); + writer.write_all(b"next segment").expect("next bytes"); + assert_eq!( + batch + .written(&next, &writer) + .expect_err("failed window") + .kind(), + ErrorKind::NotFound + ); + assert_eq!(batch.pending.len(), DURABLE_BATCH_CONCURRENT_SYNCS); + assert_eq!( + batch.synced.len(), + 0, + "no partial window may claim durability" + ); + assert_eq!( + batch.sync().expect_err("retry remains refused").kind(), + ErrorKind::NotFound + ); +}