From 07327e7c63f2895261449dbbe75ffb588b0792de Mon Sep 17 00:00:00 2001 From: Terry <186405119+tlq5l@users.noreply.github.com> Date: Tue, 20 Oct 2026 00:29:24 +0700 Subject: [PATCH] fix(cache): restore cached outputs atomically so concurrent cache hits don't fail Fixes #846 Co-authored-by: Claude Sonnet 5.5 Co-authored-by: Claude Opus 5.5 --- CHANGELOG.md | 1 + crates/vt/Cargo.toml | 4 +- crates/vt/src/session/cache/archive.rs | 343 ++++++++++++++++++++++++- 3 files changed, 343 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bca0d5da..a00b0709c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # Changelog +- **Fixed** Tasks that run at the same time and hit the same cache entry no longer intermittently fail with `File exists (os error 17)` while restoring cached outputs ([#846](https://github.com/voidzero-dev/vite-task/issues/846), [#866](https://github.com/voidzero-dev/vite-task/pull/866)). - **Fixed** A mistyped or misplaced setting in a task now gets an error that names it. For example, `cache: { outputs: [...] }` now fails with an `unknown field` error that names `outputs` and lists the accepted settings, instead of `data did not match any variant of untagged enum UserTaskDefinition` ([vite-plus#2635](https://github.com/voidzero-dev/vite-plus/issues/2635), [#815](https://github.com/voidzero-dev/vite-task/pull/815)). - **Fixed** In cached tasks on Linux and macOS, programs that pass `-1` as a directory descriptor to calls such as `openat` no longer abort. For example, `ps -p ` from procps-ng 4.0.6 or later aborted when the process had already exited ([#862](https://github.com/voidzero-dev/vite-task/pull/862)). - **Fixed** `vp run` no longer panics before running a script that contains non-ASCII characters, such as `echo こんにちは`, and no longer cuts off such commands where it displays them ([#861](https://github.com/voidzero-dev/vite-task/pull/861)). diff --git a/crates/vt/Cargo.toml b/crates/vt/Cargo.toml index 1be50ccaf..c8ee84c25 100644 --- a/crates/vt/Cargo.toml +++ b/crates/vt/Cargo.toml @@ -33,6 +33,7 @@ serde_json = { workspace = true } supports-color = { workspace = true } thiserror = { workspace = true } tar = { workspace = true } +tempfile = { workspace = true } tokio = { workspace = true, features = [ "rt-multi-thread", "io-std", @@ -66,9 +67,6 @@ vt_workspace = { workspace = true } wax = { workspace = true } zstd = { workspace = true } -[dev-dependencies] -tempfile = { workspace = true } - [target.'cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))'.dependencies] fspy = { workspace = true } diff --git a/crates/vt/src/session/cache/archive.rs b/crates/vt/src/session/cache/archive.rs index dc110e29c..6c9f9650f 100644 --- a/crates/vt/src/session/cache/archive.rs +++ b/crates/vt/src/session/cache/archive.rs @@ -3,6 +3,7 @@ use std::{ fs::File, io::{self, Read}, + path::Component, }; use vt_path::{AbsolutePath, RelativePathBuf}; @@ -48,7 +49,9 @@ pub fn create_output_archive( /// Extract a tar.zst archive, restoring files relative to workspace root. /// -/// Parent directories are created automatically. Existing files are overwritten. +/// Parent directories are created automatically. Existing files are replaced +/// atomically, so concurrent extractions of the same archive into one +/// workspace don't fail and never expose a partly written file. /// /// # Errors /// @@ -61,7 +64,67 @@ pub fn extract_output_archive( let decoder = zstd::Decoder::new(file)?; let mut archive = tar::Archive::new(decoder); - archive.unpack(workspace_root.as_path())?; + for entry in archive.entries()? { + let mut entry = entry?; + if entry.header().entry_type().is_file() { + unpack_file_atomically(&mut entry, workspace_root)?; + } else { + entry.unpack_in(workspace_root.as_path())?; + } + } + Ok(()) +} + +/// Unpack a regular file entry by writing a temporary file next to the target +/// and renaming it over the target. +/// +/// `tar` removes an existing target and then creates it with `create_new`, so +/// two extractions of the same file can make one of them fail with +/// `File exists`. The rename can't fail that way, and whichever extraction +/// renames last wins with identical bytes. +fn unpack_file_atomically( + entry: &mut tar::Entry<'_, impl Read>, + workspace_root: &AbsolutePath, +) -> anyhow::Result<()> { + let rel_path = entry.path()?.into_owned(); + if !rel_path.components().all(|component| matches!(component, Component::Normal(_))) { + return Err(anyhow::Error::msg(vt_str::format!( + "archive entry `{}` is not a path inside the workspace", + rel_path.display(), + ))); + } + let target = workspace_root.as_path().join(&rel_path); + let parent = target.parent().expect("a joined entry path has a parent"); + // Like `Entry::unpack_in`, refuse entries that resolve outside the workspace + // through a symlinked directory, and check before creating missing directories. + let canonical_root = workspace_root.as_path().canonicalize()?; + #[expect(clippy::disallowed_types, reason = "checks std paths at the filesystem boundary")] + let ensure_inside_workspace = |dir: &std::path::Path| -> anyhow::Result<()> { + if dir.canonicalize()?.starts_with(&canonical_root) { + return Ok(()); + } + Err(anyhow::Error::msg(vt_str::format!( + "archive entry `{}` resolves outside the workspace", + rel_path.display(), + ))) + }; + let mut deepest_existing = parent; + while deepest_existing.symlink_metadata().is_err() + && let Some(ancestor) = deepest_existing.parent() + { + deepest_existing = ancestor; + } + ensure_inside_workspace(deepest_existing)?; + std::fs::create_dir_all(parent)?; + ensure_inside_workspace(parent)?; + // The temporary path is removed on drop, so a failed unpack or rename leaves nothing behind. + // Plain `fs::rename` rather than `TempPath::persist`. On Windows both call `MoveFileExW`, + // which fails with access denied when the target is open or being replaced at the same + // time, but only std then retries with a POSIX-semantics rename. That replaces a target + // a reader has open with `FILE_SHARE_DELETE`, which std passes for every file it opens. + let temp_path = tempfile::NamedTempFile::new_in(parent)?.into_temp_path(); + entry.unpack(&temp_path)?; + std::fs::rename(&temp_path, &target)?; Ok(()) } @@ -84,10 +147,286 @@ pub fn check_output_archive(reader: impl Read) -> io::Result<()> { #[cfg(test)] mod tests { + use std::{ + sync::{ + Barrier, + atomic::{AtomicBool, Ordering}, + }, + thread, + }; + use vt_path::AbsolutePathBuf; use super::*; + #[test] + fn concurrent_extractions_into_one_workspace_succeed() { + let tmp = tempfile::TempDir::new().unwrap(); + let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap(); + let outputs: Vec = (0..20) + .map(|i| RelativePathBuf::new(vt_str::format!("dist/sub/out{i}.txt")).unwrap()) + .collect(); + for (i, output) in outputs.iter().enumerate() { + let path = dir.join(output); + std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap(); + std::fs::write(path.as_path(), vt_str::format!("content {i}\n").repeat(2000)).unwrap(); + } + let archive_path = dir.join("output.tar.zst"); + create_output_archive(&dir, &outputs, &archive_path).unwrap(); + + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + std::fs::create_dir(workspace.as_path()).unwrap(); + for _ in 0..50 { + let barrier = Barrier::new(2); + thread::scope(|scope| { + for _ in 0..2 { + scope.spawn(|| { + barrier.wait(); + extract_output_archive(&workspace, &archive_path).unwrap(); + }); + } + }); + for (i, output) in outputs.iter().enumerate() { + assert_eq!( + std::fs::read_to_string(workspace.join(output).as_path()).unwrap(), + vt_str::format!("content {i}\n").repeat(2000), + ); + } + } + } + + #[test] + fn extraction_replaces_existing_files_and_leaves_nothing_behind_on_failure() { + let tmp = tempfile::TempDir::new().unwrap(); + let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap(); + let output = RelativePathBuf::new("dist/out.txt").unwrap(); + let path = dir.join(&output); + std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap(); + // Incompressible content, so that truncating the archive cuts into the entry. + let content: Vec = + (0..100_000_u32).map(|i| (i.wrapping_mul(2_654_435_761) >> 24) as u8).collect(); + std::fs::write(path.as_path(), &content).unwrap(); + let archive_path = dir.join("output.tar.zst"); + create_output_archive(&dir, &[output], &archive_path).unwrap(); + + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + let restored = workspace.join("dist/out.txt"); + std::fs::create_dir_all(restored.as_path().parent().unwrap()).unwrap(); + std::fs::write(restored.as_path(), "stale").unwrap(); + extract_output_archive(&workspace, &archive_path).unwrap(); + assert_eq!(std::fs::read(restored.as_path()).unwrap(), content); + + let archive = std::fs::read(archive_path.as_path()).unwrap(); + let truncated = dir.join("truncated.tar.zst"); + std::fs::write(truncated.as_path(), &archive[..archive.len() / 2]).unwrap(); + let fresh = AbsolutePathBuf::new(tmp.path().join("fresh")).unwrap(); + std::fs::create_dir(fresh.as_path()).unwrap(); + assert!(extract_output_archive(&fresh, &truncated).is_err()); + // Neither the target nor a temporary file may be left behind. + let left_behind = + std::fs::read_dir(fresh.join("dist").as_path()).map_or(0, Iterator::count); + assert_eq!(left_behind, 0); + } + + #[test] + fn failed_rename_leaves_no_temporary_file() { + let tmp = tempfile::TempDir::new().unwrap(); + let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap(); + let output = RelativePathBuf::new("dist/out.txt").unwrap(); + let path = dir.join(&output); + std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap(); + std::fs::write(path.as_path(), "built\n").unwrap(); + let archive_path = dir.join("output.tar.zst"); + create_output_archive(&dir, &[output], &archive_path).unwrap(); + + // A non-empty directory at the target makes the final rename fail. + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + let blocker = workspace.join("dist/out.txt/child"); + std::fs::create_dir_all(blocker.as_path()).unwrap(); + assert!(extract_output_archive(&workspace, &archive_path).is_err()); + let names: Vec<_> = std::fs::read_dir(workspace.join("dist").as_path()) + .unwrap() + .map(|entry| entry.unwrap().file_name()) + .collect(); + assert_eq!(names, ["out.txt"]); + } + + #[test] + fn readers_never_see_a_missing_or_partial_file_while_restores_run() { + const FILES: usize = 8; + let expected = |i: usize| vt_str::format!("content {i}\n").repeat(20_000); + + let tmp = tempfile::TempDir::new().unwrap(); + let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap(); + let outputs: Vec = (0..FILES) + .map(|i| RelativePathBuf::new(vt_str::format!("dist/out{i}.txt")).unwrap()) + .collect(); + for (i, output) in outputs.iter().enumerate() { + let path = dir.join(output); + std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap(); + std::fs::write(path.as_path(), expected(i)).unwrap(); + } + let archive_path = dir.join("output.tar.zst"); + create_output_archive(&dir, &outputs, &archive_path).unwrap(); + + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + std::fs::create_dir(workspace.as_path()).unwrap(); + // The files exist before the readers start, as after an earlier cache hit. + extract_output_archive(&workspace, &archive_path).unwrap(); + + let done = AtomicBool::new(false); + thread::scope(|scope| { + let readers: Vec<_> = (0..2) + .map(|_| { + scope.spawn(|| { + let mut reads = 0_u32; + // Read at least once more after the restores finished. + loop { + let finished = done.load(Ordering::SeqCst); + for (i, output) in outputs.iter().enumerate() { + // `fs::read` fails if the file is missing. + let bytes = std::fs::read(workspace.join(output).as_path()) + .unwrap_or_else(|err| panic!("{output}: {err}")); + assert!(bytes == expected(i).as_bytes(), "{output}: partial file"); + reads += 1; + } + if finished { + return reads; + } + } + }) + }) + .collect(); + let restorers: Vec<_> = (0..2) + .map(|_| { + scope.spawn(|| { + for _ in 0..30 { + extract_output_archive(&workspace, &archive_path).unwrap(); + } + }) + }) + .collect(); + for restorer in restorers { + restorer.join().unwrap(); + } + done.store(true, Ordering::SeqCst); + for reader in readers { + assert!(reader.join().unwrap() > 0); + } + }); + } + + /// A tar.zst archive with one regular file entry whose path is written to + /// the header verbatim (`tar::Builder` itself refuses `..`). + fn archive_with_raw_path(path: &str) -> Vec { + let mut header = tar::Header::new_gnu(); + header.as_old_mut().name[..path.len()].copy_from_slice(path.as_bytes()); + header.set_size(5); + header.set_mode(0o644); + header.set_entry_type(tar::EntryType::Regular); + header.set_cksum(); + let mut builder = tar::Builder::new(Vec::new()); + builder.append(&header, &b"evil\n"[..]).unwrap(); + zstd::encode_all(builder.into_inner().unwrap().as_slice(), 0).unwrap() + } + + #[test] + fn entries_outside_the_workspace_are_rejected() { + let tmp = tempfile::TempDir::new().unwrap(); + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + std::fs::create_dir(workspace.as_path()).unwrap(); + let outside = tmp.path().join("evil.txt"); + let absolute = outside.to_str().unwrap().to_owned(); + + for (i, path) in + ["../evil.txt", "dist/../../evil.txt", absolute.as_str()].into_iter().enumerate() + { + let archive_path = + AbsolutePathBuf::new(tmp.path().join(vt_str::format!("evil{i}.tar.zst").as_str())) + .unwrap(); + std::fs::write(archive_path.as_path(), archive_with_raw_path(path)).unwrap(); + assert!( + extract_output_archive(&workspace, &archive_path).is_err(), + "`{path}` must be rejected", + ); + assert!(!outside.exists(), "`{path}` escaped the workspace"); + } + } + + #[test] + fn entries_through_a_symlinked_directory_out_of_the_workspace_are_rejected() { + let tmp = tempfile::TempDir::new().unwrap(); + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + let outside = tmp.path().join("outside"); + std::fs::create_dir(&outside).unwrap(); + std::fs::create_dir(workspace.as_path()).unwrap(); + #[cfg(unix)] + std::os::unix::fs::symlink(&outside, workspace.join("dist").as_path()).unwrap(); + #[cfg(windows)] + std::os::windows::fs::symlink_dir(&outside, workspace.join("dist").as_path()).unwrap(); + + let archive_path = AbsolutePathBuf::new(tmp.path().join("output.tar.zst")).unwrap(); + // Neither the file nor the missing `sub` directory may be created outside. + for path in ["dist/evil.txt", "dist/sub/evil.txt"] { + std::fs::write(archive_path.as_path(), archive_with_raw_path(path)).unwrap(); + assert!(extract_output_archive(&workspace, &archive_path).is_err(), "{path}"); + assert_eq!(std::fs::read_dir(&outside).unwrap().count(), 0, "{path}"); + } + } + + #[test] + fn restored_files_keep_their_modification_time() { + let tmp = tempfile::TempDir::new().unwrap(); + let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap(); + let output = RelativePathBuf::new("dist/out.txt").unwrap(); + let path = dir.join(&output); + std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap(); + std::fs::write(path.as_path(), "built\n").unwrap(); + // A whole second in the past, as tar stores mtime in seconds. + let mtime = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_600_000_000); + File::options().write(true).open(path.as_path()).unwrap().set_modified(mtime).unwrap(); + let archive_path = dir.join("output.tar.zst"); + create_output_archive(&dir, &[output], &archive_path).unwrap(); + + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + std::fs::create_dir(workspace.as_path()).unwrap(); + extract_output_archive(&workspace, &archive_path).unwrap(); + let restored = std::fs::metadata(workspace.join("dist/out.txt").as_path()).unwrap(); + assert_eq!(restored.modified().unwrap(), mtime); + } + + #[cfg(unix)] + #[test] + fn restored_files_keep_their_permissions() { + use std::os::unix::fs::PermissionsExt; + + let tmp = tempfile::TempDir::new().unwrap(); + let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap(); + let modes = [("dist/run.sh", 0o755), ("dist/data.txt", 0o644), ("dist/private.txt", 0o600)]; + let outputs: Vec = + modes.iter().map(|(name, _)| RelativePathBuf::new(*name).unwrap()).collect(); + std::fs::create_dir(dir.join("dist").as_path()).unwrap(); + for ((name, mode), output) in modes.iter().zip(&outputs) { + let path = dir.join(output); + std::fs::write(path.as_path(), name).unwrap(); + std::fs::set_permissions(path.as_path(), std::fs::Permissions::from_mode(*mode)) + .unwrap(); + } + let archive_path = dir.join("output.tar.zst"); + create_output_archive(&dir, &outputs, &archive_path).unwrap(); + + let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap(); + std::fs::create_dir(workspace.as_path()).unwrap(); + // Extract twice: the second run replaces files that already exist. + for _ in 0..2 { + extract_output_archive(&workspace, &archive_path).unwrap(); + for (name, mode) in modes { + let restored = std::fs::metadata(workspace.join(name).as_path()).unwrap(); + assert_eq!(restored.permissions().mode() & 0o777, mode, "{name}"); + } + } + } + #[test] fn check_accepts_only_complete_archives() { let tmp = tempfile::TempDir::new().unwrap();