diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 66403c5..01b48d2 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -34,7 +34,7 @@ use crate::types::{ use crate::utils::lock::RwLockExt; use std::io::{self, Write}; -use std::net::TcpStream; +use std::net::{TcpStream, ToSocketAddrs}; use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; @@ -46,6 +46,8 @@ use crate::{SoulseekRs, debug, error, trace, warn}; /// rather than undercutting it. const LOGIN_VERDICT_TIMEOUT: Duration = Duration::from_secs(30); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + #[derive(Debug, Clone)] pub struct PeerAddress { host: String, @@ -327,23 +329,25 @@ impl ServerActor { self.session = session; } - fn initiate_connection(&mut self) { - let stream = match TcpStream::connect(( - self.address.host.as_str(), - self.address.port, - )) { - Ok(s) => s, - Err(e) => { - error!("[server] Failed to connect to {}: {}", self.address, e); - self.disconnect_with_error(); - return; - } + fn initiate_connection(&mut self) -> bool { + let stream = (self.address.host.as_str(), self.address.port) + .to_socket_addrs() + .ok() + .and_then(|mut addrs| { + addrs.find_map(|addr| { + TcpStream::connect_timeout(&addr, CONNECT_TIMEOUT).ok() + }) + }); + let Some(stream) = stream else { + error!("[server] Failed to connect to {}", self.address); + self.disconnect_with_error(); + return false; }; if let Err(e) = stream.set_nonblocking(true) { error!("[server] Failed to set non-blocking: {}", e); self.disconnect_with_error(); - return; + return false; } stream.set_nodelay(true).ok(); @@ -351,6 +355,7 @@ impl ServerActor { self.connection_state = ConnectionState::Connecting { since: Instant::now(), }; + true } pub fn set_self_handle(&mut self, handle: ActorHandle) { @@ -420,14 +425,14 @@ impl ServerActor { } fn handle_message(&mut self, msg: ServerMessage) { - if !matches!(self.connection_state, ConnectionState::Connected) { - if matches!(&msg, ServerMessage::ProcessRead) { - // Always process read operations - } else { - // Queue all other messages when not connected - self.queued_messages.push(msg); - return; - } + if !matches!(self.connection_state, ConnectionState::Connected) + && !matches!( + &msg, + ServerMessage::ProcessRead | ServerMessage::Login { .. } + ) + { + self.queued_messages.push(msg); + return; } match msg { @@ -615,12 +620,6 @@ impl ServerActor { } fn handle_login_status(&mut self, message: bool) { - match self.context.write_safe() { - Ok(mut ctx) => ctx.logged_in = Some(message), - Err(e) => { - error!("[server] LoginStatus write: {}", e); - } - } // Send the post-login handshake exactly once, only on success, // on the live path (the old ServerActor::login did this but was // never called). Advertises real shared counts and, when @@ -634,6 +633,13 @@ impl ServerActor { ) { self.send_message(msg); } + self.session.clear(); + } + match self.context.write_safe() { + Ok(mut ctx) => ctx.logged_in = Some(message), + Err(e) => { + error!("[server] LoginStatus write: {}", e); + } } } @@ -695,6 +701,13 @@ impl ServerActor { version: ClientVersion, response: std::sync::mpsc::Sender>, ) { + if self.stream.is_none() && !self.initiate_connection() { + let _ = response.send(Err(SoulseekRs::NotConnected)); + return; + } + if let Ok(mut ctx) = self.context.write_safe() { + ctx.logged_in = None; + } self.queue_message(MessageFactory::build_login_message( &username, &password, version, )); @@ -882,13 +895,10 @@ impl ServerActor { fn disconnect_with_error(&mut self) { debug!("[server] disconnect"); - // Losing an established connection ends the session: nothing reconnects - // it, so anything still waiting on the network is waiting for good. if matches!(self.connection_state, ConnectionState::Connected) { self.session.record(SessionLoss::Disconnected); } - self.connection_state = ConnectionState::Disconnected; - self.stream.take(); + self.disconnect(); } fn disconnect(&mut self) { @@ -896,6 +906,14 @@ impl ServerActor { self.stream.take(); self.connection_state = ConnectionState::Disconnected; + self.dispatcher = None; + self.dispatcher_sender = None; + self.dispatcher_receiver = None; + self.queued_messages.clear(); + self.reader = MessageReader::new(); + if let Ok(mut ctx) = self.context.write_safe() { + ctx.logged_in = None; + } } fn check_connection_status(&mut self) { @@ -952,7 +970,7 @@ impl Actor for ServerActor { fn on_start(&mut self) { if self.stream.is_none() { - self.initiate_connection(); + let _ = self.initiate_connection(); } else { self.connection_state = ConnectionState::Connected; self.on_connection_established(); @@ -989,15 +1007,7 @@ mod tests { #[test] fn a_timed_out_connect_parks_the_actor_in_disconnected() { - let (tx, _rx) = std::sync::mpsc::channel(); - let mut actor = ServerActor::new( - PeerAddress::new("127.0.0.1".to_string(), 1), - tx, - 0, - false, - 0, - 0, - ); + let mut actor = parked_actor(1); actor.connection_state = ConnectionState::Connecting { since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(), }; @@ -1010,6 +1020,168 @@ mod tests { ); } + fn parked_actor(port: u16) -> ServerActor { + let (tx, _rx) = std::sync::mpsc::channel(); + ServerActor::new( + PeerAddress::new("127.0.0.1".to_string(), port), + tx, + 0, + false, + 0, + 0, + ) + } + + #[test] + fn dropping_the_connection_forgets_the_dispatcher_and_the_verdict() { + let mut actor = parked_actor(1); + actor.initialize_dispatcher(); + actor.connection_state = ConnectionState::Connected; + actor.handle_login_status(true); + + actor.disconnect_with_error(); + + assert_eq!(actor.session.loss(), Some(SessionLoss::Disconnected)); + assert!( + actor.dispatcher_sender.is_none(), + "a login queued now must wait in the buffer, not in a channel \ + the next connection replaces" + ); + assert_eq!( + actor.context.read().unwrap().logged_in, + None, + "the old verdict must not answer the next login" + ); + } + + #[test] + fn disconnect_clears_a_partial_frame_so_the_next_session_reframes_clean() { + use std::io::Write; + + 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(); + let partial_frame = [10u8, 0, 0, 0, 1, 2, 3]; + server_side.write_all(&partial_frame).unwrap(); + server_side.flush().unwrap(); + client.set_nonblocking(true).unwrap(); + + let mut actor = parked_actor(addr.port()); + actor.stream = Some(client); + actor.connection_state = ConnectionState::Connected; + for _ in 0..50 { + actor.process_read(); + if actor.reader.buffer_len() > 0 { + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + actor.reader.buffer_len() > 0, + "an incomplete frame should be waiting for its rest" + ); + + actor.disconnect(); + + assert_eq!( + actor.reader.buffer_len(), + 0, + "a stale partial frame carried across a reconnect reads the next \ + session's bytes as a length prefix and wedges recovery" + ); + } + + #[test] + fn a_login_on_a_parked_actor_dials_again_and_queues_the_login() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut actor = parked_actor(listener.local_addr().unwrap().port()); + actor.session.record(SessionLoss::Disconnected); + let (response, _verdict) = std::sync::mpsc::channel(); + + actor.handle_login( + "u".into(), + "p".into(), + ClientVersion::default(), + response, + ); + + assert!( + matches!( + actor.connection_state, + ConnectionState::Connecting { .. } + ), + "a parked actor must dial again on login" + ); + assert_eq!( + actor.queued_messages.len(), + 1, + "the login waits for the connection to come up" + ); + } + + #[test] + fn a_parked_actor_handles_login_rather_than_queueing_it() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut actor = parked_actor(listener.local_addr().unwrap().port()); + actor.session.record(SessionLoss::Disconnected); + let (response, _verdict) = std::sync::mpsc::channel(); + + actor.handle_message(ServerMessage::Login { + username: "u".into(), + password: "p".into(), + version: ClientVersion::default(), + response, + }); + + assert!( + matches!( + actor.connection_state, + ConnectionState::Connecting { .. } + ), + "a parked actor must act on a login, not park it in the queue" + ); + assert!( + actor + .queued_messages + .iter() + .all(|m| !matches!(m, ServerMessage::Login { .. })), + "the login must not be sitting unhandled in the queue" + ); + } + + #[test] + fn a_login_with_nobody_listening_is_refused_at_once() { + let mut actor = parked_actor(1); + let (response, verdict) = std::sync::mpsc::channel(); + + actor.handle_login( + "u".into(), + "p".into(), + ClientVersion::default(), + response, + ); + + assert!( + matches!( + verdict.recv_timeout(Duration::from_secs(5)), + Ok(Err(SoulseekRs::NotConnected)) + ), + "a refused dial must not make the caller wait out the verdict \ + timeout" + ); + } + + #[test] + fn a_successful_login_marks_the_session_live_again() { + let mut actor = parked_actor(1); + actor.session.record(SessionLoss::Disconnected); + + actor.handle_login_status(true); + + assert_eq!(actor.session.loss(), None); + } + #[test] fn post_login_messages_carry_counts_and_conditional_wait_port() { let messages = post_login_messages(true, 4321, 3, 7); diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index be94a22..4410f3a 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -510,4 +510,9 @@ impl SessionWatch { _ => None, } } + + pub fn clear(&self) { + self.0 + .store(Self::LIVE, std::sync::atomic::Ordering::Relaxed); + } } diff --git a/soulseek-rs/docs/openrpc.json b/soulseek-rs/docs/openrpc.json index 83b864e..e28e771 100644 --- a/soulseek-rs/docs/openrpc.json +++ b/soulseek-rs/docs/openrpc.json @@ -2193,7 +2193,7 @@ "type": "null" } }, - "summary": "The server session ended; nothing seen after this means anything.", + "summary": "The server session was lost. A disconnected session is reconnected on its own and events resume; a displaced one is terminal. Read daemon.status for the live state.", "x-notification": true } ], diff --git a/soulseek-rs/src/api.rs b/soulseek-rs/src/api.rs index 2f70241..b9001bd 100644 --- a/soulseek-rs/src/api.rs +++ b/soulseek-rs/src/api.rs @@ -60,6 +60,9 @@ pub trait SessionApi: Send + Sync { } fn listen_port(&self) -> Option; fn session_loss(&self) -> Option; + fn relogin(&self) -> Result { + Err(soulseek_rs::SoulseekRs::NotConnected) + } /// Search and wait out the window. Every implementation is /// `search_with_cancel` with nothing to cancel on. @@ -191,6 +194,10 @@ impl SessionApi for Client { Self::session_loss(self) } + fn relogin(&self) -> Result { + Self::login(self) + } + fn search_with_cancel( &self, query: &str, diff --git a/soulseek-rs/src/daemon/hub.rs b/soulseek-rs/src/daemon/hub.rs index 83dc43b..49a9e12 100644 --- a/soulseek-rs/src/daemon/hub.rs +++ b/soulseek-rs/src/daemon/hub.rs @@ -15,7 +15,7 @@ use super::proto::{ SessionLossEvent, UploadInfoDto, UploadStatusDto, UserMessageDto, }; use crate::api::SessionApi; -use soulseek_rs::DownloadStatus; +use soulseek_rs::{DownloadStatus, SessionLoss}; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel}; @@ -175,9 +175,7 @@ struct DrainState { /// Last upload state reported per (user, file), so an upload that has not /// moved is not re-announced every tick. uploads: HashMap<(String, String), UploadInfoDto>, - /// Whether the session-loss event has already gone out. It is terminal, so - /// it is worth saying exactly once. - reported_loss: bool, + last_loss: Option, } /// How long to keep asking whether a peer has answered a browse request. @@ -273,10 +271,12 @@ fn tick( publish_browses(session, hub, browses); publish_downloads(hub, downloads); - if !state.reported_loss - && let Some(loss) = session.session_loss() - { - state.reported_loss = true; + let loss = session.session_loss(); + if loss == state.last_loss { + return; + } + state.last_loss = loss; + if let Some(loss) = loss { hub.publish( Event::SessionLoss, serde_json::to_value(SessionLossEvent { loss: loss.into() }) diff --git a/soulseek-rs/src/daemon/mod.rs b/soulseek-rs/src/daemon/mod.rs index ad9584e..046da2f 100644 --- a/soulseek-rs/src/daemon/mod.rs +++ b/soulseek-rs/src/daemon/mod.rs @@ -24,11 +24,12 @@ pub mod server; use crate::api::SessionApi; use crate::commands::{Ctx, Session}; -use crate::output::{CliError, CliResult}; +use crate::output::{CliError, CliResult, Out}; use crate::persist::state::{PersistedDownload, PersistedMessage, StateStore}; use hub::{Hub, PendingBrowses, PendingDownload}; use proto::ChatMessageDto; use server::Daemon; +use soulseek_rs::SessionLoss; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; @@ -44,6 +45,8 @@ pub const TOKEN_NAME: &str = "daemon.token"; /// loses little, rare enough not to churn the disk on a busy queue. const SAVE_INTERVAL: Duration = Duration::from_secs(5); +const RECONNECT_BACKOFF_CAP: Duration = Duration::from_mins(1); + #[must_use] pub fn socket_path() -> Option { crate::persist::paths::state_dir().map(|dir| dir.join(SOCKET_NAME)) @@ -216,6 +219,17 @@ pub fn run( }) }; + { + let session = Arc::clone(&session.client); + let daemon = Arc::clone(&daemon); + let stop = Arc::clone(&stop); + let out = ctx.out.clone(); + let address = ctx.settings.server_address.to_string(); + std::thread::spawn(move || { + reconnect_loop(&session, &daemon, &out, &address, &stop); + }); + } + for listener in listeners { let daemon = Arc::clone(&daemon); std::thread::spawn(move || listener.serve(daemon)); @@ -302,6 +316,89 @@ fn cleanup() { } } +const fn should_reconnect(loss: Option, owned: bool) -> bool { + owned && matches!(loss, Some(SessionLoss::Disconnected)) +} + +fn reconnect_loop( + session: &Arc, + daemon: &Daemon, + out: &Out, + address: &str, + stop: &AtomicBool, +) { + let owned = session.daemon_endpoint().is_none(); + let mut failures = 0u32; + let mut announced = false; + while !stop.load(Ordering::Relaxed) { + std::thread::sleep(hub::POLL); + if !should_reconnect(session.session_loss(), owned) { + failures = 0; + announced = false; + continue; + } + if !announced { + out.status(&format!( + "the connection to {address} dropped; logging back in…" + )); + announced = true; + } + match session.relogin() { + Ok(true) => { + out.status("logged back in"); + rejoin(out, session.as_ref(), daemon); + failures = 0; + announced = false; + } + outcome => { + let wait = backoff(failures); + failures = failures.saturating_add(1); + out.warn(&format!( + "cannot log back in: {}; trying again in {}s", + relogin_error(outcome), + wait.as_secs() + )); + wait_or_stop(wait, stop); + } + } + } +} + +fn relogin_error(outcome: soulseek_rs::Result) -> String { + match outcome { + Ok(_) => "login rejected by the server".to_string(), + Err(e) => format!("login failed: {e}"), + } +} + +fn backoff(failures: u32) -> Duration { + Duration::from_secs(1 << failures.min(6)).min(RECONNECT_BACKOFF_CAP) +} + +fn wait_or_stop(total: Duration, stop: &AtomicBool) { + let deadline = Instant::now() + total; + while Instant::now() < deadline && !stop.load(Ordering::Relaxed) { + std::thread::sleep(hub::POLL); + } +} + +fn rejoin(out: &Out, session: &dyn SessionApi, daemon: &Daemon) { + join_rooms(out, session, &daemon.rooms()); + for user in session.watched_users() { + if let Err(e) = session.watch_user(&user) { + out.warn(&format!("cannot watch {user} again: {e}")); + } + } +} + +fn join_rooms(out: &Out, session: &dyn SessionApi, rooms: &[String]) { + for room in rooms { + if let Err(e) = session.join_room(room) { + out.warn(&format!("cannot rejoin {room}: {e}")); + } + } +} + /// Bring back what the last run left: re-queue unfinished transfers, rejoin /// rooms, and reload the conversation. fn restore( @@ -325,11 +422,7 @@ fn restore( ); let rooms = store.load_rooms(); - for room in &rooms { - if let Err(e) = session.join_room(room) { - ctx.out.warn(&format!("cannot rejoin {room}: {e}")); - } - } + join_rooms(&ctx.out, session, &rooms); let unfinished: Vec = store .load_downloads() @@ -425,6 +518,35 @@ mod tests { assert!(first.chars().all(|c| c.is_ascii_hexdigit())); } + #[test] + fn only_an_owned_dropped_session_reconnects() { + assert!(should_reconnect(Some(SessionLoss::Disconnected), true)); + assert!( + !should_reconnect(Some(SessionLoss::Displaced), true), + "reconnecting a displaced name starts a login war with whoever \ + took it" + ); + assert!( + !should_reconnect(None, true), + "a live session is not retried" + ); + assert!( + !should_reconnect(Some(SessionLoss::Disconnected), false), + "a borrowed session is the daemon it came from to mend" + ); + } + + #[test] + fn the_backoff_doubles_and_then_holds_at_the_cap() { + let seconds: Vec = (0..8).map(|n| backoff(n).as_secs()).collect(); + assert_eq!(seconds, [1, 2, 4, 8, 16, 32, 60, 60]); + assert_eq!( + backoff(u32::MAX), + RECONNECT_BACKOFF_CAP, + "a very long outage must not overflow into a tiny wait" + ); + } + #[test] fn two_tokens_are_not_the_same() { let dir = tempfile::tempdir().expect("a temp dir"); diff --git a/soulseek-rs/src/daemon/openrpc.rs b/soulseek-rs/src/daemon/openrpc.rs index dbb5833..f742340 100644 --- a/soulseek-rs/src/daemon/openrpc.rs +++ b/soulseek-rs/src/daemon/openrpc.rs @@ -226,7 +226,9 @@ fn event_summary(event: Event) -> &'static str { Event::DownloadStatus => "A transfer moved.", Event::Browse => "A peer answered a browse request.", Event::SessionLoss => { - "The server session ended; nothing seen after this means anything." + "The server session was lost. A disconnected session is reconnected \ + on its own and events resume; a displaced one is terminal. Read \ + daemon.status for the live state." } } } diff --git a/soulseek-rs/tests/daemon_protocol_e2e.rs b/soulseek-rs/tests/daemon_protocol_e2e.rs index 917f9a1..49b1ec0 100644 --- a/soulseek-rs/tests/daemon_protocol_e2e.rs +++ b/soulseek-rs/tests/daemon_protocol_e2e.rs @@ -29,7 +29,7 @@ struct Daemon { child: Child, state: TempDir, port: u16, - _server: Soulfind, + server: Soulfind, } /// The binary with a scrubbed environment: no test may see the developer's @@ -78,7 +78,7 @@ impl Daemon { child, state, port, - _server: server, + server, }; daemon.wait_until_ready()?; Some(daemon) @@ -130,6 +130,17 @@ impl Daemon { let read = stream.try_clone().expect("a second handle"); Some(Connection::from_parts(Box::new(read), Box::new(stream))) } + + fn status(&self) -> Option { + let mut connection = self.connect_tcp()?; + connection.authenticate(Some(&self.token()))?; + let reply = connection + .call(r#"{"jsonrpc":"2.0","id":1,"method":"daemon.status"}"#)?; + serde_json::from_str::(&reply) + .ok()? + .get_mut("result") + .map(serde_json::Value::take) + } } impl Drop for Daemon { @@ -428,10 +439,15 @@ struct Soulfind { impl Soulfind { fn start() -> Option { - let bin = soulfind_binary()?; let port = free_port()?; let db = std::env::temp_dir().join(format!("soulfind-proto-{port}.db")); let _ = std::fs::remove_file(&db); + let child = Self::spawn(port, &db)?; + Some(Self { child, db, port }) + } + + fn spawn(port: u16, db: &Path) -> Option { + let bin = soulfind_binary()?; let mut child = Command::new(bin) .args(["-d", db.to_str()?, "-p", &port.to_string()]) @@ -443,7 +459,7 @@ impl Soulfind { let deadline = Instant::now() + Duration::from_secs(10); while Instant::now() < deadline { if TcpStream::connect(("127.0.0.1", port)).is_ok() { - return Some(Self { child, db, port }); + return Some(child); } std::thread::sleep(Duration::from_millis(100)); } @@ -455,16 +471,65 @@ impl Soulfind { fn address(&self) -> String { format!("127.0.0.1:{}", self.port) } + + fn stop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + + fn restart(&mut self) -> Option<()> { + self.stop(); + self.child = Self::spawn(self.port, &self.db)?; + Some(()) + } } impl Drop for Soulfind { fn drop(&mut self) { - let _ = self.child.kill(); - let _ = self.child.wait(); + self.stop(); let _ = std::fs::remove_file(&self.db); } } +fn wait_until(timeout: Duration, mut f: impl FnMut() -> bool) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if f() { + return true; + } + std::thread::sleep(Duration::from_millis(250)); + } + false +} + +#[test] +fn the_daemon_logs_back_in_after_the_server_drops_it() { + let mut daemon = daemon_or_skip!(); + assert!( + daemon.status().expect("a status")["session_loss"].is_null(), + "the daemon starts with a live session" + ); + + daemon.server.stop(); + assert!( + wait_until(Duration::from_secs(15), || daemon + .status() + .is_some_and(|status| status["session_loss"] == "disconnected")), + "the daemon never noticed the server going away" + ); + + daemon + .server + .restart() + .expect("soulfind comes back on the same port"); + assert!( + wait_until(Duration::from_secs(45), || daemon + .status() + .is_some_and(|status| status["session_loss"].is_null())), + "the daemon never logged back in once the server was reachable again" + ); +} + /// A client configured for a remote daemon reports the *daemon's* world. /// /// This is the download-box workflow: the token lives in the client's config