cubelinux/cubelinux-2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

fix(cubefs): round-trip full record envelope through DaemonBackend + gate hardening

Browse Source 
- DaemonBackend::put/get now round-trip the FULL record envelope (header+body)
  verbatim via rawput/rawget instead of stripping to a bare body. Before this,
  a socket-backed mount wrote the bare body but get returned it as if it were a
  record, breaking the FUSE read-back path (get_record could not decode it).
  This diverged from the in-memory backend and silently corrupted reads.
- Cargo fmt --check now passes (was failing; the committed tree was never a
  clean ./check, which is why the durable mount could regress undetected).
- Fix two new clippy lints (manual_is_multiple_of) in cubefs/backend.rs and
  cubesys/commands.rs so the -D warnings gate is green.
- Daemon-backed integration tests (cubefs_daemon_smoke, daemon_backend_smoke)
  now use the real CubeStore<DaemonBackend> record path and are #[ignore]d so
  the default gate (no daemon) stays green, while ./check daemon spins up a live
  cube-server and runs them via --ignored. This makes the durable-socket
  contract an actual enforced test, not a manual ad-hoc script.
This commit is contained in:
CUBELinux-2
2026-08-13 05:58:12 -04:00
parent 8f9e7e0025
commit 0300def300
6 changed files with 138 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
check
+33 -4
View File
@@ -46,15 +46,44 @@ cargo test --workspace $FEAT --jobs "$JOBS"
step "3/3 clippy"
cargo clippy --workspace --all-targets $FEAT --jobs "$JOBS" -- -D warnings
[ "${1:-}" = mount ] && { run_mount=1; }
[ "${1:-}" = bench ] && { run_bench=1; }
[ "${1:-}" = mount ] && { run_mount=1; }
[ "${1:-}" = bench ] && { run_bench=1; }
[ "${1:-}" = stress ] && { run_stress=1; }
[ "${1:-}" = daemon ] && { run_daemon=1; }
[ -z "${run_mount:-}${run_bench:-}${run_stress:-}" ] && {
printf '\n\033[32mALL CHECKS PASSED\033[0m (run: ./check mount | bench | stress for those stages)\n'
[ -z "${run_mount:-}${run_bench:-}${run_stress:-}${run_daemon:-}" ] && {
printf '\n\033[32mALL CHECKS PASSED\033[0m (run: ./check mount | bench | stress | daemon for those stages)\n'
exit 0
}
if [ "${run_daemon:-}" ]; then
step "4/4 daemon-backed integration tests (live cube-server)"
# Spin up a real daemon on a temp socket/store, then run the `#[ignore]`d
# tests that require CUBE_SOCK. Tears the daemon down afterwards.
DSOCK=$(mktemp -u /tmp/cube-check-daemon.XXXXXX.sock)
DDIR=$(mktemp -d /tmp/cube-check-daemon.XXXXXX)
DSTORE=$DDIR/cube-store.json
./target/debug/cube-server --socket "$DSOCK" --store "$DSTORE" \
--allow-anonymous --checkpoint-ms 400 --wal-fsync-ms 30 \
>/tmp/cube-check-daemon.log 2>&1 &
DPID=$!
for i in $(seq 1 30); do [ -S "$DSOCK" ] && break; sleep 0.3; done
if [ ! -S "$DSOCK" ]; then
echo "DAEMON FAILED to start:"; cat /tmp/cube-check-daemon.log
exit 1
fi
export CUBE_SOCK="$DSOCK"
cargo test -p cubefs --features cubefs/mount \
--test cubefs_daemon_smoke --test daemon_backend_smoke \
-- --ignored --nocapture
RC=$?
kill -9 "$DPID" 2>/dev/null || true
rm -rf "$DDIR" 2>/dev/null || true
[ "$RC" -eq 0 ] || exit "$RC"
printf '\n\033[32mALL CHECKS PASSED\033[0m (incl. daemon-backed tests)\n'
exit 0
fi
if [ "${run_bench:-}" ]; then
step "4/4 cube-bench"
# Release build + timed run. Asserts correctness on every path; numbers are
cubefs/src/backend.rs
+11 -5
View File
@@ -87,11 +87,14 @@ impl DaemonBackend {
impl CubeBackend for DaemonBackend {
fn put(&mut self, key: Czyx, value: Vec<u8>) {
// `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. Stripping the header
// here would make the socket backend diverge from the in-memory backend
// and break the round-trip.
let hex = to_hex(&value);
let cmd = format!(
"rawput {} {} {} {} {}",
key.c, key.z, key.y, key.x, hex
);
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
@@ -102,6 +105,9 @@ impl CubeBackend for DaemonBackend {
}
fn get(&self, key: &Czyx) -> Option<Vec<u8>> {
// Returns the SAME full envelope that `put` stored. The daemon's
// `get_raw` is the exact backend bytes (header + body), which is what
// `CubeStore::get_record` expects to decode.
let cmd = format!("rawget {} {} {} {}", key.c, key.z, key.y, key.x);
match self.rpc(&cmd) {
Ok(resp) => {
@@ -197,7 +203,7 @@ fn to_hex(b: &[u8]) -> String {
}
fn from_hex(s: &str) -> Option<Vec<u8>> {
if s.len() % 2 != 0 {
if !s.len().is_multiple_of(2) {
return None;
}
let bytes = s.as_bytes();
cubefs/src/bin/cubefs_mount.rs
+2 -2
View File
@@ -49,13 +49,13 @@ fn main() -> ExitCode {
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>,
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>,
Box::new(HashBackend::new()) as Box<dyn CubeBackend + Send + Sync>
))
}
};
cubefs/tests/cubefs_daemon_smoke.rs
+51 -13
View File
@@ -1,25 +1,63 @@
// Ad-hoc smoke test: CubeFs + DaemonBackend (NO FUSE) against a live daemon.
// Bisects whether the FUSE layer or the cubefs/cubestore layer is the culprit.
// Run: CUBE_SOCK=<sock> cargo test --test cubefs_daemon_smoke -- --nocapture
// Daemon-backed integration test for CubeFs<DaemonBackend> against a live
// cube-server. Proves a record written through the SAME path the FUSE mount
// uses (CubeStore::put_record -> DaemonBackend::put -> rawput) is readable
// back through a second CubeStore<DaemonBackend> (the FUSE read path,
// CubeStore::get_record -> DaemonBackend::get -> rawget), byte-for-byte.
//
// Requires a live cube-server at CUBE_SOCK. Marked `#[ignore]` so the default
// gate (no daemon) stays green; `./check daemon` spins up a daemon and runs it
// via `--ignored`.
// Run manually:
// CUBE_SOCK=<sock> cargo test --test cubefs_daemon_smoke -- --ignored --nocapture
use cubecoords::Czyx;
use cubefs::{CubeFs, DaemonBackend};
use cubestore::{CubeBackend, CubeStore};
#[test]
#[ignore = "requires a live cube-server; run via `./check daemon`"]
fn cubefs_create_write_reaches_daemon() {
let sock = std::env::var("CUBE_SOCK").expect("set CUBE_SOCK");
let mut fs = CubeFs::new(CubeStore::new(DaemonBackend::new(sock)));
let sock = std::env::var("CUBE_SOCK").expect("set CUBE_SOCK to a live cube-server socket");
let mut fs = CubeFs::new(CubeStore::new(DaemonBackend::new(sock.clone())));
fs.format("cube0");
let path = "/c011/z007/y003/x009";
fs.create(path, 0, 0, 0o644).expect("create");
fs.write(path, 0, b"fuse-proof-XYZ", 0, 0).expect("write");
// Now ask the daemon directly via a second backend.
let mut probe = DaemonBackend::new(std::env::var("CUBE_SOCK").unwrap());
// Ask the daemon directly via a SECOND CubeStore (the FUSE read path),
// which calls CubeStore::get_record -> DaemonBackend::get (rawget). The
// daemon must return the full record envelope, which get_record splits.
let probe = CubeStore::new(DaemonBackend::new(sock));
let k = Czyx::new(11, 7, 3, 9);
let got = probe.get(&k);
println!("daemon-side get(11,7,3,9) => {:?}", got);
assert_eq!(got.as_deref(), Some(&b"fuse-proof-XYZ"[..]), "CubeFs write must reach daemon store");
let keys = probe.keys();
println!("daemon keys => {:?}", keys);
assert!(keys.contains(&k));
let (_, body) = probe
.get_record(&k)
.expect("record must be present in daemon store");
println!("daemon-side get_record(11,7,3,9) body => {:?}", body);
assert_eq!(
body.as_slice(),
b"fuse-proof-XYZ",
"CubeFs write must reach daemon store"
);
}
// Keep a direct raw-level check too, mirroring DaemonBackend::put/get semantics
// (full-envelope round-trip, NOT bare-body).
#[test]
#[ignore = "requires a live cube-server; run via `./check daemon`"]
fn daemon_backend_envelope_roundtrip() {
let sock = std::env::var("CUBE_SOCK").expect("set CUBE_SOCK to a live cube-server socket");
let mut b = DaemonBackend::new(sock);
let k = Czyx::new(9, 2, 2, 1);
// `put` receives the full envelope a CubeStore::put_record would emit
// (header length word + header bytes + body). We use a zero-length header
// here; the point is the envelope must survive the round-trip intact.
let mut env = Vec::new();
env.extend_from_slice(&(0u32).to_le_bytes());
env.extend_from_slice(b"fuse-persist-proof");
b.put(k, env.clone());
let got = b.get(&k).expect("envelope must round-trip");
println!("DaemonBackend::get(9,2,2,1) => {} bytes", got.len());
assert_eq!(
got, env,
"daemon backend put/get must preserve the full envelope"
);
}
cubefs/tests/daemon_backend_smoke.rs
+20 -9
View File
@@ -1,21 +1,32 @@
// Ad-hoc smoke test for DaemonBackend end-to-end against a live cube-server.
// Run with: cargo test --test daemon_backend_smoke -- --nocapture
// Exercises the raw backend put/get at the level the FUSE layer actually uses:
// `put` is handed the full record envelope (header bytes + body), and `get`
// must return exactly that envelope back so CubeStore::get_record can decode it.
// Requires CUBE_SOCK env var pointing at a running cube-server socket.
// Marked `#[ignore]` so the default gate (no daemon) stays green; `./check
// daemon` spins up a daemon and runs it via `--ignored`.
// Run manually:
// CUBE_SOCK=<sock> cargo test --test daemon_backend_smoke -- --ignored --nocapture
use cubecoords::Czyx;
use cubefs::DaemonBackend;
use cubestore::CubeBackend;
#[test]
#[ignore = "requires a live cube-server; run via `./check daemon`"]
fn daemon_backend_put_get_roundtrip() {
let sock = std::env::var("CUBE_SOCK").expect("set CUBE_SOCK to a live cube-server socket");
let mut b = DaemonBackend::new(sock);
let k = Czyx::new(9, 2, 2, 1);
b.put(k, b"fuse-persist-proof".to_vec());
let got = b.get(&k);
println!("DaemonBackend::get(9,2,2,1) => {:?}", got);
assert_eq!(got.as_deref(), Some(&b"fuse-persist-proof"[..]), "daemon backend put/get roundtrip");
// also exercise keys()
let keys = b.keys();
println!("DaemonBackend::keys() => {:?}", keys);
assert!(keys.contains(&k), "keys() should include coord after put");
// `put` is passed the full envelope (header + body), exactly as
// CubeStore::put_record would hand it. We store a header-len word + body.
let mut envelope = Vec::new();
envelope.extend_from_slice(&(0u32).to_le_bytes()); // header length (0)
envelope.extend_from_slice(b"fuse-persist-proof");
b.put(k, envelope.clone());
let got = b.get(&k).expect("envelope must be returned by get");
println!("DaemonBackend::get(9,2,2,1) => {} bytes", got.len());
assert_eq!(
got, envelope,
"daemon backend put/get roundtrip must preserve the full envelope"
);
}
cubesys/src/commands.rs
+21 -9
View File
@@ -582,8 +582,7 @@ impl Session {
let hex = it
.next()
.ok_or_else(|| "rawput needs <hex-bytes>".to_string())?;
let val = hex_decode(hex)
.ok_or_else(|| "rawput: value must be hex".to_string())?;
let val = hex_decode(hex).ok_or_else(|| "rawput: value must be hex".to_string())?;
let coord = Czyx::new(c, z, y, x);
store.put_raw(coord, val);
Ok(format!("ok: wrote {}", coord.pack_u32()))
@@ -598,10 +597,18 @@ impl Session {
Ok(format!("ok: deleted {}", coord.pack_u32()))
}
"rawkeys" => {
let ks: Vec<String> =
store.keys().iter().map(|k| k.pack_u32().to_string()).collect();
Ok(format!("ok: {} keys", ks.len()))
.map(|s| if ks.is_empty() { s } else { format!("{s}\n{}", ks.join(" ")) })
let ks: Vec<String> = store
.keys()
.iter()
.map(|k| k.pack_u32().to_string())
.collect();
Ok(format!("ok: {} keys", ks.len())).map(|s| {
if ks.is_empty() {
s
} else {
format!("{s}\n{}", ks.join(" "))
}
})
}
"rawscan" => {
let c = parse_u8(it.next(), "rawscan needs <c>")?;
@@ -612,8 +619,13 @@ impl Session {
.iter()
.map(|k| k.pack_u32().to_string())
.collect();
Ok(format!("ok: {} keys", ks.len()))
.map(|s| if ks.is_empty() { s } else { format!("{s}\n{}", ks.join(" ")) })
Ok(format!("ok: {} keys", ks.len())).map(|s| {
if ks.is_empty() {
s
} else {
format!("{s}\n{}", ks.join(" "))
}
})
}
"grant" => {
// Issue a permission grant (Task 6b / PDF flags 5-19). Only an
@@ -933,7 +945,7 @@ fn hex_encode(b: &[u8]) -> String {
/// Decode a hex string into bytes. Rejects odd length / non-hex.
fn hex_decode(s: &str) -> Option<Vec<u8>> {
if s.len() % 2 != 0 {
if !s.len().is_multiple_of(2) {
return None;
}
let bytes = s.as_bytes();