coordinode-lsm-tree 5.7.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Structured World Foundation

//! In-memory [`Fs`] implementation for testing and ephemeral trees.
//!
//! All file data lives in memory - there are no durability guarantees.
//! `sync_all`, `sync_data`, and `sync_directory` are deliberate no-ops.
//!
//! # Known limitations
//!
//! - **Compaction**: Some code paths in the compaction finalization still
//!   bypass the `Fs` trait. Write + flush + point-read works; compaction
//!   may fail with `ENOENT` on virtual paths.

use super::{Fs, FsCapabilities, FsDirEntry, FsFile, FsMetadata, FsOpenOptions};
use crate::io::{self, SeekFrom};
// Trait names referenced only by the no_std trait impls below (the std impls
// target `std::io::*` directly, so these would be unused under `std`).
#[cfg(not(feature = "std"))]
use crate::io::{Read, Seek, Write};
use crate::path::{Path, PathBuf};
#[cfg(not(feature = "std"))]
use alloc::borrow::ToOwned;
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, vec::Vec};
// no_std-capable primitives so this reference backend compiles on
// `--no-default-features --features alloc` (it's the template a no_std
// consumer copies for a real backend, e.g. WASM/IndexedDB): `spin` locks
// (no poisoning, userspace), `hashbrown` maps. The locks see no real
// contention here — a single ephemeral in-memory tree — so spin is fine.
use hashbrown::{HashMap, HashSet};
use spin::{Mutex, RwLock};

// ---------------------------------------------------------------------------
// MemFs
// ---------------------------------------------------------------------------

/// In-memory [`Fs`] backend for testing and ephemeral in-memory trees.
///
/// Backed by a `HashMap<PathBuf, Arc<Mutex<Vec<u8>>>>` - no disk I/O is
/// performed. Clones share the same backing store, and individual file
/// contents are synchronized through a per-file [`Mutex`].
///
/// # Example
///
/// ```
/// use lsm_tree::fs::MemFs;
/// use std::sync::Arc;
///
/// let fs = MemFs::new();
/// let dyn_fs: Arc<dyn lsm_tree::fs::Fs> = Arc::new(fs);
/// ```
#[derive(Clone, Debug)]
pub struct MemFs {
    state: Arc<RwLock<State>>,
    /// Per-instance namespace ID used by [`Fs::backend_id`]. Cloned
    /// `MemFs` values share the same `state` Arc AND the same ID - they
    /// are the same backend by all observable behaviour. Independently
    /// constructed `MemFs::new()` values get DIFFERENT IDs because they
    /// have disjoint file trees.
    namespace_id: u64,
    /// Total simulated disk capacity in bytes. `u64::MAX` (default) means
    /// "unbounded": [`Fs::available_space`] then reports `u64::MAX` (no disk
    /// pressure). When set to a finite value via [`MemFs::with_capacity`] /
    /// [`MemFs::set_capacity`], `available_space` reports `capacity − bytes
    /// stored`, so the simulated disk fills as data is written and reaches zero
    /// when full — a real capped disk. Shared across clones (same backend).
    ///
    /// `portable_atomic::AtomicU64` (not `core`'s): native 64-bit atomics are
    /// absent on some `no_std` targets (e.g. thumbv7em).
    capacity: Arc<portable_atomic::AtomicU64>,
    /// LIFETIME total of distinct bytes ever reclaimed by [`Fs::punch_hole`] on
    /// this simulated disk, exposed via [`MemFs::punched_bytes`] purely so a test
    /// can assert that an in-place reclaim FIRED (and roughly how much), even
    /// after the punched files are later deleted. Monotonic; counts each punch's
    /// newly-freed bytes once (overlapping re-punches add nothing). This is NOT
    /// what drives free space — [`Self::stored_bytes`] uses the per-file
    /// [`State::punched`] ranges so a removed/truncated file stops freeing space.
    punched_total: Arc<portable_atomic::AtomicU64>,
    /// Set once any [`Fs::punch_hole`] has recorded a range, so the common
    /// (never-punched) write path can skip punched-range invalidation with a
    /// single relaxed atomic load instead of taking the state lock per write.
    has_punches: Arc<portable_atomic::AtomicBool>,
    /// Whether [`Fs::capabilities`] advertises `punch_hole` (default `true`).
    /// A test sets this `false` via [`MemFs::set_punch_hole_supported`] to drive
    /// the capability-gated fallback (tight-space compaction skips on a backend
    /// that cannot punch). Shared across clones.
    punch_hole_supported: Arc<portable_atomic::AtomicBool>,
}

