coordinode-lsm-tree 7.4.2

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026-present, Dmitry Prudnikov

//! Power-loss crash simulator [`Fs`] backend for recovery testing.
//!
//! [`CrashFs`] wraps an inner [`Fs`] and models the durability contract of a
//! real disk: bytes become durable only when a file handle is `fsync`ed
//! ([`FsFile::sync_all`] / [`FsFile::sync_data`]). [`CrashFs::crash`] simulates
//! a power loss by rolling every file back to its last-synced content and
//! removing files that were never synced at all — exactly the bytes a real
//! crash would lose. Reopening the storage engine on the post-crash backend
//! (via [`CrashFs::inner`]) then exercises its recovery path against the
//! worst-case durable image.
//!
//! Unlike [`FaultFs`](crate::fs::FaultFs), which makes a *chosen* operation
//! return an error, `CrashFs` is about *durability*: every operation succeeds
//! during the run, but a `crash()` reveals which writes were actually durable.
//! The two compose — wrap `CrashFs` in a `FaultFs` to fail a specific `fsync`
//! and then crash to discard the tail that the failed sync never made durable,
//! reproducing a torn write mid-flush.
//!
//! # Model and limitations
//!
//! Durability is tracked at file-content granularity: a file's durable image is
//! its full content as of its most recent successful `sync_all` / `sync_data`.
//! A file written but never synced vanishes on `crash()`; a synced file keeps
//! exactly its last-synced bytes (a later un-synced append or truncate is
//! rolled back). Hard links the backend reports as one file
//! ([`Fs::same_file`]) share one durable image, since they share their bytes;
//! on a backend where a hard link is a copy each name keeps its own. Every
//! name is tracked resolved, as the backend resolves it: a symbolic link,
//! final or in a directory along the way, and `..` lead to the entry they
//! reach, so a file written or synced through any spelling of that kind is
//! the one file. Two names the filesystem alone makes one (a case it
//! ignores, a bind mount) are tracked as two files.
//!
//! A new directory entry (a created file, the destination of a rename, a hard
//! link or a reflink) is durable only once its parent directory is synced, as
//! POSIX promises: `crash()` removes a file whose entry was never made durable,
//! even when its content was synced. A new folder is durable from its
//! creation, unless [`CrashFs::tracking_folders`] makes its entry follow the
//! same rule: then `crash()` removes a folder whose entry was never made
//! durable, with everything under it. The directory synced is matched by what
//! the backend reports as the same directory, so any spelling of it covers
//! its entries. Removed entries are not brought back, except a durable file
//! removed behind a [`SyncMode::Barrier`] sync a power loss loses (see below),
//! and a removed folder is not put back. Opens, operations that make or remove
//! entries, and directory syncs are linearized: each
//! happens wholly before or wholly after any other, so an entry is made
//! either before a sync that then covers it or after one that does not.
//!
//! One `CrashFs` is one device. A sync under [`SyncMode::Barrier`], of a file
//! or a directory, keeps writes in order without making them durable: it is
//! queued, and [`Fs::sync_device`], or any other sync on the device, makes
//! every queued one durable. A removal or a rename issued behind a queued
//! sync is queued with it, as one step. [`CrashFs::crash`] loses them all;
//! [`CrashFs::crash_keeping`] keeps a prefix of the steps in the order they
//! were issued, the most a power loss can leave.
//!
//! This is a test/dev surface: it is gated behind the `std` feature and is not
//! part of the production storage path.
//!
//! # Examples
//!
//! ```
//! use lsm_tree::fs::{CrashFs, MemFs};
//!
//! let fs = CrashFs::new(MemFs::new());
//! // ... run a workload through `fs`, fsyncing durable checkpoints ...
//! fs.crash(); // discard everything written since the last fsync of each file
//! // ... reopen the engine on `fs.inner()` and verify recovery ...
//! ```

use super::{
    FileHint, Fs, FsCapabilities, FsDirEntry, FsFile, FsMetadata, FsOpenOptions, SyncMode,
};
use crate::io;
use crate::path::{Path, PathBuf};
use alloc::boxed::Box;
use alloc::sync::Arc;
use alloc::vec::Vec;
use hashbrown::{HashMap, HashSet};
use std::path::Component;

/// Shared crash state: the durable image of each file plus the set of files
/// written during the current run (so `crash()` can vanish never-synced ones).
#[derive(Default)]
struct CrashState {
    /// Last-synced full content per file path. Presence means "this path has
    /// been made durable at least once"; the bytes are its durable image.
    durable: HashMap<PathBuf, Vec<u8>>,
    /// Every path opened for writing this run, whether or not yet synced.
    /// `crash()` visits these (plus `durable` keys) to roll back or remove.
    touched: HashSet<PathBuf>,
    /// Paths whose directory entry was made this run and whose parent
    /// directory has not been synced since: `crash()` removes them.
    pending_entries: HashSet<PathBuf>,
    /// Names the backend reports as hard links of one file, by group: they
    /// share their bytes, so they share one durable image.
    link_group: HashMap<PathBuf, u64>,
    /// The id the next link group takes.
    next_link_group: u64,
    /// The durable target of each symbolic link a queued removal or rename
    /// took, captured before it went: a crash that loses that step puts the
    /// link back, as it puts a file back from its durable image.
    links: HashMap<PathBuf, Link>,
    /// Hard-linked names a queued removal took out of their link group, with
    /// the group: a crash that loses the removal links the name back to a
    /// name of that group still on disk.
    unlinked: HashMap<PathBuf, u64>,
    /// What [`SyncMode::Barrier`] synced, in order, and no device flush or
    /// full sync has made durable since: a crash keeps a prefix of it.
    ordered: Vec<Ordered>,
}

/// A symbolic link as a crash puts it back: its target, stored as given,
/// and whether it was made as a link to a directory, the kind Windows keeps
/// with the link whatever its target is.
#[derive(Clone)]
struct Link {
    target: PathBuf,
    to_dir: bool,
}

/// One sync [`SyncMode::Barrier`] issued, durable once the device is flushed.
enum Ordered {
    /// A file's content as of its sync.
    Image(PathBuf, Vec<u8>),
    /// The entries a directory sync covered, each with whether it names a
    /// regular file.
    Entries(Vec<(PathBuf, bool)>),
    /// A removal issued behind an ordered sync: a barrier keeps it behind
    /// that sync, so a power loss that loses the sync loses the removal too
    /// and leaves the file as the syncs before it made it. The flag says the
    /// name's entry was not durable yet when it was removed.
    Removed(PathBuf, bool),
    /// A rename issued behind an ordered sync, kept or lost whole: kept, the
    /// destination's old file and the source's name are gone and the
    /// destination holds `bytes`; lost, both names are as the syncs before it
    /// left them. The flags say each name's entry was not durable yet.
    Renamed {
        from: PathBuf,
        to: PathBuf,
        from_pending: bool,
        to_pending: bool,
        bytes: Option<Vec<u8>>,
    },
}

