vfsi-sync 0.1.4

//! Owned, shareable synchronous client and file handles.

use std::collections::{HashMap, HashSet};
use std::fmt;
use std::io::{self, Read, Seek, SeekFrom as IoSeekFrom, Write};
use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::SystemTime;

use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
use crate::{
    AttrMask, Capabilities, CopyFileSystem, DEFAULT_READ_MAX_BYTES, DirEntry, DirectoryFileSystem,
    FileSystem, LinkFileSystem, Metadata, MetadataFileSystem, MetadataQuery, MetadataUpdate,
    NamespaceFileSystem, OpenFlags, OpenRequest, Permissions, ReadAllOptions, ReadDirOptions,
    ReadOp, ReadResult, ReadStreamOptions, SetAttributes, VecFs, VectorFileSystem, VfDir, VfError,
    VfFile, VfOffset, VfResult, WriteOpRef, WriteResult,
};

/// Default aggregate payload limit for owned vector reads.
pub const DEFAULT_READV_MAX_TOTAL_BYTES: usize = DEFAULT_READ_MAX_BYTES;

/// One directory and its entries, with attributes fetched during enumeration.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DirectoryListing {
    pub path: PathBuf,
    pub entries: Vec<DirEntry>,
}

fn io_error(error: VfError) -> io::Error {
    error.into()
}

fn poisoned() -> VfError {
    VfError::client(0, crate::ERR_IO)
}

fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
    VfError::transport(
        None,
        format!("{operation} backend returned {actual} results for {expected} requests"),
    )
}

/// Cloneable owner of one synchronous backend connection.
pub struct FsClient<F> {
    inner: Arc<Mutex<F>>,
}

impl<F> Clone for FsClient<F> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

impl<F> fmt::Debug for FsClient<F> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.debug_struct("FsClient").finish_non_exhaustive()
    }
}

impl<F> FsClient<F> {
    pub fn new(filesystem: F) -> Self {
        Self {
            inner: Arc::new(Mutex::new(filesystem)),
        }
    }

    fn lock(&self) -> VfResult<std::sync::MutexGuard<'_, F>> {
        self.inner.lock().map_err(|_| poisoned())
    }

    pub fn into_inner(self) -> Result<F, Self> {
        match Arc::try_unwrap(self.inner) {
            Ok(mutex) => match mutex.into_inner() {
                Ok(filesystem) => Ok(filesystem),
                Err(poisoned) => Ok(poisoned.into_inner()),
            },
            Err(inner) => Err(Self { inner }),
        }
    }
}

impl<F: FileSystem> FsClient<F> {
    /// Open a path read-only.
    pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
        self.open_with(OpenRequest::new(path.as_ref(), OpenFlags::READ))
    }

    pub fn open_with(&self, request: OpenRequest) -> VfResult<FsFile<F>> {
        let file = self.lock()?.open_one(&request)?;
        Ok(FsFile {
            inner: Arc::clone(&self.inner),
            file: Some(file),
            path: request.path,
        })
    }

    pub fn open_options(&self) -> OpenOptions<'_, F> {
        OpenOptions::new(self)
    }

    pub fn create(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
        self.open_with(OpenRequest::new(
            path.as_ref(),
            OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
        ))
    }

    pub fn read(&self, path: impl AsRef<Path>) -> VfResult<Vec<u8>> {
        self.read_with_limit(path, DEFAULT_READ_MAX_BYTES)
    }

    /// Read one complete file while limiting the returned allocation.
    ///
    /// Use [`FsFile::read_native`] or [`std::io::Read`] to stream files that
    /// should not be held in one allocation.
    pub fn read_with_limit(&self, path: impl AsRef<Path>, max_bytes: usize) -> VfResult<Vec<u8>> {
        let path = path.as_ref();
        let mut file = self.open(path)?;
        let mut output = Vec::new();
        let mut chunk = vec![0; 64 * 1024];
        loop {
            let remaining = max_bytes.saturating_sub(output.len());
            let request = if remaining == 0 {
                1
            } else {
                remaining.min(chunk.len())
            };
            let read = file.read_native(&mut chunk[..request])?;
            if read == 0 {
                break;
            }
            if read > remaining {
                return Err(VfError::failure(0, libc::EFBIG as u32).with_context("read", path));
            }
            output.extend_from_slice(&chunk[..read]);
        }
        file.close()?;
        Ok(output)
    }

    pub fn read_to_string(&self, path: impl AsRef<Path>) -> VfResult<String> {
        self.read_to_string_with_limit(path, DEFAULT_READ_MAX_BYTES)
    }

    /// Read one complete UTF-8 file with a caller-selected allocation limit.
    pub fn read_to_string_with_limit(
        &self,
        path: impl AsRef<Path>,
        max_bytes: usize,
    ) -> VfResult<String> {
        let path = path.as_ref();
        String::from_utf8(self.read_with_limit(path, max_bytes)?)
            .map_err(|_| VfError::client(0, crate::ERR_INVAL).with_context("read_to_string", path))
    }

    pub fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> VfResult<()> {
        let path = path.as_ref();
        let mut file = self.create(path)?;
        let mut written = 0;
        while written < data.len() {
            let count = file.write_native(&data[written..])?;
            if count == 0 {
                return Err(VfError::client(0, crate::ERR_IO).with_context("write", path));
            }
            written += count;
        }
        file.close()
    }
}

