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 + ); +}