impl CrashState {
    /// Makes durable the first `count` ordered steps, in the order they were
    /// issued, and drops the rest.
    fn settle(&mut self, count: usize) {
        let ordered = core::mem::take(&mut self.ordered);
        // A kept rename made its destination's entry with it: the entry is
        // durable from there, unless a later KEPT step takes the name again
        // and leaves whatever holds it then to its own sync. A lost step
        // changes nothing on disk, so the name keeps the renamed file.
        let renamed_durable: Vec<PathBuf> = ordered
            .iter()
            .enumerate()
            .take(count)
            .filter_map(|(at, sync)| match sync {
                Ordered::Renamed { to, .. } => Some((at, to)),
                _ => None,
            })
            .filter(|&(at, to)| {
                !ordered
                    .iter()
                    .take(count)
                    .skip(at + 1)
                    .any(|later| match later {
                        Ordered::Removed(name, _) => name == to,
                        Ordered::Renamed {
                            from: source,
                            to: target,
                            ..
                        } => source == to || target == to,
                        _ => false,
                    })
            })
            .map(|(_, to)| to.clone())
            .collect();
        for to in &renamed_durable {
            self.pending_entries.remove(to);
        }
        let mut ordered = ordered.into_iter();
        for sync in ordered.by_ref().take(count) {
            match sync {
                Ordered::Image(path, bytes) => self.set_durable(&path, bytes),
                Ordered::Entries(entries) => {
                    for (entry, file) in &entries {
                        self.entry_made_durable(entry, *file);
                    }
                }
                // Kept, in its place: the file is gone from here on, and an
                // image of a file made under its name later still lands.
                Ordered::Removed(path, _) => {
                    self.durable.remove(&path);
                    self.links.remove(&path);
                    self.unlinked.remove(&path);
                }
                Ordered::Renamed {
                    from, to, bytes, ..
                } => {
                    self.durable.remove(&to);
                    self.durable.remove(&from);
                    self.links.remove(&to);
                    let link = self.links.remove(&from);
                    if let Some(bytes) = bytes {
                        self.set_durable(&to, bytes);
                    } else if let Some(target) = link {
                        // The link moved, under its new name.
                        self.links.insert(to, target);
                    }
                }
            }
        }
        // A removal behind a lost sync is lost with it, and so is everything
        // made under that name after it: no sync that could make the new
        // entry durable comes before the lost one. The file stays as the kept
        // syncs left it; an entry made only after the removal is dropped,
        // one that was pending before it stays pending. A lost rename leaves
        // both of its names so.
        for sync in ordered {
            match sync {
                Ordered::Removed(path, false) => {
                    self.pending_entries.remove(&path);
                }
                Ordered::Renamed {
                    from,
                    to,
                    from_pending,
                    to_pending,
                    ..
                } => {
                    if !from_pending {
                        self.pending_entries.remove(&from);
                    }
                    if !to_pending {
                        self.pending_entries.remove(&to);
                    }
                }
                _ => {}
            }
        }
    }

    /// Makes durable every ordered step: a device flush, or a full sync that
    /// flushes the device cache with it.
    fn settle_all(&mut self) {
        self.settle(self.ordered.len());
    }

    /// The entry `path` made durable by a sync of its directory. A regular
    /// file none of whose bytes was ever synced is still there after a power
    /// loss, empty: the file was made with its entry, and the bytes written
    /// since are what no sync covered.
    fn entry_made_durable(&mut self, path: &Path, file: bool) {
        self.pending_entries.remove(path);
        if file && !self.durable.contains_key(path) {
            self.set_durable(path, Vec::new());
        }
    }

    /// Drops what the ordered syncs still owe the name `path`, removed or
    /// replaced: a sync of its old file or its old entry does not carry over
    /// to whatever takes the name.
    fn forget_ordered(&mut self, path: &Path) {
        self.ordered.retain_mut(|sync| match sync {
            Ordered::Image(name, _) => name != path,
            Ordered::Entries(entries) => {
                entries.retain(|(entry, _)| entry != path);
                true
            }
            Ordered::Removed(..) | Ordered::Renamed { .. } => true,
        });
    }

    /// `path` and every name linked to the same file.
    fn names_of(&self, path: &Path) -> Vec<PathBuf> {
        match self.link_group.get(path) {
            Some(&group) => self
                .link_group
                .iter()
                .filter(|&(_, &other)| other == group)
                .map(|(name, _)| name.clone())
                .collect(),
            None => alloc::vec![path.to_path_buf()],
        }
    }

    /// Records `bytes` as the durable image of `path` and of every name
    /// linked to it.
    fn set_durable(&mut self, path: &Path, bytes: Vec<u8>) {
        let mut names = self.names_of(path);
        let last = names.pop().unwrap_or_else(|| path.to_path_buf());
        for name in names {
            self.durable.insert(name, bytes.clone());
        }
        self.durable.insert(last, bytes);
    }

    /// Forgets the name `path`, removed from the namespace. The file it named
    /// lives on under its other hard links, and what the file went through
    /// stays with them: a write through the removed name that was never
    /// synced is still unsynced under each of them.
    fn forget_name(&mut self, path: &Path) {
        self.forget_ordered(path);
        self.durable.remove(path);
        self.links.remove(path);
        self.unlinked.remove(path);
        self.pending_entries.remove(path);
        let touched = self.touched.remove(path);
        if let Some(group) = self.link_group.remove(path)
            && touched
        {
            let survivors: Vec<PathBuf> = self
                .link_group
                .iter()
                .filter(|&(_, &other)| other == group)
                .map(|(name, _)| name.clone())
                .collect();
            self.touched.extend(survivors);
        }
    }

    /// Forgets the folder `path` and every name under it, gone from the
    /// namespace with it.
    fn forget_under(&mut self, path: &Path) {
        // A linked name under the directory hands what its file went through
        // to the names that survive outside it.
        let linked: Vec<PathBuf> = self
            .link_group
            .keys()
            .filter(|k| k.starts_with(path))
            .cloned()
            .collect();
        for name in &linked {
            self.forget_name(name);
        }
        self.durable.retain(|k, _| !k.starts_with(path));
        self.touched.retain(|k| !k.starts_with(path));
        self.links.retain(|k, _| !k.starts_with(path));
        self.unlinked.retain(|k, _| !k.starts_with(path));
        self.pending_entries.retain(|k| !k.starts_with(path));
        self.ordered.retain_mut(|sync| match sync {
            Ordered::Image(name, _) | Ordered::Removed(name, _) => !name.starts_with(path),
            Ordered::Renamed { from, to, .. } => !from.starts_with(path) && !to.starts_with(path),
            Ordered::Entries(entries) => {
                entries.retain(|(entry, _)| !entry.starts_with(path));
                true
            }
        });
    }

