Skip to main content

vfsi_sync/
client.rs

1//! Owned, shareable synchronous client and file handles.
2
3use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::io::{self, Read, Seek, SeekFrom as IoSeekFrom, Write};
6use std::path::{Component, Path, PathBuf};
7use std::sync::{Arc, Mutex};
8use std::time::SystemTime;
9
10use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
11use crate::{
12    AttrMask, Capabilities, CopyFileSystem, DEFAULT_READ_MAX_BYTES, DirEntry, DirectoryFileSystem,
13    FileSystem, LinkFileSystem, Metadata, MetadataFileSystem, MetadataQuery, MetadataUpdate,
14    NamespaceFileSystem, OpenFlags, OpenRequest, Permissions, ReadAllOptions, ReadDirOptions,
15    ReadOp, ReadResult, ReadStreamOptions, SetAttributes, VecFs, VectorFileSystem, VfDir, VfError,
16    VfFile, VfOffset, VfResult, WriteOpRef, WriteResult,
17};
18
19/// Default aggregate payload limit for owned vector reads.
20pub const DEFAULT_READV_MAX_TOTAL_BYTES: usize = DEFAULT_READ_MAX_BYTES;
21
22/// One directory and its entries, with attributes fetched during enumeration.
23#[derive(Clone, Debug, PartialEq, Eq)]
24pub struct DirectoryListing {
25    pub path: PathBuf,
26    pub entries: Vec<DirEntry>,
27}
28
29fn io_error(error: VfError) -> io::Error {
30    error.into()
31}
32
33fn poisoned() -> VfError {
34    VfError::client(0, crate::ERR_IO)
35}
36
37fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
38    VfError::transport(
39        None,
40        format!("{operation} backend returned {actual} results for {expected} requests"),
41    )
42}
43
44/// Cloneable owner of one synchronous backend connection.
45pub struct FsClient<F> {
46    inner: Arc<Mutex<F>>,
47}
48
49impl<F> Clone for FsClient<F> {
50    fn clone(&self) -> Self {
51        Self {
52            inner: Arc::clone(&self.inner),
53        }
54    }
55}
56
57impl<F> fmt::Debug for FsClient<F> {
58    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
59        formatter.debug_struct("FsClient").finish_non_exhaustive()
60    }
61}
62
63impl<F> FsClient<F> {
64    pub fn new(filesystem: F) -> Self {
65        Self {
66            inner: Arc::new(Mutex::new(filesystem)),
67        }
68    }
69
70    fn lock(&self) -> VfResult<std::sync::MutexGuard<'_, F>> {
71        self.inner.lock().map_err(|_| poisoned())
72    }
73
74    pub fn into_inner(self) -> Result<F, Self> {
75        match Arc::try_unwrap(self.inner) {
76            Ok(mutex) => match mutex.into_inner() {
77                Ok(filesystem) => Ok(filesystem),
78                Err(poisoned) => Ok(poisoned.into_inner()),
79            },
80            Err(inner) => Err(Self { inner }),
81        }
82    }
83}
84
85impl<F: FileSystem> FsClient<F> {
86    /// Open a path read-only.
87    pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
88        self.open_with(OpenRequest::new(path.as_ref(), OpenFlags::READ))
89    }
90
91    pub fn open_with(&self, request: OpenRequest) -> VfResult<FsFile<F>> {
92        let file = self.lock()?.open_one(&request)?;
93        Ok(FsFile {
94            inner: Arc::clone(&self.inner),
95            file: Some(file),
96            path: request.path,
97        })
98    }
99
100    pub fn open_options(&self) -> OpenOptions<'_, F> {
101        OpenOptions::new(self)
102    }
103
104    pub fn create(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
105        self.open_with(OpenRequest::new(
106            path.as_ref(),
107            OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
108        ))
109    }
110
111    pub fn read(&self, path: impl AsRef<Path>) -> VfResult<Vec<u8>> {
112        self.read_with_limit(path, DEFAULT_READ_MAX_BYTES)
113    }
114
115    /// Read one complete file while limiting the returned allocation.
116    ///
117    /// Use [`FsFile::read_native`] or [`std::io::Read`] to stream files that
118    /// should not be held in one allocation.
119    pub fn read_with_limit(&self, path: impl AsRef<Path>, max_bytes: usize) -> VfResult<Vec<u8>> {
120        let path = path.as_ref();
121        let mut file = self.open(path)?;
122        let mut output = Vec::new();
123        let mut chunk = vec![0; 64 * 1024];
124        loop {
125            let remaining = max_bytes.saturating_sub(output.len());
126            let request = if remaining == 0 {
127                1
128            } else {
129                remaining.min(chunk.len())
130            };
131            let read = file.read_native(&mut chunk[..request])?;
132            if read == 0 {
133                break;
134            }
135            if read > remaining {
136                return Err(VfError::failure(0, libc::EFBIG as u32).with_context("read", path));
137            }
138            output.extend_from_slice(&chunk[..read]);
139        }
140        file.close()?;
141        Ok(output)
142    }
143
144    pub fn read_to_string(&self, path: impl AsRef<Path>) -> VfResult<String> {
145        self.read_to_string_with_limit(path, DEFAULT_READ_MAX_BYTES)
146    }
147
148    /// Read one complete UTF-8 file with a caller-selected allocation limit.
149    pub fn read_to_string_with_limit(
150        &self,
151        path: impl AsRef<Path>,
152        max_bytes: usize,
153    ) -> VfResult<String> {
154        let path = path.as_ref();
155        String::from_utf8(self.read_with_limit(path, max_bytes)?)
156            .map_err(|_| VfError::client(0, crate::ERR_INVAL).with_context("read_to_string", path))
157    }
158
159    pub fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> VfResult<()> {
160        let path = path.as_ref();
161        let mut file = self.create(path)?;
162        let mut written = 0;
163        while written < data.len() {
164            let count = file.write_native(&data[written..])?;
165            if count == 0 {
166                return Err(VfError::client(0, crate::ERR_IO).with_context("write", path));
167            }
168            written += count;
169        }
170        file.close()
171    }
172}
173
174impl<F: FileSystem> FsClient<F> {
175    /// Stream one file from offset zero in bounded chunks.
176    ///
177    /// The callback runs without holding the backend lock, so it may use this
178    /// client or drop other files owned by it. Return `Ok(false)` to stop
179    /// successfully. Callback errors
180    /// are propagated. The file is closed on success, cancellation, callback
181    /// error, or read error. At most one requested chunk is buffered at once.
182    pub fn read_stream(
183        &self,
184        path: impl AsRef<Path>,
185        callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
186    ) -> VfResult<()> {
187        self.read_stream_with_options(path, ReadStreamOptions::default(), callback)
188    }
189
190    /// Stream one file using an explicit maximum chunk size.
191    pub fn read_stream_with_options(
192        &self,
193        path: impl AsRef<Path>,
194        options: ReadStreamOptions,
195        mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
196    ) -> VfResult<()> {
197        let chunk_size = options.chunk_size_bytes();
198        if chunk_size == 0 {
199            return Err(VfError::failure(0, crate::ERR_INVAL));
200        }
201
202        let file = self.open(path)?;
203        let raw_file = file.raw()?.clone();
204        let operation = (|| -> VfResult<()> {
205            let mut offset = 0u64;
206            loop {
207                let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
208                let result = {
209                    let mut backend = self.lock()?;
210                    let result = backend.read_one(&request)?;
211                    validate_read_results(
212                        "read_stream",
213                        std::slice::from_ref(&request),
214                        std::slice::from_ref(&result),
215                    )?;
216                    result
217                };
218                let length = result.data.len();
219                if !result.data.is_empty() && !callback(offset, &result.data)? {
220                    return Ok(());
221                }
222                if result.eof {
223                    return Ok(());
224                }
225                offset = offset
226                    .checked_add(length as u64)
227                    .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
228            }
229        })();
230        let close = file.close();
231        match operation {
232            Err(error) => Err(error),
233            Ok(()) => close,
234        }
235    }
236}
237
238impl<F: MetadataFileSystem> FsClient<F> {
239    pub fn metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
240        self.lock()?.metadata_path(path.as_ref(), true)
241    }
242
243    pub fn symlink_metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
244        self.lock()?.metadata_path(path.as_ref(), false)
245    }
246
247    pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
248        SetMetadata {
249            client: self,
250            path: path.as_ref().to_path_buf(),
251            update: MetadataUpdate::new(),
252            follow: true,
253        }
254    }
255}
256
257impl<F: DirectoryFileSystem> FsClient<F> {
258    pub fn create_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
259        self.create_dir_with_mode(path, 0o777)
260    }
261
262    pub fn create_dir_with_mode(&self, path: impl AsRef<Path>, mode: u32) -> VfResult<()> {
263        self.lock()?.create_dir_one(path.as_ref(), mode)
264    }
265
266    pub fn read_dir(&self, path: impl AsRef<Path>) -> VfResult<Vec<DirEntry>> {
267        self.read_dir_with_options(path, ReadDirOptions::default())
268    }
269
270    /// Read one directory with explicit entry and path-storage limits.
271    pub fn read_dir_with_options(
272        &self,
273        path: impl AsRef<Path>,
274        options: ReadDirOptions,
275    ) -> VfResult<Vec<DirEntry>> {
276        self.lock()?.read_dir_one(path.as_ref(), options)
277    }
278
279    /// Visit one directory incrementally without retaining all entries in the
280    /// application. The callback runs under the backend lock and must not
281    /// call this client or drop other files owned by it.
282    pub fn visit_dir(
283        &self,
284        path: impl AsRef<Path>,
285        callback: impl FnMut(DirEntry) -> VfResult<bool>,
286    ) -> VfResult<()> {
287        self.visit_dir_with_options(path, ReadDirOptions::default(), callback)
288    }
289
290    /// Visit entries with explicit entry and cumulative path-byte limits.
291    pub fn visit_dir_with_options(
292        &self,
293        path: impl AsRef<Path>,
294        options: ReadDirOptions,
295        mut callback: impl FnMut(DirEntry) -> VfResult<bool>,
296    ) -> VfResult<()> {
297        self.lock()?
298            .visit_dir_one(path.as_ref(), options, &mut callback)
299    }
300}
301
302impl<F: DirectoryFileSystem + MetadataFileSystem> FsClient<F> {
303    pub fn create_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
304        let path = path.as_ref();
305        let mut current = PathBuf::new();
306        for component in path.components() {
307            match component {
308                Component::RootDir => {
309                    current.push(Path::new("/"));
310                    continue;
311                }
312                Component::Normal(part) => current.push(part),
313                Component::CurDir => continue,
314                Component::ParentDir => {
315                    // Preserve `..` so the backend resolves it relative to
316                    // its own rooted namespace. Lexically popping `/` would
317                    // accidentally turn a later absolute component into a
318                    // current-directory-relative path.
319                    current.push("..");
320                    continue;
321                }
322                Component::Prefix(_) => return Err(VfError::client(0, crate::ERR_INVAL)),
323            }
324            match self.create_dir(&current) {
325                Ok(()) => {}
326                Err(error) if error.err_no() == crate::ERR_EXIST => {
327                    if !self.metadata(&current)?.is_dir() {
328                        return Err(VfError::client(0, crate::ERR_NOTDIR)
329                            .with_context("create_dir_all", &current));
330                    }
331                }
332                Err(error) => return Err(error),
333            }
334        }
335        Ok(())
336    }
337}
338
339impl<F: NamespaceFileSystem + MetadataFileSystem> FsClient<F> {
340    fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Metadata> {
341        let mut filesystem = self.lock()?;
342        let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
343        filesystem
344            .metadata_path(path, follow)
345            .map_err(|error| error.with_context(operation, path))
346    }
347
348    pub fn remove_file(&self, path: impl AsRef<Path>) -> VfResult<()> {
349        let path = path.as_ref();
350        if self.removal_metadata(path, "remove_file")?.is_dir() {
351            return Err(VfError::client(0, crate::ERR_ISDIR).with_context("remove_file", path));
352        }
353        self.lock()?.remove_one(path, false)
354    }
355
356    pub fn remove_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
357        let path = path.as_ref();
358        if !self.removal_metadata(path, "remove_dir")?.is_dir() {
359            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir", path));
360        }
361        self.lock()?.remove_one(path, false)
362    }
363
364    pub fn remove_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
365        let path = path.as_ref();
366        if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
367            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
368        }
369        self.lock()?.remove_one(path, true)
370    }
371
372    /// Remove the contents of a directory, keeping the directory itself.
373    pub fn remove_dir_contents(&self, path: impl AsRef<Path>) -> VfResult<()> {
374        let path = path.as_ref();
375        if !self.removal_metadata(path, "remove_dir_contents")?.is_dir() {
376            return Err(
377                VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
378            );
379        }
380        self.lock()?.remove_dir_contents(path)
381    }
382
383    pub fn rename(&self, from: impl AsRef<Path>, to: impl AsRef<Path>) -> VfResult<()> {
384        self.lock()?.rename_one(from.as_ref(), to.as_ref())
385    }
386}
387
388impl<F: VecFs> FsClient<F> {
389    /// List several directories with common stat attributes and finite
390    /// allocation limits. Use `read_dirs_with_options` for richer fields.
391    pub fn read_dirs<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<DirectoryListing>> {
392        self.read_dirs_with_options(paths, AttrMask::stat(), ReadDirOptions::default())
393    }
394
395    /// Recursively enumerate a bounded tree with common stat attributes.
396    /// Use `walk_with_options` to select fields or change limits.
397    pub fn walk(&self, root: impl AsRef<Path>) -> VfResult<Vec<DirectoryListing>> {
398        self.walk_with_options(root, AttrMask::stat(), crate::WalkOptions::default())
399    }
400
401    /// Create `path` if missing, otherwise empty it. Errors if it exists and is
402    /// not a directory (a symlink to a directory is not a directory here).
403    pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
404        self.lock()?.ensure_empty_dir(path.as_ref())
405    }
406
407    /// Open a directory for handle-rooted removal. The returned handle cannot
408    /// be swapped for a symlink between opening and removal; release it with
409    /// [`close_dir`](Self::close_dir).
410    pub fn open_dir(&self, path: impl AsRef<Path>) -> VfResult<VfDir> {
411        self.lock()?.open_dir(path.as_ref())
412    }
413
414    /// Remove the contents of a directory handle, keeping the directory.
415    pub fn remove_dir_contents_handle(&self, dir: &VfDir) -> VfResult<()> {
416        self.lock()?.rm_dir_contents(dir)
417    }
418
419    /// Release a handle from [`open_dir`](Self::open_dir).
420    pub fn close_dir(&self, dir: &VfDir) -> VfResult<()> {
421        self.lock()?.close_dir(dir)
422    }
423}
424
425impl<F: LinkFileSystem> FsClient<F> {
426    pub fn symlink(&self, target: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
427        self.lock()?.symlink_one(target.as_ref(), link.as_ref())
428    }
429
430    pub fn hard_link(&self, source: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
431        self.lock()?.hard_link_one(source.as_ref(), link.as_ref())
432    }
433
434    pub fn read_link(&self, path: impl AsRef<Path>) -> VfResult<PathBuf> {
435        self.lock()?.read_link_one(path.as_ref())
436    }
437}
438
439impl<F: CopyFileSystem> FsClient<F> {
440    pub fn copy(&self, source: impl AsRef<Path>, destination: impl AsRef<Path>) -> VfResult<()> {
441        self.lock()?.copy_one(source.as_ref(), destination.as_ref())
442    }
443}
444
445impl<F: VecFs> FsClient<F> {
446    /// Fetch selected metadata for one path without following its final symlink.
447    /// Unavailable fields remain `None` on [`Metadata`].
448    pub fn symlink_metadata_with_fields(
449        &self,
450        path: impl AsRef<Path>,
451        fields: AttrMask,
452    ) -> VfResult<Metadata> {
453        let path = path.as_ref();
454        let mut attrs = crate::VfAttrs {
455            file: VfFile::from_os_path(path),
456            masks: fields | AttrMask::MODE | AttrMask::SIZE,
457            ..crate::VfAttrs::default()
458        };
459        self.lock()?
460            .lgetattrsv(std::slice::from_mut(&mut attrs))
461            .map_err(|error| error.with_context("symlink_metadata", path))?;
462        Ok(attrs.into())
463    }
464
465    /// List multiple directories in a vector call. The limits apply to the
466    /// aggregate returned entries and stored path bytes. Streaming backends
467    /// apply these limits before collecting a full listing; a backend using
468    /// the compatibility `visit_dir` fallback may buffer one directory first.
469    /// An error discards the collected prefix; callers may retry individual
470    /// directories if desired.
471    pub fn read_dirs_with_options<P: AsRef<Path>>(
472        &self,
473        paths: &[P],
474        fields: AttrMask,
475        options: ReadDirOptions,
476    ) -> VfResult<Vec<DirectoryListing>> {
477        if paths.is_empty() {
478            return Ok(Vec::new());
479        }
480        let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
481        let mut positions: HashMap<PathBuf, Vec<usize>> = HashMap::with_capacity(paths.len());
482        let mut unique_paths = Vec::with_capacity(paths.len());
483        for (index, path) in paths.iter().enumerate() {
484            if let Some(indices) = positions.get_mut(*path) {
485                indices.push(index);
486            } else {
487                positions.insert(path.to_path_buf(), vec![index]);
488                unique_paths.push(*path);
489            }
490        }
491        let mut listings: Vec<DirectoryListing> = paths
492            .iter()
493            .map(|path| DirectoryListing {
494                path: path.to_path_buf(),
495                entries: Vec::new(),
496            })
497            .collect();
498        let mut entry_count = 0usize;
499        let mut path_bytes = 0usize;
500        // Bound the number of simultaneous first READDIR pages even when the
501        // caller supplies thousands of directory operands.
502        const DIRECTORY_COHORT: usize = 32;
503        for cohort in unique_paths.chunks(DIRECTORY_COHORT) {
504            let mut callback_error = None;
505            let requested = options
506                .entry_limit()
507                .saturating_sub(entry_count)
508                .saturating_add(1);
509            let result = self.lock()?.listdirv(
510                cohort,
511                fields | AttrMask::MODE | AttrMask::SIZE,
512                requested,
513                false,
514                &mut |attributes, directory| {
515                    let Some(indices) = positions.get(directory) else {
516                        callback_error = Some(VfError::transport(
517                            None,
518                            format!(
519                                "read_dirs backend returned an unexpected directory: {}",
520                                directory.display()
521                            ),
522                        ));
523                        return false;
524                    };
525                    if !cohort.contains(&directory) {
526                        callback_error = Some(VfError::transport(
527                            None,
528                            "read_dirs backend returned a directory outside the active cohort",
529                        ));
530                        return false;
531                    }
532                    let Some(path) = attributes.file.path() else {
533                        callback_error = Some(VfError::transport(
534                            None,
535                            "read_dirs backend returned an entry without a path",
536                        ));
537                        return false;
538                    };
539                    if path.parent() != Some(directory) {
540                        callback_error = Some(VfError::transport(
541                            None,
542                            format!(
543                                "read_dirs backend returned an entry outside {}",
544                                directory.display()
545                            ),
546                        ));
547                        return false;
548                    }
549                    for &index in indices {
550                        entry_count += 1;
551                        path_bytes = path_bytes.saturating_add(path.as_os_str().len());
552                        if entry_count > options.entry_limit()
553                            || path_bytes > options.path_byte_limit()
554                        {
555                            callback_error = Some(
556                                VfError::failure(index, libc::EFBIG as u32)
557                                    .with_context("read_dirs", directory),
558                            );
559                            return false;
560                        }
561                        listings[index]
562                            .entries
563                            .push(DirEntry::new(path.to_path_buf(), attributes.clone().into()));
564                    }
565                    true
566                },
567            );
568            if let Some(error) = callback_error {
569                return Err(error);
570            }
571            result.map_err(|error| {
572                error.map_index(|index| {
573                    cohort
574                        .get(index)
575                        .and_then(|path| positions.get(*path))
576                        .and_then(|indices| indices.first())
577                        .copied()
578                        .unwrap_or(index)
579                })
580            })?;
581        }
582        Ok(listings)
583    }
584
585    /// Recursively enumerate directories with selected entry attributes.
586    /// The walk is bounded by `options`; sorting and presentation remain the
587    /// application's responsibility.
588    pub fn walk_with_options(
589        &self,
590        root: impl AsRef<Path>,
591        fields: AttrMask,
592        options: crate::WalkOptions,
593    ) -> VfResult<Vec<DirectoryListing>> {
594        let root = root.as_ref();
595        let tree = self.lock()?.walk_with_options(
596            root,
597            fields | AttrMask::MODE | AttrMask::SIZE,
598            options,
599            &mut |_, _| {},
600        )?;
601        tree.into_iter()
602            .map(|directory| {
603                let entries = directory
604                    .entries
605                    .into_iter()
606                    .enumerate()
607                    .map(|(index, attributes)| {
608                        let path = attributes
609                            .file
610                            .path()
611                            .ok_or_else(|| {
612                                VfError::transport(
613                                    None,
614                                    format!("walk backend returned entry {index} without a path"),
615                                )
616                            })?
617                            .to_path_buf();
618                        Ok(DirEntry::new(path, attributes.into()))
619                    })
620                    .collect::<VfResult<Vec<_>>>()?;
621                Ok(DirectoryListing {
622                    path: directory.path,
623                    entries,
624                })
625            })
626            .collect()
627    }
628
629    /// Copy whole files in request order. A successful prefix may remain if
630    /// a later request fails; this operation does not provide atomicity.
631    pub fn copy_files<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
632        let extents: Vec<_> = pairs
633            .iter()
634            .map(|(from, to)| {
635                crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
636            })
637            .collect();
638        self.lock()?.copyv(&extents).map_err(|error| {
639            error
640                .index_opt()
641                .and_then(|index| pairs.get(index))
642                .map_or(error.clone(), |(_, to)| {
643                    error.with_context("copy_files", to.as_ref())
644                })
645        })
646    }
647
648    /// Remove paths in request order, optionally recursing into directories.
649    /// A successful prefix may remain if a later path fails.
650    pub fn remove_paths<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
651        let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
652        self.lock()?.rm(&paths, recursive).map_err(|error| {
653            error
654                .index_opt()
655                .and_then(|index| paths.get(index))
656                .map_or(error.clone(), |path| {
657                    error.with_context("remove_paths", path)
658                })
659        })
660    }
661
662    /// Fetch no-follow metadata for many paths using the backend's vector
663    /// operation. Useful for routing without one metadata RPC per path.
664    pub fn symlink_metadatav(&self, paths: &[&Path]) -> VfResult<Vec<Metadata>> {
665        let mut attrs: Vec<_> = paths
666            .iter()
667            .map(|path| crate::VfAttrs {
668                file: VfFile::from_os_path(path),
669                masks: AttrMask::MODE,
670                ..crate::VfAttrs::default()
671            })
672            .collect();
673        self.lock()?.lgetattrsv(&mut attrs).map_err(|error| {
674            error
675                .index_opt()
676                .and_then(|index| paths.get(index))
677                .map_or(error.clone(), |path| {
678                    error.with_context("symlink_metadatav", path)
679                })
680        })?;
681        Ok(attrs.into_iter().map(Metadata::from).collect())
682    }
683}
684
685impl<F: VectorFileSystem> FsClient<F> {
686    /// Open an ordered vector of files.
687    ///
688    /// Success returns one RAII handle per request. Failure returns no
689    /// handles; VFSI does not promise transactional rollback of other
690    /// filesystem effects such as file creation.
691    pub fn openv(&self, requests: &[OpenRequest]) -> VfResult<Vec<FsFile<F>>> {
692        let mut filesystem = self.lock()?;
693        let files = filesystem.open_many(requests).map_err(|error| {
694            error
695                .index_opt()
696                .and_then(|index| requests.get(index))
697                .map_or(error.clone(), |request| {
698                    error.with_context("openv", &request.path)
699                })
700        })?;
701        if files.len() != requests.len() {
702            let error = wrong_result_count("openv", requests.len(), files.len());
703            for file in &files {
704                let _ = filesystem.close_one(file);
705            }
706            return Err(error);
707        }
708        drop(filesystem);
709        Ok(files
710            .into_iter()
711            .zip(requests)
712            .map(|(file, request)| FsFile {
713                inner: Arc::clone(&self.inner),
714                file: Some(file),
715                path: request.path.clone(),
716            })
717            .collect())
718    }
719
720    /// Try to close a group through one vector operation without consuming
721    /// the handles. On failure, all handles remain armed: the backend may
722    /// have closed a prefix, so callers must reconcile before retrying.
723    pub fn try_closev(&self, files: &mut [FsFile<F>]) -> VfResult<()> {
724        for (index, file) in files.iter().enumerate() {
725            self.validate_owner(file, index)?;
726        }
727        let descriptors: Vec<VfFile> = files
728            .iter()
729            .map(|file| file.raw().cloned())
730            .collect::<VfResult<_>>()?;
731        self.lock()?.close_many(&descriptors).map_err(|error| {
732            error
733                .index_opt()
734                .and_then(|index| files.get(index))
735                .map_or(error.clone(), |file| {
736                    error.with_context("closev", file.path())
737                })
738        })?;
739        for file in files.iter_mut() {
740            file.file = None;
741        }
742        Ok(())
743    }
744
745    /// Close a group of files through one vector operation.
746    ///
747    /// On failure, the handles are dropped and the backend receives
748    /// best-effort scalar cleanup attempts. Use [`try_closev`](Self::try_closev)
749    /// to retain the handles after an error.
750    pub fn closev(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
751        self.try_closev(&mut files)
752    }
753
754    /// Read an ordered vector with a 16 MiB aggregate request limit.
755    /// Use [`readv_with_limit`](Self::readv_with_limit) to tune the limit or
756    /// [`readv_into`](Self::readv_into) to provide bounded caller-owned buffers.
757    pub fn readv(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<ReadResult>> {
758        self.readv_with_limit(requests, DEFAULT_READV_MAX_TOTAL_BYTES)
759    }
760
761    /// Read an ordered vector with an explicit aggregate request limit.
762    pub fn readv_with_limit(
763        &self,
764        requests: &[FsRead<'_, F>],
765        max_total_bytes: usize,
766    ) -> VfResult<Vec<ReadResult>> {
767        let mut requested = 0usize;
768        for (index, request) in requests.iter().enumerate() {
769            requested = requested
770                .checked_add(request.length)
771                .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
772            if requested > max_total_bytes {
773                return Err(VfError::client(index, libc::EFBIG as u32)
774                    .with_context("readv", request.file.path()));
775            }
776        }
777        let reads = self.read_ops(requests)?;
778        let results = self.lock()?.read_many(&reads).map_err(|error| {
779            error
780                .index_opt()
781                .and_then(|index| requests.get(index))
782                .map_or(error.clone(), |request| {
783                    error.with_context("readv", request.file.path())
784                })
785        })?;
786        if results.len() != requests.len() {
787            return Err(wrong_result_count("readv", requests.len(), results.len()));
788        }
789        validate_read_results("readv", &reads, &results).map_err(|error| {
790            error
791                .index_opt()
792                .and_then(|index| requests.get(index))
793                .map_or(error.clone(), |request| {
794                    error.with_context("readv", request.file.path())
795                })
796        })?;
797        Ok(results)
798    }
799
800    fn read_ops(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<ReadOp>> {
801        let mut reads = Vec::with_capacity(requests.len());
802        for (index, request) in requests.iter().enumerate() {
803            self.validate_owner(request.file, index)?;
804            reads.push(ReadOp::new(
805                request.file.raw()?.clone(),
806                request.offset,
807                request.length,
808            ));
809        }
810        Ok(reads)
811    }
812
813    pub fn readv_into(&self, requests: &mut [FsReadInto<'_, F>]) -> VfResult<Vec<usize>> {
814        let mut reads = Vec::with_capacity(requests.len());
815        for (index, request) in requests.iter().enumerate() {
816            self.validate_owner(request.file, index)?;
817            reads.push(ReadOp::new(
818                request.file.raw()?.clone(),
819                request.offset,
820                request.buffer.len(),
821            ));
822        }
823        let results = {
824            let mut buffers: Vec<&mut [u8]> = requests
825                .iter_mut()
826                .map(|request| &mut *request.buffer)
827                .collect();
828            self.lock()?.read_many_into(&reads, &mut buffers)
829        }
830        .map_err(|error| {
831            error
832                .index_opt()
833                .and_then(|index| requests.get(index))
834                .map_or(error.clone(), |request| {
835                    error.with_context("readv_into", request.file.path())
836                })
837        })?;
838        validate_read_into_results("readv_into", &reads, &results).map_err(|error| {
839            error
840                .index_opt()
841                .and_then(|index| requests.get(index))
842                .map_or(error.clone(), |request| {
843                    error.with_context("readv_into", request.file.path())
844                })
845        })?;
846        Ok(results.into_iter().map(|result| result.read).collect())
847    }
848
849    pub fn writev(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<WriteResult>> {
850        let writes = self.write_ops(requests)?;
851        let results = self.lock()?.write_many(&writes).map_err(|error| {
852            error
853                .index_opt()
854                .and_then(|index| requests.get(index))
855                .map_or(error.clone(), |request| {
856                    error.with_context("writev", request.file.path())
857                })
858        })?;
859        if results.len() != requests.len() {
860            return Err(wrong_result_count("writev", requests.len(), results.len()));
861        }
862        validate_write_results("writev", &writes, &results).map_err(|error| {
863            error
864                .index_opt()
865                .and_then(|index| requests.get(index))
866                .map_or(error.clone(), |request| {
867                    error.with_context("writev", request.file.path())
868                })
869        })?;
870        Ok(results)
871    }
872
873    /// Write every byte in each positional request, retrying short writes in
874    /// vector waves. Like `writev`, this is not transactional: an error may
875    /// follow a successfully written prefix. Overlapping requests through the
876    /// same path complete in input order; different paths are presumed
877    /// independent (including hard-link aliases).
878    pub fn write_allv(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<WriteResult>> {
879        // Validate the entire batch before writing any prefix. In particular,
880        // empty requests must not conceal a foreign or already-closed file.
881        let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
882        let mut blocked_by = vec![Vec::new(); requests.len()];
883        for (index, request) in requests.iter().enumerate() {
884            self.validate_owner(request.file, index)?;
885            request.file.raw()?;
886            let VfOffset::At(offset) = request.offset else {
887                return Err(VfError::client(index, crate::ERR_INVAL)
888                    .with_context("write_allv", request.file.path()));
889            };
890            let end = offset
891                .checked_add(request.data.len() as u64)
892                .ok_or_else(|| {
893                    VfError::client(index, libc::EOVERFLOW as u32)
894                        .with_context("write_allv", request.file.path())
895                })?;
896            // Non-overlapping writes commute. An overlapping later request
897            // must wait until every earlier conflicting request is complete:
898            // otherwise a short-write retry can overwrite the later bytes.
899            let prior = prior_by_path.entry(request.file.path()).or_default();
900            for &(earlier, start, earlier_end) in prior.iter() {
901                if offset < earlier_end && start < end {
902                    blocked_by[index].push(earlier);
903                }
904            }
905            prior.push((index, offset, end));
906        }
907        let mut totals = vec![0usize; requests.len()];
908        let mut stable = vec![true; requests.len()];
909        let mut pending: Vec<usize> = (0..requests.len())
910            .filter(|&index| !requests[index].data.is_empty())
911            .collect();
912        while !pending.is_empty() {
913            let wave_indices: Vec<usize> = pending
914                .iter()
915                .copied()
916                .filter(|&index| {
917                    blocked_by[index]
918                        .iter()
919                        .all(|&earlier| totals[earlier] == requests[earlier].data.len())
920                })
921                .collect();
922            let wave: Vec<_> = wave_indices
923                .iter()
924                .map(|&index| {
925                    let request = &requests[index];
926                    let VfOffset::At(offset) = request.offset else {
927                        return Err(VfError::client(index, crate::ERR_INVAL)
928                            .with_context("write_allv", request.file.path()));
929                    };
930                    let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
931                        VfError::client(index, libc::EOVERFLOW as u32)
932                            .with_context("write_allv", request.file.path())
933                    })?;
934                    Ok(FsWrite {
935                        file: request.file,
936                        offset: VfOffset::At(offset),
937                        data: &request.data[totals[index]..],
938                    })
939                })
940                .collect::<VfResult<_>>()?;
941            let results = self.writev(&wave).map_err(|error| {
942                error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
943            })?;
944            for (&index, result) in wave_indices.iter().zip(results) {
945                if result.written == 0 {
946                    return Err(VfError::client(index, crate::ERR_IO)
947                        .with_context("write_allv", requests[index].file.path()));
948                }
949                totals[index] += result.written;
950                stable[index] &= result.stable;
951            }
952            pending.retain(|&index| totals[index] < requests[index].data.len());
953        }
954        requests
955            .iter()
956            .enumerate()
957            .map(|(index, request)| {
958                let VfOffset::At(offset) = request.offset else {
959                    return Err(VfError::client(index, crate::ERR_INVAL)
960                        .with_context("write_allv", request.file.path()));
961                };
962                Ok(WriteResult {
963                    file: request.file.raw()?.clone(),
964                    offset,
965                    written: totals[index],
966                    stable: stable[index],
967                })
968            })
969            .collect()
970    }
971
972    fn write_ops<'a>(&self, requests: &'a [FsWrite<'a, F>]) -> VfResult<Vec<WriteOpRef<'a>>> {
973        let mut writes = Vec::with_capacity(requests.len());
974        for (index, request) in requests.iter().enumerate() {
975            self.validate_owner(request.file, index)?;
976            writes.push(WriteOpRef::new(
977                request.file.raw()?,
978                request.offset,
979                request.data,
980            ));
981        }
982        Ok(writes)
983    }
984
985    fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
986        if Arc::ptr_eq(&self.inner, &file.inner) {
987            Ok(())
988        } else {
989            Err(VfError::client(index, crate::ERR_INVAL))
990        }
991    }
992}
993
994impl<F: VectorFileSystem + VecFs> FsClient<F> {
995    /// Read several complete files by path using vector READ operations.
996    ///
997    /// The aggregate returned data is limited to 16 MiB by default. Use
998    /// [`read_files_with_options`](Self::read_files_with_options) to choose a
999    /// different limit, or stream large files instead. This is not a snapshot
1000    /// or an atomic operation across files.
1001    pub fn read_files<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<Vec<u8>>> {
1002        self.read_files_with_options(paths, ReadAllOptions::default())
1003    }
1004
1005    /// Read several complete files with an explicit aggregate allocation limit.
1006    pub fn read_files_with_options<P: AsRef<Path>>(
1007        &self,
1008        paths: &[P],
1009        options: ReadAllOptions,
1010    ) -> VfResult<Vec<Vec<u8>>> {
1011        let files: Vec<_> = paths
1012            .iter()
1013            .map(|path| VfFile::from_os_path(path.as_ref()))
1014            .collect();
1015        let result = self
1016            .lock()?
1017            .read_allv_with_options(&files, options)
1018            .map_err(|error| {
1019                error
1020                    .index_opt()
1021                    .and_then(|index| paths.get(index))
1022                    .map_or(error.clone(), |path| {
1023                        error.with_context("read_files", path.as_ref())
1024                    })
1025            });
1026        result.and_then(|buffers| {
1027            if buffers.len() != paths.len() {
1028                Err(wrong_result_count("read_files", paths.len(), buffers.len()))
1029            } else {
1030                Ok(buffers)
1031            }
1032        })
1033    }
1034
1035    /// Replace several files from borrowed buffers using vector OPEN, WRITE,
1036    /// and CLOSE phases. Identical path spellings are rejected before opening
1037    /// anything; aliases such as hard links are still the caller's responsibility.
1038    /// The batch is not transactional: an error may follow files already
1039    /// created or written. Large inputs should be chunked by the caller rather
1040    /// than held in memory solely for this convenience method.
1041    pub fn write_files<P: AsRef<Path>, B: AsRef<[u8]>>(&self, entries: &[(P, B)]) -> VfResult<()> {
1042        let mut seen = HashSet::with_capacity(entries.len());
1043        for (index, (path, _)) in entries.iter().enumerate() {
1044            if !seen.insert(path.as_ref()) {
1045                return Err(VfError::client(index, crate::ERR_INVAL)
1046                    .with_context("write_files", path.as_ref()));
1047            }
1048        }
1049        let requests: Vec<_> = entries
1050            .iter()
1051            .map(|(path, _)| {
1052                OpenRequest::new(
1053                    path.as_ref(),
1054                    OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
1055                )
1056            })
1057            .collect();
1058        let files = self.openv(&requests)?;
1059        let writes: Vec<_> = files
1060            .iter()
1061            .zip(entries)
1062            .map(|(file, (_, data))| file.write_request_at(0, data.as_ref()))
1063            .collect();
1064        let result = self.write_allv(&writes);
1065        drop(writes);
1066        let close_result = self.closev(files);
1067        result?;
1068        close_result
1069    }
1070}
1071
1072/// `std::fs::OpenOptions`-style builder tied to an [`FsClient`].
1073pub struct OpenOptions<'a, F: FileSystem> {
1074    client: &'a FsClient<F>,
1075    flags: OpenFlags,
1076    mode: u32,
1077}
1078
1079impl<F: FileSystem> Clone for OpenOptions<'_, F> {
1080    fn clone(&self) -> Self {
1081        Self {
1082            client: self.client,
1083            flags: self.flags,
1084            mode: self.mode,
1085        }
1086    }
1087}
1088
1089impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
1090    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1091        formatter
1092            .debug_struct("OpenOptions")
1093            .field("flags", &self.flags)
1094            .field("mode", &format_args!("{:#o}", self.mode))
1095            .finish()
1096    }
1097}
1098
1099impl<'a, F: FileSystem> OpenOptions<'a, F> {
1100    fn new(client: &'a FsClient<F>) -> Self {
1101        Self {
1102            client,
1103            flags: OpenFlags::empty(),
1104            mode: 0o666,
1105        }
1106    }
1107
1108    pub fn read(&mut self, enabled: bool) -> &mut Self {
1109        self.flags.set(OpenFlags::READ, enabled);
1110        self
1111    }
1112
1113    pub fn write(&mut self, enabled: bool) -> &mut Self {
1114        self.flags.set(OpenFlags::WRITE, enabled);
1115        self
1116    }
1117
1118    pub fn append(&mut self, enabled: bool) -> &mut Self {
1119        self.flags.set(OpenFlags::APPEND, enabled);
1120        self
1121    }
1122
1123    pub fn truncate(&mut self, enabled: bool) -> &mut Self {
1124        self.flags.set(OpenFlags::TRUNCATE, enabled);
1125        self
1126    }
1127
1128    pub fn create(&mut self, enabled: bool) -> &mut Self {
1129        self.flags.set(OpenFlags::CREATE, enabled);
1130        self
1131    }
1132
1133    pub fn create_new(&mut self, enabled: bool) -> &mut Self {
1134        self.flags.set(OpenFlags::CREATE_NEW, enabled);
1135        self
1136    }
1137
1138    pub fn mode(&mut self, mode: u32) -> &mut Self {
1139        self.mode = mode;
1140        self
1141    }
1142
1143    pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
1144        self.client
1145            .open_with(OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
1146    }
1147}
1148
1149impl<F: VectorFileSystem> OpenOptions<'_, F> {
1150    pub fn openv<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
1151        let requests: Vec<OpenRequest> = paths
1152            .iter()
1153            .map(|path| OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
1154            .collect();
1155        self.client.openv(&requests)
1156    }
1157}
1158
1159/// Builder for an atomic metadata update request.
1160pub struct SetMetadata<'a, F: MetadataFileSystem> {
1161    client: &'a FsClient<F>,
1162    path: PathBuf,
1163    update: MetadataUpdate,
1164    follow: bool,
1165}
1166
1167impl<F: MetadataFileSystem> Clone for SetMetadata<'_, F> {
1168    fn clone(&self) -> Self {
1169        Self {
1170            client: self.client,
1171            path: self.path.clone(),
1172            update: self.update.clone(),
1173            follow: self.follow,
1174        }
1175    }
1176}
1177
1178impl<F: MetadataFileSystem> fmt::Debug for SetMetadata<'_, F> {
1179    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1180        formatter
1181            .debug_struct("SetMetadata")
1182            .field("path", &self.path)
1183            .field("update", &self.update)
1184            .field("follow", &self.follow)
1185            .finish_non_exhaustive()
1186    }
1187}
1188
1189impl<F: MetadataFileSystem> SetMetadata<'_, F> {
1190    pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
1191        self.update.permissions = Some(permissions);
1192        self
1193    }
1194
1195    pub fn len(&mut self, len: u64) -> &mut Self {
1196        self.update.len = Some(len);
1197        self
1198    }
1199
1200    pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
1201        self.update.accessed = Some(accessed);
1202        self
1203    }
1204
1205    pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
1206        self.update.modified = Some(modified);
1207        self
1208    }
1209
1210    pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
1211        self.follow = follow;
1212        self
1213    }
1214
1215    pub fn apply(&self) -> VfResult<()> {
1216        self.client
1217            .lock()?
1218            .set_metadata_path(&self.path, self.update.clone(), self.follow)
1219    }
1220}
1221
1222/// Owned RAII file which does not borrow the client.
1223///
1224/// Dropping an open file attempts a best-effort CLOSE. This can block on the
1225/// backend mutex and on network I/O; use [`FsFile::try_close`] when the close
1226/// result matters or when its timing must be controlled.
1227pub struct FsFile<F: FileSystem> {
1228    inner: Arc<Mutex<F>>,
1229    file: Option<VfFile>,
1230    path: PathBuf,
1231}
1232
1233impl<F: FileSystem> fmt::Debug for FsFile<F> {
1234    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1235        formatter
1236            .debug_struct("FsFile")
1237            .field("path", &self.path)
1238            .field("is_open", &self.file.is_some())
1239            .finish()
1240    }
1241}
1242
1243impl<F: FileSystem> FsFile<F> {
1244    fn raw(&self) -> VfResult<&VfFile> {
1245        self.file
1246            .as_ref()
1247            .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
1248    }
1249
1250    /// Positional read which does not alter the file cursor.
1251    pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
1252        self.read_into(buffer, VfOffset::At(offset))
1253    }
1254
1255    /// Positional write which does not alter the file cursor.
1256    pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
1257        self.write_from(buffer, VfOffset::At(offset))
1258    }
1259
1260    pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
1261        FsRead {
1262            file: self,
1263            offset: VfOffset::At(offset),
1264            length,
1265        }
1266    }
1267
1268    pub fn path(&self) -> &Path {
1269        &self.path
1270    }
1271
1272    /// Query metadata for the open object without resolving its path again.
1273    pub fn metadata(&self) -> VfResult<Metadata> {
1274        let attributes = AttrMask::MODE
1275            | AttrMask::SIZE
1276            | AttrMask::NLINK
1277            | AttrMask::FILEID
1278            | AttrMask::UID
1279            | AttrMask::GID
1280            | AttrMask::ATIME
1281            | AttrMask::MTIME
1282            | AttrMask::CTIME
1283            | AttrMask::CHANGE;
1284        self.inner
1285            .lock()
1286            .map_err(|_| poisoned())?
1287            .metadata(MetadataQuery::new(self.raw()?.clone(), attributes))
1288            .map(Metadata::from)
1289            .map_err(|error| error.with_context("metadata", &self.path))
1290    }
1291
1292    /// Truncate or extend the open file.
1293    pub fn set_len(&self, len: u64) -> VfResult<()> {
1294        let mut update = SetAttributes::new(self.raw()?.clone());
1295        update.size = Some(len);
1296        self.inner
1297            .lock()
1298            .map_err(|_| poisoned())?
1299            .set_attributes(update)
1300            .map_err(|error| error.with_context("set_len", &self.path))
1301    }
1302
1303    /// Change permissions on the open file.
1304    pub fn set_permissions(&self, permissions: Permissions) -> VfResult<()> {
1305        let mut update = SetAttributes::new(self.raw()?.clone());
1306        update.mode = Some(permissions.mode());
1307        self.inner
1308            .lock()
1309            .map_err(|_| poisoned())?
1310            .set_attributes(update)
1311            .map_err(|error| error.with_context("set_permissions", &self.path))
1312    }
1313
1314    pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
1315        FsWrite {
1316            file: self,
1317            offset: VfOffset::At(offset),
1318            data,
1319        }
1320    }
1321
1322    pub fn read_request_at_into<'a>(
1323        &'a self,
1324        offset: u64,
1325        buffer: &'a mut [u8],
1326    ) -> FsReadInto<'a, F> {
1327        FsReadInto {
1328            file: self,
1329            offset: VfOffset::At(offset),
1330            buffer,
1331        }
1332    }
1333
1334    pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
1335        self.read_into(buffer, VfOffset::Cur)
1336    }
1337
1338    pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
1339        self.write_from(buffer, VfOffset::Cur)
1340    }
1341
1342    fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
1343        if buffer.is_empty() {
1344            return Ok(0);
1345        }
1346        let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
1347        let result = self
1348            .inner
1349            .lock()
1350            .map_err(|_| poisoned())?
1351            .read_one_into(&request, buffer)
1352            .map_err(|error| error.with_context("read", &self.path))?;
1353        validate_read_into_results(
1354            "read",
1355            std::slice::from_ref(&request),
1356            std::slice::from_ref(&result),
1357        )
1358        .map_err(|error| error.with_context("read", &self.path))?;
1359        Ok(result.read)
1360    }
1361
1362    fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
1363        if buffer.is_empty() {
1364            return Ok(0);
1365        }
1366        let file = self.raw()?.clone();
1367        let result = self
1368            .inner
1369            .lock()
1370            .map_err(|_| poisoned())?
1371            .write_one(WriteOpRef::new(&file, offset, buffer))
1372            .map_err(|error| error.with_context("write", &self.path))?;
1373        if result.written > buffer.len() {
1374            return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
1375        }
1376        Ok(result.written)
1377    }
1378
1379    pub fn sync_data(&self) -> VfResult<()> {
1380        self.inner
1381            .lock()
1382            .map_err(|_| poisoned())?
1383            .sync_data(self.raw()?)
1384            .map_err(|error| error.with_context("sync_data", &self.path))
1385    }
1386
1387    pub fn sync_all(&self) -> VfResult<()> {
1388        self.inner
1389            .lock()
1390            .map_err(|_| poisoned())?
1391            .sync_all(self.raw()?)
1392            .map_err(|error| error.with_context("sync_all", &self.path))
1393    }
1394
1395    /// Attempt to close without consuming the handle. A failed close leaves
1396    /// the handle armed so the caller can reconcile or retry it explicitly.
1397    pub fn try_close(&mut self) -> VfResult<()> {
1398        let Some(file) = self.file.as_ref() else {
1399            return Ok(());
1400        };
1401        self.inner
1402            .lock()
1403            .map_err(|_| poisoned())?
1404            .close_one(file)
1405            .map_err(|error| error.with_context("close", &self.path))?;
1406        self.file = None;
1407        Ok(())
1408    }
1409
1410    /// Consume and close the handle. On failure, `Drop` makes one best-effort
1411    /// cleanup attempt; use [`try_close`](Self::try_close) to retain control.
1412    pub fn close(mut self) -> VfResult<()> {
1413        self.try_close()
1414    }
1415
1416    /// Seek while retaining [`VfError`] protocol and path information.
1417    pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
1418        self.inner
1419            .lock()
1420            .map_err(|_| poisoned())?
1421            .seek_one(self.raw()?, position)
1422            .map_err(|error| error.with_context("seek", &self.path))
1423    }
1424}
1425
1426impl<F: FileSystem> Read for FsFile<F> {
1427    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
1428        self.read_native(buffer).map_err(io_error)
1429    }
1430}
1431
1432impl<F: FileSystem> Write for FsFile<F> {
1433    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
1434        self.write_native(buffer).map_err(io_error)
1435    }
1436
1437    fn flush(&mut self) -> io::Result<()> {
1438        self.sync_data().map_err(io_error)
1439    }
1440}
1441
1442impl<F: FileSystem> Seek for FsFile<F> {
1443    fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
1444        self.seek_native(position).map_err(io_error)
1445    }
1446}
1447
1448/// Typed read request for [`FsClient::readv`].
1449pub struct FsRead<'a, F: FileSystem> {
1450    file: &'a FsFile<F>,
1451    offset: VfOffset,
1452    length: usize,
1453}
1454
1455/// Typed borrowed write request for [`FsClient::writev`].
1456pub struct FsWrite<'a, F: FileSystem> {
1457    file: &'a FsFile<F>,
1458    offset: VfOffset,
1459    data: &'a [u8],
1460}
1461
1462/// Typed vector read into caller-provided storage.
1463pub struct FsReadInto<'a, F: FileSystem> {
1464    file: &'a FsFile<F>,
1465    offset: VfOffset,
1466    buffer: &'a mut [u8],
1467}
1468
1469impl<F: FileSystem> Drop for FsFile<F> {
1470    fn drop(&mut self) {
1471        if let Some(file) = self.file.take()
1472            && let Ok(mut filesystem) = self.inner.lock()
1473        {
1474            let _ = filesystem.close_one(&file);
1475        }
1476    }
1477}