Replace custom PassthroughFs with fuse-backend-rs

fuse-pipe/Cargo.toml

@@ -30,8 +30,12 @@ async-trait = "0.1"
 # Logging
 tracing = "0.1"
 
-# Optional: FUSE client (using fork with multi-reader support)
-fuser = { path = "../../fuser-fork", optional = true }
+# fuse-backend-rs for passthrough filesystem (Cloud Hypervisor's production-grade FUSE backend)
+fuse-backend-rs = { version = "0.12", default-features = false, features = ["fusedev"] }
+vm-memory = "0.14"
+
+# Optional: FUSE client (using fork with multi-reader support via FUSE_DEV_IOC_CLONE)
+fuser = { path = "../fuser-fork", optional = true }
 
 # Concurrent data structures
 dashmap = "5.5"
@@ -47,6 +51,7 @@ clap = { version = "4.0", features = ["derive"] }
 serde_json = "1.0"
 metrics = "0.23"
 metrics-util = "0.17"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [[bench]]
 name = "protocol"

fuse-pipe/src/client/mount.rs

@@ -106,6 +106,7 @@ pub fn mount_with_options<P: AsRef<Path>>(
 
             for (reader_id, cloned_fd) in fds_vec {
                 let fs = FuseClient::new(Arc::clone(&mux_for_callback), reader_id as u32);
+                // Each cloned fd handles its own request/response pairs
                 let mut reader_session =
                     fuser::Session::from_fd_initialized(fs, cloned_fd, fuser::SessionACL::Owner);
 
@@ -254,6 +255,7 @@ pub fn mount_vsock_with_options<P: AsRef<Path>>(
 
             for (reader_id, cloned_fd) in fds_vec {
                 let fs = FuseClient::new(Arc::clone(&mux_for_callback), reader_id as u32);
+                // Each cloned fd handles its own request/response pairs
                 let mut reader_session =
                     fuser::Session::from_fd_initialized(fs, cloned_fd, fuser::SessionACL::Owner);
 

fuse-pipe/src/client/multiplexer.rs

@@ -202,16 +202,19 @@ fn reader_loop(mut socket: UnixStream, pending: Arc<DashMap<u64, Sender<Response
         }
 
         // Deserialize and route to waiting reader (lock-free lookup + remove)
-        if let Ok(wire) = bincode::deserialize::<WireResponse>(&resp_buf) {
-            // Mark client receive time on the span
-            let mut span = wire.span;
-            if let Some(ref mut s) = span {
-                s.mark("client_recv");
-            }
-
-            if let Some((_, tx)) = pending.remove(&wire.unique) {
-                let _ = tx.send((wire.response, span));
+        match bincode::deserialize::<WireResponse>(&resp_buf) {
+            Ok(wire) => {
+                // Mark client receive time on the span
+                let mut span = wire.span;
+                if let Some(ref mut s) = span {
+                    s.mark("client_recv");
+                }
+
+                if let Some((_, tx)) = pending.remove(&wire.unique) {
+                    let _ = tx.send((wire.response, span));
+                }
             }
+            Err(_) => {}
         }
     }
 }