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::{Path, PathBuf};
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::{Arc, Mutex};
9use std::time::SystemTime;
10
11use vfsi_core::api::internal::OwnedReadResult as FsReadResult;
12use vfsi_core::api::{
13    DirectoryListing, ReadIntoResult as FsReadIntoResult, ResourceLimits, StreamCompletion,
14    WriteResult as FsWriteResult,
15};
16
17use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
18use crate::{
19    AttrMask, Attrs, Backend, Capabilities, DirEntry, FileSystem, MetadataQuery, MetadataUpdate,
20    OpenFlags, OpenOp, Permissions, ReadAllOptions, ReadOp, ReadResult, RemoveOptions,
21    StreamOptions, VfDir, VfError, VfFile, VfOffset, VfResult, WriteOpRef, WriteResult,
22};
23
24fn read_result(result: ReadResult) -> FsReadResult {
25    FsReadResult {
26        offset: result.offset,
27        data: result.data,
28        eof: result.eof,
29    }
30}
31
32fn write_result(result: WriteResult) -> FsWriteResult {
33    FsWriteResult {
34        offset: result.offset,
35        written: result.written,
36        stable: result.stable,
37    }
38}
39
40fn io_error(error: VfError) -> io::Error {
41    error.into()
42}
43
44fn poisoned() -> VfError {
45    VfError::client(0, crate::ERR_IO)
46}
47
48fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
49    VfError::transport_with_kind(
50        None,
51        crate::TransportKind::InvalidReply,
52        format!("{operation} backend returned {actual} results for {expected} requests"),
53    )
54}
55
56enum CleanupTarget {
57    File(VfFile, PathBuf),
58    Directory(VfDir, PathBuf),
59}
60struct PendingClose<F> {
61    target: CleanupTarget,
62    close: fn(&mut F, &CleanupTarget) -> VfResult<()>,
63}
64
65/// The cleanup queue has a separate short-held lock: handle Drop never waits
66/// for an RPC or for the backend mutex. Final-owner teardown remains synchronous.
67struct SharedBackend<F: FileSystem> {
68    backend: Mutex<Option<F>>,
69    pending: Mutex<Vec<PendingClose<F>>>,
70    has_pending: AtomicBool,
71}
72impl<F: FileSystem> SharedBackend<F> {
73    fn new(backend: F) -> Self {
74        Self {
75            backend: Mutex::new(Some(backend)),
76            pending: Mutex::new(Vec::new()),
77            has_pending: AtomicBool::new(false),
78        }
79    }
80    fn defer(&self, target: CleanupTarget, close: fn(&mut F, &CleanupTarget) -> VfResult<()>) {
81        self.pending
82            .lock()
83            .unwrap_or_else(|e| e.into_inner())
84            .push(PendingClose { target, close });
85        self.has_pending.store(true, Ordering::Release);
86    }
87    fn lock_without_cleanup(&self) -> Result<BackendGuard<'_, F>, ()> {
88        self.backend
89            .lock()
90            .map(|guard| BackendGuard {
91                guard: Some(guard),
92                pending: &self.pending,
93                has_pending: &self.has_pending,
94            })
95            .map_err(|_| ())
96    }
97    fn lock(&self) -> Result<BackendGuard<'_, F>, ()> {
98        let mut guard = self.lock_without_cleanup()?;
99        // Ordinary I/O must not lose its result to an unrelated cleanup error.
100        // Retain failures; drain_cleanup is the explicit reporting boundary.
101        let _ = guard.drain_cleanup();
102        Ok(guard)
103    }
104}
105impl<F: FileSystem> Drop for SharedBackend<F> {
106    fn drop(&mut self) {
107        if let Some(backend) = self
108            .backend
109            .get_mut()
110            .unwrap_or_else(|e| e.into_inner())
111            .as_mut()
112        {
113            let pending = self.pending.get_mut().unwrap_or_else(|e| e.into_inner());
114            for close in pending.drain(..) {
115                let _ = (close.close)(backend, &close.target);
116            }
117            let callbacks = backend.take_notifications();
118            for callback in callbacks {
119                let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
120            }
121        }
122    }
123}
124struct BackendGuard<'a, F: FileSystem> {
125    guard: Option<std::sync::MutexGuard<'a, Option<F>>>,
126    pending: &'a Mutex<Vec<PendingClose<F>>>,
127    has_pending: &'a AtomicBool,
128}
129impl<F: FileSystem> std::ops::Deref for BackendGuard<'_, F> {
130    type Target = F;
131    fn deref(&self) -> &F {
132        self.guard
133            .as_ref()
134            .expect("live guard")
135            .as_ref()
136            .expect("live backend")
137    }
138}
139impl<F: FileSystem> std::ops::DerefMut for BackendGuard<'_, F> {
140    fn deref_mut(&mut self) -> &mut F {
141        self.guard
142            .as_mut()
143            .expect("live guard")
144            .as_mut()
145            .expect("live backend")
146    }
147}
148impl<F: FileSystem> BackendGuard<'_, F> {
149    fn drain_cleanup(&mut self) -> VfResult<()> {
150        if !self.has_pending.swap(false, Ordering::AcqRel) {
151            return Ok(());
152        }
153        let pending = std::mem::take(&mut *self.pending.lock().unwrap_or_else(|e| e.into_inner()));
154        let mut failed = Vec::new();
155        let mut first = None;
156        for close in pending {
157            if let Err(error) = (close.close)(self, &close.target) {
158                first.get_or_insert(error);
159                failed.push(close);
160            }
161        }
162        if !failed.is_empty() {
163            self.pending
164                .lock()
165                .unwrap_or_else(|e| e.into_inner())
166                .extend(failed);
167            self.has_pending.store(true, Ordering::Release);
168        }
169        first.map_or(Ok(()), Err)
170    }
171}
172impl<F: FileSystem> Drop for BackendGuard<'_, F> {
173    fn drop(&mut self) {
174        let callbacks = self.take_notifications();
175        drop(self.guard.take());
176        for callback in callbacks {
177            // Observability must not turn successful I/O into a panic or lose
178            // ownership of a newly opened descriptor before its RAII wrapping.
179            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
180        }
181    }
182}
183
184/// Cloneable owner of one synchronous backend connection.
185/// Clones share a mutex and serialize backend calls; use separate connections
186/// for parallel RPCs. Explicitly close handles to observe cleanup failures.
187/// Handle Drop queues cleanup, drained before later operations or by
188/// `drain_cleanup`. Dropping the final backend owner performs synchronous
189/// teardown and may wait for pending closes and backend request timeouts.
190pub struct FsClient<F: FileSystem> {
191    inner: Arc<SharedBackend<F>>,
192    limits: ResourceLimits,
193}
194
195impl<F: FileSystem> Clone for FsClient<F> {
196    fn clone(&self) -> Self {
197        Self {
198            inner: Arc::clone(&self.inner),
199            limits: self.limits,
200        }
201    }
202}
203
204impl<F: FileSystem> fmt::Debug for FsClient<F> {
205    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
206        formatter.debug_struct("FsClient").finish_non_exhaustive()
207    }
208}
209
210impl<F: FileSystem> FsClient<F> {
211    pub fn new(filesystem: F) -> Self {
212        Self {
213            inner: Arc::new(SharedBackend::new(filesystem)),
214            limits: ResourceLimits::default(),
215        }
216    }
217
218    /// Configure this client view and its future clones. Existing clones keep
219    /// their policy; all views still share the same connection and ownership.
220    pub fn with_limits(mut self, limits: ResourceLimits) -> Self {
221        self.limits = limits;
222        self
223    }
224
225    pub fn limits(&self) -> ResourceLimits {
226        self.limits
227    }
228
229    fn lock(&self) -> VfResult<BackendGuard<'_, F>> {
230        self.inner.lock().map_err(|_| poisoned())
231    }
232
233    pub fn into_inner(self) -> Result<F, Self> {
234        if self.drain_cleanup().is_err() {
235            return Err(self);
236        }
237        match Arc::try_unwrap(self.inner) {
238            Ok(mut shared) => Ok(shared
239                .backend
240                .get_mut()
241                .unwrap_or_else(|e| e.into_inner())
242                .take()
243                .expect("owned backend")),
244            Err(inner) => Err(Self {
245                inner,
246                limits: self.limits,
247            }),
248        }
249    }
250
251    /// Drain queued Drop cleanup and report the first failure. Failed targets
252    /// remain owned for later cleanup. No new file operation is replayed.
253    pub fn drain_cleanup(&self) -> VfResult<()> {
254        self.inner
255            .lock_without_cleanup()
256            .map_err(|_| poisoned())?
257            .drain_cleanup()
258    }
259}
260
261impl<F: FileSystem> FsClient<F> {
262    pub(crate) fn capabilities(&self) -> VfResult<Capabilities> {
263        Ok(self.lock()?.capabilities())
264    }
265    /// Open a path read-only.
266    pub(crate) fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
267        self.open_with_native(OpenOp::new(path.as_ref(), OpenFlags::READ))
268    }
269
270    pub(crate) fn open_with_native(&self, request: OpenOp) -> VfResult<FsFile<F>> {
271        let file = self.lock()?.open_impl(&request)?;
272        Ok(FsFile {
273            inner: Arc::clone(&self.inner),
274            file: Some(file),
275            path: request.path,
276        })
277    }
278
279    pub fn open_options(&self) -> OpenOptions<'_, F> {
280        OpenOptions::new(self)
281    }
282}
283
284impl<F: FileSystem> FsClient<F> {
285    /// Stream one file using an explicit maximum chunk size.
286    ///
287    /// The callback runs without holding the backend lock, so it may use this
288    /// client or drop other files owned by it. Return `Ok(false)` to stop
289    /// successfully. Callback errors propagate. At most one requested chunk is
290    /// buffered at once, and the file closes on success, cancellation, or error.
291    pub(crate) fn read_stream_with_options(
292        &self,
293        path: impl AsRef<Path>,
294        options: StreamOptions,
295        mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
296    ) -> VfResult<StreamCompletion> {
297        let chunk_size = options.chunk_size_bytes();
298        if chunk_size == 0 {
299            return Err(VfError::failure(0, crate::ERR_INVAL));
300        }
301
302        let file = self.open(path)?;
303        let raw_file = file.raw()?.clone();
304        let operation = (|| -> VfResult<StreamCompletion> {
305            let mut offset = 0u64;
306            loop {
307                let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
308                let result = {
309                    let mut backend = self.lock()?;
310                    let result = backend.read_impl(&request)?;
311                    validate_read_results(
312                        "read_stream",
313                        std::slice::from_ref(&request),
314                        std::slice::from_ref(&result),
315                    )?;
316                    result
317                };
318                let length = result.data.len();
319                if !result.data.is_empty() && !callback(offset, &result.data)? {
320                    return Ok(StreamCompletion::Stopped {
321                        next_offset: offset
322                            .checked_add(length as u64)
323                            .ok_or_else(|| VfError::client(0, crate::ERR_INVAL))?,
324                    });
325                }
326                if result.eof {
327                    return Ok(StreamCompletion::Complete);
328                }
329                offset = offset
330                    .checked_add(length as u64)
331                    .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
332            }
333        })();
334        let close = file.close();
335        match operation {
336            Err(error) => Err(error),
337            Ok(completion) => close.map(|()| completion),
338        }
339    }
340}
341
342impl<F: Backend> FsClient<F> {
343    pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
344        SetMetadata {
345            client: self,
346            path: path.as_ref().to_path_buf(),
347            update: MetadataUpdate::new(),
348            follow: true,
349        }
350    }
351}
352
353impl<F: Backend> FsClient<F> {
354    fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Attrs> {
355        let mut filesystem = self.lock()?;
356        let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
357        filesystem
358            .metadata_path_impl(path, follow)
359            .map_err(|error| error.with_context(operation, path))
360    }
361}
362
363impl<F: Backend> FsClient<F> {
364    /// Rename independent source/destination pairs in one vector phase.
365    /// Requested atomic destination semantics are applied per pair; the vector
366    /// is not transactional. Unsupported semantics are never emulated.
367    pub(crate) fn vrename<P: AsRef<Path>, Q: AsRef<Path>>(
368        &self,
369        pairs: &[(P, Q)],
370        options: vfsi_core::api::RenameOptions,
371    ) -> VfResult<()> {
372        if pairs.is_empty() {
373            return Ok(());
374        }
375        let requests: Vec<_> = pairs
376            .iter()
377            .map(|(from, to)| {
378                (
379                    VfFile::from_os_path(from.as_ref()),
380                    VfFile::from_os_path(to.as_ref()),
381                )
382            })
383            .collect();
384        self.lock()?
385            .vrename_with_options_impl(&requests, options)
386            .map_err(|error| match error.index() {
387                Some(index) if index < pairs.len() => {
388                    error.with_context("vrename", pairs[index].0.as_ref())
389                }
390                Some(_) => {
391                    VfError::transport(None, "rename backend returned an invalid error index")
392                }
393                None => error,
394            })
395    }
396
397    /// Create directories with per-request Unix permission bits.
398    pub(crate) fn vmkdir<P: AsRef<Path>>(&self, directories: &[(P, u32)]) -> VfResult<()> {
399        let mut seen = HashSet::with_capacity(directories.len());
400        for (index, (path, _)) in directories.iter().enumerate() {
401            if !seen.insert(path.as_ref()) {
402                return Err(
403                    VfError::client(index, crate::ERR_INVAL).with_context("vmkdir", path.as_ref())
404                );
405            }
406        }
407        if directories.is_empty() {
408            return Ok(());
409        }
410        let dirs: Vec<_> = directories
411            .iter()
412            .map(|(path, mode)| crate::VfAttrs {
413                file: VfFile::from_os_path(path.as_ref()),
414                masks: AttrMask::MODE,
415                mode: *mode,
416                ..Default::default()
417            })
418            .collect();
419        self.lock()?
420            .vmkdir_impl(&dirs)
421            .map_err(|error| match error.index() {
422                Some(index) if index < directories.len() => {
423                    error.with_context("vmkdir", directories[index].0.as_ref())
424                }
425                Some(_) => {
426                    VfError::transport(None, "mkdir backend returned an invalid error index")
427                }
428                None => error,
429            })
430    }
431
432    /// Create symbolic links in one native backend vector.
433    pub(crate) fn vsymlink<P: AsRef<Path>, Q: AsRef<Path>>(
434        &self,
435        pairs: &[(P, Q)],
436    ) -> VfResult<()> {
437        if pairs.is_empty() {
438            return Ok(());
439        }
440        let targets: Vec<_> = pairs.iter().map(|(target, _)| target.as_ref()).collect();
441        let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
442        self.lock()?
443            .vsymlink_impl(&targets, &links)
444            .map_err(|error| link_error(error, &links, "vsymlink"))
445    }
446
447    /// Read targets without converting Unix path bytes through UTF-8.
448    pub(crate) fn vreadlink<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<PathBuf>> {
449        if paths.is_empty() {
450            return Ok(Vec::new());
451        }
452        let paths: Vec<_> = paths.iter().map(AsRef::as_ref).collect();
453        let targets = self
454            .lock()?
455            .vreadlink_impl(&paths)
456            .map_err(|error| link_error(error, &paths, "vreadlink"))?;
457        if targets.len() != paths.len() {
458            return Err(VfError::transport(
459                None,
460                "readlink backend returned an invalid result count",
461            ));
462        }
463        Ok(targets
464            .into_iter()
465            .map(crate::native::bytes_to_path)
466            .collect())
467    }
468
469    /// Create hard links in one native backend vector.
470    pub(crate) fn vhardlink<P: AsRef<Path>, Q: AsRef<Path>>(
471        &self,
472        pairs: &[(P, Q)],
473    ) -> VfResult<()> {
474        if pairs.is_empty() {
475            return Ok(());
476        }
477        let sources: Vec<_> = pairs.iter().map(|(source, _)| source.as_ref()).collect();
478        let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
479        self.lock()?
480            .vhardlink_impl(&sources, &links)
481            .map_err(|error| link_error(error, &links, "vhardlink"))
482    }
483
484    /// Lazy between directories, with selective no-follow metadata and pruning.
485    /// Enter runs before any listing; Leave follows even a pruned directory.
486    /// Sorting buffers one bounded directory, not the whole tree. The callback
487    /// runs outside the backend lock. Limits include the starting object.
488    /// Maximum safe cohort size for incremental directory paging.
489    #[doc(hidden)]
490    pub fn directory_page_batch_size(&self) -> VfResult<usize> {
491        Ok(self.lock()?.directory_page_batch_size().clamp(1, 32))
492    }
493    /// Fetch a bounded vector of directory pages, retaining backend cursors.
494    #[doc(hidden)]
495    pub fn read_dir_pages_with_fields(
496        &self,
497        paths: &[&Path],
498        fields: AttrMask,
499        cursors: Vec<Option<crate::DirPageCursor>>,
500        page_size: usize,
501        max_entries: usize,
502    ) -> VfResult<Vec<crate::DirectoryPage>> {
503        let pages = self.lock()?.vlistdir_pages_impl(
504            paths,
505            fields | AttrMask::MODE | AttrMask::SIZE,
506            cursors,
507            page_size,
508            max_entries,
509        )?;
510        if pages.len() != paths.len() {
511            return Err(VfError::transport(
512                None,
513                "directory pages returned an invalid result count",
514            ));
515        }
516        pages
517            .into_iter()
518            .zip(paths)
519            .enumerate()
520            .map(|(index, ((attrs, next, children), path))| {
521                if children.len() != attrs.len()
522                    || attrs.len() > page_size
523                    || (attrs.is_empty() && next.is_some())
524                {
525                    return Err(VfError::transport(
526                        Some(index),
527                        "invalid directory page progress",
528                    ));
529                }
530                let mut seeds = Vec::new();
531                let entries = attrs
532                    .into_iter()
533                    .zip(children)
534                    .map(|(attrs, child)| {
535                        let entry_path = attrs
536                            .file
537                            .path()
538                            .ok_or_else(|| {
539                                VfError::transport(Some(index), "directory entry has no path")
540                            })?
541                            .to_path_buf();
542                        if entry_path.parent() != Some(*path) {
543                            return Err(VfError::transport(
544                                Some(index),
545                                "directory entry is outside its parent",
546                            ));
547                        }
548                        if let Some(child) = child {
549                            seeds.push((entry_path.clone(), child));
550                        }
551                        Ok(DirEntry::new(
552                            entry_path,
553                            vfsi_core::metadata_from_attrs(attrs),
554                        ))
555                    })
556                    .collect::<VfResult<Vec<_>>>()?;
557                Ok((
558                    DirectoryListing {
559                        path: path.to_path_buf(),
560                        entries,
561                    },
562                    next,
563                    seeds,
564                ))
565            })
566            .collect()
567    }
568}
569
570impl<F: Backend> FsClient<F> {
571    /// Create `path` if missing, otherwise empty it. Errors if it exists and is
572    /// not a directory (a symlink to a directory is not a directory here).
573    pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
574        self.lock()?.ensure_empty_dir_impl(path.as_ref())
575    }
576
577    /// Empty a directory while keeping it, with explicit removal policy.
578    pub(crate) fn remove_dir_contents_with_options(
579        &self,
580        path: impl AsRef<Path>,
581        options: RemoveOptions,
582    ) -> VfResult<()> {
583        let path = path.as_ref();
584        if !self
585            .removal_metadata(path, "remove_dir_contents_impl")?
586            .is_dir()
587        {
588            return Err(
589                VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
590            );
591        }
592        self.lock()?
593            .remove_dir_contents_path_with_options_impl(path, options)
594            .map_err(|error| error.with_context("remove_dir_contents", path))
595    }
596
597    /// Open a *genuine* directory handle for race-resistant, handle-rooted
598    /// removal. Backends that only return a path fail instead of silently
599    /// losing the handle safety guarantee.
600    pub fn open_dir_handle(&self, path: impl AsRef<Path>) -> VfResult<FsDir<F>> {
601        let path = path.as_ref();
602        let mut backend = self.lock()?;
603        let dir = backend.open_dir_impl(path)?;
604        if !matches!(dir, VfDir::Descriptor { .. }) {
605            let _ = backend.close_dir_impl(&dir);
606            return Err(VfError::unsupported(0).with_context("open_dir_handle", path));
607        }
608        Ok(FsDir {
609            inner: Arc::clone(&self.inner),
610            dir: Some(dir),
611            path: path.to_path_buf(),
612        })
613    }
614}
615
616/// Owned, handle-rooted directory. Dropping it queues backend cleanup;
617/// [`close`](Self::close) reports cleanup errors explicitly.
618pub struct FsDir<F: Backend> {
619    inner: Arc<SharedBackend<F>>,
620    dir: Option<VfDir>,
621    path: PathBuf,
622}
623
624impl<F: Backend> fmt::Debug for FsDir<F> {
625    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
626        formatter
627            .debug_struct("FsDir")
628            .field("path", &self.path)
629            .finish_non_exhaustive()
630    }
631}
632
633impl<F: Backend> FsDir<F> {
634    pub fn is_closed(&self) -> bool {
635        self.dir.is_none()
636    }
637    /// Name used at open, retained for diagnostics; not updated after rename.
638    pub fn path(&self) -> &Path {
639        &self.path
640    }
641
642    pub fn remove_contents(&self) -> VfResult<()> {
643        self.remove_contents_with_options(RemoveOptions::default())
644    }
645
646    pub fn remove_contents_with_options(&self, options: RemoveOptions) -> VfResult<()> {
647        let dir = self
648            .dir
649            .as_ref()
650            .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))?;
651        self.inner
652            .lock()
653            .map_err(|_| poisoned())?
654            .remove_dir_contents_handle_with_options_impl(dir, options)
655            .map_err(|error| error.with_context("remove_dir_contents", &self.path))
656    }
657
658    pub fn close(mut self) -> VfResult<()> {
659        self.try_close()
660    }
661
662    /// Keep cleanup ownership on failure so the caller can retry explicitly.
663    pub fn try_close(&mut self) -> VfResult<()> {
664        let Some(dir) = self.dir.as_ref() else {
665            return Ok(());
666        };
667        self.inner
668            .lock()
669            .map_err(|_| poisoned())?
670            .close_dir_impl(dir)?;
671        self.dir = None;
672        Ok(())
673    }
674}
675
676impl<F: Backend> Drop for FsDir<F> {
677    fn drop(&mut self) {
678        if let Some(dir) = self.dir.take() {
679            self.inner.defer(
680                CleanupTarget::Directory(dir, std::mem::take(&mut self.path)),
681                |backend, target| {
682                    let CleanupTarget::Directory(dir, path) = target else {
683                        unreachable!()
684                    };
685                    backend
686                        .close_dir_impl(dir)
687                        .map_err(|error| error.with_context("close_dir", path))
688                },
689            );
690        }
691    }
692}
693
694impl<F: Backend> FsClient<F> {
695    /// Copy whole files in request order. A successful prefix may remain if
696    /// a later request fails; this operation does not provide atomicity.
697    pub(crate) fn vcopy<P: AsRef<Path>, Q: AsRef<Path>>(
698        &self,
699        pairs: &[(P, Q)],
700        options: vfsi_core::api::CopyOption,
701    ) -> VfResult<()> {
702        let extents: Vec<_> = pairs
703            .iter()
704            .map(|(from, to)| {
705                crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
706            })
707            .collect();
708        self.lock()?.vcopy_impl(&extents, options).map_err(|error| {
709            error
710                .index()
711                .and_then(|index| pairs.get(index))
712                .map_or(error.clone(), |(_, to)| {
713                    error.with_context("vcopy", to.as_ref())
714                })
715        })
716    }
717
718    /// Remove paths in request order, optionally recursing into directories.
719    /// A successful prefix may remain if a later path fails.
720    #[doc(hidden)]
721    pub fn vremove_native<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
722        self.vremove_with_options_native(paths, recursive, RemoveOptions::default())
723    }
724
725    /// Remove paths with explicit error, batching, and retry policy.
726    #[doc(hidden)]
727    pub fn vremove_with_options_native<P: AsRef<Path>>(
728        &self,
729        paths: &[P],
730        recursive: bool,
731        options: RemoveOptions,
732    ) -> VfResult<()> {
733        let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
734        self.lock()?
735            .remove_paths_with_options_impl(&paths, recursive, options)
736            .map_err(|error| {
737                error
738                    .index()
739                    .and_then(|index| paths.get(index))
740                    .map_or(error.clone(), |path| {
741                        error.with_context("vremove_native", path)
742                    })
743            })
744    }
745
746    /// Vector metadata query with explicit fields and final-symlink handling.
747    /// Ancestor symlinks follow the backend's normal namespace semantics.
748    #[doc(hidden)]
749    pub fn vgetattrs_native<P: AsRef<Path>>(
750        &self,
751        paths: &[P],
752        fields: AttrMask,
753        follow: bool,
754    ) -> VfResult<Vec<Attrs>> {
755        if paths.is_empty() {
756            return Ok(Vec::new());
757        }
758        let mut attrs: Vec<_> = paths
759            .iter()
760            .map(|path| crate::VfAttrs {
761                file: VfFile::from_os_path(path.as_ref()),
762                masks: fields | AttrMask::MODE,
763                ..crate::VfAttrs::default()
764            })
765            .collect();
766        let result = {
767            let mut backend = self.lock()?;
768            if follow {
769                backend.vgetattrs_impl(&mut attrs)
770            } else {
771                backend.vgetattrs_nofollow_impl(&mut attrs)
772            }
773        };
774        result.map_err(|error| match error.index() {
775            Some(index) if index < paths.len() => {
776                error.with_context("metadatav", paths[index].as_ref())
777            }
778            Some(_) => VfError::transport(None, "metadata backend returned an invalid error index"),
779            None => error,
780        })?;
781        Ok(attrs
782            .into_iter()
783            .map(vfsi_core::metadata_from_attrs)
784            .collect())
785    }
786
787    /// Fetch no-follow metadata for many paths using the backend vector operation.
788    #[doc(hidden)]
789    pub fn vsymlink_attrs_native(&self, paths: &[&Path]) -> VfResult<Vec<Attrs>> {
790        let mut attrs: Vec<_> = paths
791            .iter()
792            .map(|path| crate::VfAttrs {
793                file: VfFile::from_os_path(path),
794                masks: AttrMask::MODE,
795                ..crate::VfAttrs::default()
796            })
797            .collect();
798        self.lock()?
799            .vgetattrs_nofollow_impl(&mut attrs)
800            .map_err(|error| {
801                error
802                    .index()
803                    .and_then(|index| paths.get(index))
804                    .map_or(error.clone(), |path| {
805                        error.with_context("vsymlink_attrs_native", path)
806                    })
807            })?;
808        Ok(attrs
809            .into_iter()
810            .map(vfsi_core::metadata_from_attrs)
811            .collect())
812    }
813}
814
815impl<F: Backend> FsClient<F> {
816    /// Open an ordered vector of files.
817    ///
818    /// Success returns one RAII handle per request. Failure returns no
819    /// handles; VFSI does not promise transactional rollback of other
820    /// filesystem effects such as file creation.
821    pub(crate) fn vopen(&self, requests: &[OpenOp]) -> VfResult<Vec<FsFile<F>>> {
822        let mut filesystem = self.lock()?;
823        let files = filesystem.vopen_impl(requests).map_err(|error| {
824            error
825                .index()
826                .and_then(|index| requests.get(index))
827                .map_or(error.clone(), |request| {
828                    error.with_context("vopen", &request.path)
829                })
830        })?;
831        if files.len() != requests.len() {
832            let error = wrong_result_count("vopen", requests.len(), files.len());
833            for file in &files {
834                let _ = filesystem.close_impl(file);
835            }
836            return Err(error);
837        }
838        drop(filesystem);
839        Ok(files
840            .into_iter()
841            .zip(requests)
842            .map(|(file, request)| FsFile {
843                inner: Arc::clone(&self.inner),
844                file: Some(file),
845                path: request.path.clone(),
846            })
847            .collect())
848    }
849
850    /// Try to close a group through one vector operation without consuming
851    /// the handles. On failure, all handles remain armed: the backend may
852    /// have closed a prefix, so callers must reconcile before retrying.
853    #[doc(hidden)]
854    pub fn vclose<'a>(&self, files: impl IntoIterator<Item = &'a mut FsFile<F>>) -> VfResult<()>
855    where
856        F: 'a,
857    {
858        let mut files: Vec<_> = files.into_iter().collect();
859        for (index, file) in files.iter().enumerate() {
860            self.validate_owner(file, index)?;
861        }
862        let positions: Vec<_> = files
863            .iter()
864            .enumerate()
865            .filter_map(|(index, file)| (!file.is_closed()).then_some(index))
866            .collect();
867        files.retain(|file| !file.is_closed());
868        if files.is_empty() {
869            return Ok(());
870        }
871        let descriptors: Vec<VfFile> = files
872            .iter()
873            .map(|file| file.raw().cloned())
874            .collect::<VfResult<_>>()?;
875        self.lock()?.vclose_impl(&descriptors).map_err(|error| {
876            error
877                .index()
878                .and_then(|index| files.get(index))
879                .map_or(error.clone(), |file| {
880                    error.with_context("vclose_owned", file.path())
881                })
882                .map_index(|index| positions.get(index).copied().unwrap_or(index))
883        })?;
884        for file in &mut files {
885            file.file = None;
886        }
887        Ok(())
888    }
889
890    /// Close a group of files through one vector operation.
891    ///
892    /// On failure, the handles are dropped and the backend receives
893    /// best-effort scalar cleanup attempts. Use [`vclose`](Self::vclose)
894    /// to retain the handles after an error.
895    #[doc(hidden)]
896    pub fn vclose_owned(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
897        self.vclose(&mut files)
898    }
899
900    /// Read an ordered vector with a 16 MiB aggregate request limit.
901    /// Use [`vread_with_limit_native`](Self::vread_with_limit_native) to tune the limit or
902    /// [`vread_into_native`](Self::vread_into_native) to provide bounded caller-owned buffers.
903    #[doc(hidden)]
904    pub fn vread_native(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<FsReadResult>> {
905        self.vread_with_limit_native(requests, self.limits.max_read_bytes)
906    }
907
908    /// Read an ordered vector with an explicit aggregate request limit.
909    #[doc(hidden)]
910    pub fn vread_with_limit_native(
911        &self,
912        requests: &[FsRead<'_, F>],
913        max_total_bytes: usize,
914    ) -> VfResult<Vec<FsReadResult>> {
915        self.vread_with_limit_projected_native(requests, max_total_bytes, |request| request)
916    }
917
918    /// Backend adapter for opaque application requests; projection does not allocate.
919    /// The projection must return the same embedded request on every invocation.
920    #[doc(hidden)]
921    pub fn vread_with_limit_projected_native<'b, T>(
922        &self,
923        requests: &[T],
924        max_total_bytes: usize,
925        project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
926    ) -> VfResult<Vec<FsReadResult>>
927    where
928        F: 'b,
929    {
930        let mut requested = 0usize;
931        for (index, request) in requests.iter().map(&project).enumerate() {
932            requested = requested
933                .checked_add(request.length)
934                .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
935            if requested > max_total_bytes {
936                return Err(VfError::client(index, libc::EFBIG as u32)
937                    .with_context("vread_native", request.file.path()));
938            }
939        }
940        let reads = self.read_ops(requests, &project)?;
941        let results = self.lock()?.vread_impl(&reads).map_err(|error| {
942            error
943                .index()
944                .and_then(|index| requests.get(index).map(&project))
945                .map_or(error.clone(), |request| {
946                    error.with_context("vread_native", request.file.path())
947                })
948        })?;
949        if results.len() != requests.len() {
950            return Err(wrong_result_count(
951                "vread_native",
952                requests.len(),
953                results.len(),
954            ));
955        }
956        validate_read_results("vread_native", &reads, &results).map_err(|error| {
957            error
958                .index()
959                .and_then(|index| requests.get(index).map(&project))
960                .map_or(error.clone(), |request| {
961                    error.with_context("vread_native", request.file.path())
962                })
963        })?;
964        Ok(results.into_iter().map(read_result).collect())
965    }
966
967    fn read_ops<'b, T>(
968        &self,
969        requests: &[T],
970        project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
971    ) -> VfResult<Vec<ReadOp>>
972    where
973        F: 'b,
974    {
975        let mut reads = Vec::with_capacity(requests.len());
976        for (index, request) in requests.iter().map(&project).enumerate() {
977            self.validate_owner(request.file, index)?;
978            reads.push(ReadOp::new(
979                request.file.raw()?.clone(),
980                request.offset,
981                request.length,
982            ));
983        }
984        Ok(reads)
985    }
986
987    #[doc(hidden)]
988    pub fn vread_into_native(
989        &self,
990        requests: &mut [FsReadInto<'_, F>],
991    ) -> VfResult<Vec<FsReadIntoResult>> {
992        self.vread_into_with_limit_native(requests, self.limits.max_read_bytes)
993    }
994
995    /// Read into caller storage with an explicit aggregate buffer budget.
996    /// This also bounds allocation in copying fallback implementations.
997    #[doc(hidden)]
998    pub fn vread_into_with_limit_native(
999        &self,
1000        requests: &mut [FsReadInto<'_, F>],
1001        max_bytes: usize,
1002    ) -> VfResult<Vec<FsReadIntoResult>> {
1003        self.vread_into_with_limit_projected_native(
1004            requests,
1005            max_bytes,
1006            |request| request,
1007            |request| request,
1008        )
1009    }
1010
1011    /// Project borrowed buffers without an intermediate request allocation.
1012    /// Both projections must identify the same embedded request, and remain
1013    /// stable across preflight, dispatch, and result validation.
1014    #[doc(hidden)]
1015    pub fn vread_into_with_limit_projected_native<'b, T>(
1016        &self,
1017        requests: &mut [T],
1018        max_bytes: usize,
1019        project: impl for<'r> Fn(&'r T) -> &'r FsReadInto<'b, F>,
1020        mut project_mut: impl for<'r> FnMut(&'r mut T) -> &'r mut FsReadInto<'b, F>,
1021    ) -> VfResult<Vec<FsReadIntoResult>>
1022    where
1023        F: 'b,
1024    {
1025        let mut requested = 0usize;
1026        let mut reads = Vec::with_capacity(requests.len());
1027        for (index, request) in requests.iter().map(&project).enumerate() {
1028            self.validate_owner(request.file, index)?;
1029            requested = requested
1030                .checked_add(request.buffer.len())
1031                .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1032            if requested > max_bytes {
1033                return Err(VfError::client(index, libc::EFBIG as u32));
1034            }
1035            reads.push(ReadOp::new(
1036                request.file.raw()?.clone(),
1037                request.offset,
1038                request.buffer.len(),
1039            ));
1040        }
1041        let results = {
1042            let mut buffers: Vec<&mut [u8]> = requests
1043                .iter_mut()
1044                .map(&mut project_mut)
1045                .map(|request| &mut *request.buffer)
1046                .collect();
1047            self.lock()?.vread_into_impl(&reads, &mut buffers)
1048        }
1049        .map_err(|error| {
1050            error
1051                .index()
1052                .and_then(|index| requests.get(index).map(&project))
1053                .map_or(error.clone(), |request| {
1054                    error.with_context("vread_into_native", request.file.path())
1055                })
1056        })?;
1057        validate_read_into_results("vread_into_native", &reads, &results).map_err(|error| {
1058            error
1059                .index()
1060                .and_then(|index| requests.get(index).map(&project))
1061                .map_or(error.clone(), |request| {
1062                    error.with_context("vread_into_native", request.file.path())
1063                })
1064        })?;
1065        Ok(results
1066            .into_iter()
1067            .map(|result| FsReadIntoResult {
1068                offset: result.offset,
1069                read: result.read,
1070                eof: result.eof,
1071            })
1072            .collect())
1073    }
1074
1075    #[doc(hidden)]
1076    pub fn vwrite_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1077        self.vwrite_projected_native(requests, |request| request)
1078    }
1079
1080    /// Backend adapter for opaque application requests, preserving borrowed payloads.
1081    /// The projection must return the same embedded request on every invocation.
1082    #[doc(hidden)]
1083    pub fn vwrite_projected_native<'b, T>(
1084        &self,
1085        requests: &[T],
1086        project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1087    ) -> VfResult<Vec<FsWriteResult>>
1088    where
1089        F: 'b,
1090    {
1091        self.vwrite_mapped_native(requests, |item| {
1092            let request = project(item);
1093            FsWrite {
1094                file: request.file,
1095                offset: request.offset,
1096                data: request.data,
1097            }
1098        })
1099    }
1100
1101    /// Adapter constructing cheap borrowed requests without a temporary vector.
1102    /// The mapper must return the same file, offset, and payload on each call.
1103    #[doc(hidden)]
1104    #[doc(hidden)]
1105    pub fn vwrite_mapped_native<'b, T>(
1106        &self,
1107        requests: &[T],
1108        project: impl Fn(&T) -> FsWrite<'b, F>,
1109    ) -> VfResult<Vec<FsWriteResult>>
1110    where
1111        F: 'b,
1112    {
1113        let writes = self.write_ops(requests, &project)?;
1114        let results = self.lock()?.vwrite_impl(&writes).map_err(|error| {
1115            error
1116                .index()
1117                .and_then(|index| requests.get(index).map(&project))
1118                .map_or(error.clone(), |request| {
1119                    error.with_context("vwrite_native", request.file.path())
1120                })
1121        })?;
1122        if results.len() != requests.len() {
1123            return Err(wrong_result_count(
1124                "vwrite_native",
1125                requests.len(),
1126                results.len(),
1127            ));
1128        }
1129        validate_write_results("vwrite_native", &writes, &results).map_err(|error| {
1130            error
1131                .index()
1132                .and_then(|index| requests.get(index).map(&project))
1133                .map_or(error.clone(), |request| {
1134                    error.with_context("vwrite_native", request.file.path())
1135                })
1136        })?;
1137        Ok(results.into_iter().map(write_result).collect())
1138    }
1139
1140    /// Write every byte in each positional request, retrying short writes in
1141    /// vector waves. Like `vwrite_native`, this is not transactional: an error may
1142    /// follow a successfully written prefix. Overlapping requests through the
1143    /// same path complete in input order; different paths are presumed
1144    /// independent (including hard-link aliases).
1145    #[doc(hidden)]
1146    pub fn vwrite_all_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1147        self.vwrite_all_projected_native(requests, |request| request)
1148    }
1149
1150    /// Complete projected requests with the same preflight and dependency waves.
1151    /// The projection must return the same embedded request on every invocation.
1152    #[doc(hidden)]
1153    pub fn vwrite_all_projected_native<'b, T>(
1154        &self,
1155        requests: &[T],
1156        project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1157    ) -> VfResult<Vec<FsWriteResult>>
1158    where
1159        F: 'b,
1160    {
1161        self.vwrite_all_mapped_native(requests, |item| {
1162            let request = project(item);
1163            FsWrite {
1164                file: request.file,
1165                offset: request.offset,
1166                data: request.data,
1167            }
1168        })
1169    }
1170
1171    /// Complete mapped writes with whole-batch preflight and dependency waves.
1172    /// The mapper must return the same file, offset, and payload on each call.
1173    #[doc(hidden)]
1174    #[doc(hidden)]
1175    pub fn vwrite_all_mapped_native<'b, T>(
1176        &self,
1177        requests: &[T],
1178        project: impl Fn(&T) -> FsWrite<'b, F>,
1179    ) -> VfResult<Vec<FsWriteResult>>
1180    where
1181        F: 'b,
1182    {
1183        // Validate the entire batch before writing any prefix. In particular,
1184        // empty requests must not conceal a foreign or already-closed file.
1185        let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
1186        let mut blocked_by = vec![Vec::new(); requests.len()];
1187        for (index, request) in requests.iter().map(&project).enumerate() {
1188            self.validate_owner(request.file, index)?;
1189            request.file.raw().map_err(|error| {
1190                error
1191                    .with_index(index)
1192                    .with_context("vwrite_all_native", request.file.path())
1193            })?;
1194            let VfOffset::At(offset) = request.offset else {
1195                return Err(VfError::client(index, crate::ERR_INVAL)
1196                    .with_context("vwrite_all_native", request.file.path()));
1197            };
1198            let end = offset
1199                .checked_add(request.data.len() as u64)
1200                .ok_or_else(|| {
1201                    VfError::client(index, libc::EOVERFLOW as u32)
1202                        .with_context("vwrite_all_native", request.file.path())
1203                })?;
1204            // Non-overlapping writes commute. An overlapping later request
1205            // must wait until every earlier conflicting request is complete:
1206            // otherwise a short-write retry can overwrite the later bytes.
1207            let prior = prior_by_path.entry(request.file.path()).or_default();
1208            for &(earlier, start, earlier_end) in prior.iter() {
1209                if offset < earlier_end && start < end {
1210                    blocked_by[index].push(earlier);
1211                }
1212            }
1213            prior.push((index, offset, end));
1214        }
1215        let mut totals = vec![0usize; requests.len()];
1216        let mut stable = vec![true; requests.len()];
1217        let mut pending: Vec<usize> = (0..requests.len())
1218            .filter(|&index| !project(&requests[index]).data.is_empty())
1219            .collect();
1220        while !pending.is_empty() {
1221            let wave_indices: Vec<usize> = pending
1222                .iter()
1223                .copied()
1224                .filter(|&index| {
1225                    blocked_by[index]
1226                        .iter()
1227                        .all(|&earlier| totals[earlier] == project(&requests[earlier]).data.len())
1228                })
1229                .collect();
1230            let wave: Vec<_> = wave_indices
1231                .iter()
1232                .map(|&index| {
1233                    let request = project(&requests[index]);
1234                    let VfOffset::At(offset) = request.offset else {
1235                        return Err(VfError::client(index, crate::ERR_INVAL)
1236                            .with_context("vwrite_all_native", request.file.path()));
1237                    };
1238                    let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
1239                        VfError::client(index, libc::EOVERFLOW as u32)
1240                            .with_context("vwrite_all_native", request.file.path())
1241                    })?;
1242                    Ok(FsWrite {
1243                        file: request.file,
1244                        offset: VfOffset::At(offset),
1245                        data: &request.data[totals[index]..],
1246                    })
1247                })
1248                .collect::<VfResult<_>>()?;
1249            let results = self.vwrite_native(&wave).map_err(|error| {
1250                error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
1251            })?;
1252            for (&index, result) in wave_indices.iter().zip(results) {
1253                if result.written == 0 {
1254                    return Err(VfError::client(index, crate::ERR_IO)
1255                        .with_context("vwrite_all_native", project(&requests[index]).file.path()));
1256                }
1257                totals[index] += result.written;
1258                stable[index] &= result.stable;
1259            }
1260            pending.retain(|&index| totals[index] < project(&requests[index]).data.len());
1261        }
1262        requests
1263            .iter()
1264            .map(&project)
1265            .enumerate()
1266            .map(|(index, request)| {
1267                let VfOffset::At(offset) = request.offset else {
1268                    return Err(VfError::client(index, crate::ERR_INVAL)
1269                        .with_context("vwrite_all_native", request.file.path()));
1270                };
1271                Ok(FsWriteResult {
1272                    offset,
1273                    written: totals[index],
1274                    stable: stable[index],
1275                })
1276            })
1277            .collect()
1278    }
1279
1280    fn write_ops<'a, 'b, T>(
1281        &self,
1282        requests: &'a [T],
1283        project: impl Fn(&T) -> FsWrite<'b, F>,
1284    ) -> VfResult<Vec<WriteOpRef<'a>>>
1285    where
1286        F: 'b,
1287        'b: 'a,
1288    {
1289        let mut writes = Vec::with_capacity(requests.len());
1290        for (index, request) in requests.iter().map(&project).enumerate() {
1291            self.validate_owner(request.file, index)?;
1292            writes.push(WriteOpRef::new(
1293                request
1294                    .file
1295                    .raw()
1296                    .map_err(|error| error.with_index(index))?,
1297                request.offset,
1298                request.data,
1299            ));
1300        }
1301        Ok(writes)
1302    }
1303
1304    fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
1305        if Arc::ptr_eq(&self.inner, &file.inner) {
1306            Ok(())
1307        } else {
1308            Err(VfError::client(index, crate::ERR_INVAL))
1309        }
1310    }
1311}
1312
1313impl<F: Backend> FsClient<F> {
1314    /// Native bounded whole-file reads for backend adapters.
1315    ///
1316    /// Applications should use [`VfsiExt::read_files_with_options`].
1317    #[doc(hidden)]
1318    pub fn read_files_native<P: AsRef<Path>>(
1319        &self,
1320        paths: &[P],
1321        options: ReadAllOptions,
1322    ) -> VfResult<Vec<Vec<u8>>> {
1323        let files: Vec<_> = paths
1324            .iter()
1325            .map(|path| VfFile::from_os_path(path.as_ref()))
1326            .collect();
1327        let result = self
1328            .lock()?
1329            .vread_all_with_options_impl(&files, options)
1330            .map_err(|error| {
1331                error
1332                    .index()
1333                    .and_then(|index| paths.get(index))
1334                    .map_or(error.clone(), |path| {
1335                        error.with_context("read_files", path.as_ref())
1336                    })
1337            });
1338        result.and_then(|buffers| {
1339            if buffers.len() != paths.len() {
1340                Err(wrong_result_count("read_files", paths.len(), buffers.len()))
1341            } else {
1342                Ok(buffers)
1343            }
1344        })
1345    }
1346}
1347
1348/// `std::fs::OpenOptions`-style builder tied to an [`FsClient`].
1349pub struct OpenOptions<'a, F: FileSystem> {
1350    client: &'a FsClient<F>,
1351    flags: OpenFlags,
1352    mode: u32,
1353}
1354
1355impl<F: FileSystem> Clone for OpenOptions<'_, F> {
1356    fn clone(&self) -> Self {
1357        Self {
1358            client: self.client,
1359            flags: self.flags,
1360            mode: self.mode,
1361        }
1362    }
1363}
1364
1365impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
1366    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1367        formatter
1368            .debug_struct("OpenOptions")
1369            .field("flags", &self.flags)
1370            .field("mode", &format_args!("{:#o}", self.mode))
1371            .finish()
1372    }
1373}
1374
1375impl<'a, F: FileSystem> OpenOptions<'a, F> {
1376    fn new(client: &'a FsClient<F>) -> Self {
1377        Self {
1378            client,
1379            flags: OpenFlags::empty(),
1380            mode: 0o666,
1381        }
1382    }
1383
1384    pub fn read(&mut self, enabled: bool) -> &mut Self {
1385        self.flags.set(OpenFlags::READ, enabled);
1386        self
1387    }
1388
1389    pub fn write(&mut self, enabled: bool) -> &mut Self {
1390        self.flags.set(OpenFlags::WRITE, enabled);
1391        self
1392    }
1393
1394    pub fn append(&mut self, enabled: bool) -> &mut Self {
1395        self.flags.set(OpenFlags::APPEND, enabled);
1396        self
1397    }
1398
1399    pub fn truncate(&mut self, enabled: bool) -> &mut Self {
1400        self.flags.set(OpenFlags::TRUNCATE, enabled);
1401        self
1402    }
1403
1404    pub fn create(&mut self, enabled: bool) -> &mut Self {
1405        self.flags.set(OpenFlags::CREATE, enabled);
1406        self
1407    }
1408
1409    pub fn create_new(&mut self, enabled: bool) -> &mut Self {
1410        self.flags.set(OpenFlags::CREATE_NEW, enabled);
1411        self
1412    }
1413
1414    pub fn mode(&mut self, mode: u32) -> &mut Self {
1415        self.mode = mode;
1416        self
1417    }
1418
1419    pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
1420        self.client
1421            .open_with_native(OpenOp::new(path.as_ref(), self.flags).mode(self.mode))
1422    }
1423}
1424
1425impl<F: Backend> OpenOptions<'_, F> {
1426    pub fn vopen<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
1427        let requests: Vec<OpenOp> = paths
1428            .iter()
1429            .map(|path| OpenOp::new(path.as_ref(), self.flags).mode(self.mode))
1430            .collect();
1431        self.client.vopen(&requests)
1432    }
1433}
1434
1435/// Builder for an atomic metadata update request.
1436pub struct SetMetadata<'a, F: Backend> {
1437    client: &'a FsClient<F>,
1438    path: PathBuf,
1439    update: MetadataUpdate,
1440    follow: bool,
1441}
1442
1443impl<F: Backend> Clone for SetMetadata<'_, F> {
1444    fn clone(&self) -> Self {
1445        Self {
1446            client: self.client,
1447            path: self.path.clone(),
1448            update: self.update.clone(),
1449            follow: self.follow,
1450        }
1451    }
1452}
1453
1454impl<F: Backend> fmt::Debug for SetMetadata<'_, F> {
1455    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1456        formatter
1457            .debug_struct("SetMetadata")
1458            .field("path", &self.path)
1459            .field("update", &self.update)
1460            .field("follow", &self.follow)
1461            .finish_non_exhaustive()
1462    }
1463}
1464
1465impl<F: Backend> SetMetadata<'_, F> {
1466    pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
1467        self.update.permissions = Some(permissions);
1468        self
1469    }
1470
1471    pub fn len(&mut self, len: u64) -> &mut Self {
1472        self.update.len = Some(len);
1473        self
1474    }
1475
1476    pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
1477        self.update.accessed = Some(accessed);
1478        self
1479    }
1480
1481    pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
1482        self.update.modified = Some(modified);
1483        self
1484    }
1485
1486    pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
1487        self.follow = follow;
1488        self
1489    }
1490
1491    pub fn uid(&mut self, uid: u32) -> &mut Self {
1492        self.update.uid = Some(uid);
1493        self
1494    }
1495    pub fn gid(&mut self, gid: u32) -> &mut Self {
1496        self.update.gid = Some(gid);
1497        self
1498    }
1499    pub fn apply(&self) -> VfResult<()> {
1500        self.client
1501            .vsetattrs(&[(&self.path, self.update.clone())], self.follow)
1502    }
1503}
1504
1505/// Owned RAII file which does not borrow the client.
1506///
1507/// Dropping an open file attempts a best-effort CLOSE. This can block on the
1508/// backend mutex and on network I/O; use [`FsFile::try_close`] when the close
1509/// result matters or when its timing must be controlled.
1510pub struct FsFile<F: FileSystem> {
1511    inner: Arc<SharedBackend<F>>,
1512    file: Option<VfFile>,
1513    path: PathBuf,
1514}
1515
1516impl<F: FileSystem> fmt::Debug for FsFile<F> {
1517    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1518        formatter
1519            .debug_struct("FsFile")
1520            .field("path", &self.path)
1521            .field("is_open", &self.file.is_some())
1522            .finish()
1523    }
1524}
1525
1526impl<F: FileSystem> FsFile<F> {
1527    /// Whether explicit close has completed successfully on this handle.
1528    pub fn is_closed(&self) -> bool {
1529        self.file.is_none()
1530    }
1531    fn raw(&self) -> VfResult<&VfFile> {
1532        self.file
1533            .as_ref()
1534            .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
1535    }
1536
1537    /// Positional read which does not alter the file cursor.
1538    pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
1539        self.read_into(buffer, VfOffset::At(offset))
1540    }
1541
1542    /// Positional write which does not alter the file cursor.
1543    pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
1544        self.write_from(buffer, VfOffset::At(offset))
1545    }
1546
1547    pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
1548        FsRead {
1549            file: self,
1550            offset: VfOffset::At(offset),
1551            length,
1552        }
1553    }
1554
1555    /// Name used at open, retained for diagnostics; not a current namespace
1556    /// lookup or proof of identity. Handle operations continue to use the
1557    /// opened object even if its pathname changes.
1558    pub fn path(&self) -> &Path {
1559        &self.path
1560    }
1561
1562    /// Query metadata for the open object without resolving its path again.
1563    pub fn attrs(&self) -> VfResult<Attrs> {
1564        let attributes = AttrMask::MODE
1565            | AttrMask::SIZE
1566            | AttrMask::NLINK
1567            | AttrMask::FILEID
1568            | AttrMask::UID
1569            | AttrMask::GID
1570            | AttrMask::ATIME
1571            | AttrMask::MTIME
1572            | AttrMask::CTIME
1573            | AttrMask::CHANGE;
1574        self.inner
1575            .lock()
1576            .map_err(|_| poisoned())?
1577            .metadata_impl(MetadataQuery::new(self.raw()?.clone(), attributes))
1578            .map(vfsi_core::metadata_from_attrs)
1579            .map_err(|error| error.with_context("metadata", &self.path))
1580    }
1581
1582    /// Truncate or extend the open file.
1583    pub fn truncate(&self, len: u64) -> VfResult<()> {
1584        self.set_metadata_update(MetadataUpdate::new().len(len))
1585    }
1586
1587    /// Change permissions on the open file through the shared vector engine.
1588    pub fn chmod(&self, permissions: Permissions) -> VfResult<()> {
1589        self.set_metadata_update(MetadataUpdate::new().permissions(permissions))
1590    }
1591
1592    fn set_metadata_update(&self, update: MetadataUpdate) -> VfResult<()> {
1593        let client = FsClient {
1594            inner: Arc::clone(&self.inner),
1595            limits: ResourceLimits::default(),
1596        };
1597        client.vsetattrs(&[(vfsi_core::MetadataTarget::File(self), update)], true)
1598    }
1599
1600    pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
1601        FsWrite {
1602            file: self,
1603            offset: VfOffset::At(offset),
1604            data,
1605        }
1606    }
1607
1608    pub fn read_request_at_into<'a>(
1609        &'a self,
1610        offset: u64,
1611        buffer: &'a mut [u8],
1612    ) -> FsReadInto<'a, F> {
1613        FsReadInto {
1614            file: self,
1615            offset: VfOffset::At(offset),
1616            buffer,
1617        }
1618    }
1619
1620    pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
1621        self.read_into(buffer, VfOffset::Cur)
1622    }
1623
1624    /// Collect the remaining bytes from this opened object, starting at its
1625    /// cursor, with an explicit logical payload limit. This does not reopen
1626    /// its path. Unlike standard `Read::read_to_end`, allocation is bounded.
1627    ///
1628    /// An exact-limit read uses at most a one-byte EOF probe. On overflow or
1629    /// I/O failure no buffer is returned and the cursor may have advanced,
1630    /// including the probe byte; this operation does not restore the cursor.
1631    pub fn read_to_end_with_limit(&mut self, max_bytes: usize) -> VfResult<Vec<u8>> {
1632        let mut data = Vec::new();
1633        loop {
1634            let remaining = max_bytes - data.len();
1635            if remaining == 0 {
1636                let mut probe = [0_u8; 1];
1637                if self.read_native(&mut probe)? == 0 {
1638                    return Ok(data);
1639                }
1640                return Err(VfError::client(0, libc::EFBIG as u32)
1641                    .with_context("read_to_end_with_limit", &self.path));
1642            }
1643            let chunk = remaining.min(crate::DEFAULT_READ_STREAM_CHUNK_BYTES);
1644            data.try_reserve(chunk).map_err(|_| {
1645                VfError::client(0, libc::ENOMEM as u32)
1646                    .with_context("read_to_end_with_limit", &self.path)
1647            })?;
1648            let start = data.len();
1649            data.resize(start + chunk, 0);
1650            let count = self.read_native(&mut data[start..])?;
1651            data.truncate(start + count);
1652            if count == 0 {
1653                return Ok(data);
1654            }
1655        }
1656    }
1657
1658    pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
1659        self.write_from(buffer, VfOffset::Cur)
1660    }
1661
1662    fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
1663        if buffer.is_empty() {
1664            return Ok(0);
1665        }
1666        let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
1667        let result = self
1668            .inner
1669            .lock()
1670            .map_err(|_| poisoned())?
1671            .read_into_impl(&request, buffer)
1672            .map_err(|error| error.with_context("read", &self.path))?;
1673        validate_read_into_results(
1674            "read",
1675            std::slice::from_ref(&request),
1676            std::slice::from_ref(&result),
1677        )
1678        .map_err(|error| error.with_context("read", &self.path))?;
1679        Ok(result.read)
1680    }
1681
1682    fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
1683        if buffer.is_empty() {
1684            return Ok(0);
1685        }
1686        let file = self.raw()?.clone();
1687        let result = self
1688            .inner
1689            .lock()
1690            .map_err(|_| poisoned())?
1691            .write_impl(WriteOpRef::new(&file, offset, buffer))
1692            .map_err(|error| error.with_context("write", &self.path))?;
1693        if result.written > buffer.len() {
1694            return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
1695        }
1696        Ok(result.written)
1697    }
1698
1699    pub fn sync_data(&self) -> VfResult<()> {
1700        self.inner
1701            .lock()
1702            .map_err(|_| poisoned())?
1703            .sync_data(self.raw()?)
1704            .map_err(|error| error.with_context("sync_data", &self.path))
1705    }
1706
1707    pub fn sync_all(&self) -> VfResult<()> {
1708        self.inner
1709            .lock()
1710            .map_err(|_| poisoned())?
1711            .sync_all(self.raw()?)
1712            .map_err(|error| error.with_context("sync_all", &self.path))
1713    }
1714
1715    /// Attempt to close without consuming the handle. A failed close leaves
1716    /// the handle armed so the caller can reconcile or retry cleanup explicitly.
1717    /// After an ambiguous close failure, remote state is unknown: do not resume
1718    /// data I/O just because `is_closed()` is false. This is local ownership,
1719    /// not proof that the server still has the descriptor open.
1720    pub fn try_close(&mut self) -> VfResult<()> {
1721        let Some(file) = self.file.as_ref() else {
1722            return Ok(());
1723        };
1724        self.inner
1725            .lock()
1726            .map_err(|_| poisoned())?
1727            .close_impl(file)
1728            .map_err(|error| error.with_context("close", &self.path))?;
1729        self.file = None;
1730        Ok(())
1731    }
1732
1733    /// Consume and close the handle. On failure, `Drop` queues cleanup for
1734    /// a later operation or explicit cleanup drain; use
1735    /// [`try_close`](Self::try_close) to retain control.
1736    pub fn close(mut self) -> VfResult<()> {
1737        self.try_close()
1738    }
1739
1740    /// Seek while retaining [`VfError`] protocol and path information.
1741    pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
1742        self.inner
1743            .lock()
1744            .map_err(|_| poisoned())?
1745            .seek_impl(self.raw()?, position)
1746            .map_err(|error| error.with_context("seek", &self.path))
1747    }
1748}
1749
1750impl<F: FileSystem> Read for FsFile<F> {
1751    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
1752        self.read_native(buffer).map_err(io_error)
1753    }
1754}
1755
1756impl<F: FileSystem> Write for FsFile<F> {
1757    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
1758        self.write_native(buffer).map_err(io_error)
1759    }
1760
1761    fn flush(&mut self) -> io::Result<()> {
1762        self.sync_data().map_err(io_error)
1763    }
1764}
1765
1766impl<F: FileSystem> Seek for FsFile<F> {
1767    fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
1768        self.seek_native(position).map_err(io_error)
1769    }
1770}
1771
1772/// Typed read request for [`FsClient::vread_native`].
1773pub struct FsRead<'a, F: FileSystem> {
1774    file: &'a FsFile<F>,
1775    offset: VfOffset,
1776    length: usize,
1777}
1778
1779/// Typed borrowed write request for [`FsClient::vwrite_native`].
1780pub struct FsWrite<'a, F: FileSystem> {
1781    file: &'a FsFile<F>,
1782    offset: VfOffset,
1783    data: &'a [u8],
1784}
1785
1786/// Typed vector read into caller-provided storage.
1787pub struct FsReadInto<'a, F: FileSystem> {
1788    file: &'a FsFile<F>,
1789    offset: VfOffset,
1790    buffer: &'a mut [u8],
1791}
1792
1793impl<F: FileSystem> Drop for FsFile<F> {
1794    fn drop(&mut self) {
1795        if let Some(file) = self.file.take() {
1796            self.inner.defer(
1797                CleanupTarget::File(file, std::mem::take(&mut self.path)),
1798                |backend, target| {
1799                    let CleanupTarget::File(file, path) = target else {
1800                        unreachable!()
1801                    };
1802                    backend
1803                        .close_deferred(file)
1804                        .map_err(|error| error.with_context("close", path))
1805                },
1806            );
1807        }
1808    }
1809}
1810
1811impl<F: FileSystem> FsClient<F> {
1812    /// Query filesystems using one native vector of paths and retained handles.
1813    pub(crate) fn vstatfs<P: vfsi_core::MetadataOperand<FsFile<F>>>(
1814        &self,
1815        targets: &[P],
1816    ) -> VfResult<Vec<crate::FilesystemStats>> {
1817        use vfsi_core::MetadataTarget;
1818        if targets.is_empty() {
1819            return Ok(Vec::new());
1820        }
1821        let mut files = Vec::with_capacity(targets.len());
1822        let mut paths = Vec::with_capacity(targets.len());
1823        for (index, target) in targets.iter().enumerate() {
1824            let (raw, path) = match target.metadata_target() {
1825                MetadataTarget::Path(path) => (VfFile::from_os_path(path), path),
1826                MetadataTarget::File(file) => {
1827                    if !Arc::ptr_eq(&self.inner, &file.inner) {
1828                        return Err(VfError::client(index, crate::ERR_INVAL)
1829                            .with_context("vstatfs", &file.path));
1830                    }
1831                    (
1832                        file.raw()
1833                            .map_err(|e| e.with_index(index).with_context("vstatfs", &file.path))?
1834                            .clone(),
1835                        file.path.as_path(),
1836                    )
1837                }
1838            };
1839            files.push(raw);
1840            paths.push(path);
1841        }
1842        let results = self
1843            .lock()?
1844            .vstatfs_impl(&files)
1845            .map_err(|error| match error.index() {
1846                Some(index) if index < paths.len() => error.with_context("vstatfs", paths[index]),
1847                Some(_) => {
1848                    VfError::transport(None, "statfs backend returned an invalid error index")
1849                }
1850                None => error,
1851            })?;
1852        if results.len() != targets.len() {
1853            return Err(VfError::transport(
1854                None,
1855                "statfs backend returned an invalid result count",
1856            ));
1857        }
1858        Ok(results)
1859    }
1860    /// Update paths and open objects with one native attribute vector.
1861    pub(crate) fn vsetattrs<P: vfsi_core::MetadataOperand<FsFile<F>>>(
1862        &self,
1863        updates: &[(P, MetadataUpdate)],
1864        follow_symlinks: bool,
1865    ) -> VfResult<()> {
1866        use vfsi_core::MetadataTarget;
1867        if updates.is_empty() {
1868            return Ok(());
1869        }
1870        let mut paths = Vec::with_capacity(updates.len());
1871        let mut attrs = Vec::with_capacity(updates.len());
1872        for (index, (target, update)) in updates.iter().enumerate() {
1873            let (raw, path) = match target.metadata_target() {
1874                MetadataTarget::Path(path) => (VfFile::from_os_path(path), path),
1875                MetadataTarget::File(file) => {
1876                    if !Arc::ptr_eq(&self.inner, &file.inner) {
1877                        return Err(VfError::client(index, crate::ERR_INVAL)
1878                            .with_context("vsetattrs", &file.path));
1879                    }
1880                    (
1881                        file.raw()
1882                            .map_err(|e| e.with_index(index).with_context("vsetattrs", &file.path))?
1883                            .clone(),
1884                        file.path.as_path(),
1885                    )
1886                }
1887            };
1888            if update.uid == Some(u32::MAX) || update.gid == Some(u32::MAX) {
1889                return Err(
1890                    VfError::client(index, crate::ERR_INVAL).with_context("vsetattrs", path)
1891                );
1892            }
1893            paths.push(path);
1894            let mut attributes = crate::SetAttributes::new(raw);
1895            attributes.mode = update.permissions.map(crate::Permissions::mode);
1896            attributes.size = update.len;
1897            attributes.uid = update.uid;
1898            attributes.gid = update.gid;
1899            attributes.atime = update
1900                .accessed
1901                .map(crate::native::system_time_parts)
1902                .transpose()
1903                .map_err(|e| e.with_index(index).with_context("vsetattrs", path))?;
1904            attributes.mtime = update
1905                .modified
1906                .map(crate::native::system_time_parts)
1907                .transpose()
1908                .map_err(|e| e.with_index(index).with_context("vsetattrs", path))?;
1909            attrs.push(attributes);
1910        }
1911        self.lock()?
1912            .vsetattrs_impl(attrs, follow_symlinks)
1913            .map_err(|error| match error.index() {
1914                Some(index) if index < updates.len() => {
1915                    error.with_context("vsetattrs", paths[index])
1916                }
1917                Some(_) => {
1918                    VfError::transport(None, "setattrs backend returned an invalid error index")
1919                }
1920                None => error,
1921            })
1922    }
1923}
1924
1925fn link_error(error: VfError, paths: &[&Path], operation: &'static str) -> VfError {
1926    match error.index() {
1927        Some(index) if index < paths.len() => error.with_context(operation, paths[index]),
1928        Some(_) => VfError::transport(None, "link backend returned an invalid error index"),
1929        None => error,
1930    }
1931}