    /// Records `dst` as a hard link of the file `src` names.
    fn link(&mut self, src: &Path, dst: &Path) {
        let group = if let Some(&group) = self.link_group.get(src) {
            group
        } else {
            let group = self.next_link_group;
            // One group per hard link a test makes, far below 2^64.
            self.next_link_group += 1;
            self.link_group.insert(src.to_path_buf(), group);
            group
        };
        self.link_group.insert(dst.to_path_buf(), group);
    }
}

/// How many symbolic links a path may pass through before the chain is
/// refused: Linux's `MAXSYMLINKS`.
const MAX_SYMLINKS: usize = 40;

/// The components of `path`, each as a path of its own, last first.
fn components_reversed(path: &Path) -> Vec<PathBuf> {
    path.components()
        .rev()
        .map(|component| PathBuf::from(component.as_os_str()))
        .collect()
}

/// `path` with `.` and `..` folded away by spelling alone; `..` of a root is
/// the root, a `..` with nothing to climb out of stays, and nothing left is
/// `.`.
fn folded(path: &Path) -> PathBuf {
    let mut out = PathBuf::new();
    for component in path.components() {
        match component {
            Component::CurDir => {}
            Component::ParentDir => match out.components().next_back() {
                Some(Component::Normal(_)) => {
                    out.pop();
                }
                Some(Component::RootDir | Component::Prefix(_)) => {}
                _ => out.push(component.as_os_str()),
            },
            other => out.push(other.as_os_str()),
        }
    }
    if out.as_os_str().is_empty() {
        out.push(".");
    }
    out
}

/// A power-loss crash simulator wrapping an inner [`Fs`].
///
/// See the module-level documentation for the durability model. The inner backend is
/// held as an [`Arc<dyn Fs>`]; obtain a clone with [`inner`](Self::inner) to
/// reopen the engine directly on the post-crash store (e.g. via
/// [`Config::with_shared_fs`](crate::Config::with_shared_fs)).
#[derive(Clone)]
pub struct CrashFs {
    inner: Arc<dyn Fs>,
    state: Arc<spin::Mutex<CrashState>>,
    /// Linearizes the namespace: every operation that makes or removes a
    /// directory entry, every open and every write that resolves the entry it
    /// lands on,
    /// holds it from its checks through the backend call to the state it
    /// records, and a directory sync holds it from the backend
    /// sync to clearing what that sync made durable. An entry is then made
    /// either before a sync, and covered by it, or after, and pending; and
    /// whether an open or a rename made an entry is decided by probes no
    /// other entry change can interleave with. Blocking, not spinning: a
    /// directory sync holds it across a system call.
    namespace: Arc<parking_lot::Mutex<()>>,
    /// Whether a new folder's entry is durable only once its parent is synced.
    folders: bool,
}

impl CrashFs {
    /// Wraps `inner`, treating its current contents as the initial durable
    /// image (nothing is rolled back until something is written and then a
    /// `crash()` occurs).
    #[must_use]
    pub fn new<F: Fs>(inner: F) -> Self {
        Self::from_shared(Arc::new(inner))
    }

    /// Wraps an existing shared backend handle.
    #[must_use]
    pub fn from_shared(inner: Arc<dyn Fs>) -> Self {
        Self {
            inner,
            state: Arc::new(spin::Mutex::new(CrashState::default())),
            namespace: Arc::new(parking_lot::Mutex::new(())),
            folders: false,
        }
    }

    /// Makes a folder made from here on durable only once the folder above it
    /// is synced, as a file is: a [`crash`](Self::crash) removes one whose
    /// entry no sync covered, with everything under it.
    ///
    /// Off by default, so that a test may make the folders it works in
    /// without syncing the ones above them.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::fs::{CrashFs, Fs, MemFs};
    /// use std::path::Path;
    ///
    /// let fs = CrashFs::new(MemFs::new()).tracking_folders();
    /// fs.create_dir_all(Path::new("/kept"))?;
    /// fs.sync_directory(Path::new("/"))?;
    /// fs.create_dir_all(Path::new("/lost"))?;
    /// fs.crash();
    /// assert!(fs.exists(Path::new("/kept"))?);
    /// assert!(!fs.exists(Path::new("/lost"))?);
    /// # Ok::<(), lsm_tree::io::Error>(())
    /// ```
    #[must_use]
    pub fn tracking_folders(mut self) -> Self {
        self.folders = true;
        self
    }

    /// Returns a clone of the wrapped backend handle, for reopening the engine
    /// on the same store after a [`crash`](Self::crash).
    #[must_use]
    pub fn inner(&self) -> Arc<dyn Fs> {
        Arc::clone(&self.inner)
    }

    /// Simulates a power loss: every file is rolled back to its last-synced
    /// content, and files written but never synced are removed. After this the
    /// backend holds exactly the bytes a real crash would have left durable.
    ///
    /// # Panics
    ///
    /// Panics if rolling a file back to its durable image fails (open / write /
    /// remove on the inner backend). A crash simulator that could not restore
    /// the durable image would silently under-test recovery, so the failure is
    /// surfaced loudly rather than swallowed. In-memory backends never hit this.
    pub fn crash(&self) {
        self.crash_keeping(0);
    }

    /// How many ordered steps no device flush has made durable yet: each sync
    /// [`SyncMode::Barrier`] issued, and each removal or rename issued behind
    /// one, which the barrier orders like a sync. A power loss keeps some
    /// prefix of them, each step whole, which
    /// [`crash_keeping`](Self::crash_keeping) picks.
    #[must_use]
    pub fn ordered_steps(&self) -> usize {
        self.state.lock().ordered.len()
    }

