hawdb-storage 0.14.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! File operations admitted before acquiring native descriptors.
//!
//! Native handles are private: cloning, directory iteration, convenience reads,
//! and temporary operations must retain their permit for the full handle lifetime.

use crate::file_descriptors::{
    context_for_path, DescriptorKind, DescriptorPermit, FileOpenContext,
};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;

pub use std::fs::{FileType, Metadata, Permissions, TryLockError};

mod source_directory;
#[cfg(any(test, feature = "test-support"))]
pub(crate) use source_directory::PowerLossRoot;
#[doc(hidden)]
pub use source_directory::{
    CapturedControlCandidate, CapturedDirectoryCandidate, PendingControlPublication,
    SourceDirectory, SourceDirectoryIdentity, SourceDirectoryLockMode,
};

pub(crate) mod captured_immutable;

mod read;
mod wal_rollback;
#[doc(hidden)]
pub use read::{AdmittedFileRead, ReadAdmissionControl};

/// A mutable native handle or an immutable logical reference. Logical files
/// retain their object identity and cursor without retaining a native handle.
#[derive(Debug)]
pub struct File {
    backing: FileBacking,
}

/// A publication scope owns lock release, independently of descriptor copies
/// retained by a concurrent fork before close-on-exec. Construct only after
/// successful acquisition; failed acquisition must never unlock another owner.
#[doc(hidden)]
#[derive(Debug)]
pub struct PublicationLockGuard(File);

impl PublicationLockGuard {
    pub(crate) fn new(locked: File) -> Self {
        Self(locked)
    }
}

#[cfg(test)]
impl std::borrow::Borrow<File> for PublicationLockGuard {
    fn borrow(&self) -> &File {
        &self.0
    }
}

impl Drop for PublicationLockGuard {
    fn drop(&mut self) {
        // Release before closing the original descriptor. Closing remains the
        // fallback on unlock failure; publication data/durability is unchanged.
        let _ = self.0.unlock();
    }
}

#[derive(Debug)]
enum FileBacking {
    Native(NativeFile),
    Immutable(Box<ImmutableFile>),
    Captured {
        // A pinned native lease closes before its logical custody can retire.
        pinned: Option<Arc<File>>,
        file: hawdb_core::SharedAllocation<ImmutableFile>,
    },
}

#[derive(Debug)]
struct NativeFile {
    // Declaration order closes the native handle before returning its capacity.
    inner: std::fs::File,
    permit: DescriptorPermit,
    #[cfg(any(test, feature = "test-support"))]
    trace: Option<Box<crate::power_loss::NativeTrace>>,
}

impl NativeFile {
    fn io<T>(&self, operation: impl FnOnce(&std::fs::File) -> io::Result<T>) -> io::Result<T> {
        #[cfg(any(test, feature = "test-support"))]
        if let Some(trace) = &self.trace {
            return trace.io(|| operation(&self.inner));
        }
        operation(&self.inner)
    }
}

#[derive(Debug, Clone)]
struct ImmutableFile {
    binding: Option<crate::immutable_files::ImmutableFileBinding>,
    handles: Arc<crate::immutable_files::ImmutableFileHandles>,
    context: FileOpenContext,
    // Explicit construction scopes can pin an admitted native lease. Ordinary
    // logical files remain lazy and evictable until a read acquires a lease.
    pinned: Option<Arc<File>>,
    // Match File::try_clone: clones share the logical sequential cursor. All
    // positioned reads bypass it and retain a cache lease for the entire read.
    cursor: Arc<std::sync::Mutex<u64>>,
    // Last: all logical-file metadata dies before custody/memory receipts.
    source: Option<hawdb_core::SharedAllocation<captured_immutable::CapturedSource>>,
}

impl File {
    fn immutable(&self) -> Option<(&ImmutableFile, Option<&Arc<File>>)> {
        match &self.backing {
            FileBacking::Native(_) => None,
            FileBacking::Immutable(file) => Some((file, file.pinned.as_ref())),
            FileBacking::Captured { file, pinned } => Some((file, pinned.as_ref())),
        }
    }

    pub fn options() -> OpenOptions {
        OpenOptions::new()
    }

    pub fn open(path: impl AsRef<Path>) -> io::Result<Self> {
        OpenOptions::new().read(true).open(path)
    }

    pub fn create(path: impl AsRef<Path>) -> io::Result<Self> {
        OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(path)
    }

