1916 lines
73 KiB
Rust
1916 lines
73 KiB
Rust
//! 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.
|
|
//! 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::{
|
|
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;
|
|
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 {
|
|
Missing,
|
|
Bytes(Vec<u8>),
|
|
}
|
|
|
|
/// `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<LogicalPath, DiskExpectation>,
|
|
pub reachable: BTreeSet<LogicalPath>,
|
|
pub diagnostics: Vec<SourceDiagnostic>,
|
|
}
|
|
|
|
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<LogicalPath>),
|
|
#[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<PathBuf, SourceFileError> {
|
|
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<fs::Metadata, SourceFileError> {
|
|
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<Vec<u8>, 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<String>,
|
|
) -> 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<Item = &'a String>,
|
|
) -> 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<DiskProject, SourceFileError> {
|
|
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<DiskProject, SourceFileError> {
|
|
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<LogicalPath, String>,
|
|
limits: ProjectLimits,
|
|
) -> Result<DiskProject, SourceFileError> {
|
|
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::<SourceLocation>)];
|
|
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::<BTreeSet<_>>();
|
|
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<JournalFile>,
|
|
}
|
|
|
|
fn journal_files<'de, D: serde::Deserializer<'de>>(
|
|
deserializer: D,
|
|
) -> Result<Vec<JournalFile>, D::Error> {
|
|
struct Files;
|
|
impl<'de> serde::de::Visitor<'de> for Files {
|
|
type Value = Vec<JournalFile>;
|
|
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
formatter.write_str("at most 4096 recovery entries")
|
|
}
|
|
fn visit_seq<A: serde::de::SeqAccess<'de>>(
|
|
self,
|
|
mut sequence: A,
|
|
) -> Result<Self::Value, A::Error> {
|
|
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 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)
|
|
.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<bool, SourceFileError> {
|
|
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<LogicalPath, DiskExpectation>,
|
|
) -> 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<DiskProject, SourceFileError> {
|
|
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<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| {
|
|
SourceFileError::Invalid(
|
|
diagnostics
|
|
.iter()
|
|
.map(|d| d.diagnostic.to_string())
|
|
.collect::<Vec<_>>()
|
|
.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::<Vec<_>>();
|
|
// 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: 2,
|
|
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())?;
|
|
phase(SavePhase::BeforePublish).map_err(|error| io_at(&journal_dir, error))?;
|
|
// Publishing a complete directory makes interrupted staging harmless.
|
|
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) {
|
|
// 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, ¤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;
|
|
phase(SavePhase::AfterReplace(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,
|
|
}),
|
|
};
|
|
}
|
|
phase(SavePhase::BeforeRetire).map_err(|error| io_at(&journal_dir, error))?;
|
|
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))?;
|
|
}
|
|
}
|
|
}
|
|
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))
|
|
}
|
|
|
|
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<bool, SourceFileError> {
|
|
recover_with_hook(entry, limits, |_| Ok(()))
|
|
}
|
|
|
|
fn recover_with_hook(
|
|
entry: &Path,
|
|
limits: ProjectLimits,
|
|
mut after_restore: impl FnMut(usize) -> io::Result<()>,
|
|
) -> Result<bool, SourceFileError> {
|
|
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 !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(),
|
|
));
|
|
}
|
|
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()]);
|
|
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()) {
|
|
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(),
|
|
));
|
|
}
|
|
}
|
|
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 {
|
|
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::*;
|
|
|
|
#[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(
|
|
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 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 {
|
|
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());
|
|
}
|
|
}
|