    /// Simulates a power loss that keeps the first `kept` of the
    /// [ordered steps](Self::ordered_steps), in the order they were issued,
    /// and loses the rest, then rolls back as [`crash`](Self::crash) does.
    ///
    /// # Panics
    ///
    /// As [`crash`](Self::crash).
    pub fn crash_keeping(&self, kept: usize) {
        let mut state = self.state.lock();
        state.settle(kept);
        // An entry its directory never made durable is lost with whatever
        // content it had.
        let lost: Vec<PathBuf> = state.pending_entries.drain().collect();
        // A folder whose entry is lost takes everything under it, whatever
        // its own syncs made durable there. Shallowest first, so a folder
        // inside another lost one is gone with it.
        let mut lost_folders: Vec<&PathBuf> = lost
            .iter()
            .filter(|path| {
                !matches!(self.inner.read_link(path), Ok(Some(_)))
                    && self.inner.metadata(path).is_ok_and(|meta| meta.is_dir)
            })
            .collect();
        lost_folders.sort_by_key(|path| path.components().count());
        for folder in lost_folders {
            match self.inner.remove_dir_all(folder) {
                Ok(()) => {}
                Err(e) if e.kind() == io::ErrorKind::NotFound => {}
                Err(e) => panic!(
                    "crash(): removing the un-synced folder {} failed: {e}",
                    folder.display()
                ),
            }
            state.forget_under(folder);
        }
        for path in &lost {
            // One gone with a lost folder above it is gone already: its
            // removal below finds nothing.
            state.durable.remove(path);
            state.links.remove(path);
            state.link_group.remove(path);
            state.touched.insert(path.clone());
        }
        // Visit every path we wrote, plus any durable path (defensive: a file
        // synced in a prior life but only read this run still gets its durable
        // image reasserted).
        let paths: Vec<PathBuf> = state
            .touched
            .iter()
            .chain(state.durable.keys())
            .chain(state.links.keys())
            .cloned()
            .collect();

        for path in paths {
            if let Some(link) = state.links.get(&path) {
                self.restore_link(&path, link);
                continue;
            }
            if state.unlinked.contains_key(&path) {
                // Linked back below, once its group's names are restored.
                continue;
            }
            match state.durable.get(&path) {
                Some(bytes) => self.restore_durable(&path, bytes),
                None => {
                    // Never synced -> the file never existed durably.
                    match self.inner.remove_file(&path) {
                        Ok(()) => {}
                        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
                        Err(e) => {
                            panic!("crash(): removing un-synced {} failed: {e}", path.display())
                        }
                    }
                }
            }
        }

        // A hard link whose removal was lost comes back as a link of a name
        // its file still has, not as a copy of it.
        let unlinked: Vec<(PathBuf, u64)> = state.unlinked.drain().collect();
        for (path, group) in unlinked {
            self.relink(&state, &path, group);
        }

        // Post-crash, only durable files exist and they are "clean"; a link
        // put back is the disk's own again, untouched by this run.
        state.touched = state.durable.keys().cloned().collect();
        state.links.clear();
        let CrashState {
            durable,
            link_group,
            ..
        } = &mut *state;
        link_group.retain(|name, _| durable.contains_key(name));
    }

    /// Purges the crash state of every tracked path under the removed folder
    /// `path`, so a crash neither resurrects nor panics recreating a file
    /// whose parent is gone.
    fn forget_folder(&self, path: &Path) {
        let name = self.name_after(path);
        self.state.lock().forget_under(&name);
    }

    /// Captures what the name `name` holds before a queued step may take it,
    /// on its first touch this run: a file's bytes as its durable image, a
    /// symbolic link's target as the link to put back.
    fn capture_name(&self, name: &Path) -> io::Result<()> {
        match self.inner.read_link(name) {
            Ok(None) => self.capture_first_touch(name),
            Ok(Some(target)) => {
                let to_dir = self.inner.symlink_to_dir(name)?;
                let mut state = self.state.lock();
                if !state.touched.contains(name) && !state.links.contains_key(name) {
                    state
                        .links
                        .insert(name.to_path_buf(), Link { target, to_dir });
                }
                state.touched.insert(name.to_path_buf());
                Ok(())
            }
            // A probe that fails says nothing to capture; the step itself
            // reports what is wrong with the name.
            Err(_) => Ok(()),
        }
    }

    /// Makes `path` a hard link of a name of link group `group` on disk, in
    /// place of whatever the run left under it; with no such name left, the
    /// file is gone with every name, and so is `path`.
    fn relink(&self, state: &CrashState, path: &Path, group: u64) {
        let source = state
            .link_group
            .iter()
            .filter(|&(_, &other)| other == group)
            .map(|(name, _)| name.clone())
            .find(|name| matches!(self.inner.exists(name), Ok(true)));
        match self.inner.remove_file(path) {
            Ok(()) => {}
            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
            Err(e) => panic!(
                "crash(): clearing {} for its link failed: {e}",
                path.display()
            ),
        }
        let Some(source) = source else {
            return;
        };
        self.inner.hard_link(&source, path).unwrap_or_else(|e| {
            panic!(
                "crash(): linking {} back to {} failed: {e}",
                path.display(),
                source.display()
            )
        });
    }

    /// Puts back the symbolic link `path`, in place of whatever the run left
    /// under the name.
    fn restore_link(&self, path: &Path, link: &Link) {
        let target = &link.target;
        if matches!(self.inner.read_link(path), Ok(Some(now)) if now == *target)
            && matches!(self.inner.symlink_to_dir(path), Ok(kind) if kind == link.to_dir)
        {
            return;
        }
        match self.inner.remove_file(path) {
            Ok(()) => {}
            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
            Err(e) => panic!(
                "crash(): clearing {} for its link failed: {e}",
                path.display()
            ),
        }
        self.inner
            .symlink(target, path, link.to_dir)
            .unwrap_or_else(|e| {
                panic!(
                    "crash(): putting back the link {} failed: {e}",
                    path.display()
                )
            });
    }

    /// Overwrites the inner file at `path` with its durable image `bytes`.
    fn restore_durable(&self, path: &Path, bytes: &[u8]) {
        let mut file = self
            .inner
            .open(
                path,
                &FsOpenOptions::new().write(true).create(true).truncate(true),
            )
            .unwrap_or_else(|e| {
                panic!(
                    "crash(): reopening {} for rollback failed: {e}",
                    path.display()
                )
            });
        std::io::Write::write_all(&mut file, bytes).unwrap_or_else(|e| {
            panic!(
                "crash(): rewriting durable image of {} failed: {e}",
                path.display()
            )
        });
    }