#[derive(Debug, Default)]
struct State {
    files: HashMap<PathBuf, Arc<Mutex<Vec<u8>>>>,
    dirs: HashSet<PathBuf>,
    /// Per-file reclaimed (punched) byte ranges, kept non-overlapping and
    /// merged. Subtracted from each file's logical length in [`MemFs::stored_bytes`]
    /// so the simulated disk reflects in-place extent reclaim (real
    /// `fallocate(PUNCH_HOLE)` frees physical blocks while the logical length is
    /// unchanged). Tracking PER FILE (not a global counter) keeps the accounting
    /// correct when a punched file is removed, renamed, truncated, or overwritten,
    /// and merging ranges keeps overlapping/cumulative punches from double-counting.
    punched: HashMap<PathBuf, Vec<(u64, u64)>>,
}

/// Merges `[start, end)` into a sorted, non-overlapping range set. Overlapping
/// or adjacent ranges coalesce, so cumulative prefix punches never double-count.
fn merge_punched_range(ranges: &mut Vec<(u64, u64)>, start: u64, end: u64) {
    if start >= end {
        return;
    }
    ranges.push((start, end));
    ranges.sort_unstable();
    let mut merged: Vec<(u64, u64)> = Vec::with_capacity(ranges.len());
    for &(s, e) in ranges.iter() {
        match merged.last_mut() {
            Some(last) if s <= last.1 => last.1 = last.1.max(e),
            _ => merged.push((s, e)),
        }
    }
    *ranges = merged;
}

/// Removes `[start, end)` from a sorted, non-overlapping range set (splitting a
/// straddling range). Used when a later write or `set_len` re-materializes
/// previously-punched bytes so they stop counting as reclaimed.
fn subtract_punched_range(ranges: &mut Vec<(u64, u64)>, start: u64, end: u64) {
    if start >= end {
        return;
    }
    let mut out: Vec<(u64, u64)> = Vec::with_capacity(ranges.len() + 1);
    for &(s, e) in ranges.iter() {
        // Keep the slice of [s, e) that falls before `start` and after `end`.
        if s < start {
            out.push((s, start.min(e)));
        }
        if e > end {
            out.push((s.max(end), e));
        }
    }
    out.retain(|&(s, e)| s < e);
    *ranges = out;
}

/// Total punched bytes of one file's range set, clipped to its current `len`
/// (ranges past a later truncation no longer count).
fn clipped_punched_len(ranges: &[(u64, u64)], len: u64) -> u64 {
    ranges
        .iter()
        .map(|&(s, e)| {
            let e = e.min(len);
            let s = s.min(e);
            e - s
        })
        .sum()
}

impl MemFs {
    /// Creates a new, empty in-memory filesystem.
    #[must_use]
    pub fn new() -> Self {
        let mut state = State::default();
        // Seed the root directory so exists("/") and read_dir("/") work.
        state.dirs.insert(PathBuf::from("/"));
        Self {
            state: Arc::new(RwLock::new(state)),
            namespace_id: next_mem_fs_namespace_id(),
            capacity: Arc::new(portable_atomic::AtomicU64::new(u64::MAX)),
            punched_total: Arc::new(portable_atomic::AtomicU64::new(0)),
            has_punches: Arc::new(portable_atomic::AtomicBool::new(false)),
            punch_hole_supported: Arc::new(portable_atomic::AtomicBool::new(true)),
        }
    }

    /// Toggles whether [`Fs::capabilities`] advertises `punch_hole`. Lets a test
    /// exercise the capability-gated fallback where tight-space compaction skips
    /// because the backend cannot reclaim extents in place.
    pub fn set_punch_hole_supported(&self, supported: bool) {
        self.punch_hole_supported
            .store(supported, portable_atomic::Ordering::Relaxed);
    }

    /// Creates an empty in-memory filesystem with a fixed total capacity in
    /// bytes — a simulated capped disk. [`Fs::available_space`] reports
    /// `capacity − bytes stored`, so the disk fills as data is written and the
    /// storage-admission gate drives the tree read-only when it is full,
    /// without any manual free-space poking. `u64::MAX` means unbounded (same
    /// as [`MemFs::new`]).
    #[must_use]
    pub fn with_capacity(capacity_bytes: u64) -> Self {
        let fs = Self::new();
        fs.set_capacity(capacity_bytes);
        fs
    }

    /// Sets the simulated total disk capacity (shared across clones). See
    /// [`MemFs::with_capacity`]. `u64::MAX` restores unbounded behaviour.
    pub fn set_capacity(&self, capacity_bytes: u64) {
        self.capacity
            .store(capacity_bytes, portable_atomic::Ordering::Relaxed);
    }