impl<F: FileSystem> FsClient<F> {
    /// Stream one file from offset zero in bounded chunks.
    ///
    /// The callback runs without holding the backend lock, so it may use this
    /// client or drop other files owned by it. Return `Ok(false)` to stop
    /// successfully. Callback errors
    /// are propagated. The file is closed on success, cancellation, callback
    /// error, or read error. At most one requested chunk is buffered at once.
    pub fn read_stream(
        &self,
        path: impl AsRef<Path>,
        callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
    ) -> VfResult<()> {
        self.read_stream_with_options(path, ReadStreamOptions::default(), callback)
    }

    /// Stream one file using an explicit maximum chunk size.
    pub fn read_stream_with_options(
        &self,
        path: impl AsRef<Path>,
        options: ReadStreamOptions,
        mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
    ) -> VfResult<()> {
        let chunk_size = options.chunk_size_bytes();
        if chunk_size == 0 {
            return Err(VfError::failure(0, crate::ERR_INVAL));
        }

        let file = self.open(path)?;
        let raw_file = file.raw()?.clone();
        let operation = (|| -> VfResult<()> {
            let mut offset = 0u64;
            loop {
                let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
                let result = {
                    let mut backend = self.lock()?;
                    let result = backend.read_one(&request)?;
                    validate_read_results(
                        "read_stream",
                        std::slice::from_ref(&request),
                        std::slice::from_ref(&result),
                    )?;
                    result
                };
                let length = result.data.len();
                if !result.data.is_empty() && !callback(offset, &result.data)? {
                    return Ok(());
                }
                if result.eof {
                    return Ok(());
                }
                offset = offset
                    .checked_add(length as u64)
                    .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
            }
        })();
        let close = file.close();
        match operation {
            Err(error) => Err(error),
            Ok(()) => close,
        }
    }
}

impl<F: MetadataFileSystem> FsClient<F> {
    pub fn metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
        self.lock()?.metadata_path(path.as_ref(), true)
    }

    pub fn symlink_metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
        self.lock()?.metadata_path(path.as_ref(), false)
    }

    pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
        SetMetadata {
            client: self,
            path: path.as_ref().to_path_buf(),
            update: MetadataUpdate::new(),
            follow: true,
        }
    }
}

impl<F: DirectoryFileSystem> FsClient<F> {
    pub fn create_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
        self.create_dir_with_mode(path, 0o777)
    }

    pub fn create_dir_with_mode(&self, path: impl AsRef<Path>, mode: u32) -> VfResult<()> {
        self.lock()?.create_dir_one(path.as_ref(), mode)
    }

    pub fn read_dir(&self, path: impl AsRef<Path>) -> VfResult<Vec<DirEntry>> {
        self.read_dir_with_options(path, ReadDirOptions::default())
    }

    /// Read one directory with explicit entry and path-storage limits.
    pub fn read_dir_with_options(
        &self,
        path: impl AsRef<Path>,
        options: ReadDirOptions,
    ) -> VfResult<Vec<DirEntry>> {
        self.lock()?.read_dir_one(path.as_ref(), options)
    }

    /// Visit one directory incrementally without retaining all entries in the
    /// application. The callback runs under the backend lock and must not
    /// call this client or drop other files owned by it.
    pub fn visit_dir(
        &self,
        path: impl AsRef<Path>,
        callback: impl FnMut(DirEntry) -> VfResult<bool>,
    ) -> VfResult<()> {
        self.visit_dir_with_options(path, ReadDirOptions::default(), callback)
    }

    /// Visit entries with explicit entry and cumulative path-byte limits.
    pub fn visit_dir_with_options(
        &self,
        path: impl AsRef<Path>,
        options: ReadDirOptions,
        mut callback: impl FnMut(DirEntry) -> VfResult<bool>,
    ) -> VfResult<()> {
        self.lock()?
            .visit_dir_one(path.as_ref(), options, &mut callback)
    }
}

impl<F: DirectoryFileSystem + MetadataFileSystem> FsClient<F> {
    pub fn create_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
        let path = path.as_ref();
        let mut current = PathBuf::new();
        for component in path.components() {
            match component {
                Component::RootDir => {
                    current.push(Path::new("/"));
                    continue;
                }
                Component::Normal(part) => current.push(part),
                Component::CurDir => continue,
                Component::ParentDir => {
                    // Preserve `..` so the backend resolves it relative to
                    // its own rooted namespace. Lexically popping `/` would
                    // accidentally turn a later absolute component into a
                    // current-directory-relative path.
                    current.push("..");
                    continue;
                }
                Component::Prefix(_) => return Err(VfError::client(0, crate::ERR_INVAL)),
            }
            match self.create_dir(&current) {
                Ok(()) => {}
                Err(error) if error.err_no() == crate::ERR_EXIST => {
                    if !self.metadata(&current)?.is_dir() {
                        return Err(VfError::client(0, crate::ERR_NOTDIR)
                            .with_context("create_dir_all", &current));
                    }
                }
                Err(error) => return Err(error),
            }
        }
        Ok(())
    }
}

impl<F: NamespaceFileSystem + MetadataFileSystem> FsClient<F> {
    fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Metadata> {
        let mut filesystem = self.lock()?;
        let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
        filesystem
            .metadata_path(path, follow)
            .map_err(|error| error.with_context(operation, path))
    }

    pub fn remove_file(&self, path: impl AsRef<Path>) -> VfResult<()> {
        let path = path.as_ref();
        if self.removal_metadata(path, "remove_file")?.is_dir() {
            return Err(VfError::client(0, crate::ERR_ISDIR).with_context("remove_file", path));
        }
        self.lock()?.remove_one(path, false)
    }