    /// Creates an anonymous, counted scratch file in the requested filesystem.
    ///
    /// No directory entry exists, including after an error, process exit, or
    /// cancellation. Filesystems/platforms without anonymous-file support return
    /// an explicit error; there is no named or uncounted fallback.
    #[doc(hidden)]
    pub fn anonymous(directory: &Path) -> io::Result<Self> {
        #[cfg(target_os = "linux")]
        {
            use std::os::unix::fs::OpenOptionsExt;
            OpenOptions::new()
                .read(true)
                .write(true)
                .custom_flags(libc::O_TMPFILE | libc::O_EXCL)
                .mode(0o600)
                .open(directory)
        }
        #[cfg(not(target_os = "linux"))]
        {
            let _ = directory;
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "anonymous scratch files are not supported on this platform",
            ))
        }
    }

    pub fn try_clone(&self) -> io::Result<Self> {
        let backing = match &self.backing {
            FileBacking::Immutable(file) => FileBacking::Immutable(file.clone()),
            FileBacking::Captured { file, pinned } => FileBacking::Captured {
                file: file.clone(),
                pinned: pinned.clone(),
            },
            FileBacking::Native(file) => {
                let permit = file.permit.context.acquire(file.permit.kind())?;
                let inner = file
                    .inner
                    .try_clone()
                    .map_err(|error| file.permit.context.map_open_error(error))?;
                FileBacking::Native(NativeFile {
                    inner,
                    permit,
                    #[cfg(any(test, feature = "test-support"))]
                    trace: file.trace.clone(),
                })
            }
        };
        Ok(Self { backing })
    }

    pub(crate) fn with_native<T>(
        &self,
        operation: impl FnOnce(&std::fs::File) -> io::Result<T>,
    ) -> io::Result<T> {
        match &self.backing {
            FileBacking::Native(file) => file.io(operation),
            _ => {
                let (file, pinned) = self.immutable().expect("logical immutable file");
                if let Some(lease) = pinned {
                    lease.with_native(operation)
                } else {
                    let lease = file.get()?;
                    lease.with_native(operation)
                }
            }
        }
    }

    /// Retain one native lease across admitted cold validation and the read.
    pub(crate) fn with_native_admitted<T>(
        &self,
        admit_validation: impl FnMut(u64) -> io::Result<()>,
        operation: impl FnOnce(&std::fs::File) -> io::Result<T>,
    ) -> io::Result<(T, u64)> {
        match &self.backing {
            FileBacking::Native(file) => file.io(operation).map(|value| (value, 0)),
            _ => {
                let (file, pinned) = self.immutable().expect("logical immutable file");
                if let Some(lease) = pinned {
                    lease.with_native(operation).map(|value| (value, 0))
                } else {
                    let (lease, validation_bytes) = file.get_admitted(admit_validation)?;
                    lease
                        .with_native(operation)
                        .map(|value| (value, validation_bytes))
                }
            }
        }
    }

    fn writable_native(&self) -> io::Result<&std::fs::File> {
        match &self.backing {
            FileBacking::Native(file) => Ok(&file.inner),
            _ => Err(io::Error::new(
                io::ErrorKind::PermissionDenied,
                "immutable logical file cannot be modified",
            )),
        }
    }

    /// Map an immutable artifact through its admitted native handle.
    /// The mapping retains the OS mapping, not the file descriptor permit.
    ///
    /// # Safety
    /// The underlying file must not be mutated or truncated while the mapping
    /// exists, including by another process.
    #[cfg(feature = "artifact-mmap")]
    pub unsafe fn map_read_only(&self) -> io::Result<memmap2::Mmap> {
        // SAFETY: the caller guarantees the immutable artifact lifetime. The
        // native handle and its permit stay together throughout map creation.
        self.with_native(|file| unsafe { memmap2::Mmap::map(file) })
    }

    pub fn metadata(&self) -> io::Result<Metadata> {
        self.with_native(std::fs::File::metadata)
    }
    /// Compares this held file with the native file currently named by `path`.
    /// Follows symlinks; it does not reserve that path or prove logical identity.
    /// For native Unix handles, path lookup uses metadata without opening and closing another descriptor
    /// for the same inode, which can release process-owned POSIX record locks.
    /// Windows uses a counted validation handle and the full 128-bit file ID.
    pub fn matches_native_path(&self, path: impl AsRef<Path>) -> io::Result<bool> {
        let identity = self.with_native(crate::native_file_identity::of_file)?;
        let path = path.as_ref();
        #[cfg(unix)]
        {
            Ok(identity == crate::native_file_identity::of_metadata(&metadata(path)?))
        }
        #[cfg(windows)]
        {
            let named = Self::open(path)?;
            Ok(identity == named.with_native(crate::native_file_identity::of_file)?)
        }
        #[cfg(not(any(unix, windows)))]
        {
            let _ = (identity, path);
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "native path identity requires Unix or Windows",
            ))
        }
    }

    pub fn sync_all(&self) -> io::Result<()> {
        self.synchronize(false)
    }
    pub fn sync_data(&self) -> io::Result<()> {
        self.synchronize(true)
    }
    pub fn set_len(&self, size: u64) -> io::Result<()> {
        #[cfg(any(test, feature = "test-support"))]
        if let FileBacking::Native(file) = &self.backing
            && let Some(trace) = &file.trace
        {
            return trace.truncate(size, || file.inner.set_len(size));
        }
        self.writable_native()?.set_len(size)
    }
    pub fn set_permissions(&self, permissions: Permissions) -> io::Result<()> {
        self.writable_native()?.set_permissions(permissions)
    }
    pub fn lock(&self) -> io::Result<()> {
        self.writable_native()?.lock()
    }
    pub fn try_lock(&self) -> Result<(), TryLockError> {
        self.writable_native()
            .map_err(TryLockError::Error)?
            .try_lock()
    }
    /// Own successful acquisition and explicit unlock independently of native
    /// descriptions inherited by a fork before close-on-exec. A failed acquire
    /// closes the input handle without constructing an unlock owner.
    #[doc(hidden)]
    pub fn try_lock_owned(self) -> Result<PublicationLockGuard, TryLockError> {
        self.try_lock()?;
        Ok(PublicationLockGuard::new(self))
    }

    pub fn unlock(&self) -> io::Result<()> {
        self.writable_native()?.unlock()
    }

    fn read_sequential(&self, buffer: &mut [u8]) -> io::Result<usize> {
        match &self.backing {
            FileBacking::Native(file) => file.io(|mut inner| inner.read(buffer)),
            _ => {
                let (file, pinned) = self.immutable().expect("logical immutable file");
                let mut cursor = file
                    .cursor
                    .lock()
                    .unwrap_or_else(|error| error.into_inner());
                let available = file.byte_length().saturating_sub(*cursor);
                let length = buffer
                    .len()
                    .min(usize::try_from(available).unwrap_or(usize::MAX));
                if length == 0 {
                    return Ok(0);
                }
                let read = if pinned.is_some() {
                    // A controlled constructor must observe short reads and
                    // EINTR between its own cancellation checkpoints.
                    self.with_native(|native| {
                        crate::io::read_native_at_once(native, &mut buffer[..length], *cursor)
                    })?
                } else {
                    crate::io::read_exact_at(self, &mut buffer[..length], *cursor)?;
                    length
                };
                *cursor += read as u64;
                Ok(read)
            }
        }
    }

    fn read_sequential_vectored(&self, buffers: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
        match &self.backing {
            FileBacking::Native(file) => file.io(|mut inner| inner.read_vectored(buffers)),
            _ => match buffers.iter_mut().find(|buffer| !buffer.is_empty()) {
                Some(buffer) => self.read_sequential(buffer),
                None => Ok(0),
            },
        }
    }

    fn seek_sequential(&self, position: SeekFrom) -> io::Result<u64> {
        match &self.backing {
            FileBacking::Native(file) => file.io(|mut inner| inner.seek(position)),
            _ => {
                let (file, _) = self.immutable().expect("logical immutable file");
                let mut cursor = file
                    .cursor
                    .lock()
                    .unwrap_or_else(|error| error.into_inner());
                let next = match position {
                    SeekFrom::Start(offset) => i128::from(offset),
                    SeekFrom::Current(offset) => i128::from(*cursor) + i128::from(offset),
                    SeekFrom::End(offset) => i128::from(file.byte_length()) + i128::from(offset),
                };
                *cursor = u64::try_from(next).map_err(|_| {
                    io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "immutable file seek is outside the representable range",
                    )
                })?;
                Ok(*cursor)
            }
        }
    }

    fn synchronize(&self, data_only: bool) -> io::Result<()> {
        match &self.backing {
            FileBacking::Native(file) => {
                let sync = || {
                    if data_only {
                        file.inner.sync_data()
                    } else {
                        file.inner.sync_all()
                    }
                };
                #[cfg(any(test, feature = "test-support"))]
                if let Some(trace) = &file.trace {
                    return trace.sync(sync);
                }
                sync()
            }
            _ => {
                let (file, pinned) = self.immutable().expect("logical immutable file");
                match pinned {
                    Some(lease) => lease.synchronize(data_only),
                    None => file.get()?.synchronize(data_only),
                }
            }
        }
    }

    fn write_sequential(&self, buffer: &[u8]) -> io::Result<usize> {
        #[cfg(any(test, feature = "test-support"))]
        if let FileBacking::Native(file) = &self.backing
            && let Some(trace) = &file.trace
        {
            return trace.write(&file.inner, buffer);
        }
        self.writable_native()?.write(buffer)
    }

    fn write_sequential_vectored(&self, buffers: &[io::IoSlice<'_>]) -> io::Result<usize> {
        #[cfg(any(test, feature = "test-support"))]
        if let FileBacking::Native(file) = &self.backing
            && let Some(trace) = &file.trace
        {
            return trace.write_vectored(&file.inner, buffers);
        }
        self.writable_native()?.write_vectored(buffers)
    }
}

