//! Bounded disk IO for source projects. Compilation stays in `source_project`. //! //! Saving checks every captured dependency, stages exact bytes, and publishes a //! recovery journal before replacing files (the entry is last). This provides //! recoverability, not multi-file atomic visibility or a cross-process CAS. //! Symbolic links/junctions are rejected inside the project root. //! Recovery metadata admits at most 4096 files and 32 MiB of JSON; source bytes //! obey the caller's aggregate limits. Reserved `.arcadia-save-*` directories //! 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. use crate::{language, source_project::*}; use serde::{Deserialize, Serialize}; use std::{ collections::{BTreeMap, BTreeSet}, fs::{self, File}, io::{self, Read, Write}, path::{Path, PathBuf}, }; use thiserror::Error; const MAX_PATH_BYTES: usize = 4096; const MAX_JOURNAL_BYTES: usize = 32 * 1024 * 1024; const MAX_JOURNAL_FILES: usize = 4096; #[derive(Debug, Clone, PartialEq, Eq)] pub enum DiskExpectation { Missing, Bytes(Vec), } /// `bundle` retains detached and invalid buffers. Compile `reachable_bundle()`; /// use `diagnostics` to explain files that discovery could not read or parse. /// Expectations always describe disk bytes before UUID stabilization. #[derive(Debug, Clone)] pub struct DiskProject { pub root: PathBuf, pub bundle: SourceBundle, pub expectations: BTreeMap, pub reachable: BTreeSet, pub diagnostics: Vec, } impl DiskProject { pub fn reachable_bundle(&self) -> SourceBundle { SourceBundle { entry: self.bundle.entry.clone(), files: self .bundle .files .iter() .filter(|(path, _)| self.reachable.contains(*path)) .map(|(path, text)| (path.clone(), text.clone())) .collect(), } } } #[derive(Debug, Error)] pub enum SourceFileError { #[error("Source file operation failed at {path}: {source}")] Io { path: PathBuf, #[source] source: io::Error, }, #[error("Unsafe source path: {0}")] Path(String), #[error("Source project admission limit: {0}")] Limit(String), #[error("Source files changed outside Arcadia Studio: {0:?}")] Conflict(Vec), #[error("No disk expectation was established for {0}. Review the destination before saving.")] UnknownExpectation(LogicalPath), #[error("Invalid source project: {0}")] Invalid(String), #[error("A source save needs recovery at {0}. Recover it before saving again.")] RecoveryRequired(PathBuf), #[error( "Save failed ({save}); recovery could not finish ({recovery}). The journal remains at {journal}." )] RecoveryFailed { save: String, recovery: String, journal: PathBuf, }, } fn io_at(path: &Path, source: io::Error) -> SourceFileError { SourceFileError::Io { path: path.into(), source, } } fn root_entry(entry: &Path) -> Result<(PathBuf, String), SourceFileError> { let name = entry .file_name() .and_then(|name| name.to_str()) .ok_or_else(|| SourceFileError::Path("Entry needs a UTF-8 file name.".into()))?; check_logical(name)?; let parent = entry .parent() .filter(|path| !path.as_os_str().is_empty()) .unwrap_or(Path::new(".")); let root = parent .canonicalize() .map_err(|error| io_at(parent, error))?; if !root.is_dir() { return Err(SourceFileError::Path( "Project root is not a directory.".into(), )); } Ok((root, name.into())) } fn check_logical(path: &str) -> Result<(), SourceFileError> { if path.len() > MAX_PATH_BYTES || resolve_include("", path).as_deref() != Ok(path) || path .split('/') .any(|part| part.to_ascii_lowercase().starts_with(".arcadia-save-")) { return Err(SourceFileError::Path(format!( "Non-canonical, reserved or overlong path '{path}'." ))); } Ok(()) } /// Reject links rather than quietly replacing a link instead of its target. /// Existing parents are checked even for a new destination. fn disk_path(root: &Path, logical: &str) -> Result { check_logical(logical)?; let canonical_root = root.canonicalize().map_err(|error| io_at(root, error))?; if canonical_root != root { return Err(SourceFileError::Path( "Project root changed physical location.".into(), )); } let mut current = root.to_path_buf(); for component in logical.split('/') { current.push(component); match fs::symlink_metadata(¤t) { Ok(metadata) => { if metadata.file_type().is_symlink() { return Err(SourceFileError::Path(format!( "Links are not source project files: '{logical}'." ))); } let physical = current .canonicalize() .map_err(|error| io_at(¤t, error))?; if !physical.starts_with(root) { return Err(SourceFileError::Path(format!( "'{logical}' escapes the project root." ))); } } Err(error) if error.kind() == io::ErrorKind::NotFound => {} Err(error) => return Err(io_at(¤t, error)), } } Ok(current) } // Check before opening: opening a FIFO for reading can wait forever. The // post-open check below still verifies the opened object. As with disk_path, // this does not promise a CAS against concurrent path replacement. fn regular_file_metadata(path: &Path) -> Result { let metadata = fs::symlink_metadata(path).map_err(|error| io_at(path, error))?; if !metadata.is_file() { return Err(SourceFileError::Path(format!( "{} is not a regular file.", path.display() ))); } Ok(metadata) } fn read_bounded(path: &Path, maximum: usize) -> Result, SourceFileError> { regular_file_metadata(path)?; let file = File::open(path).map_err(|error| io_at(path, error))?; if !file .metadata() .map_err(|error| io_at(path, error))? .is_file() { return Err(SourceFileError::Path(format!( "{} is not a regular file.", path.display() ))); } let mut bytes = Vec::new(); file.take((maximum as u64).saturating_add(1)) .read_to_end(&mut bytes) .map_err(|error| io_at(path, error))?; if bytes.len() > maximum { return Err(SourceFileError::Limit(format!( "{} exceeds the remaining {maximum} byte budget.", path.display() ))); } Ok(bytes) } fn located( bundle: &SourceBundle, path: &str, span: language::Span, code: &'static str, message: impl Into, ) -> SourceDiagnostic { let source = bundle.files.get(path).map_or("", String::as_str); let mut diagnostic = language::diagnostic( source, language::Error { code, offset: span.start.min(source.len()), message: message.into(), }, ); diagnostic.span.end = span.end.min(source.len()).max(diagnostic.span.start); SourceDiagnostic { path: path.into(), diagnostic, related: vec![], } } fn check_snapshot(snapshot: &DiskProject, limits: ProjectLimits) -> Result<(), SourceFileError> { if snapshot.bundle.entry.contains('/') { return Err(SourceFileError::Path( "Entry must be at the source root.".into(), )); } let mut paths = BTreeSet::new(); let mut folded = BTreeMap::new(); for path in std::iter::once(&snapshot.bundle.entry) .chain(snapshot.bundle.files.keys()) .chain(snapshot.expectations.keys()) { check_logical(path)?; paths.insert(path); if let Some(previous) = folded.insert(path.to_lowercase(), path) && previous != path { return Err(SourceFileError::Path(format!( "Case-colliding paths '{previous}' and '{path}'." ))); } } if paths.len() > limits.max_files { return Err(SourceFileError::Limit(format!( "More than {} files.", limits.max_files ))); } for total in [ snapshot .bundle .files .values() .fold(0usize, |sum, text| sum.saturating_add(text.len())), snapshot .expectations .values() .fold(0usize, |sum, expected| { sum.saturating_add(match expected { DiskExpectation::Missing => 0, DiskExpectation::Bytes(bytes) => bytes.len(), }) }), ] { if total > limits.model.max_source_bytes { return Err(SourceFileError::Limit( "Combined source bytes exceed the project budget.".into(), )); } } Ok(()) } fn check_physical_aliases<'a>( root: &Path, paths: impl IntoIterator, ) -> Result<(), SourceFileError> { let mut handles: Vec<(&str, same_file::Handle)> = Vec::new(); let mut visited = BTreeSet::new(); for logical in paths { if !visited.insert(logical) { continue; } let path = disk_path(root, logical)?; match regular_file_metadata(&path) { Ok(_) => {} Err(SourceFileError::Io { source, .. }) if source.kind() == io::ErrorKind::NotFound => { continue; } Err(error) => return Err(error), } match same_file::Handle::from_path(&path) { Ok(handle) => { if let Some((previous, _)) = handles.iter().find(|(_, old)| *old == handle) { return Err(SourceFileError::Path(format!( "'{previous}' and '{logical}' refer to the same physical file." ))); } handles.push((logical, handle)); } Err(error) if error.kind() == io::ErrorKind::NotFound => {} Err(error) => return Err(io_at(&path, error)), } } Ok(()) } /// Load original bytes, including invalid UTF-8/syntax diagnostics. Missing /// includes are diagnostics, not a reason to discard editable files. pub fn load_project(entry: &Path, limits: ProjectLimits) -> Result { let (root, entry) = root_entry(entry)?; let snapshot = DiskProject { root, bundle: SourceBundle { entry, files: BTreeMap::new(), }, expectations: BTreeMap::new(), reachable: BTreeSet::new(), diagnostics: vec![], }; discover_missing(&snapshot, &BTreeMap::new(), limits) } /// Establish an unsaved project destination. An existing entry is a collision, /// never implicit permission to overwrite it. Parent directory must exist. pub fn new_project(entry: &Path, limits: ProjectLimits) -> Result { let (root, entry) = root_entry(entry)?; let mut snapshot = DiskProject { root, bundle: SourceBundle { entry: entry.clone(), files: BTreeMap::new(), }, expectations: BTreeMap::new(), reachable: BTreeSet::from([entry.clone()]), diagnostics: vec![], }; expect_new_file(&mut snapshot, &entry, limits)?; Ok(snapshot) } pub fn expect_new_file( snapshot: &mut DiskProject, path: &str, limits: ProjectLimits, ) -> Result<(), SourceFileError> { let destination = disk_path(&snapshot.root, path)?; match fs::symlink_metadata(&destination) { Ok(_) => return Err(SourceFileError::Conflict(vec![path.into()])), Err(error) if error.kind() == io::ErrorKind::NotFound => {} Err(error) => return Err(io_at(&destination, error)), } let mut candidate = snapshot.clone(); candidate .expectations .insert(path.into(), DiskExpectation::Missing); check_snapshot(&candidate, limits)?; snapshot.expectations = candidate.expectations; Ok(()) } /// Merge editor overlays and discover only files absent from retained buffers. /// Existing drafts (including detached ones) and their disk baselines survive. pub fn discover_missing( snapshot: &DiskProject, overlays: &BTreeMap, limits: ProjectLimits, ) -> Result { let mut result = snapshot.clone(); result.bundle.files.extend(overlays.clone()); result.reachable.clear(); result.diagnostics.clear(); check_snapshot(&result, limits)?; let mut stack = vec![(result.bundle.entry.clone(), 0usize, None::)]; let mut edges = 0usize; let mut attempted = BTreeSet::new(); while let Some((path, depth, origin)) = stack.pop() { let at = origin.unwrap_or(SourceLocation { path: path.clone(), span: language::Span::default(), }); if depth > limits.max_include_depth { result.diagnostics.push(located( &result.bundle, &at.path, at.span, "project_limit", "Include depth exceeds the project budget.", )); continue; } result.reachable.insert(path.clone()); if !attempted.insert(path.clone()) { continue; } if !result.bundle.files.contains_key(&path) { let unique = result .expectations .keys() .chain(result.bundle.files.keys()) .chain(std::iter::once(&path)) .collect::>(); if unique.len() > limits.max_files { result.diagnostics.push(located( &result.bundle, &at.path, at.span, "project_limit", "File count exceeds the project budget.", )); continue; } let used = result .bundle .files .values() .fold(0usize, |sum, text| sum.saturating_add(text.len())) .max(result.expectations.values().fold(0usize, |sum, expected| { sum.saturating_add(match expected { DiskExpectation::Missing => 0, DiskExpectation::Bytes(bytes) => bytes.len(), }) })); let loaded = disk_path(&result.root, &path).and_then(|physical| { read_bounded( &physical, limits.model.max_source_bytes.saturating_sub(used), ) }); match loaded { Ok(bytes) => { result .expectations .insert(path.clone(), DiskExpectation::Bytes(bytes.clone())); match String::from_utf8(bytes) { Ok(text) => { result.bundle.files.insert(path.clone(), text); } Err(_) => { result.diagnostics.push(located( &result.bundle, &at.path, at.span, "source_encoding", format!("'{path}' is not UTF-8 source."), )); continue; } } } Err(error) => { if matches!(&error, SourceFileError::Io { source, .. } if source.kind() == io::ErrorKind::NotFound) { result .expectations .entry(path.clone()) .or_insert(DiskExpectation::Missing); } let code = match &error { SourceFileError::Limit(_) => "source_limit", SourceFileError::Path(_) => "invalid_path", _ => "missing_include", }; result.diagnostics.push(located( &result.bundle, &at.path, at.span, code, format!("Cannot load '{path}': {error}"), )); continue; } } } match inspect_includes( &path, &result.bundle.files[&path], path == result.bundle.entry, limits, ) { Ok(includes) => { edges = edges.saturating_add(includes.len()); if edges > limits.max_include_edges { result.diagnostics.push(located( &result.bundle, &path, language::Span::default(), "project_limit", "Include edges exceed the project budget.", )); continue; } stack.extend( includes .into_iter() .rev() .map(|include| (include.target, depth + 1, Some(include.location))), ); } Err(diagnostics) => result.diagnostics.extend(diagnostics), } } check_snapshot(&result, limits)?; check_physical_aliases( &result.root, result.expectations.keys().chain(result.bundle.files.keys()), )?; if result.diagnostics.is_empty() { match compile_bundle(&result.reachable_bundle(), limits) { Ok(compiled) => result.diagnostics = compiled.diagnostics, Err(diagnostics) => result.diagnostics = diagnostics, } } let pending = journal_path(&result.root, &result.bundle.entry); if fs::symlink_metadata(&pending).is_ok() { result.diagnostics.push(located( &result.bundle, &result.bundle.entry, language::Span::default(), "pending_save", format!( "A save was interrupted. Review and recover {} before saving this project.", pending.display() ), )); } Ok(result) } #[derive(Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct Journal { version: u8, entry: String, #[serde(deserialize_with = "journal_files")] files: Vec, } fn journal_files<'de, D: serde::Deserializer<'de>>( deserializer: D, ) -> Result, D::Error> { struct Files; impl<'de> serde::de::Visitor<'de> for Files { type Value = Vec; fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str("at most 4096 recovery entries") } fn visit_seq>( self, mut sequence: A, ) -> Result { let mut files = Vec::new(); while let Some(file) = sequence.next_element()? { if files.len() >= MAX_JOURNAL_FILES { return Err(serde::de::Error::custom("Recovery entry budget exceeded.")); } files.push(file); } Ok(files) } } deserializer.deserialize_seq(Files) } #[derive(Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct JournalFile { path: String, before: bool, write: bool, } fn journal_path(root: &Path, entry: &str) -> PathBuf { root.join(format!(".arcadia-save-{entry}.journal")) } fn staged_path(journal: &Path, index: usize, suffix: &str) -> PathBuf { journal.join(format!("{index}.{suffix}")) } fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), SourceFileError> { let mut file = File::options() .write(true) .create_new(true) .open(path) .map_err(|error| io_at(path, error))?; file.write_all(bytes) .and_then(|()| file.sync_all()) .map_err(|error| io_at(path, error)) } fn sync_directory(path: &Path) -> Result<(), SourceFileError> { #[cfg(unix)] File::open(path) .and_then(|file| file.sync_all()) .map_err(|error| io_at(path, error))?; #[cfg(not(unix))] let _ = path; Ok(()) } fn atomic_write(root: &Path, logical: &str, bytes: &[u8]) -> Result<(), SourceFileError> { let destination = disk_path(root, logical)?; let parent = destination.parent().unwrap(); fs::create_dir_all(parent).map_err(|error| io_at(parent, error))?; // Recheck after directory creation; an external process may replace parents. disk_path(root, logical)?; let permissions = match regular_file_metadata(&destination) { Ok(metadata) => Some(metadata.permissions()), Err(SourceFileError::Io { source, .. }) if source.kind() == io::ErrorKind::NotFound => None, Err(error) => return Err(error), }; let mut staged = tempfile::NamedTempFile::new_in(parent).map_err(|error| io_at(parent, error))?; staged .write_all(bytes) .map_err(|error| io_at(&destination, error))?; // Preserve the existing file's portable permissions through replacement // and recovery. New files retain the private temporary-file default. if let Some(permissions) = permissions { staged .as_file() .set_permissions(permissions) .map_err(|error| io_at(&destination, error))?; } staged .as_file() .sync_all() .map_err(|error| io_at(&destination, error))?; staged .persist(&destination) .map_err(|error| io_at(&destination, error.error))?; sync_directory(parent) } fn matches_disk( root: &Path, logical: &str, expected: &DiskExpectation, ) -> Result { let path = disk_path(root, logical)?; match expected { DiskExpectation::Missing => match fs::symlink_metadata(&path) { Ok(_) => Ok(false), Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(true), Err(error) => Err(io_at(&path, error)), }, DiskExpectation::Bytes(expected) => match read_bounded(&path, expected.len()) { Ok(actual) => Ok(actual == *expected), Err(SourceFileError::Limit(_)) => Ok(false), Err(SourceFileError::Io { source, .. }) if source.kind() == io::ErrorKind::NotFound => { Ok(false) } Err(error) => Err(error), }, } } fn check_conflicts( root: &Path, expectations: &BTreeMap, ) -> Result<(), SourceFileError> { check_physical_aliases(root, expectations.keys())?; let mut conflicts = Vec::new(); for (path, expected) in expectations { if !matches_disk(root, path, expected)? { conflicts.push(path.clone()); } } if conflicts.is_empty() { Ok(()) } else { Err(SourceFileError::Conflict(conflicts)) } } /// Save an already accepted, identity-stabilized bundle. Every destination needs /// an explicit original/Missing expectation. Detached files are never deleted. pub fn save_project( snapshot: &DiskProject, compiled: &CompiledSourceProject, limits: ProjectLimits, ) -> Result { save_with_hook(snapshot, compiled, limits, |_| Ok(())) } fn save_with_hook( snapshot: &DiskProject, compiled: &CompiledSourceProject, limits: ProjectLimits, mut after_replace: impl FnMut(usize) -> io::Result<()>, ) -> Result { check_snapshot(snapshot, limits)?; let checked = compile_bundle(&compiled.bundle, limits).map_err(|diagnostics| { SourceFileError::Invalid( diagnostics .iter() .map(|d| d.diagnostic.to_string()) .collect::>() .join("\n"), ) })?; if checked.bundle != compiled.bundle || checked.decoded.project != compiled.decoded.project || checked.index != compiled.index || snapshot.bundle.entry != compiled.bundle.entry { return Err(SourceFileError::Invalid( "Compiled model, source, index or entry is stale.".into(), )); } let compiled = &checked; for path in compiled.bundle.files.keys() { disk_path(&snapshot.root, path)?; if !snapshot.expectations.contains_key(path) { return Err(SourceFileError::UnknownExpectation(path.clone())); } } check_snapshot(&saved_snapshot(snapshot, compiled), limits)?; let journal_dir = journal_path(&snapshot.root, &snapshot.bundle.entry); if fs::symlink_metadata(&journal_dir).is_ok() { return Err(SourceFileError::RecoveryRequired(journal_dir)); } check_conflicts(&snapshot.root, &snapshot.expectations)?; let mut entries = snapshot.expectations.iter().map(|(path, before)| JournalFile { path: path.clone(), before: matches!(before, DiskExpectation::Bytes(_)), write: compiled.bundle.files.get(path).is_some_and(|after| !matches!(before, DiskExpectation::Bytes(bytes) if bytes == after.as_bytes())), }).collect::>(); // 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, entry: snapshot.bundle.entry.clone(), files: entries, }; if journal.files.len() > MAX_JOURNAL_FILES { return Err(SourceFileError::Limit( "Recovery is limited to 4096 files.".into(), )); } if !journal.files.iter().any(|file| file.write) { return Ok(saved_snapshot(snapshot, compiled)); } let metadata = serde_json::to_vec(&journal) .map_err(|error| SourceFileError::Invalid(error.to_string()))?; if metadata.len() > MAX_JOURNAL_BYTES { return Err(SourceFileError::Limit( "Recovery metadata is too large.".into(), )); } let stage = tempfile::Builder::new() .prefix(".arcadia-save-stage-") .tempdir_in(&snapshot.root) .map_err(|error| io_at(&snapshot.root, error))?; for (index, file) in journal.files.iter().enumerate() { if let DiskExpectation::Bytes(bytes) = &snapshot.expectations[&file.path] { write_synced(&staged_path(stage.path(), index, "before"), bytes)?; } if file.write { write_synced( &staged_path(stage.path(), index, "after"), compiled.bundle.files[&file.path].as_bytes(), )?; } } write_synced(&stage.path().join("journal.json"), &metadata)?; sync_directory(stage.path())?; // Publishing a complete directory makes interrupted staging harmless. fs::rename(stage.path(), &journal_dir).map_err(|error| io_at(&journal_dir, error))?; let _ = stage.keep(); // The renamed directory now belongs to recovery. sync_directory(&snapshot.root)?; 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, ¤t)?; 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) .map_err(|error| io_at(&snapshot.root.join(&file.path), error))?; } check_conflicts(&snapshot.root, ¤t)?; Ok(()) })(); if let Err(error) = result { return match recover_project(&snapshot.root.join(&snapshot.bundle.entry), limits) { Ok(_) => Err(error), Err(recovery) => Err(SourceFileError::RecoveryFailed { save: error.to_string(), recovery: recovery.to_string(), journal: journal_dir, }), }; } retire_journal(&snapshot.root, &journal_dir, &journal)?; Ok(saved_snapshot(snapshot, compiled)) } fn saved_snapshot(snapshot: &DiskProject, compiled: &CompiledSourceProject) -> DiskProject { let mut saved = snapshot.clone(); for (path, text) in &compiled.bundle.files { saved.bundle.files.insert(path.clone(), text.clone()); saved.expectations.insert( path.clone(), DiskExpectation::Bytes(text.as_bytes().to_vec()), ); } saved.reachable = compiled.bundle.files.keys().cloned().collect(); saved.diagnostics = compiled.diagnostics.clone(); saved } fn remove_journal(directory: &Path, journal: &Journal) -> Result<(), SourceFileError> { // Never recursively delete an untrusted journal directory. Only validated, // fixed-name files are removed; unexpected content keeps the directory. for (index, file) in journal.files.iter().enumerate() { for (present, suffix) in [(file.before, "before"), (file.write, "after")] { if present { let path = staged_path(directory, index, suffix); fs::remove_file(&path).map_err(|error| io_at(&path, 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)) } fn retire_journal(root: &Path, directory: &Path, journal: &Journal) -> Result<(), SourceFileError> { // Atomic retirement precedes cleanup: interruption during cleanup cannot // leave a pending journal with half its recovery information missing. let completed = root.join(format!(".arcadia-save-completed-{}", crate::Id::new_v4())); fs::rename(directory, &completed).map_err(|error| io_at(directory, error))?; // Cleanup failures leave a uniquely named completed directory, never a // misleading pending transaction. It is safe to remove that directory later. let _ = sync_directory(root); let _ = remove_journal(&completed, journal); Ok(()) } /// Explicit interrupted-save recovery. All original/written bytes are checked /// before rollback, and each restoration rechecks its target. External changes /// are never knowingly overwritten. `false` means there was no pending journal. pub fn recover_project(entry: &Path, limits: ProjectLimits) -> Result { recover_with_hook(entry, limits, |_| Ok(())) } fn recover_with_hook( entry: &Path, limits: ProjectLimits, mut after_restore: impl FnMut(usize) -> io::Result<()>, ) -> Result { let (root, entry) = root_entry(entry)?; let directory = journal_path(&root, &entry); match fs::symlink_metadata(&directory) { Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false), Err(error) => return Err(io_at(&directory, error)), Ok(metadata) if !metadata.is_dir() || metadata.file_type().is_symlink() => { return Err(SourceFileError::Path( "Recovery journal is not a real directory.".into(), )); } Ok(_) => {} } if directory .canonicalize() .map_err(|error| io_at(&directory, error))? .parent() != Some(root.as_path()) { return Err(SourceFileError::Path( "Recovery directory escapes project root.".into(), )); } let metadata_path = directory.join("journal.json"); let read_journal_file = |path: &Path, budget| { if fs::symlink_metadata(path) .map_err(|error| io_at(path, error))? .file_type() .is_symlink() { return Err(SourceFileError::Path("Recovery file is a link.".into())); } read_bounded(path, budget) }; 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 { return Err(SourceFileError::Invalid( "Recovery journal version, entry or file budget is invalid.".into(), )); } let mut original = BTreeMap::new(); let mut written = BTreeMap::new(); let mut folded = BTreeSet::new(); let (mut before_bytes, mut after_bytes) = (0usize, 0usize); let mut allowed = BTreeSet::from(["journal.json".to_string()]); for (index, file) in journal.files.iter().enumerate() { disk_path(&root, &file.path)?; if !folded.insert(file.path.to_lowercase()) { return Err(SourceFileError::Path("Duplicate recovery paths.".into())); } let before = if file.before { let bytes = read_journal_file( &staged_path(&directory, index, "before"), limits.model.max_source_bytes.saturating_sub(before_bytes), )?; before_bytes += bytes.len(); allowed.insert(format!("{index}.before")); DiskExpectation::Bytes(bytes) } else { DiskExpectation::Missing }; if file.write { let bytes = read_journal_file( &staged_path(&directory, index, "after"), limits.model.max_source_bytes.saturating_sub(after_bytes), )?; after_bytes += bytes.len(); allowed.insert(format!("{index}.after")); written.insert(file.path.clone(), DiskExpectation::Bytes(bytes)); } original.insert(file.path.clone(), before); } if !original.contains_key(&entry) { return Err(SourceFileError::Invalid( "Recovery journal has no entry expectation.".into(), )); } for file in fs::read_dir(&directory).map_err(|error| io_at(&directory, error))? { let file = file.map_err(|error| io_at(&directory, error))?; if !file .file_name() .to_str() .is_some_and(|name| allowed.contains(name)) { return Err(SourceFileError::Invalid( "Recovery directory contains unexpected files.".into(), )); } } let mut current = BTreeMap::new(); let mut conflicts = Vec::new(); for (path, before) in &original { if matches_disk(&root, path, before)? { current.insert(path.clone(), before.clone()); } else if let Some(after) = written.get(path) && matches_disk(&root, path, after)? { current.insert(path.clone(), after.clone()); } else { conflicts.push(path.clone()); } } if !conflicts.is_empty() { return Err(SourceFileError::Conflict(conflicts)); } // Reverse save order; a second recovery can continue after interruption. let mut restored = 0; for file in journal.files.iter().rev().filter(|file| file.write) { check_conflicts(&root, ¤t)?; if current[&file.path] == original[&file.path] { continue; } match &original[&file.path] { DiskExpectation::Bytes(bytes) => atomic_write(&root, &file.path, bytes)?, DiskExpectation::Missing => { let path = disk_path(&root, &file.path)?; fs::remove_file(&path).map_err(|error| io_at(&path, error))?; sync_directory(path.parent().unwrap())?; } } current.insert(file.path.clone(), original[&file.path].clone()); restored += 1; after_restore(restored).map_err(|error| io_at(&directory, error))?; } retire_journal(&root, &directory, &journal)?; Ok(true) } #[cfg(test)] mod transaction_tests { use super::*; fn fixture() -> (tempfile::TempDir, DiskProject, CompiledSourceProject) { let directory = tempfile::tempdir().unwrap(); fs::write( directory.path().join("project.arcadia"), "arcadia 5; project p \"P\" { include \"a.arcadia\"; include \"b.arcadia\"; }", ) .unwrap(); fs::write(directory.path().join("a.arcadia"), "oa_activity a \"A\" {}").unwrap(); fs::write(directory.path().join("b.arcadia"), "oa_activity b \"B\" {}").unwrap(); let snapshot = load_project( &directory.path().join("project.arcadia"), ProjectLimits::default(), ) .unwrap(); let compiled = compile_bundle(&snapshot.reachable_bundle(), ProjectLimits::default()).unwrap(); (directory, snapshot, compiled) } #[test] fn partial_save_failures_restore_every_original_and_write_entry_last() { for stop in 1..=3 { let (directory, snapshot, compiled) = fixture(); let error = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| { if count < 3 { assert!( matches_disk( &snapshot.root, &snapshot.bundle.entry, &snapshot.expectations[&snapshot.bundle.entry] ) .unwrap() ); } if count == stop { Err(io::Error::other("injected replacement failure")) } else { Ok(()) } }) .unwrap_err(); assert!(matches!(error, SourceFileError::Io { .. }), "{error}"); check_conflicts(&snapshot.root, &snapshot.expectations).unwrap(); assert!( !recover_project( &directory.path().join("project.arcadia"), ProjectLimits::default() ) .unwrap() ); } } #[test] fn interrupted_save_and_interrupted_recovery_can_both_be_resumed() { let (directory, snapshot, compiled) = fixture(); let interrupted = std::panic::catch_unwind(|| { let _ = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| { assert!(count < 3, "simulated process interruption"); Ok(()) }); }); assert!(interrupted.is_err()); assert!( load_project( &directory.path().join("project.arcadia"), ProjectLimits::default() ) .unwrap() .diagnostics .iter() .any(|d| d.diagnostic.code == "pending_save") ); assert!(matches!( save_project(&snapshot, &compiled, ProjectLimits::default()), Err(SourceFileError::RecoveryRequired(_)) )); let entry = directory.path().join("project.arcadia"); let interrupted = recover_with_hook(&entry, ProjectLimits::default(), |_| { Err(io::Error::other("recovery interrupted")) }); assert!(interrupted.is_err()); assert!(recover_project(&entry, ProjectLimits::default()).unwrap()); check_conflicts(&snapshot.root, &snapshot.expectations).unwrap(); assert!(!recover_project(&entry, ProjectLimits::default()).unwrap()); } #[test] fn rollback_refuses_external_edits_even_in_an_unchanged_dependency() { let (directory, mut snapshot, initial) = fixture(); snapshot = save_project(&snapshot, &initial, ProjectLimits::default()).unwrap(); let mut bundle = initial.bundle.clone(); bundle .files .get_mut("a.arcadia") .unwrap() .push_str("\n// accepted change\n"); let compiled = compile_bundle(&bundle, ProjectLimits::default()).unwrap(); let error = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |_| { fs::write(directory.path().join("b.arcadia"), "// external edit").unwrap(); Err(io::Error::other("injected failure after external edit")) }) .unwrap_err(); assert!(matches!(error, SourceFileError::RecoveryFailed { .. })); assert_eq!( fs::read_to_string(directory.path().join("b.arcadia")).unwrap(), "// external edit" ); assert_eq!( fs::read_to_string(directory.path().join("a.arcadia")).unwrap(), compiled.bundle.files["a.arcadia"] ); fs::write( directory.path().join("b.arcadia"), &initial.bundle.files["b.arcadia"], ) .unwrap(); assert!( recover_project( &directory.path().join("project.arcadia"), ProjectLimits::default() ) .unwrap() ); check_conflicts(&snapshot.root, &snapshot.expectations).unwrap(); } #[test] fn recovery_rechecks_external_changes_between_restorations() { let (directory, snapshot, compiled) = fixture(); let _ = std::panic::catch_unwind(|| { let _ = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| { assert!(count < 3, "interrupted after all replacements"); Ok(()) }); }); let entry = directory.path().join("project.arcadia"); let error = recover_with_hook(&entry, ProjectLimits::default(), |_| { fs::write( directory.path().join("b.arcadia"), "// external edit during recovery", ) .unwrap(); Ok(()) }) .unwrap_err(); assert!(matches!(error, SourceFileError::Conflict(_))); assert_eq!( fs::read_to_string(directory.path().join("b.arcadia")).unwrap(), "// external edit during recovery" ); assert!(journal_path(&snapshot.root, &snapshot.bundle.entry).exists()); fs::write( directory.path().join("b.arcadia"), &compiled.bundle.files["b.arcadia"], ) .unwrap(); assert!(recover_project(&entry, ProjectLimits::default()).unwrap()); check_conflicts(&snapshot.root, &snapshot.expectations).unwrap(); } #[test] fn rollback_removes_only_our_new_file_and_retains_detached_files() { let (directory, mut snapshot, _) = fixture(); fs::write(directory.path().join("detached.arcadia"), "// retained").unwrap(); let mut bundle = snapshot.reachable_bundle(); let source = bundle.files.get_mut("project.arcadia").unwrap(); *source = source.replacen( "include \"a.arcadia\";", "include \"new/first.arcadia\"; include \"a.arcadia\";", 1, ); bundle .files .insert("new/first.arcadia".into(), "oa_activity c \"C\" {}".into()); expect_new_file(&mut snapshot, "new/first.arcadia", ProjectLimits::default()).unwrap(); let compiled = compile_bundle(&bundle, ProjectLimits::default()).unwrap(); let error = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| { if count == 3 { Err(io::Error::other("after new file")) } else { Ok(()) } }) .unwrap_err(); assert!(matches!(error, SourceFileError::Io { .. })); assert!(!directory.path().join("new/first.arcadia").exists()); assert_eq!( fs::read_to_string(directory.path().join("detached.arcadia")).unwrap(), "// retained" ); check_conflicts(&snapshot.root, &snapshot.expectations).unwrap(); } #[test] fn recovery_rejects_untrusted_escape_paths_before_touching_files() { let (directory, snapshot, compiled) = fixture(); let _ = std::panic::catch_unwind(|| { let _ = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |_| { panic!("stop") }); }); let journal_dir = journal_path(&snapshot.root, &snapshot.bundle.entry); let metadata = journal_dir.join("journal.json"); let mut journal: Journal = serde_json::from_slice(&fs::read(&metadata).unwrap()).unwrap(); journal.files[0].path = "../outside.arcadia".into(); fs::write(&metadata, serde_json::to_vec(&journal).unwrap()).unwrap(); assert!(matches!( recover_project( &directory.path().join("project.arcadia"), ProjectLimits::default() ), Err(SourceFileError::Path(_)) )); assert!(journal_dir.exists()); } }