    pub fn remove_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
        let path = path.as_ref();
        if !self.removal_metadata(path, "remove_dir")?.is_dir() {
            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir", path));
        }
        self.lock()?.remove_one(path, false)
    }

    pub fn remove_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
        let path = path.as_ref();
        if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
        }
        self.lock()?.remove_one(path, true)
    }

    /// Remove the contents of a directory, keeping the directory itself.
    pub fn remove_dir_contents(&self, path: impl AsRef<Path>) -> VfResult<()> {
        let path = path.as_ref();
        if !self.removal_metadata(path, "remove_dir_contents")?.is_dir() {
            return Err(
                VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
            );
        }
        self.lock()?.remove_dir_contents(path)
    }

    pub fn rename(&self, from: impl AsRef<Path>, to: impl AsRef<Path>) -> VfResult<()> {
        self.lock()?.rename_one(from.as_ref(), to.as_ref())
    }
}

impl<F: VecFs> FsClient<F> {
    /// List several directories with common stat attributes and finite
    /// allocation limits. Use `read_dirs_with_options` for richer fields.
    pub fn read_dirs<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<DirectoryListing>> {
        self.read_dirs_with_options(paths, AttrMask::stat(), ReadDirOptions::default())
    }

    /// Recursively enumerate a bounded tree with common stat attributes.
    /// Use `walk_with_options` to select fields or change limits.
    pub fn walk(&self, root: impl AsRef<Path>) -> VfResult<Vec<DirectoryListing>> {
        self.walk_with_options(root, AttrMask::stat(), crate::WalkOptions::default())
    }

    /// Create `path` if missing, otherwise empty it. Errors if it exists and is
    /// not a directory (a symlink to a directory is not a directory here).
    pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
        self.lock()?.ensure_empty_dir(path.as_ref())
    }

    /// Open a directory for handle-rooted removal. The returned handle cannot
    /// be swapped for a symlink between opening and removal; release it with
    /// [`close_dir`](Self::close_dir).
    pub fn open_dir(&self, path: impl AsRef<Path>) -> VfResult<VfDir> {
        self.lock()?.open_dir(path.as_ref())
    }

    /// Remove the contents of a directory handle, keeping the directory.
    pub fn remove_dir_contents_handle(&self, dir: &VfDir) -> VfResult<()> {
        self.lock()?.rm_dir_contents(dir)
    }

    /// Release a handle from [`open_dir`](Self::open_dir).
    pub fn close_dir(&self, dir: &VfDir) -> VfResult<()> {
        self.lock()?.close_dir(dir)
    }
}

impl<F: LinkFileSystem> FsClient<F> {
    pub fn symlink(&self, target: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
        self.lock()?.symlink_one(target.as_ref(), link.as_ref())
    }

    pub fn hard_link(&self, source: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
        self.lock()?.hard_link_one(source.as_ref(), link.as_ref())
    }

    pub fn read_link(&self, path: impl AsRef<Path>) -> VfResult<PathBuf> {
        self.lock()?.read_link_one(path.as_ref())
    }
}

impl<F: CopyFileSystem> FsClient<F> {
    pub fn copy(&self, source: impl AsRef<Path>, destination: impl AsRef<Path>) -> VfResult<()> {
        self.lock()?.copy_one(source.as_ref(), destination.as_ref())
    }
}

impl<F: VecFs> FsClient<F> {
    /// Fetch selected metadata for one path without following its final symlink.
    /// Unavailable fields remain `None` on [`Metadata`].
    pub fn symlink_metadata_with_fields(
        &self,
        path: impl AsRef<Path>,
        fields: AttrMask,
    ) -> VfResult<Metadata> {
        let path = path.as_ref();
        let mut attrs = crate::VfAttrs {
            file: VfFile::from_os_path(path),
            masks: fields | AttrMask::MODE | AttrMask::SIZE,
            ..crate::VfAttrs::default()
        };
        self.lock()?
            .lgetattrsv(std::slice::from_mut(&mut attrs))
            .map_err(|error| error.with_context("symlink_metadata", path))?;
        Ok(attrs.into())
    }