impl Read for File {
    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
        self.read_sequential(buffer)
    }
    fn read_vectored(&mut self, buffers: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
        self.read_sequential_vectored(buffers)
    }
}
impl Read for &File {
    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
        self.read_sequential(buffer)
    }
    fn read_vectored(&mut self, buffers: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
        self.read_sequential_vectored(buffers)
    }
}
impl Write for File {
    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
        self.write_sequential(buffer)
    }
    fn write_vectored(&mut self, buffers: &[io::IoSlice<'_>]) -> io::Result<usize> {
        self.write_sequential_vectored(buffers)
    }
    fn flush(&mut self) -> io::Result<()> {
        self.writable_native()?.flush()
    }
}
impl Write for &File {
    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
        self.write_sequential(buffer)
    }
    fn write_vectored(&mut self, buffers: &[io::IoSlice<'_>]) -> io::Result<usize> {
        self.write_sequential_vectored(buffers)
    }
    fn flush(&mut self) -> io::Result<()> {
        self.writable_native()?.flush()
    }
}
impl Seek for File {
    fn seek(&mut self, position: SeekFrom) -> io::Result<u64> {
        self.seek_sequential(position)
    }
    fn stream_position(&mut self) -> io::Result<u64> {
        self.seek_sequential(SeekFrom::Current(0))
    }
}
impl Seek for &File {
    fn seek(&mut self, position: SeekFrom) -> io::Result<u64> {
        self.seek_sequential(position)
    }
    fn stream_position(&mut self) -> io::Result<u64> {
        self.seek_sequential(SeekFrom::Current(0))
    }
}

