Skip to main content

vfsi_sync/
client.rs

1//! Owned, shareable synchronous client and file handles.
2
3use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::io::{self, Read, Seek, SeekFrom as IoSeekFrom, Write};
6use std::path::{Component, Path, PathBuf};
7use std::sync::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    TraversalCompletion, WriteResult as FsWriteResult,
15};
16
17use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
18use crate::{
19    AttrMask, Capabilities, CopyFileSystem, DirEntry, DirectoryFileSystem, FileSystem,
20    LinkFileSystem, Metadata, MetadataFileSystem, MetadataQuery, MetadataUpdate,
21    NamespaceFileSystem, OpenFlags, OpenRequest, Permissions, ReadAllOptions, ReadDirOptions,
22    ReadOp, ReadResult, ReadStreamOptions, RemoveOptions, SetAttributes, VecFs, VectorFileSystem,
23    VfDir, VfError, VfFile, VfOffset, VfResult, WriteOpRef, WriteResult,
24};
25
26fn read_result(result: ReadResult) -> FsReadResult {
27    FsReadResult {
28        offset: result.offset,
29        data: result.data,
30        eof: result.eof,
31    }
32}
33
34fn write_result(result: WriteResult) -> FsWriteResult {
35    FsWriteResult {
36        offset: result.offset,
37        written: result.written,
38        stable: result.stable,
39    }
40}
41
42fn io_error(error: VfError) -> io::Error {
43    error.into()
44}
45
46fn poisoned() -> VfError {
47    VfError::client(0, crate::ERR_IO)
48}
49
50fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
51    VfError::transport_with_kind(
52        None,
53        crate::TransportKind::InvalidReply,
54        format!("{operation} backend returned {actual} results for {expected} requests"),
55    )
56}
57
58enum CleanupTarget {
59    File(VfFile, PathBuf),
60    Directory(VfDir, PathBuf),
61}
62struct PendingClose<F> {
63    target: CleanupTarget,
64    close: fn(&mut F, &CleanupTarget) -> VfResult<()>,
65}
66
67/// The cleanup queue has a separate short-held lock: handle Drop never waits
68/// for an RPC or for the backend mutex. Final-owner teardown remains synchronous.
69struct SharedBackend<F: FileSystem> {
70    backend: Mutex<Option<F>>,
71    pending: Mutex<Vec<PendingClose<F>>>,
72    has_pending: AtomicBool,
73}
74impl<F: FileSystem> SharedBackend<F> {
75    fn new(backend: F) -> Self {
76        Self {
77            backend: Mutex::new(Some(backend)),
78            pending: Mutex::new(Vec::new()),
79            has_pending: AtomicBool::new(false),
80        }
81    }
82    fn defer(&self, target: CleanupTarget, close: fn(&mut F, &CleanupTarget) -> VfResult<()>) {
83        self.pending
84            .lock()
85            .unwrap_or_else(|e| e.into_inner())
86            .push(PendingClose { target, close });
87        self.has_pending.store(true, Ordering::Release);
88    }
89    fn lock_without_cleanup(&self) -> Result<BackendGuard<'_, F>, ()> {
90        self.backend
91            .lock()
92            .map(|guard| BackendGuard {
93                guard: Some(guard),
94                pending: &self.pending,
95                has_pending: &self.has_pending,
96            })
97            .map_err(|_| ())
98    }
99    fn lock(&self) -> Result<BackendGuard<'_, F>, ()> {
100        let mut guard = self.lock_without_cleanup()?;
101        // Ordinary I/O must not lose its result to an unrelated cleanup error.
102        // Retain failures; drain_cleanup is the explicit reporting boundary.
103        let _ = guard.drain_cleanup();
104        Ok(guard)
105    }
106}
107impl<F: FileSystem> Drop for SharedBackend<F> {
108    fn drop(&mut self) {
109        if let Some(backend) = self
110            .backend
111            .get_mut()
112            .unwrap_or_else(|e| e.into_inner())
113            .as_mut()
114        {
115            let pending = self.pending.get_mut().unwrap_or_else(|e| e.into_inner());
116            for close in pending.drain(..) {
117                let _ = (close.close)(backend, &close.target);
118            }
119            let callbacks = backend.take_notifications();
120            for callback in callbacks {
121                let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
122            }
123        }
124    }
125}
126struct BackendGuard<'a, F: FileSystem> {
127    guard: Option<std::sync::MutexGuard<'a, Option<F>>>,
128    pending: &'a Mutex<Vec<PendingClose<F>>>,
129    has_pending: &'a AtomicBool,
130}
131impl<F: FileSystem> std::ops::Deref for BackendGuard<'_, F> {
132    type Target = F;
133    fn deref(&self) -> &F {
134        self.guard
135            .as_ref()
136            .expect("live guard")
137            .as_ref()
138            .expect("live backend")
139    }
140}
141impl<F: FileSystem> std::ops::DerefMut for BackendGuard<'_, F> {
142    fn deref_mut(&mut self) -> &mut F {
143        self.guard
144            .as_mut()
145            .expect("live guard")
146            .as_mut()
147            .expect("live backend")
148    }
149}
150impl<F: FileSystem> BackendGuard<'_, F> {
151    fn drain_cleanup(&mut self) -> VfResult<()> {
152        if !self.has_pending.swap(false, Ordering::AcqRel) {
153            return Ok(());
154        }
155        let pending = std::mem::take(&mut *self.pending.lock().unwrap_or_else(|e| e.into_inner()));
156        let mut failed = Vec::new();
157        let mut first = None;
158        for close in pending {
159            if let Err(error) = (close.close)(self, &close.target) {
160                first.get_or_insert(error);
161                failed.push(close);
162            }
163        }
164        if !failed.is_empty() {
165            self.pending
166                .lock()
167                .unwrap_or_else(|e| e.into_inner())
168                .extend(failed);
169            self.has_pending.store(true, Ordering::Release);
170        }
171        first.map_or(Ok(()), Err)
172    }
173}
174impl<F: FileSystem> Drop for BackendGuard<'_, F> {
175    fn drop(&mut self) {
176        let callbacks = self.take_notifications();
177        drop(self.guard.take());
178        for callback in callbacks {
179            // Observability must not turn successful I/O into a panic or lose
180            // ownership of a newly opened descriptor before its RAII wrapping.
181            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
182        }
183    }
184}
185
186/// Cloneable owner of one synchronous backend connection.
187/// Clones share a mutex and serialize backend calls; use separate connections
188/// for parallel RPCs. Explicitly close handles to observe cleanup failures.
189/// Handle Drop queues cleanup, drained before later operations or by
190/// `drain_cleanup`. Dropping the final backend owner performs synchronous
191/// teardown and may wait for pending closes and backend request timeouts.
192pub struct FsClient<F: FileSystem> {
193    inner: Arc<SharedBackend<F>>,
194    limits: ResourceLimits,
195}
196
197impl<F: FileSystem> Clone for FsClient<F> {
198    fn clone(&self) -> Self {
199        Self {
200            inner: Arc::clone(&self.inner),
201            limits: self.limits,
202        }
203    }
204}
205
206impl<F: FileSystem> fmt::Debug for FsClient<F> {
207    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
208        formatter.debug_struct("FsClient").finish_non_exhaustive()
209    }
210}
211
212impl<F: FileSystem> FsClient<F> {
213    pub fn new(filesystem: F) -> Self {
214        Self {
215            inner: Arc::new(SharedBackend::new(filesystem)),
216            limits: ResourceLimits::default(),
217        }
218    }
219
220    /// Configure this client view and its future clones. Existing clones keep
221    /// their policy; all views still share the same connection and ownership.
222    pub fn with_limits(mut self, limits: ResourceLimits) -> Self {
223        self.limits = limits;
224        self
225    }
226
227    pub fn limits(&self) -> ResourceLimits {
228        self.limits
229    }
230
231    fn lock(&self) -> VfResult<BackendGuard<'_, F>> {
232        self.inner.lock().map_err(|_| poisoned())
233    }
234
235    pub fn into_inner(self) -> Result<F, Self> {
236        if self.drain_cleanup().is_err() {
237            return Err(self);
238        }
239        match Arc::try_unwrap(self.inner) {
240            Ok(mut shared) => Ok(shared
241                .backend
242                .get_mut()
243                .unwrap_or_else(|e| e.into_inner())
244                .take()
245                .expect("owned backend")),
246            Err(inner) => Err(Self {
247                inner,
248                limits: self.limits,
249            }),
250        }
251    }
252
253    /// Drain queued Drop cleanup and report the first failure. Failed targets
254    /// remain owned for later cleanup. No new file operation is replayed.
255    pub fn drain_cleanup(&self) -> VfResult<()> {
256        self.inner
257            .lock_without_cleanup()
258            .map_err(|_| poisoned())?
259            .drain_cleanup()
260    }
261}
262
263impl<F: FileSystem> FsClient<F> {
264    pub fn capabilities(&self) -> VfResult<Capabilities> {
265        Ok(self.lock()?.capabilities())
266    }
267    /// Open a path read-only.
268    pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
269        self.open_with(OpenRequest::new(path.as_ref(), OpenFlags::READ))
270    }
271
272    pub fn open_with(&self, request: OpenRequest) -> VfResult<FsFile<F>> {
273        let file = self.lock()?.open_one(&request)?;
274        Ok(FsFile {
275            inner: Arc::clone(&self.inner),
276            file: Some(file),
277            path: request.path,
278        })
279    }
280
281    pub fn open_options(&self) -> OpenOptions<'_, F> {
282        OpenOptions::new(self)
283    }
284
285    pub fn create(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
286        self.open_with(OpenRequest::new(
287            path.as_ref(),
288            OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
289        ))
290    }
291
292    pub fn read(&self, path: impl AsRef<Path>) -> VfResult<Vec<u8>> {
293        self.read_with_limit(path, self.limits.max_read_bytes)
294    }
295
296    /// Read one complete file while limiting the returned allocation.
297    ///
298    /// Use [`FsFile::read_native`] or [`std::io::Read`] to stream files that
299    /// should not be held in one allocation.
300    pub fn read_with_limit(&self, path: impl AsRef<Path>, max_bytes: usize) -> VfResult<Vec<u8>> {
301        let path = path.as_ref();
302        let file = self.open(path)?;
303        let operation = self
304            .lock()?
305            .read_file(file.raw()?, max_bytes)
306            .map_err(|error| error.with_context("read", path))
307            .and_then(|data| {
308                if data.len() > max_bytes {
309                    Err(VfError::client(0, libc::EFBIG as u32).with_context("read", path))
310                } else {
311                    Ok(data)
312                }
313            });
314        // Owned cleanup queues a retry through Drop on close failure, and keeps
315        // the read error primary when both read and close fail.
316        let cleanup = file.close();
317        operation.and_then(|data| cleanup.map(|()| data))
318    }
319
320    pub fn read_to_string(&self, path: impl AsRef<Path>) -> VfResult<String> {
321        self.read_to_string_with_limit(path, self.limits.max_read_bytes)
322    }
323
324    /// Read one complete UTF-8 file with a caller-selected allocation limit.
325    pub fn read_to_string_with_limit(
326        &self,
327        path: impl AsRef<Path>,
328        max_bytes: usize,
329    ) -> VfResult<String> {
330        let path = path.as_ref();
331        String::from_utf8(self.read_with_limit(path, max_bytes)?)
332            .map_err(|_| VfError::client(0, crate::ERR_INVAL).with_context("read_to_string", path))
333    }
334
335    pub fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> VfResult<()> {
336        let path = path.as_ref();
337        let mut file = self.create(path)?;
338        let mut written = 0;
339        while written < data.len() {
340            let count = file.write_native(&data[written..])?;
341            if count == 0 {
342                return Err(VfError::client(0, crate::ERR_IO).with_context("write", path));
343            }
344            written += count;
345        }
346        file.close()
347    }
348}
349
350impl<F: FileSystem> FsClient<F> {
351    /// Stream one file from offset zero in bounded chunks.
352    ///
353    /// The callback runs without holding the backend lock, so it may use this
354    /// client or drop other files owned by it. Return `Ok(false)` to stop
355    /// successfully. Callback errors
356    /// are propagated. The file is closed on success, cancellation, callback
357    /// error, or read error. At most one requested chunk is buffered at once.
358    pub fn read_stream(
359        &self,
360        path: impl AsRef<Path>,
361        callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
362    ) -> VfResult<StreamCompletion> {
363        self.read_stream_with_options(
364            path,
365            ReadStreamOptions::new().chunk_size(self.limits.stream_chunk_bytes),
366            callback,
367        )
368    }
369
370    /// Stream one file using an explicit maximum chunk size.
371    pub fn read_stream_with_options(
372        &self,
373        path: impl AsRef<Path>,
374        options: ReadStreamOptions,
375        mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
376    ) -> VfResult<StreamCompletion> {
377        let chunk_size = options.chunk_size_bytes();
378        if chunk_size == 0 {
379            return Err(VfError::failure(0, crate::ERR_INVAL));
380        }
381
382        let file = self.open(path)?;
383        let raw_file = file.raw()?.clone();
384        let operation = (|| -> VfResult<StreamCompletion> {
385            let mut offset = 0u64;
386            loop {
387                let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
388                let result = {
389                    let mut backend = self.lock()?;
390                    let result = backend.read_one(&request)?;
391                    validate_read_results(
392                        "read_stream",
393                        std::slice::from_ref(&request),
394                        std::slice::from_ref(&result),
395                    )?;
396                    result
397                };
398                let length = result.data.len();
399                if !result.data.is_empty() && !callback(offset, &result.data)? {
400                    return Ok(StreamCompletion::Stopped {
401                        next_offset: offset
402                            .checked_add(length as u64)
403                            .ok_or_else(|| VfError::client(0, crate::ERR_INVAL))?,
404                    });
405                }
406                if result.eof {
407                    return Ok(StreamCompletion::Complete);
408                }
409                offset = offset
410                    .checked_add(length as u64)
411                    .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
412            }
413        })();
414        let close = file.close();
415        match operation {
416            Err(error) => Err(error),
417            Ok(completion) => close.map(|()| completion),
418        }
419    }
420}
421
422impl<F: MetadataFileSystem> FsClient<F> {
423    pub fn metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
424        self.lock()?.metadata_path(path.as_ref(), true)
425    }
426
427    pub fn symlink_metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
428        self.lock()?.metadata_path(path.as_ref(), false)
429    }
430
431    pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
432        SetMetadata {
433            client: self,
434            path: path.as_ref().to_path_buf(),
435            update: MetadataUpdate::new(),
436            follow: true,
437        }
438    }
439}
440
441impl<F: DirectoryFileSystem> FsClient<F> {
442    pub fn create_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
443        self.create_dir_with_mode(path, 0o777)
444    }
445
446    pub fn create_dir_with_mode(&self, path: impl AsRef<Path>, mode: u32) -> VfResult<()> {
447        self.lock()?.create_dir_one(path.as_ref(), mode)
448    }
449
450    pub fn read_dir(&self, path: impl AsRef<Path>) -> VfResult<Vec<DirEntry>> {
451        self.read_dir_with_options(path, self.limits.directory_options())
452    }
453
454    /// Read one directory with explicit entry and path-storage limits.
455    pub fn read_dir_with_options(
456        &self,
457        path: impl AsRef<Path>,
458        options: ReadDirOptions,
459    ) -> VfResult<Vec<DirEntry>> {
460        self.lock()?.read_dir_one(path.as_ref(), options)
461    }
462
463    /// Visit one directory one bounded page at a time. `Continue(())` requests
464    /// the next entry; `Break(())` stops the entire visit successfully.
465    /// The callback runs without the lock and may reenter or drop its files.
466    pub fn visit_dir(
467        &self,
468        path: impl AsRef<Path>,
469        callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
470    ) -> VfResult<TraversalCompletion> {
471        self.visit_dir_with_options(path, self.limits.directory_options(), callback)
472    }
473
474    /// Visit entries with explicit entry and cumulative path-byte limits.
475    pub fn visit_dir_with_options(
476        &self,
477        path: impl AsRef<Path>,
478        options: ReadDirOptions,
479        callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
480    ) -> VfResult<TraversalCompletion> {
481        self.visit_dir_with_fields(path, crate::native::metadata_mask(), options, callback)
482    }
483
484    pub fn visit_dir_with_fields(
485        &self,
486        path: impl AsRef<Path>,
487        fields: AttrMask,
488        options: ReadDirOptions,
489        mut callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
490    ) -> VfResult<TraversalCompletion> {
491        const PAGE_SIZE: usize = 1024;
492        let path = path.as_ref();
493        let mut cursor = None;
494        let mut count = 0usize;
495        let mut path_bytes = 0usize;
496        let max_entries = if options.entry_limit() == usize::MAX {
497            0 // VecFs uses zero for an explicitly unlimited listing.
498        } else {
499            options.entry_limit().saturating_add(1)
500        };
501        loop {
502            // Start with one entry so early-stop callbacks and tight limits
503            // do not trigger a large local scan before application code runs.
504            let page_size = if cursor.is_none() {
505                1
506            } else {
507                PAGE_SIZE.min(
508                    options
509                        .entry_limit()
510                        .saturating_sub(count)
511                        .saturating_add(1),
512                )
513            };
514            let (entries, next) = {
515                self.lock()?.read_dir_page_with_fields(
516                    path,
517                    fields,
518                    cursor,
519                    page_size,
520                    max_entries,
521                )?
522            };
523            if entries.is_empty() && next.is_some() {
524                return Err(VfError::transport(None, "directory page made no progress"));
525            }
526            for entry in entries {
527                if count >= options.entry_limit() {
528                    return Err(
529                        VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
530                    );
531                }
532                path_bytes = path_bytes
533                    .checked_add(entry.path().as_os_str().len())
534                    .ok_or_else(|| {
535                        VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
536                    })?;
537                if path_bytes > options.path_byte_limit() {
538                    return Err(
539                        VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
540                    );
541                }
542                count += 1;
543                if callback(entry)?.is_break() {
544                    return Ok(TraversalCompletion::Stopped);
545                }
546            }
547            match next {
548                Some(value) => cursor = Some(value),
549                None => return Ok(TraversalCompletion::Complete),
550            }
551        }
552    }
553}
554
555impl<F: DirectoryFileSystem + MetadataFileSystem> FsClient<F> {
556    pub fn create_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
557        let path = path.as_ref();
558        let mut current = PathBuf::new();
559        for component in path.components() {
560            match component {
561                Component::RootDir => {
562                    current.push(Path::new("/"));
563                    continue;
564                }
565                Component::Normal(part) => current.push(part),
566                Component::CurDir => continue,
567                Component::ParentDir => {
568                    // Preserve `..` so the backend resolves it relative to
569                    // its own rooted namespace. Lexically popping `/` would
570                    // accidentally turn a later absolute component into a
571                    // current-directory-relative path.
572                    current.push("..");
573                    continue;
574                }
575                Component::Prefix(_) => return Err(VfError::client(0, crate::ERR_INVAL)),
576            }
577            match self.create_dir(&current) {
578                Ok(()) => {}
579                Err(error) if error.err_no() == crate::ERR_EXIST => {
580                    if !self.metadata(&current)?.is_dir() {
581                        return Err(VfError::client(0, crate::ERR_NOTDIR)
582                            .with_context("create_dir_all", &current));
583                    }
584                }
585                Err(error) => return Err(error),
586            }
587        }
588        Ok(())
589    }
590}
591
592impl<F: NamespaceFileSystem + MetadataFileSystem> FsClient<F> {
593    fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Metadata> {
594        let mut filesystem = self.lock()?;
595        let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
596        filesystem
597            .metadata_path(path, follow)
598            .map_err(|error| error.with_context(operation, path))
599    }
600
601    pub fn remove_file(&self, path: impl AsRef<Path>) -> VfResult<()> {
602        let path = path.as_ref();
603        if self.removal_metadata(path, "remove_file")?.is_dir() {
604            return Err(VfError::client(0, crate::ERR_ISDIR).with_context("remove_file", path));
605        }
606        self.lock()?.remove_one(path, false)
607    }
608
609    pub fn remove_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
610        let path = path.as_ref();
611        if !self.removal_metadata(path, "remove_dir")?.is_dir() {
612            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir", path));
613        }
614        self.lock()?.remove_one(path, false)
615    }
616
617    pub fn remove_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
618        let path = path.as_ref();
619        if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
620            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
621        }
622        self.lock()?.remove_one(path, true)
623    }
624
625    /// Remove the contents of a directory, keeping the directory itself.
626    pub fn remove_dir_contents(&self, path: impl AsRef<Path>) -> VfResult<()> {
627        let path = path.as_ref();
628        if !self.removal_metadata(path, "remove_dir_contents")?.is_dir() {
629            return Err(
630                VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
631            );
632        }
633        self.lock()?.remove_dir_contents(path)
634    }
635
636    pub fn rename(&self, from: impl AsRef<Path>, to: impl AsRef<Path>) -> VfResult<()> {
637        self.lock()?.rename_one(from.as_ref(), to.as_ref())
638    }
639}
640
641impl<F: VecFs> FsClient<F> {
642    /// Rename independent source/destination pairs in one vector phase.
643    pub fn renamev<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
644        if pairs.is_empty() {
645            return Ok(());
646        }
647        let requests: Vec<_> = pairs
648            .iter()
649            .map(|(from, to)| {
650                (
651                    VfFile::from_os_path(from.as_ref()),
652                    VfFile::from_os_path(to.as_ref()),
653                )
654            })
655            .collect();
656        self.lock()?
657            .renamev(&requests)
658            .map_err(|error| match error.index() {
659                Some(index) if index < pairs.len() => {
660                    error.with_context("renamev", pairs[index].0.as_ref())
661                }
662                Some(_) => {
663                    VfError::transport(None, "rename backend returned an invalid error index")
664                }
665                None => error,
666            })
667    }
668
669    /// Create directories in input order using vector MKDIR. Parents must
670    /// already exist. This is not transactional: failure may leave a prefix.
671    pub fn mkdirv<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<()> {
672        let mut seen = HashSet::with_capacity(paths.len());
673        for (index, path) in paths.iter().enumerate() {
674            if !seen.insert(path.as_ref()) {
675                return Err(
676                    VfError::client(index, crate::ERR_INVAL).with_context("mkdirv", path.as_ref())
677                );
678            }
679        }
680        if paths.is_empty() {
681            return Ok(());
682        }
683        let dirs: Vec<_> = paths
684            .iter()
685            .map(|path| crate::VfAttrs {
686                file: VfFile::from_os_path(path.as_ref()),
687                masks: AttrMask::MODE,
688                mode: 0o777,
689                ..Default::default()
690            })
691            .collect();
692        self.lock()?.mkdirv(&dirs).map_err(|error| {
693            match error.index().and_then(|index| paths.get(index)) {
694                Some(path) => error.with_context("mkdirv", path.as_ref()),
695                None => error,
696            }
697        })
698    }
699
700    /// Lazy between directories, with selective no-follow metadata and pruning.
701    /// Enter runs before any listing; Leave follows even a pruned directory.
702    /// Sorting buffers one bounded directory, not the whole tree. The callback
703    /// runs outside the backend lock. Limits include the starting object.
704    pub fn walk_events_with_options(
705        &self,
706        root: impl AsRef<Path>,
707        fields: AttrMask,
708        options: crate::WalkOptions,
709        sort_by_name: bool,
710        callback: impl FnMut(&crate::WalkEvent) -> VfResult<crate::WalkControl>,
711    ) -> VfResult<TraversalCompletion> {
712        let root = root.as_ref();
713        let metadata = self.symlink_metadata_with_fields(root, fields | AttrMask::MODE)?;
714        crate::walk_events(
715            DirEntry::new(root.to_path_buf(), metadata),
716            options,
717            sort_by_name,
718            |path, limits| {
719                let mut listings =
720                    self.read_dirs_with_options(&[path], fields | AttrMask::MODE, limits)?;
721                if listings.len() != 1 {
722                    return Err(VfError::transport(None, "invalid directory result count"));
723                }
724                Ok(listings.remove(0).entries)
725            },
726            callback,
727        )
728    }
729    /// Visit a tree using bounded directory pages,
730    /// never follows symlinks, and invokes the callback outside the backend
731    /// lock. Unlike collecting `walk`, this trades multi-directory batching
732    /// for bounded incremental delivery. A backend without native paging
733    /// may retain one bounded listing; finish its pages before descending,
734    /// so snapshots never accumulate across ancestor directories.
735    /// `ControlFlow::Break(())` stops the entire walk successfully, not merely
736    /// the current subtree. Directory order is backend-defined.
737    pub fn visit_walk(
738        &self,
739        root: impl AsRef<Path>,
740        callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
741    ) -> VfResult<TraversalCompletion> {
742        self.visit_walk_with_options(root, self.limits.walk_options(), callback)
743    }
744
745    pub fn visit_walk_with_options(
746        &self,
747        root: impl AsRef<Path>,
748        options: crate::WalkOptions,
749        callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
750    ) -> VfResult<TraversalCompletion> {
751        self.visit_walk_with_fields(root, crate::native::metadata_mask(), options, callback)
752    }
753
754    pub fn visit_walk_with_fields(
755        &self,
756        root: impl AsRef<Path>,
757        fields: AttrMask,
758        options: crate::WalkOptions,
759        callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
760    ) -> VfResult<TraversalCompletion> {
761        let root = root.as_ref();
762        if !self.symlink_metadata(root)?.is_dir() {
763            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("visit_walk", root));
764        }
765        visit_walk_pages(
766            root,
767            options,
768            |path, cursor, page_size, max_entries| {
769                self.lock()?
770                    .read_dir_page_with_fields(path, fields, cursor, page_size, max_entries)
771            },
772            callback,
773        )
774    }
775    /// Maximum safe cohort size for incremental directory paging.
776    pub fn directory_page_batch_size(&self) -> VfResult<usize> {
777        Ok(self.lock()?.directory_page_batch_size().clamp(1, 32))
778    }
779    /// Fetch a bounded vector of directory pages, retaining backend cursors.
780    pub fn read_dir_pages_with_fields(
781        &self,
782        paths: &[&Path],
783        fields: AttrMask,
784        cursors: Vec<Option<crate::DirPageCursor>>,
785        page_size: usize,
786        max_entries: usize,
787    ) -> VfResult<Vec<crate::DirectoryPage>> {
788        let pages = self.lock()?.listdir_pages(
789            paths,
790            fields | AttrMask::MODE | AttrMask::SIZE,
791            cursors,
792            page_size,
793            max_entries,
794        )?;
795        if pages.len() != paths.len() {
796            return Err(VfError::transport(
797                None,
798                "directory pages returned an invalid result count",
799            ));
800        }
801        pages
802            .into_iter()
803            .zip(paths)
804            .enumerate()
805            .map(|(index, ((attrs, next, children), path))| {
806                if children.len() != attrs.len()
807                    || attrs.len() > page_size
808                    || (attrs.is_empty() && next.is_some())
809                {
810                    return Err(VfError::transport(
811                        Some(index),
812                        "invalid directory page progress",
813                    ));
814                }
815                let mut seeds = Vec::new();
816                let entries = attrs
817                    .into_iter()
818                    .zip(children)
819                    .map(|(attrs, child)| {
820                        let entry_path = attrs
821                            .file
822                            .path()
823                            .ok_or_else(|| {
824                                VfError::transport(Some(index), "directory entry has no path")
825                            })?
826                            .to_path_buf();
827                        if entry_path.parent() != Some(*path) {
828                            return Err(VfError::transport(
829                                Some(index),
830                                "directory entry is outside its parent",
831                            ));
832                        }
833                        if let Some(child) = child {
834                            seeds.push((entry_path.clone(), child));
835                        }
836                        Ok(DirEntry::new(
837                            entry_path,
838                            vfsi_core::metadata_from_attrs(attrs),
839                        ))
840                    })
841                    .collect::<VfResult<Vec<_>>>()?;
842                Ok((
843                    DirectoryListing {
844                        path: path.to_path_buf(),
845                        entries,
846                    },
847                    next,
848                    seeds,
849                ))
850            })
851            .collect()
852    }
853
854    /// List several directories with common stat attributes and finite
855    /// allocation limits. Use `read_dirs_with_options` for richer fields.
856    pub fn read_dirs<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<DirectoryListing>> {
857        self.read_dirs_with_options(paths, AttrMask::stat(), self.limits.directory_options())
858    }
859}
860
861type DirectoryPage = (Vec<DirEntry>, Option<crate::DirPageCursor>);
862
863fn visit_walk_pages(
864    root: &Path,
865    options: crate::WalkOptions,
866    mut read_page: impl FnMut(
867        &Path,
868        Option<crate::DirPageCursor>,
869        usize,
870        usize,
871    ) -> VfResult<DirectoryPage>,
872    mut callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
873) -> VfResult<TraversalCompletion> {
874    let mut pending = vec![(root.to_path_buf(), 0usize)];
875    let mut count = 0usize;
876    let mut path_bytes = root.as_os_str().len();
877    if path_bytes > options.path_byte_limit() {
878        return Err(VfError::client(0, libc::EFBIG as u32));
879    }
880    while let Some((path, depth)) = pending.pop() {
881        let mut cursor = None;
882        let mut children = Vec::new();
883        loop {
884            let requested = options
885                .entry_limit()
886                .saturating_sub(count)
887                .saturating_add(1);
888            let (entries, next) = read_page(&path, cursor, requested.min(128), requested)?;
889            if entries.is_empty() && next.is_some() {
890                return Err(VfError::transport(None, "directory page made no progress")
891                    .with_context("visit_walk", &path));
892            }
893            for entry in entries {
894                count = count
895                    .checked_add(1)
896                    .ok_or_else(|| VfError::client(0, libc::EFBIG as u32))?;
897                path_bytes = path_bytes
898                    .checked_add(entry.path().as_os_str().len())
899                    .ok_or_else(|| VfError::client(0, libc::EFBIG as u32))?;
900                if count > options.entry_limit() || path_bytes > options.path_byte_limit() {
901                    return Err(
902                        VfError::client(0, libc::EFBIG as u32).with_context("visit_walk", &path)
903                    );
904                }
905                if callback(&entry)?.is_break() {
906                    return Ok(TraversalCompletion::Stopped);
907                }
908                if entry.metadata().is_dir() {
909                    if depth >= options.depth_limit() {
910                        if !options.truncates_at_depth_limit() {
911                            return Err(VfError::client(0, libc::EFBIG as u32)
912                                .with_context("visit_walk", entry.path()));
913                        }
914                    } else {
915                        children.push((entry.path().to_path_buf(), depth + 1));
916                    }
917                }
918            }
919            cursor = next;
920            if cursor.is_none() {
921                break;
922            }
923        }
924        pending.extend(children.into_iter().rev());
925    }
926    Ok(TraversalCompletion::Complete)
927}
928
929impl<F: VecFs> FsClient<F> {
930    /// Recursively enumerate a bounded tree with common stat attributes.
931    /// Use `walk_with_options` to select fields or change limits.
932    pub fn walk(&self, root: impl AsRef<Path>) -> VfResult<Vec<DirectoryListing>> {
933        self.walk_with_options(root, AttrMask::stat(), self.limits.walk_options())
934    }
935
936    /// Create `path` if missing, otherwise empty it. Errors if it exists and is
937    /// not a directory (a symlink to a directory is not a directory here).
938    pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
939        self.lock()?.ensure_empty_dir(path.as_ref())
940    }
941
942    /// Remove a directory tree with explicit error, batching, and retry policy.
943    pub fn remove_dir_all_with_options(
944        &self,
945        path: impl AsRef<Path>,
946        options: RemoveOptions,
947    ) -> VfResult<()> {
948        let path = path.as_ref();
949        if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
950            return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
951        }
952        self.lock()?
953            .rm_with_options(&[path], true, options)
954            .map_err(|error| error.with_context("remove_dir_all", path))
955    }
956
957    /// Empty a directory while keeping it, with explicit removal policy.
958    pub fn remove_dir_contents_with_options(
959        &self,
960        path: impl AsRef<Path>,
961        options: RemoveOptions,
962    ) -> VfResult<()> {
963        let path = path.as_ref();
964        if !self.removal_metadata(path, "remove_dir_contents")?.is_dir() {
965            return Err(
966                VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
967            );
968        }
969        self.lock()?
970            .rm_contents_with_options(path, options)
971            .map_err(|error| error.with_context("remove_dir_contents", path))
972    }
973
974    /// Open a *genuine* directory handle for race-resistant, handle-rooted
975    /// removal. Backends that only return a path fail instead of silently
976    /// losing the handle safety guarantee.
977    pub fn open_dir_handle(&self, path: impl AsRef<Path>) -> VfResult<FsDir<F>> {
978        let path = path.as_ref();
979        let mut backend = self.lock()?;
980        let dir = backend.open_dir(path)?;
981        if !matches!(dir, VfDir::Descriptor { .. }) {
982            let _ = backend.close_dir(&dir);
983            return Err(VfError::unsupported(0).with_context("open_dir_handle", path));
984        }
985        Ok(FsDir {
986            inner: Arc::clone(&self.inner),
987            dir: Some(dir),
988            path: path.to_path_buf(),
989        })
990    }
991}
992
993/// Owned, handle-rooted directory. Dropping it queues backend cleanup;
994/// [`close`](Self::close) reports cleanup errors explicitly.
995pub struct FsDir<F: VecFs> {
996    inner: Arc<SharedBackend<F>>,
997    dir: Option<VfDir>,
998    path: PathBuf,
999}
1000
1001impl<F: VecFs> fmt::Debug for FsDir<F> {
1002    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1003        formatter
1004            .debug_struct("FsDir")
1005            .field("path", &self.path)
1006            .finish_non_exhaustive()
1007    }
1008}
1009
1010impl<F: VecFs> FsDir<F> {
1011    pub fn is_closed(&self) -> bool {
1012        self.dir.is_none()
1013    }
1014    /// Name used at open, retained for diagnostics; not updated after rename.
1015    pub fn path(&self) -> &Path {
1016        &self.path
1017    }
1018
1019    pub fn remove_contents(&self) -> VfResult<()> {
1020        self.remove_contents_with_options(RemoveOptions::default())
1021    }
1022
1023    pub fn remove_contents_with_options(&self, options: RemoveOptions) -> VfResult<()> {
1024        let dir = self
1025            .dir
1026            .as_ref()
1027            .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))?;
1028        self.inner
1029            .lock()
1030            .map_err(|_| poisoned())?
1031            .rm_dir_contents_with_options(dir, options)
1032            .map_err(|error| error.with_context("remove_dir_contents", &self.path))
1033    }
1034
1035    pub fn close(mut self) -> VfResult<()> {
1036        self.try_close()
1037    }
1038
1039    /// Keep cleanup ownership on failure so the caller can retry explicitly.
1040    pub fn try_close(&mut self) -> VfResult<()> {
1041        let Some(dir) = self.dir.as_ref() else {
1042            return Ok(());
1043        };
1044        self.inner.lock().map_err(|_| poisoned())?.close_dir(dir)?;
1045        self.dir = None;
1046        Ok(())
1047    }
1048}
1049
1050impl<F: VecFs> Drop for FsDir<F> {
1051    fn drop(&mut self) {
1052        if let Some(dir) = self.dir.take() {
1053            self.inner.defer(
1054                CleanupTarget::Directory(dir, std::mem::take(&mut self.path)),
1055                |backend, target| {
1056                    let CleanupTarget::Directory(dir, path) = target else {
1057                        unreachable!()
1058                    };
1059                    backend
1060                        .close_dir(dir)
1061                        .map_err(|error| error.with_context("close_dir", path))
1062                },
1063            );
1064        }
1065    }
1066}
1067
1068impl<F: LinkFileSystem> FsClient<F> {
1069    pub fn symlink(&self, target: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
1070        self.lock()?.symlink_one(target.as_ref(), link.as_ref())
1071    }
1072
1073    pub fn hard_link(&self, source: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
1074        self.lock()?.hard_link_one(source.as_ref(), link.as_ref())
1075    }
1076
1077    pub fn read_link(&self, path: impl AsRef<Path>) -> VfResult<PathBuf> {
1078        self.lock()?.read_link_one(path.as_ref())
1079    }
1080}
1081
1082impl<F: CopyFileSystem> FsClient<F> {
1083    pub fn copy(&self, source: impl AsRef<Path>, destination: impl AsRef<Path>) -> VfResult<()> {
1084        self.lock()?.copy_one(source.as_ref(), destination.as_ref())
1085    }
1086}
1087
1088impl<F: VecFs> FsClient<F> {
1089    /// Fetch selected metadata for one path without following its final symlink.
1090    /// Unavailable fields remain `None` on [`Metadata`].
1091    pub fn symlink_metadata_with_fields(
1092        &self,
1093        path: impl AsRef<Path>,
1094        fields: AttrMask,
1095    ) -> VfResult<Metadata> {
1096        let path = path.as_ref();
1097        let mut attrs = crate::VfAttrs {
1098            file: VfFile::from_os_path(path),
1099            masks: fields | AttrMask::MODE | AttrMask::SIZE,
1100            ..crate::VfAttrs::default()
1101        };
1102        self.lock()?
1103            .lgetattrsv(std::slice::from_mut(&mut attrs))
1104            .map_err(|error| error.with_context("symlink_metadata", path))?;
1105        Ok(vfsi_core::metadata_from_attrs(attrs))
1106    }
1107
1108    /// List multiple directories in a vector call. The limits apply to the
1109    /// aggregate returned entries and stored path bytes. Streaming backends
1110    /// apply these limits before collecting a full listing; a backend using
1111    /// the compatibility `visit_dir` fallback may buffer one directory first.
1112    /// An error discards the collected prefix; callers may retry individual
1113    /// directories if desired.
1114    pub fn read_dirs_with_options<P: AsRef<Path>>(
1115        &self,
1116        paths: &[P],
1117        fields: AttrMask,
1118        options: ReadDirOptions,
1119    ) -> VfResult<Vec<DirectoryListing>> {
1120        if paths.is_empty() {
1121            return Ok(Vec::new());
1122        }
1123        let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
1124        let mut positions: HashMap<PathBuf, Vec<usize>> = HashMap::with_capacity(paths.len());
1125        let mut unique_paths = Vec::with_capacity(paths.len());
1126        for (index, path) in paths.iter().enumerate() {
1127            if let Some(indices) = positions.get_mut(*path) {
1128                indices.push(index);
1129            } else {
1130                positions.insert(path.to_path_buf(), vec![index]);
1131                unique_paths.push(*path);
1132            }
1133        }
1134        let mut listings: Vec<DirectoryListing> = paths
1135            .iter()
1136            .map(|path| DirectoryListing {
1137                path: path.to_path_buf(),
1138                entries: Vec::new(),
1139            })
1140            .collect();
1141        let mut entry_count = 0usize;
1142        let mut path_bytes = 0usize;
1143        // Bound the number of simultaneous first READDIR pages even when the
1144        // caller supplies thousands of directory operands.
1145        const DIRECTORY_COHORT: usize = 32;
1146        for cohort in unique_paths.chunks(DIRECTORY_COHORT) {
1147            let mut callback_error = None;
1148            let requested = options
1149                .entry_limit()
1150                .saturating_sub(entry_count)
1151                .saturating_add(1);
1152            let result = self.lock()?.listdirv(
1153                cohort,
1154                fields | AttrMask::MODE | AttrMask::SIZE,
1155                requested,
1156                false,
1157                &mut |attributes, directory| {
1158                    let Some(indices) = positions.get(directory) else {
1159                        callback_error = Some(VfError::transport(
1160                            None,
1161                            format!(
1162                                "read_dirs backend returned an unexpected directory: {}",
1163                                directory.display()
1164                            ),
1165                        ));
1166                        return false;
1167                    };
1168                    if !cohort.contains(&directory) {
1169                        callback_error = Some(VfError::transport(
1170                            None,
1171                            "read_dirs backend returned a directory outside the active cohort",
1172                        ));
1173                        return false;
1174                    }
1175                    let Some(path) = attributes.file.path() else {
1176                        callback_error = Some(VfError::transport(
1177                            None,
1178                            "read_dirs backend returned an entry without a path",
1179                        ));
1180                        return false;
1181                    };
1182                    if path.parent() != Some(directory) {
1183                        callback_error = Some(VfError::transport(
1184                            None,
1185                            format!(
1186                                "read_dirs backend returned an entry outside {}",
1187                                directory.display()
1188                            ),
1189                        ));
1190                        return false;
1191                    }
1192                    for &index in indices {
1193                        entry_count += 1;
1194                        path_bytes = path_bytes.saturating_add(path.as_os_str().len());
1195                        if entry_count > options.entry_limit()
1196                            || path_bytes > options.path_byte_limit()
1197                        {
1198                            callback_error = Some(
1199                                VfError::failure(index, libc::EFBIG as u32)
1200                                    .with_context("read_dirs", directory),
1201                            );
1202                            return false;
1203                        }
1204                        listings[index].entries.push(DirEntry::new(
1205                            path.to_path_buf(),
1206                            vfsi_core::metadata_from_attrs(attributes.clone()),
1207                        ));
1208                    }
1209                    true
1210                },
1211            );
1212            if let Some(error) = callback_error {
1213                return Err(error);
1214            }
1215            result.map_err(|error| {
1216                error.map_index(|index| {
1217                    cohort
1218                        .get(index)
1219                        .and_then(|path| positions.get(*path))
1220                        .and_then(|indices| indices.first())
1221                        .copied()
1222                        .unwrap_or(index)
1223                })
1224            })?;
1225        }
1226        Ok(listings)
1227    }
1228
1229    /// Recursively enumerate directories with selected entry attributes.
1230    /// The walk is bounded by `options`; sorting and presentation remain the
1231    /// application's responsibility.
1232    pub fn walk_with_options(
1233        &self,
1234        root: impl AsRef<Path>,
1235        fields: AttrMask,
1236        options: crate::WalkOptions,
1237    ) -> VfResult<Vec<DirectoryListing>> {
1238        let root = root.as_ref();
1239        let tree = self.lock()?.walk_with_options(
1240            root,
1241            fields | AttrMask::MODE | AttrMask::SIZE,
1242            options,
1243            &mut |_, _| {},
1244        )?;
1245        tree.into_iter()
1246            .map(|directory| {
1247                let entries = directory
1248                    .entries
1249                    .into_iter()
1250                    .enumerate()
1251                    .map(|(index, attributes)| {
1252                        let path = attributes
1253                            .file
1254                            .path()
1255                            .ok_or_else(|| {
1256                                VfError::transport(
1257                                    None,
1258                                    format!("walk backend returned entry {index} without a path"),
1259                                )
1260                            })?
1261                            .to_path_buf();
1262                        Ok(DirEntry::new(
1263                            path,
1264                            vfsi_core::metadata_from_attrs(attributes),
1265                        ))
1266                    })
1267                    .collect::<VfResult<Vec<_>>>()?;
1268                Ok(DirectoryListing {
1269                    path: directory.path,
1270                    entries,
1271                })
1272            })
1273            .collect()
1274    }
1275
1276    /// Copy whole files in request order. A successful prefix may remain if
1277    /// a later request fails; this operation does not provide atomicity.
1278    pub fn copyv<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
1279        let extents: Vec<_> = pairs
1280            .iter()
1281            .map(|(from, to)| {
1282                crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
1283            })
1284            .collect();
1285        self.lock()?.copyv(&extents).map_err(|error| {
1286            error
1287                .index()
1288                .and_then(|index| pairs.get(index))
1289                .map_or(error.clone(), |(_, to)| {
1290                    error.with_context("copyv", to.as_ref())
1291                })
1292        })
1293    }
1294
1295    /// Remove paths in request order, optionally recursing into directories.
1296    /// A successful prefix may remain if a later path fails.
1297    pub fn remove_paths<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
1298        self.remove_paths_with_options(paths, recursive, RemoveOptions::default())
1299    }
1300
1301    /// Remove paths with explicit error, batching, and retry policy.
1302    pub fn remove_paths_with_options<P: AsRef<Path>>(
1303        &self,
1304        paths: &[P],
1305        recursive: bool,
1306        options: RemoveOptions,
1307    ) -> VfResult<()> {
1308        let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
1309        self.lock()?
1310            .rm_with_options(&paths, recursive, options)
1311            .map_err(|error| {
1312                error
1313                    .index()
1314                    .and_then(|index| paths.get(index))
1315                    .map_or(error.clone(), |path| {
1316                        error.with_context("remove_paths", path)
1317                    })
1318            })
1319    }
1320
1321    /// Vector metadata query with explicit fields and final-symlink handling.
1322    /// Ancestor symlinks follow the backend's normal namespace semantics.
1323    pub fn metadata_many<P: AsRef<Path>>(
1324        &self,
1325        paths: &[P],
1326        fields: AttrMask,
1327        follow: bool,
1328    ) -> VfResult<Vec<Metadata>> {
1329        if paths.is_empty() {
1330            return Ok(Vec::new());
1331        }
1332        let mut attrs: Vec<_> = paths
1333            .iter()
1334            .map(|path| crate::VfAttrs {
1335                file: VfFile::from_os_path(path.as_ref()),
1336                masks: fields | AttrMask::MODE,
1337                ..crate::VfAttrs::default()
1338            })
1339            .collect();
1340        let result = {
1341            let mut backend = self.lock()?;
1342            if follow {
1343                backend.getattrsv(&mut attrs)
1344            } else {
1345                backend.lgetattrsv(&mut attrs)
1346            }
1347        };
1348        result.map_err(|error| match error.index() {
1349            Some(index) if index < paths.len() => {
1350                error.with_context("metadatav", paths[index].as_ref())
1351            }
1352            Some(_) => VfError::transport(None, "metadata backend returned an invalid error index"),
1353            None => error,
1354        })?;
1355        Ok(attrs
1356            .into_iter()
1357            .map(vfsi_core::metadata_from_attrs)
1358            .collect())
1359    }
1360
1361    /// Fetch no-follow metadata for many paths using the backend vector operation.
1362    pub fn symlink_metadatav(&self, paths: &[&Path]) -> VfResult<Vec<Metadata>> {
1363        let mut attrs: Vec<_> = paths
1364            .iter()
1365            .map(|path| crate::VfAttrs {
1366                file: VfFile::from_os_path(path),
1367                masks: AttrMask::MODE,
1368                ..crate::VfAttrs::default()
1369            })
1370            .collect();
1371        self.lock()?.lgetattrsv(&mut attrs).map_err(|error| {
1372            error
1373                .index()
1374                .and_then(|index| paths.get(index))
1375                .map_or(error.clone(), |path| {
1376                    error.with_context("symlink_metadatav", path)
1377                })
1378        })?;
1379        Ok(attrs
1380            .into_iter()
1381            .map(vfsi_core::metadata_from_attrs)
1382            .collect())
1383    }
1384}
1385
1386impl<F: VectorFileSystem> FsClient<F> {
1387    /// Open an ordered vector of files.
1388    ///
1389    /// Success returns one RAII handle per request. Failure returns no
1390    /// handles; VFSI does not promise transactional rollback of other
1391    /// filesystem effects such as file creation.
1392    pub fn openv(&self, requests: &[OpenRequest]) -> VfResult<Vec<FsFile<F>>> {
1393        let mut filesystem = self.lock()?;
1394        let files = filesystem.open_many(requests).map_err(|error| {
1395            error
1396                .index()
1397                .and_then(|index| requests.get(index))
1398                .map_or(error.clone(), |request| {
1399                    error.with_context("openv", &request.path)
1400                })
1401        })?;
1402        if files.len() != requests.len() {
1403            let error = wrong_result_count("openv", requests.len(), files.len());
1404            for file in &files {
1405                let _ = filesystem.close_one(file);
1406            }
1407            return Err(error);
1408        }
1409        drop(filesystem);
1410        Ok(files
1411            .into_iter()
1412            .zip(requests)
1413            .map(|(file, request)| FsFile {
1414                inner: Arc::clone(&self.inner),
1415                file: Some(file),
1416                path: request.path.clone(),
1417            })
1418            .collect())
1419    }
1420
1421    /// Try to close a group through one vector operation without consuming
1422    /// the handles. On failure, all handles remain armed: the backend may
1423    /// have closed a prefix, so callers must reconcile before retrying.
1424    pub fn try_closev<'a>(&self, files: impl IntoIterator<Item = &'a mut FsFile<F>>) -> VfResult<()>
1425    where
1426        F: 'a,
1427    {
1428        let mut files: Vec<_> = files.into_iter().collect();
1429        for (index, file) in files.iter().enumerate() {
1430            self.validate_owner(file, index)?;
1431        }
1432        let positions: Vec<_> = files
1433            .iter()
1434            .enumerate()
1435            .filter_map(|(index, file)| (!file.is_closed()).then_some(index))
1436            .collect();
1437        files.retain(|file| !file.is_closed());
1438        if files.is_empty() {
1439            return Ok(());
1440        }
1441        let descriptors: Vec<VfFile> = files
1442            .iter()
1443            .map(|file| file.raw().cloned())
1444            .collect::<VfResult<_>>()?;
1445        self.lock()?.close_many(&descriptors).map_err(|error| {
1446            error
1447                .index()
1448                .and_then(|index| files.get(index))
1449                .map_or(error.clone(), |file| {
1450                    error.with_context("closev", file.path())
1451                })
1452                .map_index(|index| positions.get(index).copied().unwrap_or(index))
1453        })?;
1454        for file in &mut files {
1455            file.file = None;
1456        }
1457        Ok(())
1458    }
1459
1460    /// Close a group of files through one vector operation.
1461    ///
1462    /// On failure, the handles are dropped and the backend receives
1463    /// best-effort scalar cleanup attempts. Use [`try_closev`](Self::try_closev)
1464    /// to retain the handles after an error.
1465    pub fn closev(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
1466        self.try_closev(&mut files)
1467    }
1468
1469    /// Read an ordered vector with a 16 MiB aggregate request limit.
1470    /// Use [`readv_with_limit`](Self::readv_with_limit) to tune the limit or
1471    /// [`readv_into`](Self::readv_into) to provide bounded caller-owned buffers.
1472    pub fn readv(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<FsReadResult>> {
1473        self.readv_with_limit(requests, self.limits.max_read_bytes)
1474    }
1475
1476    /// Read an ordered vector with an explicit aggregate request limit.
1477    pub fn readv_with_limit(
1478        &self,
1479        requests: &[FsRead<'_, F>],
1480        max_total_bytes: usize,
1481    ) -> VfResult<Vec<FsReadResult>> {
1482        self.readv_with_limit_projected(requests, max_total_bytes, |request| request)
1483    }
1484
1485    /// Backend adapter for opaque application requests; projection does not allocate.
1486    /// The projection must return the same embedded request on every invocation.
1487    pub fn readv_with_limit_projected<'b, T>(
1488        &self,
1489        requests: &[T],
1490        max_total_bytes: usize,
1491        project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
1492    ) -> VfResult<Vec<FsReadResult>>
1493    where
1494        F: 'b,
1495    {
1496        let mut requested = 0usize;
1497        for (index, request) in requests.iter().map(&project).enumerate() {
1498            requested = requested
1499                .checked_add(request.length)
1500                .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1501            if requested > max_total_bytes {
1502                return Err(VfError::client(index, libc::EFBIG as u32)
1503                    .with_context("readv", request.file.path()));
1504            }
1505        }
1506        let reads = self.read_ops(requests, &project)?;
1507        let results = self.lock()?.read_many(&reads).map_err(|error| {
1508            error
1509                .index()
1510                .and_then(|index| requests.get(index).map(&project))
1511                .map_or(error.clone(), |request| {
1512                    error.with_context("readv", request.file.path())
1513                })
1514        })?;
1515        if results.len() != requests.len() {
1516            return Err(wrong_result_count("readv", requests.len(), results.len()));
1517        }
1518        validate_read_results("readv", &reads, &results).map_err(|error| {
1519            error
1520                .index()
1521                .and_then(|index| requests.get(index).map(&project))
1522                .map_or(error.clone(), |request| {
1523                    error.with_context("readv", request.file.path())
1524                })
1525        })?;
1526        Ok(results.into_iter().map(read_result).collect())
1527    }
1528
1529    fn read_ops<'b, T>(
1530        &self,
1531        requests: &[T],
1532        project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
1533    ) -> VfResult<Vec<ReadOp>>
1534    where
1535        F: 'b,
1536    {
1537        let mut reads = Vec::with_capacity(requests.len());
1538        for (index, request) in requests.iter().map(&project).enumerate() {
1539            self.validate_owner(request.file, index)?;
1540            reads.push(ReadOp::new(
1541                request.file.raw()?.clone(),
1542                request.offset,
1543                request.length,
1544            ));
1545        }
1546        Ok(reads)
1547    }
1548
1549    pub fn readv_into(
1550        &self,
1551        requests: &mut [FsReadInto<'_, F>],
1552    ) -> VfResult<Vec<FsReadIntoResult>> {
1553        self.readv_into_with_limit(requests, self.limits.max_read_bytes)
1554    }
1555
1556    /// Read into caller storage with an explicit aggregate buffer budget.
1557    /// This also bounds allocation in copying fallback implementations.
1558    pub fn readv_into_with_limit(
1559        &self,
1560        requests: &mut [FsReadInto<'_, F>],
1561        max_bytes: usize,
1562    ) -> VfResult<Vec<FsReadIntoResult>> {
1563        self.readv_into_with_limit_projected(
1564            requests,
1565            max_bytes,
1566            |request| request,
1567            |request| request,
1568        )
1569    }
1570
1571    /// Project borrowed buffers without an intermediate request allocation.
1572    /// Both projections must identify the same embedded request, and remain
1573    /// stable across preflight, dispatch, and result validation.
1574    pub fn readv_into_with_limit_projected<'b, T>(
1575        &self,
1576        requests: &mut [T],
1577        max_bytes: usize,
1578        project: impl for<'r> Fn(&'r T) -> &'r FsReadInto<'b, F>,
1579        mut project_mut: impl for<'r> FnMut(&'r mut T) -> &'r mut FsReadInto<'b, F>,
1580    ) -> VfResult<Vec<FsReadIntoResult>>
1581    where
1582        F: 'b,
1583    {
1584        let mut requested = 0usize;
1585        let mut reads = Vec::with_capacity(requests.len());
1586        for (index, request) in requests.iter().map(&project).enumerate() {
1587            self.validate_owner(request.file, index)?;
1588            requested = requested
1589                .checked_add(request.buffer.len())
1590                .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1591            if requested > max_bytes {
1592                return Err(VfError::client(index, libc::EFBIG as u32));
1593            }
1594            reads.push(ReadOp::new(
1595                request.file.raw()?.clone(),
1596                request.offset,
1597                request.buffer.len(),
1598            ));
1599        }
1600        let results = {
1601            let mut buffers: Vec<&mut [u8]> = requests
1602                .iter_mut()
1603                .map(&mut project_mut)
1604                .map(|request| &mut *request.buffer)
1605                .collect();
1606            self.lock()?.read_many_into(&reads, &mut buffers)
1607        }
1608        .map_err(|error| {
1609            error
1610                .index()
1611                .and_then(|index| requests.get(index).map(&project))
1612                .map_or(error.clone(), |request| {
1613                    error.with_context("readv_into", request.file.path())
1614                })
1615        })?;
1616        validate_read_into_results("readv_into", &reads, &results).map_err(|error| {
1617            error
1618                .index()
1619                .and_then(|index| requests.get(index).map(&project))
1620                .map_or(error.clone(), |request| {
1621                    error.with_context("readv_into", request.file.path())
1622                })
1623        })?;
1624        Ok(results
1625            .into_iter()
1626            .map(|result| FsReadIntoResult {
1627                offset: result.offset,
1628                read: result.read,
1629                eof: result.eof,
1630            })
1631            .collect())
1632    }
1633
1634    pub fn writev(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1635        self.writev_projected(requests, |request| request)
1636    }
1637
1638    /// Backend adapter for opaque application requests, preserving borrowed payloads.
1639    /// The projection must return the same embedded request on every invocation.
1640    pub fn writev_projected<'b, T>(
1641        &self,
1642        requests: &[T],
1643        project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1644    ) -> VfResult<Vec<FsWriteResult>>
1645    where
1646        F: 'b,
1647    {
1648        self.writev_mapped(requests, |item| {
1649            let request = project(item);
1650            FsWrite {
1651                file: request.file,
1652                offset: request.offset,
1653                data: request.data,
1654            }
1655        })
1656    }
1657
1658    /// Adapter constructing cheap borrowed requests without a temporary vector.
1659    /// The mapper must return the same file, offset, and payload on each call.
1660    #[doc(hidden)]
1661    pub fn writev_mapped<'b, T>(
1662        &self,
1663        requests: &[T],
1664        project: impl Fn(&T) -> FsWrite<'b, F>,
1665    ) -> VfResult<Vec<FsWriteResult>>
1666    where
1667        F: 'b,
1668    {
1669        let writes = self.write_ops(requests, &project)?;
1670        let results = self.lock()?.write_many(&writes).map_err(|error| {
1671            error
1672                .index()
1673                .and_then(|index| requests.get(index).map(&project))
1674                .map_or(error.clone(), |request| {
1675                    error.with_context("writev", request.file.path())
1676                })
1677        })?;
1678        if results.len() != requests.len() {
1679            return Err(wrong_result_count("writev", requests.len(), results.len()));
1680        }
1681        validate_write_results("writev", &writes, &results).map_err(|error| {
1682            error
1683                .index()
1684                .and_then(|index| requests.get(index).map(&project))
1685                .map_or(error.clone(), |request| {
1686                    error.with_context("writev", request.file.path())
1687                })
1688        })?;
1689        Ok(results.into_iter().map(write_result).collect())
1690    }
1691
1692    /// Write every byte in each positional request, retrying short writes in
1693    /// vector waves. Like `writev`, this is not transactional: an error may
1694    /// follow a successfully written prefix. Overlapping requests through the
1695    /// same path complete in input order; different paths are presumed
1696    /// independent (including hard-link aliases).
1697    pub fn write_allv(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1698        self.write_allv_projected(requests, |request| request)
1699    }
1700
1701    /// Complete projected requests with the same preflight and dependency waves.
1702    /// The projection must return the same embedded request on every invocation.
1703    pub fn write_allv_projected<'b, T>(
1704        &self,
1705        requests: &[T],
1706        project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1707    ) -> VfResult<Vec<FsWriteResult>>
1708    where
1709        F: 'b,
1710    {
1711        self.write_allv_mapped(requests, |item| {
1712            let request = project(item);
1713            FsWrite {
1714                file: request.file,
1715                offset: request.offset,
1716                data: request.data,
1717            }
1718        })
1719    }
1720
1721    /// Complete mapped writes with whole-batch preflight and dependency waves.
1722    /// The mapper must return the same file, offset, and payload on each call.
1723    #[doc(hidden)]
1724    pub fn write_allv_mapped<'b, T>(
1725        &self,
1726        requests: &[T],
1727        project: impl Fn(&T) -> FsWrite<'b, F>,
1728    ) -> VfResult<Vec<FsWriteResult>>
1729    where
1730        F: 'b,
1731    {
1732        // Validate the entire batch before writing any prefix. In particular,
1733        // empty requests must not conceal a foreign or already-closed file.
1734        let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
1735        let mut blocked_by = vec![Vec::new(); requests.len()];
1736        for (index, request) in requests.iter().map(&project).enumerate() {
1737            self.validate_owner(request.file, index)?;
1738            request.file.raw().map_err(|error| {
1739                error
1740                    .with_index(index)
1741                    .with_context("write_allv", request.file.path())
1742            })?;
1743            let VfOffset::At(offset) = request.offset else {
1744                return Err(VfError::client(index, crate::ERR_INVAL)
1745                    .with_context("write_allv", request.file.path()));
1746            };
1747            let end = offset
1748                .checked_add(request.data.len() as u64)
1749                .ok_or_else(|| {
1750                    VfError::client(index, libc::EOVERFLOW as u32)
1751                        .with_context("write_allv", request.file.path())
1752                })?;
1753            // Non-overlapping writes commute. An overlapping later request
1754            // must wait until every earlier conflicting request is complete:
1755            // otherwise a short-write retry can overwrite the later bytes.
1756            let prior = prior_by_path.entry(request.file.path()).or_default();
1757            for &(earlier, start, earlier_end) in prior.iter() {
1758                if offset < earlier_end && start < end {
1759                    blocked_by[index].push(earlier);
1760                }
1761            }
1762            prior.push((index, offset, end));
1763        }
1764        let mut totals = vec![0usize; requests.len()];
1765        let mut stable = vec![true; requests.len()];
1766        let mut pending: Vec<usize> = (0..requests.len())
1767            .filter(|&index| !project(&requests[index]).data.is_empty())
1768            .collect();
1769        while !pending.is_empty() {
1770            let wave_indices: Vec<usize> = pending
1771                .iter()
1772                .copied()
1773                .filter(|&index| {
1774                    blocked_by[index]
1775                        .iter()
1776                        .all(|&earlier| totals[earlier] == project(&requests[earlier]).data.len())
1777                })
1778                .collect();
1779            let wave: Vec<_> = wave_indices
1780                .iter()
1781                .map(|&index| {
1782                    let request = project(&requests[index]);
1783                    let VfOffset::At(offset) = request.offset else {
1784                        return Err(VfError::client(index, crate::ERR_INVAL)
1785                            .with_context("write_allv", request.file.path()));
1786                    };
1787                    let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
1788                        VfError::client(index, libc::EOVERFLOW as u32)
1789                            .with_context("write_allv", request.file.path())
1790                    })?;
1791                    Ok(FsWrite {
1792                        file: request.file,
1793                        offset: VfOffset::At(offset),
1794                        data: &request.data[totals[index]..],
1795                    })
1796                })
1797                .collect::<VfResult<_>>()?;
1798            let results = self.writev(&wave).map_err(|error| {
1799                error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
1800            })?;
1801            for (&index, result) in wave_indices.iter().zip(results) {
1802                if result.written == 0 {
1803                    return Err(VfError::client(index, crate::ERR_IO)
1804                        .with_context("write_allv", project(&requests[index]).file.path()));
1805                }
1806                totals[index] += result.written;
1807                stable[index] &= result.stable;
1808            }
1809            pending.retain(|&index| totals[index] < project(&requests[index]).data.len());
1810        }
1811        requests
1812            .iter()
1813            .map(&project)
1814            .enumerate()
1815            .map(|(index, request)| {
1816                let VfOffset::At(offset) = request.offset else {
1817                    return Err(VfError::client(index, crate::ERR_INVAL)
1818                        .with_context("write_allv", request.file.path()));
1819                };
1820                Ok(FsWriteResult {
1821                    offset,
1822                    written: totals[index],
1823                    stable: stable[index],
1824                })
1825            })
1826            .collect()
1827    }
1828
1829    fn write_ops<'a, 'b, T>(
1830        &self,
1831        requests: &'a [T],
1832        project: impl Fn(&T) -> FsWrite<'b, F>,
1833    ) -> VfResult<Vec<WriteOpRef<'a>>>
1834    where
1835        F: 'b,
1836        'b: 'a,
1837    {
1838        let mut writes = Vec::with_capacity(requests.len());
1839        for (index, request) in requests.iter().map(&project).enumerate() {
1840            self.validate_owner(request.file, index)?;
1841            writes.push(WriteOpRef::new(
1842                request
1843                    .file
1844                    .raw()
1845                    .map_err(|error| error.with_index(index))?,
1846                request.offset,
1847                request.data,
1848            ));
1849        }
1850        Ok(writes)
1851    }
1852
1853    fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
1854        if Arc::ptr_eq(&self.inner, &file.inner) {
1855            Ok(())
1856        } else {
1857            Err(VfError::client(index, crate::ERR_INVAL))
1858        }
1859    }
1860}
1861
1862impl<F: VectorFileSystem + VecFs> FsClient<F> {
1863    /// Read several complete files by path using vector READ operations.
1864    ///
1865    /// The aggregate returned data is limited to 16 MiB by default. Use
1866    /// [`read_files_with_options`](Self::read_files_with_options) to choose a
1867    /// different limit, or stream large files instead. This is not a snapshot
1868    /// or an atomic operation across files.
1869    pub fn read_files<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<Vec<u8>>> {
1870        self.read_files_with_options(
1871            paths,
1872            ReadAllOptions::new().max_total_bytes(self.limits.max_read_bytes),
1873        )
1874    }
1875
1876    /// Read several complete files with an explicit aggregate allocation limit.
1877    pub fn read_files_with_options<P: AsRef<Path>>(
1878        &self,
1879        paths: &[P],
1880        options: ReadAllOptions,
1881    ) -> VfResult<Vec<Vec<u8>>> {
1882        let files: Vec<_> = paths
1883            .iter()
1884            .map(|path| VfFile::from_os_path(path.as_ref()))
1885            .collect();
1886        let result = self
1887            .lock()?
1888            .read_allv_with_options(&files, options)
1889            .map_err(|error| {
1890                error
1891                    .index()
1892                    .and_then(|index| paths.get(index))
1893                    .map_or(error.clone(), |path| {
1894                        error.with_context("read_files", path.as_ref())
1895                    })
1896            });
1897        result.and_then(|buffers| {
1898            if buffers.len() != paths.len() {
1899                Err(wrong_result_count("read_files", paths.len(), buffers.len()))
1900            } else {
1901                Ok(buffers)
1902            }
1903        })
1904    }
1905
1906    /// Replace several files from borrowed buffers using vector OPEN, WRITE,
1907    /// and CLOSE phases. Identical path spellings are rejected before opening
1908    /// anything; aliases such as hard links are still the caller's responsibility.
1909    /// The batch is not transactional: an error may follow files already
1910    /// created or written. Large inputs should be chunked by the caller rather
1911    /// than held in memory solely for this convenience method.
1912    pub fn write_files<P: AsRef<Path>, B: AsRef<[u8]>>(&self, entries: &[(P, B)]) -> VfResult<()> {
1913        let mut seen = HashSet::with_capacity(entries.len());
1914        for (index, (path, _)) in entries.iter().enumerate() {
1915            if !seen.insert(path.as_ref()) {
1916                return Err(VfError::client(index, crate::ERR_INVAL)
1917                    .with_context("write_files", path.as_ref()));
1918            }
1919        }
1920        let requests: Vec<_> = entries
1921            .iter()
1922            .map(|(path, _)| {
1923                OpenRequest::new(
1924                    path.as_ref(),
1925                    OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
1926                )
1927            })
1928            .collect();
1929        let files = self.openv(&requests)?;
1930        let writes: Vec<_> = files
1931            .iter()
1932            .zip(entries)
1933            .map(|(file, (_, data))| file.write_request_at(0, data.as_ref()))
1934            .collect();
1935        let result = self.write_allv(&writes);
1936        drop(writes);
1937        let close_result = self.closev(files);
1938        result?;
1939        close_result
1940    }
1941}
1942
1943/// `std::fs::OpenOptions`-style builder tied to an [`FsClient`].
1944pub struct OpenOptions<'a, F: FileSystem> {
1945    client: &'a FsClient<F>,
1946    flags: OpenFlags,
1947    mode: u32,
1948}
1949
1950impl<F: FileSystem> Clone for OpenOptions<'_, F> {
1951    fn clone(&self) -> Self {
1952        Self {
1953            client: self.client,
1954            flags: self.flags,
1955            mode: self.mode,
1956        }
1957    }
1958}
1959
1960impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
1961    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1962        formatter
1963            .debug_struct("OpenOptions")
1964            .field("flags", &self.flags)
1965            .field("mode", &format_args!("{:#o}", self.mode))
1966            .finish()
1967    }
1968}
1969
1970impl<'a, F: FileSystem> OpenOptions<'a, F> {
1971    fn new(client: &'a FsClient<F>) -> Self {
1972        Self {
1973            client,
1974            flags: OpenFlags::empty(),
1975            mode: 0o666,
1976        }
1977    }
1978
1979    pub fn read(&mut self, enabled: bool) -> &mut Self {
1980        self.flags.set(OpenFlags::READ, enabled);
1981        self
1982    }
1983
1984    pub fn write(&mut self, enabled: bool) -> &mut Self {
1985        self.flags.set(OpenFlags::WRITE, enabled);
1986        self
1987    }
1988
1989    pub fn append(&mut self, enabled: bool) -> &mut Self {
1990        self.flags.set(OpenFlags::APPEND, enabled);
1991        self
1992    }
1993
1994    pub fn truncate(&mut self, enabled: bool) -> &mut Self {
1995        self.flags.set(OpenFlags::TRUNCATE, enabled);
1996        self
1997    }
1998
1999    pub fn create(&mut self, enabled: bool) -> &mut Self {
2000        self.flags.set(OpenFlags::CREATE, enabled);
2001        self
2002    }
2003
2004    pub fn create_new(&mut self, enabled: bool) -> &mut Self {
2005        self.flags.set(OpenFlags::CREATE_NEW, enabled);
2006        self
2007    }
2008
2009    pub fn mode(&mut self, mode: u32) -> &mut Self {
2010        self.mode = mode;
2011        self
2012    }
2013
2014    pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
2015        self.client
2016            .open_with(OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
2017    }
2018}
2019
2020impl<F: VectorFileSystem> OpenOptions<'_, F> {
2021    pub fn openv<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
2022        let requests: Vec<OpenRequest> = paths
2023            .iter()
2024            .map(|path| OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
2025            .collect();
2026        self.client.openv(&requests)
2027    }
2028}
2029
2030/// Builder for an atomic metadata update request.
2031pub struct SetMetadata<'a, F: MetadataFileSystem> {
2032    client: &'a FsClient<F>,
2033    path: PathBuf,
2034    update: MetadataUpdate,
2035    follow: bool,
2036}
2037
2038impl<F: MetadataFileSystem> Clone for SetMetadata<'_, F> {
2039    fn clone(&self) -> Self {
2040        Self {
2041            client: self.client,
2042            path: self.path.clone(),
2043            update: self.update.clone(),
2044            follow: self.follow,
2045        }
2046    }
2047}
2048
2049impl<F: MetadataFileSystem> fmt::Debug for SetMetadata<'_, F> {
2050    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2051        formatter
2052            .debug_struct("SetMetadata")
2053            .field("path", &self.path)
2054            .field("update", &self.update)
2055            .field("follow", &self.follow)
2056            .finish_non_exhaustive()
2057    }
2058}
2059
2060impl<F: MetadataFileSystem> SetMetadata<'_, F> {
2061    pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
2062        self.update.permissions = Some(permissions);
2063        self
2064    }
2065
2066    pub fn len(&mut self, len: u64) -> &mut Self {
2067        self.update.len = Some(len);
2068        self
2069    }
2070
2071    pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
2072        self.update.accessed = Some(accessed);
2073        self
2074    }
2075
2076    pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
2077        self.update.modified = Some(modified);
2078        self
2079    }
2080
2081    pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
2082        self.follow = follow;
2083        self
2084    }
2085
2086    pub fn apply(&self) -> VfResult<()> {
2087        self.client
2088            .lock()?
2089            .set_metadata_path(&self.path, self.update.clone(), self.follow)
2090    }
2091}
2092
2093/// Owned RAII file which does not borrow the client.
2094///
2095/// Dropping an open file attempts a best-effort CLOSE. This can block on the
2096/// backend mutex and on network I/O; use [`FsFile::try_close`] when the close
2097/// result matters or when its timing must be controlled.
2098pub struct FsFile<F: FileSystem> {
2099    inner: Arc<SharedBackend<F>>,
2100    file: Option<VfFile>,
2101    path: PathBuf,
2102}
2103
2104impl<F: FileSystem> fmt::Debug for FsFile<F> {
2105    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2106        formatter
2107            .debug_struct("FsFile")
2108            .field("path", &self.path)
2109            .field("is_open", &self.file.is_some())
2110            .finish()
2111    }
2112}
2113
2114impl<F: FileSystem> FsFile<F> {
2115    /// Whether explicit close has completed successfully on this handle.
2116    pub fn is_closed(&self) -> bool {
2117        self.file.is_none()
2118    }
2119    fn raw(&self) -> VfResult<&VfFile> {
2120        self.file
2121            .as_ref()
2122            .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
2123    }
2124
2125    /// Positional read which does not alter the file cursor.
2126    pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
2127        self.read_into(buffer, VfOffset::At(offset))
2128    }
2129
2130    /// Positional write which does not alter the file cursor.
2131    pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
2132        self.write_from(buffer, VfOffset::At(offset))
2133    }
2134
2135    pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
2136        FsRead {
2137            file: self,
2138            offset: VfOffset::At(offset),
2139            length,
2140        }
2141    }
2142
2143    /// Name used at open, retained for diagnostics; not a current namespace
2144    /// lookup or proof of identity. Handle operations continue to use the
2145    /// opened object even if its pathname changes.
2146    pub fn path(&self) -> &Path {
2147        &self.path
2148    }
2149
2150    /// Query metadata for the open object without resolving its path again.
2151    pub fn metadata(&self) -> VfResult<Metadata> {
2152        let attributes = AttrMask::MODE
2153            | AttrMask::SIZE
2154            | AttrMask::NLINK
2155            | AttrMask::FILEID
2156            | AttrMask::UID
2157            | AttrMask::GID
2158            | AttrMask::ATIME
2159            | AttrMask::MTIME
2160            | AttrMask::CTIME
2161            | AttrMask::CHANGE;
2162        self.inner
2163            .lock()
2164            .map_err(|_| poisoned())?
2165            .metadata(MetadataQuery::new(self.raw()?.clone(), attributes))
2166            .map(vfsi_core::metadata_from_attrs)
2167            .map_err(|error| error.with_context("metadata", &self.path))
2168    }
2169
2170    /// Truncate or extend the open file.
2171    pub fn set_len(&self, len: u64) -> VfResult<()> {
2172        let mut update = SetAttributes::new(self.raw()?.clone());
2173        update.size = Some(len);
2174        self.inner
2175            .lock()
2176            .map_err(|_| poisoned())?
2177            .set_attributes(update)
2178            .map_err(|error| error.with_context("set_len", &self.path))
2179    }
2180
2181    /// Change permissions on the open file.
2182    pub fn set_permissions(&self, permissions: Permissions) -> VfResult<()> {
2183        let mut update = SetAttributes::new(self.raw()?.clone());
2184        update.mode = Some(permissions.mode());
2185        self.inner
2186            .lock()
2187            .map_err(|_| poisoned())?
2188            .set_attributes(update)
2189            .map_err(|error| error.with_context("set_permissions", &self.path))
2190    }
2191
2192    pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
2193        FsWrite {
2194            file: self,
2195            offset: VfOffset::At(offset),
2196            data,
2197        }
2198    }
2199
2200    pub fn read_request_at_into<'a>(
2201        &'a self,
2202        offset: u64,
2203        buffer: &'a mut [u8],
2204    ) -> FsReadInto<'a, F> {
2205        FsReadInto {
2206            file: self,
2207            offset: VfOffset::At(offset),
2208            buffer,
2209        }
2210    }
2211
2212    pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
2213        self.read_into(buffer, VfOffset::Cur)
2214    }
2215
2216    /// Collect the remaining bytes from this opened object, starting at its
2217    /// cursor, with an explicit logical payload limit. This does not reopen
2218    /// its path. Unlike standard `Read::read_to_end`, allocation is bounded.
2219    ///
2220    /// An exact-limit read uses at most a one-byte EOF probe. On overflow or
2221    /// I/O failure no buffer is returned and the cursor may have advanced,
2222    /// including the probe byte; this operation does not restore the cursor.
2223    pub fn read_to_end_with_limit(&mut self, max_bytes: usize) -> VfResult<Vec<u8>> {
2224        let mut data = Vec::new();
2225        loop {
2226            let remaining = max_bytes - data.len();
2227            if remaining == 0 {
2228                let mut probe = [0_u8; 1];
2229                if self.read_native(&mut probe)? == 0 {
2230                    return Ok(data);
2231                }
2232                return Err(VfError::client(0, libc::EFBIG as u32)
2233                    .with_context("read_to_end_with_limit", &self.path));
2234            }
2235            let chunk = remaining.min(crate::DEFAULT_READ_STREAM_CHUNK_BYTES);
2236            data.try_reserve(chunk).map_err(|_| {
2237                VfError::client(0, libc::ENOMEM as u32)
2238                    .with_context("read_to_end_with_limit", &self.path)
2239            })?;
2240            let start = data.len();
2241            data.resize(start + chunk, 0);
2242            let count = self.read_native(&mut data[start..])?;
2243            data.truncate(start + count);
2244            if count == 0 {
2245                return Ok(data);
2246            }
2247        }
2248    }
2249
2250    pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
2251        self.write_from(buffer, VfOffset::Cur)
2252    }
2253
2254    fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
2255        if buffer.is_empty() {
2256            return Ok(0);
2257        }
2258        let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
2259        let result = self
2260            .inner
2261            .lock()
2262            .map_err(|_| poisoned())?
2263            .read_one_into(&request, buffer)
2264            .map_err(|error| error.with_context("read", &self.path))?;
2265        validate_read_into_results(
2266            "read",
2267            std::slice::from_ref(&request),
2268            std::slice::from_ref(&result),
2269        )
2270        .map_err(|error| error.with_context("read", &self.path))?;
2271        Ok(result.read)
2272    }
2273
2274    fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
2275        if buffer.is_empty() {
2276            return Ok(0);
2277        }
2278        let file = self.raw()?.clone();
2279        let result = self
2280            .inner
2281            .lock()
2282            .map_err(|_| poisoned())?
2283            .write_one(WriteOpRef::new(&file, offset, buffer))
2284            .map_err(|error| error.with_context("write", &self.path))?;
2285        if result.written > buffer.len() {
2286            return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
2287        }
2288        Ok(result.written)
2289    }
2290
2291    pub fn sync_data(&self) -> VfResult<()> {
2292        self.inner
2293            .lock()
2294            .map_err(|_| poisoned())?
2295            .sync_data(self.raw()?)
2296            .map_err(|error| error.with_context("sync_data", &self.path))
2297    }
2298
2299    pub fn sync_all(&self) -> VfResult<()> {
2300        self.inner
2301            .lock()
2302            .map_err(|_| poisoned())?
2303            .sync_all(self.raw()?)
2304            .map_err(|error| error.with_context("sync_all", &self.path))
2305    }
2306
2307    /// Attempt to close without consuming the handle. A failed close leaves
2308    /// the handle armed so the caller can reconcile or retry cleanup explicitly.
2309    /// After an ambiguous close failure, remote state is unknown: do not resume
2310    /// data I/O just because `is_closed()` is false. This is local ownership,
2311    /// not proof that the server still has the descriptor open.
2312    pub fn try_close(&mut self) -> VfResult<()> {
2313        let Some(file) = self.file.as_ref() else {
2314            return Ok(());
2315        };
2316        self.inner
2317            .lock()
2318            .map_err(|_| poisoned())?
2319            .close_one(file)
2320            .map_err(|error| error.with_context("close", &self.path))?;
2321        self.file = None;
2322        Ok(())
2323    }
2324
2325    /// Consume and close the handle. On failure, `Drop` queues cleanup for
2326    /// a later operation or explicit cleanup drain; use
2327    /// [`try_close`](Self::try_close) to retain control.
2328    pub fn close(mut self) -> VfResult<()> {
2329        self.try_close()
2330    }
2331
2332    /// Seek while retaining [`VfError`] protocol and path information.
2333    pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
2334        self.inner
2335            .lock()
2336            .map_err(|_| poisoned())?
2337            .seek_one(self.raw()?, position)
2338            .map_err(|error| error.with_context("seek", &self.path))
2339    }
2340}
2341
2342impl<F: FileSystem> Read for FsFile<F> {
2343    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
2344        self.read_native(buffer).map_err(io_error)
2345    }
2346}
2347
2348impl<F: FileSystem> Write for FsFile<F> {
2349    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
2350        self.write_native(buffer).map_err(io_error)
2351    }
2352
2353    fn flush(&mut self) -> io::Result<()> {
2354        self.sync_data().map_err(io_error)
2355    }
2356}
2357
2358impl<F: FileSystem> Seek for FsFile<F> {
2359    fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
2360        self.seek_native(position).map_err(io_error)
2361    }
2362}
2363
2364/// Typed read request for [`FsClient::readv`].
2365pub struct FsRead<'a, F: FileSystem> {
2366    file: &'a FsFile<F>,
2367    offset: VfOffset,
2368    length: usize,
2369}
2370
2371/// Typed borrowed write request for [`FsClient::writev`].
2372pub struct FsWrite<'a, F: FileSystem> {
2373    file: &'a FsFile<F>,
2374    offset: VfOffset,
2375    data: &'a [u8],
2376}
2377
2378/// Typed vector read into caller-provided storage.
2379pub struct FsReadInto<'a, F: FileSystem> {
2380    file: &'a FsFile<F>,
2381    offset: VfOffset,
2382    buffer: &'a mut [u8],
2383}
2384
2385impl<F: FileSystem> Drop for FsFile<F> {
2386    fn drop(&mut self) {
2387        if let Some(file) = self.file.take() {
2388            self.inner.defer(
2389                CleanupTarget::File(file, std::mem::take(&mut self.path)),
2390                |backend, target| {
2391                    let CleanupTarget::File(file, path) = target else {
2392                        unreachable!()
2393                    };
2394                    backend
2395                        .close_deferred(file)
2396                        .map_err(|error| error.with_context("close", path))
2397                },
2398            );
2399        }
2400    }
2401}
2402
2403#[cfg(test)]
2404mod traversal_tests {
2405    use super::*;
2406    use std::sync::atomic::{AtomicUsize, Ordering};
2407
2408    // The generic listdir_page fallback's cursor owns the full unconsumed
2409    // listing, not just its next page. Count these retained allocations.
2410    struct Snapshot {
2411        remaining: std::vec::IntoIter<DirEntry>,
2412        live: Arc<AtomicUsize>,
2413    }
2414    impl Drop for Snapshot {
2415        fn drop(&mut self) {
2416            self.live.fetch_sub(1, Ordering::SeqCst);
2417        }
2418    }
2419    fn snapshot_pages(
2420        live: Arc<AtomicUsize>,
2421        peak: Arc<AtomicUsize>,
2422    ) -> impl FnMut(&Path, Option<crate::DirPageCursor>, usize, usize) -> VfResult<DirectoryPage>
2423    {
2424        move |path, cursor, page_size, max_entries| {
2425            let mut snapshot = match cursor {
2426                Some(cursor) => cursor.into_state::<Snapshot>()?,
2427                None => {
2428                    let depth = path.components().count();
2429                    let entries = (0..129)
2430                        .take(max_entries)
2431                        .map(|index| {
2432                            let path = path.join(format!("entry-{index}"));
2433                            let attrs = crate::VfAttrs {
2434                                file: VfFile::from_os_path(&path),
2435                                ftype: if index == 0 && depth < 5 {
2436                                    crate::VfType::Directory
2437                                } else {
2438                                    crate::VfType::Regular
2439                                },
2440                                ..Default::default()
2441                            };
2442                            DirEntry::new(path, vfsi_core::metadata_from_attrs(attrs))
2443                        })
2444                        .collect::<Vec<_>>();
2445                    peak.fetch_max(live.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
2446                    Snapshot {
2447                        remaining: entries.into_iter(),
2448                        live: Arc::clone(&live),
2449                    }
2450                }
2451            };
2452            let entries = snapshot.remaining.by_ref().take(page_size).collect();
2453            let next = if snapshot.remaining.len() == 0 {
2454                None
2455            } else {
2456                Some(crate::DirPageCursor::new(snapshot))
2457            };
2458            Ok((entries, next))
2459        }
2460    }
2461    #[test]
2462    fn traversal_never_retains_multiple_fallback_snapshots() {
2463        let live = Arc::new(AtomicUsize::new(0));
2464        let peak = Arc::new(AtomicUsize::new(0));
2465        let mut seen = 0;
2466        let completion = visit_walk_pages(
2467            Path::new("/tree"),
2468            crate::WalkOptions::new().max_entries(1000),
2469            snapshot_pages(Arc::clone(&live), Arc::clone(&peak)),
2470            |_| {
2471                seen += 1;
2472                Ok(std::ops::ControlFlow::Continue(()))
2473            },
2474        )
2475        .unwrap();
2476        assert_eq!(completion, TraversalCompletion::Complete);
2477        assert_eq!(seen, 4 * 129);
2478        assert_eq!(live.load(Ordering::SeqCst), 0);
2479        assert_eq!(peak.load(Ordering::SeqCst), 1);
2480    }
2481    #[test]
2482    fn traversal_drops_fallback_snapshot_on_stop_error_and_limit() {
2483        for outcome in 0..3 {
2484            let live = Arc::new(AtomicUsize::new(0));
2485            let peak = Arc::new(AtomicUsize::new(0));
2486            let result = visit_walk_pages(
2487                Path::new("/tree"),
2488                crate::WalkOptions::new().max_entries(if outcome == 2 { 2 } else { 1000 }),
2489                snapshot_pages(Arc::clone(&live), peak),
2490                |_| match outcome {
2491                    0 => Ok(std::ops::ControlFlow::Break(())),
2492                    1 => Err(VfError::client(0, libc::EIO as u32)),
2493                    _ => Ok(std::ops::ControlFlow::Continue(())),
2494                },
2495            );
2496            match outcome {
2497                0 => assert_eq!(result.unwrap(), TraversalCompletion::Stopped),
2498                1 => assert_eq!(result.unwrap_err().err_no(), libc::EIO as u32),
2499                _ => assert_eq!(result.unwrap_err().err_no(), libc::EFBIG as u32),
2500            }
2501            assert_eq!(live.load(Ordering::SeqCst), 0);
2502        }
2503    }
2504}