    /// List multiple directories in a vector call. The limits apply to the
    /// aggregate returned entries and stored path bytes. Streaming backends
    /// apply these limits before collecting a full listing; a backend using
    /// the compatibility `visit_dir` fallback may buffer one directory first.
    /// An error discards the collected prefix; callers may retry individual
    /// directories if desired.
    pub fn read_dirs_with_options<P: AsRef<Path>>(
        &self,
        paths: &[P],
        fields: AttrMask,
        options: ReadDirOptions,
    ) -> VfResult<Vec<DirectoryListing>> {
        if paths.is_empty() {
            return Ok(Vec::new());
        }
        let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
        let mut positions: HashMap<PathBuf, Vec<usize>> = HashMap::with_capacity(paths.len());
        let mut unique_paths = Vec::with_capacity(paths.len());
        for (index, path) in paths.iter().enumerate() {
            if let Some(indices) = positions.get_mut(*path) {
                indices.push(index);
            } else {
                positions.insert(path.to_path_buf(), vec![index]);
                unique_paths.push(*path);
            }
        }
        let mut listings: Vec<DirectoryListing> = paths
            .iter()
            .map(|path| DirectoryListing {
                path: path.to_path_buf(),
                entries: Vec::new(),
            })
            .collect();
        let mut entry_count = 0usize;
        let mut path_bytes = 0usize;
        // Bound the number of simultaneous first READDIR pages even when the
        // caller supplies thousands of directory operands.
        const DIRECTORY_COHORT: usize = 32;
        for cohort in unique_paths.chunks(DIRECTORY_COHORT) {
            let mut callback_error = None;
            let requested = options
                .entry_limit()
                .saturating_sub(entry_count)
                .saturating_add(1);
            let result = self.lock()?.listdirv(
                cohort,
                fields | AttrMask::MODE | AttrMask::SIZE,
                requested,
                false,
                &mut |attributes, directory| {
                    let Some(indices) = positions.get(directory) else {
                        callback_error = Some(VfError::transport(
                            None,
                            format!(
                                "read_dirs backend returned an unexpected directory: {}",
                                directory.display()
                            ),
                        ));
                        return false;
                    };
                    if !cohort.contains(&directory) {
                        callback_error = Some(VfError::transport(
                            None,
                            "read_dirs backend returned a directory outside the active cohort",
                        ));
                        return false;
                    }
                    let Some(path) = attributes.file.path() else {
                        callback_error = Some(VfError::transport(
                            None,
                            "read_dirs backend returned an entry without a path",
                        ));
                        return false;
                    };
                    if path.parent() != Some(directory) {
                        callback_error = Some(VfError::transport(
                            None,
                            format!(
                                "read_dirs backend returned an entry outside {}",
                                directory.display()
                            ),
                        ));
                        return false;
                    }
                    for &index in indices {
                        entry_count += 1;
                        path_bytes = path_bytes.saturating_add(path.as_os_str().len());
                        if entry_count > options.entry_limit()
                            || path_bytes > options.path_byte_limit()
                        {
                            callback_error = Some(
                                VfError::failure(index, libc::EFBIG as u32)
                                    .with_context("read_dirs", directory),
                            );
                            return false;
                        }
                        listings[index]
                            .entries
                            .push(DirEntry::new(path.to_path_buf(), attributes.clone().into()));
                    }
                    true
                },
            );
            if let Some(error) = callback_error {
                return Err(error);
            }
            result.map_err(|error| {
                error.map_index(|index| {
                    cohort
                        .get(index)
                        .and_then(|path| positions.get(*path))
                        .and_then(|indices| indices.first())
                        .copied()
                        .unwrap_or(index)
                })
            })?;
        }
        Ok(listings)
    }

    /// Recursively enumerate directories with selected entry attributes.
    /// The walk is bounded by `options`; sorting and presentation remain the
    /// application's responsibility.
    pub fn walk_with_options(
        &self,
        root: impl AsRef<Path>,
        fields: AttrMask,
        options: crate::WalkOptions,
    ) -> VfResult<Vec<DirectoryListing>> {
        let root = root.as_ref();
        let tree = self.lock()?.walk_with_options(
            root,
            fields | AttrMask::MODE | AttrMask::SIZE,
            options,
            &mut |_, _| {},
        )?;
        tree.into_iter()
            .map(|directory| {
                let entries = directory
                    .entries
                    .into_iter()
                    .enumerate()
                    .map(|(index, attributes)| {
                        let path = attributes
                            .file
                            .path()
                            .ok_or_else(|| {
                                VfError::transport(
                                    None,
                                    format!("walk backend returned entry {index} without a path"),
                                )
                            })?
                            .to_path_buf();
                        Ok(DirEntry::new(path, attributes.into()))
                    })
                    .collect::<VfResult<Vec<_>>>()?;
                Ok(DirectoryListing {
                    path: directory.path,
                    entries,
                })
            })
            .collect()
    }

    /// Copy whole files in request order. A successful prefix may remain if
    /// a later request fails; this operation does not provide atomicity.
    pub fn copy_files<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
        let extents: Vec<_> = pairs
            .iter()
            .map(|(from, to)| {
                crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
            })
            .collect();
        self.lock()?.copyv(&extents).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| pairs.get(index))
                .map_or(error.clone(), |(_, to)| {
                    error.with_context("copy_files", to.as_ref())
                })
        })
    }

    /// Remove paths in request order, optionally recursing into directories.
    /// A successful prefix may remain if a later path fails.
    pub fn remove_paths<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
        let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
        self.lock()?.rm(&paths, recursive).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| paths.get(index))
                .map_or(error.clone(), |path| {
                    error.with_context("remove_paths", path)
                })
        })
    }

    /// Fetch no-follow metadata for many paths using the backend's vector
    /// operation. Useful for routing without one metadata RPC per path.
    pub fn symlink_metadatav(&self, paths: &[&Path]) -> VfResult<Vec<Metadata>> {
        let mut attrs: Vec<_> = paths
            .iter()
            .map(|path| crate::VfAttrs {
                file: VfFile::from_os_path(path),
                masks: AttrMask::MODE,
                ..crate::VfAttrs::default()
            })
            .collect();
        self.lock()?.lgetattrsv(&mut attrs).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| paths.get(index))
                .map_or(error.clone(), |path| {
                    error.with_context("symlink_metadatav", path)
                })
        })?;
        Ok(attrs.into_iter().map(Metadata::from).collect())
    }
}

