diff --git a/libshpool/src/daemon/server.rs b/libshpool/src/daemon/server.rs index 20814ea0..7f7a6766 100644 --- a/libshpool/src/daemon/server.rs +++ b/libshpool/src/daemon/server.rs @@ -625,36 +625,76 @@ impl Server { fn handle_detach(&self, mut stream: UnixStream, request: DetachRequest) -> anyhow::Result<()> { let mut not_found_sessions = vec![]; let mut not_attached_sessions = vec![]; + + // Resolve the requested names to control handles while the shells lock + // is held, then drop it. The ctl handshake below MUST NOT run under + // that lock: client_connection and client_connection_ack are both + // rendezvous channels (bounded(0)), so each half only completes when + // the shell->client thread is sitting in its select loop. A client + // whose socket has stopped draining (a stalled ssh window, a suspended + // laptop) leaves that thread blocked in write() instead, and an + // unbounded exchange here then parks the global shells lock forever -- + // every list, attach, detach and kill in the daemon wedges behind a + // single unresponsive session. Holding only an Arc keeps the ctl alive + // if the session is removed while we talk to it. + let mut targets = Vec::with_capacity(request.sessions.len()); { let _s = span!(Level::INFO, "lock(shells)").entered(); let shells = self.shells.lock(); for session in request.sessions.into_iter() { if let Some(s) = shells.get(&session) { - let _s = span!(Level::INFO, "lock(shell_to_client_ctl)", s = session).entered(); - let shell_to_client_ctl = s.shell_to_client_ctl.lock(); - shell_to_client_ctl - .client_connection - .send(shell::ClientConnectionMsg::Disconnect) - .context("sending client detach to shell->client")?; - let status = shell_to_client_ctl - .client_connection_ack - .recv() - .context("getting client conn ack")?; - info!("detached session({}), status = {:?}", session, status); - if let shell::ClientConnectionStatus::DetachNone = status { - not_attached_sessions.push(session); - } else { - // The bidi-loop unwind in handle_attach owns the SessionDetached publish; - // we just update the lifecycle state eagerly so a concurrent list() - // reflects the detach immediately. - s.lifecycle.record_detached(); - } + targets.push((session, Arc::clone(&s.shell_to_client_ctl))); } else { not_found_sessions.push(session); } } } + // Both halves are bounded, matching the session-message detach path. + // A session that cannot complete the handshake in time is reported as + // not attached rather than being allowed to stall the daemon. + let mut detached_sessions = vec![]; + for (session, shell_to_client_ctl) in targets.into_iter() { + let _s = span!(Level::INFO, "lock(shell_to_client_ctl)", s = session).entered(); + let shell_to_client_ctl = shell_to_client_ctl.lock(); + if let Err(err) = shell_to_client_ctl + .client_connection + .send_timeout(shell::ClientConnectionMsg::Disconnect, SESSION_MSG_TIMEOUT) + { + error!("sending client detach to shell->client for {}: {:?}", session, err); + not_attached_sessions.push(session); + continue; + } + let status = + match shell_to_client_ctl.client_connection_ack.recv_timeout(SESSION_MSG_TIMEOUT) { + Ok(status) => status, + Err(err) => { + error!("getting client conn ack for {}: {:?}", session, err); + not_attached_sessions.push(session); + continue; + } + }; + info!("detached session({}), status = {:?}", session, status); + if let shell::ClientConnectionStatus::DetachNone = status { + not_attached_sessions.push(session); + } else { + detached_sessions.push(session); + } + } + + // The bidi-loop unwind in handle_attach owns the SessionDetached + // publish; we just update the lifecycle state eagerly so a concurrent + // list() reflects the detach immediately. + if !detached_sessions.is_empty() { + let _s = span!(Level::INFO, "timestamp_lock(shells)").entered(); + let shells = self.shells.lock(); + for session in detached_sessions.iter() { + if let Some(s) = shells.get(session) { + s.lifecycle.record_detached(); + } + } + } + write_reply(&mut stream, DetachReply { not_found_sessions, not_attached_sessions }) .context("writing detach reply")?; diff --git a/shpool/tests/regression.rs b/shpool/tests/regression.rs index e8a8e14d..1b51b668 100644 --- a/shpool/tests/regression.rs +++ b/shpool/tests/regression.rs @@ -334,3 +334,60 @@ fn pager_exit_transitions_to_shell() -> anyhow::Result<()> { Ok(()) } + +/// Regression test for a daemon-wide wedge in the detach handler. The +/// client_connection/client_connection_ack exchange is a rendezvous, so it +/// only completes while the shell->client thread is parked in its select +/// loop. A client whose socket has stopped draining (a stalled ssh window, a +/// suspended laptop) leaves that thread blocked in write() instead, and +/// handle_detach used to run the exchange while still holding the global +/// shells lock -- one unresponsive client wedged every list, attach, detach +/// and kill in the daemon. +/// +/// We stop the attach client with SIGSTOP, flood the session with output +/// until the kernel socket buffers fill and the shell->client thread is stuck +/// in write(), then detach. The daemon must answer the detach (reporting the +/// session rather than hanging) and a follow-up list must come back. +#[test] +#[timeout(30000)] +fn detach_of_stalled_client_does_not_wedge_daemon() -> anyhow::Result<()> { + let mut daemon_proc = support::daemon::Proc::new("norc.toml", DaemonArgs::default()) + .context("starting daemon proc")?; + + let mut attach_proc = + daemon_proc.attach("sh1", Default::default()).context("starting attach proc")?; + daemon_proc.await_event("daemon-bidi-stream-enter")?; + + let mut line_matcher = attach_proc.line_matcher()?; + attach_proc.run_cmd("echo ready")?; + line_matcher.scan_until_re("ready$")?; + + // Ask the shell for far more output than the socket buffers hold, then + // immediately stop the client so nothing drains. + attach_proc.run_cmd("yes | head -c 8000000; echo flood-done")?; + let client_pid = attach_proc.proc.id().to_string(); + let stopped = Command::new("kill") + .args(["-STOP", &client_pid]) + .status() + .context("stopping attach client")?; + assert!(stopped.success(), "SIGSTOP failed"); + + // Give the flood time to fill the kernel buffers behind the stopped + // client so the shell->client thread is genuinely parked in write(). + std::thread::sleep(Duration::from_millis(1500)); + + // On buggy code this call never returns: the rendezvous send blocks + // under the shells lock and the whole daemon wedges behind it. The exit + // status does not matter here -- a stalled client is correctly reported + // as not attached -- only that the daemon answered at all. + let _detach_out = + daemon_proc.detach(vec![String::from("sh1")]).context("detaching stalled client")?; + + // The real assertion: the daemon still answers. + let list_out = daemon_proc.list().context("listing after detach")?; + assert!(list_out.status.success(), "list did not complete"); + + let _ = Command::new("kill").args(["-CONT", &client_pid]).status(); + + Ok(()) +}