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