    /// Total bytes currently stored across all files (the simulated disk
    /// usage). Sums every file's length under the state read lock.
    fn stored_bytes(&self) -> u64 {
        let state = self.state.read();
        // Each file contributes its logical length minus the bytes punched out of
        // it (clipped to the current length). Per-file accounting means a removed
        // or truncated file stops subtracting stale reclaim. The sum is bounded by
        // the simulated capacity, so it cannot overflow u64.
        state
            .files
            .iter()
            .map(|(path, data)| {
                let len = data.lock().len() as u64;
                let punched = state
                    .punched
                    .get(path)
                    .map_or(0, |ranges| clipped_punched_len(ranges, len));
                len - punched
            })
            .sum()
    }

    /// LIFETIME total of distinct bytes reclaimed by [`Fs::punch_hole`] on this
    /// simulated disk. Lets a test assert that an in-place extent reclaim (e.g.
    /// the tight-space compaction prefix punch) actually fired and roughly how
    /// much — it stays counted even after the punched files are deleted, so it is
    /// the right metric for "did the rewrite punch incrementally?". It is NOT the
    /// current free space: [`Fs::available_space`] reflects that, dropping a
    /// removed/truncated file's reclaim.
    #[must_use]
    pub fn punched_bytes(&self) -> u64 {
        self.punched_total.load(portable_atomic::Ordering::Relaxed)
    }
}

/// Allocates the next per-instance `MemFs` namespace ID. Values are
/// process-unique (monotonic atomic counter) so two `MemFs::new()`
/// values never collide; cloned `MemFs` instances reuse the same ID
/// because `MemFs` derives `Clone`.
fn next_mem_fs_namespace_id() -> u64 {
    use core::sync::atomic::{AtomicU32, Ordering};
    // `AtomicU32`, not `AtomicU64`: 64-bit atomics are unavailable on some
    // no_std targets (e.g. thumbv7em). u32 IDs are ample for distinct
    // in-memory backends in one process; widened to u64 at the call site.
    // Start at 1 so a future `0` sentinel stays available if needed.
    static COUNTER: AtomicU32 = AtomicU32::new(1);
    u64::from(COUNTER.fetch_add(1, Ordering::Relaxed))
}

