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

vfsi_sync/client/
mod.rs