#[derive(Debug, Clone)]
pub struct OpenOptions {
    inner: std::fs::OpenOptions,
    kind: DescriptorKind,
    read: bool,
    mutable: [bool; 5],
    native_options: bool,
    #[cfg(any(test, feature = "test-support"))]
    capture_opaque_options: bool,
}

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

impl OpenOptions {
    pub fn new() -> Self {
        Self {
            inner: std::fs::OpenOptions::new(),
            kind: DescriptorKind::Transient,
            read: false,
            mutable: [false; 5],
            native_options: false,
            #[cfg(any(test, feature = "test-support"))]
            capture_opaque_options: false,
        }
    }
    pub fn read(&mut self, value: bool) -> &mut Self {
        self.read = value;
        self.inner.read(value);
        self
    }
    pub fn write(&mut self, value: bool) -> &mut Self {
        self.mutable[0] = value;
        self.inner.write(value);
        self
    }
    pub fn append(&mut self, value: bool) -> &mut Self {
        self.mutable[1] = value;
        self.inner.append(value);
        self
    }
    pub fn truncate(&mut self, value: bool) -> &mut Self {
        self.mutable[2] = value;
        self.inner.truncate(value);
        self
    }
    pub fn create(&mut self, value: bool) -> &mut Self {
        self.mutable[3] = value;
        self.inner.create(value);
        self
    }
    pub fn create_new(&mut self, value: bool) -> &mut Self {
        self.mutable[4] = value;
        self.inner.create_new(value);
        self
    }
    pub fn descriptor_kind(&mut self, kind: DescriptorKind) -> &mut Self {
        self.kind = kind;
        self
    }