impl Default for MemFs {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// MemFile
// ---------------------------------------------------------------------------

/// An open file handle backed by an in-memory buffer.
struct MemFile {
    data: Arc<Mutex<Vec<u8>>>,
    cursor: u64,
    readable: bool,
    writable: bool,
    is_append: bool,
    /// Shared `MemFs` state + this file's path, so a write / `set_len` that
    /// re-materializes previously-punched bytes can drop the stale reclaim from
    /// [`State::punched`]. Gated by `has_punches` so the never-punched common
    /// path stays lock-free.
    state: Arc<RwLock<State>>,
    path: PathBuf,
    has_punches: Arc<portable_atomic::AtomicBool>,
}

/// Copies bytes from `data[pos..]` into `buf`, returning byte count.
fn copy_from_data(buf: &mut [u8], data: &[u8], pos: usize) -> usize {
    let available = data.get(pos..).unwrap_or_default();
    let n = buf.len().min(available.len());
    if let (Some(dst), Some(src)) = (buf.get_mut(..n), available.get(..n)) {
        dst.copy_from_slice(src);
    }
    n
}

// Bodies live on inherent `*_impl` methods returning `crate::io::Result`; the
// trait impls are dual-gated thin wrappers. Under `std`, `crate::io::{Read,
// Write,Seek}` are method-less supertrait aliases (blanket-impl'd for
// `std::io::*`), so the real impl must target `std::io::*` there and bridge the
// error back via `Into`; under `no_std` it targets the native `crate::io::*`.
impl MemFile {
    fn read_impl(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if !self.readable {
            return Err(io::Error::other("file not opened for reading"));
        }
        let data = lock(&self.data)?;
        let pos = usize::try_from(self.cursor).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "cursor exceeds addressable memory",
            )
        })?;
        let n = copy_from_data(buf, &data, pos);
        drop(data);
        self.cursor += n as u64;
        Ok(n)
    }

    fn write_impl(&mut self, buf: &[u8]) -> io::Result<usize> {
        if !self.writable {
            return Err(io::Error::other("file not opened for writing"));
        }
        if buf.is_empty() {
            return Ok(0);
        }
        let mut data = lock(&self.data)?;

        let pos = if self.is_append {
            data.len()
        } else {
            usize::try_from(self.cursor).map_err(|_| {
                io::Error::new(
                    io::ErrorKind::InvalidInput,
                    "write position exceeds addressable memory",
                )
            })?
        };

        let end = pos.checked_add(buf.len()).ok_or_else(|| {
            io::Error::new(io::ErrorKind::InvalidInput, "write position overflow")
        })?;
        if end > data.len() {
            data.resize(end, 0);
        }
        if let Some(dst) = data.get_mut(pos..end) {
            dst.copy_from_slice(buf);
        }
        drop(data);
        // The written span re-materializes any punched bytes it overlaps.
        self.drop_punched_overlap(pos as u64, end as u64);
        self.cursor = end as u64;
        Ok(buf.len())
    }

    /// Resolves the path this handle's backing buffer lives under RIGHT NOW.
    /// The captured `self.path` goes stale after a `rename`, but the data `Arc`
    /// is the stable identity, so match it against `state.files` (fast-path the
    /// captured path). Returns `None` if the file was removed.
    fn current_path(&self, state: &State) -> Option<PathBuf> {
        if state
            .files
            .get(&self.path)
            .is_some_and(|data| Arc::ptr_eq(data, &self.data))
        {
            return Some(self.path.clone());
        }
        state
            .files
            .iter()
            .find(|(_, data)| Arc::ptr_eq(data, &self.data))
            .map(|(path, _)| path.clone())
    }

    /// Drops `[start, end)` from this file's punched ranges (a write re-wrote
    /// those bytes, so they are no longer reclaimed). Lock-free no-op until some
    /// punch has happened. Acquired AFTER the data lock is released to keep the
    /// state→data lock order one-directional with `punch_hole`.
    fn drop_punched_overlap(&self, start: u64, end: u64) {
        if !self.has_punches.load(portable_atomic::Ordering::Relaxed) {
            return;
        }
        let mut state = self.state.write();
        let Some(path) = self.current_path(&state) else {
            return;
        };
        if let Some(ranges) = state.punched.get_mut(&path) {
            subtract_punched_range(ranges, start, end);
            if ranges.is_empty() {
                state.punched.remove(&path);
            }
        }
    }

    /// Clips this file's punched ranges to `new_len` after a `set_len`, so a
    /// shrink permanently removes past-end reclaim (it cannot resurrect on a
    /// later grow).
    fn clip_punched_to(&self, new_len: u64) {
        if !self.has_punches.load(portable_atomic::Ordering::Relaxed) {
            return;
        }
        let mut state = self.state.write();
        let Some(path) = self.current_path(&state) else {
            return;
        };
        if let Some(ranges) = state.punched.get_mut(&path) {
            ranges.retain_mut(|(s, e)| {
                *e = (*e).min(new_len);
                *s < *e
            });
            if ranges.is_empty() {
                state.punched.remove(&path);
            }
        }
    }

    fn seek_impl(&mut self, pos: SeekFrom) -> io::Result<u64> {
        let new_pos: u64 = match pos {
            SeekFrom::Start(n) => n,
            SeekFrom::End(n) => {
                let len = {
                    let data = lock(&self.data)?;
                    u64::try_from(data.len()).map_err(|_| {
                        io::Error::other("in-memory file length does not fit in u64")
                    })?
                };
                let result = i128::from(len) + i128::from(n);
                if result < 0 {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "seek to negative position",
                    ));
                }
                u64::try_from(result).map_err(|_| {
                    io::Error::new(io::ErrorKind::InvalidInput, "seek position overflow")
                })?
            }
            SeekFrom::Current(n) => {
                let result = i128::from(self.cursor) + i128::from(n);
                if result < 0 {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "seek to negative position",
                    ));
                }
                u64::try_from(result).map_err(|_| {
                    io::Error::new(io::ErrorKind::InvalidInput, "seek position overflow")
                })?
            }
        };

        self.cursor = new_pos;
        Ok(self.cursor)
    }
}

#[cfg(feature = "std")]
impl std::io::Read for MemFile {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.read_impl(buf).map_err(Into::into)
    }
}
#[cfg(not(feature = "std"))]
impl Read for MemFile {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.read_impl(buf)
    }
}

#[cfg(feature = "std")]
impl std::io::Write for MemFile {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.write_impl(buf).map_err(Into::into)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}
#[cfg(not(feature = "std"))]
impl Write for MemFile {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.write_impl(buf)
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

#[cfg(feature = "std")]
impl std::io::Seek for MemFile {
    fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
        self.seek_impl(pos.into()).map_err(Into::into)
    }
}
#[cfg(not(feature = "std"))]
impl Seek for MemFile {
    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
        self.seek_impl(pos)
    }
}

impl FsFile for MemFile {
    fn sync_all(&self) -> io::Result<()> {
        Ok(())
    }

    fn sync_data(&self) -> io::Result<()> {
        Ok(())
    }

    fn metadata(&self) -> io::Result<FsMetadata> {
        let data = lock(&self.data)?;
        Ok(FsMetadata {
            len: data.len() as u64,
            is_dir: false,
            is_file: true,
        })
    }