    /// Reads the current content of `path` if it already exists, for use as the
    /// initial durable baseline. Returns `Ok(None)` only when the file does not
    /// exist (a brand-new file has no durable image until its first sync); any
    /// other open/read error is propagated rather than swallowed, so a transient
    /// read failure cannot silently drop a pre-existing file's durable image.
    fn read_baseline(&self, path: &Path) -> io::Result<Option<Vec<u8>>> {
        let mut f = match self.inner.open(path, &FsOpenOptions::new().read(true)) {
            Ok(f) => f,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e),
        };
        let mut buf = Vec::new();
        std::io::Read::read_to_end(&mut f, &mut buf)?;
        Ok(Some(buf))
    }

    /// Records `path`'s pre-existing content as its durable baseline on the FIRST
    /// touch of this run, then marks it touched. Re-touching is a no-op for the
    /// baseline (a file already touched is tracked by sync; re-reading it would
    /// wrongly promote un-synced bytes to durable). Used by every entry point
    /// that creates or mutates a file's content: `open` (writable), `punch_hole`,
    /// `truncate_file`.
    fn capture_first_touch(&self, path: &Path) -> io::Result<()> {
        let pb = path.to_path_buf();
        // A file is first touched when none of its names was: a write through
        // a linked name already put un-synced bytes in it, and its durable
        // image is the one its names share.
        let first_touch = {
            let state = self.state.lock();
            !state
                .names_of(path)
                .iter()
                .any(|name| state.touched.contains(name))
        };
        if first_touch {
            let baseline = self.read_baseline(path)?;
            if let Some(bytes) = baseline {
                self.state.lock().set_durable(path, bytes);
            }
        }
        self.state.lock().touched.insert(pb);
        Ok(())
    }

    /// The entry a write through `path` lands on: `path` with every symlink
    /// it passes through resolved, the final one included, which is the
    /// entry an open that creates makes and the one whose bytes a sync makes
    /// durable.
    fn entry_of(&self, path: &Path) -> io::Result<PathBuf> {
        self.resolve(path, true)
    }

    /// The entry `path` names itself, as `rename`, `unlink` and `link` see
    /// it: its directory resolved, its final component kept even when it is
    /// a symlink.
    fn name_of(&self, path: &Path) -> io::Result<PathBuf> {
        self.resolve(path, false)
    }

    /// [`Self::name_of`] for an operation the backend has already done: the
    /// operation stands, so a probe that cannot answer leaves the name as it
    /// was given rather than failing it.
    fn name_after(&self, path: &Path) -> PathBuf {
        self.name_of(path).unwrap_or_else(|_| path.to_path_buf())
    }

    /// `path` spelled the one way every name of its entry is tracked under,
    /// so a file reached through a symlinked directory or through `..` is the
    /// file reached through its real directory.
    ///
    /// Symlinks are resolved component by component, as the kernel walks a
    /// path, through at most `MAX_SYMLINKS` links. Then `.` and `..` are
    /// folded away, which is what the walk would do over a prefix that holds
    /// no symlink, but only when the backend agrees that the folded directory
    /// is the one it reaches: a backend that keeps `..` as part of a name
    /// (`MemFs`) gets the name it was given.
    fn resolve(&self, path: &Path, follow_final: bool) -> io::Result<PathBuf> {
        let mut resolved = PathBuf::new();
        // Components still to walk, the next one last.
        let mut rest: Vec<PathBuf> = components_reversed(path);
        let mut followed = 0;
        while let Some(part) = rest.pop() {
            if !matches!(part.components().next(), Some(Component::Normal(_))) {
                // A root replaces what was walked, `.` adds nothing, and `..`
                // is folded below, once the prefix it climbs out of is known
                // to hold no symlink.
                if !matches!(part.components().next(), Some(Component::CurDir)) {
                    resolved.push(part);
                }
                continue;
            }
            let candidate = resolved.join(&part);
            if rest.is_empty() && !follow_final {
                resolved = candidate;
                break;
            }
            match self.inner.read_link(&candidate)? {
                Some(target) => {
                    // MAX_SYMLINKS links are followed; only a link past them
                    // is refused.
                    if followed == MAX_SYMLINKS {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidInput,
                            "too many levels of symbolic links",
                        ));
                    }
                    followed += 1;
                    // A relative target continues from the directory holding
                    // the link; an absolute one starts again at its root.
                    rest.extend(components_reversed(&target));
                }
                None => resolved = candidate,
            }
        }
        if resolved.as_os_str().is_empty() {
            // `.` alone: the current directory, as the engine names it.
            resolved.push(".");
        }
        let folded = folded(&resolved);
        if folded == resolved {
            return Ok(resolved);
        }
        // The entry itself may not exist yet (an open that creates), its
        // directory does: that is what the backend is asked about.
        let (spelled, plain) = match (resolved.components().next_back(), folded.file_name()) {
            (Some(Component::Normal(_)), Some(_)) => (
                crate::file::entry_directory(&resolved).to_path_buf(),
                crate::file::entry_directory(&folded).to_path_buf(),
            ),
            _ => (resolved.clone(), folded.clone()),
        };
        Ok(
            if matches!(self.inner.same_file(&spelled, &plain), Ok(true)) {
                folded
            } else {
                resolved
            },
        )
    }

    /// Whether `path`'s directory lists an entry spelled exactly as its final
    /// component. A listing that cannot be read answers no.
    fn listed_as_spelled(&self, path: &Path) -> bool {
        let Some(name) = path.file_name() else {
            return false;
        };
        self.inner
            .read_dir(crate::file::entry_directory(path))
            .is_ok_and(|entries| {
                entries
                    .iter()
                    .any(|entry| std::ffi::OsStr::new(&entry.file_name) == name)
            })
    }

    /// Records that `path` got a new directory entry, durable once its parent
    /// directory is synced.
    fn new_entry(&self, path: &Path) {
        self.state.lock().pending_entries.insert(path.to_path_buf());
    }

    /// Runs the backend sync of `directory` and then makes its pending
    /// entries durable, with the namespace held throughout: no entry can be
    /// made or removed between the backend sync and what it is credited with.
    fn sync_entries_of(
        &self,
        directory: &Path,
        mode: SyncMode,
        sync: impl FnOnce() -> io::Result<()>,
    ) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        sync()?;
        // Entries are tracked under their resolved names, so the directory
        // is matched resolved too: a sync through a symlink or `..` covers
        // the entries of the directory it reaches. The sync is done, so a
        // probe that cannot answer matches the directory as it was named.
        let directory = self
            .entry_of(directory)
            .unwrap_or_else(|_| directory.to_path_buf());
        // A spelling that resolving does not fold, a case the filesystem
        // ignores or a bind mount, still reaches the same directory: what
        // the backend says about identity decides, outside the state lock.
        // The namespace is held, so the pending set cannot change meanwhile.
        let pending = self.state.lock().pending_entries.clone();
        // Each with whether it names a regular file, which a power loss
        // leaves in place once its entry is durable (see
        // `CrashState::entry_made_durable`); asked of the backend here, the
        // state lock not held.
        let covered: Vec<(PathBuf, bool)> = pending
            .into_iter()
            .filter(|entry| {
                let parent = crate::file::entry_directory(entry);
                parent == directory || matches!(self.inner.same_file(parent, &directory), Ok(true))
            })
            .map(|entry| {
                let file = !matches!(self.inner.read_link(&entry), Ok(Some(_)))
                    && self.inner.metadata(&entry).is_ok_and(|meta| meta.is_file);
                (entry, file)
            })
            .collect();
        let mut state = self.state.lock();
        // A barrier makes the entries durable only once the device is
        // flushed; any other sync flushes what was ordered before it too.
        if mode == SyncMode::Barrier {
            state.ordered.push(Ordered::Entries(covered));
            return Ok(());
        }
        state.settle_all();
        for (entry, file) in &covered {
            state.entry_made_durable(entry, *file);
        }
        Ok(())
    }

    /// Records the destination of a copy-style op (`hard_link` / `reflink`): it
    /// mirrors the source's durability so a crash either restores the
    /// linked/cloned bytes (durable source) or removes an un-synced copy.
    ///
    /// The source's durable image comes from one of two places: an entry already
    /// in `durable` (synced or captured this run), or, for a source that is
    /// pre-existing and never touched this run, its on-disk baseline (the same
    /// "current contents are initially durable" rule [`Self::capture_first_touch`]
    /// applies). A source that is touched but not durable holds un-synced bytes,
    /// so the copy correctly inherits no durable image and a crash removes it.
    ///
    /// Reads the baseline outside the state lock (mirroring
    /// [`Self::capture_first_touch`]) so backend I/O never runs under the mutex.
    fn track_copy(&self, src: &Path, dst: &Path) -> io::Result<()> {
        // The backend made `dst` already: record it as a written, pending
        // entry before the fallible baseline read, so a read that fails
        // leaves an entry a crash removes rather than one it never saw.
        let (src_durable, src_touched) = {
            let mut state = self.state.lock();
            state.touched.insert(dst.to_path_buf());
            state.pending_entries.insert(dst.to_path_buf());
            (state.durable.get(src).cloned(), state.touched.contains(src))
        };
        let dst_image = match src_durable {
            Some(bytes) => Some(bytes),
            None if !src_touched => self.read_baseline(src)?,
            None => None,
        };
        if let Some(bytes) = dst_image {
            self.state.lock().durable.insert(dst.to_path_buf(), bytes);
        }
        Ok(())
    }
}

