From c1a80b0d302d4e79064bda369eac8cb6f3e91071 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 14 Sep 2026 12:07:14 +0200 Subject: [PATCH] feat(lib): let another thread cancel a client's connect and login A caller that gives up on a login had no way to reach the client: the thread that owns it sits in login() for up to 45 seconds, and disconnect() needs the client to itself. Until that thread returned, the abandoned client kept connecting and still wrote its login frame the moment the server finished the TCP handshake. When that happened after a replacement had logged in under the same name, the server kicked the replacement. Client::cancel_handle() hands out a CancelHandle that can be moved to any thread. After cancel(): - the server actor writes nothing more. The check and the write share a lock with cancel(), so once cancel() returns no frame leaves, whether it was waiting in the queue, in the dispatcher channel, or mid-flush; - login() returns NotConnected within 100 ms, and hands no login to the actor at all when the client was already cancelled; - connect() refuses before binding the listener when the client was already cancelled. A cancel that lands during that check can still let the bind through. Either way the owner drops the client as before, which disconnects it and gives its port back. A frame already written before cancel() returned cannot be recalled. The API is additive, and login() on a client nobody cancels behaves as before. Soulseek.NET tears the whole connection down when ConnectAsync is cancelled, and Nicotine+ drops outgoing messages for a socket it closed; this gives soulseek-rs the same end state from another thread. --- soulseek-rs-lib/src/actor/server_actor.rs | 12 +++- .../src/actor/server_actor/tests.rs | 38 ++++++++++ soulseek-rs-lib/src/client/cancel.rs | 24 +++++++ soulseek-rs-lib/src/client/connection.rs | 27 +++++-- soulseek-rs-lib/src/client/mod.rs | 10 +++ soulseek-rs-lib/src/lib.rs | 2 +- soulseek-rs-lib/tests/e2e.rs | 71 +++++++++++++++++++ 7 files changed, 178 insertions(+), 6 deletions(-) create mode 100644 soulseek-rs-lib/src/client/cancel.rs diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 150de3e..658387f 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -1,5 +1,5 @@ use crate::actor::{Actor, ActorHandle, ConnectionState}; -use crate::client::ClientOperation; +use crate::client::{CancelHandle, ClientOperation}; use crate::dispatcher::MessageDispatcher; use crate::message::server::AdminMessageHandler; use crate::message::server::CantConnectToPeerHandler; @@ -288,6 +288,7 @@ pub struct ServerActor { shared_folder_count: u32, shared_file_count: u32, session: SessionWatch, + cancel: CancelHandle, } /// The messages a client sends right after a successful login: its shared-file @@ -340,6 +341,7 @@ impl ServerActor { shared_folder_count, shared_file_count, session: SessionWatch::default(), + cancel: CancelHandle::new(), } } @@ -348,6 +350,10 @@ impl ServerActor { self.session = session; } + pub fn set_cancel(&mut self, cancel: CancelHandle) { + self.cancel = cancel; + } + fn initiate_connection(&mut self) -> bool { let stream = (self.address.get_host(), self.address.get_port()) .to_socket_addrs() @@ -563,6 +569,10 @@ impl ServerActor { } fn send_message(&mut self, message: Message) { + let cancel = self.cancel.clone(); + let Some(_live) = cancel.live() else { + return; + }; let Some(stream) = self.stream.as_mut() else { error!("[server] Cannot send message: stream is None"); return; diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs index 0809d99..064ba7e 100644 --- a/soulseek-rs-lib/src/actor/server_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -188,6 +188,44 @@ fn a_login_with_nobody_listening_is_refused_at_once() { ); } +#[test] +fn a_cancelled_actor_never_writes_the_login_it_queued() { + use std::io::Read; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let client = TcpStream::connect(addr).unwrap(); + let (mut server_side, _) = listener.accept().unwrap(); + client.set_nonblocking(true).unwrap(); + + let cancel = CancelHandle::new(); + let mut actor = parked_actor(addr.port()); + actor.set_cancel(cancel.clone()); + actor.stream = Some(client); + actor.connection_state = ConnectionState::Connecting { + since: Instant::now(), + }; + let (response, _verdict) = std::sync::mpsc::channel(); + actor.handle_login( + "u".into(), + "p".into(), + ClientVersion::default(), + response, + ); + assert_eq!(actor.queued_messages.len(), 1); + + cancel.cancel(); + actor.tick(); + assert!(matches!(actor.connection_state, ConnectionState::Connected)); + actor.on_stop(); + + server_side + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut written = Vec::new(); + assert_eq!(server_side.read_to_end(&mut written).unwrap(), 0); +} + #[test] fn a_successful_login_marks_the_session_live_again() { let mut actor = parked_actor(1); diff --git a/soulseek-rs-lib/src/client/cancel.rs b/soulseek-rs-lib/src/client/cancel.rs new file mode 100644 index 0000000..ba8ebbe --- /dev/null +++ b/soulseek-rs-lib/src/client/cancel.rs @@ -0,0 +1,24 @@ +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; + +#[derive(Clone)] +pub struct CancelHandle(Arc>); + +impl CancelHandle { + pub(crate) fn new() -> Self { + Self(Arc::new(Mutex::new(false))) + } + + pub fn cancel(&self) { + *self.0.lock().unwrap_or_else(PoisonError::into_inner) = true; + } + + #[must_use] + pub fn is_cancelled(&self) -> bool { + *self.0.lock().unwrap_or_else(PoisonError::into_inner) + } + + pub(crate) fn live(&self) -> Option> { + let guard = self.0.lock().unwrap_or_else(PoisonError::into_inner); + (!*guard).then_some(guard) + } +} diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index 56103ec..cf0bb09 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -5,9 +5,11 @@ use super::{ TcpStream, error, info, mpsc, thread, trace, warn, }; use std::sync::atomic::Ordering; +use std::sync::mpsc::RecvTimeoutError; use std::time::{Duration, Instant}; const LISTENER_RELEASE_TIMEOUT: Duration = Duration::from_secs(2); +const LOGIN_POLL_INTERVAL: Duration = Duration::from_millis(100); /// Ceiling on the wait for a login verdict. Generous enough for a slow server /// and a retrying connect, short enough that an unattended caller ends. @@ -90,6 +92,10 @@ impl Client { ctx.shares = shares; ctx.shared_directories.clone_from(&self.shared_directories); + if self.cancel.is_cancelled() { + return Err(SoulseekRs::NotConnected); + } + // Bind before logging in: the port we advertise has to be the port we // hold, and a bind that fails outright is the caller's to see rather // than a panic on a thread nobody is watching. @@ -111,6 +117,7 @@ impl Client { shared_file_count, ); server_actor.set_session_watch(self.session.clone()); + server_actor.set_cancel(self.cancel.clone()); self.server_handle = Some(ctx.actor_system.spawn_with_handle( server_actor, @@ -202,7 +209,9 @@ impl Client { /// control. pub fn login(&self) -> Result { info!("Logging in as {}", self.username); - if let Some(handle) = &self.server_handle { + if let Some(handle) = &self.server_handle + && !self.cancel.is_cancelled() + { let (tx, rx) = std::sync::mpsc::channel(); let _ = handle.send(ServerMessage::Login { username: self.username.clone(), @@ -211,9 +220,19 @@ impl Client { response: tx, }); - match rx.recv_timeout(LOGIN_RESPONSE_TIMEOUT) { - Ok(result) => result, - Err(_) => Err(SoulseekRs::Timeout), + let deadline = Instant::now() + LOGIN_RESPONSE_TIMEOUT; + loop { + match rx.recv_timeout(LOGIN_POLL_INTERVAL) { + Ok(result) => return result, + Err(RecvTimeoutError::Timeout) + if self.cancel.is_cancelled() => + { + return Err(SoulseekRs::NotConnected); + } + Err(RecvTimeoutError::Timeout) + if Instant::now() < deadline => {} + Err(_) => return Err(SoulseekRs::Timeout), + } } } else { Err(SoulseekRs::NotConnected) diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index eefc812..6ce58e5 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -649,6 +649,7 @@ pub struct Client { /// port it bound. listener_stopped: Arc, listener_released: Arc, + cancel: CancelHandle, } impl Drop for Client { @@ -684,6 +685,7 @@ impl Client { session: SessionWatch::default(), listener_stopped: Arc::new(AtomicBool::new(false)), listener_released: Arc::new(AtomicBool::new(false)), + cancel: CancelHandle::new(), } } @@ -724,6 +726,11 @@ impl Client { self.session.loss() } + #[must_use] + pub fn cancel_handle(&self) -> CancelHandle { + self.cancel.clone() + } + /// The directories whose files are currently shared with other peers. #[must_use] pub fn shared_directories(&self) -> Vec { @@ -849,6 +856,7 @@ impl Client { } } +mod cancel; mod children; mod connection; mod context; @@ -861,5 +869,7 @@ mod social; mod upload_queue; mod uploads; +pub use cancel::CancelHandle; + #[cfg(test)] mod tests; diff --git a/soulseek-rs-lib/src/lib.rs b/soulseek-rs-lib/src/lib.rs index a174fbd..45d3263 100644 --- a/soulseek-rs-lib/src/lib.rs +++ b/soulseek-rs-lib/src/lib.rs @@ -32,7 +32,7 @@ pub mod prelude { // Re-export commonly used types pub use actor::server_actor::{PeerAddress, UserMessage}; -pub use client::{Client, ClientSettings}; +pub use client::{CancelHandle, Client, ClientSettings}; pub use error::{Result, SoulseekRs}; pub use message::peer::{SharedDirectory, SharedFileEntry}; pub use peer::{ConnectionType, ParseConnectionTypeError}; diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 01c905b..0e45c9f 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -5621,3 +5621,74 @@ fn a_dropped_client_has_released_its_listener_port_when_drop_returns() { ); } } + +#[test] +fn a_cancelled_client_never_binds_its_port() { + let _gate = SERVER_GATE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let port = free_port().expect("a port to hold"); + let mut client = Client::with_settings(ClientSettings { + username: "e2e_cancelled_bind".to_string(), + password: "pw".to_string(), + server_address: PeerAddress::new("127.0.0.1".to_string(), 1), + enable_listen: true, + listen_port: port, + shared_directories: Vec::new(), + accept_children: false, + version: ClientVersion::default(), + }); + + client.cancel_handle().cancel(); + + assert!(client.connect().is_err()); + assert_eq!(client.listen_port(), None); + assert!(std::net::TcpListener::bind(("0.0.0.0", port)).is_ok()); +} + +#[test] +fn a_cancel_ends_a_login_waiting_for_its_verdict_and_nothing_follows_it() { + let server = + std::net::TcpListener::bind("127.0.0.1:0").expect("a silent server"); + let mut client = Client::with_settings(ClientSettings { + username: "e2e_cancelled_login".to_string(), + password: "pw".to_string(), + server_address: PeerAddress::new( + "127.0.0.1".to_string(), + server.local_addr().expect("server address").port(), + ), + enable_listen: false, + listen_port: 0, + shared_directories: Vec::new(), + accept_children: false, + version: ClientVersion::default(), + }); + client.connect().expect("start the session"); + let (mut conn, _) = server.accept().expect("the client dials the server"); + conn.set_read_timeout(Some(Duration::from_secs(5))) + .expect("read timeout"); + let cancel = client.cancel_handle(); + + std::thread::scope(|scope| { + let login = scope.spawn(|| client.login()); + let frame = + read_framed(&mut conn).expect("the login reaches the server"); + assert_eq!(frame.get_message_code(), 1); + + let cancelled_at = Instant::now(); + cancel.cancel(); + assert!(login.join().expect("the login thread").is_err()); + assert!(cancelled_at.elapsed() < Duration::from_secs(5)); + }); + client + .connect_peer("e2e_nobody") + .expect("the actor takes a request"); + drop(client); + + let mut rest = Vec::new(); + assert_eq!( + conn.read_to_end(&mut rest) + .expect("the server sees the session end"), + 0 + ); +}