impl<F: VectorFileSystem> FsClient<F> {
    /// Open an ordered vector of files.
    ///
    /// Success returns one RAII handle per request. Failure returns no
    /// handles; VFSI does not promise transactional rollback of other
    /// filesystem effects such as file creation.
    pub fn openv(&self, requests: &[OpenRequest]) -> VfResult<Vec<FsFile<F>>> {
        let mut filesystem = self.lock()?;
        let files = filesystem.open_many(requests).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| requests.get(index))
                .map_or(error.clone(), |request| {
                    error.with_context("openv", &request.path)
                })
        })?;
        if files.len() != requests.len() {
            let error = wrong_result_count("openv", requests.len(), files.len());
            for file in &files {
                let _ = filesystem.close_one(file);
            }
            return Err(error);
        }
        drop(filesystem);
        Ok(files
            .into_iter()
            .zip(requests)
            .map(|(file, request)| FsFile {
                inner: Arc::clone(&self.inner),
                file: Some(file),
                path: request.path.clone(),
            })
            .collect())
    }

    /// Try to close a group through one vector operation without consuming
    /// the handles. On failure, all handles remain armed: the backend may
    /// have closed a prefix, so callers must reconcile before retrying.
    pub fn try_closev(&self, files: &mut [FsFile<F>]) -> VfResult<()> {
        for (index, file) in files.iter().enumerate() {
            self.validate_owner(file, index)?;
        }
        let descriptors: Vec<VfFile> = files
            .iter()
            .map(|file| file.raw().cloned())
            .collect::<VfResult<_>>()?;
        self.lock()?.close_many(&descriptors).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| files.get(index))
                .map_or(error.clone(), |file| {
                    error.with_context("closev", file.path())
                })
        })?;
        for file in files.iter_mut() {
            file.file = None;
        }
        Ok(())
    }

    /// Close a group of files through one vector operation.
    ///
    /// On failure, the handles are dropped and the backend receives
    /// best-effort scalar cleanup attempts. Use [`try_closev`](Self::try_closev)
    /// to retain the handles after an error.
    pub fn closev(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
        self.try_closev(&mut files)
    }

    /// Read an ordered vector with a 16 MiB aggregate request limit.
    /// Use [`readv_with_limit`](Self::readv_with_limit) to tune the limit or
    /// [`readv_into`](Self::readv_into) to provide bounded caller-owned buffers.
    pub fn readv(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<ReadResult>> {
        self.readv_with_limit(requests, DEFAULT_READV_MAX_TOTAL_BYTES)
    }

    /// Read an ordered vector with an explicit aggregate request limit.
    pub fn readv_with_limit(
        &self,
        requests: &[FsRead<'_, F>],
        max_total_bytes: usize,
    ) -> VfResult<Vec<ReadResult>> {
        let mut requested = 0usize;
        for (index, request) in requests.iter().enumerate() {
            requested = requested
                .checked_add(request.length)
                .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
            if requested > max_total_bytes {
                return Err(VfError::client(index, libc::EFBIG as u32)
                    .with_context("readv", request.file.path()));
            }
        }
        let reads = self.read_ops(requests)?;
        let results = self.lock()?.read_many(&reads).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| requests.get(index))
                .map_or(error.clone(), |request| {
                    error.with_context("readv", request.file.path())
                })
        })?;
        if results.len() != requests.len() {
            return Err(wrong_result_count("readv", requests.len(), results.len()));
        }
        validate_read_results("readv", &reads, &results).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| requests.get(index))
                .map_or(error.clone(), |request| {
                    error.with_context("readv", request.file.path())
                })
        })?;
        Ok(results)
    }

    fn read_ops(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<ReadOp>> {
        let mut reads = Vec::with_capacity(requests.len());
        for (index, request) in requests.iter().enumerate() {
            self.validate_owner(request.file, index)?;
            reads.push(ReadOp::new(
                request.file.raw()?.clone(),
                request.offset,
                request.length,
            ));
        }
        Ok(reads)
    }

    pub fn readv_into(&self, requests: &mut [FsReadInto<'_, F>]) -> VfResult<Vec<usize>> {
        let mut reads = Vec::with_capacity(requests.len());
        for (index, request) in requests.iter().enumerate() {
            self.validate_owner(request.file, index)?;
            reads.push(ReadOp::new(
                request.file.raw()?.clone(),
                request.offset,
                request.buffer.len(),
            ));
        }
        let results = {
            let mut buffers: Vec<&mut [u8]> = requests
                .iter_mut()
                .map(|request| &mut *request.buffer)
                .collect();
            self.lock()?.read_many_into(&reads, &mut buffers)
        }
        .map_err(|error| {
            error
                .index_opt()
                .and_then(|index| requests.get(index))
                .map_or(error.clone(), |request| {
                    error.with_context("readv_into", request.file.path())
                })
        })?;
        validate_read_into_results("readv_into", &reads, &results).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| requests.get(index))
                .map_or(error.clone(), |request| {
                    error.with_context("readv_into", request.file.path())
                })
        })?;
        Ok(results.into_iter().map(|result| result.read).collect())
    }

    pub fn writev(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<WriteResult>> {
        let writes = self.write_ops(requests)?;
        let results = self.lock()?.write_many(&writes).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| requests.get(index))
                .map_or(error.clone(), |request| {
                    error.with_context("writev", request.file.path())
                })
        })?;
        if results.len() != requests.len() {
            return Err(wrong_result_count("writev", requests.len(), results.len()));
        }
        validate_write_results("writev", &writes, &results).map_err(|error| {
            error
                .index_opt()
                .and_then(|index| requests.get(index))
                .map_or(error.clone(), |request| {
                    error.with_context("writev", request.file.path())
                })
        })?;
        Ok(results)
    }

    /// Write every byte in each positional request, retrying short writes in
    /// vector waves. Like `writev`, this is not transactional: an error may
    /// follow a successfully written prefix. Overlapping requests through the
    /// same path complete in input order; different paths are presumed
    /// independent (including hard-link aliases).
    pub fn write_allv(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<WriteResult>> {
        // Validate the entire batch before writing any prefix. In particular,
        // empty requests must not conceal a foreign or already-closed file.
        let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
        let mut blocked_by = vec![Vec::new(); requests.len()];
        for (index, request) in requests.iter().enumerate() {
            self.validate_owner(request.file, index)?;
            request.file.raw()?;
            let VfOffset::At(offset) = request.offset else {
                return Err(VfError::client(index, crate::ERR_INVAL)
                    .with_context("write_allv", request.file.path()));
            };
            let end = offset
                .checked_add(request.data.len() as u64)
                .ok_or_else(|| {
                    VfError::client(index, libc::EOVERFLOW as u32)
                        .with_context("write_allv", request.file.path())
                })?;
            // Non-overlapping writes commute. An overlapping later request
            // must wait until every earlier conflicting request is complete:
            // otherwise a short-write retry can overwrite the later bytes.
            let prior = prior_by_path.entry(request.file.path()).or_default();
            for &(earlier, start, earlier_end) in prior.iter() {
                if offset < earlier_end && start < end {
                    blocked_by[index].push(earlier);
                }
            }
            prior.push((index, offset, end));
        }
        let mut totals = vec![0usize; requests.len()];
        let mut stable = vec![true; requests.len()];
        let mut pending: Vec<usize> = (0..requests.len())
            .filter(|&index| !requests[index].data.is_empty())
            .collect();
        while !pending.is_empty() {
            let wave_indices: Vec<usize> = pending
                .iter()
                .copied()
                .filter(|&index| {
                    blocked_by[index]
                        .iter()
                        .all(|&earlier| totals[earlier] == requests[earlier].data.len())
                })
                .collect();
            let wave: Vec<_> = wave_indices
                .iter()
                .map(|&index| {
                    let request = &requests[index];
                    let VfOffset::At(offset) = request.offset else {
                        return Err(VfError::client(index, crate::ERR_INVAL)
                            .with_context("write_allv", request.file.path()));
                    };
                    let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
                        VfError::client(index, libc::EOVERFLOW as u32)
                            .with_context("write_allv", request.file.path())
                    })?;
                    Ok(FsWrite {
                        file: request.file,
                        offset: VfOffset::At(offset),
                        data: &request.data[totals[index]..],
                    })
                })
                .collect::<VfResult<_>>()?;
            let results = self.writev(&wave).map_err(|error| {
                error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
            })?;
            for (&index, result) in wave_indices.iter().zip(results) {
                if result.written == 0 {
                    return Err(VfError::client(index, crate::ERR_IO)
                        .with_context("write_allv", requests[index].file.path()));
                }
                totals[index] += result.written;
                stable[index] &= result.stable;
            }
            pending.retain(|&index| totals[index] < requests[index].data.len());
        }
        requests
            .iter()
            .enumerate()
            .map(|(index, request)| {
                let VfOffset::At(offset) = request.offset else {
                    return Err(VfError::client(index, crate::ERR_INVAL)
                        .with_context("write_allv", request.file.path()));
                };
                Ok(WriteResult {
                    file: request.file.raw()?.clone(),
                    offset,
                    written: totals[index],
                    stable: stable[index],
                })
            })
            .collect()
    }

    fn write_ops<'a>(&self, requests: &'a [FsWrite<'a, F>]) -> VfResult<Vec<WriteOpRef<'a>>> {
        let mut writes = Vec::with_capacity(requests.len());
        for (index, request) in requests.iter().enumerate() {
            self.validate_owner(request.file, index)?;
            writes.push(WriteOpRef::new(
                request.file.raw()?,
                request.offset,
                request.data,
            ));
        }
        Ok(writes)
    }

    fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
        if Arc::ptr_eq(&self.inner, &file.inner) {
            Ok(())
        } else {
            Err(VfError::client(index, crate::ERR_INVAL))
        }
    }
}