impl Fs for CrashFs {
    fn open(&self, path: &Path, opts: &FsOpenOptions) -> io::Result<Box<dyn FsFile>> {
        let writable = opts.write || opts.create || opts.create_new || opts.append || opts.truncate;
        // An open holds the namespace from resolving its entry to the
        // registration: the backend follows the same symlinks to the same
        // file, and whether an open that may create made the entry is what
        // the open did. A handle opened only for reading resolves too, since
        // a sync through it makes the file it reached durable.
        let may_create = opts.create || opts.create_new;
        let namespace = self.namespace.lock();
        // An open that may only create (`O_EXCL`) does not follow a final
        // symlink and refuses any existing name: the backend answers whether
        // it can, and if it does the file is new, at the path itself, with no
        // prior image to capture.
        if opts.create_new {
            let inner = self.inner.open(path, opts)?;
            let entry = self.name_after(path);
            self.state.lock().touched.insert(entry.clone());
            self.new_entry(&entry);
            drop(namespace);
            return Ok(Box::new(CrashFile {
                inner,
                path: entry,
                fs: Arc::clone(&self.inner),
                state: Arc::clone(&self.state),
            }));
        }
        // A write follows symlinks: through a dangling one it creates the
        // target, so the target is the entry made and the file tracked.
        if !writable {
            // The open stands, so a probe that cannot answer leaves the
            // handle tracked under the name it was opened with.
            let inner = self.inner.open(path, opts)?;
            let entry = self.entry_of(path).unwrap_or_else(|_| path.to_path_buf());
            drop(namespace);
            return Ok(Box::new(CrashFile {
                inner,
                path: entry,
                fs: Arc::clone(&self.inner),
                state: Arc::clone(&self.state),
            }));
        }
        let entry = self.entry_of(path)?;
        let creates = may_create && !self.inner.exists(&entry)?;
        // Capture the pre-existing durable image BEFORE the open (which may
        // truncate); a brand-new file captures nothing, so a crash before its
        // first sync removes it.
        self.capture_first_touch(&entry)?;
        let inner = self.inner.open(path, opts)?;
        if creates {
            self.new_entry(&entry);
        }
        drop(namespace);
        Ok(Box::new(CrashFile {
            inner,
            path: entry,
            fs: Arc::clone(&self.inner),
            state: Arc::clone(&self.state),
        }))
    }

    fn create_dir_all(&self, path: &Path) -> io::Result<()> {
        if !self.folders {
            return self.inner.create_dir_all(path);
        }
        let _namespace = self.namespace.lock();
        // Each folder of the chain that did not exist gets an entry in the one
        // above it, durable only once that one is synced.
        // A root has no entry anywhere, whatever the backend says of it.
        let mut made = Vec::new();
        let mut at = path;
        while !self.inner.exists(at)? {
            let Some(parent) = at.parent() else {
                break;
            };
            made.push(at.to_path_buf());
            // A bare name's entry is in the current directory.
            if parent.as_os_str().is_empty() {
                break;
            }
            at = parent;
        }
        self.inner.create_dir_all(path)?;
        for folder in made {
            self.new_entry(&self.name_after(&folder));
        }
        Ok(())
    }

    fn create_dir(&self, path: &Path) -> io::Result<()> {
        if !self.folders {
            return self.inner.create_dir(path);
        }
        let _namespace = self.namespace.lock();
        self.inner.create_dir(path)?;
        self.new_entry(&self.name_after(path));
        Ok(())
    }

    fn read_dir(&self, path: &Path) -> io::Result<Vec<FsDirEntry>> {
        self.inner.read_dir(path)
    }

