zerofs_client/
failover.rs

1//! A failover client: given all HA node addresses, connects to the current
2//! leader and transparently re-routes across a failover.
3//!
4//! The leader is found by probing for the node whose 9P both accepts a
5//! connection and answers a sanity op (a standby is not serving; a fenced or
6//! lapsed leader fails the lease gate).
7//!
8//! Retry safety (a prior design had a silent-duplication bug here). A
9//! retry-across-failover is safe two ways: naturally idempotent ops (`read`,
10//! and `write` = create+truncate), and idempotency op-ids for the non-idempotent
11//! mutating ops. Each op-id is generated ONCE per logical op (here, before the
12//! retry loop) and reused on every attempt, so the server recognizes a resend as
13//! already-applied instead of double-applying or spuriously EEXIST/ENOENT-ing.
14
15use crate::file::File;
16use crate::session::{FidGuard, Session};
17use crate::types::OpenOptions;
18use crate::{Client, Metadata, NodeKind, ZeroFsError};
19use arc_swap::ArcSwapOption;
20use bytes::Bytes;
21use std::future::Future;
22use std::path::Path;
23use std::sync::Arc;
24use std::time::Duration;
25
26/// A fresh op-id, generated ONCE per logical mutating op (before its retry loop)
27/// and reused on every retry. Regenerating per retry would break cross-failover
28/// dedup: each retry would mint an id the new leader has never seen.
29fn new_op_id() -> [u8; 16] {
30    use rand::RngCore;
31    let mut id = [0u8; 16];
32    rand::thread_rng().fill_bytes(&mut id);
33    id
34}
35
36pub(crate) const MAX_ATTEMPTS: usize = 60;
37pub(crate) const RETRY_DELAY: Duration = Duration::from_millis(500);
38const PROBE_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
39const PROBE_OP_TIMEOUT: Duration = Duration::from_secs(3);
40/// Per-op cap before treating the leader as dead and re-routing: a killed
41/// leader's cached connection otherwise hangs forever. Must exceed the leader's
42/// ship timeout so a healthy (shipping) write is not falsely re-routed. Also
43/// bounds each op on an open [`crate::File`] handle, which re-opens on failover.
44pub(crate) const OP_TIMEOUT: Duration = Duration::from_secs(8);
45
46/// A client that connects to the current HA leader and transparently re-routes
47/// across a failover. See the module docs for retry-safety.
48pub struct FailoverClient {
49    targets: Vec<String>,
50    /// Current leader connection, loaded lock-free per op, swapped on failover.
51    /// `None` means disconnected (re-probe needed).
52    current: ArcSwapOption<Client>,
53}
54
55/// Errors meaning "this node is no longer the leader": re-routing and retrying an
56/// idempotent op is safe. FS-semantic errors (NotFound, etc.) must surface.
57pub(crate) fn is_transport_failure(e: &ZeroFsError) -> bool {
58    match e {
59        // NotLeader is the explicit "this node was deposed/fenced" signal
60        // (P9_ENOTLEADER); a dropped connection / failed negotiation are the others.
61        ZeroFsError::NotLeader { .. }
62        | ZeroFsError::ConnectFailed { .. }
63        | ZeroFsError::Closed
64        | ZeroFsError::Protocol { .. } => true,
65        // The per-op db gate (a lapse racing an already-dispatched op) flattens to EIO.
66        ZeroFsError::Io { errno, .. } => *errno == libc::EIO,
67        _ => false,
68    }
69}
70
71impl FailoverClient {
72    /// Connect to the serving leader among `targets`, retrying while none serves
73    /// yet (bring-up or a failover gap).
74    pub async fn connect(targets: Vec<String>) -> Result<Arc<Self>, ZeroFsError> {
75        if targets.is_empty() {
76            return Err(ZeroFsError::ConnectFailed {
77                message: "no targets given".into(),
78            });
79        }
80        let fc = Arc::new(Self {
81            targets,
82            current: ArcSwapOption::empty(),
83        });
84        for _ in 0..MAX_ATTEMPTS {
85            if fc.obtain().await.is_some() {
86                return Ok(fc);
87            }
88            tokio::time::sleep(RETRY_DELAY).await;
89        }
90        Err(ZeroFsError::ConnectFailed {
91            message: "no serving leader found among targets".into(),
92        })
93    }
94
95    fn cached(&self) -> Option<Arc<Client>> {
96        self.current.load_full()
97    }
98
99    fn invalidate(&self) {
100        self.current.store(None);
101    }
102
103    /// The cached connection, else a freshly probed one (cached on success).
104    async fn obtain(&self) -> Option<Arc<Client>> {
105        if let Some(c) = self.cached() {
106            return Some(c);
107        }
108        let client = self.probe().await?;
109        self.current.store(Some(client.clone()));
110        Some(client)
111    }
112
113    /// Probe the targets for the serving leader (connectable AND answers a sanity op).
114    async fn probe(&self) -> Option<Arc<Client>> {
115        for target in &self.targets {
116            let connected =
117                tokio::time::timeout(PROBE_CONNECT_TIMEOUT, Client::connect(target)).await;
118            let Ok(Ok(client)) = connected else {
119                continue;
120            };
121            let healthy = matches!(
122                tokio::time::timeout(PROBE_OP_TIMEOUT, client.stat("/")).await,
123                Ok(Ok(_))
124            );
125            if healthy {
126                return Some(client);
127            }
128            client.close().await;
129        }
130        None
131    }
132
133    /// Run `op` against the current leader, re-routing across a failover. `op`
134    /// must be SAFE to replay (naturally idempotent, or carrying a stable op-id
135    /// the caller generated once outside this helper). A transport failure or a
136    /// hang past [`OP_TIMEOUT`] drops the cached connection and retries on the
137    /// re-probed leader; any other error surfaces immediately.
138    async fn with_failover<T, F, Fut>(&self, op: F) -> Result<T, ZeroFsError>
139    where
140        F: Fn(Arc<Client>) -> Fut,
141        Fut: Future<Output = Result<T, ZeroFsError>>,
142    {
143        let mut last = None;
144        for _ in 0..MAX_ATTEMPTS {
145            let Some(client) = self.obtain().await else {
146                tokio::time::sleep(RETRY_DELAY).await;
147                continue;
148            };
149            match tokio::time::timeout(OP_TIMEOUT, op(client)).await {
150                Ok(Ok(v)) => return Ok(v),
151                Ok(Err(e)) if is_transport_failure(&e) => {
152                    self.invalidate();
153                    last = Some(e);
154                    tokio::time::sleep(RETRY_DELAY).await;
155                }
156                Ok(Err(e)) => return Err(e),
157                Err(_) => {
158                    // Hung op: leader likely dead. Drop it and re-route.
159                    self.invalidate();
160                    last = Some(ZeroFsError::ConnectFailed {
161                        message: "op timed out; leader unreachable".into(),
162                    });
163                }
164            }
165        }
166        Err(last.unwrap_or(ZeroFsError::ConnectFailed {
167            message: "failover attempts exhausted".into(),
168        }))
169    }
170
171    /// Re-open `path` on a freshly probed leader, returning a new (session, fid)
172    /// binding for a failover-aware [`File`] to swap in after a leader change.
173    /// Invalidates the stale cached leader first. A plain open is idempotent, so
174    /// `with_failover`'s retry is safe.
175    pub(crate) async fn reopen_handle(
176        &self,
177        path: &Path,
178        opts: &OpenOptions,
179    ) -> Result<(Arc<Session>, FidGuard), ZeroFsError> {
180        self.invalidate();
181        let path = path.to_path_buf();
182        let opts = *opts;
183        self.with_failover(move |c| {
184            let path = path.clone();
185            async move { c.open_guard(&path, &opts).await }
186        })
187        .await
188    }
189
190    /// Read a file, re-routing across a failover. Idempotent.
191    pub async fn read(&self, path: impl AsRef<Path>) -> Result<Bytes, ZeroFsError> {
192        let path = path.as_ref().to_path_buf();
193        self.with_failover(move |c| {
194            let path = path.clone();
195            async move { c.read(path).await }
196        })
197        .await
198    }
199
200    /// Write a file (create + truncate = last-write-wins), re-routing across a
201    /// failover. Idempotent.
202    pub async fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> Result<(), ZeroFsError> {
203        let path = path.as_ref().to_path_buf();
204        let data = data.to_vec();
205        self.with_failover(move |c| {
206            let path = path.clone();
207            let data = data.clone();
208            async move { c.write(path, &data).await }
209        })
210        .await
211    }
212
213    /// Create a directory, re-routing across a failover. Retry-safe via a stable op-id.
214    pub async fn create_dir(
215        &self,
216        path: impl AsRef<Path>,
217        mode: u32,
218    ) -> Result<Metadata, ZeroFsError> {
219        let path = path.as_ref().to_path_buf();
220        let op_id = new_op_id();
221        self.with_failover(move |c| {
222            let path = path.clone();
223            async move { c.create_dir_op_id(path, mode, op_id).await }
224        })
225        .await
226    }
227
228    /// Create-or-truncate a file (mode 0o644) and return a failover-aware handle:
229    /// its I/O is bounded by a per-op timeout and, on a leader failover, the handle
230    /// re-opens the path on the re-probed leader and retries. Retry-safe via a
231    /// stable op-id.
232    pub async fn create(
233        self: &Arc<Self>,
234        path: impl AsRef<Path>,
235    ) -> Result<Arc<File>, ZeroFsError> {
236        let orig = path.as_ref().to_path_buf();
237        let op_id = new_op_id();
238        let create_opts = OpenOptions::read_write().create(true).truncate(true);
239        let (session, guard) = {
240            let orig = orig.clone();
241            self.with_failover(move |c| {
242                let orig = orig.clone();
243                async move { c.open_guard_op_id(&orig, &create_opts, op_id).await }
244            })
245            .await?
246        };
247        // Re-open after a failover with the same access but NO create/truncate:
248        // the file already exists, so re-truncating would discard durable data.
249        let reopen_opts = OpenOptions::read_write();
250        Ok(File::new_failover(
251            Arc::clone(self),
252            session,
253            guard,
254            orig,
255            reopen_opts,
256        ))
257    }
258
259    /// Remove a file, symlink, or device node, re-routing across a failover.
260    /// Retry-safe via a stable op-id.
261    pub async fn remove_file(&self, path: impl AsRef<Path>) -> Result<(), ZeroFsError> {
262        let path = path.as_ref().to_path_buf();
263        let op_id = new_op_id();
264        self.with_failover(move |c| {
265            let path = path.clone();
266            async move { c.remove_file_op_id(path, op_id).await }
267        })
268        .await
269    }
270
271    /// Remove an empty directory, re-routing across a failover. Retry-safe via a stable op-id.
272    pub async fn remove_dir(&self, path: impl AsRef<Path>) -> Result<(), ZeroFsError> {
273        let path = path.as_ref().to_path_buf();
274        let op_id = new_op_id();
275        self.with_failover(move |c| {
276            let path = path.clone();
277            async move { c.remove_dir_op_id(path, op_id).await }
278        })
279        .await
280    }
281
282    /// Atomically rename/move, re-routing across a failover. Retry-safe via a stable op-id.
283    pub async fn rename(
284        &self,
285        from: impl AsRef<Path>,
286        to: impl AsRef<Path>,
287    ) -> Result<(), ZeroFsError> {
288        let from = from.as_ref().to_path_buf();
289        let to = to.as_ref().to_path_buf();
290        let op_id = new_op_id();
291        self.with_failover(move |c| {
292            let from = from.clone();
293            let to = to.clone();
294            async move { c.rename_op_id(from, to, op_id).await }
295        })
296        .await
297    }
298
299    /// Create a hard link, re-routing across a failover. Retry-safe via a stable op-id.
300    pub async fn hard_link(
301        &self,
302        original: impl AsRef<Path>,
303        link: impl AsRef<Path>,
304    ) -> Result<Metadata, ZeroFsError> {
305        let original = original.as_ref().to_path_buf();
306        let link = link.as_ref().to_path_buf();
307        let op_id = new_op_id();
308        self.with_failover(move |c| {
309            let original = original.clone();
310            let link = link.clone();
311            async move { c.hard_link_op_id(original, link, op_id).await }
312        })
313        .await
314    }
315
316    /// Create a symlink, re-routing across a failover. Retry-safe via a stable op-id.
317    pub async fn symlink(
318        &self,
319        target: impl AsRef<Path>,
320        link_path: impl AsRef<Path>,
321    ) -> Result<Metadata, ZeroFsError> {
322        let target = target.as_ref().to_path_buf();
323        let link_path = link_path.as_ref().to_path_buf();
324        let op_id = new_op_id();
325        self.with_failover(move |c| {
326            let target = target.clone();
327            let link_path = link_path.clone();
328            async move { c.symlink_op_id(target, link_path, op_id).await }
329        })
330        .await
331    }
332
333    /// Create a fifo, socket, or device node, re-routing across a failover.
334    /// Retry-safe via a stable op-id.
335    pub async fn mknod(
336        &self,
337        path: impl AsRef<Path>,
338        kind: NodeKind,
339        mode: u32,
340    ) -> Result<Metadata, ZeroFsError> {
341        let path = path.as_ref().to_path_buf();
342        let op_id = new_op_id();
343        self.with_failover(move |c| {
344            let path = path.clone();
345            async move { c.mknod_op_id(path, kind, mode, op_id).await }
346        })
347        .await
348    }
349}
zerofs_client/failover.rs

zerofs_client/
failover.rs
