diff --git a/rust/core/src/clients/irc.rs b/rust/core/src/clients/irc.rs index 7d6d15b..b1e0fc3 100644 --- a/rust/core/src/clients/irc.rs +++ b/rust/core/src/clients/irc.rs @@ -211,7 +211,7 @@ impl Client { *self.out.lock() = Some(out_tx); let client = self.clone(); - std::thread::spawn(move || client.io_thread(out_rx)); + std::thread::spawn(super::guarded(self.cfg.on_error.clone(), self.cfg.on_close.clone(), move || client.io_thread(out_rx))); let timeout = self.cfg.connect_timeout; match ready_rx.recv_timeout(timeout) { diff --git a/rust/core/src/clients/local.rs b/rust/core/src/clients/local.rs index 5e6f401..55c4d0c 100644 --- a/rust/core/src/clients/local.rs +++ b/rust/core/src/clients/local.rs @@ -76,7 +76,8 @@ impl LocalShell { drop(st); let me = self.clone(); - std::thread::Builder::new().name("nexus-local-read".into()).spawn(move || { + let (on_error, on_close) = (self.cb.on_error.clone(), self.cb.on_close.clone()); + std::thread::Builder::new().name("nexus-local-read".into()).spawn(super::guarded(on_error, on_close, move || { let mut buf = vec![0u8; 32 * 1024]; loop { match reader.read(&mut buf) { @@ -89,7 +90,7 @@ impl LocalShell { } } } - })?; + }))?; let me = self.clone(); std::thread::Builder::new().name("nexus-local-wait".into()).spawn(move || { let _ = child.wait(); diff --git a/rust/core/src/clients/mod.rs b/rust/core/src/clients/mod.rs index 32a40a3..ab04765 100644 --- a/rust/core/src/clients/mod.rs +++ b/rust/core/src/clients/mod.rs @@ -44,6 +44,38 @@ pub mod serial; pub mod telnet; pub mod vnc; +/// Wraps a client's I/O loop so a panic in it (a decoder bug, say) ends only that session — +/// reported through `on_error`, then `on_close` — instead of leaving it looking connected with +/// nothing reading. +pub(crate) fn guarded(on_error: ErrorFn, on_close: CloseFn, f: impl FnOnce() + Send + 'static) -> impl FnOnce() + Send + 'static { + move || { + if let Err(p) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f)) { + on_error(format!("internal error: {}", panic_message(&*p))); + on_close(); + } + } +} + +/// `guarded` for a tokio task: the panic surfaces when the task's handle is awaited. +pub(crate) fn guarded_task(on_error: ErrorFn, on_close: CloseFn, fut: F) +where + F: std::future::Future + Send + 'static, +{ + let h = tokio::spawn(fut); + tokio::spawn(async move { + if let Err(e) = h.await + && e.is_panic() + { + on_error(format!("internal error: {}", panic_message(&*e.into_panic()))); + on_close(); + } + }); +} + +pub(crate) fn panic_message(p: &(dyn std::any::Any + Send)) -> String { + p.downcast_ref::<&str>().map(|s| s.to_string()).or_else(|| p.downcast_ref::().cloned()).unwrap_or_else(|| "unknown panic".into()) +} + /// Called with protocol negotiation debug messages (Go's `onDebug`). pub type DebugFn = Arc; @@ -91,7 +123,7 @@ impl TcpConn { *guard = Some(stream); } let me = self.clone(); - std::thread::spawn(move || { + std::thread::spawn(guarded(self.on_error.clone(), self.on_close.clone(), move || { use std::io::Read; let mut buf = [0u8; 4096]; loop { @@ -111,7 +143,7 @@ impl TcpConn { } return; } - }); + })); Ok(()) } @@ -151,3 +183,33 @@ pub(crate) fn utf8_to_latin1(data: &[u8]) -> Vec { pub(crate) fn latin1_to_utf8(data: &[u8]) -> String { data.iter().map(|&b| b as char).collect() } + +#[cfg(test)] +mod guard_tests { + use super::*; + use parking_lot::Mutex; + + #[test] + fn a_panicking_io_loop_reports_and_closes_once() { + let log: Arc>> = Arc::default(); + let (l1, l2) = (log.clone(), log.clone()); + let on_error: ErrorFn = Arc::new(move |e| l1.lock().push(e)); + let on_close: CloseFn = Arc::new(move || l2.lock().push("closed".into())); + std::thread::spawn(guarded(on_error.clone(), on_close.clone(), || panic!("bad tile"))).join().unwrap(); + assert_eq!(*log.lock(), ["internal error: bad tile", "closed"]); + // A loop that ends normally is left to report its own close. + std::thread::spawn(guarded(on_error.clone(), on_close.clone(), || {})).join().unwrap(); + assert_eq!(log.lock().len(), 2); + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.block_on(async { + guarded_task(on_error, on_close, async { panic!("{}", "task boom") }); + for _ in 0..50 { + if log.lock().len() == 4 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }); + assert_eq!(log.lock()[2..], ["internal error: task boom", "closed"]); + } +} diff --git a/rust/core/src/clients/rdp.rs b/rust/core/src/clients/rdp.rs index 0f5098f..d56aaff 100644 --- a/rust/core/src/clients/rdp.rs +++ b/rust/core/src/clients/rdp.rs @@ -82,7 +82,8 @@ impl Client { let (tx, rx) = mpsc::channel(); *self.tx.lock() = Some(tx); let me = self.clone(); - std::thread::Builder::new().name("nexus-rdp".into()).spawn(move || { + let (on_error, on_close) = (self.cfg.on_error.clone(), self.cfg.on_close.clone()); + std::thread::Builder::new().name("nexus-rdp".into()).spawn(super::guarded(on_error, on_close, move || { let r = me.run(framed, result, rx); let closing = me.tx.lock().take().is_none(); if let Err(e) = r @@ -91,7 +92,7 @@ impl Client { (me.cfg.on_error)(e.to_string()); } (me.cfg.on_close)(); - })?; + }))?; Ok(()) } diff --git a/rust/core/src/clients/serial.rs b/rust/core/src/clients/serial.rs index 14ff7b2..4773e3d 100644 --- a/rust/core/src/clients/serial.rs +++ b/rust/core/src/clients/serial.rs @@ -90,7 +90,7 @@ impl SerialClient { *guard = Some((port, stop.clone())); } let me = inner.clone(); - std::thread::spawn(move || me.read_loop(reader, stop)); + std::thread::spawn(super::guarded(inner.cb.on_error.clone(), inner.cb.on_close.clone(), move || me.read_loop(reader, stop))); Ok(()) } diff --git a/rust/core/src/clients/ssh.rs b/rust/core/src/clients/ssh.rs index e0b6844..e24d105 100644 --- a/rust/core/src/clients/ssh.rs +++ b/rust/core/src/clients/ssh.rs @@ -502,7 +502,7 @@ impl SshClient { } }); let me = self.clone(); - tokio::spawn(async move { + super::guarded_task(self.cb.on_error.clone(), self.cb.on_close.clone(), async move { while let Some(msg) = read.wait().await { match msg { ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, .. } => (me.cb.on_data)(&data), diff --git a/rust/core/src/clients/vnc.rs b/rust/core/src/clients/vnc.rs index 0d11a4b..7f194bf 100644 --- a/rust/core/src/clients/vnc.rs +++ b/rust/core/src/clients/vnc.rs @@ -164,7 +164,7 @@ impl Client { *self.out.lock() = Some(out_tx); let client = self.clone(); - std::thread::spawn(move || client.io_thread(out_rx)); + std::thread::spawn(super::guarded(self.cfg.on_error.clone(), self.cfg.on_close.clone(), move || client.io_thread(out_rx))); let timeout = self.cfg.connect_timeout; match ready_rx.recv_timeout(timeout) {