From 3baf986a847280dcd1556d6cb82c2802a57fea09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=B6=BE=E7=80=AC=E6=A1=83=E6=A1=83?= Date: Wed, 4 Nov 2026 12:28:52 +0800 Subject: [PATCH] fix(io): isolate concurrent atomic output writes --- senbei-io/src/atomic.rs | 209 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 198 insertions(+), 11 deletions(-) diff --git a/senbei-io/src/atomic.rs b/senbei-io/src/atomic.rs index cf98683..223621f 100644 --- a/senbei-io/src/atomic.rs +++ b/senbei-io/src/atomic.rs @@ -1,16 +1,203 @@ //! Shared atomic filesystem writes for native orchestration. -use std::path::{Path, PathBuf}; +use std::io::{self, Write}; +use std::path::Path; +use std::thread; +use std::time::Duration; -/// Write `bytes` through a sibling temporary file and replace `dest` only after -/// the complete write succeeds. -pub(crate) fn write_atomic(dest: &Path, bytes: &[u8]) -> std::io::Result<()> { - let mut temporary_name = dest.as_os_str().to_os_string(); - temporary_name.push(".senbei-tmp"); - let temporary = PathBuf::from(temporary_name); - let result = std::fs::write(&temporary, bytes).and_then(|()| std::fs::rename(&temporary, dest)); - if result.is_err() { - let _ = std::fs::remove_file(&temporary); +use tempfile::Builder; + +/// Write `bytes` through a uniquely named temporary file in `dest`'s directory +/// and replace `dest` only after the complete write succeeds. +/// +/// The temporary name is random, so concurrent writers of the same destination +/// do not share one sibling file. An existing `dest` is replaced. On any +/// failure the temporary file is removed and `dest` is left unchanged. +pub(crate) fn write_atomic(dest: &Path, bytes: &[u8]) -> io::Result<()> { + let parent = dest + .parent() + .filter(|parent| !parent.as_os_str().is_empty()); + let parent = parent.unwrap_or_else(|| Path::new(".")); + // Same directory as `dest`: the final replace must stay on one filesystem. + // The temporary is private to this call and is removed if persist is not + // reached, or if persist fails. + let mut temporary = Builder::new() + .prefix(".senbei-") + .suffix(".tmp") + .tempfile_in(parent)?; + temporary.write_all(bytes)?; + temporary.flush()?; + temporary.as_file().sync_all()?; + // `persist` replaces an existing file and, on failure, hands the temporary + // back. Drop deletes that file, so a failed replace cannot leave `dest` + // half-written. Windows can refuse the replace while another handle still + // has `dest` open (the file is mapped, or a concurrent replace has not + // dropped its handle yet). Retry these refusals briefly; a persistent + // permission or sharing failure still leaves the old destination intact. + const ATTEMPTS: u32 = 8; + let mut attempt = 0; + loop { + match temporary.persist(dest) { + Ok(_) => return Ok(()), + Err(error) if replace_in_progress(&error.error) && attempt + 1 < ATTEMPTS => { + thread::sleep(Duration::from_millis(1 << attempt)); + attempt += 1; + temporary = error.file; + } + Err(error) => return Err(error.error), + } + } +} + +/// A replace refused because the destination is momentarily in use. +/// +/// `ERROR_ACCESS_DENIED` (5) is what Windows returns when `MoveFileExW` cannot +/// replace a file another handle still has open. `ERROR_SHARING_VIOLATION` +/// (32) is the same condition reported through the sharing check. These codes +/// can also mean a permanent denial, so retries are bounded and never remove +/// the existing destination to force replacement. +fn replace_in_progress(error: &io::Error) -> bool { + #[cfg(windows)] + { + matches!(error.raw_os_error(), Some(5 | 32)) + } + #[cfg(not(windows))] + { + let _ = error; + false + } +} + +#[cfg(test)] +mod tests { + use std::fs; + use std::sync::{Arc, Barrier}; + use std::thread; + + use super::*; + + fn stray_names(dir: &Path) -> Vec { + fs::read_dir(dir) + .expect("read destination directory") + .map(|entry| { + entry + .expect("directory entry") + .file_name() + .to_string_lossy() + .into_owned() + }) + .filter(|name| name.starts_with(".senbei-") || name.ends_with(".senbei-tmp")) + .collect() + } + + #[test] + fn writes_new_file_and_leaves_no_temporary() { + let dir = tempfile::tempdir().expect("temp dir"); + let dest = dir.path().join("out.bin"); + + write_atomic(&dest, b"payload").expect("write"); + + assert_eq!(fs::read(&dest).unwrap(), b"payload"); + assert!(stray_names(dir.path()).is_empty()); + } + + #[test] + fn overwrites_existing_destination() { + let dir = tempfile::tempdir().expect("temp dir"); + let dest = dir.path().join("out.bin"); + fs::write(&dest, b"previous").expect("seed"); + + write_atomic(&dest, b"replacement").expect("overwrite"); + + assert_eq!(fs::read(&dest).unwrap(), b"replacement"); + assert!(stray_names(dir.path()).is_empty()); + } + + #[test] + fn failure_leaves_existing_destination_and_no_temporary() { + let dir = tempfile::tempdir().expect("temp dir"); + let dest = dir.path().join("out.bin"); + fs::create_dir(&dest).expect("destination directory"); + + let error = write_atomic(&dest, b"replacement").expect_err("cannot replace a directory"); + + assert_ne!(error.kind(), io::ErrorKind::AlreadyExists); + assert!(dest.is_dir()); + assert!(stray_names(dir.path()).is_empty()); + assert!( + stray_names(&dest).is_empty(), + "temporary file was left inside the destination" + ); + } + + #[test] + fn missing_parent_writes_nothing() { + let dir = tempfile::tempdir().expect("temp dir"); + let parent = dir.path().join("missing"); + let dest = parent.join("out.bin"); + + let error = write_atomic(&dest, b"payload").expect_err("missing parent"); + + assert_eq!(error.kind(), io::ErrorKind::NotFound); + assert!(!parent.exists()); + } + + #[test] + fn concurrent_writers_leave_one_complete_output() { + let dir = tempfile::tempdir().expect("temp dir"); + let dest = Arc::new(dir.path().join("out.bin")); + let barrier = Arc::new(Barrier::new(8)); + let mut handles = Vec::new(); + for id in 0..8u8 { + let dest = Arc::clone(&dest); + let barrier = Arc::clone(&barrier); + handles.push(thread::spawn(move || { + barrier.wait(); + let payload = vec![id; 64 * 1024]; + write_atomic(&dest, &payload).expect("concurrent write"); + id + })); + } + let ids: Vec = handles + .into_iter() + .map(|handle| handle.join().expect("writer thread")) + .collect(); + + let written = fs::read(dest.as_path()).expect("read destination"); + let writer = written[0]; + assert!(ids.contains(&writer)); + assert!(written.iter().all(|byte| *byte == writer)); + assert_eq!(written.len(), 64 * 1024); + assert!(stray_names(dir.path()).is_empty()); + } + + #[test] + fn distinct_temporary_names_survive_a_failed_replace() { + // Two writers no longer share `.senbei-tmp`. Each temporary is + // removed when its own replace fails, including when the destination + // cannot be replaced at all. + let dir = tempfile::tempdir().expect("temp dir"); + let dest = dir.path().join("out.bin"); + fs::create_dir(&dest).expect("destination directory"); + let barrier = Arc::new(Barrier::new(2)); + let mut handles = Vec::new(); + for payload in [b"first".as_slice(), b"second".as_slice()] { + let dest = dest.clone(); + let barrier = Arc::clone(&barrier); + let payload = payload.to_vec(); + handles.push(thread::spawn(move || { + barrier.wait(); + write_atomic(&dest, &payload) + })); + } + for handle in handles { + handle + .join() + .expect("writer thread") + .expect_err("directory destination"); + } + assert!(dest.is_dir()); + assert!(stray_names(dir.path()).is_empty()); + assert!(stray_names(&dest).is_empty()); } - result }