    pub fn open(&self, path: impl AsRef<Path>) -> io::Result<File> {
        let path = path.as_ref();
        self.open_with_context(path, &context_for_path(path)?)
    }

    pub(crate) fn open_with_context(
        &self,
        path: &Path,
        context: &FileOpenContext,
    ) -> io::Result<File> {
        // Unbound native files do not need an immutable cache allocation.
        let handles = context.state.existing_immutable_handles();
        let mutable = self.mutable.iter().any(|flag| *flag);
        if self.read
            && !mutable
            && !self.native_options
            && self.kind == DescriptorKind::Transient
            && let Some(handles) = handles.as_ref()
            && let Some(binding) = handles.binding(path)?
        {
            return Ok(File {
                backing: FileBacking::Immutable(Box::new(ImmutableFile {
                    binding: Some(binding),
                    handles: handles.clone(),
                    context: context.clone(),
                    pinned: None,
                    cursor: Arc::new(std::sync::Mutex::new(0)),
                    source: None,
                })),
            });
        }
        if (mutable || self.native_options)
            && !self.mutable[4]
            && let Some(handles) = handles.as_ref()
            && handles.binding(path)?.is_some()
        {
            // Native options can override access mode or request truncation.
            // Give opaque native opens a private inode too. create_new must
            // still reject an existing alias without changing its binding.
            handles.detach_for_write(path, context, self.mutable[2])?;
        }
        // COW charges each of its native handles before opening it, then
        // closes the candidate before acquiring this final handle's capacity.
        let permit = context.acquire(self.kind)?;
        // Later readers must see the mutable path. Existing logical readers
        // retain their captured immutable identity and remain snapshot-safe.
        // Invalidate before the native open can truncate or modify the path.
        if (mutable || self.native_options)
            && !self.mutable[4]
            && let Some(handles) = handles.as_ref()
        {
            handles.unbind(path)?;
        }
        #[cfg(any(test, feature = "test-support"))]
        let (inner, trace) = match crate::power_loss::for_path(context, path)? {
            Some(core) => crate::power_loss::NativeTrace::open(
                core,
                path,
                &self.inner,
                self.mutable,
                self.capture_opaque_options,
            )
            .map(|(inner, trace)| (inner, Some(Box::new(trace))))
            .map_err(|error| context.map_open_error(error))?,
            None => (
                self.inner
                    .open(path)
                    .map_err(|error| context.map_open_error(error))?,
                None,
            ),
        };
        #[cfg(not(any(test, feature = "test-support")))]
        let inner = self
            .inner
            .open(path)
            .map_err(|error| context.map_open_error(error))?;
        if self.mutable[4]
            && let Some(handles) = handles
        {
            // A rejected create_new must preserve the existing alias. Clear
            // a stale binding only after native creation actually succeeds.
            handles.unbind(path)?;
        }
        Ok(File {
            backing: FileBacking::Native(NativeFile {
                inner,
                permit,
                #[cfg(any(test, feature = "test-support"))]
                trace,
            }),
        })
    }
}

#[cfg(unix)]
impl std::os::unix::fs::OpenOptionsExt for OpenOptions {
    fn mode(&mut self, mode: u32) -> &mut Self {
        self.inner.mode(mode);
        self
    }
    fn custom_flags(&mut self, flags: i32) -> &mut Self {
        self.native_options = true;
        #[cfg(any(test, feature = "test-support"))]
        {
            self.capture_opaque_options = true;
        }
        self.inner.custom_flags(flags);
        self
    }
}

#[cfg(windows)]
impl std::os::windows::fs::OpenOptionsExt for OpenOptions {
    fn access_mode(&mut self, mode: u32) -> &mut Self {
        self.native_options = true;
        #[cfg(any(test, feature = "test-support"))]
        {
            self.capture_opaque_options = true;
        }
        self.inner.access_mode(mode);
        self
    }
    fn share_mode(&mut self, mode: u32) -> &mut Self {
        self.native_options = true;
        #[cfg(any(test, feature = "test-support"))]
        {
            self.capture_opaque_options = true;
        }
        self.inner.share_mode(mode);
        self
    }
    fn custom_flags(&mut self, flags: u32) -> &mut Self {
        self.native_options = true;
        #[cfg(any(test, feature = "test-support"))]
        {
            // BACKUP_SEMANTICS permits the production directory-barrier open;
            // it does not alter the captured byte/cursor/namespace operations.
            // Any opaque option remains opaque even if later flags are known.
            const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000;
            self.capture_opaque_options |= flags & !FILE_FLAG_BACKUP_SEMANTICS != 0;
        }
        self.inner.custom_flags(flags);
        self
    }
    fn attributes(&mut self, attributes: u32) -> &mut Self {
        self.native_options = true;
        #[cfg(any(test, feature = "test-support"))]
        {
            self.capture_opaque_options = true;
        }
        self.inner.attributes(attributes);
        self
    }
    fn security_qos_flags(&mut self, flags: u32) -> &mut Self {
        self.native_options = true;
        #[cfg(any(test, feature = "test-support"))]
        {
            self.capture_opaque_options = true;
        }
        self.inner.security_qos_flags(flags);
        self
    }
}

