sorokin.oleg.edu/ARCADIA-Studio
1
0
Fork 0
Code Issues 37 Pull Requests 2 Packages Projects Releases 9 Wiki Activity

Complete source 6 engineering workflows and preserve release validation evidence

Browse Source
This commit is contained in:
Codex
2026-09-06 21:49:04 -07:00
parent 2b2d2bc18b
commit 4b6b5fc197
290 changed files with 48271 additions and 1791 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/oa-core/src/source_files.rs
+706 -5
View File
@@ -9,6 +9,13 @@
//! may remain after interruption during staging/completed cleanup. They contain
//! no pending transaction unless named `.arcadia-save-ENTRY.journal`. Windows
//! has no portable directory fsync; this is not a power-loss durability promise.
//! Journal v2 records whether writes may have begun with a flushed marker.
//! Recovery of a v2 journal without that marker discards only journal metadata;
//! v1 journals retain their prior rollback rules.
//! Journal publication refuses replacement on Windows and Linux, including a
//! competing empty directory. Linux requires kernel/filesystem support for
//! `renameat2(RENAME_NOREPLACE)` and fails without a fallback when unavailable.
//! Other operating systems retain standard rename semantics.
use crate::{language, source_project::*};
use serde::{Deserialize, Serialize};
use std::{
@@ -22,6 +29,8 @@ use thiserror::Error;
const MAX_PATH_BYTES: usize = 4096;
const MAX_JOURNAL_BYTES: usize = 32 * 1024 * 1024;
const MAX_JOURNAL_FILES: usize = 4096;
const WRITE_STARTED: &str = "write-started";
const WRITE_STARTED_BYTES: &[u8] = b"Arcadia journal v2: model writes may have started\n";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DiskExpectation {
@@ -588,6 +597,127 @@ fn staged_path(journal: &Path, index: usize, suffix: &str) -> PathBuf {
journal.join(format!("{index}.{suffix}"))
}
fn publish_journal(stage: &Path, journal: &Path) -> Result<(), SourceFileError> {
#[cfg(windows)]
{
publish_journal_with_retry(
stage,
journal,
rename_journal_without_replacement,
std::thread::sleep,
)
}
#[cfg(target_os = "linux")]
{
rename_journal_without_replacement(stage, journal).map_err(|error| io_at(journal, error))
}
// Other platforms retain their existing rename behavior; the native
// no-replacement publication guarantee is currently Windows/Linux only.
#[cfg(not(any(windows, target_os = "linux")))]
{
fs::rename(stage, journal).map_err(|error| io_at(journal, error))
}
}
#[cfg(target_os = "linux")]
fn rename_journal_without_replacement(stage: &Path, journal: &Path) -> io::Result<()> {
use std::{ffi::CString, os::unix::ffi::OsStrExt};
let path_bytes = |path: &Path| {
CString::new(path.as_os_str().as_bytes()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
"Journal path contains an interior NUL.",
)
})
};
let (stage, journal) = (path_bytes(stage)?, path_bytes(journal)?);
// Use libc's architecture-specific syscall number, avoiding a dependency on
// the newer glibc renameat2 wrapper. Both NUL-terminated byte buffers remain
// alive for the call. RENAME_NOREPLACE makes even an empty competing directory
// an error. Unsupported kernels/filesystems fail closed: never fall back to
// ordinary rename, whose replacement semantics would lose journal ownership.
let result = unsafe {
libc::syscall(
libc::SYS_renameat2,
libc::AT_FDCWD as libc::c_long,
stage.as_ptr(),
libc::AT_FDCWD as libc::c_long,
journal.as_ptr(),
libc::RENAME_NOREPLACE as libc::c_ulong,
)
};
if result == -1 {
Err(io::Error::last_os_error())
} else {
Ok(())
}
}
#[cfg(windows)]
fn rename_journal_without_replacement(stage: &Path, journal: &Path) -> io::Result<()> {
use std::os::windows::ffi::OsStrExt;
#[link(name = "kernel32")]
unsafe extern "system" {
fn MoveFileExW(existing: *const u16, destination: *const u16, flags: u32) -> i32;
}
let wide = |path: &Path| -> io::Result<Vec<u16>> {
let mut value: Vec<_> = path.as_os_str().encode_wide().collect();
if value.contains(&0) {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Journal path contains an interior NUL.",
));
}
value.push(0);
Ok(value)
};
let (stage, journal) = (wide(stage)?, wide(journal)?);
// Both terminated UTF-16 buffers live through the call. Flags=0 requests
// neither replacement, cross-volume copying nor deferred execution. Unlike
// fs::rename on modern Windows, an existing empty directory is not replaced.
if unsafe { MoveFileExW(stage.as_ptr(), journal.as_ptr(), 0) } == 0 {
Err(io::Error::last_os_error())
} else {
Ok(())
}
}
// Windows scanners may briefly open newly written stage files without delete
// sharing. Retry only publication, before any model file has been replaced.
// Ownership still transfers only after one successful rename. Never overwrite
// a journal that appeared while waiting, and retain the final OS error/path.
#[cfg(any(windows, test))]
fn publish_journal_with_retry(
stage: &Path,
journal: &Path,
mut rename: impl FnMut(&Path, &Path) -> io::Result<()>,
mut pause: impl FnMut(std::time::Duration),
) -> Result<(), SourceFileError> {
let mut delays = [5, 10, 25, 50, 100, 200].into_iter();
loop {
match fs::symlink_metadata(journal) {
Ok(_) => return Err(SourceFileError::RecoveryRequired(journal.into())),
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => return Err(io_at(journal, error)),
}
match rename(stage, journal) {
Ok(()) => return Ok(()),
Err(error) => {
// ERROR_ACCESS_DENIED, ERROR_SHARING_VIOLATION, ERROR_LOCK_VIOLATION.
// A permanent ACL failure is still returned after at most 390 ms.
if matches!(error.raw_os_error(), Some(5 | 32 | 33))
&& let Some(delay) = delays.next()
{
pause(std::time::Duration::from_millis(delay));
continue;
}
return Err(io_at(journal, error));
}
}
}
}
fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), SourceFileError> {
let mut file = File::options()
.write(true)
@@ -699,6 +829,26 @@ fn save_with_hook(
compiled: &CompiledSourceProject,
limits: ProjectLimits,
mut after_replace: impl FnMut(usize) -> io::Result<()>,
) -> Result<DiskProject, SourceFileError> {
save_with_phases(snapshot, compiled, limits, |phase| match phase {
SavePhase::AfterReplace(count) => after_replace(count),
_ => Ok(()),
})
}
enum SavePhase {
BeforePublish,
AfterPublish,
AfterWriteMarker,
AfterReplace(usize),
BeforeRetire,
}
fn save_with_phases(
snapshot: &DiskProject,
compiled: &CompiledSourceProject,
limits: ProjectLimits,
mut phase: impl FnMut(SavePhase) -> io::Result<()>,
) -> Result<DiskProject, SourceFileError> {
check_snapshot(snapshot, limits)?;
let checked = compile_bundle(&compiled.bundle, limits).map_err(|diagnostics| {
@@ -739,7 +889,7 @@ fn save_with_hook(
// Stable dependency order, with the project entry committed last.
entries.sort_by_key(|file| (file.path == snapshot.bundle.entry, file.path.clone()));
let journal = Journal {
version: 1,
version: 2,
entry: snapshot.bundle.entry.clone(),
files: entries,
};
@@ -775,20 +925,46 @@ fn save_with_hook(
}
write_synced(&stage.path().join("journal.json"), &metadata)?;
sync_directory(stage.path())?;
phase(SavePhase::BeforePublish).map_err(|error| io_at(&journal_dir, error))?;
// Publishing a complete directory makes interrupted staging harmless.
fs::rename(stage.path(), &journal_dir).map_err(|error| io_at(&journal_dir, error))?;
publish_journal(stage.path(), &journal_dir)?;
let _ = stage.keep(); // The renamed directory now belongs to recovery.
sync_directory(&snapshot.root)?;
phase(SavePhase::AfterPublish).map_err(|error| io_at(&journal_dir, error))?;
// Publication may have waited while another writer completed the same
// candidate. Nothing has been written by this save: never interpret their
// matching after-bytes as ours and roll them back. The absent v2 marker
// also preserves this distinction after interruption or retirement failure.
if let Err(error) = check_conflicts(&snapshot.root, &snapshot.expectations) {
let retirement = phase(SavePhase::BeforeRetire)
.map_err(|error| io_at(&journal_dir, error))
.and_then(|()| retire_journal(&snapshot.root, &journal_dir, &journal));
return match retirement {
Ok(()) => Err(error),
Err(retirement) => Err(SourceFileError::RecoveryFailed {
save: error.to_string(),
recovery: retirement.to_string(),
journal: journal_dir,
}),
};
}
write_synced(&journal_dir.join(WRITE_STARTED), WRITE_STARTED_BYTES)?;
sync_directory(&journal_dir)?;
phase(SavePhase::AfterWriteMarker).map_err(|error| io_at(&journal_dir, error))?;
let result = (|| {
let mut current = snapshot.expectations.clone();
let mut replacements = 0;
for file in journal.files.iter().filter(|file| file.write) {
check_conflicts(&snapshot.root, &current)?;
// The first check is above, before the durable write-attempt marker.
// Further checks follow real write attempts and may need rollback.
if replacements != 0 {
check_conflicts(&snapshot.root, &current)?;
}
let bytes = compiled.bundle.files[&file.path].as_bytes();
atomic_write(&snapshot.root, &file.path, bytes)?;
current.insert(file.path.clone(), DiskExpectation::Bytes(bytes.to_vec()));
replacements += 1;
after_replace(replacements)
phase(SavePhase::AfterReplace(replacements))
.map_err(|error| io_at(&snapshot.root.join(&file.path), error))?;
}
check_conflicts(&snapshot.root, &current)?;
@@ -804,6 +980,7 @@ fn save_with_hook(
}),
};
}
phase(SavePhase::BeforeRetire).map_err(|error| io_at(&journal_dir, error))?;
retire_journal(&snapshot.root, &journal_dir, &journal)?;
Ok(saved_snapshot(snapshot, compiled))
}
@@ -833,6 +1010,14 @@ fn remove_journal(directory: &Path, journal: &Journal) -> Result<(), SourceFileE
}
}
}
if journal.version == 2 {
let marker = directory.join(WRITE_STARTED);
match fs::remove_file(&marker) {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => return Err(io_at(&marker, error)),
}
}
let metadata = directory.join("journal.json");
fs::remove_file(&metadata).map_err(|error| io_at(&metadata, error))?;
fs::remove_dir(directory).map_err(|error| io_at(directory, error))
@@ -898,7 +1083,10 @@ fn recover_with_hook(
let bytes = read_journal_file(&metadata_path, MAX_JOURNAL_BYTES)?;
let journal: Journal = serde_json::from_slice(&bytes)
.map_err(|error| SourceFileError::Invalid(format!("Invalid recovery journal: {error}")))?;
if journal.version != 1 || journal.entry != entry || journal.files.len() > limits.max_files {
if !matches!(journal.version, 1 | 2)
|| journal.entry != entry
|| journal.files.len() > limits.max_files
{
return Err(SourceFileError::Invalid(
"Recovery journal version, entry or file budget is invalid.".into(),
));
@@ -908,6 +1096,25 @@ fn recover_with_hook(
let mut folded = BTreeSet::new();
let (mut before_bytes, mut after_bytes) = (0usize, 0usize);
let mut allowed = BTreeSet::from(["journal.json".to_string()]);
let write_started = if journal.version == 2 {
let marker = directory.join(WRITE_STARTED);
match fs::symlink_metadata(&marker) {
Ok(_) => {
if read_journal_file(&marker, WRITE_STARTED_BYTES.len())? != WRITE_STARTED_BYTES {
return Err(SourceFileError::Invalid(
"Invalid journal write-started marker.".into(),
));
}
allowed.insert(WRITE_STARTED.into());
true
}
Err(error) if error.kind() == io::ErrorKind::NotFound => false,
Err(error) => return Err(io_at(&marker, error)),
}
} else {
// Frozen v1 journals predate the marker and retain their rollback rules.
true
};
for (index, file) in journal.files.iter().enumerate() {
disk_path(&root, &file.path)?;
if !folded.insert(file.path.to_lowercase()) {
@@ -952,6 +1159,10 @@ fn recover_with_hook(
));
}
}
if !write_started {
retire_journal(&root, &directory, &journal)?;
return Ok(true);
}
let mut current = BTreeMap::new();
let mut conflicts = Vec::new();
for (path, before) in &original {
@@ -995,6 +1206,342 @@ fn recover_with_hook(
mod transaction_tests {
use super::*;
#[cfg(target_os = "linux")]
#[test]
fn linux_journal_publication_preserves_competing_destinations_and_our_stage() {
use std::os::unix::{fs::MetadataExt, fs::symlink};
for kind in ["empty directory", "nonempty directory", "file", "symlink"] {
let directory = tempfile::tempdir().unwrap();
let stage = directory.path().join("stage");
let journal = directory.path().join("journal");
fs::create_dir(&stage).unwrap();
fs::write(stage.join("journal.json"), b"our complete metadata").unwrap();
// Simulate a competitor arriving after the caller's absence check.
assert!(!journal.exists());
match kind {
"empty directory" => fs::create_dir(&journal).unwrap(),
"nonempty directory" => {
fs::create_dir(&journal).unwrap();
fs::write(journal.join("journal.json"), b"competitor metadata").unwrap();
}
"file" => fs::write(&journal, b"competitor file").unwrap(),
"symlink" => symlink("absent competitor target", &journal).unwrap(),
_ => unreachable!(),
}
let before = fs::symlink_metadata(&journal).unwrap();
let error = publish_journal(&stage, &journal).unwrap_err();
assert!(
matches!(error, SourceFileError::Io { path, source }
if path == journal && source.raw_os_error() == Some(libc::EEXIST)),
"{kind} must refuse publication"
);
let after = fs::symlink_metadata(&journal).unwrap();
assert_eq!((before.dev(), before.ino()), (after.dev(), after.ino()));
assert_eq!(
fs::read(stage.join("journal.json")).unwrap(),
b"our complete metadata"
);
match kind {
"empty directory" => assert_eq!(fs::read_dir(&journal).unwrap().count(), 0),
"nonempty directory" => assert_eq!(
fs::read(journal.join("journal.json")).unwrap(),
b"competitor metadata"
),
"file" => assert_eq!(fs::read(&journal).unwrap(), b"competitor file"),
"symlink" => assert_eq!(
fs::read_link(&journal).unwrap(),
Path::new("absent competitor target")
),
_ => unreachable!(),
}
}
}
#[cfg(target_os = "linux")]
#[test]
fn linux_journal_publication_preserves_raw_path_bytes_and_rejects_nul() {
use std::{ffi::OsStr, os::unix::ffi::OsStrExt};
let directory = tempfile::tempdir().unwrap();
let stage = directory.path().join(OsStr::from_bytes(b"stage-\xff"));
let journal = directory.path().join(OsStr::from_bytes(b"journal-\xfe"));
fs::create_dir(&stage).unwrap();
fs::write(stage.join("journal.json"), b"complete metadata").unwrap();
for (a, b) in [
(Path::new("bad\0path"), journal.as_path()),
(stage.as_path(), Path::new("bad\0path")),
] {
let error = rename_journal_without_replacement(a, b).unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
assert!(stage.join("journal.json").exists());
assert!(!journal.exists());
}
publish_journal(&stage, &journal).unwrap();
assert!(!stage.exists());
assert_eq!(
fs::read(journal.join("journal.json")).unwrap(),
b"complete metadata"
);
}
#[cfg(target_os = "linux")]
#[test]
fn linux_journal_publication_has_one_winner_for_concurrent_stages() {
use std::sync::Barrier;
let directory = tempfile::tempdir().unwrap();
let journal = directory.path().join("journal");
let stages = [
directory.path().join("stage-a"),
directory.path().join("stage-b"),
];
for (i, stage) in stages.iter().enumerate() {
fs::create_dir(stage).unwrap();
fs::write(stage.join("journal.json"), format!("complete writer {i}")).unwrap();
}
let barrier = Barrier::new(2);
let results = std::thread::scope(|scope| {
let journal = &journal;
let barrier = &barrier;
let handles: Vec<_> = stages
.iter()
.map(|stage| {
scope.spawn(move || {
assert!(!journal.exists());
barrier.wait();
publish_journal(stage, journal)
})
})
.collect();
handles
.into_iter()
.map(|handle| handle.join().unwrap())
.collect::<Vec<_>>()
});
assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
for (i, result) in results.into_iter().enumerate() {
match result {
Ok(()) => {
assert!(!stages[i].exists());
assert_eq!(
fs::read_to_string(journal.join("journal.json")).unwrap(),
format!("complete writer {i}")
);
}
Err(SourceFileError::Io { path, source }) => {
assert_eq!(path, journal);
assert_eq!(source.raw_os_error(), Some(libc::EEXIST));
assert_eq!(
fs::read_to_string(stages[i].join("journal.json")).unwrap(),
format!("complete writer {i}")
);
}
Err(error) => panic!("unexpected publication error: {error}"),
}
}
}
#[cfg(target_os = "linux")]
#[test]
#[ignore = "requires strace renameat2 error injection and OA_TEST_RENAMEAT2_ERRNO"]
fn linux_journal_publication_injected_unsupported_error_is_fail_closed() {
let errno: i32 = std::env::var("OA_TEST_RENAMEAT2_ERRNO")
.expect("set OA_TEST_RENAMEAT2_ERRNO to the injected Linux errno")
.parse()
.unwrap();
assert!(matches!(
errno,
libc::ENOSYS | libc::EINVAL | libc::EOPNOTSUPP
));
let (directory, snapshot, compiled) = fixture();
let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
let original: Vec<_> = snapshot
.bundle
.files
.keys()
.map(|path| (path.clone(), fs::read(directory.path().join(path)).unwrap()))
.collect();
let error = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
if matches!(
phase,
SavePhase::AfterPublish | SavePhase::AfterWriteMarker | SavePhase::AfterReplace(_)
) {
panic!("unsupported no-replace publication must stop before ownership transfer");
}
Ok(())
})
.unwrap_err();
assert!(matches!(error, SourceFileError::Io { path, source }
if path == journal && source.raw_os_error() == Some(errno)));
assert!(!journal.exists());
for (path, bytes) in original {
assert_eq!(fs::read(directory.path().join(path)).unwrap(), bytes);
}
check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
}
#[cfg(target_os = "linux")]
#[test]
fn linux_journal_publication_refusal_never_starts_model_writes() {
let (directory, snapshot, compiled) = fixture();
let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
let original: Vec<_> = snapshot
.bundle
.files
.keys()
.map(|path| (path.clone(), fs::read(directory.path().join(path)).unwrap()))
.collect();
let error = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
match phase {
SavePhase::BeforePublish => fs::create_dir(&journal).unwrap(),
SavePhase::AfterPublish
| SavePhase::AfterWriteMarker
| SavePhase::AfterReplace(_) => {
panic!("a rejected publisher must not obtain ownership or write source files")
}
_ => {}
}
Ok(())
})
.unwrap_err();
assert!(matches!(error, SourceFileError::Io { path, source }
if path == journal && source.raw_os_error() == Some(libc::EEXIST)));
assert_eq!(fs::read_dir(&journal).unwrap().count(), 0);
assert!(!journal.join(WRITE_STARTED).exists());
for (path, bytes) in original {
assert_eq!(fs::read(directory.path().join(path)).unwrap(), bytes);
}
check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
}
#[cfg(windows)]
#[test]
fn journal_publication_does_not_replace_empty_target_created_after_precheck() {
let directory = tempfile::tempdir().unwrap();
let stage = directory.path().join("stage");
let journal = directory.path().join("journal");
fs::create_dir(&stage).unwrap();
fs::write(stage.join("journal.json"), b"our journal").unwrap();
let mut called = false;
let result = publish_journal_with_retry(
&stage,
&journal,
|a, b| {
assert!(!called);
called = true;
fs::create_dir(b).unwrap();
rename_journal_without_replacement(a, b)
},
|_| {},
);
assert!(result.is_err());
assert!(called);
assert_eq!(
fs::read(stage.join("journal.json")).unwrap(),
b"our journal"
);
assert_eq!(fs::read_dir(&journal).unwrap().count(), 0);
assert!(
matches!(rename_journal_without_replacement(Path::new("bad\0path"), &journal), Err(error) if error.kind() == io::ErrorKind::InvalidInput)
);
}
#[test]
fn journal_publication_retries_only_transient_windows_errors_before_transferring_ownership() {
let directory = tempfile::tempdir().unwrap();
let stage = directory.path().join("stage");
let journal = directory.path().join("journal");
fs::create_dir(&stage).unwrap();
fs::write(stage.join("journal.json"), b"complete staged metadata").unwrap();
let mut errors = [5, 32, 33].into_iter();
let mut waits = vec![];
publish_journal_with_retry(
&stage,
&journal,
|a, b| {
assert!(a.join("journal.json").exists());
assert!(!b.exists());
match errors.next() {
Some(code) => Err(io::Error::from_raw_os_error(code)),
None => fs::rename(a, b),
}
},
|delay| waits.push(delay.as_millis()),
)
.unwrap();
assert_eq!(waits, [5, 10, 25]);
assert!(!stage.exists());
assert_eq!(
fs::read(journal.join("journal.json")).unwrap(),
b"complete staged metadata"
);
}
#[test]
fn journal_publication_exhaustion_and_permanent_errors_keep_stage_and_report_original_path() {
for (code, expected_calls, expected_wait) in [(5, 7, 390), (123, 1, 0)] {
let directory = tempfile::tempdir().unwrap();
let stage = directory.path().join("stage");
let journal = directory.path().join("journal");
fs::create_dir(&stage).unwrap();
fs::write(stage.join("journal.json"), b"before publication").unwrap();
let (mut calls, mut waited) = (0, 0);
let error = publish_journal_with_retry(
&stage,
&journal,
|_, _| {
calls += 1;
Err(io::Error::from_raw_os_error(code))
},
|delay| waited += delay.as_millis(),
)
.unwrap_err();
assert_eq!((calls, waited), (expected_calls, expected_wait));
assert!(
matches!(error, SourceFileError::Io { path, source } if path == journal && source.raw_os_error() == Some(code))
);
assert_eq!(
fs::read(stage.join("journal.json")).unwrap(),
b"before publication"
);
assert!(!journal.exists());
}
}
#[test]
fn journal_publication_never_replaces_another_journal_that_appears_during_retry() {
let directory = tempfile::tempdir().unwrap();
let stage = directory.path().join("stage");
let journal = directory.path().join("journal");
fs::create_dir(&stage).unwrap();
fs::write(stage.join("journal.json"), b"our staged journal").unwrap();
let mut calls = 0;
let error = publish_journal_with_retry(
&stage,
&journal,
|_, _| {
calls += 1;
Err(io::Error::from_raw_os_error(32))
},
|_| {
fs::create_dir(&journal).unwrap();
fs::write(journal.join("journal.json"), b"other save owns this").unwrap();
},
)
.unwrap_err();
assert_eq!(calls, 1);
assert!(matches!(error, SourceFileError::RecoveryRequired(path) if path == journal));
assert_eq!(
fs::read(journal.join("journal.json")).unwrap(),
b"other save owns this"
);
assert_eq!(
fs::read(stage.join("journal.json")).unwrap(),
b"our staged journal"
);
}
fn fixture() -> (tempfile::TempDir, DiskProject, CompiledSourceProject) {
let directory = tempfile::tempdir().unwrap();
fs::write(
@@ -1014,6 +1561,160 @@ mod transaction_tests {
(directory, snapshot, compiled)
}
#[test]
fn unwritten_journal_never_rolls_back_a_completed_writer_with_identical_after_bytes() {
for interrupt_retirement in [false, true] {
let (directory, snapshot, compiled) = fixture();
let error = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
match phase {
SavePhase::BeforePublish => {
// A competing save finishes during our publication wait.
save_project(&snapshot, &compiled, ProjectLimits::default()).unwrap();
}
SavePhase::BeforeRetire if interrupt_retirement => {
return Err(io::Error::other("injected no-write retirement failure"));
}
SavePhase::AfterWriteMarker | SavePhase::AfterReplace(_) => {
panic!("losing writer must not start writes")
}
_ => {}
}
Ok(())
})
.unwrap_err();
assert!(matches!(
error,
SourceFileError::Conflict(_) | SourceFileError::RecoveryFailed { .. }
));
for (path, bytes) in &compiled.bundle.files {
assert_eq!(
fs::read_to_string(directory.path().join(path)).unwrap(),
*bytes
);
}
let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
assert_eq!(journal.exists(), interrupt_retirement);
if interrupt_retirement {
assert!(!journal.join(WRITE_STARTED).exists());
// Explicit recovery after a failed retirement must also leave
// the completed other writer untouched, even matching after bytes.
assert!(
recover_project(
&directory.path().join("project.arcadia"),
ProjectLimits::default()
)
.unwrap()
);
for (path, bytes) in &compiled.bundle.files {
assert_eq!(
fs::read_to_string(directory.path().join(path)).unwrap(),
*bytes
);
}
}
assert!(!journal.exists());
}
}
#[test]
fn recovery_before_any_write_accepts_v2_marker_and_keeps_v1_rollback_compatibility() {
for version in [1, 2] {
let (directory, snapshot, compiled) = fixture();
let interrupted = std::panic::catch_unwind(|| {
let _ = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
if matches!(phase, SavePhase::AfterWriteMarker) {
panic!("interrupted before atomic write");
}
Ok(())
});
});
assert!(interrupted.is_err());
check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
assert_eq!(
fs::read(journal.join(WRITE_STARTED)).unwrap(),
WRITE_STARTED_BYTES
);
if version == 1 {
// A frozen v1 journal has no marker and means writes may have begun.
let path = journal.join("journal.json");
let mut metadata: Journal =
serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
metadata.version = 1;
fs::write(path, serde_json::to_vec(&metadata).unwrap()).unwrap();
fs::remove_file(journal.join(WRITE_STARTED)).unwrap();
for (path, bytes) in &compiled.bundle.files {
fs::write(directory.path().join(path), bytes).unwrap();
}
}
assert!(
recover_project(
&directory.path().join("project.arcadia"),
ProjectLimits::default()
)
.unwrap()
);
check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
}
}
#[test]
fn interrupted_publication_without_write_marker_never_rolls_back_external_matching_bytes() {
let (directory, snapshot, compiled) = fixture();
let interrupted = std::panic::catch_unwind(|| {
let _ = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
if matches!(phase, SavePhase::AfterPublish) {
panic!("interrupted before write marker");
}
Ok(())
});
});
assert!(interrupted.is_err());
let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
assert!(!journal.join(WRITE_STARTED).exists());
for (path, bytes) in &compiled.bundle.files {
fs::write(directory.path().join(path), bytes).unwrap();
}
assert!(
recover_project(
&directory.path().join("project.arcadia"),
ProjectLimits::default()
)
.unwrap()
);
for (path, bytes) in &compiled.bundle.files {
assert_eq!(
fs::read_to_string(directory.path().join(path)).unwrap(),
*bytes
);
}
assert!(!journal.exists());
}
#[test]
fn malformed_v2_write_marker_refuses_recovery_without_changing_source_bytes() {
let (directory, snapshot, compiled) = fixture();
let _ = std::panic::catch_unwind(|| {
let _ = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
if matches!(phase, SavePhase::AfterWriteMarker) {
panic!("interrupted");
}
Ok(())
});
});
let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
fs::write(journal.join(WRITE_STARTED), b"partial").unwrap();
assert!(matches!(
recover_project(
&directory.path().join("project.arcadia"),
ProjectLimits::default()
),
Err(SourceFileError::Invalid(_))
));
check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
assert!(journal.exists());
}
#[test]
fn partial_save_failures_restore_every_original_and_write_entry_last() {
for stop in 1..=3 {