    fn set_len(&self, size: u64) -> io::Result<()> {
        if !self.writable {
            return Err(io::Error::other("set_len requires write access"));
        }
        let new_len = usize::try_from(size).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "set_len size exceeds usize::MAX",
            )
        })?;
        lock(&self.data)?.resize(new_len, 0);
        // A shrink permanently removes past-end reclaim; a grow keeps the
        // in-bounds ranges (they stay valid).
        self.clip_punched_to(size);
        Ok(())
    }

    fn read_at(&self, buf: &mut [u8], offset: u64) -> io::Result<usize> {
        if !self.readable {
            return Err(io::Error::other("read_at requires read access"));
        }
        let offset = usize::try_from(offset).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "read_at offset exceeds usize::MAX",
            )
        })?;
        let data = lock(&self.data)?;
        Ok(copy_from_data(buf, &data, offset))
    }

    /// No-op: in-memory files are not shared across processes. `MemFs` is a
    /// test/ephemeral backend - cross-process exclusivity is not meaningful.
    fn lock_exclusive(&self) -> io::Result<()> {
        Ok(())
    }

    fn try_lock_exclusive(&self) -> io::Result<bool> {
        // `MemFs` is a single-process in-memory backend: there is no other
        // process to contend with, so the directory lock is vacuously held.
        // Opt in explicitly (the trait default fails closed for backends that
        // have not implemented non-blocking locking).
        Ok(true)
    }
}

/// Rejects empty paths before they can create entries in the `/`-rooted namespace.
fn ensure_non_empty_path(path: &Path) -> io::Result<()> {
    if path.as_os_str().is_empty() {
        return Err(io::Error::new(io::ErrorKind::InvalidInput, "empty path"));
    }
    Ok(())
}

