fix(cubefs): make FUSE mount a durable view of cube-server daemon store
The cubefs-mount --socket path was never actually built: CubeFs<B> is generic, so the --socket (DaemonBackend) and default (HashBackend) arms of the match were incompatible types (E0308), and --features mount failed to compile. The running binary was therefore the in-memory build, so --socket was silently ignored and every FUSE write went to RAM and never reached the daemon (rawget/rawkeys returned none / 0 keys, WAL stayed 0). Fix: type-erase the backend. Add (forwards to inner) in cubestore, and build in cubefs-mount via Box::new(DaemonBackend::new(p)) / Box::new(HashBackend::new()). Keeps cubefs free of a cubesys dep (acyclic graph). Verified end-to-end on host (shared code path as the VM): a FUSE write via lands in the daemon store (rawget returns the record, 5 keys present, WAL grows), and survives a of the daemon + relaunch with the same --store (byte-identical read-back). Also includes (from RESUME-cubefs-daemon.md): cubesys raw* command family (rawget/rawput/rawdel/rawkeys/rawscan + parse/hex helpers) and the DaemonBackend client + smoke tests. Report/verification docs added. Note: VM cubefs.service still mounts in-memory (no --socket); update the unit to as a follow-up so the deployed VM FUSE is durable too.
This commit is contained in:
CUBELinux-2
@@ -0,0 +1,213 @@
|
||||
//! `DaemonBackend` — a [`CubeBackend`] that talks to a running `cube-server`
|
||||
//! over its Unix-domain socket.
|
||||
//!
|
||||
//! This is what makes cubefs a *real* FUSE view of the daemon's durable store
|
||||
//! (PDF Package 3: "cubefs: optional FUSE filesystem view so cube records
|
||||
//! appear as files/directories"), instead of an isolated in-memory map that
|
||||
//! loses every write on unmount. The daemon serves the raw coordinate
|
||||
//! commands (`rawget`/`rawput`/`rawdel`/`rawkeys`/`rawscan`) over the same
|
||||
//! text protocol it uses for `cubec`; `DaemonBackend` is the matching client.
|
||||
//!
|
||||
//! Wire format: length-prefixed UTF-8 frames (see `cubesys::net`). We reuse
|
||||
//! the framing by hand so this crate stays free of any cubesys dependency —
|
||||
//! that keeps the dependency graph acyclic (cubefs is already depended on by
|
||||
//! cubesys, so adding a cubesys dep here would create the very cycle we are
|
||||
//! avoiding).
|
||||
|
||||
use cubecoords::Czyx;
|
||||
use cubestore::CubeBackend;
|
||||
use std::io::{Read, Write};
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
/// A [`CubeBackend`] backed by a `cube-server` Unix socket.
|
||||
///
|
||||
/// Cheaply cloneable: each clone opens its own connection on first use, so
|
||||
/// the backend can be shared across the FUSE worker threads without a global
|
||||
/// mutex on the socket (the daemon already serializes per-command internally
|
||||
/// via its store lock).
|
||||
pub struct DaemonBackend {
|
||||
socket: PathBuf,
|
||||
connect_timeout: Duration,
|
||||
}
|
||||
|
||||
impl Clone for DaemonBackend {
|
||||
/// Clones share the socket path but NOT any open connection — each clone
|
||||
/// opens its own connection per call, so FUSE worker threads never contend
|
||||
/// on one shared `UnixStream` (which is not `Clone` anyway).
|
||||
fn clone(&self) -> Self {
|
||||
DaemonBackend {
|
||||
socket: self.socket.clone(),
|
||||
connect_timeout: self.connect_timeout,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl DaemonBackend {
|
||||
/// Connect to the daemon at `socket`. The connection is established lazily
|
||||
/// on the first `get`/`put`/`delete` (so construction never blocks).
|
||||
pub fn new(socket: impl Into<PathBuf>) -> Self {
|
||||
DaemonBackend {
|
||||
socket: socket.into(),
|
||||
connect_timeout: Duration::from_secs(5),
|
||||
}
|
||||
}
|
||||
|
||||
/// The default daemon socket path (`$XDG_RUNTIME_DIR/cube/cube.sock`).
|
||||
pub fn default_socket() -> PathBuf {
|
||||
if let Ok(r) = std::env::var("XDG_RUNTIME_DIR") {
|
||||
PathBuf::from(r).join("cube/cube.sock")
|
||||
} else {
|
||||
PathBuf::from("/run/cube/cube.sock")
|
||||
}
|
||||
}
|
||||
|
||||
fn connect(&self) -> std::io::Result<UnixStream> {
|
||||
let s = UnixStream::connect(&self.socket)?;
|
||||
s.set_read_timeout(Some(Duration::from_secs(10)))?;
|
||||
s.set_write_timeout(Some(self.connect_timeout))?;
|
||||
Ok(s)
|
||||
}
|
||||
|
||||
fn rpc(&self, cmd: &str) -> Result<String, String> {
|
||||
// Open a FRESH connection per call. A cached stream goes stale the
|
||||
// instant the daemon restarts (kill -9 / crash / redeploy): the socket
|
||||
// is still `Some` but dead, so every later write fails silently and
|
||||
// FUSE reports success while nothing reaches the store. That was the
|
||||
// bug. The daemon already serializes per command, so per-call
|
||||
// connections are correct and contention-free.
|
||||
let mut s = self
|
||||
.connect()
|
||||
.map_err(|e| format!("daemon socket {cmd}: {e}"))?;
|
||||
write_frame(&mut s, cmd).map_err(|e| format!("daemon socket {cmd}: {e}"))?;
|
||||
read_frame(&mut s).map_err(|e| format!("daemon socket {cmd}: {e}"))
|
||||
}
|
||||
}
|
||||
|
||||
impl CubeBackend for DaemonBackend {
|
||||
fn put(&mut self, key: Czyx, value: Vec<u8>) {
|
||||
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}");
|
||||
}
|
||||
}
|
||||
|
||||
fn get(&self, key: &Czyx) -> Option<Vec<u8>> {
|
||||
let cmd = format!("rawget {} {} {} {}", key.c, key.z, key.y, key.x);
|
||||
match self.rpc(&cmd) {
|
||||
Ok(resp) => {
|
||||
let resp = resp.trim();
|
||||
if let Some(hex) = resp.strip_prefix("ok: ") {
|
||||
from_hex(hex)
|
||||
} else {
|
||||
// "none" or an error reply => absent.
|
||||
None
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("cubefs: rawget {key:?} failed: {e}");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn delete(&mut self, key: &Czyx) {
|
||||
let cmd = format!("rawdel {} {} {} {}", key.c, key.z, key.y, key.x);
|
||||
if let Err(e) = self.rpc(&cmd) {
|
||||
eprintln!("cubefs: rawdel {key:?} failed: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
fn keys(&self) -> Vec<Czyx> {
|
||||
match self.rpc("rawkeys") {
|
||||
Ok(resp) => parse_key_list(&resp),
|
||||
Err(e) => {
|
||||
eprintln!("cubefs: rawkeys failed: {e}");
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn scan_prefix(&self, c: u8, z: Option<u8>, y: Option<u8>) -> Vec<Czyx> {
|
||||
let cmd = match (z, y) {
|
||||
(Some(z), Some(y)) => format!("rawscan {c} {z} {y}"),
|
||||
(Some(z), None) => format!("rawscan {c} {z}"),
|
||||
_ => format!("rawscan {c}"),
|
||||
};
|
||||
match self.rpc(&cmd) {
|
||||
Ok(resp) => parse_key_list(&resp),
|
||||
Err(e) => {
|
||||
eprintln!("cubefs: rawscan {c} {z:?} {y:?} failed: {e}");
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse the `ok: N keys\n<space-separated pack_u32 list>` reply into
|
||||
/// [`Czyx`] coordinates. Tolerant of the `ok: N keys` prefix being absent.
|
||||
fn parse_key_list(resp: &str) -> Vec<Czyx> {
|
||||
let body = resp.trim();
|
||||
// Drop the "ok: N keys" summary line if present; the payload is the rest.
|
||||
let payload = match body.split_once('\n') {
|
||||
Some((head, rest)) if head.starts_with("ok:") => rest,
|
||||
_ => body,
|
||||
};
|
||||
payload
|
||||
.split_whitespace()
|
||||
.filter_map(|tok| tok.parse::<u32>().ok())
|
||||
.map(Czyx::unpack_u32)
|
||||
.collect()
|
||||
}
|
||||
|
||||
// --- local copies of the length-prefixed framing used by cubesys::net ---
|
||||
// Duplicated (not imported) to keep cubefs free of a cubesys dependency.
|
||||
|
||||
fn write_frame<W: Write>(w: &mut W, payload: &str) -> std::io::Result<()> {
|
||||
let bytes = payload.as_bytes();
|
||||
w.write_all(&(bytes.len() as u32).to_le_bytes())?;
|
||||
w.write_all(bytes)?;
|
||||
w.flush()
|
||||
}
|
||||
|
||||
fn read_frame<R: Read>(r: &mut R) -> std::io::Result<String> {
|
||||
let mut len_buf = [0u8; 4];
|
||||
r.read_exact(&mut len_buf)?;
|
||||
let len = u32::from_le_bytes(len_buf) as usize;
|
||||
let mut buf = vec![0u8; len];
|
||||
r.read_exact(&mut buf)?;
|
||||
String::from_utf8(buf).map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
|
||||
}
|
||||
|
||||
fn to_hex(b: &[u8]) -> String {
|
||||
let mut s = String::with_capacity(b.len() * 2);
|
||||
for byte in b {
|
||||
s.push_str(&format!("{byte:02x}"));
|
||||
}
|
||||
s
|
||||
}
|
||||
|
||||
fn from_hex(s: &str) -> Option<Vec<u8>> {
|
||||
if s.len() % 2 != 0 {
|
||||
return None;
|
||||
}
|
||||
let bytes = s.as_bytes();
|
||||
let mut out = Vec::with_capacity(s.len() / 2);
|
||||
let mut i = 0;
|
||||
while i < bytes.len() {
|
||||
let hi = (bytes[i] as char).to_digit(16)?;
|
||||
let lo = (bytes[i + 1] as char).to_digit(16)?;
|
||||
out.push(((hi << 4) | lo) as u8);
|
||||
i += 2;
|
||||
}
|
||||
Some(out)
|
||||
}
|
||||
@@ -1,25 +1,31 @@
|
||||
//! `cubefs-mount` — mount a cube as a POSIX filesystem.
|
||||
//!
|
||||
//! Usage: `cubefs-mount <mountpoint> [--label NAME] [--seed] [--allow-other]`
|
||||
//! Usage:
|
||||
//! cubefs-mount <mountpoint> [--label NAME] [--seed] [--allow-other]
|
||||
//! [--socket PATH]
|
||||
//!
|
||||
//! `--socket PATH` mounts the **daemon's** durable store over a Unix socket
|
||||
//! (see `cubefs::DaemonBackend`). A write through the FUSE mount then lands in
|
||||
//! the daemon's WAL and survives an unmount — or a daemon restart — exactly as
|
||||
//! the PDF's "cubefs is a FUSE view of the cube" intends. Without `--socket`
|
||||
//! the mount uses the in-memory `HashBackend` (ephemeral; every write is lost
|
||||
//! on unmount), which is still useful for namespace-mapping tests and for
|
||||
//! machines with no `cube-server` running.
|
||||
//!
|
||||
//! `--allow-other` lets users other than the mounting user reach the
|
||||
//! filesystem. Without it the kernel rejects them at the mountpoint before any
|
||||
//! request reaches us, so multi-user ACL behaviour cannot be observed.
|
||||
//!
|
||||
//! The backing store is the in-memory [`HashBackend`] for now: Package 3's job
|
||||
//! is the *namespace mapping*, and a durable on-disk backend is a cubestore
|
||||
//! concern that gets swapped in by changing one type parameter here. `--seed`
|
||||
//! populates a few records so the mount has something to `ls`.
|
||||
|
||||
use cubefs::fuse::CubeFuse;
|
||||
use cubefs::CubeFs;
|
||||
use cubestore::{CubeStore, HashBackend};
|
||||
use cubefs::{CubeFs, DaemonBackend};
|
||||
use cubestore::{CubeBackend, CubeStore, HashBackend};
|
||||
use std::process::ExitCode;
|
||||
|
||||
fn main() -> ExitCode {
|
||||
let args: Vec<String> = std::env::args().collect();
|
||||
let Some(mountpoint) = args.get(1).filter(|a| !a.starts_with("--")) else {
|
||||
eprintln!("usage: cubefs-mount <mountpoint> [--label NAME] [--seed]");
|
||||
eprintln!(
|
||||
"usage: cubefs-mount <mountpoint> [--label NAME] [--seed] [--socket PATH] [--allow-other]"
|
||||
);
|
||||
return ExitCode::from(2);
|
||||
};
|
||||
let label = args
|
||||
@@ -30,8 +36,29 @@ fn main() -> ExitCode {
|
||||
.unwrap_or_else(|| "cube0".to_string());
|
||||
let seed = args.iter().any(|a| a == "--seed");
|
||||
let allow_other = args.iter().any(|a| a == "--allow-other");
|
||||
let socket = args
|
||||
.iter()
|
||||
.position(|a| a == "--socket")
|
||||
.and_then(|i| args.get(i + 1).cloned());
|
||||
|
||||
let mut fs = CubeFs::new(CubeStore::new(HashBackend::new()));
|
||||
// Backing store: the daemon's durable store (over a socket) when `--socket`
|
||||
// is given, otherwise an ephemeral in-memory map. Both are type-erased to
|
||||
// `Box<dyn CubeBackend + Send + Sync>` so `CubeFs` has a single concrete
|
||||
// type regardless of which backend was chosen at runtime.
|
||||
let mut fs: CubeFs<Box<dyn CubeBackend + Send + Sync>> = match &socket {
|
||||
Some(path) => {
|
||||
eprintln!("cubefs: mounting daemon store at socket {path}");
|
||||
CubeFs::new(CubeStore::new(
|
||||
Box::new(DaemonBackend::new(path.clone())) as Box<dyn CubeBackend + Send + Sync>,
|
||||
))
|
||||
}
|
||||
None => {
|
||||
eprintln!("cubefs: mounting IN-MEMORY store (writes are NOT durable; use --socket PATH to mount the daemon's store)");
|
||||
CubeFs::new(CubeStore::new(
|
||||
Box::new(HashBackend::new()) as Box<dyn CubeBackend + Send + Sync>,
|
||||
))
|
||||
}
|
||||
};
|
||||
fs.format(&label);
|
||||
|
||||
if seed {
|
||||
@@ -73,9 +100,10 @@ fn main() -> ExitCode {
|
||||
opts.push(fuser::MountOption::AllowOther);
|
||||
}
|
||||
eprintln!(
|
||||
"mounting cubefs at {mountpoint} (label={label}, seed={seed}, allow_other={allow_other}) — ctrl-c to unmount"
|
||||
"mounting cubefs at {mountpoint} (label={label}, socket={:?}, seed={seed}, allow_other={allow_other}) — ctrl-c to unmount",
|
||||
socket
|
||||
);
|
||||
match fuser::mount2(CubeFuse::new(fs), mountpoint, &opts) {
|
||||
match fuser::mount2(cubefs::fuse::CubeFuse::new(fs), mountpoint, &opts) {
|
||||
Ok(()) => ExitCode::SUCCESS,
|
||||
Err(e) => {
|
||||
eprintln!("mount failed: {e}");
|
||||
|
||||
@@ -39,6 +39,7 @@
|
||||
#![forbid(unsafe_code)]
|
||||
#![warn(missing_docs)]
|
||||
|
||||
pub mod backend;
|
||||
pub mod nullspace;
|
||||
pub mod path;
|
||||
pub mod vfs;
|
||||
@@ -46,6 +47,7 @@ pub mod vfs;
|
||||
#[cfg(feature = "mount")]
|
||||
pub mod fuse;
|
||||
|
||||
pub use backend::DaemonBackend;
|
||||
pub use nullspace::{Acl, JournalEntry, JournalOp, NullSpace, VolumeMeta};
|
||||
pub use path::{czyx_to_ino, ino_to_czyx, parse_path, render_path, PathError, ROOT_INO};
|
||||
pub use vfs::{Attr, CubeFs, FsError, Kind};
|
||||
|
||||
Reference in New Issue
Block a user