impl<F: VectorFileSystem + VecFs> FsClient<F> {
    /// Read several complete files by path using vector READ operations.
    ///
    /// The aggregate returned data is limited to 16 MiB by default. Use
    /// [`read_files_with_options`](Self::read_files_with_options) to choose a
    /// different limit, or stream large files instead. This is not a snapshot
    /// or an atomic operation across files.
    pub fn read_files<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<Vec<u8>>> {
        self.read_files_with_options(paths, ReadAllOptions::default())
    }

    /// Read several complete files with an explicit aggregate allocation limit.
    pub fn read_files_with_options<P: AsRef<Path>>(
        &self,
        paths: &[P],
        options: ReadAllOptions,
    ) -> VfResult<Vec<Vec<u8>>> {
        let files: Vec<_> = paths
            .iter()
            .map(|path| VfFile::from_os_path(path.as_ref()))
            .collect();
        let result = self
            .lock()?
            .read_allv_with_options(&files, options)
            .map_err(|error| {
                error
                    .index_opt()
                    .and_then(|index| paths.get(index))
                    .map_or(error.clone(), |path| {
                        error.with_context("read_files", path.as_ref())
                    })
            });
        result.and_then(|buffers| {
            if buffers.len() != paths.len() {
                Err(wrong_result_count("read_files", paths.len(), buffers.len()))
            } else {
                Ok(buffers)
            }
        })
    }