/// Validates that the parent directory of `path` exists and is a directory.
///
/// Returns `Ok(())` when the parent is root, empty, or an existing directory.
/// Returns `Err(Other)` when the parent is a file, or `Err(NotFound)` when
/// it does not exist at all.
fn ensure_parent_dir(path: &Path, state: &State) -> io::Result<()> {
    if let Some(parent) = path.parent()
        && !parent.as_os_str().is_empty()
        && parent != Path::new("/")
        && !state.dirs.contains(parent)
    {
        if state.files.contains_key(parent) {
            return Err(io::Error::other(format!(
                "parent is not a directory: {}",
                parent.display()
            )));
        }
        return Err(io::Error::new(
            io::ErrorKind::NotFound,
            format!("parent directory does not exist: {}", parent.display()),
        ));
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Fs for MemFs
// ---------------------------------------------------------------------------

impl Fs for MemFs {
    fn open(&self, path: &Path, opts: &FsOpenOptions) -> io::Result<Box<dyn FsFile>> {
        ensure_non_empty_path(path)?;
        let mut state = write_state(&self.state)?;
        let path = path.to_path_buf();
        let wants_write = opts.write || opts.append;

        // Validate flag combinations first (path-independent), before any
        // filesystem lookups. This ensures consistent InvalidInput errors
        // regardless of whether the parent directory exists.
        if !opts.read && !wants_write {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "open requires at least read, write, or append access",
            ));
        }
        if opts.truncate && opts.append {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "truncate and append cannot be used together",
            ));
        }
        if opts.truncate && !opts.write {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "truncate requires write access",
            ));
        }
        if (opts.create || opts.create_new) && !wants_write {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "create/create_new requires write or append access",
            ));
        }

        ensure_parent_dir(&path, &state)?;

        let exists = state.files.contains_key(&path);
        let is_dir = state.dirs.contains(&path);

        // Opening a directory path without create flags is an error (mirrors EISDIR).
        if is_dir && !opts.create && !opts.create_new {
            return Err(io::Error::other(format!(
                "path is a directory: {}",
                path.display()
            )));
        }

        // Reject creating a file at a path that is already a directory.
        if is_dir && (opts.create || opts.create_new) {
            return Err(io::Error::new(
                io::ErrorKind::AlreadyExists,
                format!("path is a directory: {}", path.display()),
            ));
        }

        if opts.create_new {
            if exists {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    format!("file already exists: {}", path.display()),
                ));
            }
            let data = Arc::new(Mutex::new(Vec::new()));
            state.files.insert(path.clone(), Arc::clone(&data));
            return Ok(Box::new(MemFile {
                data,
                cursor: 0,
                readable: opts.read,
                writable: opts.write || opts.append,
                is_append: opts.append,
                state: Arc::clone(&self.state),
                path,
                has_punches: Arc::clone(&self.has_punches),
            }));
        }

        if exists {
            let data = state
                .files
                .get(&path)
                .map(Arc::clone)
                .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "concurrent removal"))?;

            if opts.truncate {
                lock(&data)?.clear();
                // The bytes are gone; drop stale punched ranges so they cannot
                // resurrect and over-free once the file is rewritten and grows.
                state.punched.remove(&path);
            }

            // Cursor starts at 0 even in append mode - append only affects
            // where writes land (Write::write checks is_append), not the
            // read cursor. This matches std::fs::File behaviour.
            let cursor = 0;

            Ok(Box::new(MemFile {
                data,
                cursor,
                readable: opts.read,
                writable: opts.write || opts.append,
                is_append: opts.append,
                state: Arc::clone(&self.state),
                path,
                has_punches: Arc::clone(&self.has_punches),
            }))
        } else if opts.create {
            let data = Arc::new(Mutex::new(Vec::new()));
            state.files.insert(path.clone(), Arc::clone(&data));
            Ok(Box::new(MemFile {
                data,
                cursor: 0,
                readable: opts.read,
                writable: opts.write || opts.append,
                is_append: opts.append,
                state: Arc::clone(&self.state),
                path,
                has_punches: Arc::clone(&self.has_punches),
            }))
        } else {
            Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!("file not found: {}", path.display()),
            ))
        }
    }

    fn create_dir_all(&self, path: &Path) -> io::Result<()> {
        ensure_non_empty_path(path)?;
        let mut state = write_state(&self.state)?;

        // Collect all components first, then validate, then insert.
        // This avoids partial insertion if an ancestor is a regular file.
        let mut to_create = Vec::new();
        let mut current = path.to_path_buf();
        loop {
            if state.files.contains_key(&current) {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    format!("path conflicts with existing file: {}", current.display()),
                ));
            }
            to_create.push(current.clone());
            if !current.pop() || current.as_os_str().is_empty() {
                break;
            }
        }

        for dir in to_create {
            state.dirs.insert(dir);
        }
        Ok(())
    }

    fn create_dir(&self, path: &Path) -> io::Result<()> {
        ensure_non_empty_path(path)?;
        let mut state = write_state(&self.state)?;

        // Atomic single-leaf create: reject if anything (file OR dir)
        // already occupies the path. Mirrors POSIX `mkdir(2)` semantics.
        if state.dirs.contains(path) || state.files.contains_key(path) {
            return Err(io::Error::new(
                io::ErrorKind::AlreadyExists,
                format!("path already exists: {}", path.display()),
            ));
        }

        // Parent must exist AND be a directory. Delegating to
        // `ensure_parent_dir` gives the caller a `NotFound` vs
        // `parent-is-a-file` diagnostic (matching POSIX `ENOTDIR`),
        // instead of a single ambiguous `NotFound` for both cases.
        ensure_parent_dir(path, &state)?;

        state.dirs.insert(path.to_path_buf());
        Ok(())
    }

    fn read_dir(&self, path: &Path) -> io::Result<Vec<FsDirEntry>> {
        let state = read_state(&self.state)?;

        if !state.dirs.contains(path) {
            // Distinguish "path is a file" from "path does not exist".
            if state.files.contains_key(path) {
                return Err(io::Error::other(format!(
                    "not a directory: {}",
                    path.display()
                )));
            }
            return Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!("directory not found: {}", path.display()),
            ));
        }

        let mut entries = Vec::new();

        for file_path in state.files.keys() {
            if file_path.parent() == Some(path)
                && let Some(name) = file_path.file_name()
            {
                // Match StdFs contract: reject non-UTF-8 names with InvalidData.
                #[cfg(feature = "std")]
                let file_name = name.to_str().ok_or_else(|| {
                    io::Error::new(
                        io::ErrorKind::InvalidData,
                        format!(
                            "non-UTF-8 filename in directory {}: {}",
                            path.display(),
                            name.display()
                        ),
                    )
                })?;
                // no_std: keys are UTF-8 `&str` by construction.
                #[cfg(not(feature = "std"))]
                let file_name = name;
                entries.push(FsDirEntry {
                    path: file_path.clone(),
                    file_name: file_name.to_owned(),
                    is_dir: false,
                });
            }
        }

        for dir_path in &state.dirs {
            if dir_path.parent() == Some(path)
                && dir_path != path
                && let Some(name) = dir_path.file_name()
            {
                #[cfg(feature = "std")]
                let file_name = name.to_str().ok_or_else(|| {
                    io::Error::new(
                        io::ErrorKind::InvalidData,
                        format!(
                            "non-UTF-8 filename in directory {}: {}",
                            path.display(),
                            name.display()
                        ),
                    )
                })?;
                // no_std: keys are UTF-8 `&str` by construction.
                #[cfg(not(feature = "std"))]
                let file_name = name;
                entries.push(FsDirEntry {
                    path: dir_path.clone(),
                    file_name: file_name.to_owned(),
                    is_dir: true,
                });
            }
        }

        Ok(entries)
    }

    fn remove_file(&self, path: &Path) -> io::Result<()> {
        let mut state = write_state(&self.state)?;
        if state.dirs.contains(path) {
            return Err(io::Error::other(format!(
                "cannot remove_file on directory: {}",
                path.display()
            )));
        }
        if state.files.remove(path).is_none() {
            return Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!("file not found: {}", path.display()),
            ));
        }
        // Drop any punched-range accounting so a removed file stops freeing space.
        state.punched.remove(path);
        Ok(())
    }

    fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
        let mut state = write_state(&self.state)?;

        // Reject files - std::fs::remove_dir_all errors on non-directories.
        if state.files.contains_key(path) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("path is not a directory: {}", path.display()),
            ));
        }

        if !state.dirs.contains(path) {
            return Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!("path not found: {}", path.display()),
            ));
        }

        state.files.retain(|p, _| !p.starts_with(path));
        state.dirs.retain(|p| !p.starts_with(path));
        state.punched.retain(|p, _| !p.starts_with(path));

        // Re-seed root so exists("/") and read_dir("/") remain valid.
        state.dirs.insert(PathBuf::from("/"));
        Ok(())
    }

    fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
        ensure_non_empty_path(from)?;
        ensure_non_empty_path(to)?;
        let mut state = write_state(&self.state)?;

        ensure_parent_dir(to, &state)?;

        // Reject renaming onto an existing directory. Otherwise `to` would end
        // up present in both `files` and `dirs`, corrupting MemFs state.
        if state.dirs.contains(to) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("destination is a directory: {}", to.display()),
            ));
        }

        // Directory renames are not implemented in MemFs because they require
        // updating descendant paths in both `dirs` and `files`.
        if state.dirs.contains(from) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("path is a directory: {}", from.display()),
            ));
        }

        if let Some(data) = state.files.remove(from) {
            state.files.insert(to.to_path_buf(), data);
            // Move the punched-range accounting with the file; the destination is
            // replaced, so its old ranges (if any) are dropped first.
            match state.punched.remove(from) {
                Some(ranges) => {
                    state.punched.insert(to.to_path_buf(), ranges);
                }
                None => {
                    state.punched.remove(to);
                }
            }
            Ok(())
        } else {
            Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!("file not found: {}", from.display()),
            ))
        }
    }

    fn metadata(&self, path: &Path) -> io::Result<FsMetadata> {
        let state = read_state(&self.state)?;

        if let Some(data) = state.files.get(path) {
            let d = lock(data)?;
            Ok(FsMetadata {
                len: d.len() as u64,
                is_dir: false,
                is_file: true,
            })
        } else if state.dirs.contains(path) {
            Ok(FsMetadata {
                len: 0,
                is_dir: true,
                is_file: false,
            })
        } else {
            Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!("path not found: {}", path.display()),
            ))
        }
    }

    fn available_space(&self, _path: &Path) -> io::Result<u64> {
        let capacity = self.capacity.load(portable_atomic::Ordering::Relaxed);
        // Unbounded → no disk pressure. Otherwise the simulated free space is
        // capacity minus what is currently stored (saturating: an over-capacity
        // state reports zero free, never wraps).
        if capacity == u64::MAX {
            Ok(u64::MAX)
        } else {
            Ok(capacity.saturating_sub(self.stored_bytes()))
        }
    }

    fn sync_directory(&self, path: &Path) -> io::Result<()> {
        // Durability is a no-op, but validate the path is an existing directory.
        let state = read_state(&self.state)?;
        if !state.dirs.contains(path) {
            if state.files.contains_key(path) {
                return Err(io::Error::other(format!(
                    "sync_directory: not a directory: {}",
                    path.display()
                )));
            }
            return Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!("sync_directory: path not found: {}", path.display()),
            ));
        }
        Ok(())
    }

    fn exists(&self, path: &Path) -> io::Result<bool> {
        let state = read_state(&self.state)?;
        Ok(state.files.contains_key(path) || state.dirs.contains(path))
    }

    fn hard_link(&self, src: &Path, dst: &Path) -> io::Result<()> {
        ensure_non_empty_path(src)?;
        ensure_non_empty_path(dst)?;
        let mut state = write_state(&self.state)?;

        ensure_parent_dir(dst, &state)?;

        if state.dirs.contains(dst) {
            return Err(io::Error::new(
                io::ErrorKind::AlreadyExists,
                format!("destination is a directory: {}", dst.display()),
            ));
        }
        if state.files.contains_key(dst) {
            return Err(io::Error::new(
                io::ErrorKind::AlreadyExists,
                format!("destination already exists: {}", dst.display()),
            ));
        }

        // MemFs has no inode concept - produce an independent copy so the
        // destination has the same byte contents but its own backing buffer.
        // This matches the documented [`Fs::hard_link`] semantics for
        // in-memory backends.
        let bytes = {
            let src_data = state.files.get(src).ok_or_else(|| {
                io::Error::new(
                    io::ErrorKind::NotFound,
                    format!("source file not found: {}", src.display()),
                )
            })?;
            let guard = lock(src_data)?;
            guard.clone()
        };

        state
            .files
            .insert(dst.to_path_buf(), Arc::new(Mutex::new(bytes)));
        Ok(())
    }

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

    fn volume_id(&self, _path: &Path) -> Option<u64> {
        // One `MemFs` instance is one simulated disk with a single capacity /
        // free-space pool (shared across clones via the same `state` Arc), so the
        // per-instance namespace ID also identifies the volume. Independently
        // constructed instances are independent volumes.
        Some(self.namespace_id)
    }

    /// In-memory backend: no filesystem-level guarantees on any path.
    /// Explicitly returns the all-`false` default so the "no integrity / no
    /// `CoW` / no reflink" stance is intentional rather than inherited by
    /// accident. Only `punch_hole` is set: [`Self::punch_hole`] simulates
    /// in-place extent reclaim, so tight-space compaction (and its tests) can
    /// run against this backend.
    fn capabilities(&self, _path: &Path) -> FsCapabilities {
        FsCapabilities {
            punch_hole: self
                .punch_hole_supported
                .load(portable_atomic::Ordering::Relaxed),
            ..FsCapabilities::default()
        }
    }

    /// Simulates `fallocate(PUNCH_HOLE)`: zeroes `[offset, offset+len)` in the
    /// file (so the hole reads back as zeros) and records the reclaimed bytes so
    /// [`Fs::available_space`] reflects the freed space, while the file's logical
    /// length stays unchanged. The range is clamped to the current file length;
    /// a punch wholly past EOF is a no-op.
    fn punch_hole(&self, path: &Path, offset: u64, len: u64) -> io::Result<()> {
        // Write lock: the punched-range bookkeeping lives in `State`, and the
        // reclaim must be atomic with zeroing the bytes.
        let mut state = self.state.write();
        let (start, end) = {
            let data = state.files.get(path).ok_or_else(|| {
                io::Error::new(io::ErrorKind::NotFound, "punch_hole: file not found")
            })?;
            let mut buf = data.lock();
            let file_len = buf.len() as u64;
            // Clamp to the file: a hole cannot extend the logical length.
            let start = offset.min(file_len);
            // offset + len can exceed u64 for an adversarial caller; the result is
            // immediately clamped to the file length, so the saturating add only
            // avoids a wraparound that would defeat that clamp.
            let end = offset.saturating_add(len).min(file_len);
            if start >= end {
                return Ok(());
            }
            #[expect(
                clippy::cast_possible_truncation,
                reason = "start/end are clamped to buf.len() (a usize), so they fit usize"
            )]
            let (s, e) = (start as usize, end as usize);
            if let Some(slice) = buf.get_mut(s..e) {
                slice.fill(0);
            }
            (start, end)
        };
        // Update the per-file ranges (drives free space) and, by the union delta,
        // the lifetime counter (drives the `punched_bytes` test metric).
        let ranges = state.punched.entry(path.to_path_buf()).or_default();
        let before: u64 = ranges.iter().map(|&(s, e)| e - s).sum();
        merge_punched_range(ranges, start, end);
        let after: u64 = ranges.iter().map(|&(s, e)| e - s).sum();
        self.punched_total
            .fetch_add(after - before, portable_atomic::Ordering::Relaxed);
        // Arm the write/set_len invalidation fast-path now that a punch exists.
        self.has_punches
            .store(true, portable_atomic::Ordering::Relaxed);
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Lock helpers - convert PoisonError to io::Error
// ---------------------------------------------------------------------------

// `spin` locks cannot be poisoned (no unwind-during-hold concept), so these
// always succeed; the `io::Result` return is kept so the `?`-using call sites
// stay unchanged.
// Kept returning `io::Result` (always `Ok`) so the `?`-using call sites are
// untouched — spin locks never poison, but a future fallible lock layer would
// slot in here without churning every caller.
#[expect(
    clippy::unnecessary_wraps,
    reason = "Result kept for ?-compatible call sites and future fallible-lock parity"
)]
fn lock<T>(m: &Mutex<T>) -> io::Result<impl core::ops::DerefMut<Target = T> + '_> {
    Ok(m.lock())
}

#[expect(
    clippy::unnecessary_wraps,
    reason = "Result kept for ?-compatible call sites and future fallible-lock parity"
)]
fn read_state(rw: &RwLock<State>) -> io::Result<impl core::ops::Deref<Target = State> + '_> {
    Ok(rw.read())
}

#[expect(
    clippy::unnecessary_wraps,
    reason = "Result kept for ?-compatible call sites and future fallible-lock parity"
)]
fn write_state(rw: &RwLock<State>) -> io::Result<impl core::ops::DerefMut<Target = State> + '_> {
    Ok(rw.write())
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

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