use crate::atomic::{
create_dir_all_durable, publish_file_with, publish_file_with_access_denied_retry, try_sync_dir,
AccessDeniedRetry, AtomicWriteOptions, AtomicWriter, Privacy, ReplaceMode, RetryPolicy,
};
use crate::error::{FsError, Result};
use crate::file_key::FileKey;
use crate::lock::{FileLock, LockOptions};
use crate::plain::canonicalize_plain;
use std::fs::{self, File};
use std::io::{self, Read};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
const AREA: &str = ".rktxn";
const LOCK_NAME: &str = "lock";
const RETAINED_DIR: &str = "retained";
const JOURNAL: &str = "journal";
const COMMIT: &str = "commit";
const JOURNAL_MAGIC: &str = "rktxn-journal 1";
const COMMIT_MAGIC: &str = "rktxn-commit 1";
const SIDE_MARKER: &str = ".rktxn-";
const SIDE_PREFIX_MAX: usize = 64;
const COPY_BUFFER: usize = 64 * 1024;
const MAX_COMPONENTS: usize = 64;
const MAX_JOURNAL_BYTES: u64 = 4 * 1024 * 1024;
pub const MAX_ENTRIES: usize = 256;
pub const MAX_RELATIVE_BYTES: usize = 1024;
static ID_COUNTER: AtomicU64 = AtomicU64::new(0);
macro_rules! define_step {
($vis:vis) => {
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
#[allow(dead_code)]
$vis enum Step {
BeforeJournalStaging,
AfterJournalStaging,
BeforeStage(usize),
BeforeStageClaim(usize),
AfterStageClaim(usize),
AfterStage(usize),
BeforeJournalStaged,
AfterJournalStaged,
BeforeBackup(usize),
BeforeLink(usize),
AfterBackup(usize),
BeforePublish(usize),
AfterPublish(usize),
BeforeCommitMarker,
AfterCommitMarker,
BeforeCleanup(usize),
AfterCleanup(usize),
BeforeRollbackMark(usize),
AfterRollbackMark(usize),
BeforeRestore(usize),
AfterRestore(usize),
BeforeJournalRemove,
AfterJournalRemove,
AfterMarkerRemove,
SetAsideSyncSource,
SetAsideSyncTarget,
SetAsideProbe,
DeleteBackoff,
}
};
}
#[cfg(feature = "test-seams")]
define_step!(pub);
#[cfg(not(feature = "test-seams"))]
define_step!(pub(crate));
#[cfg(feature = "test-seams")]
pub mod seams {
pub use super::Step;
use std::cell::{Cell, RefCell};
type Hook = Box<dyn Fn(&Step) -> Option<std::io::Error>>;
thread_local! {
static HOOK: RefCell<Option<Hook>> = const { RefCell::new(None) };
static FORCE_ASIDE: Cell<bool> = const { Cell::new(false) };
static FORCE_DELETE_FAILURES: Cell<u32> = const { Cell::new(0) };
}
#[derive(Debug)]
pub struct DeleteGuard(());
impl Drop for DeleteGuard {
fn drop(&mut self) {
FORCE_DELETE_FAILURES.with(|count| count.set(0));
}
}
pub fn force_delete_failures(count: u32) -> DeleteGuard {
FORCE_DELETE_FAILURES.with(|failures| failures.set(count));
DeleteGuard(())
}
pub(super) fn take_delete_failure() -> bool {
FORCE_DELETE_FAILURES.with(|count| {
let left = count.get();
count.set(left.saturating_sub(1));
left > 0
})
}
#[derive(Debug)]
pub struct AsideGuard(());
impl Drop for AsideGuard {
fn drop(&mut self) {
FORCE_ASIDE.with(|flag| flag.set(false));
}
}
pub fn force_aside_backup() -> AsideGuard {
FORCE_ASIDE.with(|flag| flag.set(true));
AsideGuard(())
}
pub(super) fn aside_forced() -> bool {
FORCE_ASIDE.with(Cell::get)
}
#[derive(Debug)]
pub struct StepGuard(());
impl Drop for StepGuard {
fn drop(&mut self) {
HOOK.with(|hook| *hook.borrow_mut() = None);
}
}
pub fn on_step(hook: impl Fn(&Step) -> Option<std::io::Error> + 'static) -> StepGuard {
HOOK.with(|slot| *slot.borrow_mut() = Some(Box::new(hook)));
StepGuard(())
}
pub(super) fn run(step: &Step) -> Option<std::io::Error> {
HOOK.with(|slot| slot.borrow().as_ref().and_then(|hook| hook(step)))
}
}
#[cfg(feature = "test-seams")]
fn fire(step: Step, at: &Path) -> Result<()> {
match seams::run(&step) {
Some(error) => Err(FsError::io(at, error)),
None => Ok(()),
}
}
#[cfg(not(feature = "test-seams"))]
#[inline(always)]
fn fire(_step: Step, _at: &Path) -> Result<()> {
Ok(())
}
#[cfg(feature = "test-seams")]
fn forced_delete_failure() -> Option<io::Error> {
seams::take_delete_failure()
.then(|| io::Error::new(io::ErrorKind::ResourceBusy, "injected sharing failure"))
}
#[cfg(not(feature = "test-seams"))]
#[inline(always)]
fn forced_delete_failure() -> Option<io::Error> {
None
}
#[cfg(feature = "test-seams")]
fn aside_forced() -> bool {
seams::aside_forced()
}
#[cfg(not(feature = "test-seams"))]
#[inline(always)]
fn aside_forced() -> bool {
false
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum ExpectedPrior {
Any,
Absent,
Identity(FileKey),
#[cfg(feature = "digest")]
Sha256(String),
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum Publication {
#[default]
Replace,
CreateNew,
}
#[derive(Clone, Debug)]
pub struct TransactionEntry {
destination: PathBuf,
candidate: PathBuf,
expect: ExpectedPrior,
publication: Publication,
private: bool,
}
impl TransactionEntry {
pub fn new(destination: impl Into<PathBuf>, candidate: impl Into<PathBuf>) -> Self {
Self {
destination: destination.into(),
candidate: candidate.into(),
expect: ExpectedPrior::Any,
publication: Publication::Replace,
private: false,
}
}
pub fn create_new(destination: impl Into<PathBuf>, candidate: impl Into<PathBuf>) -> Self {
Self {
publication: Publication::CreateNew,
expect: ExpectedPrior::Absent,
..Self::new(destination, candidate)
}
}
pub fn expect(mut self, expect: ExpectedPrior) -> Self {
self.expect = expect;
self
}
pub fn private(mut self) -> Self {
self.private = true;
self
}
pub fn destination(&self) -> &Path {
&self.destination
}
pub fn candidate(&self) -> &Path {
&self.candidate
}
}
#[derive(Clone, Debug)]
pub struct TransactionOptions {
lock_timeout: Duration,
retry: Option<RetryPolicy>,
access_denied: Option<AccessDeniedRetry>,
}
impl Default for TransactionOptions {
fn default() -> Self {
Self {
lock_timeout: Duration::from_secs(30),
retry: Some(RetryPolicy::new(5, Duration::from_millis(50))),
access_denied: None,
}
}
}
impl TransactionOptions {
pub fn lock_timeout(mut self, timeout: Duration) -> Self {
self.lock_timeout = timeout;
self
}
pub fn retry(mut self, retry: Option<RetryPolicy>) -> Self {
self.retry = retry;
self
}
pub fn access_denied_retry(mut self, retry: AccessDeniedRetry) -> Self {
self.access_denied = Some(retry);
self
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct RetainedItem {
pub path: PathBuf,
pub reason: String,
}
impl RetainedItem {
fn new(path: impl Into<PathBuf>, reason: impl std::fmt::Display) -> Self {
Self {
path: path.into(),
reason: reason.to_string(),
}
}
}
#[derive(Debug)]
#[non_exhaustive]
pub struct CommittedReport {
pub id: String,
pub published: Vec<PathBuf>,
pub retained: Vec<RetainedItem>,
}
#[derive(Debug)]
#[non_exhaustive]
pub struct RolledBackReport {
pub id: String,
pub cause: FsError,
pub restored: Vec<PathBuf>,
}
#[derive(Debug)]
#[non_exhaustive]
pub struct RecoveryRequiredReport {
pub id: String,
pub cause: FsError,
pub retained: Vec<RetainedItem>,
pub journal: PathBuf,
}
#[derive(Debug)]
#[non_exhaustive]
pub enum TransactionReport {
Committed(CommittedReport),
RolledBack(RolledBackReport),
RecoveryRequired(RecoveryRequiredReport),
}
impl TransactionReport {
pub fn id(&self) -> &str {
match self {
Self::Committed(report) => &report.id,
Self::RolledBack(report) => &report.id,
Self::RecoveryRequired(report) => &report.id,
}
}
pub fn is_committed(&self) -> bool {
matches!(self, Self::Committed(_))
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum RecoveryOutcome {
Discarded,
RolledBack,
Completed,
CompletedSetAside {
items: Vec<RetainedItem>,
},
Retained {
reason: String,
items: Vec<RetainedItem>,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct RecoveredTransaction {
pub id: String,
pub outcome: RecoveryOutcome,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct RecoveryReport {
pub transactions: Vec<RecoveredTransaction>,
}
impl RecoveryReport {
pub fn is_clean(&self) -> bool {
!self
.transactions
.iter()
.any(|t| matches!(t.outcome, RecoveryOutcome::Retained { .. }))
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct Ident {
key: FileKey,
len: u64,
modified_ns: u128,
}
impl Ident {
fn of_file(file: &File, shown: &Path) -> Result<Self> {
let key = FileKey::of_file(file, shown)?;
let meta = file.metadata().map_err(|e| FsError::io(shown, e))?;
let modified_ns = meta
.modified()
.ok()
.and_then(|time| time.duration_since(UNIX_EPOCH).ok())
.map_or(0, |elapsed| elapsed.as_nanos());
Ok(Self {
key,
len: meta.len(),
modified_ns,
})
}
fn render(&self) -> String {
format!(
"{:x}.{:x}.{}.{}",
self.key.volume, self.key.index, self.len, self.modified_ns
)
}
fn parse(text: &str) -> Option<Self> {
let mut parts = text.split('.');
let volume = u64::from_str_radix(parts.next()?, 16).ok()?;
let index = u64::from_str_radix(parts.next()?, 16).ok()?;
let len = parts.next()?.parse().ok()?;
let modified_ns = parts.next()?.parse().ok()?;
if parts.next().is_some() {
return None;
}
Some(Self {
key: FileKey { volume, index },
len,
modified_ns,
})
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
enum Probe {
Missing,
File(Ident),
Other(String),
}
fn is_link(meta: &fs::Metadata) -> bool {
if meta.file_type().is_symlink() {
return true;
}
#[cfg(windows)]
{
use std::os::windows::fs::MetadataExt;
const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x400;
if meta.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0 {
return true;
}
}
false
}
fn probe(path: &Path) -> Result<Probe> {
let meta = match fs::symlink_metadata(path) {
Ok(meta) => meta,
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(Probe::Missing),
Err(e) => return Err(FsError::io(path, e)),
};
if is_link(&meta) {
return Ok(Probe::Other("a symbolic link or reparse point".into()));
}
if !meta.is_file() {
return Ok(Probe::Other("not a regular file".into()));
}
let file = match File::open(path) {
Ok(file) => file,
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(Probe::Missing),
Err(e) => return Err(FsError::io(path, e)),
};
Ok(Probe::File(Ident::of_file(&file, path)?))
}
fn conflict(path: &Path, reason: &str) -> FsError {
FsError::Conflict {
path: path.to_path_buf(),
reason: reason.to_string(),
}
}
fn remove_if_exists(path: &Path) -> io::Result<()> {
match fs::remove_file(path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e),
}
}
fn transient_delete_error(error: &io::Error) -> bool {
#[cfg(windows)]
if let Some(code) = error.raw_os_error() {
return matches!(code, 5 | 32 | 33);
}
let _ = error;
false
}
fn transient_removal_error(error: &io::Error) -> bool {
transient_delete_error(error)
|| (cfg!(feature = "test-seams") && error.kind() == io::ErrorKind::ResourceBusy)
}
fn remove_while_identical(
path: &Path,
identical: impl Fn(&Ident) -> bool,
mismatch: &str,
retry: Option<RetryPolicy>,
) -> std::result::Result<(), RetainedItem> {
let stop: std::cell::RefCell<Option<RetainedItem>> = std::cell::RefCell::new(None);
let attempts = std::cell::Cell::new(0u32);
let result = retry_while(retry, transient_removal_error, || {
if attempts.get() > 0 {
if let Err(error) = fire(Step::DeleteBackoff, path) {
*stop.borrow_mut() = Some(RetainedItem::new(path, error));
return Err(io::Error::other("stopped before the next attempt"));
}
}
attempts.set(attempts.get() + 1);
match probe(path) {
Ok(Probe::Missing) => Ok(()),
Ok(Probe::File(now)) if identical(&now) => match forced_delete_failure() {
Some(error) => Err(error),
None => remove_if_exists(path),
},
Ok(_) => {
*stop.borrow_mut() = Some(RetainedItem::new(path, mismatch));
Err(io::Error::other("identity no longer proven"))
}
Err(error) => {
*stop.borrow_mut() = Some(RetainedItem::new(path, error));
Err(io::Error::other("identity cannot be read"))
}
}
});
if let Some(item) = stop.into_inner() {
return Err(item);
}
result.map_err(|e| RetainedItem::new(path, e))?;
sync_parent_strict(path).map_err(|e| RetainedItem::new(path, e))
}
#[cfg(test)]
fn remove_with_retry(path: &Path, retry: Option<RetryPolicy>) -> io::Result<()> {
retry_while(retry, transient_delete_error, || remove_if_exists(path))
}
fn retry_while(
retry: Option<RetryPolicy>,
transient: impl Fn(&io::Error) -> bool,
mut operation: impl FnMut() -> io::Result<()>,
) -> io::Result<()> {
let attempts = retry.map_or(0, |policy| policy.attempts);
let first = retry.map_or(Duration::ZERO, |policy| policy.delay);
let cap = first.max(Duration::from_secs(1));
let mut delay = first;
let mut tried = 0;
loop {
match operation() {
Err(error) if tried < attempts && transient(&error) => {
tried += 1;
std::thread::sleep(delay);
delay = delay.saturating_mul(2).min(cap);
}
other => return other,
}
}
}
fn sync_dir_strict(dir: &Path) -> Result<()> {
#[cfg(windows)]
let result = crate::atomic::flush_dir_windows(dir);
#[cfg(not(windows))]
let result = try_sync_dir(dir);
match result {
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
other => other.map_err(|e| FsError::io(dir, e)),
}
}
fn sync_parent_strict(path: &Path) -> Result<()> {
if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
sync_dir_strict(parent)?;
}
Ok(())
}
fn sync_parent(path: &Path) {
if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
let _ = try_sync_dir(parent);
}
}
fn fold(text: &str) -> String {
if cfg!(any(windows, target_vendor = "apple")) {
text.to_lowercase()
} else {
text.to_string()
}
}
fn bad_name(rel: &Path) -> FsError {
FsError::BadName(rel.display().to_string())
}
fn valid_component(name: &str) -> bool {
!name.is_empty()
&& name != "."
&& name != ".."
&& !name.contains(['/', '\0'])
&& !(cfg!(windows) && crate::names::is_unsafe_windows_component(name))
}
fn split_relative(rel: &Path) -> Result<Vec<String>> {
let text = rel.to_str().ok_or_else(|| bad_name(rel))?;
if text.len() > MAX_RELATIVE_BYTES || text.contains('\0') {
return Err(bad_name(rel));
}
let mut components = Vec::new();
for component in rel.components() {
match component {
Component::Normal(name) => {
let name = name.to_str().ok_or_else(|| bad_name(rel))?;
if !valid_component(name) {
return Err(bad_name(rel));
}
components.push(name.to_string());
}
Component::CurDir => {}
_ => return Err(bad_name(rel)),
}
}
if components.is_empty() || components.len() > MAX_COMPONENTS {
return Err(bad_name(rel));
}
if fold(&components[0]) == AREA {
return Err(bad_name(rel));
}
let leaf = &components[components.len() - 1];
crate::atomic::reject_reserved_name(leaf)?;
if leaf.contains(SIDE_MARKER) {
return Err(bad_name(rel));
}
Ok(components)
}
fn split_journal_rel(text: &str) -> Option<Vec<String>> {
if text.is_empty() || text.len() > MAX_RELATIVE_BYTES + 256 {
return None;
}
let components: Vec<String> = text.split('/').map(str::to_string).collect();
if components.len() > MAX_COMPONENTS + 1 || !components.iter().all(|c| valid_component(c)) {
return None;
}
Some(components)
}
fn join_rel(dir: &[String], name: &str) -> String {
let mut out = String::new();
for component in dir {
out.push_str(component);
out.push('/');
}
out.push_str(name);
out
}
fn resolve_entry_path(root: &Path, components: &[String]) -> Result<PathBuf> {
let mut current = root.to_path_buf();
let mut missing = false;
for (index, component) in components.iter().enumerate() {
current.push(component);
if missing {
continue;
}
match fs::symlink_metadata(¤t) {
Ok(meta) => {
if is_link(&meta) {
return Err(FsError::Symlink(current));
}
if index + 1 == components.len() {
if !meta.is_file() {
return Err(FsError::NotAFile(current));
}
} else if !meta.is_dir() {
return Err(FsError::NotADirectory(current));
}
}
Err(e) if e.kind() == io::ErrorKind::NotFound => missing = true,
Err(e) => return Err(FsError::io(¤t, e)),
}
}
Ok(current)
}
struct EntryPaths {
dest: PathBuf,
stage: PathBuf,
backup: PathBuf,
}
fn entry_paths(root: &Path, entry: &JournalEntry) -> Result<EntryPaths> {
let corrupt = |reason: &str| FsError::Corrupt {
path: root.join(AREA).join(JOURNAL),
reason: reason.to_string(),
};
let dest = split_journal_rel(&entry.dest)
.ok_or_else(|| corrupt("an entry destination is not a plain relative path"))?;
let stage = split_journal_rel(&entry.stage)
.ok_or_else(|| corrupt("an entry staged name is not a plain relative path"))?;
let backup = split_journal_rel(&entry.backup)
.ok_or_else(|| corrupt("an entry backup name is not a plain relative path"))?;
let dir = &dest[..dest.len() - 1];
if stage.len() != dest.len()
|| backup.len() != dest.len()
|| stage[..stage.len() - 1] != *dir
|| backup[..backup.len() - 1] != *dir
{
return Err(corrupt(
"an entry's staged and backup names are not beside its destination",
));
}
if fold(&dest[0]) == AREA {
return Err(corrupt(
"an entry destination is inside the transaction directory",
));
}
Ok(EntryPaths {
dest: resolve_entry_path(root, &dest)?,
stage: resolve_entry_path(root, &stage)?,
backup: resolve_entry_path(root, &backup)?,
})
}
fn bounded_prefix(name: &str) -> &str {
if name.len() <= SIDE_PREFIX_MAX {
return name;
}
let mut end = SIDE_PREFIX_MAX;
while !name.is_char_boundary(end) {
end -= 1;
}
&name[..end]
}
fn side_name(leaf: &str, id: &str, index: usize, kind: char) -> String {
format!(".{}{SIDE_MARKER}{id}-{index}{kind}", bounded_prefix(leaf))
}
fn new_id() -> String {
let millis = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |elapsed| elapsed.as_millis());
let count = ID_COUNTER.fetch_add(1, Ordering::Relaxed);
format!("{millis:012x}-{:x}-{count:x}", std::process::id())
}
fn valid_id(id: &str) -> bool {
!id.is_empty()
&& id.len() <= 64
&& id
.bytes()
.all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f' | b'-'))
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum JournalPhase {
Staging,
Staged,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct JournalEntry {
dest: String,
stage: String,
backup: String,
create_new: bool,
private: bool,
prior: Option<Ident>,
staged: Option<Ident>,
unpublished: bool,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct Journal {
id: String,
root: FileKey,
phase: JournalPhase,
entries: Vec<JournalEntry>,
crc: u32,
}
fn hex_encode(text: &str) -> String {
text.bytes().map(|b| format!("{b:02x}")).collect()
}
fn hex_decode(text: &str) -> Option<String> {
if !text.len().is_multiple_of(2) || !text.bytes().all(|b| b.is_ascii_hexdigit()) {
return None;
}
let bytes: Option<Vec<u8>> = (0..text.len())
.step_by(2)
.map(|at| u8::from_str_radix(text.get(at..at + 2)?, 16).ok())
.collect();
String::from_utf8(bytes?).ok()
}
fn render_ident(ident: &Option<Ident>) -> String {
ident
.as_ref()
.map_or_else(|| "-".to_string(), Ident::render)
}
fn parse_ident(text: &str) -> Option<Option<Ident>> {
if text == "-" {
Some(None)
} else {
Ident::parse(text).map(Some)
}
}
impl Journal {
fn render(&self) -> (String, u32) {
let mut out = String::new();
out.push_str(JOURNAL_MAGIC);
out.push('\n');
out.push_str(&format!("id {}\n", self.id));
out.push_str(&format!(
"root {:x}.{:x}\n",
self.root.volume, self.root.index
));
out.push_str(match self.phase {
JournalPhase::Staging => "phase staging\n",
JournalPhase::Staged => "phase staged\n",
});
out.push_str(&format!("entries {}\n", self.entries.len()));
for (index, entry) in self.entries.iter().enumerate() {
out.push_str(&format!(
"entry {index} {} {} {} {} {} {} {} {}\n",
if entry.create_new { 'N' } else { 'R' },
if entry.private { 1 } else { 0 },
hex_encode(&entry.dest),
hex_encode(&entry.stage),
hex_encode(&entry.backup),
render_ident(&entry.prior),
render_ident(&entry.staged),
if entry.unpublished { 1 } else { 0 },
));
}
let crc = crate::zip::crc32(out.as_bytes());
out.push_str(&format!("crc {crc:08x}\n"));
(out, crc)
}
fn parse(text: &str, shown: &Path) -> Result<Self> {
let corrupt = |reason: &str| FsError::Corrupt {
path: shown.to_path_buf(),
reason: reason.to_string(),
};
let split = text
.rfind("crc ")
.ok_or_else(|| corrupt("the checksum line is missing"))?;
if split != 0 && !text[..split].ends_with('\n') {
return Err(corrupt("the checksum line is misplaced"));
}
let (body, tail) = text.split_at(split);
let crc_text = tail
.strip_prefix("crc ")
.and_then(|rest| rest.strip_suffix('\n'))
.ok_or_else(|| corrupt("the checksum line is malformed"))?;
if crc_text.len() != 8 {
return Err(corrupt("the checksum is not eight hex digits"));
}
let crc =
u32::from_str_radix(crc_text, 16).map_err(|_| corrupt("the checksum is not hex"))?;
if crate::zip::crc32(body.as_bytes()) != crc {
return Err(corrupt("the checksum does not match the journal"));
}
let mut lines = body.lines();
if lines.next() != Some(JOURNAL_MAGIC) {
return Err(corrupt("not a version 1 journal"));
}
let mut field = |name: &str| -> Result<String> {
let line = lines
.next()
.ok_or_else(|| corrupt("the journal is truncated"))?;
line.strip_prefix(name)
.and_then(|rest| rest.strip_prefix(' '))
.map(str::to_string)
.ok_or_else(|| corrupt("a journal header line is out of place"))
};
let id = field("id")?;
if !valid_id(&id) {
return Err(corrupt("the transaction id is malformed"));
}
let root_text = field("root")?;
let root = {
let mut parts = root_text.split('.');
let volume = parts.next().and_then(|p| u64::from_str_radix(p, 16).ok());
let index = parts.next().and_then(|p| u64::from_str_radix(p, 16).ok());
match (volume, index, parts.next()) {
(Some(volume), Some(index), None) => FileKey { volume, index },
_ => return Err(corrupt("the root identity is malformed")),
}
};
let phase = match field("phase")?.as_str() {
"staging" => JournalPhase::Staging,
"staged" => JournalPhase::Staged,
_ => return Err(corrupt("the journal phase is unknown")),
};
let count: usize = field("entries")?
.parse()
.map_err(|_| corrupt("the entry count is malformed"))?;
if count == 0 || count > MAX_ENTRIES {
return Err(corrupt("the entry count is out of range"));
}
let mut entries = Vec::with_capacity(count);
for index in 0..count {
let line = field("entry")?;
let tokens: Vec<&str> = line.split(' ').collect();
if tokens.len() != 9 || tokens[0] != index.to_string() {
return Err(corrupt("an entry line is malformed or out of order"));
}
let create_new = match tokens[1] {
"N" => true,
"R" => false,
_ => return Err(corrupt("an entry mode is unknown")),
};
let private = match tokens[2] {
"1" => true,
"0" => false,
_ => return Err(corrupt("an entry privacy flag is unknown")),
};
let unpublished = match tokens[8] {
"1" => true,
"0" => false,
_ => return Err(corrupt("an entry rollback mark is unknown")),
};
let decode = |text: &str| {
hex_decode(text).ok_or_else(|| corrupt("an entry path is not hex of UTF-8"))
};
let prior = parse_ident(tokens[6])
.ok_or_else(|| corrupt("an entry prior identity is malformed"))?;
let staged = parse_ident(tokens[7])
.ok_or_else(|| corrupt("an entry staged identity is malformed"))?;
if phase == JournalPhase::Staged && staged.is_none() {
return Err(corrupt(
"a staged journal has an entry without a staged identity",
));
}
let dest = decode(tokens[3])?;
let stage = decode(tokens[4])?;
let backup = decode(tokens[5])?;
let components = split_journal_rel(&dest)
.ok_or_else(|| corrupt("an entry destination is not a plain relative path"))?;
let leaf = &components[components.len() - 1];
let dir = &components[..components.len() - 1];
if stage != join_rel(dir, &side_name(leaf, &id, index, 's'))
|| backup != join_rel(dir, &side_name(leaf, &id, index, 'b'))
{
return Err(corrupt(
"an entry side name does not match its transaction and index",
));
}
entries.push(JournalEntry {
dest,
stage,
backup,
create_new,
private,
prior,
staged,
unpublished,
});
}
if lines.next().is_some() {
return Err(corrupt("the journal has trailing lines"));
}
Ok(Self {
id,
root,
phase,
entries,
crc,
})
}
}
fn load_journal(path: &Path) -> Result<Journal> {
let file = File::open(path).map_err(|e| FsError::io(path, e))?;
let mut text = String::new();
file.take(MAX_JOURNAL_BYTES + 1)
.read_to_string(&mut text)
.map_err(|e| FsError::io(path, e))?;
if text.len() as u64 > MAX_JOURNAL_BYTES {
return Err(FsError::Corrupt {
path: path.to_path_buf(),
reason: "the journal is larger than any journal this crate writes".into(),
});
}
Journal::parse(&text, path)
}
fn marker_text(id: &str, journal_crc: u32) -> String {
format!("{COMMIT_MAGIC} {id} {journal_crc:08x}\n")
}
fn write_record(ctx: &Ctx<'_>, path: &Path, text: &str, replace: ReplaceMode) -> Result<()> {
let options = AtomicWriteOptions {
private: true,
create_parents: false,
durable: true,
replace,
retry_sharing_violation: ctx.options.retry,
privacy: Privacy::OwnerOnly,
};
let mut writer = AtomicWriter::create_with(path, options)?;
writer.write_all(text.as_bytes())?;
let claim = read_record_claim(writer.temp_path(), text)?;
let result = writer.commit();
if result.is_ok() || result.as_ref().is_err_and(|error| error.target_replaced()) {
ctx.claim_record(path, claim);
}
result
}
fn write_journal(
ctx: &Ctx<'_>,
journal: &mut Journal,
before: Step,
after: Step,
replace: ReplaceMode,
) -> Result<()> {
let path = ctx.txn_dir.join(JOURNAL);
fire(before, &path)?;
let (text, crc) = journal.render();
journal.crc = crc;
write_record(ctx, &path, &text, replace)?;
fire(after, &path)
}
#[derive(Clone, Debug, PartialEq, Eq)]
enum EntryState {
Untouched,
LinkedBackup,
AsideBackup,
Published { backup: bool },
Conflict(String),
}
#[cfg(test)]
fn classify(
prior: Option<&Ident>,
staged: &Ident,
dest: &Probe,
stage: &Probe,
backup: &Probe,
) -> EntryState {
classify_entry(prior, staged, dest, stage, backup, false)
}
fn classify_entry(
prior: Option<&Ident>,
staged: &Ident,
dest: &Probe,
stage: &Probe,
backup: &Probe,
unpublished: bool,
) -> EntryState {
for (name, state) in [
("destination", dest),
("staged copy", stage),
("backup", backup),
] {
if let Probe::Other(why) = state {
return EntryState::Conflict(format!("the {name} is {why}"));
}
}
if let Probe::File(now) = stage {
if now != staged {
return EntryState::Conflict(
"the staged copy no longer matches its recorded identity".into(),
);
}
}
match (backup, prior) {
(Probe::File(now), Some(prior)) if now != prior => {
return EntryState::Conflict(
"the backup no longer matches the prior file's recorded identity".into(),
);
}
(Probe::File(_), None) => {
return EntryState::Conflict(
"a backup exists for a destination that did not exist".into(),
);
}
_ => {}
}
let backup_present = matches!(backup, Probe::File(_));
let dest_is_staged = matches!(dest, Probe::File(now) if now == staged);
let dest_is_prior = match (dest, prior) {
(Probe::File(now), Some(prior)) => now == prior,
_ => false,
};
if dest_is_staged && dest_is_prior {
return EntryState::Conflict(
"the staged and prior identities are indistinguishable".into(),
);
}
if dest_is_staged {
return match (prior, backup_present) {
(Some(_), false) => EntryState::Conflict(
"the destination holds the staged content but the prior file's backup is missing"
.into(),
),
(Some(_), true) => EntryState::Published { backup: true },
(None, _) => EntryState::Published { backup: false },
};
}
match (prior, dest) {
(Some(_), _) if dest_is_prior => {
if backup_present {
EntryState::LinkedBackup
} else {
EntryState::Untouched
}
}
(Some(_), Probe::Missing) => {
if backup_present {
EntryState::AsideBackup
} else {
EntryState::Conflict(
"the destination and the prior file's backup are both missing".into(),
)
}
}
(Some(_), Probe::File(_))
if !backup_present && (matches!(stage, Probe::File(_)) || unpublished) =>
{
EntryState::Untouched
}
(Some(_), _) => EntryState::Conflict("the destination was changed by someone else".into()),
(None, Probe::Missing) => EntryState::Untouched,
(None, Probe::File(_)) if matches!(stage, Probe::File(_)) || unpublished => {
EntryState::Untouched
}
(None, _) => EntryState::Conflict(
"a destination that did not exist was created by someone else".into(),
),
}
}
#[derive(Clone, Copy, Debug)]
pub struct FileTransaction;
#[derive(Debug)]
struct PlannedEntry {
rel: String,
components: Vec<String>,
candidate: PathBuf,
candidate_ident: Ident,
parents: Vec<(PathBuf, FileKey)>,
expect: ExpectedPrior,
create_new: bool,
private: bool,
}
#[derive(Debug)]
pub struct PreparedTransaction {
root: PathBuf,
requested_root: PathBuf,
root_key: FileKey,
entries: Vec<PlannedEntry>,
options: TransactionOptions,
}
impl FileTransaction {
pub fn prepare(
root: &Path,
entries: Vec<TransactionEntry>,
options: TransactionOptions,
) -> Result<PreparedTransaction> {
let requested_root = std::path::absolute(root).map_err(|e| FsError::io(root, e))?;
let root = canonicalize_plain(root).map_err(|e| FsError::io(root, e))?;
let root_key = FileKey::of_path(&root)?;
let meta = fs::metadata(&root).map_err(|e| FsError::io(&root, e))?;
if !meta.is_dir() {
return Err(FsError::NotADirectory(root));
}
if entries.is_empty() {
return Err(FsError::io(
&root,
io::Error::new(
io::ErrorKind::InvalidInput,
"a transaction needs at least one entry",
),
));
}
if entries.len() > MAX_ENTRIES {
return Err(FsError::TooManyEntries(MAX_ENTRIES));
}
let mut planned: Vec<PlannedEntry> = Vec::with_capacity(entries.len());
let mut folded: Vec<Vec<String>> = Vec::with_capacity(entries.len());
for entry in entries {
let components = split_relative(&entry.destination)?;
let key: Vec<String> = components.iter().map(|c| fold(c)).collect();
for other in &folded {
let shared = key.len().min(other.len());
if key[..shared] == other[..shared] {
return Err(bad_name(&entry.destination));
}
}
folded.push(key);
if entry.publication == Publication::CreateNew && entry.expect != ExpectedPrior::Absent
{
return Err(FsError::io(
&entry.destination,
io::Error::new(
io::ErrorKind::InvalidInput,
"a create-new entry can only expect an absent destination",
),
));
}
let candidate_ident = match probe(&entry.candidate)? {
Probe::File(ident) => ident,
Probe::Missing => {
return Err(FsError::io(
&entry.candidate,
io::Error::from(io::ErrorKind::NotFound),
))
}
Probe::Other(_) => return Err(FsError::NotAFile(entry.candidate)),
};
let mut parents = vec![(root.clone(), root_key)];
let mut parent = root.clone();
for component in components.iter().take(components.len() - 1) {
parent.push(component);
match fs::symlink_metadata(&parent) {
Ok(meta) if is_link(&meta) => return Err(FsError::Symlink(parent)),
Ok(meta) if meta.is_dir() => {
parents.push((parent.clone(), FileKey::of_path(&parent)?));
}
Ok(_) => return Err(FsError::NotADirectory(parent)),
Err(error) if error.kind() == io::ErrorKind::NotFound => break,
Err(error) => return Err(FsError::io(&parent, error)),
}
}
let planned_entry = PlannedEntry {
parents,
rel: components.join("/"),
components,
candidate: entry.candidate,
candidate_ident,
expect: entry.expect,
create_new: entry.publication == Publication::CreateNew,
private: entry.private,
};
let dest = resolve_entry_path(&root, &planned_entry.components)?;
check_expectation(&dest, &planned_entry)?;
planned.push(planned_entry);
}
Ok(PreparedTransaction {
root,
requested_root,
root_key,
entries: planned,
options,
})
}
}
fn check_expectation(dest: &Path, entry: &PlannedEntry) -> Result<Option<Ident>> {
let now = probe(dest)?;
let ident = match now {
Probe::Other(_) => return Err(FsError::NotAFile(dest.to_path_buf())),
Probe::Missing => None,
Probe::File(ident) => Some(ident),
};
if entry.create_new && ident.is_some() {
return Err(FsError::AlreadyExists(dest.to_path_buf()));
}
match (&entry.expect, ident) {
(ExpectedPrior::Any, found) => Ok(found),
(ExpectedPrior::Absent, None) => Ok(None),
(ExpectedPrior::Absent, Some(_)) => Err(conflict(
dest,
"the destination exists but was expected to be absent",
)),
(ExpectedPrior::Identity(_), None) => Err(conflict(
dest,
"the destination is gone but was expected to exist",
)),
(ExpectedPrior::Identity(key), Some(found)) => {
if found.key == *key {
Ok(Some(found))
} else {
Err(conflict(
dest,
"the destination is not the file that was expected (its identity differs)",
))
}
}
#[cfg(feature = "digest")]
(ExpectedPrior::Sha256(_), None) => Err(conflict(
dest,
"the destination is gone but was expected to exist",
)),
#[cfg(feature = "digest")]
(ExpectedPrior::Sha256(expected), Some(found)) => {
let actual = crate::digest::sha256_file(dest).map_err(|e| FsError::io(dest, e))?;
if !actual.eq_ignore_ascii_case(expected) {
return Err(conflict(
dest,
"the destination's content hash is not the expected one",
));
}
match probe(dest)? {
Probe::File(again) if again == found => Ok(Some(found)),
_ => Err(conflict(
dest,
"the destination changed while it was being hashed",
)),
}
}
}
}
fn verify_parent_identities(parents: &[(PathBuf, FileKey)]) -> Result<()> {
for (path, expected) in parents {
let meta = fs::symlink_metadata(path).map_err(|e| FsError::io(path, e))?;
if is_link(&meta) || !meta.is_dir() || FileKey::of_path(path)? != *expected {
return Err(conflict(
path,
"a directory recorded during preparation was substituted",
));
}
}
Ok(())
}
impl PreparedTransaction {
fn verify_prepared_directories(&self) -> Result<()> {
let resolved = canonicalize_plain(&self.requested_root)
.map_err(|e| FsError::io(&self.requested_root, e))?;
if resolved != self.root || FileKey::of_path(&resolved)? != self.root_key {
return Err(conflict(
&self.requested_root,
"the prepared root was substituted",
));
}
for entry in &self.entries {
verify_parent_identities(&entry.parents)?;
}
Ok(())
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn destinations(&self) -> Vec<PathBuf> {
self.entries
.iter()
.map(|entry| {
let mut path = self.root.clone();
path.extend(&entry.components);
path
})
.collect()
}
pub fn commit(self) -> Result<TransactionReport> {
self.verify_prepared_directories()?;
let area = self.root.join(AREA);
create_dir_all_durable(&area, true)?;
let _lock = FileLock::acquire(&area.join(LOCK_NAME), &lock_options(&self.options))?;
if let Some(pending) = first_pending(&area)? {
return Err(FsError::Locked {
path: area.join(&pending),
owner: format!(
"an earlier transaction ({pending}) awaits recovery; call recover_transactions"
),
});
}
self.verify_prepared_directories()?;
let root_key = self.root_key;
let id = new_id();
let mut entries = Vec::with_capacity(self.entries.len());
for (index, planned) in self.entries.iter().enumerate() {
entries.push(plan_entry(&self.root, &id, index, planned)?);
}
let mut journal = Journal {
id: id.clone(),
root: root_key,
phase: JournalPhase::Staging,
entries,
crc: 0,
};
let txn_dir = area.join(&id);
if let Err(error) = create_dir_all_durable(&txn_dir, true) {
let _ = fs::remove_dir(&txn_dir);
return Err(error);
}
let ctx = Ctx {
records: std::cell::RefCell::new(Vec::new()),
root: &self.root,
options: &self.options,
txn_dir: &txn_dir,
};
run_commit(&ctx, &self.entries, &mut journal)
}
}
fn lock_options(options: &TransactionOptions) -> LockOptions {
LockOptions::wait(options.lock_timeout).labeled("rightkit-fs transaction")
}
fn first_pending(area: &Path) -> Result<Option<String>> {
let mut pending: Vec<String> = Vec::new();
for entry in fs::read_dir(area).map_err(|e| FsError::io(area, e))? {
let entry = entry.map_err(|e| FsError::io(area, e))?;
let file_type = entry
.file_type()
.map_err(|e| FsError::io(entry.path(), e))?;
let name = entry.file_name().to_string_lossy().into_owned();
if file_type.is_dir() && name != RETAINED_DIR {
pending.push(name);
}
}
pending.sort();
Ok(pending.into_iter().next())
}
fn plan_entry(root: &Path, id: &str, index: usize, planned: &PlannedEntry) -> Result<JournalEntry> {
let dest = resolve_entry_path(root, &planned.components)?;
let prior = check_expectation(&dest, planned)?;
match probe(&planned.candidate)? {
Probe::File(now) if now == planned.candidate_ident => {}
Probe::File(_) => {
return Err(conflict(
&planned.candidate,
"the candidate file changed after the transaction was prepared",
))
}
Probe::Missing => return Err(conflict(&planned.candidate, "the candidate file is gone")),
Probe::Other(_) => return Err(FsError::NotAFile(planned.candidate.clone())),
}
let (dir, leaf) = planned.components.split_at(planned.components.len() - 1);
let leaf = &leaf[0];
Ok(JournalEntry {
dest: planned.rel.clone(),
stage: join_rel(dir, &side_name(leaf, id, index, 's')),
backup: join_rel(dir, &side_name(leaf, id, index, 'b')),
create_new: planned.create_new,
private: planned.private,
prior,
staged: None,
unpublished: false,
})
}
#[derive(Clone)]
struct RecordClaim {
ident: Ident,
text: String,
}
fn read_record_claim(path: &Path, expected: &str) -> Result<RecordClaim> {
if !matches!(probe(path)?, Probe::File(_)) {
return Err(conflict(path, "the protocol record is not a regular file"));
}
let mut file = File::open(path).map_err(|e| FsError::io(path, e))?;
let before = Ident::of_file(&file, path)?;
let mut text = String::new();
(&mut file)
.take(MAX_JOURNAL_BYTES + 1)
.read_to_string(&mut text)
.map_err(|e| FsError::io(path, e))?;
if text != expected || Ident::of_file(&file, path)? != before {
return Err(conflict(
path,
"the protocol record's content changed or is unrecognized",
));
}
Ok(RecordClaim {
ident: before,
text,
})
}
struct Ctx<'a> {
records: std::cell::RefCell<Vec<(String, RecordClaim)>>,
root: &'a Path,
options: &'a TransactionOptions,
txn_dir: &'a Path,
}
impl Ctx<'_> {
fn claim_record(&self, path: &Path, claim: RecordClaim) {
let name = path
.file_name()
.expect("a protocol record has a name")
.to_string_lossy()
.into_owned();
let mut records = self.records.borrow_mut();
if let Some(record) = records.iter_mut().find(|record| record.0 == name) {
record.1 = claim;
} else {
records.push((name, claim));
}
}
fn record_claim(&self, name: &str) -> Option<RecordClaim> {
self.records
.borrow()
.iter()
.find(|record| record.0 == name)
.map(|record| record.1.clone())
}
}
fn run_commit(
ctx: &Ctx<'_>,
planned: &[PlannedEntry],
journal: &mut Journal,
) -> Result<TransactionReport> {
if let Err(cause) = begin_staging(ctx, planned, journal) {
return abandon(ctx, journal, cause);
}
for (index, planned_entry) in planned.iter().enumerate() {
if let Err(cause) = verify_parent_identities(&planned_entry.parents)
.and_then(|()| apply_entry(ctx, index, &journal.entries[index]))
{
return Ok(roll_back_here(ctx, journal, cause));
}
}
let marker_path = ctx.txn_dir.join(COMMIT);
if let Err(cause) = fire(Step::BeforeCommitMarker, &marker_path) {
return Ok(roll_back_here(ctx, journal, cause));
}
let marker = marker_text(&journal.id, journal.crc);
if let Err(cause) = write_record(ctx, &marker_path, &marker, ReplaceMode::NoReplace) {
return Ok(TransactionReport::RecoveryRequired(
RecoveryRequiredReport {
id: journal.id.clone(),
cause,
retained: vec![RetainedItem::new(
&marker_path,
"the commit marker could not be confirmed; recover_transactions decides",
)],
journal: ctx.txn_dir.join(JOURNAL),
},
));
}
let published: Vec<PathBuf> = journal
.entries
.iter()
.filter_map(|entry| entry_paths(ctx.root, entry).ok().map(|paths| paths.dest))
.collect();
let retained = match fire(Step::AfterCommitMarker, &marker_path) {
Ok(()) => complete_committed(ctx, journal).into_items(),
Err(error) => vec![RetainedItem::new(ctx.txn_dir, error)],
};
Ok(TransactionReport::Committed(CommittedReport {
id: journal.id.clone(),
published,
retained,
}))
}
fn begin_staging(ctx: &Ctx<'_>, planned: &[PlannedEntry], journal: &mut Journal) -> Result<()> {
write_journal(
ctx,
journal,
Step::BeforeJournalStaging,
Step::AfterJournalStaging,
ReplaceMode::NoReplace,
)?;
for (index, planned_entry) in planned.iter().enumerate() {
let staged = stage_entry(ctx, index, planned_entry, journal)?;
journal.entries[index].staged = Some(staged);
}
journal.phase = JournalPhase::Staged;
write_journal(
ctx,
journal,
Step::BeforeJournalStaged,
Step::AfterJournalStaged,
ReplaceMode::MoveWriteThrough,
)
}
fn abandon(ctx: &Ctx<'_>, journal: &Journal, cause: FsError) -> Result<TransactionReport> {
let retained = discard(ctx, journal);
if retained.is_empty() {
Err(cause)
} else {
Ok(TransactionReport::RecoveryRequired(
RecoveryRequiredReport {
id: journal.id.clone(),
cause,
retained,
journal: ctx.txn_dir.join(JOURNAL),
},
))
}
}
fn roll_back_here(ctx: &Ctx<'_>, journal: &mut Journal, cause: FsError) -> TransactionReport {
let outcome = rollback(ctx, journal);
if outcome.retained.is_empty() {
TransactionReport::RolledBack(RolledBackReport {
id: journal.id.clone(),
cause,
restored: outcome.restored,
})
} else {
TransactionReport::RecoveryRequired(RecoveryRequiredReport {
id: journal.id.clone(),
cause,
retained: outcome.retained,
journal: ctx.txn_dir.join(JOURNAL),
})
}
}
fn stage_entry(
ctx: &Ctx<'_>,
index: usize,
planned: &PlannedEntry,
journal: &mut Journal,
) -> Result<Ident> {
let entry = journal.entries[index].clone();
let entry = &entry;
let paths = entry_paths(ctx.root, entry)?;
fire(Step::BeforeStage(index), &paths.stage)?;
verify_parent_identities(&planned.parents)?;
if let Some(parent) = paths.dest.parent() {
match fs::symlink_metadata(parent) {
Ok(_) => {}
Err(e) if e.kind() == io::ErrorKind::NotFound => {
create_dir_all_durable(parent, entry.private)?;
}
Err(e) => return Err(FsError::io(parent, e)),
}
}
let paths = entry_paths(ctx.root, entry)?;
let staged = {
let mut claim = |key: FileKey| -> Result<()> {
journal.entries[index].staged = Some(Ident {
key,
len: 0,
modified_ns: 0,
});
write_journal(
ctx,
journal,
Step::BeforeStageClaim(index),
Step::AfterStageClaim(index),
ReplaceMode::MoveWriteThrough,
)
};
stage_copy(ctx, planned, entry, &paths, &mut claim)?
};
fire(Step::AfterStage(index), &paths.stage)?;
Ok(staged)
}
fn stage_copy(
ctx: &Ctx<'_>,
planned: &PlannedEntry,
entry: &JournalEntry,
paths: &EntryPaths,
claim: &mut dyn FnMut(FileKey) -> Result<()>,
) -> Result<Ident> {
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
let mut source =
File::open(&planned.candidate).map_err(|e| FsError::io(&planned.candidate, e))?;
let before = Ident::of_file(&source, &planned.candidate)?;
if before != planned.candidate_ident {
return Err(conflict(
&planned.candidate,
"the candidate file changed after the transaction was prepared",
));
}
let options = AtomicWriteOptions {
private: entry.private,
create_parents: false,
durable: true,
replace: ReplaceMode::NoReplace,
retry_sharing_violation: ctx.options.retry,
privacy: if entry.private {
Privacy::OwnerOnly
} else {
Privacy::Inherit
},
};
let mut writer = AtomicWriter::create_with(&paths.stage, options)?;
let claimed = FileKey::of_file(writer.file_mut(), &paths.stage)?;
claim(claimed)?;
let mut buffer = vec![0u8; COPY_BUFFER];
let mut copied = 0u64;
loop {
let read = match source.read(&mut buffer) {
Ok(0) => break,
Ok(read) => read,
Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) => return Err(FsError::io(&planned.candidate, e)),
};
writer.write_all(&buffer[..read])?;
copied += read as u64;
}
#[cfg(unix)]
if !entry.private {
let candidate_mode = source
.metadata()
.map(|meta| meta.permissions().mode())
.unwrap_or(0o644);
let prior_or_candidate = if entry.prior.is_some() {
fs::symlink_metadata(&paths.dest)
.map(|meta| meta.permissions().mode())
.unwrap_or(candidate_mode)
} else {
candidate_mode
};
let mode = prior_or_candidate & 0o7777;
let temp = writer.temp_path().to_path_buf();
writer
.file_mut()
.set_permissions(fs::Permissions::from_mode(mode))
.map_err(|e| FsError::io(&temp, e))?;
}
let after = Ident::of_file(&source, &planned.candidate)?;
if after != before || copied != before.len {
return Err(conflict(
&planned.candidate,
"the candidate file changed while it was being staged",
));
}
writer.commit()?;
match probe(&paths.stage)? {
Probe::File(ident) if ident.len == copied && ident.key == claimed => Ok(ident),
_ => Err(conflict(
&paths.stage,
"the staged copy is not what was written",
)),
}
}
enum BackupKind {
Link,
Aside,
}
fn link_unsupported(error: &io::Error) -> bool {
if error.kind() == io::ErrorKind::Unsupported {
return true;
}
#[cfg(unix)]
if let Some(code) = error.raw_os_error() {
return matches!(
code,
libc::EPERM | libc::ENOTSUP | libc::ENOSYS | libc::EINVAL | libc::EMLINK
);
}
#[cfg(windows)]
if let Some(code) = error.raw_os_error() {
return matches!(code, 1 | 50);
}
false
}
fn rename_durable_exclusive(from: &Path, to: &Path) -> Result<()> {
publish_file_with(
from,
to,
AtomicWriteOptions {
durable: true,
create_parents: false,
replace: ReplaceMode::NoReplace,
..AtomicWriteOptions::default()
},
)?;
sync_parent_strict(from)?;
sync_parent_strict(to)
}
fn make_backup(dest: &Path, backup: &Path) -> Result<BackupKind> {
if aside_forced() {
rename_durable_exclusive(dest, backup)?;
return Ok(BackupKind::Aside);
}
match fs::hard_link(dest, backup) {
Ok(()) => Ok(BackupKind::Link),
Err(e) if link_unsupported(&e) => {
rename_durable_exclusive(dest, backup)?;
Ok(BackupKind::Aside)
}
Err(e) => Err(FsError::io(backup, e)),
}
}
fn verify_backup(paths: &EntryPaths, prior: &Ident, kind: &BackupKind) -> Result<()> {
let backup = probe(&paths.backup)?;
if matches!(&backup, Probe::File(now) if now == prior) {
return Ok(());
}
if let (BackupKind::Link, Probe::File(linked)) = (kind, &backup) {
if matches!(probe(&paths.dest), Ok(Probe::File(now)) if now == *linked) {
let _ = remove_if_exists(&paths.backup);
sync_parent(&paths.backup);
}
}
if let (BackupKind::Aside, Probe::File(_)) = (kind, &backup) {
if rename_durable_exclusive(&paths.backup, &paths.dest).is_ok() {
sync_parent(&paths.dest);
}
}
Err(conflict(
&paths.dest,
"the destination changed while its backup was being made",
))
}
fn publish_step(ctx: &Ctx<'_>, from: &Path, to: &Path, replace: ReplaceMode) -> Result<()> {
let options = AtomicWriteOptions {
private: false,
create_parents: false,
durable: true,
replace,
retry_sharing_violation: ctx.options.retry,
privacy: Privacy::Inherit,
};
match ctx.options.access_denied {
Some(retry) => publish_file_with_access_denied_retry(from, to, options, retry),
None => publish_file_with(from, to, options),
}
}
fn apply_entry(ctx: &Ctx<'_>, index: usize, entry: &JournalEntry) -> Result<()> {
let paths = entry_paths(ctx.root, entry)?;
let staged = entry.staged.ok_or_else(|| FsError::Corrupt {
path: ctx.txn_dir.join(JOURNAL),
reason: "an entry has no staged identity".into(),
})?;
let mut replace = ReplaceMode::NoReplace;
match &entry.prior {
Some(prior) => {
if entry.create_new {
return Err(FsError::AlreadyExists(paths.dest));
}
fire(Step::BeforeBackup(index), &paths.dest)?;
match probe(&paths.dest)? {
Probe::File(now) if now == *prior => {}
_ => {
return Err(conflict(
&paths.dest,
"the destination changed after the transaction recorded it",
))
}
}
fire(Step::BeforeLink(index), &paths.dest)?;
let kind = make_backup(&paths.dest, &paths.backup)?;
verify_backup(&paths, prior, &kind)?;
if let BackupKind::Link = kind {
let mut open = fs::OpenOptions::new();
open.read(true);
#[cfg(windows)]
open.write(true);
let backup = open
.open(&paths.backup)
.map_err(|e| FsError::io(&paths.backup, e))?;
if Ident::of_file(&backup, &paths.backup)? != *prior {
return Err(conflict(
&paths.backup,
"the opened backup changed before its flush",
));
}
crate::atomic::durable_sync_with(&backup, crate::atomic::SyncPolicy::Fallback)
.map_err(|e| FsError::io(&paths.backup, e))?;
sync_parent_strict(&paths.backup)?;
verify_backup(&paths, prior, &kind)?;
replace = ReplaceMode::MoveWriteThrough;
}
fire(Step::AfterBackup(index), &paths.backup)?;
}
None => {
if probe(&paths.dest)? != Probe::Missing {
return Err(FsError::AlreadyExists(paths.dest));
}
}
}
fire(Step::BeforePublish(index), &paths.dest)?;
match probe(&paths.stage)? {
Probe::File(now) if now == staged => {}
_ => {
return Err(conflict(
&paths.stage,
"the staged copy changed before it was published",
))
}
}
publish_step(ctx, &paths.stage, &paths.dest, replace)?;
fire(Step::AfterPublish(index), &paths.dest)
}
struct RollbackOutcome {
restored: Vec<PathBuf>,
retained: Vec<RetainedItem>,
}
fn remove_matching(
ctx: &Ctx<'_>,
path: &Path,
expect: &Ident,
) -> std::result::Result<(), RetainedItem> {
remove_while_identical(
path,
|now| now == expect,
"it no longer matches the identity the transaction recorded, so it was not removed",
ctx.options.retry,
)
}
fn rollback(ctx: &Ctx<'_>, journal: &mut Journal) -> RollbackOutcome {
let mut outcome = RollbackOutcome {
restored: Vec::new(),
retained: Vec::new(),
};
for index in (0..journal.entries.len()).rev() {
let entry = journal.entries[index].clone();
match rollback_entry(ctx, index, &entry, journal) {
Ok(Some(restored)) => outcome.restored.push(restored),
Ok(None) => {}
Err(item) => outcome.retained.push(item),
}
}
outcome.restored.reverse();
if outcome.retained.is_empty() {
for entry in &journal.entries {
if let Err(error) = sync_parent_strict(&ctx.root.join(&entry.dest)) {
outcome
.retained
.push(RetainedItem::new(ctx.root.join(&entry.dest), error));
}
}
}
if outcome.retained.is_empty() {
outcome.retained.extend(finish_dir(ctx));
}
outcome
}
fn mark_unpublished(ctx: &Ctx<'_>, index: usize, journal: &mut Journal) -> Result<()> {
journal.entries[index].unpublished = true;
write_journal(
ctx,
journal,
Step::BeforeRollbackMark(index),
Step::AfterRollbackMark(index),
ReplaceMode::MoveWriteThrough,
)
}
fn rollback_entry(
ctx: &Ctx<'_>,
index: usize,
entry: &JournalEntry,
journal: &mut Journal,
) -> std::result::Result<Option<PathBuf>, RetainedItem> {
let shown = ctx.root.join(&entry.dest);
let paths = entry_paths(ctx.root, entry).map_err(|e| RetainedItem::new(&shown, e))?;
let staged = entry
.staged
.ok_or_else(|| RetainedItem::new(&paths.dest, "the entry has no staged identity"))?;
let dest = probe(&paths.dest).map_err(|e| RetainedItem::new(&paths.dest, e))?;
let stage = probe(&paths.stage).map_err(|e| RetainedItem::new(&paths.stage, e))?;
let backup = probe(&paths.backup).map_err(|e| RetainedItem::new(&paths.backup, e))?;
let state = classify_entry(
entry.prior.as_ref(),
&staged,
&dest,
&stage,
&backup,
entry.unpublished,
);
let step_error = |path: &Path, error: FsError| RetainedItem::new(path, error);
let external = match &dest {
Probe::File(now) => entry.prior.as_ref() != Some(now),
_ => false,
};
if matches!(state, EntryState::Untouched) && external && !entry.unpublished {
mark_unpublished(ctx, index, journal).map_err(|e| step_error(&paths.dest, e))?;
}
let restored = match state {
EntryState::Conflict(reason) => return Err(RetainedItem::new(&paths.dest, reason)),
EntryState::Untouched => None,
EntryState::LinkedBackup => {
let prior = entry.prior.as_ref().ok_or_else(|| {
RetainedItem::new(&paths.backup, "a linked backup without a prior identity")
})?;
fire(Step::BeforeRestore(index), &paths.backup)
.map_err(|e| step_error(&paths.backup, e))?;
remove_matching(ctx, &paths.backup, prior)?;
fire(Step::AfterRestore(index), &paths.backup)
.map_err(|e| step_error(&paths.backup, e))?;
None
}
EntryState::AsideBackup => {
fire(Step::BeforeRestore(index), &paths.dest)
.map_err(|e| step_error(&paths.dest, e))?;
publish_step(ctx, &paths.backup, &paths.dest, ReplaceMode::NoReplace)
.map_err(|e| step_error(&paths.dest, e))?;
sync_parent_strict(&paths.dest).map_err(|e| step_error(&paths.dest, e))?;
fire(Step::AfterRestore(index), &paths.dest).map_err(|e| step_error(&paths.dest, e))?;
Some(paths.dest.clone())
}
EntryState::Published { backup: true } => {
fire(Step::BeforeRestore(index), &paths.dest)
.map_err(|e| step_error(&paths.dest, e))?;
publish_step(
ctx,
&paths.backup,
&paths.dest,
ReplaceMode::MoveWriteThrough,
)
.map_err(|e| step_error(&paths.dest, e))?;
fire(Step::AfterRestore(index), &paths.dest).map_err(|e| step_error(&paths.dest, e))?;
Some(paths.dest.clone())
}
EntryState::Published { backup: false } => {
fire(Step::BeforeRestore(index), &paths.dest)
.map_err(|e| step_error(&paths.dest, e))?;
remove_matching(ctx, &paths.dest, &staged)?;
fire(Step::AfterRestore(index), &paths.dest).map_err(|e| step_error(&paths.dest, e))?;
Some(paths.dest.clone())
}
};
remove_matching(ctx, &paths.stage, &staged)?;
Ok(restored)
}
struct Cleanup {
blocking: Vec<RetainedItem>,
set_aside: Vec<RetainedItem>,
}
impl Cleanup {
fn note(&mut self, outcome: std::result::Result<Option<RetainedItem>, RetainedItem>) {
match outcome {
Ok(Some(moved)) => self.set_aside.push(moved),
Ok(None) => {}
Err(item) => self.blocking.push(item),
}
}
fn into_items(self) -> Vec<RetainedItem> {
let mut items = self.blocking;
items.extend(self.set_aside);
items
}
}
fn complete_committed(ctx: &Ctx<'_>, journal: &Journal) -> Cleanup {
let mut cleanup = Cleanup {
blocking: Vec::new(),
set_aside: Vec::new(),
};
for (index, entry) in journal.entries.iter().enumerate() {
let shown = ctx.root.join(&entry.dest);
let paths = match entry_paths(ctx.root, entry) {
Ok(paths) => paths,
Err(error) => {
cleanup.note(Err(RetainedItem::new(&shown, error)));
continue;
}
};
if let Some(prior) = &entry.prior {
let attempt = fire(Step::BeforeCleanup(index), &paths.backup)
.map_err(|e| RetainedItem::new(&paths.backup, e))
.and_then(|()| finish_leftover(ctx, &paths.backup, prior))
.and_then(|moved| {
fire(Step::AfterCleanup(index), &paths.backup)
.map_err(|e| RetainedItem::new(&paths.backup, e))
.map(|()| moved)
});
cleanup.note(attempt);
}
if let Some(staged) = &entry.staged {
cleanup.note(finish_leftover(ctx, &paths.stage, staged));
}
}
if cleanup.blocking.is_empty() {
cleanup.blocking.extend(finish_dir(ctx));
}
cleanup
}
fn finish_leftover(
ctx: &Ctx<'_>,
path: &Path,
expect: &Ident,
) -> std::result::Result<Option<RetainedItem>, RetainedItem> {
match probe(path) {
Ok(Probe::File(now)) if now != *expect => set_aside(ctx, path).map(Some),
Ok(Probe::Missing) => resume_set_aside(ctx, path),
_ => remove_matching(ctx, path, expect).map(|()| None),
}
}
fn set_aside_target(ctx: &Ctx<'_>, path: &Path) -> Option<(PathBuf, PathBuf)> {
let (id, name) = (ctx.txn_dir.file_name()?, path.file_name()?);
let dir = ctx.root.join(AREA).join(RETAINED_DIR).join(id);
let target = dir.join(name);
Some((dir, target))
}
fn set_aside_item(path: &Path, target: &Path, how: &str) -> RetainedItem {
RetainedItem::new(
target,
format!(
"{how} from {}: it no longer matched what the transaction recorded",
path.display()
),
)
}
fn set_aside(ctx: &Ctx<'_>, path: &Path) -> std::result::Result<RetainedItem, RetainedItem> {
let Some((dir, target)) = set_aside_target(ctx, path) else {
return Err(RetainedItem::new(path, "it has no name to set aside under"));
};
create_dir_all_durable(&dir, true).map_err(|e| RetainedItem::new(path, e))?;
rename_durable_exclusive(path, &target).map_err(|e| RetainedItem::new(path, e))?;
sync_set_aside(path, &target)?;
Ok(set_aside_item(path, &target, "moved here"))
}
fn resume_set_aside(
ctx: &Ctx<'_>,
path: &Path,
) -> std::result::Result<Option<RetainedItem>, RetainedItem> {
let Some((_, target)) = set_aside_target(ctx, path) else {
return Ok(None);
};
let looked = fire(Step::SetAsideProbe, &target)
.and_then(|()| fs::symlink_metadata(&target).map_err(|e| FsError::io(&target, e)));
match looked {
Ok(_) => {}
Err(error) if error.is_not_found() => return Ok(None),
Err(error) => return Err(RetainedItem::new(&target, error)),
}
sync_set_aside(path, &target)?;
Ok(Some(set_aside_item(path, &target, "moved here earlier")))
}
fn sync_set_aside(path: &Path, target: &Path) -> std::result::Result<(), RetainedItem> {
let failed = |dir: &Path, error: FsError| {
RetainedItem::new(
target,
format!(
"moved from {} but {} could not be flushed ({error}); the journal stays so recovery repeats it",
path.display(),
dir.display()
),
)
};
if let Some(source_dir) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
fire(Step::SetAsideSyncSource, source_dir)
.and_then(|()| sync_dir_strict(source_dir))
.map_err(|e| failed(source_dir, e))?;
}
if let Some(target_dir) = target.parent() {
fire(Step::SetAsideSyncTarget, target_dir)
.and_then(|()| sync_dir_strict(target_dir))
.map_err(|e| failed(target_dir, e))?;
}
Ok(())
}
fn remove_claimed(
path: &Path,
claim: Option<&Ident>,
full: bool,
retry: Option<RetryPolicy>,
) -> std::result::Result<(), RetainedItem> {
let Some(claim) = claim else {
return match probe(path) {
Ok(Probe::Missing) => Ok(()),
Ok(Probe::File(_)) => Err(RetainedItem::new(
path,
"the journal records no staged file for this name, so it was not removed",
)),
Ok(Probe::Other(why)) => Err(RetainedItem::new(path, format!("it is {why}"))),
Err(error) => Err(RetainedItem::new(path, error)),
};
};
remove_while_identical(
path,
|now| {
if full {
now == claim
} else {
now.key == claim.key
}
},
"it is not the file the transaction journaled for this staged name, so it was not removed",
retry,
)
}
fn discard(ctx: &Ctx<'_>, journal: &Journal) -> Vec<RetainedItem> {
let mut retained = Vec::new();
let full = journal.phase == JournalPhase::Staged;
for entry in &journal.entries {
match entry_paths(ctx.root, entry) {
Ok(paths) => {
if let Err(item) =
remove_claimed(&paths.stage, entry.staged.as_ref(), full, ctx.options.retry)
{
retained.push(item);
}
}
Err(error) => retained.push(RetainedItem::new(ctx.root.join(&entry.dest), error)),
}
}
if retained.is_empty() {
retained.extend(finish_dir(ctx));
}
retained
}
fn remove_record(ctx: &Ctx<'_>, name: &str) -> std::result::Result<(), RetainedItem> {
let path = ctx.txn_dir.join(name);
let claim = ctx.record_claim(name);
remove_while_identical(
&path,
|now| {
claim.as_ref().is_some_and(|claim| {
*now == claim.ident
&& read_record_claim(&path, &claim.text)
.is_ok_and(|found| found.ident == claim.ident)
})
},
"the protocol record no longer has its recorded identity and exact content",
ctx.options.retry,
)
}
fn finish_dir(ctx: &Ctx<'_>) -> Vec<RetainedItem> {
let txn_dir = ctx.txn_dir;
let mut retained = Vec::new();
match fs::read_dir(txn_dir) {
Ok(listing) => {
for entry in listing {
let entry = match entry {
Ok(entry) => entry,
Err(error) => {
retained.push(RetainedItem::new(txn_dir, error));
continue;
}
};
let name = entry.file_name().to_string_lossy().into_owned();
let valid = ctx.record_claim(&name).is_some_and(|claim| {
read_record_claim(&entry.path(), &claim.text)
.is_ok_and(|found| found.ident == claim.ident)
});
if !matches!(name.as_str(), JOURNAL | COMMIT) || !valid {
retained.push(RetainedItem::new(
entry.path(),
"unclaimed or changed protocol material",
));
}
}
}
Err(error) if error.kind() == io::ErrorKind::NotFound => return retained,
Err(error) => return vec![RetainedItem::new(txn_dir, error)],
}
if !retained.is_empty() {
return retained;
}
let journal = txn_dir.join(JOURNAL);
if let Err(error) = fire(Step::BeforeJournalRemove, &journal) {
return vec![RetainedItem::new(&journal, error)];
}
if let Err(item) = remove_record(ctx, JOURNAL) {
return vec![item];
}
let removed = sync_dir_strict(txn_dir).and_then(|()| fire(Step::AfterJournalRemove, &journal));
if let Err(error) = removed {
return vec![RetainedItem::new(&journal, error)];
}
if let Err(item) = remove_record(ctx, COMMIT) {
return vec![item];
}
let marker = txn_dir.join(COMMIT);
let synced = sync_dir_strict(txn_dir).and_then(|()| fire(Step::AfterMarkerRemove, &marker));
if let Err(error) = synced {
return vec![RetainedItem::new(txn_dir, error)];
}
match fs::remove_dir(txn_dir) {
Ok(()) => {
if let Err(error) = sync_parent_strict(txn_dir) {
retained.push(RetainedItem::new(txn_dir, error));
}
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => retained.push(RetainedItem::new(txn_dir, error)),
}
retained
}
pub fn recover_transactions(root: &Path) -> Result<RecoveryReport> {
recover_transactions_with(root, &TransactionOptions::default())
}
pub fn recover_transactions_with(
root: &Path,
options: &TransactionOptions,
) -> Result<RecoveryReport> {
let root = canonicalize_plain(root).map_err(|e| FsError::io(root, e))?;
let area = root.join(AREA);
match fs::symlink_metadata(&area) {
Ok(meta) => {
if is_link(&meta) {
return Err(FsError::Symlink(area));
}
if !meta.is_dir() {
return Err(FsError::NotADirectory(area));
}
}
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(RecoveryReport::default()),
Err(e) => return Err(FsError::io(&area, e)),
}
let _lock = FileLock::acquire(&area.join(LOCK_NAME), &lock_options(options))?;
let root_key = FileKey::of_path(&root)?;
let mut names: Vec<String> = Vec::new();
for entry in fs::read_dir(&area).map_err(|e| FsError::io(&area, e))? {
let entry = entry.map_err(|e| FsError::io(&area, e))?;
let file_type = entry
.file_type()
.map_err(|e| FsError::io(entry.path(), e))?;
let name = entry.file_name().to_string_lossy().into_owned();
if file_type.is_dir() && name != RETAINED_DIR {
names.push(name);
}
}
names.sort();
names.reverse();
let mut report = RecoveryReport::default();
for name in names {
let txn_dir = area.join(&name);
let ctx = Ctx {
records: std::cell::RefCell::new(Vec::new()),
root: &root,
options,
txn_dir: &txn_dir,
};
let outcome = recover_one(&ctx, &name, root_key);
report
.transactions
.push(RecoveredTransaction { id: name, outcome });
}
Ok(report)
}
fn retained_outcome(reason: &str, items: Vec<RetainedItem>) -> RecoveryOutcome {
RecoveryOutcome::Retained {
reason: reason.to_string(),
items,
}
}
fn present(path: &Path) -> bool {
fs::symlink_metadata(path).is_ok()
}
fn recover_one(ctx: &Ctx<'_>, name: &str, root_key: FileKey) -> RecoveryOutcome {
let journal_path = ctx.txn_dir.join(JOURNAL);
let marker_path = ctx.txn_dir.join(COMMIT);
if !valid_id(name) {
return retained_outcome(
"an unrecognized directory under .rktxn",
vec![RetainedItem::new(ctx.txn_dir, "not a transaction id")],
);
}
let has_journal = present(&journal_path);
let has_marker = present(&marker_path);
if !has_journal {
if has_marker {
let text = match fs::read_to_string(&marker_path) {
Ok(text) => text,
Err(error) => {
return retained_outcome(
"the remaining marker cannot be read",
vec![RetainedItem::new(&marker_path, error)],
)
}
};
let prefix = format!("{COMMIT_MAGIC} {name} ");
let valid = text
.strip_prefix(&prefix)
.and_then(|tail| tail.strip_suffix('\n'))
.is_some_and(|crc| crc.len() == 8 && u32::from_str_radix(crc, 16).is_ok());
if !valid {
return retained_outcome(
"unrecognized marker without a journal",
vec![RetainedItem::new(
&marker_path,
"marker does not identify this transaction",
)],
);
}
match read_record_claim(&marker_path, &text) {
Ok(claim) => ctx.claim_record(&marker_path, claim),
Err(error) => {
return retained_outcome(
"the remaining marker changed",
vec![RetainedItem::new(&marker_path, error)],
)
}
}
}
let left = finish_dir(ctx);
return match (left.is_empty(), has_marker) {
(true, true) => RecoveryOutcome::Completed,
(true, false) => RecoveryOutcome::Discarded,
(false, _) => retained_outcome(
"the transaction directory holds material recovery will not delete",
left,
),
};
}
let mut journal = match load_journal(&journal_path) {
Ok(journal) => journal,
Err(error) => {
return retained_outcome(
"the journal cannot be read",
vec![RetainedItem::new(&journal_path, error)],
)
}
};
match read_record_claim(&journal_path, &journal.render().0) {
Ok(claim) => ctx.claim_record(&journal_path, claim),
Err(error) => {
return retained_outcome(
"the journal changed while loading",
vec![RetainedItem::new(&journal_path, error)],
)
}
}
if journal.id != name {
return retained_outcome(
"the journal belongs to another transaction",
vec![RetainedItem::new(
&journal_path,
"its id differs from its directory name",
)],
);
}
if journal.root.index != root_key.index {
return retained_outcome(
"the journal was written for a different root directory",
vec![RetainedItem::new(
&journal_path,
"the root's file identity differs (the root was copied or replaced)",
)],
);
}
if journal.root.volume != root_key.volume {
return retained_outcome(
"the journal's root directory is on a different device than recorded",
vec![RetainedItem::new(
&journal_path,
"the root has the recorded file index but another device number; a renumbering after a restart is not proven, so the transaction is kept for a person to resolve",
)],
);
}
if has_marker {
let expected = marker_text(&journal.id, journal.crc);
let matches = fs::read_to_string(&marker_path)
.map(|text| text == expected)
.unwrap_or(false);
if !matches || journal.phase != JournalPhase::Staged {
return retained_outcome(
"the commit marker does not match the journal",
vec![RetainedItem::new(
&marker_path,
"unreadable, or bound to a different journal",
)],
);
}
match read_record_claim(&marker_path, &expected) {
Ok(claim) => ctx.claim_record(&marker_path, claim),
Err(error) => {
return retained_outcome(
"the marker changed while loading",
vec![RetainedItem::new(&marker_path, error)],
)
}
}
let cleanup = complete_committed(ctx, &journal);
return if !cleanup.blocking.is_empty() {
retained_outcome(
"committed, but cleanup could not finish",
cleanup.into_items(),
)
} else if cleanup.set_aside.is_empty() {
RecoveryOutcome::Completed
} else {
RecoveryOutcome::CompletedSetAside {
items: cleanup.set_aside,
}
};
}
match journal.phase {
JournalPhase::Staging => {
let left = discard(ctx, &journal);
if left.is_empty() {
RecoveryOutcome::Discarded
} else {
retained_outcome("staged copies could not be removed", left)
}
}
JournalPhase::Staged => {
let outcome = rollback(ctx, &mut journal);
if outcome.retained.is_empty() {
RecoveryOutcome::RolledBack
} else {
retained_outcome(
"rollback stopped at material that does not match what the journal recorded",
outcome.retained,
)
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn ident(index: u64, len: u64, modified_ns: u128) -> Ident {
Ident {
key: FileKey { volume: 7, index },
len,
modified_ns,
}
}
fn file(ident: Ident) -> Probe {
Probe::File(ident)
}
#[test]
fn ident_round_trips_and_rejects_noise() {
let value = ident(0xabc, 12, 99);
assert_eq!(Ident::parse(&value.render()), Some(value));
for bad in ["", "1.2.3", "1.2.3.4.5", "x.2.3.4", "1.2.-3.4", "1.2.3.y"] {
assert_eq!(Ident::parse(bad), None, "{bad:?}");
}
}
fn on_device(ident: Ident, volume: u64) -> Ident {
Ident {
key: FileKey {
volume,
index: ident.key.index,
},
..ident
}
}
#[test]
fn identity_compares_the_device_number_too() {
let value = ident(5, 10, 100);
assert_eq!(value, ident(5, 10, 100));
assert_ne!(
value,
on_device(value, 99),
"another device is another file"
);
assert_ne!(value, ident(6, 10, 100));
assert_ne!(value, ident(5, 11, 100));
assert_ne!(value, ident(5, 10, 101));
}
#[test]
fn a_changed_device_is_never_classified_as_published_or_restorable() {
let moved = |ident: Ident| file(on_device(ident, 1234));
let conflict = |s: EntryState| matches!(s, EntryState::Conflict(_));
assert!(conflict(classify(
Some(&PRIOR()),
&STAGED(),
&moved(STAGED()),
&Probe::Missing,
&moved(PRIOR()),
)));
assert!(conflict(classify(
Some(&PRIOR()),
&STAGED(),
&moved(PRIOR()),
&moved(STAGED()),
&Probe::Missing,
)));
assert!(conflict(classify(
None,
&STAGED(),
&moved(STAGED()),
&Probe::Missing,
&Probe::Missing,
)));
}
#[test]
fn a_transient_failure_is_retried_exactly_the_bounded_number_of_times() {
let policy = Some(RetryPolicy::new(3, Duration::from_millis(1)));
let calls = std::cell::Cell::new(0u32);
let started = std::time::Instant::now();
let result = retry_while(
policy,
|_| true,
|| {
calls.set(calls.get() + 1);
Err(io::Error::other("busy"))
},
);
let elapsed = started.elapsed();
assert!(result.is_err());
assert_eq!(calls.get(), 4, "one attempt and three retries, no more");
assert!(
elapsed >= Duration::from_millis(7),
"the pauses double (1 + 2 + 4 ms): {elapsed:?}"
);
assert!(elapsed < Duration::from_secs(5), "bounded: {elapsed:?}");
let calls = std::cell::Cell::new(0u32);
let result = retry_while(
policy,
|_| true,
|| {
calls.set(calls.get() + 1);
if calls.get() < 3 {
Err(io::Error::other("busy"))
} else {
Ok(())
}
},
);
assert!(result.is_ok());
assert_eq!(calls.get(), 3);
}
#[test]
fn a_permanent_failure_or_no_policy_is_one_attempt() {
let policy = Some(RetryPolicy::new(3, Duration::from_millis(1)));
let calls = std::cell::Cell::new(0u32);
let attempt = || {
calls.set(calls.get() + 1);
Err(io::Error::other("denied"))
};
assert!(retry_while(policy, |_| false, attempt).is_err());
assert_eq!(calls.get(), 1, "a permanent error is never retried");
calls.set(0);
assert!(retry_while(None, |_| true, attempt).is_err());
assert_eq!(calls.get(), 1, "without a policy there is one attempt");
let busy = |code| io::Error::from_raw_os_error(code);
assert!(!transient_delete_error(&busy(2)));
assert_eq!(transient_delete_error(&busy(32)), cfg!(windows));
assert_eq!(transient_delete_error(&busy(5)), cfg!(windows));
}
#[test]
fn removing_a_missing_file_is_fine_and_a_directory_is_an_error() {
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("f");
std::fs::write(&file, b"x").unwrap();
let retry = Some(RetryPolicy::new(3, Duration::from_millis(1)));
remove_with_retry(&file, retry).unwrap();
remove_with_retry(&file, retry).expect("a missing file is removed");
let inner = dir.path().join("d");
std::fs::create_dir(&inner).unwrap();
std::fs::write(inner.join("x"), b"x").unwrap();
assert!(remove_with_retry(&inner, retry).is_err());
assert!(remove_with_retry(&inner, None).is_err());
}
#[test]
fn a_staged_name_is_removed_only_when_it_is_the_journaled_file() {
let dir = tempfile::tempdir().unwrap();
let (mine, theirs) = (dir.path().join("mine"), dir.path().join("theirs"));
std::fs::write(&mine, b"m").unwrap();
std::fs::write(&theirs, b"foreign").unwrap();
let Probe::File(recorded) = probe(&mine).unwrap() else {
panic!("not a file");
};
let claim = Ident {
key: recorded.key,
len: 0,
modified_ns: 0,
};
let none = None;
assert!(remove_claimed(&theirs, Some(&claim), false, none).is_err());
assert!(remove_claimed(&theirs, None, false, none).is_err());
assert_eq!(std::fs::read(&theirs).unwrap(), b"foreign");
assert!(remove_claimed(&mine, Some(&claim), true, none).is_err());
assert!(mine.exists());
let elsewhere = Ident {
key: FileKey {
volume: claim.key.volume.wrapping_add(1),
index: claim.key.index,
},
..claim
};
assert!(remove_claimed(&mine, Some(&elsewhere), false, none).is_err());
assert!(mine.exists());
remove_claimed(&mine, Some(&claim), false, none).unwrap();
assert!(!mine.exists());
remove_claimed(&mine, None, false, none).unwrap();
remove_claimed(&mine, Some(&claim), true, none).unwrap();
let folder = dir.path().join("folder");
std::fs::create_dir(&folder).unwrap();
assert!(remove_claimed(&folder, Some(&claim), false, none).is_err());
assert!(folder.is_dir());
}
#[test]
fn removal_needs_the_identity_to_hold_and_a_vanished_file_is_success() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("f");
std::fs::write(&path, b"x").unwrap();
let none = None;
let kept = remove_while_identical(&path, |_| false, "not ours", none).unwrap_err();
assert!(kept.reason.contains("not ours"), "{kept:?}");
assert!(path.exists());
remove_while_identical(&path, |_| true, "not ours", none).unwrap();
assert!(!path.exists());
remove_while_identical(&path, |_| false, "not ours", none).unwrap();
let folder = dir.path().join("d");
std::fs::create_dir(&folder).unwrap();
assert!(remove_while_identical(&folder, |_| true, "not ours", none).is_err());
assert!(folder.is_dir());
}
#[test]
fn an_entry_marked_unpublished_stays_untouched_after_its_staged_copy_is_gone() {
let other = ident(9, 99, 999);
let untouched = |s: EntryState| s == EntryState::Untouched;
let replay = |prior: Option<Ident>, unpublished: bool| {
classify_entry(
prior.as_ref(),
&STAGED(),
&file(other),
&Probe::Missing,
&Probe::Missing,
unpublished,
)
};
assert!(matches!(
replay(Some(PRIOR()), false),
EntryState::Conflict(_)
));
assert!(untouched(replay(Some(PRIOR()), true)));
assert!(matches!(replay(None, false), EntryState::Conflict(_)));
assert!(untouched(replay(None, true)));
assert_eq!(
classify_entry(
Some(&PRIOR()),
&STAGED(),
&file(STAGED()),
&Probe::Missing,
&file(PRIOR()),
true
),
EntryState::Published { backup: true }
);
assert!(matches!(
classify_entry(
Some(&PRIOR()),
&STAGED(),
&file(other),
&Probe::Missing,
&file(other),
true
),
EntryState::Conflict(_)
));
}
#[test]
fn hex_round_trips_unicode() {
let text = "dir/\u{e9}t\u{e9}.txt";
assert_eq!(hex_decode(&hex_encode(text)).as_deref(), Some(text));
assert_eq!(hex_decode("abc"), None);
assert_eq!(hex_decode("zz"), None);
assert_eq!(hex_decode("ff"), None, "0xff alone is not UTF-8");
}
fn sample_journal() -> Journal {
Journal {
id: "00000000abcd-1f-0".into(),
root: FileKey {
volume: 3,
index: 9,
},
phase: JournalPhase::Staged,
entries: vec![
JournalEntry {
dest: "a/b.txt".into(),
stage: "a/.b.txt.rktxn-00000000abcd-1f-0-0s".into(),
backup: "a/.b.txt.rktxn-00000000abcd-1f-0-0b".into(),
create_new: false,
private: true,
prior: Some(ident(1, 2, 3)),
staged: Some(ident(4, 5, 6)),
unpublished: true,
},
JournalEntry {
dest: "c.txt".into(),
stage: ".c.txt.rktxn-00000000abcd-1f-0-1s".into(),
backup: ".c.txt.rktxn-00000000abcd-1f-0-1b".into(),
create_new: true,
private: false,
prior: None,
staged: Some(ident(7, 8, 9)),
unpublished: false,
},
],
crc: 0,
}
}
#[test]
fn journal_round_trips() {
let mut journal = sample_journal();
let (text, crc) = journal.render();
journal.crc = crc;
let parsed = Journal::parse(&text, Path::new("journal")).unwrap();
assert_eq!(parsed, journal);
}
#[test]
fn journal_refuses_damage() {
let (text, _) = sample_journal().render();
let path = Path::new("journal");
let flipped = text.replacen("entries 2", "entries 3", 1);
assert!(Journal::parse(&flipped, path).is_err());
assert!(Journal::parse(&text[..text.len() - 12], path).is_err());
assert!(Journal::parse(text.split("crc ").next().unwrap(), path).is_err());
assert!(Journal::parse(&format!("{text}extra\n"), path).is_err());
let magic = text.replacen("rktxn-journal 1", "rktxn-journal 2", 1);
assert!(Journal::parse(&magic, path).is_err());
assert!(Journal::parse("", path).is_err());
}
#[test]
fn a_staged_journal_needs_staged_identities() {
let mut journal = sample_journal();
journal.entries[0].staged = None;
let (text, _) = journal.render();
assert!(Journal::parse(&text, Path::new("journal")).is_err());
journal.phase = JournalPhase::Staging;
let (text, _) = journal.render();
assert!(Journal::parse(&text, Path::new("journal")).is_ok());
}
#[test]
fn relative_destinations_are_confined_and_bounded() {
assert_eq!(
split_relative(Path::new("a/./b.txt")).unwrap(),
vec!["a".to_string(), "b.txt".to_string()]
);
for bad in [
"",
".",
"/abs.txt",
"../x",
"a/../x",
".rktxn/x",
"a/.x.rktxn-1-0s",
"a/.x.rktmp-1-2-3",
] {
assert!(split_relative(Path::new(bad)).is_err(), "{bad:?}");
}
let long = "x".repeat(MAX_RELATIVE_BYTES + 1);
assert!(split_relative(Path::new(&long)).is_err());
let deep = vec!["d"; MAX_COMPONENTS + 1].join("/");
assert!(split_relative(Path::new(&deep)).is_err());
}
#[test]
fn side_names_are_bounded_and_not_temp_names() {
let long = "n".repeat(250);
let stage = side_name(&long, "00000000abcd-1f-0", 3, 's');
assert!(stage.len() < 120);
assert!(stage.contains(SIDE_MARKER));
assert!(crate::atomic::generated_temp_pid(&stage).is_none());
let wide = "\u{e9}".repeat(100);
assert!(side_name(&wide, "i", 0, 'b').is_char_boundary(1));
assert!(valid_id(&new_id()));
}
#[test]
fn journal_paths_must_sit_beside_the_destination() {
let entry = JournalEntry {
dest: "a/b.txt".into(),
stage: "elsewhere/.b.txt.rktxn-00000000abcd-1f-0-0s".into(),
backup: "a/.b.txt.rktxn-00000000abcd-1f-0-0b".into(),
create_new: false,
private: false,
prior: None,
staged: Some(ident(1, 1, 1)),
unpublished: false,
};
let dir = tempfile::tempdir().unwrap();
assert!(matches!(
entry_paths(dir.path(), &entry),
Err(FsError::Corrupt { .. })
));
let inside = JournalEntry {
dest: ".rktxn/x".into(),
stage: ".rktxn/.x.rktxn-x-0s".into(),
backup: ".rktxn/.x.rktxn-x-0b".into(),
..entry
};
assert!(matches!(
entry_paths(dir.path(), &inside),
Err(FsError::Corrupt { .. })
));
}
const PRIOR: fn() -> Ident = || ident(1, 10, 100);
const STAGED: fn() -> Ident = || ident(2, 20, 200);
fn state(prior: Option<Ident>, dest: Probe, stage: Probe, backup: Probe) -> EntryState {
classify(prior.as_ref(), &STAGED(), &dest, &stage, &backup)
}
#[test]
fn an_untouched_replace_entry() {
assert_eq!(
state(Some(PRIOR()), file(PRIOR()), file(STAGED()), Probe::Missing),
EntryState::Untouched
);
assert_eq!(
state(Some(PRIOR()), file(PRIOR()), Probe::Missing, Probe::Missing),
EntryState::Untouched
);
}
#[test]
fn a_link_backup_before_the_publish() {
assert_eq!(
state(Some(PRIOR()), file(PRIOR()), file(STAGED()), file(PRIOR())),
EntryState::LinkedBackup
);
}
#[test]
fn an_aside_backup_before_the_publish() {
assert_eq!(
state(Some(PRIOR()), Probe::Missing, file(STAGED()), file(PRIOR())),
EntryState::AsideBackup
);
assert_eq!(
state(Some(PRIOR()), Probe::Missing, Probe::Missing, file(PRIOR())),
EntryState::AsideBackup
);
}
#[test]
fn a_published_replace_entry() {
assert_eq!(
state(Some(PRIOR()), file(STAGED()), Probe::Missing, file(PRIOR())),
EntryState::Published { backup: true }
);
assert_eq!(
state(Some(PRIOR()), file(STAGED()), file(STAGED()), file(PRIOR())),
EntryState::Published { backup: true }
);
}
#[test]
fn a_destination_edited_before_the_backup_was_made_is_untouched_and_rolls_back() {
let edited = ident(1, 11, 101);
assert_eq!(
state(Some(PRIOR()), file(edited), file(STAGED()), Probe::Missing),
EntryState::Untouched
);
assert!(matches!(
state(Some(PRIOR()), file(edited), Probe::Missing, Probe::Missing),
EntryState::Conflict(_)
));
}
#[test]
fn a_foreign_file_that_appeared_before_our_publish_is_left_alone() {
let other = ident(9, 99, 999);
assert_eq!(
state(None, file(other), file(STAGED()), Probe::Missing),
EntryState::Untouched
);
}
#[test]
fn a_new_file_entry() {
assert_eq!(
state(None, Probe::Missing, file(STAGED()), Probe::Missing),
EntryState::Untouched
);
assert_eq!(
state(None, Probe::Missing, Probe::Missing, Probe::Missing),
EntryState::Untouched
);
assert_eq!(
state(None, file(STAGED()), Probe::Missing, Probe::Missing),
EntryState::Published { backup: false }
);
}
#[test]
fn anything_unexpected_is_a_conflict_never_a_guess() {
let other = ident(9, 99, 999);
let conflict = |s: EntryState| matches!(s, EntryState::Conflict(_));
assert!(conflict(state(
Some(PRIOR()),
file(other),
Probe::Missing,
file(PRIOR())
)));
assert!(conflict(state(
Some(PRIOR()),
file(other),
Probe::Missing,
Probe::Missing
)));
assert!(conflict(state(
Some(PRIOR()),
file(other),
file(STAGED()),
file(other)
)));
assert!(conflict(state(
Some(PRIOR()),
file(other),
Probe::Missing,
file(PRIOR())
)));
assert!(conflict(state(
Some(PRIOR()),
file(STAGED()),
Probe::Missing,
file(other)
)));
assert!(conflict(state(
Some(PRIOR()),
file(PRIOR()),
file(other),
Probe::Missing
)));
assert!(conflict(state(
Some(PRIOR()),
Probe::Missing,
Probe::Missing,
Probe::Missing
)));
assert!(conflict(state(
Some(PRIOR()),
file(STAGED()),
Probe::Missing,
Probe::Missing
)));
assert!(conflict(state(
None,
file(other),
Probe::Missing,
Probe::Missing
)));
assert!(conflict(state(
None,
Probe::Missing,
file(STAGED()),
file(PRIOR())
)));
let dir = Probe::Other("not a regular file".into());
assert!(conflict(state(
Some(PRIOR()),
dir.clone(),
file(STAGED()),
Probe::Missing
)));
assert!(conflict(state(
Some(PRIOR()),
file(PRIOR()),
dir.clone(),
Probe::Missing
)));
assert!(conflict(state(
Some(PRIOR()),
file(PRIOR()),
Probe::Missing,
dir
)));
assert!(conflict(classify(
Some(&STAGED()),
&STAGED(),
&file(STAGED()),
&Probe::Missing,
&file(STAGED()),
)));
}
}