vfsi_sync/client/
mod.rs

1//! Owned, shareable synchronous client and file handles.
2
3use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::path::{Path, PathBuf};
6use std::sync::Arc;
7
8use vfsi_core::api::internal::OwnedReadResult as FsReadResult;
9use vfsi_core::api::{
10    DirectoryListing, ReadIntoResult as FsReadIntoResult, ResourceLimits, StreamCompletion,
11    WriteResult as FsWriteResult,
12};
13
14use crate::backend::{HandleBackend, VectorBackend};
15use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
16use crate::{
17    AttrMask, Attrs, Capabilities, DirEntry, FileOperand, OpenFlags, OpenOp, ReadAllOptions,
18    ReadOp, ReadResult, RemoveOptions, StreamOptions, VfDir, VfError, VfOffset, VfResult, WriteOp,
19    WriteResult,
20};
21
22fn with_indexed_context<'a>(
23    error: VfError,
24    operation: &'static str,
25    path: impl FnOnce(usize) -> Option<&'a Path>,
26) -> VfError {
27    match error.index().and_then(path) {
28        Some(path) => error.with_context(operation, path),
29        None => error,
30    }
31}
32
33fn read_result(result: ReadResult) -> FsReadResult {
34    FsReadResult {
35        offset: result.offset,
36        data: result.data,
37        eof: result.eof,
38    }
39}
40
41fn write_result(result: WriteResult) -> FsWriteResult {
42    FsWriteResult {
43        offset: result.offset,
44        bytes_written: result.bytes_written,
45        stable: result.stable,
46    }
47}
48
49fn poisoned() -> VfError {
50    VfError::client(0, crate::ERR_IO)
51}
52
53fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
54    VfError::transport_with_kind(
55        None,
56        crate::TransportKind::InvalidReply,
57        format!("{operation} backend returned {actual} results for {expected} requests"),
58    )
59}
60
61/// Cloneable owner of one synchronous backend connection.
62/// Clones share a mutex and serialize backend calls; use separate connections
63/// for parallel RPCs. Explicitly close handles to observe cleanup failures.
64/// Handle Drop queues cleanup, drained before later operations or by
65/// `drain_cleanup`. Dropping the final backend owner performs synchronous
66/// teardown and may wait for pending closes and backend request timeouts.
67pub struct FsClient<F: HandleBackend> {
68    inner: Arc<SharedBackend<F>>,
69    limits: ResourceLimits,
70}
71
72impl<F: HandleBackend> Clone for FsClient<F> {
73    fn clone(&self) -> Self {
74        Self {
75            inner: Arc::clone(&self.inner),
76            limits: self.limits,
77        }
78    }
79}
80
81impl<F: HandleBackend> fmt::Debug for FsClient<F> {
82    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
83        formatter.debug_struct("FsClient").finish_non_exhaustive()
84    }
85}
86
87impl<F: HandleBackend> FsClient<F> {
88    pub fn new(filesystem: F) -> Self {
89        Self {
90            inner: Arc::new(SharedBackend::new(filesystem)),
91            limits: ResourceLimits::default(),
92        }
93    }
94
95    /// Configure this client view and its future clones. Existing clones keep
96    /// their policy; all views still share the same connection and ownership.
97    pub fn with_limits(mut self, limits: ResourceLimits) -> Self {
98        self.limits = limits;
99        self
100    }
101
102    pub fn limits(&self) -> ResourceLimits {
103        self.limits
104    }
105
106    fn lock(&self) -> VfResult<BackendGuard<'_, F>> {
107        self.inner.lock().map_err(|_| poisoned())
108    }
109
110    pub fn into_inner(self) -> Result<F, Self> {
111        if self.drain_cleanup().is_err() {
112            return Err(self);
113        }
114        let limits = self.limits;
115        match Arc::try_unwrap(self.inner) {
116            Ok(shared) => shared.into_backend().map_err(|shared| Self {
117                inner: Arc::new(shared),
118                limits,
119            }),
120            Err(inner) => Err(Self { inner, limits }),
121        }
122    }
123
124    /// Drain queued Drop cleanup and report the first failure. Failed targets
125    /// remain owned for later cleanup. No new file operation is replayed.
126    pub fn drain_cleanup(&self) -> VfResult<()> {
127        self.inner
128            .lock_without_cleanup()
129            .map_err(|_| poisoned())?
130            .drain_cleanup()
131    }
132}
133
134impl<F: HandleBackend> FsClient<F> {
135    pub(crate) fn capabilities(&self) -> VfResult<Capabilities> {
136        Ok(self.lock()?.capabilities())
137    }
138    /// Open a path read-only.
139    pub(crate) fn open(&self, path: impl AsRef<Path>) -> VfResult<OpenFile<F>> {
140        self.open_with_native(OpenOp::new(path.as_ref(), OpenFlags::READ))
141    }
142
143    pub(crate) fn open_with_native(&self, request: OpenOp) -> VfResult<OpenFile<F>> {
144        vfsi_core::internal::validate_open_requests(std::slice::from_ref(&request))?;
145        let file = self.lock()?.open_impl(&request)?;
146        Ok(OpenFile {
147            inner: Arc::clone(&self.inner),
148            file: Some(file),
149            append: request.flags().contains(OpenFlags::APPEND),
150            path: request.into_path(),
151        })
152    }
153}
154
155impl<F: VectorBackend> FsClient<F> {
156    /// Open an ordered vector of files.
157    ///
158    /// Success returns one RAII handle per request. Failure returns no
159    /// handles; VFSI does not promise transactional rollback of other
160    /// filesystem effects such as file creation.
161    pub(crate) fn vopen(&self, requests: &[OpenOp]) -> VfResult<Vec<OpenFile<F>>> {
162        vfsi_core::internal::validate_open_requests(requests)?;
163        let mut filesystem = self.lock()?;
164        let files = filesystem.vopen_impl(requests).map_err(|error| {
165            with_indexed_context(error, "vopen", |index| {
166                requests.get(index).map(|request| request.path())
167            })
168        })?;
169        if files.len() != requests.len() {
170            let error = wrong_result_count("vopen", requests.len(), files.len());
171            for (index, file) in files.into_iter().enumerate() {
172                if filesystem.close_impl(&file).is_err() {
173                    // Keep failed cleanup under the same RAII/deferred-close
174                    // ownership as ordinary handles, preserving the shape error.
175                    drop(OpenFile {
176                        inner: Arc::clone(&self.inner),
177                        file: Some(file),
178                        append: false,
179                        path: requests
180                            .get(index)
181                            .map(|request| request.path().to_path_buf())
182                            .unwrap_or_default(),
183                    });
184                }
185            }
186            return Err(error);
187        }
188        drop(filesystem);
189        Ok(files
190            .into_iter()
191            .zip(requests)
192            .map(|(file, request)| OpenFile {
193                inner: Arc::clone(&self.inner),
194                file: Some(file),
195                append: request.flags().contains(OpenFlags::APPEND),
196                path: request.path().to_path_buf(),
197            })
198            .collect())
199    }
200
201    /// Try to close a group through one vector operation without consuming
202    /// the handles. On failure, all handles remain armed: the backend may
203    /// have closed a prefix, so callers must reconcile before retrying.
204    #[doc(hidden)]
205    pub fn vclose<'a>(&self, files: impl IntoIterator<Item = &'a mut OpenFile<F>>) -> VfResult<()>
206    where
207        F: 'a,
208    {
209        let mut files: Vec<_> = files.into_iter().collect();
210        for (index, file) in files.iter().enumerate() {
211            self.validate_owner(file, index)?;
212        }
213        let positions: Vec<_> = files
214            .iter()
215            .enumerate()
216            .filter_map(|(index, file)| (!file.is_closed()).then_some(index))
217            .collect();
218        files.retain(|file| !file.is_closed());
219        if files.is_empty() {
220            return Ok(());
221        }
222        let descriptors: Vec<FileOperand> = files
223            .iter()
224            .map(|file| file.raw().cloned())
225            .collect::<VfResult<_>>()?;
226        self.lock()?.vclose_impl(&descriptors).map_err(|error| {
227            with_indexed_context(error, "vclose_owned", |index| {
228                files.get(index).map(|file| file.path())
229            })
230            .map_index(|index| positions.get(index).copied().unwrap_or(index))
231        })?;
232        for file in &mut files {
233            file.file = None;
234        }
235        Ok(())
236    }
237
238    /// Close a group of files through one vector operation.
239    ///
240    /// On failure, the handles are dropped and the backend receives
241    /// best-effort scalar cleanup attempts. Use [`vclose`](Self::vclose)
242    /// to retain the handles after an error.
243    #[doc(hidden)]
244    pub fn vclose_owned(&self, mut files: Vec<OpenFile<F>>) -> VfResult<()> {
245        self.vclose(&mut files)
246    }
247
248    fn validate_owner(&self, file: &OpenFile<F>, index: usize) -> VfResult<()> {
249        if Arc::ptr_eq(&self.inner, &file.inner) {
250            Ok(())
251        } else {
252            Err(VfError::client(index, crate::ERR_INVAL))
253        }
254    }
255}
256
257fn link_error(error: VfError, paths: &[&Path], operation: &'static str) -> VfError {
258    match error.index() {
259        Some(index) if index < paths.len() => error.with_context(operation, paths[index]),
260        Some(_) => VfError::transport(None, "link backend returned an invalid error index"),
261        None => error,
262    }
263}
264
265mod shared;
266use shared::{BackendGuard, CleanupTarget, SharedBackend};
267mod handles;
268pub use handles::{FsDir, FsRead, FsReadInto, FsWrite, OpenFile};
269mod directory;
270mod metadata;
271mod namespace;
272mod read;
273mod write;
vfsi_sync/client/mod.rs

vfsi_sync/client/
mod.rs
