Give up on a stalled 9p request instead of hanging the mount forever

src/client.rs

@@ -14,6 +14,7 @@ use std::collections::HashMap;
 use std::io;
 use std::sync::atomic::{AtomicU16, AtomicU32, Ordering};
 use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
 use tokio::sync::oneshot;
 
 const NOTAG: u16 = 0xffff;
@@ -25,9 +26,17 @@ struct Frame {
     body: Vec<u8>,
 }
 
+/// A request that has been written to the transport and is waiting for its reply.
+struct PendingReq {
+    tx: oneshot::Sender<Frame>,
+    /// T-message type, so a stuck request can be named when the stall watchdog fires.
+    mtype: u8,
+    sent_at: Instant,
+}
+
 pub struct NineClient {
     sink: tokio::sync::Mutex<ByteSink>,
-    pending: Mutex<HashMap<u16, oneshot::Sender<Frame>>>,
+    pending: Mutex<HashMap<u16, PendingReq>>,
     next_tag: AtomicU16,
     next_fid: AtomicU32,
     pub msize: u32,
@@ -50,6 +59,7 @@ impl NineClient {
         msize: u32,
         n_uname: u32,
         aname: &str,
+        stall_timeout: Duration,
     ) -> Result<(Arc<NineClient>, Qid), Box<dyn std::error::Error>> {
         let (sink, mut stream) = transport.split();
 
@@ -86,6 +96,42 @@ impl NineClient {
             let _ = pump.transport_gone.send(true);
         });
 
+        // Stall watchdog: a reply that never arrives -- a silently dead tunnel, or a server wedged
+        // on its backing store -- would otherwise hang its waiter (and with it the serial FUSE
+        // session loop) forever with no other symptom. Declare the transport dead instead, which
+        // fails all waiters and makes the mount exit so its supervisor can remount.
+        if !stall_timeout.is_zero() {
+            let watch = client.clone();
+            tokio::spawn(async move {
+                let mut ticker = tokio::time::interval(Duration::from_secs(5));
+                loop {
+                    ticker.tick().await;
+                    if *watch.transport_gone.subscribe().borrow() {
+                        return;
+                    }
+                    let oldest = {
+                        let map = watch.pending.lock().unwrap();
+                        map.iter()
+                            .map(|(tag, p)| (*tag, p.mtype, p.sent_at.elapsed()))
+                            .max_by_key(|(_, _, age)| *age)
+                    };
+                    if let Some((tag, mtype, age)) = oldest {
+                        if age > stall_timeout {
+                            tracing::error!(
+                                tag,
+                                mtype = tmsg_name(mtype),
+                                age_secs = age.as_secs(),
+                                "9p request stalled; declaring the transport dead"
+                            );
+                            watch.pending.lock().unwrap().clear();
+                            let _ = watch.transport_gone.send(true);
+                            return;
+                        }
+                    }
+                }
+            });
+        }
+
         // Version handshake (tag NOTAG), then attach.
         let negotiated = client.version(msize).await?;
         // Re-stamp msize is immutable on the struct; we only ever send <= negotiated, and our reads
@@ -111,8 +157,8 @@ impl NineClient {
             typ: frame.typ,
             body: frame.body[2..].to_vec(),
         };
-        if let Some(tx) = self.pending.lock().unwrap().remove(&tag) {
-            let _ = tx.send(payload);
+        if let Some(p) = self.pending.lock().unwrap().remove(&tag) {
+            let _ = p.tx.send(payload);
         }
     }
 
@@ -133,7 +179,14 @@ impl NineClient {
     /// transport/protocol failure (mapped to EIO).
     async fn transact(&self, mtype: u8, tag: u16, body: &[u8]) -> Result<Frame, i32> {
         let (tx, rx) = oneshot::channel();
-        self.pending.lock().unwrap().insert(tag, tx);
+        self.pending.lock().unwrap().insert(
+            tag,
+            PendingReq {
+                tx,
+                mtype,
+                sent_at: Instant::now(),
+            },
+        );
 
         let size = (4 + 1 + 2 + body.len()) as u32;
         let mut frame = Vec::with_capacity(size as usize);

src/fuse9p.rs

@@ -289,11 +289,15 @@ impl Fuse9p {
         detach_on_transport_loss: bool,
         // Whether to mount with `default_permissions`.
         default_permissions: bool,
+        // Give up on a 9p request unanswered for this long: the transport is declared dead, the
+        // mount exits, and the supervisor remounts. Zero disables the watchdog.
+        stall_timeout: std::time::Duration,
     ) -> Result<(), Box<dyn std::error::Error>> {
         tracing::info!(?tuning, "mount9p-fuse: tuning");
         // Attach as `uid` so the server acts as that user for file ops (a multiuser server like diod
         // setfsuids to it per attach), so files are owned by `uid` and chmod works.
-        let (client, root_qid) = NineClient::connect(transport, msize, uid, aname).await?;
+        let (client, root_qid) =
+            NineClient::connect(transport, msize, uid, aname, stall_timeout).await?;
         // Watch for the 9p transport closing: if it does, the mount is dead (every op would just
         // return EIO), so we exit below and let whatever supervises this process remount.
         let mut transport_gone = client.transport_gone();

src/lib.rs

@@ -50,5 +50,16 @@ pub async fn mount(
     aname: &str,
     tuning: Tuning,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    Fuse9p::run(transport, mountpoint, msize, uid, aname, tuning, true, true).await
+    Fuse9p::run(
+        transport,
+        mountpoint,
+        msize,
+        uid,
+        aname,
+        tuning,
+        true,
+        true,
+        std::time::Duration::from_secs(30),
+    )
+    .await
 }

src/main.rs

@@ -45,6 +45,12 @@ enum Cmd {
         /// revalidates, so out-of-band writes show up.
         #[arg(long, default_value = "mmap")]
         cache: String,
+        /// Seconds an outstanding 9p request may go unanswered before the daemon gives up, detaches
+        /// the mount, and exits (so a supervisor can remount). The kernel v9fs client itself waits
+        /// forever, so without this a wedged server or silently dead tunnel hangs every process
+        /// touching the mount. 0 disables the watchdog.
+        #[arg(long, default_value_t = 30)]
+        stall_timeout: u64,
         mountpoint: PathBuf,
     },
 
@@ -105,6 +111,11 @@ enum Cmd {
         /// Let the kernel enforce permissions against the owner/mode the 9p server reports.
         #[arg(long, default_value_t = true, action = clap::ArgAction::Set)]
         default_permissions: bool,
+        /// Seconds an outstanding 9p request may go unanswered before the daemon gives up and exits
+        /// (so a supervisor can remount). Without this a wedged server or silently dead tunnel hangs
+        /// every process touching the mount. 0 disables the watchdog.
+        #[arg(long, default_value_t = 30)]
+        stall_timeout: u64,
 
         mountpoint: PathBuf,
     },
@@ -131,10 +142,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             headers,
             msize,
             cache,
+            stall_timeout,
             mountpoint,
         } => {
             let transport = build_transport(&connect, &parse_headers(&headers)?).await?;
-            mount9p::mount9p(transport, &mountpoint, msize, &cache).await
+            mount9p::mount9p(
+                transport,
+                &mountpoint,
+                msize,
+                &cache,
+                std::time::Duration::from_secs(stall_timeout),
+            )
+            .await
         }
         Cmd::Mount9pFuse {
             connect,
@@ -150,6 +169,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             wb_depth,
             detach_on_transport_loss,
             default_permissions,
+            stall_timeout,
             mountpoint,
         } => {
             let transport = build_transport(&connect, &parse_headers(&headers)?).await?;
@@ -171,6 +191,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 tuning,
                 detach_on_transport_loss,
                 default_permissions,
+                std::time::Duration::from_secs(stall_timeout),
             )
             .await
         }