    /// Replace several files from borrowed buffers using vector OPEN, WRITE,
    /// and CLOSE phases. Identical path spellings are rejected before opening
    /// anything; aliases such as hard links are still the caller's responsibility.
    /// The batch is not transactional: an error may follow files already
    /// created or written. Large inputs should be chunked by the caller rather
    /// than held in memory solely for this convenience method.
    pub fn write_files<P: AsRef<Path>, B: AsRef<[u8]>>(&self, entries: &[(P, B)]) -> VfResult<()> {
        let mut seen = HashSet::with_capacity(entries.len());
        for (index, (path, _)) in entries.iter().enumerate() {
            if !seen.insert(path.as_ref()) {
                return Err(VfError::client(index, crate::ERR_INVAL)
                    .with_context("write_files", path.as_ref()));
            }
        }
        let requests: Vec<_> = entries
            .iter()
            .map(|(path, _)| {
                OpenRequest::new(
                    path.as_ref(),
                    OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
                )
            })
            .collect();
        let files = self.openv(&requests)?;
        let writes: Vec<_> = files
            .iter()
            .zip(entries)
            .map(|(file, (_, data))| file.write_request_at(0, data.as_ref()))
            .collect();
        let result = self.write_allv(&writes);
        drop(writes);
        let close_result = self.closev(files);
        result?;
        close_result
    }
}

/// `std::fs::OpenOptions`-style builder tied to an [`FsClient`].
pub struct OpenOptions<'a, F: FileSystem> {
    client: &'a FsClient<F>,
    flags: OpenFlags,
    mode: u32,
}

impl<F: FileSystem> Clone for OpenOptions<'_, F> {
    fn clone(&self) -> Self {
        Self {
            client: self.client,
            flags: self.flags,
            mode: self.mode,
        }
    }
}

impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OpenOptions")
            .field("flags", &self.flags)
            .field("mode", &format_args!("{:#o}", self.mode))
            .finish()
    }
}

impl<'a, F: FileSystem> OpenOptions<'a, F> {
    fn new(client: &'a FsClient<F>) -> Self {
        Self {
            client,
            flags: OpenFlags::empty(),
            mode: 0o666,
        }
    }

    pub fn read(&mut self, enabled: bool) -> &mut Self {
        self.flags.set(OpenFlags::READ, enabled);
        self
    }

    pub fn write(&mut self, enabled: bool) -> &mut Self {
        self.flags.set(OpenFlags::WRITE, enabled);
        self
    }

    pub fn append(&mut self, enabled: bool) -> &mut Self {
        self.flags.set(OpenFlags::APPEND, enabled);
        self
    }

    pub fn truncate(&mut self, enabled: bool) -> &mut Self {
        self.flags.set(OpenFlags::TRUNCATE, enabled);
        self
    }

    pub fn create(&mut self, enabled: bool) -> &mut Self {
        self.flags.set(OpenFlags::CREATE, enabled);
        self
    }

    pub fn create_new(&mut self, enabled: bool) -> &mut Self {
        self.flags.set(OpenFlags::CREATE_NEW, enabled);
        self
    }

    pub fn mode(&mut self, mode: u32) -> &mut Self {
        self.mode = mode;
        self
    }

    pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
        self.client
            .open_with(OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
    }
}

impl<F: VectorFileSystem> OpenOptions<'_, F> {
    pub fn openv<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
        let requests: Vec<OpenRequest> = paths
            .iter()
            .map(|path| OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
            .collect();
        self.client.openv(&requests)
    }
}

/// Builder for an atomic metadata update request.
pub struct SetMetadata<'a, F: MetadataFileSystem> {
    client: &'a FsClient<F>,
    path: PathBuf,
    update: MetadataUpdate,
    follow: bool,
}

impl<F: MetadataFileSystem> Clone for SetMetadata<'_, F> {
    fn clone(&self) -> Self {
        Self {
            client: self.client,
            path: self.path.clone(),
            update: self.update.clone(),
            follow: self.follow,
        }
    }
}

impl<F: MetadataFileSystem> fmt::Debug for SetMetadata<'_, F> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SetMetadata")
            .field("path", &self.path)
            .field("update", &self.update)
            .field("follow", &self.follow)
            .finish_non_exhaustive()
    }
}

impl<F: MetadataFileSystem> SetMetadata<'_, F> {
    pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
        self.update.permissions = Some(permissions);
        self
    }

    pub fn len(&mut self, len: u64) -> &mut Self {
        self.update.len = Some(len);
        self
    }

    pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
        self.update.accessed = Some(accessed);
        self
    }

    pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
        self.update.modified = Some(modified);
        self
    }

    pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
        self.follow = follow;
        self
    }

    pub fn apply(&self) -> VfResult<()> {
        self.client
            .lock()?
            .set_metadata_path(&self.path, self.update.clone(), self.follow)
    }
}

/// Owned RAII file which does not borrow the client.
///
/// Dropping an open file attempts a best-effort CLOSE. This can block on the
/// backend mutex and on network I/O; use [`FsFile::try_close`] when the close
/// result matters or when its timing must be controlled.
pub struct FsFile<F: FileSystem> {
    inner: Arc<Mutex<F>>,
    file: Option<VfFile>,
    path: PathBuf,
}

impl<F: FileSystem> fmt::Debug for FsFile<F> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("FsFile")
            .field("path", &self.path)
            .field("is_open", &self.file.is_some())
            .finish()
    }
}

impl<F: FileSystem> FsFile<F> {
    fn raw(&self) -> VfResult<&VfFile> {
        self.file
            .as_ref()
            .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
    }

