From 889b1bdb573d016d41d482ed5be55ae6f7b6bd4b Mon Sep 17 00:00:00 2001 From: bsodfather Date: Thu, 27 Aug 2026 07:04:22 +0300 Subject: [PATCH] feat(artifacts): add bounded reconciliation --- crates/crank-artifacts/src/housekeeping.rs | 491 +++++++++++++++++- crates/crank-artifacts/src/lib.rs | 5 +- crates/crank-artifacts/src/model.rs | 49 ++ crates/crank-artifacts/src/store.rs | 19 +- .../crank-artifacts/tests/reconciliation.rs | 430 +++++++++++++++ 5 files changed, 985 insertions(+), 9 deletions(-) create mode 100644 crates/crank-artifacts/tests/reconciliation.rs diff --git a/crates/crank-artifacts/src/housekeeping.rs b/crates/crank-artifacts/src/housekeeping.rs index ac628dc..61fecd7 100644 --- a/crates/crank-artifacts/src/housekeeping.rs +++ b/crates/crank-artifacts/src/housekeeping.rs @@ -1,13 +1,66 @@ use std::{ + fmt, os::fd::AsRawFd, time::{Duration, SystemTime, UNIX_EPOCH}, }; use crate::{ - ArtifactError, ArtifactStore, - store::{RootLock, ensure_root_unchanged, fsync_fd, open_existing_dir, unlinkat}, + ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace, + ReconciliationScan, + store::{ + QUARANTINE_DIR, RootLock, check_file, checkpointed, ensure_root_unchanged, fsync_fd, + open_existing_dir, open_or_create_dir, rename_no_replace_at, stat_fd, unlinkat, + }, }; +const FINAL_NAMESPACE: u8 = 0; +const QUARANTINE_NAMESPACE: u8 = 1; + +/// Opaque continuation for a bounded reconciliation traversal. It is valid +/// only for the unchanged pinned root from which it was returned. +pub struct ReconciliationCursor { + root_dev: u64, + root_ino: u64, + namespace: u8, + shard: u8, + position: i64, +} + +impl fmt::Debug for ReconciliationCursor { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ReconciliationCursor(..)") + } +} + +/// Opaque, inode-bound evidence for one valid final or quarantined artifact. +/// It intentionally exposes neither its digest nor filesystem path. +#[derive(Clone)] +pub struct ReconciliationCandidate { + root_dev: u64, + root_ino: u64, + namespace: ReconciliationNamespace, + shard: String, + shard_dev: u64, + shard_ino: u64, + name: String, + dev: u64, + ino: u64, + modified_seconds: i64, + modified_nanoseconds: i64, +} + +impl fmt::Debug for ReconciliationCandidate { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ReconciliationCandidate(..)") + } +} + +impl ReconciliationCandidate { + pub fn namespace(&self) -> ReconciliationNamespace { + self.namespace + } +} + /// Opaque, single-use evidence that a particular stale temporary inode was /// observed under this store's pinned root. It is not a blob reference. #[derive(Debug)] @@ -30,6 +83,259 @@ pub struct TempScan { } impl ArtifactStore { + /// Streams valid final and quarantine entries with an opaque continuation. + /// Neither the report nor the returned capabilities disclose paths/digests. + pub fn scan_reconciliation( + &self, + continuation: Option, + scan_budget: usize, + result_limit: usize, + ) -> Result<(ReconciliationScan, Vec), ArtifactError> { + let root = self.root()?; + let _lock = RootLock::shared(root)?; + ensure_root_unchanged(root)?; + let mut cursor = match continuation { + Some(cursor) => { + if cursor.root_dev != root.dev + || cursor.root_ino != root.ino + || cursor.namespace > QUARANTINE_NAMESPACE + { + return Err(ArtifactError::UnsafeRoot); + } + cursor + } + None => reconciliation_cursor(root.dev, root.ino, FINAL_NAMESPACE, 0, 0), + }; + let mut report = ReconciliationScan::empty(None); + if scan_budget == 0 || result_limit == 0 { + report.continuation = Some(cursor); + return Ok((report, Vec::new())); + } + + let mut remaining = scan_budget; + let mut entries = Vec::new(); + while cursor.namespace <= QUARANTINE_NAMESPACE { + if remaining == 0 || entries.len() == result_limit { + report.continuation = Some(cursor); + return Ok((report, entries)); + } + let namespace_name = namespace_name(cursor.namespace); + let namespace = match open_existing_dir(root.fd.as_raw_fd(), namespace_name) { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => { + advance_namespace(&mut cursor); + continue; + } + Err(error) => return Err(error), + }; + loop { + if remaining == 0 { + report.continuation = Some(cursor); + return Ok((report, entries)); + } + let shard_name = format!("{:02x}", cursor.shard); + let shard = match open_existing_dir(namespace.as_raw_fd(), shard_name.as_bytes()) { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => { + consume_scan_work(&mut report, &mut remaining); + if cursor.shard == u8::MAX { + break; + } + advance_shard(&mut cursor); + continue; + } + Err(error) => return Err(error), + }; + let shard_stat = stat_fd(shard.as_raw_fd())?; + let mut directory = DirectoryStream::open(shard.as_raw_fd(), cursor.position)?; + loop { + if remaining == 0 || entries.len() == result_limit { + report.continuation = Some(cursor); + return Ok((report, entries)); + } + let Some((name, after)) = directory.next()? else { + // An empty directory still consumes one bounded + // traversal unit, so a sparse namespace cannot make + // us probe all 256 shards in one page. + consume_scan_work(&mut report, &mut remaining); + break; + }; + cursor.position = after; + consume_scan_work(&mut report, &mut remaining); + let Ok(name) = String::from_utf8(name) else { + report.malformed += 1; + continue; + }; + if !valid_final_name(&name, &shard_name) { + report.malformed += 1; + continue; + } + #[cfg(debug_assertions)] + let _ = crate::test_support::checkpoint("reconciliation_before_stat"); + let stat = match checkpointed("reconciliation_stat", || { + nofollow_stat(shard.as_raw_fd(), &name) + }) { + Ok(stat) => stat, + Err(ArtifactError::NotFound) => { + // The entry was concurrently removed. It is no + // longer retryable, so the cursor may advance. + continue; + } + Err(ArtifactError::UnsafeRoot) => { + report.unsafe_entries += 1; + continue; + } + Err(error) => return Err(error), + }; + if check_file(&stat).is_err() { + report.unsafe_entries += 1; + continue; + } + let namespace = if cursor.namespace == FINAL_NAMESPACE { + report.final_entries += 1; + ReconciliationNamespace::Final + } else { + report.quarantined_entries += 1; + ReconciliationNamespace::Quarantine + }; + entries.push(ReconciliationCandidate { + root_dev: root.dev, + root_ino: root.ino, + namespace, + shard: shard_name.clone(), + shard_dev: shard_stat.st_dev, + shard_ino: shard_stat.st_ino, + name, + dev: stat.st_dev, + ino: stat.st_ino, + modified_seconds: stat.st_mtime, + modified_nanoseconds: stat.st_mtime_nsec, + }); + // The continuation is deliberately kept at `after`, + // immediately before the next unissued item. + } + if cursor.shard == u8::MAX { + break; + } + advance_shard(&mut cursor); + } + advance_namespace(&mut cursor); + } + Ok((report, entries)) + } + + /// Moves a revalidated final inode into the private quarantine namespace + /// without replacing any existing quarantine inode. + pub fn quarantine_reconciliation( + &self, + candidate: ReconciliationCandidate, + ) -> Result { + if candidate.namespace != ReconciliationNamespace::Final { + return Err(ArtifactError::UnsafeRoot); + } + let root = self.root()?; + let _lock = RootLock::exclusive(root)?; + ensure_reconciliation_root(root, &candidate)?; + let final_root = open_existing_dir(root.fd.as_raw_fd(), namespace_name(FINAL_NAMESPACE))?; + let final_shard = open_existing_dir(final_root.as_raw_fd(), candidate.shard.as_bytes())?; + match revalidate_candidate(&candidate, final_shard.as_raw_fd()) { + Ok(()) => {} + Err(ArtifactError::NotFound) => { + return self.recover_quarantine(root, &candidate); + } + Err(ArtifactError::UnsafeRoot) => { + return match self.recover_quarantine(root, &candidate) { + Ok(outcome) => Ok(outcome), + Err(ArtifactError::NotFound) => Err(ArtifactError::UnsafeRoot), + Err(error) => Err(error), + }; + } + Err(error) => return Err(error), + } + let quarantine_root = open_or_create_dir(root.fd.as_raw_fd(), QUARANTINE_DIR)?; + let quarantine_shard = + open_or_create_dir(quarantine_root.as_raw_fd(), candidate.shard.as_bytes())?; + match checkpointed("reconciliation_quarantine_rename", || { + rename_no_replace_at( + final_shard.as_raw_fd(), + candidate.name.as_bytes(), + quarantine_shard.as_raw_fd(), + candidate.name.as_bytes(), + ) + })? { + true => match fsync_quarantine_dirs(&final_shard, &quarantine_shard) { + Ok(()) => Ok(ReconciliationMutation::Quarantined), + Err(_) => Ok(ReconciliationMutation::Retryable), + }, + false => self.recover_quarantine(root, &candidate), + } + } + + /// Idempotently removes a revalidated quarantined inode and fsyncs its + /// directory. An fsync ambiguity is explicitly retryable. + pub fn delete_quarantined_reconciliation( + &self, + candidate: ReconciliationCandidate, + ) -> Result { + if candidate.namespace != ReconciliationNamespace::Quarantine { + return Err(ArtifactError::UnsafeRoot); + } + let root = self.root()?; + let _lock = RootLock::exclusive(root)?; + ensure_reconciliation_root(root, &candidate)?; + let quarantine_root = match open_existing_dir(root.fd.as_raw_fd(), QUARANTINE_DIR) { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => return Ok(ReconciliationMutation::AlreadyAbsent), + Err(error) => return Err(error), + }; + let shard = match open_existing_dir(quarantine_root.as_raw_fd(), candidate.shard.as_bytes()) + { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => return Ok(ReconciliationMutation::AlreadyAbsent), + Err(error) => return Err(error), + }; + match revalidate_candidate(&candidate, shard.as_raw_fd()) { + Ok(()) => {} + Err(ArtifactError::NotFound) => { + return fsync_mutation(&shard, ReconciliationMutation::AlreadyAbsent); + } + Err(error) => return Err(error), + } + match checkpointed("reconciliation_delete_unlink", || { + unlinkat(shard.as_raw_fd(), candidate.name.as_bytes()) + }) { + Ok(()) => fsync_mutation(&shard, ReconciliationMutation::Deleted), + Err(ArtifactError::NotFound) => { + fsync_mutation(&shard, ReconciliationMutation::AlreadyAbsent) + } + Err(error) => Err(error), + } + } + + fn recover_quarantine( + &self, + root: &crate::store::Root, + candidate: &ReconciliationCandidate, + ) -> Result { + let final_root = open_existing_dir(root.fd.as_raw_fd(), namespace_name(FINAL_NAMESPACE))?; + let final_shard = open_existing_dir(final_root.as_raw_fd(), candidate.shard.as_bytes())?; + let quarantine_root = match open_existing_dir(root.fd.as_raw_fd(), QUARANTINE_DIR) { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => return Err(ArtifactError::NotFound), + Err(error) => return Err(error), + }; + let shard = match open_existing_dir(quarantine_root.as_raw_fd(), candidate.shard.as_bytes()) + { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => return Err(ArtifactError::NotFound), + Err(error) => return Err(error), + }; + revalidate_moved_candidate(candidate, shard.as_raw_fd())?; + match fsync_quarantine_dirs(&final_shard, &shard) { + Ok(()) => Ok(ReconciliationMutation::AlreadyQuarantined), + Err(_) => Ok(ReconciliationMutation::Retryable), + } + } /// Streams temporary entries, reading at most `scan_budget` directory /// entries and returning at most `result_limit` stale capabilities. pub fn scan_stale_temps( @@ -147,6 +453,187 @@ impl ArtifactStore { } } +fn reconciliation_cursor( + root_dev: u64, + root_ino: u64, + namespace: u8, + shard: u8, + position: i64, +) -> ReconciliationCursor { + ReconciliationCursor { + root_dev, + root_ino, + namespace, + shard, + position, + } +} + +fn namespace_name(namespace: u8) -> &'static [u8] { + match namespace { + FINAL_NAMESPACE => b"sha256", + QUARANTINE_NAMESPACE => QUARANTINE_DIR, + _ => unreachable!("validated reconciliation namespace"), + } +} + +fn advance_shard(cursor: &mut ReconciliationCursor) { + debug_assert_ne!(cursor.shard, u8::MAX); + cursor.shard += 1; + cursor.position = 0; +} + +fn advance_namespace(cursor: &mut ReconciliationCursor) { + cursor.namespace = cursor.namespace.saturating_add(1); + cursor.shard = 0; + cursor.position = 0; +} + +fn consume_scan_work(report: &mut ReconciliationScan, remaining: &mut usize) { + debug_assert!(*remaining > 0); + *remaining -= 1; + report.scanned += 1; +} + +fn ensure_reconciliation_root( + root: &crate::store::Root, + candidate: &ReconciliationCandidate, +) -> Result<(), ArtifactError> { + ensure_root_unchanged(root)?; + if candidate.root_dev != root.dev + || candidate.root_ino != root.ino + || candidate.shard.len() != 2 + || !candidate.shard.bytes().all(is_lower_hex) + || !valid_final_name(&candidate.name, &candidate.shard) + { + return Err(ArtifactError::UnsafeRoot); + } + Ok(()) +} + +fn revalidate_candidate( + candidate: &ReconciliationCandidate, + shard: i32, +) -> Result<(), ArtifactError> { + let shard_stat = stat_fd(shard)?; + if shard_stat.st_dev != candidate.shard_dev || shard_stat.st_ino != candidate.shard_ino { + return Err(ArtifactError::UnsafeRoot); + } + let stat = nofollow_stat(shard, &candidate.name)?; + if stat.st_dev != candidate.dev + || stat.st_ino != candidate.ino + || stat.st_mtime != candidate.modified_seconds + || stat.st_mtime_nsec != candidate.modified_nanoseconds + { + return Err(ArtifactError::UnsafeRoot); + } + check_file(&stat).map_err(|_| ArtifactError::UnsafeRoot) +} + +/// Recovery follows a successful cross-directory rename. The candidate remains +/// bound to the original final shard, so the quarantine shard identity cannot +/// equal its recorded shard identity; only the canonical name and inode are +/// revalidated here. +fn revalidate_moved_candidate( + candidate: &ReconciliationCandidate, + shard: i32, +) -> Result<(), ArtifactError> { + let stat = nofollow_stat(shard, &candidate.name)?; + if stat.st_dev != candidate.dev + || stat.st_ino != candidate.ino + || stat.st_mtime != candidate.modified_seconds + || stat.st_mtime_nsec != candidate.modified_nanoseconds + { + return Err(ArtifactError::UnsafeRoot); + } + check_file(&stat).map_err(|_| ArtifactError::UnsafeRoot) +} + +fn fsync_mutation( + shard: &std::os::fd::OwnedFd, + success: ReconciliationMutation, +) -> Result { + match checkpointed("reconciliation_delete_fsync", || { + fsync_fd(shard.as_raw_fd()) + }) { + Ok(()) => Ok(success), + Err(_) => Ok(ReconciliationMutation::Retryable), + } +} + +/// Persist both directory entry changes even if the first fsync reports an +/// error. Either failure remains an ambiguity, but skipping the second fsync +/// would needlessly enlarge the crash window. +fn fsync_quarantine_dirs( + final_shard: &std::os::fd::OwnedFd, + quarantine_shard: &std::os::fd::OwnedFd, +) -> Result<(), ArtifactError> { + let source = checkpointed("reconciliation_quarantine_source_fsync", || { + fsync_fd(final_shard.as_raw_fd()) + }); + let destination = checkpointed("reconciliation_quarantine_destination_fsync", || { + fsync_fd(quarantine_shard.as_raw_fd()) + }); + source.and(destination) +} + +fn valid_final_name(name: &str, shard: &str) -> bool { + name.len() == 64 && name.starts_with(shard) && name.bytes().all(is_lower_hex) +} + +struct DirectoryStream { + directory: *mut libc::DIR, +} + +impl DirectoryStream { + fn open(parent: i32, position: i64) -> Result { + let duplicate = crate::store::duplicate_fd(parent)?; + let raw = std::os::fd::IntoRawFd::into_raw_fd(duplicate); + let directory = unsafe { libc::fdopendir(raw) }; + if directory.is_null() { + unsafe { + libc::close(raw); + } + return Err(ArtifactError::Storage); + } + if position != 0 { + unsafe { + libc::seekdir(directory, position as libc::c_long); + } + } + Ok(Self { directory }) + } + + fn next(&mut self) -> Result, i64)>, ArtifactError> { + loop { + unsafe { + *libc::__errno_location() = 0; + } + let entry = unsafe { libc::readdir(self.directory) }; + if entry.is_null() { + if unsafe { *libc::__errno_location() } != 0 { + return Err(ArtifactError::Storage); + } + return Ok(None); + } + let after = unsafe { libc::telldir(self.directory) } as i64; + let name = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes(); + if name == b"." || name == b".." { + continue; + } + return Ok(Some((name.to_vec(), after))); + } + } +} + +impl Drop for DirectoryStream { + fn drop(&mut self) { + unsafe { + libc::closedir(self.directory); + } + } +} + fn list_names( parent: i32, remaining: &mut usize, diff --git a/crates/crank-artifacts/src/lib.rs b/crates/crank-artifacts/src/lib.rs index 320ed72..ce1d1f9 100644 --- a/crates/crank-artifacts/src/lib.rs +++ b/crates/crank-artifacts/src/lib.rs @@ -18,8 +18,9 @@ mod store; pub mod test_support; pub use error::ArtifactError; -pub use housekeeping::{StaleTemp, TempScan}; +pub use housekeeping::{ReconciliationCandidate, ReconciliationCursor, StaleTemp, TempScan}; pub use model::{ - ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, RegisteredArtifact, StoredArtifact, + ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, ReconciliationMutation, + ReconciliationNamespace, ReconciliationScan, RegisteredArtifact, StoredArtifact, }; pub use store::ArtifactStore; diff --git a/crates/crank-artifacts/src/model.rs b/crates/crank-artifacts/src/model.rs index 3195b5b..b0eb9b7 100644 --- a/crates/crank-artifacts/src/model.rs +++ b/crates/crank-artifacts/src/model.rs @@ -77,6 +77,55 @@ pub struct RegisteredArtifact { pub(crate) size_bytes: usize, } +/// Namespace in which reconciliation observed an artifact inode. It does not +/// disclose the artifact's digest or filesystem location. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ReconciliationNamespace { + Final, + Quarantine, +} + +/// Bounded, redacted progress report from a reconciliation page. +/// +/// `continuation` is opaque evidence for resuming an unchanged namespace. It +/// intentionally cannot be created from a caller-supplied path or digest. +#[derive(Debug)] +pub struct ReconciliationScan { + pub scanned: usize, + pub malformed: usize, + pub unsafe_entries: usize, + pub final_entries: usize, + pub quarantined_entries: usize, + pub continuation: Option, +} + +impl ReconciliationScan { + pub(crate) fn empty(continuation: Option) -> Self { + Self { + scanned: 0, + malformed: 0, + unsafe_entries: 0, + final_entries: 0, + quarantined_entries: 0, + continuation, + } + } +} + +/// The durable/retryable result of a reconciliation mutation. +/// +/// A successful rename or unlink followed by an fsync failure is intentionally +/// reported as `Retryable`: a subsequent bounded scan determines the durable +/// state without treating the operation as a failed no-op. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ReconciliationMutation { + Quarantined, + AlreadyQuarantined, + Deleted, + AlreadyAbsent, + Retryable, +} + impl RegisteredArtifact { pub fn artifact_ref(&self) -> &ArtifactRef { &self.artifact_ref diff --git a/crates/crank-artifacts/src/store.rs b/crates/crank-artifacts/src/store.rs index 4e80278..45bfd57 100644 --- a/crates/crank-artifacts/src/store.rs +++ b/crates/crank-artifacts/src/store.rs @@ -16,6 +16,7 @@ use sha2::{Digest, Sha256}; use crate::{ArtifactError, ArtifactRef, MAX_ARTIFACT_BYTES, RegisteredArtifact, StoredArtifact}; const SHA_DIR: &[u8] = b"sha256"; +pub(crate) const QUARANTINE_DIR: &[u8] = b"quarantine"; const TEMP_PREFIX: &str = ".crank-artifact-tmp-v1-"; const RESOLVE_NO_SYMLINKS: u64 = 0x04; static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0); @@ -382,7 +383,7 @@ fn check_private_dir(stat: &libc::stat) -> Result<(), ArtifactError> { } Ok(()) } -fn stat_fd(fd: i32) -> Result { +pub(crate) fn stat_fd(fd: i32) -> Result { let mut stat = unsafe { std::mem::zeroed() }; if unsafe { libc::fstat(fd, &mut stat) } != 0 { Err(ArtifactError::Storage) @@ -390,7 +391,7 @@ fn stat_fd(fd: i32) -> Result { Ok(stat) } } -fn check_file(stat: &libc::stat) -> Result<(), ArtifactError> { +pub(crate) fn check_file(stat: &libc::stat) -> Result<(), ArtifactError> { if (stat.st_mode & libc::S_IFMT) != libc::S_IFREG || stat.st_uid != unsafe { libc::geteuid() } || stat.st_nlink != 1 @@ -508,7 +509,7 @@ fn write_all(fd: &OwnedFd, bytes: &[u8]) -> Result<(), ArtifactError> { /// Debug/test-only fault seam coupled to one syscall wrapper. `Fail` models a /// syscall that did not run; crash/hold actions happen only after success. -fn checkpointed( +pub(crate) fn checkpointed( stage: &str, operation: impl FnOnce() -> Result, ) -> Result { @@ -608,14 +609,22 @@ fn chmod_read_only(fd: i32) -> Result<(), ArtifactError> { } } fn rename_no_replace(parent: i32, old: &[u8], new: &[u8]) -> Result { + rename_no_replace_at(parent, old, parent, new) +} +pub(crate) fn rename_no_replace_at( + old_parent: i32, + old: &[u8], + new_parent: i32, + new: &[u8], +) -> Result { let old = c_name(old)?; let new = c_name(new)?; let result = unsafe { libc::syscall( libc::SYS_renameat2, - parent, + old_parent, old.as_ptr(), - parent, + new_parent, new.as_ptr(), libc::RENAME_NOREPLACE, ) diff --git a/crates/crank-artifacts/tests/reconciliation.rs b/crates/crank-artifacts/tests/reconciliation.rs new file mode 100644 index 0000000..903dee5 --- /dev/null +++ b/crates/crank-artifacts/tests/reconciliation.rs @@ -0,0 +1,430 @@ +use std::{ + fs, + os::unix::fs::PermissionsExt, + path::PathBuf, + process::Command, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + thread, + time::Duration, +}; + +use crank_artifacts::{ + ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace, +}; + +#[cfg(debug_assertions)] +use crank_artifacts::test_support::{ + FaultAction, clear_checkpoint, set_checkpoint, wait_until_held, +}; +#[cfg(debug_assertions)] +use std::sync::{Mutex, OnceLock}; + +static NEXT_ROOT: AtomicU64 = AtomicU64::new(0); + +struct TestRoot(PathBuf); + +impl TestRoot { + fn new(name: &str) -> Self { + let path = std::env::temp_dir().join(format!( + "crank-artifacts-reconciliation-{name}-{}-{}", + std::process::id(), + NEXT_ROOT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&path).unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap(); + Self(path) + } +} + +impl Drop for TestRoot { + fn drop(&mut self) { + let _ = fs::set_permissions(&self.0, fs::Permissions::from_mode(0o700)); + let _ = fs::remove_dir_all(&self.0); + } +} + +#[cfg(debug_assertions)] +fn fault_guard() -> std::sync::MutexGuard<'static, ()> { + static GUARD: OnceLock> = OnceLock::new(); + GUARD.get_or_init(|| Mutex::new(())).lock().unwrap() +} + +#[test] +fn paginates_final_entries_without_disclosing_locations() { + let root = TestRoot::new("pages"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"reconciliation first").unwrap(); + store.put(b"reconciliation second").unwrap(); + + let mut continuation = None; + let mut candidates = Vec::new(); + loop { + let (report, mut page) = store.scan_reconciliation(continuation, 1, 1).unwrap(); + assert!(report.scanned <= 1); + assert!(page.len() <= 1); + candidates.append(&mut page); + continuation = report.continuation; + if continuation.is_none() { + break; + } + } + assert_eq!(candidates.len(), 2); + assert!( + candidates + .iter() + .all(|candidate| candidate.namespace() == ReconciliationNamespace::Final) + ); + for candidate in candidates { + assert_eq!( + store.quarantine_reconciliation(candidate).unwrap(), + ReconciliationMutation::Quarantined + ); + } + let (_, quarantined_entries) = store.scan_reconciliation(None, 512, 8).unwrap(); + assert_eq!(quarantined_entries.len(), 2); + assert!( + quarantined_entries + .iter() + .all(|candidate| candidate.namespace() == ReconciliationNamespace::Quarantine) + ); +} + +#[test] +fn classifies_malformed_and_unsafe_entries_with_a_usable_page() { + let root = TestRoot::new("malformed"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reconciliation valid").unwrap(); + let shard = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]); + fs::write(shard.join("not-a-digest"), b"junk").unwrap(); + let unsafe_name = format!( + "{}{}", + &stored.artifact_ref.digest_hex()[..2], + "f".repeat(62) + ); + std::os::unix::fs::symlink(shard.join("not-a-digest"), shard.join(unsafe_name)).unwrap(); + let directory_name = format!( + "{}{}", + &stored.artifact_ref.digest_hex()[..2], + "e".repeat(62) + ); + fs::create_dir(shard.join(directory_name)).unwrap(); + let hardlink_name = format!( + "{}{}", + &stored.artifact_ref.digest_hex()[..2], + "d".repeat(62) + ); + fs::hard_link(shard.join("not-a-digest"), shard.join(hardlink_name)).unwrap(); + + let (report, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + assert!(report.malformed >= 1); + assert!(report.unsafe_entries >= 3); + assert_eq!(candidates.len(), 1); + assert!(report.continuation.is_none()); +} + +#[test] +fn rejects_a_cursor_from_another_store() { + let first_root = TestRoot::new("foreign-cursor-first"); + let second_root = TestRoot::new("foreign-cursor-second"); + let first = ArtifactStore::open(&first_root.0).unwrap(); + let second = ArtifactStore::open(&second_root.0).unwrap(); + first.put(b"cursor source").unwrap(); + let (report, _) = first.scan_reconciliation(None, 1, 1).unwrap(); + + assert!(matches!( + second.scan_reconciliation(report.continuation, 1, 1), + Err(ArtifactError::UnsafeRoot) + )); +} + +#[test] +#[cfg(debug_assertions)] +fn disappearing_entry_after_readdir_is_skipped() { + let _guard = fault_guard(); + let root = TestRoot::new("disappearing"); + let store = Arc::new(ArtifactStore::open(&root.0).unwrap()); + let stored = store.put(b"disappearing reconciliation entry").unwrap(); + let path = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(stored.artifact_ref.digest_hex()); + set_checkpoint("reconciliation_before_stat", FaultAction::Hold); + let scan_store = Arc::clone(&store); + let scan = thread::spawn(move || scan_store.scan_reconciliation(None, 512, 8)); + assert!(wait_until_held(Duration::from_secs(2))); + fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap(); + fs::remove_file(&path).unwrap(); + clear_checkpoint(); + + let (_, candidates) = scan.join().unwrap().unwrap(); + assert!(candidates.is_empty()); +} + +#[test] +fn quarantine_and_delete_are_idempotent() { + let root = TestRoot::new("mutation"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reconciliation mutation").unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + + assert_eq!( + store.quarantine_reconciliation(candidate.clone()).unwrap(), + ReconciliationMutation::Quarantined + ); + assert_eq!( + store.read(&stored.artifact_ref), + Err(ArtifactError::NotFound) + ); + assert_eq!( + store.quarantine_reconciliation(candidate).unwrap(), + ReconciliationMutation::AlreadyQuarantined + ); + + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let quarantined = candidates + .into_iter() + .find(|candidate| candidate.namespace() == ReconciliationNamespace::Quarantine) + .unwrap(); + assert_eq!( + store + .delete_quarantined_reconciliation(quarantined.clone()) + .unwrap(), + ReconciliationMutation::Deleted + ); + assert_eq!( + store + .delete_quarantined_reconciliation(quarantined) + .unwrap(), + ReconciliationMutation::AlreadyAbsent + ); +} + +#[test] +fn quarantine_preserves_old_inode_when_the_digest_is_republished() { + let root = TestRoot::new("no-clobber"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reconciliation no-clobber").unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let original = candidates.into_iter().next().unwrap(); + assert_eq!( + store.quarantine_reconciliation(original.clone()).unwrap(), + ReconciliationMutation::Quarantined + ); + store.put(b"reconciliation no-clobber").unwrap(); + assert_eq!( + store.quarantine_reconciliation(original).unwrap(), + ReconciliationMutation::AlreadyQuarantined + ); + assert_eq!( + store.read(&stored.artifact_ref).unwrap(), + b"reconciliation no-clobber" + ); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + assert_eq!(candidates.len(), 2); + assert!( + candidates + .iter() + .any(|candidate| candidate.namespace() == ReconciliationNamespace::Final) + ); + assert!( + candidates + .iter() + .any(|candidate| candidate.namespace() == ReconciliationNamespace::Quarantine) + ); +} + +#[test] +fn delete_fails_closed_for_a_replaced_or_hardlinked_quarantined_inode() { + let root = TestRoot::new("quarantine-replaced"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store + .put(b"reconciliation quarantined replacement") + .unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + let path = root + .0 + .join("quarantine") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(stored.artifact_ref.digest_hex()); + fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap(); + fs::remove_file(&path).unwrap(); + fs::write(&path, b"replacement").unwrap(); + let sibling = path.with_extension("link"); + fs::hard_link(&path, &sibling).unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o400)).unwrap(); + + assert_eq!( + store.delete_quarantined_reconciliation(candidate), + Err(ArtifactError::UnsafeRoot) + ); + assert!(path.exists()); + assert!(sibling.exists()); +} + +#[test] +#[cfg(debug_assertions)] +fn delete_fsync_ambiguity_is_retryable_and_recovers_as_absent() { + let _guard = fault_guard(); + let root = TestRoot::new("delete-fsync"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"reconciliation delete fsync").unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + set_checkpoint("reconciliation_delete_fsync", FaultAction::Fail); + assert_eq!( + store + .delete_quarantined_reconciliation(candidate.clone()) + .unwrap(), + ReconciliationMutation::Retryable + ); + clear_checkpoint(); + assert_eq!( + store.delete_quarantined_reconciliation(candidate).unwrap(), + ReconciliationMutation::AlreadyAbsent + ); +} + +#[test] +fn rejects_a_replaced_final_inode() { + let root = TestRoot::new("replaced"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reconciliation original").unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + let path = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(stored.artifact_ref.digest_hex()); + fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap(); + fs::remove_file(&path).unwrap(); + fs::write(&path, b"replacement").unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o400)).unwrap(); + assert_eq!( + store.quarantine_reconciliation(candidate), + Err(ArtifactError::UnsafeRoot) + ); + assert!(path.exists()); +} + +#[test] +#[cfg(debug_assertions)] +fn retries_after_post_rename_fsync_ambiguity() { + let _guard = fault_guard(); + let root = TestRoot::new("fsync"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"reconciliation fault").unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + set_checkpoint("reconciliation_quarantine_source_fsync", FaultAction::Fail); + assert_eq!( + store.quarantine_reconciliation(candidate.clone()).unwrap(), + ReconciliationMutation::Retryable + ); + clear_checkpoint(); + assert_eq!( + store.quarantine_reconciliation(candidate).unwrap(), + ReconciliationMutation::AlreadyQuarantined + ); +} + +#[test] +#[cfg(debug_assertions)] +fn storage_fault_during_stat_propagates() { + let _guard = fault_guard(); + let root = TestRoot::new("stat-storage"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"reconciliation stat storage").unwrap(); + set_checkpoint("reconciliation_stat", FaultAction::Fail); + assert!(matches!( + store.scan_reconciliation(None, 512, 8), + Err(ArtifactError::Storage) + )); + clear_checkpoint(); +} + +#[test] +#[cfg(debug_assertions)] +fn crash_windows_are_recoverable() { + if let Some(root) = std::env::var_os("CRANK_RECONCILIATION_CHILD_ROOT") { + let action = std::env::var("CRANK_RECONCILIATION_CHILD_ACTION").unwrap(); + set_checkpoint( + std::env::var("CRANK_RECONCILIATION_CHILD_STAGE").unwrap(), + FaultAction::Exit, + ); + let store = ArtifactStore::open(PathBuf::from(root)).unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + if action == "quarantine" { + let _ = store.quarantine_reconciliation(candidate); + } else { + let _ = store.delete_quarantined_reconciliation(candidate); + } + panic!("checkpoint did not terminate the child"); + } + + for (stage, action) in [ + ("reconciliation_quarantine_rename", "quarantine"), + ("reconciliation_quarantine_source_fsync", "quarantine"), + ("reconciliation_quarantine_destination_fsync", "quarantine"), + ("reconciliation_delete_unlink", "delete"), + ("reconciliation_delete_fsync", "delete"), + ] { + let root = TestRoot::new(&format!("crash-{stage}")); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reconciliation crash bytes").unwrap(); + if action == "delete" { + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + } + let status = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "crash_windows_are_recoverable", "--nocapture"]) + .env("CRANK_RECONCILIATION_CHILD_ROOT", &root.0) + .env("CRANK_RECONCILIATION_CHILD_ACTION", action) + .env("CRANK_RECONCILIATION_CHILD_STAGE", stage) + .status() + .unwrap(); + assert_eq!(status.code(), Some(86), "stage={stage}"); + + let store = ArtifactStore::open(&root.0).unwrap(); + let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + for candidate in candidates { + match candidate.namespace() { + ReconciliationNamespace::Final => { + let _ = store.quarantine_reconciliation(candidate).unwrap(); + } + ReconciliationNamespace::Quarantine => { + let _ = store.delete_quarantined_reconciliation(candidate).unwrap(); + } + } + } + let (_, remaining) = store.scan_reconciliation(None, 512, 8).unwrap(); + assert!(remaining.is_empty(), "stage={stage}"); + assert_eq!( + store + .put(b"reconciliation crash bytes") + .unwrap() + .artifact_ref, + stored.artifact_ref + ); + } +}