    fn remove_file(&self, path: &Path) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        // A removal that may be queued (see below) can be lost, and then the
        // file is back as it was: a file this run never touched has its bytes
        // captured now, before they are gone. A symlink names no bytes of
        // its own. Captured whether or not a step is queued yet: a barrier
        // sync on another thread can queue one before the removal is
        // recorded, and only the namespace changes are serialized here.
        let name = self.name_after(path);
        self.capture_name(&name)?;
        self.inner.remove_file(path)?;
        let name = self.name_after(path);
        let mut state = self.state.lock();
        // Behind an ordered sync, the removal is ordered too: a power loss
        // that keeps only a prefix of the syncs may lose it, so what the file
        // went through stays queued, in order, for the crash to replay.
        if state.ordered.is_empty() {
            state.forget_name(&name);
            return Ok(());
        }
        let was_pending = state.pending_entries.contains(&name);
        // A name with hard links leaves its group now, so a file made under
        // it later is its own; a lost removal links it back to the group.
        if state.names_of(&name).len() > 1
            && let Some(group) = state.link_group.remove(&name)
        {
            if state.touched.contains(&name) {
                let survivors: Vec<PathBuf> = state
                    .link_group
                    .iter()
                    .filter(|&(_, &other)| other == group)
                    .map(|(other, _)| other.clone())
                    .collect();
                state.touched.extend(survivors);
            }
            state.durable.remove(&name);
            state.unlinked.insert(name.clone(), group);
        }
        state.touched.insert(name.clone());
        state.ordered.push(Ordered::Removed(name, was_pending));
        Ok(())
    }

    fn remove_dir(&self, path: &Path) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        self.inner.remove_dir(path)?;
        self.forget_folder(path);
        Ok(())
    }

    fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        self.inner.remove_dir_all(path)?;
        self.forget_folder(path);
        Ok(())
    }

    fn same_file(&self, a: &Path, b: &Path) -> io::Result<bool> {
        self.inner.same_file(a, b)
    }

    fn read_link(&self, path: &Path) -> io::Result<Option<PathBuf>> {
        self.inner.read_link(path)
    }

    fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        // POSIX rename(2): when both entries are links to one file (one path,
        // or two hard links of one inode) the call succeeds and changes
        // nothing, and the crash state stays as it was. Two names the backend
        // reports as one file may still be two entries (two symlinks to one
        // file) or one entry under two spellings (a filesystem that ignores
        // case), so whether the rename changed anything is told by the exact
        // spellings its directories list, before and after it. The namespace
        // is held, so nothing else changes them meanwhile, and the inner
        // backend is the disk this simulates: a fault layer composes above the
        // simulator, never below it.
        let spelled = || (self.listed_as_spelled(from), self.listed_as_spelled(to));
        let before = matches!(self.inner.same_file(from, to), Ok(true)).then(spelled);
        // A rename that may be queued (see below) can be lost, and then both
        // names are back as they were: files this run never touched have
        // their bytes captured now, before the rename moves or replaces them.
        // A symlink names no bytes of its own. Captured whether or not a step
        // is queued yet, as a removal's are.
        if from != to {
            for name in [self.name_after(from), self.name_after(to)] {
                self.capture_name(&name)?;
            }
        }
        self.inner.rename(from, to)?;
        // Without a file identity to go on (a backend that has none, a
        // dangling symlink), the source still being there, the link itself
        // rather than what it points to, under its own spelling, is what
        // tells. A probe that fails is read as an ordinary rename.
        let unchanged = match before {
            Some(before) => before == spelled(),
            None => {
                (matches!(self.inner.read_link(from), Ok(Some(_)))
                    || matches!(self.inner.exists(from), Ok(true)))
                    && self.listed_as_spelled(from)
            }
        };
        if from == to || unchanged {
            return Ok(());
        }
        let (from, to) = (self.name_after(from), self.name_after(to));
        let (from, to) = (from.as_path(), to.as_path());
        let mut state = self.state.lock();
        // Behind an ordered sync, the rename is ordered too, as a removal is,
        // and as one step: a power loss that keeps only a prefix of the syncs
        // either keeps it whole or loses it, and with it everything after,
        // leaving both names as the kept syncs made them. Kept, the
        // destination's old file and the source's name are gone and the
        // destination holds the bytes the source had durable or queued by
        // then. Names with hard links keep the immediate rename.
        if !state.ordered.is_empty()
            && state.names_of(from).len() == 1
            && state.names_of(to).len() <= 1
        {
            // The last queued step that took the source's name from the file
            // it held (a removal, a rename away or onto it): what the name
            // had before belongs to that file. The file renamed now has only
            // what that step and the ones after gave it.
            let since = state.ordered.iter().rposition(|sync| match sync {
                Ordered::Removed(name, _) => name.as_path() == from,
                Ordered::Renamed {
                    from: source,
                    to: target,
                    ..
                } => source.as_path() == from || target.as_path() == from,
                _ => false,
            });
            let queued = state
                .ordered
                .iter()
                .skip(since.unwrap_or(0))
                .rev()
                .find_map(|sync| match sync {
                    Ordered::Image(name, bytes) if name.as_path() == from => Some(bytes.clone()),
                    Ordered::Renamed {
                        to: target,
                        bytes: Some(bytes),
                        ..
                    } if target.as_path() == from => Some(bytes.clone()),
                    _ => None,
                });
            let bytes = queued.or_else(|| {
                since
                    .is_none()
                    .then(|| state.durable.get(from).cloned())
                    .flatten()
            });
            let to_pending = state.pending_entries.contains(to);
            let from_pending = state.pending_entries.contains(from);
            state.ordered.push(Ordered::Renamed {
                from: from.to_path_buf(),
                to: to.to_path_buf(),
                from_pending,
                to_pending,
                bytes,
            });
            state.touched.insert(from.to_path_buf());
            state.touched.insert(to.to_path_buf());
            // The destination's entry is new until its directory is synced.
            state.pending_entries.insert(to.to_path_buf());
            if let Some(group) = state.link_group.remove(from) {
                state.link_group.insert(to.to_path_buf(), group);
            }
            return Ok(());
        }
        // The source's state moves with its file, under the new name.
        let from_durable = state.durable.remove(from);
        let from_link = state.links.remove(from);
        let from_touched = state.touched.remove(from);
        let from_group = state.link_group.remove(from);
        state.pending_entries.remove(from);
        // The destination is replaced on disk, as a removal of that name: its
        // durable image and write-tracking go (so a replaced, previously
        // synced destination is not resurrected to its stale content on
        // crash), and what its file went through stays with the file's other
        // hard links.
        state.forget_name(to);
        // An ordered sync of the source's content follows its file; one of
        // its entry does not, since the new entry is pending until its own
        // directory is synced.
        for sync in &mut state.ordered {
            match sync {
                Ordered::Image(name, _) if name.as_path() == from => *name = to.to_path_buf(),
                Ordered::Image(..) | Ordered::Removed(..) | Ordered::Renamed { .. } => {}
                Ordered::Entries(entries) => entries.retain(|(entry, _)| entry != from),
            }
        }
        // The rename is as durable as the source was.
        if let Some(bytes) = from_durable {
            state.durable.insert(to.to_path_buf(), bytes);
        }
        if let Some(target) = from_link {
            state.links.insert(to.to_path_buf(), target);
        }
        if from_touched {
            state.touched.insert(to.to_path_buf());
        }
        // The destination's entry is new until its directory is synced.
        state.pending_entries.insert(to.to_path_buf());
        if let Some(group) = from_group {
            state.link_group.insert(to.to_path_buf(), group);
        }
        Ok(())
    }

    fn metadata(&self, path: &Path) -> io::Result<FsMetadata> {
        self.inner.metadata(path)
    }

    fn sync_directory(&self, path: &Path) -> io::Result<()> {
        self.sync_entries_of(path, SyncMode::Full, || self.inner.sync_directory(path))
    }

    fn sync_directory_with(&self, path: &Path, mode: SyncMode) -> io::Result<()> {
        self.sync_entries_of(path, mode, || self.inner.sync_directory_with(path, mode))
    }

    fn sync_device(&self, path: &Path) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        self.inner.sync_device(path)?;
        self.state.lock().settle_all();
        Ok(())
    }

    fn exists(&self, path: &Path) -> io::Result<bool> {
        self.inner.exists(path)
    }

    fn hard_link(&self, src: &Path, dst: &Path) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        self.inner.hard_link(src, dst)?;
        // A symlink linked as itself (Linux `linkat(2)` without
        // `AT_SYMLINK_FOLLOW`) is a new entry with no bytes of its own: no
        // image is kept under it, and it is grouped with nothing. The link is
        // made, so a probe that cannot answer does not fail it either.
        let link = self.name_after(dst);
        if matches!(self.inner.read_link(dst), Ok(Some(_))) {
            self.new_entry(&link);
            return Ok(());
        }
        // A backend that followed a symlink source made `dst` a name of the
        // file the link chain ends at, so that file is the one copied and
        // the name `dst` is grouped with.
        let file = self.entry_of(src).unwrap_or_else(|_| self.name_after(src));
        self.track_copy(&file, &link)?;
        // On a backend where a hard link is a second name of one file, a sync
        // through either name makes the other's bytes durable too; on one
        // where it is a copy (`MemFs`), each name keeps its own image. A
        // probe that cannot answer leaves the names separate.
        if matches!(self.inner.same_file(&file, &link), Ok(true)) {
            let mut state = self.state.lock();
            state.link(&file, &link);
            // The image `dst` took is the one file's, `src`'s baseline
            // included: every name holds it, so a later first touch through
            // any of them, which finds the group already touched, still has
            // durable bytes to roll back to.
            if let Some(bytes) = state.durable.get(&link).cloned() {
                state.set_durable(&link, bytes);
            }
        }
        Ok(())
    }

    fn backend_id(&self) -> Option<u64> {
        self.inner.backend_id()
    }

    fn write_io_stats(&self) -> Option<&Arc<super::WriteIoStats>> {
        self.inner.write_io_stats()
    }

    fn write_budget(&self) -> Option<&Arc<crate::write_budget::WriteBudget>> {
        self.inner.write_budget()
    }

    fn volume_id(&self, path: &Path) -> Option<u64> {
        self.inner.volume_id(path)
    }

    fn capabilities(&self, path: &Path) -> FsCapabilities {
        self.inner.capabilities(path)
    }

    fn try_disable_cow(&self, path: &Path) -> io::Result<()> {
        self.inner.try_disable_cow(path)
    }

    fn punch_hole(&self, path: &Path, offset: u64, len: u64) -> io::Result<()> {
        // Content-mutating: capture the pre-mutation durable image so an
        // un-synced punch rolls back on crash. The namespace is held so the
        // backend reaches the file that was resolved.
        let _namespace = self.namespace.lock();
        self.capture_first_touch(&self.entry_of(path)?)?;
        self.inner.punch_hole(path, offset, len)
    }

    fn reflink_file(&self, src: &Path, dst: &Path) -> io::Result<()> {
        let _namespace = self.namespace.lock();
        self.inner.reflink_file(src, dst)?;
        let file = self.entry_of(src).unwrap_or_else(|_| self.name_after(src));
        self.track_copy(&file, &self.name_after(dst))?;
        Ok(())
    }

    fn truncate_file(&self, path: &Path) -> io::Result<()> {
        // Content-mutating: capture the pre-truncate image so an un-synced
        // reclaim rolls back on crash. The namespace is held so the backend
        // reaches the file that was resolved.
        let _namespace = self.namespace.lock();
        self.capture_first_touch(&self.entry_of(path)?)?;
        self.inner.truncate_file(path)
    }

    fn hard_link_count(&self, path: &Path) -> io::Result<u64> {
        self.inner.hard_link_count(path)
    }

    fn available_space(&self, path: &Path) -> io::Result<u64> {
        self.inner.available_space(path)
    }

    fn allocated_size(&self, path: &Path) -> io::Result<Option<u64>> {
        self.inner.allocated_size(path)
    }

    fn extent_is_hole(&self, path: &Path, offset: u64, len: u64) -> io::Result<Option<bool>> {
        self.inner.extent_is_hole(path, offset, len)
    }

    fn extent_contains_hole(&self, path: &Path, offset: u64, len: u64) -> io::Result<Option<bool>> {
        self.inner.extent_contains_hole(path, offset, len)
    }
}