pub fn read(path: impl AsRef<Path>) -> io::Result<Vec<u8>> {
    let mut file = File::open(path)?;
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes)?;
    Ok(bytes)
}

pub fn read_to_string(path: impl AsRef<Path>) -> io::Result<String> {
    let mut file = File::open(path)?;
    let mut text = String::new();
    file.read_to_string(&mut text)?;
    Ok(text)
}

pub fn write(path: impl AsRef<Path>, contents: impl AsRef<[u8]>) -> io::Result<()> {
    File::create(path)?.write_all(contents.as_ref())
}

pub fn copy(source: impl AsRef<Path>, destination: impl AsRef<Path>) -> io::Result<u64> {
    let mut source = File::open(source)?;
    let metadata = source.metadata()?;
    if !metadata.is_file() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "copy source is not a regular file",
        ));
    }
    let mut destination = File::create(destination)?;
    let copied = io::copy(&mut source, &mut destination)?;
    destination.set_permissions(metadata.permissions())?;
    Ok(copied)
}

fn temporary<T>(path: &Path, operation: impl FnOnce() -> io::Result<T>) -> io::Result<T> {
    let context = context_for_path(path)?;
    let _permit = context.acquire(DescriptorKind::Transient)?;
    operation().map_err(|error| context.map_open_error(error))
}

fn two_paths<T>(
    source: &Path,
    destination: &Path,
    operation: impl FnOnce() -> io::Result<T>,
) -> io::Result<T> {
    let source_context = context_for_path(source)?;
    let destination_context = context_for_path(destination)?;
    let _source_permit = source_context.acquire(DescriptorKind::Transient)?;
    if Arc::ptr_eq(&source_context.state, &destination_context.state) {
        // One filesystem operation in one project is one temporary admission.
        operation().map_err(|error| source_context.map_open_error(error))
    } else {
        let _destination_permit = destination_context.acquire(DescriptorKind::Transient)?;
        operation().map_err(|error| destination_context.map_open_error(error))
    }
}

pub fn metadata(path: impl AsRef<Path>) -> io::Result<Metadata> {
    temporary(path.as_ref(), || std::fs::metadata(path.as_ref()))
}
pub fn try_exists(path: impl AsRef<Path>) -> io::Result<bool> {
    match metadata(path) {
        Ok(_) => Ok(true),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
        Err(error) => Err(error),
    }
}
pub fn symlink_metadata(path: impl AsRef<Path>) -> io::Result<Metadata> {
    temporary(path.as_ref(), || std::fs::symlink_metadata(path.as_ref()))
}
pub fn canonicalize(path: impl AsRef<Path>) -> io::Result<PathBuf> {
    temporary(path.as_ref(), || std::fs::canonicalize(path.as_ref()))
}
#[derive(Clone, Copy)]
enum NamespaceOperation {
    CreateDirectory,
    CreateDirectories,
    Remove,
    Rename,
    RenameNew,
    HardLink,
}

