From ce244e4e91c619e2c77d213276ac0c64e486283a Mon Sep 17 00:00:00 2001 From: michel Date: Sat, 18 Jul 2026 14:33:39 +0200 Subject: [PATCH] fix: contain panicking jobs so a thread-pool worker is never lost MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A job that panicked unwound out of the worker's run loop and permanently killed that worker; the pool has a fixed worker count and never respawns, so repeated panics (e.g. from malformed network messages) would exhaust the pool and wedge the whole actor system — a DoS. The worker now runs each job under catch_unwind (the shared receiver lock is already released, so a panic cannot poison it) and survives to process the next job. Drop and execute no longer unwrap their channel/join results, so shutdown can never itself panic. --- soulseek-rs-lib/src/utils/thread_pool.rs | 45 +++++++++++++++++++++--- 1 file changed, 41 insertions(+), 4 deletions(-) diff --git a/soulseek-rs-lib/src/utils/thread_pool.rs b/soulseek-rs-lib/src/utils/thread_pool.rs index eafd0a1..371c5bc 100644 --- a/soulseek-rs-lib/src/utils/thread_pool.rs +++ b/soulseek-rs-lib/src/utils/thread_pool.rs @@ -39,7 +39,7 @@ impl ThreadPool { { let job = Box::new(f); if let Some(ref sender) = self.sender { - sender.send(Message::NewJob(job)).unwrap(); + let _ = sender.send(Message::NewJob(job)); } } } @@ -48,18 +48,47 @@ impl Drop for ThreadPool { fn drop(&mut self) { if let Some(sender) = self.sender.take() { for _ in &self.workers { - sender.send(Message::Terminate).unwrap(); + let _ = sender.send(Message::Terminate); } } + // Ignore join errors: a worker whose job panicked returns Err here, but + // Drop must never itself panic (that would abort the process). for worker in &mut self.workers { if let Some(thread) = worker.thread.take() { - thread.join().unwrap(); + let _ = thread.join(); } } } } +#[cfg(test)] +mod tests { + use super::ThreadPool; + use std::sync::mpsc; + use std::time::Duration; + + // A single panicking job must not permanently kill its worker: subsequent + // jobs still need to run. Otherwise one malformed network message could + // leak a pool worker until the whole pool is exhausted (a DoS). + #[test] + fn pool_survives_a_panicking_job() { + let pool = ThreadPool::new(1); + pool.execute(|| panic!("boom")); + + let (tx, rx) = mpsc::channel(); + pool.execute(move || { + let _ = tx.send(42); + }); + + assert_eq!( + rx.recv_timeout(Duration::from_secs(5)).ok(), + Some(42), + "worker died after a panicking job" + ); + } +} + struct Worker { thread: Option>, } @@ -73,7 +102,15 @@ impl Worker { Err(_) => break, }; match message { - Ok(Message::NewJob(job)) => job(), + // Contain a panicking job so it kills only that job, not the + // worker: the pool has a fixed number of workers and a lost + // one is never replaced. The lock is already released here, + // so a panic cannot poison the shared receiver. + Ok(Message::NewJob(job)) => { + let _ = std::panic::catch_unwind( + std::panic::AssertUnwindSafe(job), + ); + } Ok(Message::Terminate) | Err(_) => break, } }