zerofs_client/
file.rs

1use crate::error::{ClientResultExt, ZeroFsError};
2use crate::failover::{
3    FailoverClient, MAX_ATTEMPTS, OP_TIMEOUT, RETRY_DELAY, is_transport_failure,
4};
5use crate::session::{FidGuard, Session};
6use crate::types::{Metadata, OpenOptions, SetAttrs};
7use arc_swap::ArcSwap;
8use std::future::Future;
9use std::path::PathBuf;
10use std::sync::Arc;
11use std::sync::atomic::{AtomicBool, Ordering};
12
13/// The (session, fid) an open handle is currently bound to. A failover-aware
14/// handle swaps this atomically when it re-opens on a new leader; the old `Bound`
15/// drops once no in-flight op still holds it, clunking the old fid.
16struct Bound {
17    session: Arc<Session>,
18    guard: FidGuard,
19}
20
21/// What a failover-aware handle needs to re-bind after a leader change.
22struct Reconnect {
23    fc: Arc<FailoverClient>,
24    /// The path to RE-open on the new leader (walkable, unlike the display string).
25    path: PathBuf,
26    /// How to re-open: the original access, but never create/truncate. The file
27    /// already exists, so re-truncating would discard durable data.
28    opts: OpenOptions,
29    /// Serializes re-opens so concurrent failed ops produce exactly one.
30    relock: tokio::sync::Mutex<()>,
31}
32
33/// An open file. All I/O is positioned (pread/pwrite); there is no shared
34/// cursor, so an `Arc<File>` is safe to use from many tasks at once.
35///
36/// A handle from [`crate::FailoverClient`] is failover-aware: each op is bounded
37/// by a timeout and, on a transport failure or a hang against a dead/frozen
38/// leader, the handle re-opens the path on the re-probed leader and retries. A
39/// handle from a plain [`crate::Client`] is bound to its one connection.
40pub struct File {
41    bound: ArcSwap<Bound>,
42    /// `None` for a plain single-connection handle; `Some` for a failover-aware one.
43    reconnect: Option<Reconnect>,
44    closed: AtomicBool,
45    path: String,
46}
47
48impl std::fmt::Debug for File {
49    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
50        f.debug_struct("File")
51            .field("path", &self.path)
52            .field("closed", &self.closed.load(Ordering::Relaxed))
53            .field("failover", &self.reconnect.is_some())
54            .finish_non_exhaustive()
55    }
56}
57
58impl File {
59    /// A plain handle bound to one connection (no failover).
60    pub(crate) fn new(session: Arc<Session>, guard: FidGuard, path: String) -> Arc<Self> {
61        Arc::new(Self {
62            bound: ArcSwap::from_pointee(Bound { session, guard }),
63            reconnect: None,
64            closed: AtomicBool::new(false),
65            path,
66        })
67    }
68
69    /// A failover-aware handle: on a transport failure it re-opens `path` (with
70    /// `reopen_opts`, which must not create/truncate) on the re-probed leader.
71    pub(crate) fn new_failover(
72        fc: Arc<FailoverClient>,
73        session: Arc<Session>,
74        guard: FidGuard,
75        path: PathBuf,
76        reopen_opts: OpenOptions,
77    ) -> Arc<Self> {
78        let display = crate::path::display(&path);
79        Arc::new(Self {
80            bound: ArcSwap::from_pointee(Bound { session, guard }),
81            reconnect: Some(Reconnect {
82                fc,
83                path,
84                opts: reopen_opts,
85                relock: tokio::sync::Mutex::new(()),
86            }),
87            closed: AtomicBool::new(false),
88            path: display,
89        })
90    }
91
92    /// Run `op` against the current (session, fid). A plain handle does one shot.
93    /// A failover-aware handle bounds each try by [`OP_TIMEOUT`] and, on a transport
94    /// failure or a hang, re-opens on the re-probed leader and retries. FS-semantic
95    /// errors (NotFound, etc.) surface immediately.
96    async fn with_binding<T, F, Fut>(&self, op: F) -> Result<T, ZeroFsError>
97    where
98        F: Fn(Arc<Session>, u32) -> Fut,
99        Fut: Future<Output = Result<T, ZeroFsError>>,
100    {
101        if self.closed.load(Ordering::Acquire) {
102            return Err(ZeroFsError::Closed);
103        }
104        let Some(rc) = &self.reconnect else {
105            let b = self.bound.load_full();
106            return op(Arc::clone(&b.session), b.guard.fid()).await;
107        };
108        let mut last: Option<ZeroFsError> = None;
109        for _ in 0..MAX_ATTEMPTS {
110            if self.closed.load(Ordering::Acquire) {
111                return Err(ZeroFsError::Closed);
112            }
113            let b = self.bound.load_full();
114            match tokio::time::timeout(OP_TIMEOUT, op(Arc::clone(&b.session), b.guard.fid())).await
115            {
116                Ok(Ok(v)) => return Ok(v),
117                Ok(Err(e)) if is_transport_failure(&e) => {
118                    last = Some(e);
119                    self.rebind(rc, &b).await?;
120                    tokio::time::sleep(RETRY_DELAY).await;
121                }
122                Ok(Err(e)) => return Err(e),
123                Err(_) => {
124                    last = Some(ZeroFsError::ConnectFailed {
125                        message: format!("{}: file op timed out; leader unreachable", self.path),
126                    });
127                    self.rebind(rc, &b).await?;
128                }
129            }
130        }
131        Err(last.unwrap_or(ZeroFsError::ConnectFailed {
132            message: format!("{}: file failover attempts exhausted", self.path),
133        }))
134    }
135
136    /// Re-open the path on a freshly probed leader and swap the binding, unless a
137    /// concurrent op already re-bound past this failure (single-flight via `relock`
138    /// + pointer identity of the failed `Bound`).
139    async fn rebind(&self, rc: &Reconnect, failed: &Arc<Bound>) -> Result<(), ZeroFsError> {
140        let _g = rc.relock.lock().await;
141        if !Arc::ptr_eq(&self.bound.load_full(), failed) {
142            return Ok(());
143        }
144        let (session, guard) = rc.fc.reopen_handle(&rc.path, &rc.opts).await?;
145        // Carry this handle's un-fsync'd `.zerofs4` durability obligation from the
146        // failed connection's fid onto the freshly re-opened one, so a verified fsync
147        // after the re-route still verifies the write (and ESTALEs if its lineage broke)
148        // instead of treating the fresh handle as clean.
149        if let Some(token) = failed.session.client.unsynced_oldest(failed.guard.fid()) {
150            session.client.seed_unsynced(guard.fid(), token);
151        }
152        self.bound.store(Arc::new(Bound { session, guard }));
153        Ok(())
154    }
155
156    /// Largest payload a single `read_at` round trip can return; used by the
157    /// cursor to keep one `poll_read` to one round trip.
158    #[cfg(feature = "tokio-io")]
159    pub(crate) fn max_read_chunk(&self) -> u32 {
160        self.bound.load().session.client.max_io()
161    }
162
163    /// Read up to `len` bytes at `offset`; a shorter result means EOF. Returns
164    /// [`bytes::Bytes`]: a read served by one round trip comes back with no copy.
165    pub async fn read_at(&self, offset: u64, len: u32) -> Result<bytes::Bytes, ZeroFsError> {
166        let path = self.path.clone();
167        self.with_binding(move |session, fid| {
168            let path = path.clone();
169            async move { session.client.read_bytes(fid, offset, len).await.ctx(&path) }
170        })
171        .await
172    }
173
174    /// Write all of `data` at `offset` (any size, chunked internally); errors
175    /// on a short write.
176    pub async fn write_at(&self, offset: u64, data: &[u8]) -> Result<(), ZeroFsError> {
177        let data = data.to_vec();
178        let path = self.path.clone();
179        self.with_binding(move |session, fid| {
180            let data = data.clone();
181            let path = path.clone();
182            async move { session.write_all(fid, offset, &data, &path).await }
183        })
184        .await
185    }
186
187    /// Current metadata of this open file (fstat).
188    pub async fn metadata(&self) -> Result<Metadata, ZeroFsError> {
189        let path = self.path.clone();
190        let stat = self
191            .with_binding(move |session, fid| {
192                let path = path.clone();
193                async move { session.stat_fid(fid, &path).await }
194            })
195            .await?;
196        Ok(Metadata::from_stat(&stat))
197    }
198
199    /// Truncate or extend to `size` bytes.
200    pub async fn set_len(&self, size: u64) -> Result<(), ZeroFsError> {
201        self.set_attr(SetAttrs {
202            size: Some(size),
203            ..Default::default()
204        })
205        .await?;
206        Ok(())
207    }
208
209    /// Apply metadata changes through this handle.
210    pub async fn set_attr(&self, attrs: SetAttrs) -> Result<Metadata, ZeroFsError> {
211        let path = self.path.clone();
212        let stat = self
213            .with_binding(move |session, fid| {
214                let path = path.clone();
215                async move { session.setattr_fid(fid, &attrs, &path).await }
216            })
217            .await?;
218        Ok(Metadata::from_stat(&stat))
219    }
220
221    /// Flush data and metadata to durable (S3-backed) storage.
222    ///
223    /// Against a `.zerofs4` server this is a verified fsync: success means every write
224    /// acked on this handle before it is durable and survives any failover. If the
225    /// durability lineage broke (a failover the surviving node could not prove it
226    /// inherited), it returns [`ZeroFsError::Stale`] rather than report success: treat
227    /// the prior writes as lost, redo them, and call `sync_all` again. A pre-`.zerofs4`
228    /// server gives a plain, unverified flush.
229    pub async fn sync_all(&self) -> Result<(), ZeroFsError> {
230        let path = self.path.clone();
231        self.with_binding(move |session, fid| {
232            let path = path.clone();
233            async move { session.client.fsync(fid, 0).await.ctx(&path) }
234        })
235        .await
236    }
237
238    /// Flush file data only. Like [`Self::sync_all`] this is verified against a
239    /// `.zerofs4` server: it returns [`ZeroFsError::Stale`] if the prior writes' lineage
240    /// broke, rather than report success.
241    pub async fn sync_data(&self) -> Result<(), ZeroFsError> {
242        let path = self.path.clone();
243        self.with_binding(move |session, fid| {
244            let path = path.clone();
245            async move { session.client.fsync(fid, 1).await.ctx(&path) }
246        })
247        .await
248    }
249
250    /// Mark the handle closed (later calls return `Closed`). The fid is clunked
251    /// and its number recycled when the handle is dropped (for scope-bound use,
252    /// right after this call). Always succeeds; idempotent; never blocks.
253    pub async fn close(&self) {
254        self.closed.store(true, Ordering::Release);
255    }
256
257    /// An independent `AsyncRead + AsyncWrite + AsyncSeek` cursor over this
258    /// file, starting at offset 0. Multiple cursors over one `File` are safe:
259    /// each carries its own position and the underlying I/O is positioned.
260    /// Rust-only sugar (`tokio-io` feature); never crosses FFI.
261    #[cfg(feature = "tokio-io")]
262    pub fn cursor(self: &Arc<File>) -> crate::io::FileCursor {
263        crate::io::FileCursor::new(Arc::clone(self))
264    }
265}
zerofs_client/file.rs

zerofs_client/
file.rs