/// A file handle that records its durable image on `fsync`.
struct CrashFile {
    inner: Box<dyn FsFile>,
    path: PathBuf,
    /// Backend handle, used to reopen `path` read-only when snapshotting its
    /// content on `fsync` (the write handle may lack read access).
    fs: Arc<dyn Fs>,
    state: Arc<spin::Mutex<CrashState>>,
}

impl CrashFile {
    /// Captures this path's full current content as its durable image. Called
    /// after a successful `fsync`. Reopens the path read-only because the write
    /// handle being synced is not necessarily readable.
    fn snapshot(&self, mode: SyncMode) -> io::Result<()> {
        let mut rf = self.fs.open(&self.path, &FsOpenOptions::new().read(true))?;
        let mut buf = Vec::new();
        std::io::Read::read_to_end(&mut rf, &mut buf)?;
        let mut state = self.state.lock();
        // A barrier makes the content durable only once the device is
        // flushed; any other sync flushes what was ordered before it too.
        if mode == SyncMode::Barrier {
            state.ordered.push(Ordered::Image(self.path.clone(), buf));
        } else {
            state.settle_all();
            state.set_durable(&self.path, buf);
        }
        Ok(())
    }
}

impl std::io::Read for CrashFile {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.inner.read(buf)
    }
}

impl std::io::Write for CrashFile {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.inner.write(buf)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        // flush pushes to the backend but is NOT a durability barrier; the
        // durable image is captured on sync, not flush.
        self.inner.flush()
    }
}

impl std::io::Seek for CrashFile {
    fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(pos)
    }
}

impl FsFile for CrashFile {
    fn sync_all(&self) -> io::Result<()> {
        self.inner.sync_all()?;
        self.snapshot(SyncMode::Full)
    }

    fn sync_data(&self) -> io::Result<()> {
        self.inner.sync_data()?;
        self.snapshot(SyncMode::Full)
    }

    fn sync_all_with(&self, mode: SyncMode) -> io::Result<()> {
        self.inner.sync_all_with(mode)?;
        self.snapshot(mode)
    }

    fn sync_data_with(&self, mode: SyncMode) -> io::Result<()> {
        self.inner.sync_data_with(mode)?;
        self.snapshot(mode)
    }

    fn start_writeback(&self, offset: u64, len: u64) -> io::Result<()> {
        // Writeback makes nothing durable; only a sync does.
        self.inner.start_writeback(offset, len)
    }

    fn metadata(&self) -> io::Result<FsMetadata> {
        self.inner.metadata()
    }

    fn hard_link_count(&self) -> io::Result<u64> {
        self.inner.hard_link_count()
    }

    fn set_len(&self, size: u64) -> io::Result<()> {
        self.inner.set_len(size)
    }

    fn read_at(&self, buf: &mut [u8], offset: u64) -> io::Result<usize> {
        self.inner.read_at(buf, offset)
    }

    fn read_at_buf(&self, buf: &mut super::ReadBuf<'_>, offset: u64) -> io::Result<()> {
        self.inner.read_at_buf(buf, offset)
    }

    fn read_many_buf(&self, regions: &mut [(u64, super::ReadBuf<'_>)]) -> io::Result<()> {
        super::read_buf::read_each(self, regions)
    }

    fn lock_exclusive(&self) -> io::Result<()> {
        self.inner.lock_exclusive()
    }

    fn try_lock_exclusive(&self) -> io::Result<bool> {
        self.inner.try_lock_exclusive()
    }

    fn hint(&self, hint: FileHint) -> io::Result<()> {
        self.inner.hint(hint)
    }

    fn set_write_lifetime(&self, lifetime: super::WriteLifetime) -> io::Result<()> {
        self.inner.set_write_lifetime(lifetime)
    }
}

#[cfg(test)]
#[expect(clippy::unwrap_used, clippy::expect_used, reason = "test code")]
mod tests;