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

@@ -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<u8>) -> 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}")),
         }
     }
 

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<B: CubeBackend> CubeFs<B> {
         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<B: CubeBackend> CubeFs<B> {
         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<B: CubeBackend> CubeFs<B> {
         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<u8>) {}
+        fn put_checked(&mut self, _key: Czyx, _value: Vec<u8>) -> Result<(), String> {
+            Err("simulated daemon unreachable".to_string())
+        }
+        fn get(&self, _key: &Czyx) -> Option<Vec<u8>> {
+            // 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<Czyx> {
+            vec![]
+        }
+        fn scan_prefix(&self, _c: u8, _z: Option<u8>, _y: Option<u8>) -> Vec<Czyx> {
+            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<u8>) {}
+        fn put_checked(&mut self, _key: Czyx, _value: Vec<u8>) -> Result<(), String> {
+            Err("simulated daemon unreachable".to_string())
+        }
+        fn get(&self, _key: &Czyx) -> Option<Vec<u8>> {
+            None
+        }
+        fn delete(&mut self, _key: &Czyx) {}
+        fn keys(&self) -> Vec<Czyx> {
+            vec![]
+        }
+        fn scan_prefix(&self, _c: u8, _z: Option<u8>, _y: Option<u8>) -> Vec<Czyx> {
+            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);
+    }
 }

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<u8>);
+    /// 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<u8>) -> Result<(), String> {
+        self.put(key, value);
+        Ok(())
+    }
     /// Fetch the value at `key`, if present.
     fn get(&self, key: &Czyx) -> Option<Vec<u8>>;
     /// Remove the value at `key`.
@@ -356,7 +365,25 @@ impl<B: CubeBackend> CubeStore<B> {
         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<B: CubeBackend> CubeStore<B> {
         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<u8>) {
-        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<u8>) -> Result<(), String> {
+        self.backend.put_checked(key, value)
+    }
+
     /// Raw backend get.
     pub fn get_raw(&self, key: &Czyx) -> Option<Vec<u8>> {
         self.backend.get(key)