From f40b44823d42e752d67048ff0206eef560356be6 Mon Sep 17 00:00:00 2001 From: luulu Date: Thu, 13 Aug 2026 07:50:05 -0400 Subject: [PATCH] cubefs: surface daemon put failures as FUSE EIO via put_checked Keep CubeBackend::put returning () for source compat across cubesys/cubecrypt/ cube-bench; add put_checked() (default Ok(())) letting DaemonBackend report a real socket error. FUSE create/write/truncate now call put_record_checked and map the failure to FsError::WriteFailed -> errno 5 (EIO). Verified: ./check (fmt+test+clippy) green; canonical ./check mount 57/57. --- cubefs/src/backend.rs | 34 ++++++++++++--- cubefs/src/vfs.rs | 96 +++++++++++++++++++++++++++++++++++++++++-- cubestore/src/lib.rs | 41 ++++++++++++++++-- 3 files changed, 160 insertions(+), 11 deletions(-) diff --git a/cubefs/src/backend.rs b/cubefs/src/backend.rs index 32c1682..318ba59 100644 --- a/cubefs/src/backend.rs +++ b/cubefs/src/backend.rs @@ -93,14 +93,38 @@ impl CubeBackend for DaemonBackend { // (`get_record`) can split header/body back out. Stripping the header // here would make the socket backend diverge from the in-memory backend // and break the round-trip. + // + // A failed put is logged (best-effort) here; the FUSE write path calls + // `put_checked` instead so the failure becomes EIO rather than silent. let hex = to_hex(&value); let cmd = format!("rawput {} {} {} {} {}", key.c, key.z, key.y, key.x, hex); - // A failed durability write is reported via stderr; the FUSE layer - // surfaces the prior successful state to the kernel. We do not panic - // here because an unreachable daemon should not crash the mount — it - // should surface as a write error to the caller (best-effort here). if let Err(e) = self.rpc(&cmd) { - eprintln!("cubefs: rawput {key:?} failed: {e}"); + eprintln!("cubefs: daemon socket rawput {:?}: {e}", key); + } + } + + fn put_checked(&mut self, key: Czyx, value: Vec) -> Result<(), String> { + // `value` is the FULL record envelope (header bytes + body) emitted by + // `CubeStore::put_record` — NOT a bare body. We must store it verbatim + // so that `get` returns an identical envelope and the FUSE read path + // (`get_record`) can split header/body back out. + let hex = to_hex(&value); + let cmd = format!("rawput {} {} {} {} {}", key.c, key.z, key.y, key.x, hex); + // A failed durability write must NOT be silently swallowed: the FUSE + // write path turns this `Err` into EIO so the caller's write() fails + // loudly instead of the kernel believing a durable write landed. An + // unreachable daemon is a genuine I/O error, not a no-op. + match self.rpc(&cmd) { + Ok(resp) => { + // The daemon may still reject (quota, sealed, auth) — treat a + // non-ok reply as a write failure too. + if resp.trim().starts_with("ok") { + Ok(()) + } else { + Err(format!("daemon rejected rawput {key:?}: {}", resp.trim())) + } + } + Err(e) => Err(format!("daemon socket rawput {key:?}: {e}")), } } diff --git a/cubefs/src/vfs.rs b/cubefs/src/vfs.rs index 5a065da..8378eb4 100644 --- a/cubefs/src/vfs.rs +++ b/cubefs/src/vfs.rs @@ -62,6 +62,11 @@ pub enum FsError { /// What the path actually names. actual: Kind, }, + /// EIO — a backend write could not be durably committed (e.g. the daemon + /// socket behind a `--socket` FUSE mount is unreachable or rejected the + /// write). Surfaced so the application's write() fails loudly instead of + /// the kernel believing a durable write landed. + WriteFailed(String), } impl FsError { @@ -77,6 +82,7 @@ impl FsError { Kind::Directory => 20, // ENOTDIR Kind::File => 21, // EISDIR }, + FsError::WriteFailed(_) => 5, // EIO } } } @@ -358,7 +364,9 @@ impl CubeFs { h.created_at = Some(now()); h.size_bytes = Some(0); h.refresh_flags(); - self.store.put_record(c, &h, &[]); + self.store + .put_record_checked(c, &h, &[]) + .map_err(FsError::WriteFailed)?; NullSpace::set_acl(&mut self.store, c, Acl { uid, gid, mode }); NullSpace::journal_append(&mut self.store, JournalOp::Create, c); self.getattr(p) @@ -417,7 +425,9 @@ impl CubeFs { body[start..end].copy_from_slice(data); h.size_bytes = Some(body.len() as u64); h.refresh_flags(); - self.store.put_record(c, &h, &body); + self.store + .put_record_checked(c, &h, &body) + .map_err(FsError::WriteFailed)?; NullSpace::journal_append(&mut self.store, JournalOp::Write, c); Ok(data.len() as u32) } @@ -436,7 +446,9 @@ impl CubeFs { body.resize(len as usize, 0); h.size_bytes = Some(body.len() as u64); h.refresh_flags(); - self.store.put_record(c, &h, &body); + self.store + .put_record_checked(c, &h, &body) + .map_err(FsError::WriteFailed)?; NullSpace::journal_append(&mut self.store, JournalOp::Write, c); Ok(()) } @@ -965,4 +977,82 @@ mod tests { assert_eq!(f.read(&p, 0, 1, ROOT.0, ROOT.1).unwrap(), b"z".to_vec()); } } + + /// Backend whose `put` is a no-op and whose `put_checked` always fails + /// (models an unreachable daemon). `get` returns an empty existing record + /// so `vfs::write` (which reads-then-writes) reaches the failing + /// `put_checked` path used by the FUSE write code. + struct FailingWriteBackend; + impl CubeBackend for FailingWriteBackend { + fn put(&mut self, _key: Czyx, _value: Vec) {} + fn put_checked(&mut self, _key: Czyx, _value: Vec) -> Result<(), String> { + Err("simulated daemon unreachable".to_string()) + } + fn get(&self, _key: &Czyx) -> Option> { + // Envelope = [u32 header_len=0][header][body] => [0,0,0,0] + Some(vec![0, 0, 0, 0]) + } + fn delete(&mut self, _key: &Czyx) {} + fn keys(&self) -> Vec { + vec![] + } + fn scan_prefix(&self, _c: u8, _z: Option, _y: Option) -> Vec { + vec![] + } + } + + /// Backend whose `put` is a no-op and whose `put_checked` always fails, + /// with `get` returning nothing, so `vfs::create` (which checks existence + /// first) proceeds to `put_record_checked`. + struct FailingCreateBackend; + impl CubeBackend for FailingCreateBackend { + fn put(&mut self, _key: Czyx, _value: Vec) {} + fn put_checked(&mut self, _key: Czyx, _value: Vec) -> Result<(), String> { + Err("simulated daemon unreachable".to_string()) + } + fn get(&self, _key: &Czyx) -> Option> { + None + } + fn delete(&mut self, _key: &Czyx) {} + fn keys(&self) -> Vec { + vec![] + } + fn scan_prefix(&self, _c: u8, _z: Option, _y: Option) -> Vec { + vec![] + } + } + + #[test] + fn write_failed_backend_surfaces_as_eio() { + // The FsError variant must map to EIO (errno 5) so the FUSE layer + // reports a failed daemon write back to the application. + assert_eq!(FsError::WriteFailed("x".into()).errno(), 5); + } + + #[test] + fn create_reaches_failing_put_and_returns_eio() { + // create() performs a put_record; a failing backend must surface it as + // FsError::WriteFailed (mapped to EIO by the FUSE layer). + let mut f = CubeFs::new(CubeStore::new(FailingCreateBackend)); + f.format("test"); + let e = f + .create("/c001/z001/y001/x001", USER.0, USER.1, 0o644) + .unwrap_err(); + assert!(matches!(e, FsError::WriteFailed(_))); + assert_eq!(e.errno(), 5); + } + + #[test] + fn write_reaches_failing_put_and_returns_eio() { + // write() on an existing record performs a put_record_checked; a + // failing backend must surface it as FsError::WriteFailed/EIO. ROOT + // credentials bypass the (absent) ACL so we reach the write path. + let mut f = CubeFs::new(CubeStore::new(FailingWriteBackend)); + f.format("test"); + let e = f + .write("/c002/z002/y002/x002", 0, b"data", ROOT.0, ROOT.1) + .unwrap_err(); + assert!(matches!(e, FsError::WriteFailed(_))); + assert_eq!(e.errno(), 5); + } } diff --git a/cubestore/src/lib.rs b/cubestore/src/lib.rs index ad401a7..982099e 100644 --- a/cubestore/src/lib.rs +++ b/cubestore/src/lib.rs @@ -24,6 +24,15 @@ use std::collections::HashMap; pub trait CubeBackend { /// Store `value` at `key`. fn put(&mut self, key: Czyx, value: Vec); + /// Store `value` at `key`, returning `Err` if the write cannot be + /// durably committed (e.g. the daemon socket is unreachable for + /// `DaemonBackend`). The default implementation ignores the result so + /// existing backends stay source-compatible; `DaemonBackend` overrides it. + /// The FUSE write path calls this variant and surfaces `Err` as `EIO`. + fn put_checked(&mut self, key: Czyx, value: Vec) -> Result<(), String> { + self.put(key, value); + Ok(()) + } /// Fetch the value at `key`, if present. fn get(&self, key: &Czyx) -> Option>; /// Remove the value at `key`. @@ -356,7 +365,25 @@ impl CubeStore { buf.extend_from_slice(&(hdr_bytes.len() as u32).to_le_bytes()); buf.extend_from_slice(&hdr_bytes); buf.extend_from_slice(body); - self.backend.put(label, buf); + self.backend.put(label, buf) + } + + /// Like [`CubeStore::put_record`] but returns `Err` if the backend cannot + /// durably commit (e.g. `DaemonBackend` with an unreachable socket). The + /// FUSE write path uses this variant so a failed daemon write becomes + /// `EIO` instead of a silently-dropped write. + pub fn put_record_checked( + &mut self, + label: Czyx, + header: &CubeHeader, + body: &[u8], + ) -> Result<(), String> { + let hdr_bytes = record_codec::encode_header(header); + let mut buf = Vec::with_capacity(4 + hdr_bytes.len() + body.len()); + buf.extend_from_slice(&(hdr_bytes.len() as u32).to_le_bytes()); + buf.extend_from_slice(&hdr_bytes); + buf.extend_from_slice(body); + self.backend.put_checked(label, buf) } /// Fetch and split a record into `(header, body)`. @@ -376,10 +403,18 @@ impl CubeStore { Some((hdr, body)) } - /// Raw backend access (delegates put/get/delete for non-record payloads). + /// Raw backend write (non-record payloads, e.g. ACL/xattr/volume buckets). pub fn put_raw(&mut self, key: Czyx, value: Vec) { - self.backend.put(key, value); + self.backend.put(key, value) } + + /// Like [`CubeStore::put_raw`] but returns `Err` if the backend cannot + /// durably commit. Used by the FUSE write path for best-effort metadata; + /// callers may `.ok()` it or surface the error. + pub fn put_raw_checked(&mut self, key: Czyx, value: Vec) -> Result<(), String> { + self.backend.put_checked(key, value) + } + /// Raw backend get. pub fn get_raw(&self, key: &Czyx) -> Option> { self.backend.get(key)