diff --git a/src/ding/mod.rs b/src/ding/mod.rs index 1b9e4463..b7650104 100644 --- a/src/ding/mod.rs +++ b/src/ding/mod.rs @@ -3292,6 +3292,87 @@ Enter to select · ↑/↓ to navigate · Esc to cancel"; ); } + /// Platforms that cannot hardlink through the open-file descriptor path (macOS fdescfs + /// rejects linkat(AT_SYMLINK_FOLLOW) on /dev/fd/N with EPERM) fall back to a byte-copy + /// archive receipt. The differentiated supersession semantics must hold unchanged on that + /// path, the receipt must carry exactly the validated bytes, and no staging file may leak + /// into the archive. + #[test] + #[cfg(debug_assertions)] + fn archive_copy_fallback_preserves_supersede_ownership_without_staging_leftovers() { + crate::event::TEST_FORCE_ARCHIVE_RECEIPT_COPY.store(true, Ordering::Relaxed); + struct ResetGuard; + impl Drop for ResetGuard { + fn drop(&mut self) { + crate::event::TEST_FORCE_ARCHIVE_RECEIPT_COPY.store(false, Ordering::Relaxed); + } + } + let _guard = ResetGuard; + + let (catalog, inbox) = event_catalog(); + let root = catalog.path(); + let archive = crate::message::archive_dir(&root.join("hetz").join("worker")); + + let failure_filename = emit_ci(root, "failure", true); + let mut seen = HashSet::new(); + let mut pending: VecDeque = new_arrivals(&inbox, &mut seen) + .into_iter() + .map(PendingNotice::message) + .collect(); + assert_eq!(pending.len(), 1); + let failure_bytes = + std::fs::read(inbox.join(&failure_filename)).expect("staged event bytes"); + let failure_text = pending[0].text( + DingContext { + catalog_root: root, + this_host: "hetz", + recipient: "hetz.worker", + }, + &mut None, + ); + + let poker = OwnershipPoker { + pokes: Mutex::new(Vec::new()), + retries: Mutex::new(Vec::new()), + poke_outcomes: Mutex::new(VecDeque::from([PokeOutcome::Staged])), + retry_outcomes: Mutex::new(VecDeque::from([PokeOutcome::Staged])), + }; + flush_in(root, &mut pending, &poker); + + emit_ci(root, "success", true); + pending.extend( + new_arrivals(&inbox, &mut seen) + .into_iter() + .map(PendingNotice::message), + ); + prune_archived_pending(&inbox, &mut pending); + flush_in(root, &mut pending, &poker); + + assert_eq!(pending.len(), 2, "later FIFO work remains blocked"); + assert_eq!( + poker.pokes.lock().unwrap().as_slice(), + [failure_text.as_str()], + "the successor is never pasted on top of a retained payload" + ); + assert!(!inbox.join(&failure_filename).exists(), "head was archived"); + let receipt = std::fs::read(archive.join(&failure_filename)) + .expect("byte-copy archive receipt exists"); + assert_eq!( + receipt, failure_bytes, + "the copy receipt carries exactly the validated bytes" + ); + let staging_leftovers: Vec<_> = std::fs::read_dir(&archive) + .unwrap() + .filter_map(Result::ok) + .map(|entry| entry.file_name().to_string_lossy().into_owned()) + .filter(|name| name.starts_with(".st2-archive-")) + .collect(); + assert!( + staging_leftovers.is_empty(), + "no staging files survive in the archive: {staging_leftovers:?}" + ); + } + #[test] fn archived_not_retained_releases_fifo_without_repasting_owned_notice() { let agent = tempfile::tempdir().unwrap(); diff --git a/src/event.rs b/src/event.rs index 8b955ae3..298e624d 100644 --- a/src/event.rs +++ b/src/event.rs @@ -11,7 +11,7 @@ use std::os::unix::fs::DirBuilderExt as _; use std::os::unix::fs::MetadataExt as _; use std::os::unix::fs::OpenOptionsExt as _; use std::path::Path; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use anyhow::Context as _; use serde::{Deserialize, Serialize}; @@ -672,7 +672,14 @@ fn archive_validated_file( }; if result < 0 { let error = std::io::Error::last_os_error(); - if error.kind() != std::io::ErrorKind::AlreadyExists { + if error.kind() == std::io::ErrorKind::AlreadyExists { + // A receipt for this predecessor already exists; the readback below proves it + // carries exactly the validated bytes. + } else if capability_link_unsupported(&error) { + // The platform cannot hardlink through the open-file descriptor path at all; + // degrade to a byte-copy receipt instead of failing publication. + write_archive_receipt_copy(file, &archive_dir, archive, filename)?; + } else { return Err(error).context("archive the validated predecessor capability"); } } @@ -690,6 +697,77 @@ fn archive_validated_file( Ok(()) } +/// Whether linkat through the open-file capability path is unsupported by the platform rather +/// than a real failure. macOS fdescfs answers linkat(AT_SYMLINK_FOLLOW) on /dev/fd/N with +/// EPERM; ENOSYS/EOPNOTSUPP cover kernels lacking the syscall or its symlink-follow semantics. +/// Everything else stays a hard error so genuine failures (permissions, cross-device, ...) +/// surface instead of being silently degraded to a copy. +fn capability_link_unsupported(error: &std::io::Error) -> bool { + #[cfg(debug_assertions)] + if TEST_FORCE_ARCHIVE_RECEIPT_COPY.load(Ordering::Relaxed) { + return true; + } + matches!( + error.raw_os_error(), + Some(libc::EPERM) | Some(libc::ENOSYS) | Some(libc::EOPNOTSUPP) + ) +} + +/// Debug-only switch letting tests exercise the byte-copy fallback on platforms where the real +/// capability linkat would succeed. Not a supported configuration knob. Flipping this mid-run +/// is safe: the fallback receipt is verified against the validated bytes exactly like the +/// hardlink path, and the same-inode unlink treats a copy receipt as "archived" regardless. +#[cfg(debug_assertions)] +pub(crate) static TEST_FORCE_ARCHIVE_RECEIPT_COPY: AtomicBool = AtomicBool::new(false); + +/// Materialize the archive receipt as a byte copy of the validated file, for platforms that +/// cannot hardlink through the open-file descriptor path. +/// +/// The staged temp keeps a concurrent archiver from observing a partial receipt, and +/// rename_noreplace turns an install race into a no-op: whichever receipt wins, the caller's +/// readback proves it carries exactly the validated bytes. The tradeoff against the hardlink +/// fast path is inode identity -- a crash between the copy and the conditional unlink leaves +/// the retained inbox entry in place until revalidation, which the same-inode checks read as +/// "still present", never as data loss. +fn write_archive_receipt_copy( + file: &File, + archive_dir: &File, + archive: &Path, + filename: &str, +) -> anyhow::Result<()> { + let mut source = file.try_clone()?; + source.rewind()?; + let mut bytes = Vec::new(); + source.read_to_end(&mut bytes)?; + drop(source); + + let staged = archive.join(format!( + ".st2-archive-{}-{}", + std::process::id(), + TMP_COUNTER.fetch_add(1, Ordering::Relaxed) + )); + let mut staged_file = OpenOptions::new() + .read(true) + .write(true) + .create_new(true) + .mode(0o600) + .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) + .open(&staged)?; + staged_file.write_all(&bytes)?; + staged_file.sync_all()?; + drop(staged_file); + + let target = archive.join(filename); + if let Err(error) = crate::catalog_transaction::rename_noreplace(&staged, &target) { + fs::remove_file(&staged).context("remove staging copy after failed receipt install")?; + if error.kind() != std::io::ErrorKind::AlreadyExists { + return Err(error).context("install archived predecessor receipt"); + } + } + archive_dir.sync_all()?; + Ok(()) +} + fn conditional_unlink_same_inode( inbox_file: &File, expected_file: &File,