/// Dispatch native namespace mutation and its fault-model observation together.
/// Descriptor admission remains with the caller and precedes both paths.
fn namespace_io<T>(
    operation: NamespaceOperation,
    path: &Path,
    destination: Option<&Path>,
    native: impl FnOnce() -> io::Result<T>,
) -> io::Result<T> {
    #[cfg(any(test, feature = "test-support"))]
    {
        let context = context_for_path(path)?;
        match operation {
            NamespaceOperation::Rename
            | NamespaceOperation::RenameNew
            | NamespaceOperation::HardLink => {
                let destination = destination.expect("two-path namespace operation has a target");
                let event = match operation {
                    NamespaceOperation::Rename | NamespaceOperation::RenameNew => {
                        crate::power_loss::IoEvent::Rename
                    }
                    _ => crate::power_loss::IoEvent::HardLink,
                };
                crate::power_loss::two_paths(
                    &context,
                    &context_for_path(destination)?,
                    path,
                    destination,
                    Some(event),
                    |engine, source, destination| match operation {
                        NamespaceOperation::Rename => engine.rename(source, destination),
                        NamespaceOperation::RenameNew => {
                            engine.rename_directory_new(source, destination)
                        }
                        _ => engine.hard_link(source, destination),
                    },
                    native,
                )
            }
            _ => crate::power_loss::namespace(
                &context,
                path,
                matches!(operation, NamespaceOperation::Remove)
                    .then_some(crate::power_loss::IoEvent::Remove),
                |engine, path| match operation {
                    NamespaceOperation::CreateDirectory => {
                        engine.create_directory(path).map(|_| ())
                    }
                    NamespaceOperation::CreateDirectories => {
                        crate::power_loss::create_directories(engine, path)
                    }
                    NamespaceOperation::Remove => engine.remove(path),
                    _ => unreachable!("two-path operations are dispatched separately"),
                },
                native,
            ),
        }
    }
    #[cfg(not(any(test, feature = "test-support")))]
    {
        let _ = (operation, path, destination);
        native()
    }
}

pub fn create_dir(path: impl AsRef<Path>) -> io::Result<()> {
    let path = path.as_ref();
    temporary(path, || {
        namespace_io(NamespaceOperation::CreateDirectory, path, None, || {
            std::fs::create_dir(path)
        })
    })
}
pub fn create_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
    let path = path.as_ref();
    temporary(path, || {
        namespace_io(NamespaceOperation::CreateDirectories, path, None, || {
            std::fs::create_dir_all(path)
        })
    })
}
pub fn remove_file(path: impl AsRef<Path>) -> io::Result<()> {
    let path = path.as_ref();
    temporary(path, || {
        namespace_io(NamespaceOperation::Remove, path, None, || {
            std::fs::remove_file(path)
        })
    })?;
    unbind_immutable_path(path)
}

pub(crate) fn unbind_immutable_path(path: &Path) -> io::Result<()> {
    if let Some(handles) = context_for_path(path)?.state.existing_immutable_handles() {
        handles.unbind(path)?;
    }
    Ok(())
}
pub fn remove_dir(path: impl AsRef<Path>) -> io::Result<()> {
    let path = path.as_ref();
    temporary(path, || {
        namespace_io(NamespaceOperation::Remove, path, None, || {
            std::fs::remove_dir(path)
        })
    })
}
pub fn rename(source: impl AsRef<Path>, destination: impl AsRef<Path>) -> io::Result<()> {
    two_paths(source.as_ref(), destination.as_ref(), || {
        namespace_io(
            NamespaceOperation::Rename,
            source.as_ref(),
            Some(destination.as_ref()),
            || std::fs::rename(source.as_ref(), destination.as_ref()),
        )
    })?;
    // Captured logical readers retain their immutable identity, while future
    // opens must observe the replacement or the removed source name.
    for path in [source.as_ref(), destination.as_ref()] {
        unbind_immutable_path(path)?;
    }
    Ok(())
}
/// Atomically move a directory only if the destination name is absent.
/// No existence-check/remove/rename fallback is used. Unsupported kernels or
/// filesystems refuse. Success still needs both parent-directory barriers.
/// An error may follow a completed move; callers must retain custody/evidence.
#[doc(hidden)]
pub fn rename_directory_new(
    source: impl AsRef<Path>,
    destination: impl AsRef<Path>,
) -> io::Result<()> {
    let source = source.as_ref();
    let destination = destination.as_ref();
    two_paths(source, destination, || {
        namespace_io(
            NamespaceOperation::RenameNew,
            source,
            Some(destination),
            || {
                if !std::fs::symlink_metadata(source)?.is_dir() {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "directory move requires a direct directory",
                    ));
                }
                // Darwin can report a successful no-op when both names denote
                // the same entry, even with RENAME_EXCL. Reject existing names
                // early; the exclusive syscall still closes the ensuing race.
                match std::fs::symlink_metadata(destination) {
                    Ok(_) => {
                        return Err(io::Error::new(
                            io::ErrorKind::AlreadyExists,
                            "directory move target already exists",
                        ))
                    }
                    Err(error) if error.kind() == io::ErrorKind::NotFound => {}
                    Err(error) => return Err(error),
                }
                #[cfg(test)]
                rename_new::after_destination_check();
                rename_new::native(source, destination)
            },
        )
    })?;
    for path in [source, destination] {
        unbind_immutable_path(path)?;
    }
    Ok(())
}