    /// Positional read which does not alter the file cursor.
    pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
        self.read_into(buffer, VfOffset::At(offset))
    }

    /// Positional write which does not alter the file cursor.
    pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
        self.write_from(buffer, VfOffset::At(offset))
    }

    pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
        FsRead {
            file: self,
            offset: VfOffset::At(offset),
            length,
        }
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Query metadata for the open object without resolving its path again.
    pub fn metadata(&self) -> VfResult<Metadata> {
        let attributes = AttrMask::MODE
            | AttrMask::SIZE
            | AttrMask::NLINK
            | AttrMask::FILEID
            | AttrMask::UID
            | AttrMask::GID
            | AttrMask::ATIME
            | AttrMask::MTIME
            | AttrMask::CTIME
            | AttrMask::CHANGE;
        self.inner
            .lock()
            .map_err(|_| poisoned())?
            .metadata(MetadataQuery::new(self.raw()?.clone(), attributes))
            .map(Metadata::from)
            .map_err(|error| error.with_context("metadata", &self.path))
    }

    /// Truncate or extend the open file.
    pub fn set_len(&self, len: u64) -> VfResult<()> {
        let mut update = SetAttributes::new(self.raw()?.clone());
        update.size = Some(len);
        self.inner
            .lock()
            .map_err(|_| poisoned())?
            .set_attributes(update)
            .map_err(|error| error.with_context("set_len", &self.path))
    }

    /// Change permissions on the open file.
    pub fn set_permissions(&self, permissions: Permissions) -> VfResult<()> {
        let mut update = SetAttributes::new(self.raw()?.clone());
        update.mode = Some(permissions.mode());
        self.inner
            .lock()
            .map_err(|_| poisoned())?
            .set_attributes(update)
            .map_err(|error| error.with_context("set_permissions", &self.path))
    }

    pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
        FsWrite {
            file: self,
            offset: VfOffset::At(offset),
            data,
        }
    }

    pub fn read_request_at_into<'a>(
        &'a self,
        offset: u64,
        buffer: &'a mut [u8],
    ) -> FsReadInto<'a, F> {
        FsReadInto {
            file: self,
            offset: VfOffset::At(offset),
            buffer,
        }
    }

    pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
        self.read_into(buffer, VfOffset::Cur)
    }

    pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
        self.write_from(buffer, VfOffset::Cur)
    }

    fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
        if buffer.is_empty() {
            return Ok(0);
        }
        let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
        let result = self
            .inner
            .lock()
            .map_err(|_| poisoned())?
            .read_one_into(&request, buffer)
            .map_err(|error| error.with_context("read", &self.path))?;
        validate_read_into_results(
            "read",
            std::slice::from_ref(&request),
            std::slice::from_ref(&result),
        )
        .map_err(|error| error.with_context("read", &self.path))?;
        Ok(result.read)
    }

    fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
        if buffer.is_empty() {
            return Ok(0);
        }
        let file = self.raw()?.clone();
        let result = self
            .inner
            .lock()
            .map_err(|_| poisoned())?
            .write_one(WriteOpRef::new(&file, offset, buffer))
            .map_err(|error| error.with_context("write", &self.path))?;
        if result.written > buffer.len() {
            return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
        }
        Ok(result.written)
    }

    pub fn sync_data(&self) -> VfResult<()> {
        self.inner
            .lock()
            .map_err(|_| poisoned())?
            .sync_data(self.raw()?)
            .map_err(|error| error.with_context("sync_data", &self.path))
    }

    pub fn sync_all(&self) -> VfResult<()> {
        self.inner
            .lock()
            .map_err(|_| poisoned())?
            .sync_all(self.raw()?)
            .map_err(|error| error.with_context("sync_all", &self.path))
    }

    /// Attempt to close without consuming the handle. A failed close leaves
    /// the handle armed so the caller can reconcile or retry it explicitly.
    pub fn try_close(&mut self) -> VfResult<()> {
        let Some(file) = self.file.as_ref() else {
            return Ok(());
        };
        self.inner
            .lock()
            .map_err(|_| poisoned())?
            .close_one(file)
            .map_err(|error| error.with_context("close", &self.path))?;
        self.file = None;
        Ok(())
    }

    /// Consume and close the handle. On failure, `Drop` makes one best-effort
    /// cleanup attempt; use [`try_close`](Self::try_close) to retain control.
    pub fn close(mut self) -> VfResult<()> {
        self.try_close()
    }

    /// Seek while retaining [`VfError`] protocol and path information.
    pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
        self.inner
            .lock()
            .map_err(|_| poisoned())?
            .seek_one(self.raw()?, position)
            .map_err(|error| error.with_context("seek", &self.path))
    }
}

impl<F: FileSystem> Read for FsFile<F> {
    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
        self.read_native(buffer).map_err(io_error)
    }
}

impl<F: FileSystem> Write for FsFile<F> {
    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
        self.write_native(buffer).map_err(io_error)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.sync_data().map_err(io_error)
    }
}

impl<F: FileSystem> Seek for FsFile<F> {
    fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
        self.seek_native(position).map_err(io_error)
    }
}

/// Typed read request for [`FsClient::readv`].
pub struct FsRead<'a, F: FileSystem> {
    file: &'a FsFile<F>,
    offset: VfOffset,
    length: usize,
}

/// Typed borrowed write request for [`FsClient::writev`].
pub struct FsWrite<'a, F: FileSystem> {
    file: &'a FsFile<F>,
    offset: VfOffset,
    data: &'a [u8],
}

/// Typed vector read into caller-provided storage.
pub struct FsReadInto<'a, F: FileSystem> {
    file: &'a FsFile<F>,
    offset: VfOffset,
    buffer: &'a mut [u8],
}

impl<F: FileSystem> Drop for FsFile<F> {
    fn drop(&mut self) {
        if let Some(file) = self.file.take()
            && let Ok(mut filesystem) = self.inner.lock()
        {
            let _ = filesystem.close_one(&file);
        }
    }
}