diff --git a/crates/crank-artifacts/src/housekeeping.rs b/crates/crank-artifacts/src/housekeeping.rs index 61fecd7..9fdef3c 100644 --- a/crates/crank-artifacts/src/housekeeping.rs +++ b/crates/crank-artifacts/src/housekeeping.rs @@ -1,12 +1,13 @@ use std::{ fmt, - os::fd::AsRawFd, + os::fd::{AsRawFd, FromRawFd, OwnedFd}, time::{Duration, SystemTime, UNIX_EPOCH}, }; +use crate::temp_scan::{list_names, valid_temp_name}; use crate::{ ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace, - ReconciliationScan, + ReconciliationScan, ReconciliationScanStop, 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, @@ -16,24 +17,54 @@ use crate::{ 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. +/// Opaque bounded-scan continuation, valid only for an unchanged namespace. +/// Discard it after any put, quarantine, delete, or external mutation. pub struct ReconciliationCursor { root_dev: u64, root_ino: u64, namespace: u8, shard: u8, - position: i64, + state: ReconciliationCursorState, } +enum ReconciliationCursorState { + OpenNamespace, + ValidateNamespace(OwnedFd), + OpenShard(OwnedFd), + ValidateShard { + namespace: OwnedFd, + shard: OwnedFd, + }, + ReadShard { + namespace: OwnedFd, + shard: OwnedFd, + shard_dev: u64, + shard_ino: u64, + buffer: Vec, + offset: usize, + filled: usize, + cookie: i64, + }, + SeekShard { + namespace: OwnedFd, + shard: OwnedFd, + shard_dev: u64, + shard_ino: u64, + buffer: Vec, + cookie: i64, + }, + Transition, +} + +struct ShardIdentity(u64, u64); + 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. +/// Opaque inode-bound evidence that exposes neither digest nor filesystem path. #[derive(Clone)] pub struct ReconciliationCandidate { root_dev: u64, @@ -61,8 +92,7 @@ impl ReconciliationCandidate { } } -/// Opaque, single-use evidence that a particular stale temporary inode was -/// observed under this store's pinned root. It is not a blob reference. +/// Opaque single-use evidence for a stale temporary inode under the pinned root. #[derive(Debug)] pub struct StaleTemp { name: String, @@ -83,8 +113,7 @@ 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. + /// Streams entries without disclosing paths/digests, with opaque continuation. pub fn scan_reconciliation( &self, continuation: Option, @@ -92,8 +121,6 @@ impl ArtifactStore { 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 @@ -104,128 +131,89 @@ impl ArtifactStore { } cursor } - None => reconciliation_cursor(root.dev, root.ino, FINAL_NAMESPACE, 0, 0), + None => reconciliation_cursor(root.dev, root.ino), }; - let mut report = ReconciliationScan::empty(None); - if scan_budget == 0 || result_limit == 0 { - report.continuation = Some(cursor); - return Ok((report, Vec::new())); + let report = ReconciliationScan::empty(None); + let _lock = match RootLock::shared(root) { + Ok(lock) => lock, + Err(ArtifactError::Storage) => { + return Ok(reconciliation_page( + report, + Vec::new(), + cursor, + ReconciliationScanStop::Retryable, + )); + } + Err(error) => return Err(error), + }; + match ensure_root_unchanged(root) { + Ok(()) => {} + Err(ArtifactError::Storage) => { + return Ok(reconciliation_page( + report, + Vec::new(), + cursor, + ReconciliationScanStop::Retryable, + )); + } + Err(error) => return Err(error), + } + let mut report = report; + if scan_budget == 0 { + return Ok(reconciliation_page( + report, + Vec::new(), + cursor, + ReconciliationScanStop::ScanBudget, + )); + } + if result_limit == 0 { + return Ok(reconciliation_page( + report, + Vec::new(), + cursor, + ReconciliationScanStop::ResultLimit, + )); } - 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)); + if entries.len() == result_limit { + return Ok(reconciliation_page( + report, + entries, + cursor, + ReconciliationScanStop::ResultLimit, + )); } - 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); + if report.traversal_syscalls == scan_budget { + return Ok(reconciliation_page( + report, + entries, + cursor, + ReconciliationScanStop::ScanBudget, + )); + } + let state = std::mem::replace(&mut cursor.state, ReconciliationCursorState::Transition); + match reconciliation_step(root, &mut cursor, state, &mut report, &mut entries) { + Ok(()) => {} + Err((state, ArtifactError::Storage)) => { + cursor.state = state; + return Ok(reconciliation_page( + report, + entries, + cursor, + ReconciliationScanStop::Retryable, + )); + } + Err((_, error)) => return Err(error), } - advance_namespace(&mut cursor); } + report.stop = ReconciliationScanStop::Complete; Ok((report, entries)) } - /// Moves a revalidated final inode into the private quarantine namespace - /// without replacing any existing quarantine inode. + /// Moves a revalidated final inode to quarantine without replacing an inode. pub fn quarantine_reconciliation( &self, candidate: ReconciliationCandidate, @@ -271,8 +259,7 @@ impl ArtifactStore { } } - /// Idempotently removes a revalidated quarantined inode and fsyncs its - /// directory. An fsync ambiguity is explicitly retryable. + /// Idempotently removes a revalidated quarantined inode and fsyncs its directory. pub fn delete_quarantined_reconciliation( &self, candidate: ReconciliationCandidate, @@ -336,8 +323,7 @@ impl ArtifactStore { Err(_) => Ok(ReconciliationMutation::Retryable), } } - /// Streams temporary entries, reading at most `scan_budget` directory - /// entries and returning at most `result_limit` stale capabilities. + /// Scans at most `scan_budget` entries and returns `result_limit` stale capabilities. pub fn scan_stale_temps( &self, grace: Duration, @@ -409,8 +395,7 @@ impl ArtifactStore { Ok((report, result)) } - /// Deletes only a previously discovered temporary inode after an exclusive - /// cooperative root lock and inode revalidation, then fsyncs its shard. + /// Deletes a revalidated stale inode under an exclusive lock, then fsyncs. pub fn delete_stale_temp(&self, candidate: StaleTemp) -> Result<(), ArtifactError> { let root = self.root()?; let _lock = RootLock::exclusive(root)?; @@ -453,22 +438,433 @@ impl ArtifactStore { } } -fn reconciliation_cursor( - root_dev: u64, - root_ino: u64, - namespace: u8, - shard: u8, - position: i64, -) -> ReconciliationCursor { +fn reconciliation_cursor(root_dev: u64, root_ino: u64) -> ReconciliationCursor { ReconciliationCursor { root_dev, root_ino, + namespace: FINAL_NAMESPACE, + shard: 0, + state: ReconciliationCursorState::OpenNamespace, + } +} + +fn reconciliation_page( + mut report: ReconciliationScan, + entries: Vec, + cursor: ReconciliationCursor, + stop: ReconciliationScanStop, +) -> (ReconciliationScan, Vec) { + report.stop = stop; + report.continuation = Some(cursor); + (report, entries) +} + +fn reconciliation_step( + root: &crate::store::Root, + cursor: &mut ReconciliationCursor, + state: ReconciliationCursorState, + report: &mut ReconciliationScan, + entries: &mut Vec, +) -> Result<(), (ReconciliationCursorState, ArtifactError)> { + match state { + ReconciliationCursorState::OpenNamespace => { + count_traversal_syscall(report); + match open_traversal_dir(root.fd.as_raw_fd(), namespace_name(cursor.namespace)) { + Ok(namespace) => { + cursor.state = ReconciliationCursorState::ValidateNamespace(namespace); + } + Err(ArtifactError::NotFound) => advance_namespace(cursor), + Err(ArtifactError::UnsafeRoot) => { + report.unsafe_entries += 1; + advance_namespace(cursor); + } + Err(error) => { + return Err((ReconciliationCursorState::OpenNamespace, error)); + } + } + } + ReconciliationCursorState::ValidateNamespace(namespace) => { + count_traversal_syscall(report); + match validate_traversal_dir(&namespace) { + Ok(_) => { + cursor.state = ReconciliationCursorState::OpenShard(namespace); + } + Err(ArtifactError::UnsafeRoot) => { + report.unsafe_entries += 1; + advance_namespace(cursor); + } + Err(error) => { + return Err(( + ReconciliationCursorState::ValidateNamespace(namespace), + error, + )); + } + } + } + ReconciliationCursorState::OpenShard(namespace) => { + let shard_name = format!("{:02x}", cursor.shard); + count_traversal_syscall(report); + match open_traversal_dir(namespace.as_raw_fd(), shard_name.as_bytes()) { + Ok(shard) => { + cursor.state = ReconciliationCursorState::ValidateShard { namespace, shard }; + } + Err(ArtifactError::NotFound) => advance_reconciliation_shard(cursor, namespace), + Err(ArtifactError::UnsafeRoot) => { + report.unsafe_entries += 1; + advance_reconciliation_shard(cursor, namespace); + } + Err(error) => { + return Err((ReconciliationCursorState::OpenShard(namespace), error)); + } + } + } + ReconciliationCursorState::ValidateShard { namespace, shard } => { + count_traversal_syscall(report); + match validate_traversal_dir(&shard) { + Ok(stat) => { + cursor.state = read_shard_state( + namespace, + shard, + ShardIdentity(stat.st_dev, stat.st_ino), + vec![0; 8192], + 0, + 0, + 0, + ); + } + Err(ArtifactError::UnsafeRoot) => { + report.unsafe_entries += 1; + advance_reconciliation_shard(cursor, namespace); + } + Err(error) => { + return Err(( + ReconciliationCursorState::ValidateShard { namespace, shard }, + error, + )); + } + } + } + ReconciliationCursorState::ReadShard { + namespace, + shard, + shard_dev, + shard_ino, + mut buffer, + mut offset, + mut filled, + mut cookie, + } => { + if offset == filled { + count_traversal_syscall(report); + match checkpointed("reconciliation_getdents", || { + getdents64(shard.as_raw_fd(), &mut buffer) + }) { + Ok(0) => advance_reconciliation_shard(cursor, namespace), + Ok(count) => { + offset = 0; + filled = count; + cursor.state = read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + offset, + filled, + cookie, + ); + } + Err(error) => { + return Err(( + ReconciliationCursorState::SeekShard { + namespace, + shard, + shard_dev, + shard_ino, + buffer, + cookie, + }, + error, + )); + } + } + return Ok(()); + } + + let (name, after, after_cookie) = match parse_dirent(&buffer[..filled], offset) { + Ok(entry) => entry, + Err(error) => { + return Err(( + read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + offset, + filled, + cookie, + ), + error, + )); + } + }; + if name == b"." || name == b".." { + offset = after; + cookie = after_cookie; + cursor.state = read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + offset, + filled, + cookie, + ); + return Ok(()); + } + let shard_name = format!("{:02x}", cursor.shard); + let Ok(name) = String::from_utf8(name) else { + report.scanned += 1; + report.malformed += 1; + offset = after; + cookie = after_cookie; + cursor.state = read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + offset, + filled, + cookie, + ); + return Ok(()); + }; + if !valid_final_name(&name, &shard_name) { + report.scanned += 1; + if cursor.namespace != FINAL_NAMESPACE || !valid_temp_name(&name) { + report.malformed += 1; + } + offset = after; + cookie = after_cookie; + cursor.state = read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + offset, + filled, + cookie, + ); + return Ok(()); + } + + #[cfg(debug_assertions)] + let _ = crate::test_support::checkpoint("reconciliation_before_stat"); + count_traversal_syscall(report); + let stat = match checkpointed("reconciliation_stat", || { + record_traversal_call(); + nofollow_stat(shard.as_raw_fd(), &name) + }) { + Ok(stat) => Some(stat), + Err(ArtifactError::NotFound) => None, + Err(ArtifactError::UnsafeRoot) => { + report.unsafe_entries += 1; + None + } + Err(error) => { + return Err(( + read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + offset, + filled, + cookie, + ), + error, + )); + } + }; + report.scanned += 1; + offset = after; + cookie = after_cookie; + if let Some(stat) = stat { + if check_file(&stat).is_err() { + report.unsafe_entries += 1; + } else { + let candidate_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: candidate_namespace, + shard: shard_name, + shard_dev, + shard_ino, + name, + dev: stat.st_dev, + ino: stat.st_ino, + modified_seconds: stat.st_mtime, + modified_nanoseconds: stat.st_mtime_nsec, + }); + } + } + cursor.state = read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + offset, + filled, + cookie, + ); + } + ReconciliationCursorState::SeekShard { + namespace, + shard, + shard_dev, + shard_ino, + buffer, + cookie, + } => { + count_traversal_syscall(report); + record_traversal_call(); + if unsafe { libc::lseek(shard.as_raw_fd(), cookie, libc::SEEK_SET) } != cookie { + return Err(( + ReconciliationCursorState::SeekShard { + namespace, + shard, + shard_dev, + shard_ino, + buffer, + cookie, + }, + ArtifactError::Storage, + )); + } + cursor.state = read_shard_state( + namespace, + shard, + ShardIdentity(shard_dev, shard_ino), + buffer, + 0, + 0, + cookie, + ); + } + ReconciliationCursorState::Transition => unreachable!("temporary scanner state"), + } + Ok(()) +} + +fn count_traversal_syscall(report: &mut ReconciliationScan) { + report.traversal_syscalls += 1; +} + +fn read_shard_state( + namespace: OwnedFd, + shard: OwnedFd, + shard_identity: ShardIdentity, + buffer: Vec, + offset: usize, + filled: usize, + cookie: i64, +) -> ReconciliationCursorState { + let ShardIdentity(shard_dev, shard_ino) = shard_identity; + ReconciliationCursorState::ReadShard { namespace, shard, - position, + shard_dev, + shard_ino, + buffer, + offset, + filled, + cookie, } } +fn open_traversal_dir(parent: i32, name: &[u8]) -> Result { + let name = crate::store::c_name(name)?; + record_traversal_call(); + let raw = unsafe { + libc::openat( + parent, + name.as_ptr(), + libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC | libc::O_NOFOLLOW, + ) + }; + if raw < 0 { + match unsafe { *libc::__errno_location() } { + libc::ENOENT => Err(ArtifactError::NotFound), + libc::EACCES | libc::ELOOP | libc::ENOTDIR => Err(ArtifactError::UnsafeRoot), + _ => Err(ArtifactError::Storage), + } + } else { + Ok(unsafe { OwnedFd::from_raw_fd(raw) }) + } +} + +fn validate_traversal_dir(directory: &OwnedFd) -> Result { + record_traversal_call(); + let stat = stat_fd(directory.as_raw_fd())?; + if stat.st_uid != unsafe { libc::geteuid() } + || (stat.st_mode & libc::S_IFMT) != libc::S_IFDIR + || (stat.st_mode & 0o777) != 0o700 + { + Err(ArtifactError::UnsafeRoot) + } else { + Ok(stat) + } +} + +fn getdents64(directory: i32, buffer: &mut [u8]) -> Result { + record_traversal_call(); + let count = unsafe { + libc::syscall( + libc::SYS_getdents64, + directory, + buffer.as_mut_ptr(), + buffer.len(), + ) + }; + if count < 0 { + Err(ArtifactError::Storage) + } else { + usize::try_from(count).map_err(|_| ArtifactError::Storage) + } +} + +fn parse_dirent(buffer: &[u8], offset: usize) -> Result<(Vec, usize, i64), ArtifactError> { + const HEADER: usize = 19; + if buffer.len().saturating_sub(offset) < HEADER { + return Err(ArtifactError::Storage); + } + let record_length = usize::from(u16::from_ne_bytes([ + buffer[offset + 16], + buffer[offset + 17], + ])); + let after = offset + .checked_add(record_length) + .filter(|after| record_length >= HEADER && *after <= buffer.len()) + .ok_or(ArtifactError::Storage)?; + let raw_name = &buffer[offset + HEADER..after]; + let end = raw_name + .iter() + .position(|byte| *byte == 0) + .ok_or(ArtifactError::Storage)?; + let cookie = i64::from_ne_bytes( + buffer[offset + 8..offset + 16] + .try_into() + .map_err(|_| ArtifactError::Storage)?, + ); + Ok((raw_name[..end].to_vec(), after, cookie)) +} + fn namespace_name(namespace: u8) -> &'static [u8] { match namespace { FINAL_NAMESPACE => b"sha256", @@ -477,22 +873,19 @@ fn namespace_name(namespace: u8) -> &'static [u8] { } } -fn advance_shard(cursor: &mut ReconciliationCursor) { - debug_assert_ne!(cursor.shard, u8::MAX); - cursor.shard += 1; - cursor.position = 0; +fn advance_reconciliation_shard(cursor: &mut ReconciliationCursor, namespace: OwnedFd) { + if cursor.shard == u8::MAX { + advance_namespace(cursor); + } else { + cursor.shard += 1; + cursor.state = ReconciliationCursorState::OpenShard(namespace); + } } 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; + cursor.state = ReconciliationCursorState::OpenNamespace; } fn ensure_reconciliation_root( @@ -530,10 +923,8 @@ fn revalidate_candidate( 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. +/// After cross-directory rename, only canonical name and inode remain common; +/// the quarantine shard cannot equal the capability's recorded final shard. fn revalidate_moved_candidate( candidate: &ReconciliationCandidate, shard: i32, @@ -561,9 +952,7 @@ fn fsync_mutation( } } -/// 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. +/// Attempts both directory fsyncs; either failure leaves a retryable ambiguity. fn fsync_quarantine_dirs( final_shard: &std::os::fd::OwnedFd, quarantine_shard: &std::os::fd::OwnedFd, @@ -581,132 +970,6 @@ 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, - report: &mut TempScan, -) -> Result, ArtifactError> { - let dot = std::ffi::CString::new(".").map_err(|_| ArtifactError::Storage)?; - let raw = unsafe { - libc::openat( - parent, - dot.as_ptr(), - libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC, - ) - }; - if raw < 0 { - return Err(ArtifactError::Storage); - } - let directory = unsafe { libc::fdopendir(raw) }; - if directory.is_null() { - unsafe { - libc::close(raw); - } - return Err(ArtifactError::Storage); - } - let mut names = Vec::new(); - loop { - if *remaining == 0 { - break; - } - unsafe { - *libc::__errno_location() = 0; - } - let entry = unsafe { libc::readdir(directory) }; - if entry.is_null() { - if unsafe { *libc::__errno_location() } != 0 { - unsafe { - libc::closedir(directory); - } - return Err(ArtifactError::Storage); - } - break; - } - let name = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes(); - if name == b"." || name == b".." { - continue; - } - report.scanned += 1; - *remaining -= 1; - if let Ok(name) = std::str::from_utf8(name) { - names.push(name.to_owned()); - } - } - unsafe { - libc::closedir(directory); - } - Ok(names) -} -fn valid_temp_name(name: &str) -> bool { - let Some(rest) = name.strip_prefix(ArtifactStore::temp_prefix()) else { - return false; - }; - let mut fields = rest.split('-'); - let (Some(nonce), Some(pid), Some(sequence), None) = - (fields.next(), fields.next(), fields.next(), fields.next()) - else { - return false; - }; - nonce.len() == 32 - && nonce.bytes().all(is_lower_hex) - && !pid.is_empty() - && pid.bytes().all(|byte| byte.is_ascii_digit()) - && !sequence.is_empty() - && sequence.bytes().all(|byte| byte.is_ascii_digit()) -} fn is_lower_hex(byte: u8) -> bool { byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte) } @@ -719,3 +982,8 @@ fn nofollow_stat(parent: i32, name: &str) -> Result { Err(crate::store::classify_errno()) } } + +fn record_traversal_call() { + #[cfg(debug_assertions)] + crate::test_support::record_traversal_call(); +} diff --git a/crates/crank-artifacts/src/lib.rs b/crates/crank-artifacts/src/lib.rs index ce1d1f9..39d475a 100644 --- a/crates/crank-artifacts/src/lib.rs +++ b/crates/crank-artifacts/src/lib.rs @@ -12,6 +12,7 @@ mod housekeeping; mod legacy; mod model; mod store; +mod temp_scan; #[cfg(debug_assertions)] #[doc(hidden)] @@ -21,6 +22,7 @@ pub use error::ArtifactError; pub use housekeeping::{ReconciliationCandidate, ReconciliationCursor, StaleTemp, TempScan}; pub use model::{ ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, ReconciliationMutation, - ReconciliationNamespace, ReconciliationScan, RegisteredArtifact, StoredArtifact, + ReconciliationNamespace, ReconciliationScan, ReconciliationScanStop, RegisteredArtifact, + StoredArtifact, }; pub use store::ArtifactStore; diff --git a/crates/crank-artifacts/src/model.rs b/crates/crank-artifacts/src/model.rs index b0eb9b7..912fa9c 100644 --- a/crates/crank-artifacts/src/model.rs +++ b/crates/crank-artifacts/src/model.rs @@ -85,17 +85,32 @@ pub enum ReconciliationNamespace { Quarantine, } +/// Why a bounded reconciliation page stopped. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ReconciliationScanStop { + Complete, + ScanBudget, + ResultLimit, + Retryable, +} + /// 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 { + /// Non-dot directory entries classified by this page. pub scanned: usize, + /// Traversal syscall budget consumed by this page. It equals syscall + /// attempts in production; debug fault injection may stop an attempt just + /// before the syscall. Root validation and locking are fixed overhead. + pub traversal_syscalls: usize, pub malformed: usize, pub unsafe_entries: usize, pub final_entries: usize, pub quarantined_entries: usize, + pub stop: ReconciliationScanStop, pub continuation: Option, } @@ -103,10 +118,12 @@ impl ReconciliationScan { pub(crate) fn empty(continuation: Option) -> Self { Self { scanned: 0, + traversal_syscalls: 0, malformed: 0, unsafe_entries: 0, final_entries: 0, quarantined_entries: 0, + stop: ReconciliationScanStop::Complete, continuation, } } diff --git a/crates/crank-artifacts/src/temp_scan.rs b/crates/crank-artifacts/src/temp_scan.rs new file mode 100644 index 0000000..21700dd --- /dev/null +++ b/crates/crank-artifacts/src/temp_scan.rs @@ -0,0 +1,78 @@ +use crate::{ArtifactError, ArtifactStore, TempScan}; + +pub(crate) fn valid_temp_name(name: &str) -> bool { + let Some(rest) = name.strip_prefix(ArtifactStore::temp_prefix()) else { + return false; + }; + let mut fields = rest.split('-'); + let (Some(nonce), Some(pid), Some(sequence), None) = + (fields.next(), fields.next(), fields.next(), fields.next()) + else { + return false; + }; + nonce.len() == 32 + && nonce + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + && !pid.is_empty() + && pid.bytes().all(|byte| byte.is_ascii_digit()) + && !sequence.is_empty() + && sequence.bytes().all(|byte| byte.is_ascii_digit()) +} + +pub(crate) fn list_names( + parent: i32, + remaining: &mut usize, + report: &mut TempScan, +) -> Result, ArtifactError> { + let dot = std::ffi::CString::new(".").map_err(|_| ArtifactError::Storage)?; + let raw = unsafe { + libc::openat( + parent, + dot.as_ptr(), + libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC, + ) + }; + if raw < 0 { + return Err(ArtifactError::Storage); + } + let directory = unsafe { libc::fdopendir(raw) }; + if directory.is_null() { + unsafe { + libc::close(raw); + } + return Err(ArtifactError::Storage); + } + let mut names = Vec::new(); + loop { + if *remaining == 0 { + break; + } + unsafe { + *libc::__errno_location() = 0; + } + let entry = unsafe { libc::readdir(directory) }; + if entry.is_null() { + if unsafe { *libc::__errno_location() } != 0 { + unsafe { + libc::closedir(directory); + } + return Err(ArtifactError::Storage); + } + break; + } + let name = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes(); + if name == b"." || name == b".." { + continue; + } + report.scanned += 1; + *remaining -= 1; + if let Ok(name) = std::str::from_utf8(name) { + names.push(name.to_owned()); + } + } + unsafe { + libc::closedir(directory); + } + Ok(names) +} diff --git a/crates/crank-artifacts/src/test_support.rs b/crates/crank-artifacts/src/test_support.rs index 6f2342b..beef1da 100644 --- a/crates/crank-artifacts/src/test_support.rs +++ b/crates/crank-artifacts/src/test_support.rs @@ -1,10 +1,28 @@ //! Test-only crash and fault control for syscall-coupled storage checkpoints. use std::{ - sync::{Condvar, Mutex, OnceLock}, + collections::BTreeMap, + sync::{ + Condvar, Mutex, OnceLock, + atomic::{AtomicUsize, Ordering}, + }, time::Duration, }; +static TRAVERSAL_CALLS: AtomicUsize = AtomicUsize::new(0); + +pub fn reset_traversal_calls() { + TRAVERSAL_CALLS.store(0, Ordering::SeqCst); +} + +pub fn traversal_calls() -> usize { + TRAVERSAL_CALLS.load(Ordering::SeqCst) +} + +pub(crate) fn record_traversal_call() { + TRAVERSAL_CALLS.fetch_add(1, Ordering::SeqCst); +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum FaultAction { Exit, @@ -13,7 +31,8 @@ pub enum FaultAction { } struct State { - checkpoint: Option<(String, FaultAction)>, + checkpoint: Option<(String, FaultAction, usize)>, + hits: BTreeMap, held: bool, } fn slot() -> &'static (Mutex, Condvar) { @@ -22,6 +41,7 @@ fn slot() -> &'static (Mutex, Condvar) { ( Mutex::new(State { checkpoint: None, + hits: BTreeMap::new(), held: false, }), Condvar::new(), @@ -30,10 +50,25 @@ fn slot() -> &'static (Mutex, Condvar) { } pub fn set_checkpoint(stage: impl Into, action: FaultAction) { + set_checkpoint_on_hit(stage, action, 1); +} + +pub fn set_checkpoint_on_hit(stage: impl Into, action: FaultAction, hit: usize) { + assert!(hit > 0, "checkpoint hit is one-based"); + let (lock, _) = slot(); + let mut state = lock.lock().expect("artifact test checkpoint lock poisoned"); + state.checkpoint = Some((stage.into(), action, hit)); + state.hits.clear(); +} + +pub fn checkpoint_hits(stage: &str) -> usize { let (lock, _) = slot(); lock.lock() .expect("artifact test checkpoint lock poisoned") - .checkpoint = Some((stage.into(), action)); + .hits + .get(stage) + .copied() + .unwrap_or(0) } pub fn clear_checkpoint() { @@ -55,10 +90,17 @@ pub fn wait_until_held(timeout: Duration) -> bool { pub(crate) fn checkpoint(stage: &str) -> Option { let (lock, wake) = slot(); let mut state = lock.lock().expect("artifact test checkpoint lock poisoned"); - let action = state - .checkpoint - .as_ref() - .and_then(|(expected, action)| (expected == stage).then_some(*action)); + let configured = state.checkpoint.clone(); + let action = configured.and_then(|(expected, action, target_hit)| { + if expected != stage { + return None; + } + let hit = state.hits.entry(stage.to_owned()).or_insert(0); + if *hit == 0 { + *hit = 1; + } + (*hit == target_hit).then_some(action) + }); if action == Some(FaultAction::Hold) { state.held = true; wake.notify_all(); @@ -73,9 +115,13 @@ pub(crate) fn checkpoint(stage: &str) -> Option { pub(crate) fn action(stage: &str) -> Option { let (lock, _) = slot(); - let state = lock.lock().expect("artifact test checkpoint lock poisoned"); + let mut state = lock.lock().expect("artifact test checkpoint lock poisoned"); + *state.hits.entry(stage.to_owned()).or_insert(0) += 1; + let hit = state.hits[stage]; state .checkpoint .as_ref() - .and_then(|(expected, action)| (expected == stage).then_some(*action)) + .and_then(|(expected, action, target_hit)| { + (expected == stage && hit == *target_hit).then_some(*action) + }) } diff --git a/crates/crank-artifacts/tests/reconciliation.rs b/crates/crank-artifacts/tests/reconciliation.rs index 903dee5..58baf5b 100644 --- a/crates/crank-artifacts/tests/reconciliation.rs +++ b/crates/crank-artifacts/tests/reconciliation.rs @@ -1,28 +1,28 @@ use std::{ fs, - os::unix::fs::PermissionsExt, + os::unix::fs::{MetadataExt, PermissionsExt}, path::PathBuf, - process::Command, - sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }, - thread, - time::Duration, + sync::atomic::{AtomicU64, Ordering}, }; +#[cfg(debug_assertions)] +use std::{process::Command, sync::Arc, thread, time::Duration}; + use crank_artifacts::{ ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace, + ReconciliationScanStop, }; #[cfg(debug_assertions)] use crank_artifacts::test_support::{ - FaultAction, clear_checkpoint, set_checkpoint, wait_until_held, + FaultAction, checkpoint_hits, clear_checkpoint, reset_traversal_calls, set_checkpoint, + set_checkpoint_on_hit, traversal_calls, wait_until_held, }; #[cfg(debug_assertions)] use std::sync::{Mutex, OnceLock}; static NEXT_ROOT: AtomicU64 = AtomicU64::new(0); +const FULL_SCAN_BUDGET: usize = 2048; struct TestRoot(PathBuf); @@ -61,16 +61,31 @@ fn paginates_final_entries_without_disclosing_locations() { let mut continuation = None; let mut candidates = Vec::new(); + let mut final_entries = 0; + let mut scanned = 0; loop { let (report, mut page) = store.scan_reconciliation(continuation, 1, 1).unwrap(); + assert_eq!(report.traversal_syscalls, 1); assert!(report.scanned <= 1); + assert_eq!(report.final_entries, page.len()); + assert_eq!(report.quarantined_entries, 0); assert!(page.len() <= 1); + scanned += report.scanned; + final_entries += report.final_entries; + if let Some(cursor) = report.continuation.as_ref() { + assert_eq!(format!("{cursor:?}"), "ReconciliationCursor(..)"); + } + for candidate in &page { + assert_eq!(format!("{candidate:?}"), "ReconciliationCandidate(..)"); + } candidates.append(&mut page); continuation = report.continuation; if continuation.is_none() { break; } } + assert_eq!(scanned, 2); + assert_eq!(final_entries, 2); assert_eq!(candidates.len(), 2); assert!( candidates @@ -83,8 +98,14 @@ fn paginates_final_entries_without_disclosing_locations() { ReconciliationMutation::Quarantined ); } - let (_, quarantined_entries) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (report, quarantined_entries) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); assert_eq!(quarantined_entries.len(), 2); + assert_eq!(report.scanned, 2); + assert_eq!(report.final_entries, 0); + assert_eq!(report.quarantined_entries, 2); + assert_eq!(report.stop, ReconciliationScanStop::Complete); assert!( quarantined_entries .iter() @@ -121,13 +142,98 @@ fn classifies_malformed_and_unsafe_entries_with_a_usable_page() { ); 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); + let (report, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.scanned, 5); + assert_eq!(report.malformed, 1); + assert_eq!(report.unsafe_entries, 3); + assert_eq!(report.final_entries, 1); + assert_eq!(report.quarantined_entries, 0); assert_eq!(candidates.len(), 1); assert!(report.continuation.is_none()); } +#[test] +fn skips_a_valid_publish_temp_without_classifying_it_as_malformed() { + let root = TestRoot::new("valid-temp"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reconciliation with concurrent temp").unwrap(); + let shard = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]); + fs::write( + shard.join(".crank-artifact-tmp-v1-0123456789abcdef0123456789abcdef-7-9"), + b"in-flight", + ) + .unwrap(); + + let (report, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.scanned, 2); + assert_eq!(report.malformed, 0); + assert_eq!(report.final_entries, 1); + assert_eq!(candidates.len(), 1); +} + +#[test] +fn unsafe_canonical_shard_does_not_starve_later_shards() { + let root = TestRoot::new("unsafe-shard"); + let store = ArtifactStore::open(&root.0).unwrap(); + let sha = root.0.join("sha256"); + fs::create_dir(&sha).unwrap(); + fs::set_permissions(&sha, fs::Permissions::from_mode(0o700)).unwrap(); + let unsafe_shard = sha.join("00"); + fs::create_dir(&unsafe_shard).unwrap(); + fs::set_permissions(&unsafe_shard, fs::Permissions::from_mode(0o000)).unwrap(); + let shard = sha.join("ff"); + fs::create_dir(&shard).unwrap(); + fs::set_permissions(&shard, fs::Permissions::from_mode(0o700)).unwrap(); + let name = "f".repeat(64); + fs::write(shard.join(&name), b"later valid entry").unwrap(); + fs::set_permissions(shard.join(&name), fs::Permissions::from_mode(0o400)).unwrap(); + + let (report, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.stop, ReconciliationScanStop::Complete); + assert_eq!(report.unsafe_entries, 1); + assert_eq!(report.final_entries, 1); + assert!(candidates.iter().any(|candidate| { + candidate.namespace() == ReconciliationNamespace::Final + && format!("{candidate:?}") == "ReconciliationCandidate(..)" + })); + fs::set_permissions(unsafe_shard, fs::Permissions::from_mode(0o700)).unwrap(); +} + +#[test] +#[cfg(debug_assertions)] +fn traversal_never_exceeds_the_exact_syscall_budget() { + let root = TestRoot::new("syscall-budget"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"bounded traversal syscall accounting").unwrap(); + + for budget in 0..=8 { + reset_traversal_calls(); + let (report, _) = store.scan_reconciliation(None, budget, 8).unwrap(); + assert!(report.traversal_syscalls <= budget, "budget={budget}"); + assert_eq!(report.traversal_syscalls, budget, "budget={budget}"); + assert_eq!(traversal_calls(), report.traversal_syscalls); + assert_eq!(report.stop, ReconciliationScanStop::ScanBudget); + } + + reset_traversal_calls(); + let (report, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 1) + .unwrap(); + assert_eq!(candidates.len(), 1); + assert_eq!(report.stop, ReconciliationScanStop::ResultLimit); + assert!(report.traversal_syscalls <= FULL_SCAN_BUDGET); + assert_eq!(traversal_calls(), report.traversal_syscalls); +} + #[test] fn rejects_a_cursor_from_another_store() { let first_root = TestRoot::new("foreign-cursor-first"); @@ -157,7 +263,7 @@ fn disappearing_entry_after_readdir_is_skipped() { .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)); + let scan = thread::spawn(move || scan_store.scan_reconciliation(None, FULL_SCAN_BUDGET, 8)); assert!(wait_until_held(Duration::from_secs(2))); fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap(); fs::remove_file(&path).unwrap(); @@ -172,7 +278,9 @@ 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 (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let candidate = candidates.into_iter().next().unwrap(); assert_eq!( @@ -188,7 +296,9 @@ fn quarantine_and_delete_are_idempotent() { ReconciliationMutation::AlreadyQuarantined ); - let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let quarantined = candidates .into_iter() .find(|candidate| candidate.namespace() == ReconciliationNamespace::Quarantine) @@ -212,7 +322,9 @@ 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 (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let original = candidates.into_iter().next().unwrap(); assert_eq!( store.quarantine_reconciliation(original.clone()).unwrap(), @@ -227,7 +339,9 @@ fn quarantine_preserves_old_inode_when_the_digest_is_republished() { store.read(&stored.artifact_ref).unwrap(), b"reconciliation no-clobber" ); - let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); assert_eq!(candidates.len(), 2); assert!( candidates @@ -241,6 +355,42 @@ fn quarantine_preserves_old_inode_when_the_digest_is_republished() { ); } +#[test] +fn quarantine_refuses_to_replace_an_unrelated_inode() { + let root = TestRoot::new("unrelated-collision"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"canonical final survives collision").unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + let digest = stored.artifact_ref.digest_hex(); + let quarantine = root.0.join("quarantine"); + let shard = quarantine.join(&digest[..2]); + fs::create_dir(&quarantine).unwrap(); + fs::create_dir(&shard).unwrap(); + fs::set_permissions(&quarantine, fs::Permissions::from_mode(0o700)).unwrap(); + fs::set_permissions(&shard, fs::Permissions::from_mode(0o700)).unwrap(); + let collision = shard.join(digest); + fs::write(&collision, b"unrelated quarantine inode").unwrap(); + fs::set_permissions(&collision, fs::Permissions::from_mode(0o400)).unwrap(); + let collision_ino = fs::metadata(&collision).unwrap().ino(); + + assert_eq!( + store.quarantine_reconciliation(candidate), + Err(ArtifactError::UnsafeRoot) + ); + assert_eq!( + store.read(&stored.artifact_ref).unwrap(), + b"canonical final survives collision" + ); + assert_eq!(fs::read(collision).unwrap(), b"unrelated quarantine inode"); + assert_eq!( + fs::metadata(shard.join(digest)).unwrap().ino(), + collision_ino + ); +} + #[test] fn delete_fails_closed_for_a_replaced_or_hardlinked_quarantined_inode() { let root = TestRoot::new("quarantine-replaced"); @@ -248,11 +398,15 @@ fn delete_fails_closed_for_a_replaced_or_hardlinked_quarantined_inode() { let stored = store .put(b"reconciliation quarantined replacement") .unwrap(); - let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); store .quarantine_reconciliation(candidates.into_iter().next().unwrap()) .unwrap(); - let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let candidate = candidates.into_iter().next().unwrap(); let path = root .0 @@ -274,6 +428,40 @@ fn delete_fails_closed_for_a_replaced_or_hardlinked_quarantined_inode() { assert!(sibling.exists()); } +#[test] +fn delete_revalidates_a_post_scan_hardlink_without_inode_replacement() { + let root = TestRoot::new("quarantine-hardlink"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"same quarantined inode gains a link").unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 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()); + let sibling = path.with_extension("link"); + let original_ino = fs::metadata(&path).unwrap().ino(); + fs::hard_link(&path, &sibling).unwrap(); + + assert_eq!( + store.delete_quarantined_reconciliation(candidate), + Err(ArtifactError::UnsafeRoot) + ); + assert_eq!(fs::metadata(&path).unwrap().nlink(), 2); + assert_eq!(fs::metadata(&path).unwrap().ino(), original_ino); + assert!(path.exists()); + assert!(sibling.exists()); +} + #[test] #[cfg(debug_assertions)] fn delete_fsync_ambiguity_is_retryable_and_recovers_as_absent() { @@ -281,11 +469,15 @@ fn delete_fsync_ambiguity_is_retryable_and_recovers_as_absent() { 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(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); store .quarantine_reconciliation(candidates.into_iter().next().unwrap()) .unwrap(); - let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let candidate = candidates.into_iter().next().unwrap(); set_checkpoint("reconciliation_delete_fsync", FaultAction::Fail); assert_eq!( @@ -306,7 +498,9 @@ 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 (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let candidate = candidates.into_iter().next().unwrap(); let path = root .0 @@ -328,36 +522,175 @@ fn rejects_a_replaced_final_inode() { #[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(); + for stage in [ + "reconciliation_quarantine_source_fsync", + "reconciliation_quarantine_destination_fsync", + ] { + let root = TestRoot::new(stage); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(stage.as_bytes()).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + set_checkpoint(stage, FaultAction::Fail); + assert_eq!( + store.quarantine_reconciliation(candidate.clone()).unwrap(), + ReconciliationMutation::Retryable, + "stage={stage}" + ); + assert_eq!( + checkpoint_hits("reconciliation_quarantine_source_fsync"), + 1, + "stage={stage}" + ); + assert_eq!( + checkpoint_hits("reconciliation_quarantine_destination_fsync"), + 1, + "stage={stage}" + ); + clear_checkpoint(); + assert_eq!( + store.quarantine_reconciliation(candidate).unwrap(), + ReconciliationMutation::AlreadyQuarantined, + "stage={stage}" + ); + } +} + +#[test] +#[cfg(debug_assertions)] +fn mutation_syscall_failures_preserve_retryable_state() { + let _guard = fault_guard(); + + let rename_root = TestRoot::new("rename-fail"); + let rename_store = ArtifactStore::open(&rename_root.0).unwrap(); + let stored = rename_store.put(b"rename failure bytes").unwrap(); + let (_, candidates) = rename_store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let candidate = candidates.into_iter().next().unwrap(); - set_checkpoint("reconciliation_quarantine_source_fsync", FaultAction::Fail); + set_checkpoint("reconciliation_quarantine_rename", FaultAction::Fail); assert_eq!( - store.quarantine_reconciliation(candidate.clone()).unwrap(), - ReconciliationMutation::Retryable + rename_store.quarantine_reconciliation(candidate.clone()), + Err(ArtifactError::Storage) + ); + assert_eq!( + rename_store.read(&stored.artifact_ref).unwrap(), + b"rename failure bytes" ); clear_checkpoint(); assert_eq!( - store.quarantine_reconciliation(candidate).unwrap(), - ReconciliationMutation::AlreadyQuarantined + rename_store.quarantine_reconciliation(candidate).unwrap(), + ReconciliationMutation::Quarantined + ); + + let delete_root = TestRoot::new("unlink-fail"); + let delete_store = ArtifactStore::open(&delete_root.0).unwrap(); + delete_store.put(b"unlink failure bytes").unwrap(); + let (_, candidates) = delete_store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + delete_store + .quarantine_reconciliation(candidates.into_iter().next().unwrap()) + .unwrap(); + let (_, candidates) = delete_store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + let candidate = candidates.into_iter().next().unwrap(); + set_checkpoint("reconciliation_delete_unlink", FaultAction::Fail); + assert_eq!( + delete_store.delete_quarantined_reconciliation(candidate.clone()), + Err(ArtifactError::Storage) + ); + clear_checkpoint(); + assert_eq!( + delete_store + .delete_quarantined_reconciliation(candidate) + .unwrap(), + ReconciliationMutation::Deleted ); } #[test] #[cfg(debug_assertions)] -fn storage_fault_during_stat_propagates() { +fn storage_fault_returns_a_retryable_page_before_the_failed_entry() { 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) - )); + let sha = root.0.join("sha256"); + let shard = sha.join("aa"); + fs::create_dir(&sha).unwrap(); + fs::create_dir(&shard).unwrap(); + fs::set_permissions(&sha, fs::Permissions::from_mode(0o700)).unwrap(); + fs::set_permissions(&shard, fs::Permissions::from_mode(0o700)).unwrap(); + for suffix in ['1', '2'] { + let path = shard.join(format!("aa{}", suffix.to_string().repeat(62))); + fs::write(&path, b"valid opaque candidate").unwrap(); + fs::set_permissions(path, fs::Permissions::from_mode(0o400)).unwrap(); + } + + set_checkpoint_on_hit("reconciliation_stat", FaultAction::Fail, 2); + let (report, mut candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.stop, ReconciliationScanStop::Retryable); + assert_eq!(report.scanned, 1); + assert_eq!(report.final_entries, 1); + assert_eq!(candidates.len(), 1); + let continuation = report.continuation; clear_checkpoint(); + + let (report, resumed) = store + .scan_reconciliation(continuation, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.stop, ReconciliationScanStop::Complete); + assert_eq!(report.scanned, 1); + assert_eq!(report.final_entries, 1); + candidates.extend(resumed); + assert_eq!(candidates.len(), 2); + for candidate in candidates { + assert_eq!( + store.quarantine_reconciliation(candidate).unwrap(), + ReconciliationMutation::Quarantined + ); + } + let (report, quarantined) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.final_entries, 0); + assert_eq!(report.quarantined_entries, 2); + assert_eq!(quarantined.len(), 2); +} + +#[test] +#[cfg(debug_assertions)] +fn getdents_retry_restores_the_pre_call_directory_cookie() { + let _guard = fault_guard(); + let root = TestRoot::new("getdents-storage"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"getdents retry bytes").unwrap(); + + set_checkpoint("reconciliation_getdents", FaultAction::Fail); + let (report, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.stop, ReconciliationScanStop::Retryable); + assert!(candidates.is_empty()); + clear_checkpoint(); + + let (report, candidates) = store + .scan_reconciliation(report.continuation, 1, 8) + .unwrap(); + assert_eq!(report.stop, ReconciliationScanStop::ScanBudget); + assert_eq!(report.traversal_syscalls, 1); + assert!(candidates.is_empty()); + + let (report, candidates) = store + .scan_reconciliation(report.continuation, FULL_SCAN_BUDGET, 8) + .unwrap(); + assert_eq!(report.stop, ReconciliationScanStop::Complete); + assert_eq!(candidates.len(), 1); } #[test] @@ -370,7 +703,9 @@ fn crash_windows_are_recoverable() { FaultAction::Exit, ); let store = ArtifactStore::open(PathBuf::from(root)).unwrap(); - let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); let candidate = candidates.into_iter().next().unwrap(); if action == "quarantine" { let _ = store.quarantine_reconciliation(candidate); @@ -391,7 +726,9 @@ fn crash_windows_are_recoverable() { 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(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); store .quarantine_reconciliation(candidates.into_iter().next().unwrap()) .unwrap(); @@ -406,7 +743,9 @@ fn crash_windows_are_recoverable() { assert_eq!(status.code(), Some(86), "stage={stage}"); let store = ArtifactStore::open(&root.0).unwrap(); - let (_, candidates) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, candidates) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); for candidate in candidates { match candidate.namespace() { ReconciliationNamespace::Final => { @@ -417,7 +756,9 @@ fn crash_windows_are_recoverable() { } } } - let (_, remaining) = store.scan_reconciliation(None, 512, 8).unwrap(); + let (_, remaining) = store + .scan_reconciliation(None, FULL_SCAN_BUDGET, 8) + .unwrap(); assert!(remaining.is_empty(), "stage={stage}"); assert_eq!( store