Skip to content

Commit c66fd7f

Browse files
tlq5lclaude
andcommitted
fix(cache): restore cached outputs atomically so concurrent cache hits don't fail
Fixes #846 Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 8db455e commit c66fd7f

3 files changed

Lines changed: 343 additions & 5 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
# Changelog
22

3+
- **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), [#???](https://github.com/voidzero-dev/vite-task/pull/???)).
34
- **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)).
45
- **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 <pid>` 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)).
56
- **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)).

‎crates/vt/Cargo.toml‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ serde_json = { workspace = true }
3333
supports-color = { workspace = true }
3434
thiserror = { workspace = true }
3535
tar = { workspace = true }
36+
tempfile = { workspace = true }
3637
tokio = { workspace = true, features = [
3738
"rt-multi-thread",
3839
"io-std",
@@ -66,9 +67,6 @@ vt_workspace = { workspace = true }
6667
wax = { workspace = true }
6768
zstd = { workspace = true }
6869

69-
[dev-dependencies]
70-
tempfile = { workspace = true }
71-
7270
[target.'cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))'.dependencies]
7371
fspy = { workspace = true }
7472

‎crates/vt/src/session/cache/archive.rs‎

Lines changed: 341 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
use std::{
44
fs::File,
55
io::{self, Read},
6+
path::Component,
67
};
78

89
use vt_path::{AbsolutePath, RelativePathBuf};
@@ -48,7 +49,9 @@ pub fn create_output_archive(
4849

4950
/// Extract a tar.zst archive, restoring files relative to workspace root.
5051
///
51-
/// Parent directories are created automatically. Existing files are overwritten.
52+
/// Parent directories are created automatically. Existing files are replaced
53+
/// atomically, so concurrent extractions of the same archive into one
54+
/// workspace don't fail and never expose a partly written file.
5255
///
5356
/// # Errors
5457
///
@@ -61,7 +64,67 @@ pub fn extract_output_archive(
6164
let decoder = zstd::Decoder::new(file)?;
6265
let mut archive = tar::Archive::new(decoder);
6366

64-
archive.unpack(workspace_root.as_path())?;
67+
for entry in archive.entries()? {
68+
let mut entry = entry?;
69+
if entry.header().entry_type().is_file() {
70+
unpack_file_atomically(&mut entry, workspace_root)?;
71+
} else {
72+
entry.unpack_in(workspace_root.as_path())?;
73+
}
74+
}
75+
Ok(())
76+
}
77+
78+
/// Unpack a regular file entry by writing a temporary file next to the target
79+
/// and renaming it over the target.
80+
///
81+
/// `tar` removes an existing target and then creates it with `create_new`, so
82+
/// two extractions of the same file can make one of them fail with
83+
/// `File exists`. The rename can't fail that way, and whichever extraction
84+
/// renames last wins with identical bytes.
85+
fn unpack_file_atomically(
86+
entry: &mut tar::Entry<'_, impl Read>,
87+
workspace_root: &AbsolutePath,
88+
) -> anyhow::Result<()> {
89+
let rel_path = entry.path()?.into_owned();
90+
if !rel_path.components().all(|component| matches!(component, Component::Normal(_))) {
91+
return Err(anyhow::Error::msg(vt_str::format!(
92+
"archive entry `{}` is not a path inside the workspace",
93+
rel_path.display(),
94+
)));
95+
}
96+
let target = workspace_root.as_path().join(&rel_path);
97+
let parent = target.parent().expect("a joined entry path has a parent");
98+
// Like `Entry::unpack_in`, refuse entries that resolve outside the workspace
99+
// through a symlinked directory, and check before creating missing directories.
100+
let canonical_root = workspace_root.as_path().canonicalize()?;
101+
#[expect(clippy::disallowed_types, reason = "checks std paths at the filesystem boundary")]
102+
let ensure_inside_workspace = |dir: &std::path::Path| -> anyhow::Result<()> {
103+
if dir.canonicalize()?.starts_with(&canonical_root) {
104+
return Ok(());
105+
}
106+
Err(anyhow::Error::msg(vt_str::format!(
107+
"archive entry `{}` resolves outside the workspace",
108+
rel_path.display(),
109+
)))
110+
};
111+
let mut deepest_existing = parent;
112+
while deepest_existing.symlink_metadata().is_err()
113+
&& let Some(ancestor) = deepest_existing.parent()
114+
{
115+
deepest_existing = ancestor;
116+
}
117+
ensure_inside_workspace(deepest_existing)?;
118+
std::fs::create_dir_all(parent)?;
119+
ensure_inside_workspace(parent)?;
120+
// The temporary path is removed on drop, so a failed unpack or rename leaves nothing behind.
121+
// Plain `fs::rename` rather than `TempPath::persist`. On Windows both call `MoveFileExW`,
122+
// which fails with access denied when the target is open or being replaced at the same
123+
// time, but only std then retries with a POSIX-semantics rename. That replaces a target
124+
// a reader has open with `FILE_SHARE_DELETE`, which std passes for every file it opens.
125+
let temp_path = tempfile::NamedTempFile::new_in(parent)?.into_temp_path();
126+
entry.unpack(&temp_path)?;
127+
std::fs::rename(&temp_path, &target)?;
65128
Ok(())
66129
}
67130

@@ -84,10 +147,286 @@ pub fn check_output_archive(reader: impl Read) -> io::Result<()> {
84147

85148
#[cfg(test)]
86149
mod tests {
150+
use std::{
151+
sync::{
152+
Barrier,
153+
atomic::{AtomicBool, Ordering},
154+
},
155+
thread,
156+
};
157+
87158
use vt_path::AbsolutePathBuf;
88159

89160
use super::*;
90161

162+
#[test]
163+
fn concurrent_extractions_into_one_workspace_succeed() {
164+
let tmp = tempfile::TempDir::new().unwrap();
165+
let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap();
166+
let outputs: Vec<RelativePathBuf> = (0..20)
167+
.map(|i| RelativePathBuf::new(vt_str::format!("dist/sub/out{i}.txt")).unwrap())
168+
.collect();
169+
for (i, output) in outputs.iter().enumerate() {
170+
let path = dir.join(output);
171+
std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap();
172+
std::fs::write(path.as_path(), vt_str::format!("content {i}\n").repeat(2000)).unwrap();
173+
}
174+
let archive_path = dir.join("output.tar.zst");
175+
create_output_archive(&dir, &outputs, &archive_path).unwrap();
176+
177+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
178+
std::fs::create_dir(workspace.as_path()).unwrap();
179+
for _ in 0..50 {
180+
let barrier = Barrier::new(2);
181+
thread::scope(|scope| {
182+
for _ in 0..2 {
183+
scope.spawn(|| {
184+
barrier.wait();
185+
extract_output_archive(&workspace, &archive_path).unwrap();
186+
});
187+
}
188+
});
189+
for (i, output) in outputs.iter().enumerate() {
190+
assert_eq!(
191+
std::fs::read_to_string(workspace.join(output).as_path()).unwrap(),
192+
vt_str::format!("content {i}\n").repeat(2000),
193+
);
194+
}
195+
}
196+
}
197+
198+
#[test]
199+
fn extraction_replaces_existing_files_and_leaves_nothing_behind_on_failure() {
200+
let tmp = tempfile::TempDir::new().unwrap();
201+
let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap();
202+
let output = RelativePathBuf::new("dist/out.txt").unwrap();
203+
let path = dir.join(&output);
204+
std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap();
205+
// Incompressible content, so that truncating the archive cuts into the entry.
206+
let content: Vec<u8> =
207+
(0..100_000_u32).map(|i| (i.wrapping_mul(2_654_435_761) >> 24) as u8).collect();
208+
std::fs::write(path.as_path(), &content).unwrap();
209+
let archive_path = dir.join("output.tar.zst");
210+
create_output_archive(&dir, &[output], &archive_path).unwrap();
211+
212+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
213+
let restored = workspace.join("dist/out.txt");
214+
std::fs::create_dir_all(restored.as_path().parent().unwrap()).unwrap();
215+
std::fs::write(restored.as_path(), "stale").unwrap();
216+
extract_output_archive(&workspace, &archive_path).unwrap();
217+
assert_eq!(std::fs::read(restored.as_path()).unwrap(), content);
218+
219+
let archive = std::fs::read(archive_path.as_path()).unwrap();
220+
let truncated = dir.join("truncated.tar.zst");
221+
std::fs::write(truncated.as_path(), &archive[..archive.len() / 2]).unwrap();
222+
let fresh = AbsolutePathBuf::new(tmp.path().join("fresh")).unwrap();
223+
std::fs::create_dir(fresh.as_path()).unwrap();
224+
assert!(extract_output_archive(&fresh, &truncated).is_err());
225+
// Neither the target nor a temporary file may be left behind.
226+
let left_behind =
227+
std::fs::read_dir(fresh.join("dist").as_path()).map_or(0, Iterator::count);
228+
assert_eq!(left_behind, 0);
229+
}
230+
231+
#[test]
232+
fn failed_rename_leaves_no_temporary_file() {
233+
let tmp = tempfile::TempDir::new().unwrap();
234+
let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap();
235+
let output = RelativePathBuf::new("dist/out.txt").unwrap();
236+
let path = dir.join(&output);
237+
std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap();
238+
std::fs::write(path.as_path(), "built\n").unwrap();
239+
let archive_path = dir.join("output.tar.zst");
240+
create_output_archive(&dir, &[output], &archive_path).unwrap();
241+
242+
// A non-empty directory at the target makes the final rename fail.
243+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
244+
let blocker = workspace.join("dist/out.txt/child");
245+
std::fs::create_dir_all(blocker.as_path()).unwrap();
246+
assert!(extract_output_archive(&workspace, &archive_path).is_err());
247+
let names: Vec<_> = std::fs::read_dir(workspace.join("dist").as_path())
248+
.unwrap()
249+
.map(|entry| entry.unwrap().file_name())
250+
.collect();
251+
assert_eq!(names, ["out.txt"]);
252+
}
253+
254+
#[test]
255+
fn readers_never_see_a_missing_or_partial_file_while_restores_run() {
256+
const FILES: usize = 8;
257+
let expected = |i: usize| vt_str::format!("content {i}\n").repeat(20_000);
258+
259+
let tmp = tempfile::TempDir::new().unwrap();
260+
let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap();
261+
let outputs: Vec<RelativePathBuf> = (0..FILES)
262+
.map(|i| RelativePathBuf::new(vt_str::format!("dist/out{i}.txt")).unwrap())
263+
.collect();
264+
for (i, output) in outputs.iter().enumerate() {
265+
let path = dir.join(output);
266+
std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap();
267+
std::fs::write(path.as_path(), expected(i)).unwrap();
268+
}
269+
let archive_path = dir.join("output.tar.zst");
270+
create_output_archive(&dir, &outputs, &archive_path).unwrap();
271+
272+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
273+
std::fs::create_dir(workspace.as_path()).unwrap();
274+
// The files exist before the readers start, as after an earlier cache hit.
275+
extract_output_archive(&workspace, &archive_path).unwrap();
276+
277+
let done = AtomicBool::new(false);
278+
thread::scope(|scope| {
279+
let readers: Vec<_> = (0..2)
280+
.map(|_| {
281+
scope.spawn(|| {
282+
let mut reads = 0_u32;
283+
// Read at least once more after the restores finished.
284+
loop {
285+
let finished = done.load(Ordering::SeqCst);
286+
for (i, output) in outputs.iter().enumerate() {
287+
// `fs::read` fails if the file is missing.
288+
let bytes = std::fs::read(workspace.join(output).as_path())
289+
.unwrap_or_else(|err| panic!("{output}: {err}"));
290+
assert!(bytes == expected(i).as_bytes(), "{output}: partial file");
291+
reads += 1;
292+
}
293+
if finished {
294+
return reads;
295+
}
296+
}
297+
})
298+
})
299+
.collect();
300+
let restorers: Vec<_> = (0..2)
301+
.map(|_| {
302+
scope.spawn(|| {
303+
for _ in 0..30 {
304+
extract_output_archive(&workspace, &archive_path).unwrap();
305+
}
306+
})
307+
})
308+
.collect();
309+
for restorer in restorers {
310+
restorer.join().unwrap();
311+
}
312+
done.store(true, Ordering::SeqCst);
313+
for reader in readers {
314+
assert!(reader.join().unwrap() > 0);
315+
}
316+
});
317+
}
318+
319+
/// A tar.zst archive with one regular file entry whose path is written to
320+
/// the header verbatim (`tar::Builder` itself refuses `..`).
321+
fn archive_with_raw_path(path: &str) -> Vec<u8> {
322+
let mut header = tar::Header::new_gnu();
323+
header.as_old_mut().name[..path.len()].copy_from_slice(path.as_bytes());
324+
header.set_size(5);
325+
header.set_mode(0o644);
326+
header.set_entry_type(tar::EntryType::Regular);
327+
header.set_cksum();
328+
let mut builder = tar::Builder::new(Vec::new());
329+
builder.append(&header, &b"evil\n"[..]).unwrap();
330+
zstd::encode_all(builder.into_inner().unwrap().as_slice(), 0).unwrap()
331+
}
332+
333+
#[test]
334+
fn entries_outside_the_workspace_are_rejected() {
335+
let tmp = tempfile::TempDir::new().unwrap();
336+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
337+
std::fs::create_dir(workspace.as_path()).unwrap();
338+
let outside = tmp.path().join("evil.txt");
339+
let absolute = outside.to_str().unwrap().to_owned();
340+
341+
for (i, path) in
342+
["../evil.txt", "dist/../../evil.txt", absolute.as_str()].into_iter().enumerate()
343+
{
344+
let archive_path =
345+
AbsolutePathBuf::new(tmp.path().join(vt_str::format!("evil{i}.tar.zst").as_str()))
346+
.unwrap();
347+
std::fs::write(archive_path.as_path(), archive_with_raw_path(path)).unwrap();
348+
assert!(
349+
extract_output_archive(&workspace, &archive_path).is_err(),
350+
"`{path}` must be rejected",
351+
);
352+
assert!(!outside.exists(), "`{path}` escaped the workspace");
353+
}
354+
}
355+
356+
#[test]
357+
fn entries_through_a_symlinked_directory_out_of_the_workspace_are_rejected() {
358+
let tmp = tempfile::TempDir::new().unwrap();
359+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
360+
let outside = tmp.path().join("outside");
361+
std::fs::create_dir(&outside).unwrap();
362+
std::fs::create_dir(workspace.as_path()).unwrap();
363+
#[cfg(unix)]
364+
std::os::unix::fs::symlink(&outside, workspace.join("dist").as_path()).unwrap();
365+
#[cfg(windows)]
366+
std::os::windows::fs::symlink_dir(&outside, workspace.join("dist").as_path()).unwrap();
367+
368+
let archive_path = AbsolutePathBuf::new(tmp.path().join("output.tar.zst")).unwrap();
369+
// Neither the file nor the missing `sub` directory may be created outside.
370+
for path in ["dist/evil.txt", "dist/sub/evil.txt"] {
371+
std::fs::write(archive_path.as_path(), archive_with_raw_path(path)).unwrap();
372+
assert!(extract_output_archive(&workspace, &archive_path).is_err(), "{path}");
373+
assert_eq!(std::fs::read_dir(&outside).unwrap().count(), 0, "{path}");
374+
}
375+
}
376+
377+
#[test]
378+
fn restored_files_keep_their_modification_time() {
379+
let tmp = tempfile::TempDir::new().unwrap();
380+
let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap();
381+
let output = RelativePathBuf::new("dist/out.txt").unwrap();
382+
let path = dir.join(&output);
383+
std::fs::create_dir_all(path.as_path().parent().unwrap()).unwrap();
384+
std::fs::write(path.as_path(), "built\n").unwrap();
385+
// A whole second in the past, as tar stores mtime in seconds.
386+
let mtime = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_600_000_000);
387+
File::options().write(true).open(path.as_path()).unwrap().set_modified(mtime).unwrap();
388+
let archive_path = dir.join("output.tar.zst");
389+
create_output_archive(&dir, &[output], &archive_path).unwrap();
390+
391+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
392+
std::fs::create_dir(workspace.as_path()).unwrap();
393+
extract_output_archive(&workspace, &archive_path).unwrap();
394+
let restored = std::fs::metadata(workspace.join("dist/out.txt").as_path()).unwrap();
395+
assert_eq!(restored.modified().unwrap(), mtime);
396+
}
397+
398+
#[cfg(unix)]
399+
#[test]
400+
fn restored_files_keep_their_permissions() {
401+
use std::os::unix::fs::PermissionsExt;
402+
403+
let tmp = tempfile::TempDir::new().unwrap();
404+
let dir = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap();
405+
let modes = [("dist/run.sh", 0o755), ("dist/data.txt", 0o644), ("dist/private.txt", 0o600)];
406+
let outputs: Vec<RelativePathBuf> =
407+
modes.iter().map(|(name, _)| RelativePathBuf::new(*name).unwrap()).collect();
408+
std::fs::create_dir(dir.join("dist").as_path()).unwrap();
409+
for ((name, mode), output) in modes.iter().zip(&outputs) {
410+
let path = dir.join(output);
411+
std::fs::write(path.as_path(), name).unwrap();
412+
std::fs::set_permissions(path.as_path(), std::fs::Permissions::from_mode(*mode))
413+
.unwrap();
414+
}
415+
let archive_path = dir.join("output.tar.zst");
416+
create_output_archive(&dir, &outputs, &archive_path).unwrap();
417+
418+
let workspace = AbsolutePathBuf::new(tmp.path().join("workspace")).unwrap();
419+
std::fs::create_dir(workspace.as_path()).unwrap();
420+
// Extract twice: the second run replaces files that already exist.
421+
for _ in 0..2 {
422+
extract_output_archive(&workspace, &archive_path).unwrap();
423+
for (name, mode) in modes {
424+
let restored = std::fs::metadata(workspace.join(name).as_path()).unwrap();
425+
assert_eq!(restored.permissions().mode() & 0o777, mode, "{name}");
426+
}
427+
}
428+
}
429+
91430
#[test]
92431
fn check_accepts_only_complete_archives() {
93432
let tmp = tempfile::TempDir::new().unwrap();

0 commit comments

Comments
 (0)