mod rename_new;

pub fn hard_link(source: impl AsRef<Path>, destination: impl AsRef<Path>) -> io::Result<()> {
    two_paths(source.as_ref(), destination.as_ref(), || {
        namespace_io(
            NamespaceOperation::HardLink,
            source.as_ref(),
            Some(destination.as_ref()),
            || std::fs::hard_link(source.as_ref(), destination.as_ref()),
        )
    })
}

#[derive(Debug)]
pub struct ReadDir {
    inner: std::fs::ReadDir,
    permit: Arc<DescriptorPermit>,
}

#[derive(Debug)]
pub struct DirEntry {
    inner: std::fs::DirEntry,
    // Unix entries can retain the directory FD after ReadDir itself is dropped.
    permit: Arc<DescriptorPermit>,
}

pub fn read_dir(path: impl AsRef<Path>) -> io::Result<ReadDir> {
    let path = path.as_ref();
    let context = context_for_path(path)?;
    read_dir_with_context(path, &context)
}

pub(crate) fn read_dir_with_context(path: &Path, context: &FileOpenContext) -> io::Result<ReadDir> {
    let permit = Arc::new(context.acquire(DescriptorKind::Transient)?);
    let inner = std::fs::read_dir(path).map_err(|error| context.map_open_error(error))?;
    Ok(ReadDir { inner, permit })
}

impl Iterator for ReadDir {
    type Item = io::Result<DirEntry>;
    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|entry| {
            entry.map(|inner| DirEntry {
                inner,
                permit: self.permit.clone(),
            })
        })
    }
}

impl DirEntry {
    pub fn path(&self) -> PathBuf {
        self.inner.path()
    }
    pub fn file_name(&self) -> std::ffi::OsString {
        self.inner.file_name()
    }
    pub fn file_type(&self) -> io::Result<FileType> {
        self.inner.file_type()
    }
    pub fn metadata(&self) -> io::Result<Metadata> {
        let context = &self.permit.context;
        let _permit = context.acquire(DescriptorKind::Transient)?;
        self.inner
            .metadata()
            .map_err(|error| context.map_open_error(error))
    }
}

/// Maximum children retained while removing one admitted directory batch.
#[doc(hidden)]
pub const DIRECTORY_REMOVAL_BATCH_ENTRIES: usize = 64;

/// Close each directory batch before descending: depth and breadth never retain
/// one native directory descriptor per node in the deletion tree.
pub fn remove_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
    remove_dir_all_with_batch_size(path, DIRECTORY_REMOVAL_BATCH_ENTRIES)
}

/// Internal callers can bound retained path memory with a smaller batch.
#[doc(hidden)]
pub fn remove_dir_all_with_batch_size(
    path: impl AsRef<Path>,
    batch_entries: usize,
) -> io::Result<()> {
    if batch_entries == 0 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "directory removal batch must be nonzero",
        ));
    }
    let path = path.as_ref();
    let root = symlink_metadata(path)?;
    if root.file_type().is_symlink() {
        #[cfg(windows)]
        if root.is_dir() {
            return remove_dir(path);
        }
        return remove_file(path);
    }
    if !root.is_dir() {
        return Err(io::Error::new(
            io::ErrorKind::NotADirectory,
            "removal root is not a directory",
        ));
    }
    let mut pending = vec![path.to_path_buf()];
    while let Some(directory) = pending.pop() {
        let children = {
            let entries = read_dir(&directory)?;
            entries
                .take(batch_entries)
                .map(|entry| {
                    let entry = entry?;
                    Ok((entry.path(), entry.file_type()?))
                })
                .collect::<io::Result<Vec<_>>>()?
        };
        if children.is_empty() {
            remove_dir(&directory)?;
        } else {
            pending.push(directory);
            for (child, kind) in children {
                if kind.is_dir() && !kind.is_symlink() {
                    pending.push(child);
                } else {
                    #[cfg(windows)]
                    if kind.is_dir() && kind.is_symlink() {
                        remove_dir(child)?;
                        continue;
                    }
                    remove_file(child)?;
                }
            }
        }
    }
    Ok(())
}