diff --git a/README.md b/README.md index 888704a..40978b6 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,7 @@ The one-shot commands follow three rules, so they compose with other tools: | 4 | the command worked but found nothing | | 5 | timed out waiting for a response or a transfer | | 6 | a transfer started but did not finish | +| 7 | the session ended mid-command; nothing it saw is reliable | #### Commands @@ -464,6 +465,10 @@ forward the port yourself. `--listener-port` you chose) to this machine on your router. - Pass `--no-listener` to turn the listener (and port mapping) off, or `--listener` to force it on when the config file disables it. +- If another process on this machine already holds that port — the usual case + when several one-shot commands run at once — the client takes one the OS + picks instead, says so on stderr, and advertises and maps that one. `whoami` + reports the port it really holds. Check whether it works on your network without launching the whole client: diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index b76d732..5c5d393 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -481,6 +481,25 @@ impl PeerActor { ); } } + // A peer hanging up a control connection it had already + // established is ordinary: other clients close idle peer + // sockets while a transfer runs on its own connection. Report + // the clean disconnect it is — an error here fails every + // download queued with that peer, the one streaming included. + // A connection that never established is the other story: the + // peer accepted and vanished, which is what the server-brokered + // fallback exists for, so that one keeps the error path. + Err(ref e) + if e.kind() == io::ErrorKind::UnexpectedEof + && self.established => + { + debug!( + "[peer:{}] peer closed the connection", + self.peer_username() + ); + self.disconnect(); + return; + } Err(e) => { let username = self.peer_username(); error!( diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index cfa239e..f5b27c4 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -14,6 +14,7 @@ use crate::message::server::MessageUser; use crate::message::server::ParentMinSpeedHandler; use crate::message::server::ParentSpeedRatioHandler; use crate::message::server::PrivilegedUsersHandler; +use crate::message::server::ReloggedHandler; use crate::message::server::SayChatroomHandler; use crate::message::server::UserJoinedRoomHandler; use crate::message::server::UserLeftRoomHandler; @@ -25,7 +26,9 @@ use crate::message::{Handlers, MessageType}; use crate::message::{Message, MessageReader}; use crate::peer::ConnectionType; use crate::peer::Peer; -use crate::types::{ClientVersion, RoomEvent, RoomInfo}; +use crate::types::{ + ClientVersion, RoomEvent, RoomInfo, SessionLoss, SessionWatch, +}; use crate::utils::lock::RwLockExt; use std::io::{self, Error, Write}; @@ -156,6 +159,9 @@ impl UserMessage { pub enum ServerMessage { ProcessRead, LoginStatus(bool), + /// The server is closing this connection: the same username logged in + /// elsewhere. + Relogged, SendMessage(Message), Login { username: String, @@ -252,6 +258,7 @@ pub struct ServerActor { queued_messages: Vec, shared_folder_count: u32, shared_file_count: u32, + session: SessionWatch, } /// The messages a client sends right after a successful login: its shared-file @@ -304,9 +311,15 @@ impl ServerActor { queued_messages: Vec::new(), shared_folder_count, shared_file_count, + session: SessionWatch::default(), } } + /// Share the client's view of whether this session is still alive. + pub fn set_session_watch(&mut self, session: SessionWatch) { + self.session = session; + } + #[must_use] pub const fn get_address(&self) -> &PeerAddress { &self.address @@ -392,6 +405,7 @@ impl ServerActor { let mut handlers = Handlers::new(); handlers.register_handler(LoginHandler); + handlers.register_handler(ReloggedHandler); handlers.register_handler(RoomListHandler); handlers.register_handler(GetUserStatusHandler); handlers.register_handler(GetUserStatsHandler); @@ -459,6 +473,7 @@ impl ServerActor { ServerMessage::LoginStatus(message) => { self.handle_login_status(message); } + ServerMessage::Relogged => self.handle_relogged(), ServerMessage::PierceFirewall(token) => { self.send_message( MessageFactory::build_pierce_firewall_message(token), @@ -866,9 +881,27 @@ impl ServerActor { } } + /// The server is closing this connection because the same account logged + /// in elsewhere. Nothing reconnects it, so the session is over. + fn handle_relogged(&mut self) { + error!( + "[server] another login took this username; this session has been \ + closed by the server" + ); + self.session.record(SessionLoss::Displaced); + self.connection_state = ConnectionState::Disconnected; + self.disconnect(); + } + fn disconnect_with_error(&mut self, _error: Error) { 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(); } diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index e4ade93..2aa7d15 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -53,14 +53,26 @@ impl Client { ctx.shares = shares; ctx.shared_directories.clone_from(&self.shared_directories); - let server_actor = ServerActor::new( + // 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. + let listener = if self.enable_listen { + let listener = Listen::bind(self.listen_port)?; + self.bound_port = Some(listener.local_addr()?.port()); + Some(listener) + } else { + None + }; + + let mut server_actor = ServerActor::new( self.address.clone(), sender, - self.listen_port, + self.bound_port.unwrap_or(self.listen_port), self.enable_listen, shared_folder_count, shared_file_count, ); + server_actor.set_session_watch(self.session.clone()); self.server_handle = Some(ctx.actor_system.spawn_with_handle( server_actor, @@ -69,19 +81,13 @@ impl Client { }, )); - if self.enable_listen { - let listen_port = self.listen_port; + if let Some(listener) = listener { let client_sender = listen_sender; let context = self.context.clone(); let own_username = self.username.clone(); thread::spawn(move || { - Listen::start( - listen_port, - client_sender, - context, - own_username, - ); + Listen::serve(&listener, client_sender, context, own_username); }); } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index e2b1e5a..094557c 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -5,7 +5,7 @@ use crate::actor::server_actor::{ use crate::download_store::{DownloadStore, collect_failed_tokens}; use crate::types::{ ClientVersion, DownloadMetadata, DownloadStatus, RoomEvent, RoomInfo, - UserInfo, UserPresence, UserStats, UserStatus, + SessionLoss, SessionWatch, UserInfo, UserPresence, UserStats, UserStatus, }; use crate::utils::logger; use crate::{ @@ -811,6 +811,9 @@ impl ClientContext { pub struct Client { enable_listen: bool, listen_port: u16, + /// The port the listener actually holds, known once [`Client::connect`] + /// has bound it. + bound_port: Option, address: PeerAddress, username: String, password: String, @@ -818,6 +821,7 @@ pub struct Client { shared_directories: Vec, server_handle: Option>, context: Arc>, + session: SessionWatch, } impl Client { @@ -834,6 +838,7 @@ impl Client { Self { enable_listen: settings.enable_listen, listen_port: settings.listen_port, + bound_port: None, address: settings.server_address, username: settings.username, password: settings.password, @@ -841,6 +846,7 @@ impl Client { shared_directories: settings.shared_directories, context: Arc::new(RwLock::new(ClientContext::new())), server_handle: None, + session: SessionWatch::default(), } } @@ -850,6 +856,23 @@ impl Client { &self.username } + /// The port peers can reach this client on, or `None` when it is not + /// listening (or has not connected yet). This is the port that was really + /// bound, which is not always the one that was configured. + #[must_use] + pub const fn listen_port(&self) -> Option { + self.bound_port + } + + /// Why the server session ended, or `None` while it is alive. + /// + /// A lost session sees nothing on the network, so an empty result set from + /// one says nothing about what the network holds. + #[must_use] + pub fn session_loss(&self) -> Option { + self.session.loss() + } + /// The directories whose files are currently shared with other peers. #[must_use] pub fn shared_directories(&self) -> Vec { diff --git a/soulseek-rs-lib/src/lib.rs b/soulseek-rs-lib/src/lib.rs index 010a577..518c9d7 100644 --- a/soulseek-rs-lib/src/lib.rs +++ b/soulseek-rs-lib/src/lib.rs @@ -33,6 +33,6 @@ pub use error::{Result, SoulseekRs}; pub use message::peer::SharedDirectory; pub use types::{ ClientVersion, DownloadStatus, File, RoomEvent, RoomInfo, Search, - SearchResult, Transfer, UploadInfo, UploadStatus, UserInfo, UserPresence, - UserStats, UserStatus, + SearchResult, SessionLoss, Transfer, UploadInfo, UploadStatus, UserInfo, + UserPresence, UserStats, UserStatus, }; diff --git a/soulseek-rs-lib/src/message/message_reader.rs b/soulseek-rs-lib/src/message/message_reader.rs index b4352fd..33ccb07 100644 --- a/soulseek-rs-lib/src/message/message_reader.rs +++ b/soulseek-rs-lib/src/message/message_reader.rs @@ -39,8 +39,14 @@ impl MessageReader { ) -> io::Result<()> { let mut temp_buffer = [0; 1024]; // Temporary buffer for reading from the socket let bytes_read = stream.read(&mut temp_buffer)?; + // On a TCP socket a zero-byte read is the other end hanging up. Passing + // that back as success left the caller polling a dead socket forever, + // which is how a server that closes cleanly went unnoticed. if bytes_read == 0 { - return Ok(()); + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "the remote end closed the connection", + )); } // Add the read bytes to the internal buffer @@ -121,4 +127,32 @@ mod tests { assert!(buffered_reader.buffer.is_empty()); assert_eq!(vec![1, 2, 3], rest); } + + /// A hangup has to reach the caller as an error. Reported as success it + /// leaves the actors polling a socket that will never speak again, which is + /// how a server that closed cleanly went unnoticed. + #[test] + fn a_closed_connection_reads_as_an_error_not_as_success() { + use std::io::Write; + use std::net::{TcpListener, TcpStream}; + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let mut ours = TcpStream::connect(addr).unwrap(); + let (mut theirs, _) = listener.accept().unwrap(); + + theirs.write_all(&[1, 2, 3]).unwrap(); + theirs.flush().unwrap(); + drop(theirs); + + let mut reader = MessageReader::new(); + // Whatever was sent before the close still arrives. + while reader.buffer_len() < 3 { + reader.read_from_socket(&mut ours).unwrap(); + } + let error = reader + .read_from_socket(&mut ours) + .expect_err("a closed connection is not a successful read"); + assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof); + } } diff --git a/soulseek-rs-lib/src/message/server/mod.rs b/soulseek-rs-lib/src/message/server/mod.rs index f3fa4bc..104082b 100644 --- a/soulseek-rs-lib/src/message/server/mod.rs +++ b/soulseek-rs-lib/src/message/server/mod.rs @@ -10,6 +10,7 @@ mod message_user; mod parent_min_speed; mod parent_speed_ratio; mod privileged_users; +mod relogged; mod room_list; mod say_chatroom; mod user_info; @@ -29,6 +30,7 @@ pub use message_user::MessageUser; pub use parent_min_speed::ParentMinSpeedHandler; pub use parent_speed_ratio::ParentSpeedRatioHandler; pub use privileged_users::{CheckPrivilegesHandler, PrivilegedUsersHandler}; +pub use relogged::ReloggedHandler; pub use room_list::RoomListHandler; pub use say_chatroom::SayChatroomHandler; pub use user_info::{GetUserStatsHandler, GetUserStatusHandler}; diff --git a/soulseek-rs-lib/src/message/server/relogged.rs b/soulseek-rs-lib/src/message/server/relogged.rs new file mode 100644 index 0000000..84ce7a7 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/relogged.rs @@ -0,0 +1,18 @@ +use crate::actor::server_actor::ServerMessage; +use crate::message::{Message, MessageHandler}; +use std::sync::mpsc::Sender; + +/// `Relogged` (code 41): the server is about to cut this connection because +/// the same username logged in somewhere else. +pub struct ReloggedHandler; + +impl MessageHandler for ReloggedHandler { + fn get_code(&self) -> u8 { + 41 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let _ = message; + let _ = sender.send(ServerMessage::Relogged); + } +} diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index ed1e855..72d5e1b 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -36,20 +36,11 @@ fn read_peer_init_message( reader: &mut MessageReader, ) -> io::Result { // An untrusted peer gets a bounded handshake. Without this a peer that - // connects and stays silent parks this read forever, and one that hangs up - // spins it (a zero-byte read is EOF, which `read_from_socket` reports as - // success) — either way pinning a thread that owes us a peer init. + // connects and stays silent parks this read forever, pinning a thread that + // owes us a peer init. stream.set_read_timeout(Some(PEER_INIT_TIMEOUT))?; let message = loop { - let buffered = reader.buffer_len(); reader.read_from_socket(stream)?; - if reader.buffer_len() == buffered { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "peer closed before sending a peer init", - )); - } - if let Ok(Some(msg)) = reader.extract_message() { break msg; } @@ -374,16 +365,34 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { pub struct Listen {} impl Listen { - pub fn start( - port: u16, + /// Take the peer-listening socket, falling back to a port the operating + /// system picks when the configured one is already held. + /// + /// Several clients on one machine share one configured port, and losing + /// the race for it must not cost a session its listener: an unreachable + /// client is one that never receives search responses or transfers. The + /// caller advertises [`TcpListener::local_addr`], so peers are always told + /// the port that was really bound. + pub fn bind(port: u16) -> io::Result { + match TcpListener::bind(("0.0.0.0", port)) { + Err(e) if e.kind() == io::ErrorKind::AddrInUse && port != 0 => { + info!( + "[listener] port {port} is taken, falling back to an \ + ephemeral port" + ); + TcpListener::bind(("0.0.0.0", 0)) + } + other => other, + } + } + + pub fn serve( + listener: &TcpListener, client_sender: Sender, client_context: Arc>, own_username: String, ) { - info!("[listener] starting listener on port {port}"); - - let listener = TcpListener::bind(format!("0.0.0.0:{port}")) - .expect("Failed to bind listener to port"); + info!("[listener] listening on {:?}", listener.local_addr()); let context = ConnectionContext { client_sender, diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index f133b15..f873981 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -399,3 +399,62 @@ pub struct UploadInfo { /// download does. Zero unless the upload is in progress. pub speed_bytes_per_sec: f64, } + +/// Why a server session ended before the client was done with it. +/// +/// A session that ends stops seeing the network entirely, which is not the +/// same thing as the network having nothing to show — telling the two apart is +/// the difference between "retry" and "give up". +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum SessionLoss { + /// Another login claimed this username. The server keeps one session per + /// name, so the older one is cut off. + Displaced = 1, + /// The connection to the server dropped. + Disconnected = 2, +} + +impl std::fmt::Display for SessionLoss { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Displaced => f.write_str( + "another login took over this username; the server allows one \ + session per account", + ), + Self::Disconnected => { + f.write_str("the connection to the server dropped") + } + } + } +} + +/// A lock-free view, shared with the server actor, of whether the session is +/// still alive. +/// +/// The first loss recorded wins: a displaced session also drops its socket a +/// moment later, and "displaced" is the reason worth reporting. +#[derive(Debug, Clone, Default)] +pub struct SessionWatch(std::sync::Arc); + +impl SessionWatch { + const LIVE: u8 = 0; + + pub fn record(&self, loss: SessionLoss) { + let _ = self.0.compare_exchange( + Self::LIVE, + loss as u8, + std::sync::atomic::Ordering::Relaxed, + std::sync::atomic::Ordering::Relaxed, + ); + } + + #[must_use] + pub fn loss(&self) -> Option { + match self.0.load(std::sync::atomic::Ordering::Relaxed) { + 1 => Some(SessionLoss::Displaced), + 2 => Some(SessionLoss::Disconnected), + _ => None, + } + } +} diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 8f3567d..8ca5519 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -30,7 +30,7 @@ use soulseek_rs::message::server::MessageFactory; use soulseek_rs::peer::ConnectionType; use soulseek_rs::{ Client, ClientSettings, ClientVersion, DownloadStatus, PeerAddress, - UploadStatus, + SessionLoss, UploadStatus, }; /// Only one test at a time may drive a server. @@ -930,6 +930,9 @@ struct MockDirectUpload { content: Vec, token: u32, ready: Sender<()>, + /// Hang up the control connection before streaming, the way clients that + /// drop idle peer sockets do while a transfer runs on its own connection. + close_control_first: bool, } fn run_mock_direct_peer(cfg: &MockDirectUpload) -> std::io::Result<()> { @@ -1016,6 +1019,13 @@ fn run_mock_direct_peer(cfg: &MockDirectUpload) -> std::io::Result<()> { } } + if cfg.close_control_first { + drop(p); + // Give the downloader time to notice the hangup before the bytes move, + // so the test measures what it does about it rather than outrunning it. + std::thread::sleep(Duration::from_secs(2)); + } + // 5. Stream the bytes over an F connection to the downloader's listener. // `srv` stays in scope so the peer remains online for the whole transfer. serve_file_over_f( @@ -1060,6 +1070,7 @@ fn a_file_downloads_from_a_peer_via_direct_connection() { content: content.clone(), token, ready: ready_tx, + close_control_first: false, }; let uploader = std::thread::spawn(move || { if let Err(e) = run_mock_direct_peer(&cfg) { @@ -2210,3 +2221,252 @@ fn a_user_who_never_logged_in_reads_as_offline() { .expect("with a presence"); assert!(!presence.status.is_reachable(), "got {}", presence.status); } + +// --------------------------------------------------------------------------- +// Concurrent sessions: several one-shot invocations running at once. +// --------------------------------------------------------------------------- + +/// Ask the server what address it hands out for `username`. +fn advertised_port(server_addr: &str, asker: &str, username: &str) -> u32 { + let mut peer = login_raw(server_addr, asker, "pw").expect("asker login"); + peer.write_all( + &MessageFactory::build_get_peer_address(username).get_buffer(), + ) + .expect("ask for the address"); + peer.flush().expect("flush"); + let mut reply = expect_code(&mut peer, 3, Duration::from_secs(5)) + .expect("the server should answer with an address"); + reply.set_pointer(8); + let _name = reply.read_string(); + for _ in 0..4 { + let _ = reply.read_int8(); + } + reply.read_int32() +} + +#[test] +fn a_taken_listen_port_does_not_cost_us_the_listener() { + let server = server_or_skip!(); + let addr = format!("{}:{}", server.host, server.port); + + // A sibling process already holds the configured port: the ordinary case + // when several invocations run at once with one config file between them. + let taken = free_port().expect("port to occupy"); + let _squatter = + std::net::TcpListener::bind(("0.0.0.0", taken)).expect("occupy"); + + let mut client = Client::with_settings(server.listening_settings( + "e2e_port_clash", + "pw", + taken, + )); + client.connect().expect("connect despite the taken port"); + assert!(client.login().expect("login")); + + let bound = client.listen_port().expect("a listener was still bound"); + assert_ne!(bound, taken, "a port someone else holds cannot be ours"); + std::thread::sleep(Duration::from_secs(1)); + + // Whatever port we ended up on, the server must hand peers that one: + // advertising a port held by someone else sends our search responses to a + // stranger, and we wait out the search seeing nothing. + let advertised = advertised_port(&addr, "e2e_port_asker", "e2e_port_clash"); + assert_eq!( + advertised, + u32::from(bound), + "the server must advertise the port we really bound" + ); +} + +#[test] +fn a_listening_client_reports_the_port_it_bound() { + let server = server_or_skip!(); + let port = free_port().expect("free port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_port_ok", + "pw", + port, + )); + client.connect().expect("connect"); + assert_eq!(client.listen_port(), Some(port)); + + let mut quiet = + Client::with_settings(server.settings("e2e_port_none", "pw")); + quiet.connect().expect("connect"); + assert_eq!( + quiet.listen_port(), + None, + "a client that does not listen has no port to report" + ); +} + +#[test] +fn a_second_login_under_one_name_reports_the_first_session_as_lost() { + let server = server_or_skip!(); + + let mut first = + Client::with_settings(server.settings("e2e_displaced", "pw")); + first.connect().expect("first connect"); + assert!(first.login().expect("first login")); + assert_eq!(first.session_loss(), None, "a fresh session is alive"); + std::thread::sleep(Duration::from_secs(1)); + + // The server allows one session per account, so this login evicts the one + // above. Silently seeing nothing from then on is what made concurrent runs + // report files as unavailable when they were not. + let mut second = + Client::with_settings(server.settings("e2e_displaced", "pw")); + second.connect().expect("second connect"); + assert!(second.login().expect("second login")); + + let deadline = Instant::now() + Duration::from_secs(15); + while first.session_loss().is_none() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + assert_eq!( + first.session_loss(), + Some(SessionLoss::Displaced), + "the evicted session must say why it went quiet" + ); + assert_eq!(second.session_loss(), None, "the winner keeps its session"); +} + +#[test] +fn clients_with_distinct_names_search_the_same_sharer_at_once() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + let content: Vec = (0..2048u32).map(|i| (i % 251) as u8).collect(); + std::fs::write(share_dir.join("e2e_probe_parallel.bin"), &content).unwrap(); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.listening_settings( + "e2e_par_sharer", + "pw", + free_port().expect("sharer port"), + ) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + + // Three sessions on one machine, all configured with the same listen port: + // exactly what several agents running the CLI at once look like. + let shared_port = free_port().expect("shared port"); + let seekers: Vec = (0..3) + .map(|i| { + let mut client = Client::with_settings(server.listening_settings( + &format!("e2e_par_seeker_{i}"), + "pw", + shared_port, + )); + client.connect().expect("seeker connect"); + assert!(client.login().expect("seeker login")); + client + }) + .collect(); + + let ports: Vec = + seekers.iter().filter_map(Client::listen_port).collect(); + assert_eq!(ports.len(), 3, "every seeker keeps a listener"); + let mut unique = ports.clone(); + unique.sort_unstable(); + unique.dedup(); + assert_eq!(unique.len(), 3, "each seeker holds its own port: {ports:?}"); + std::thread::sleep(Duration::from_secs(1)); + + std::thread::scope(|scope| { + for client in &seekers { + scope.spawn(move || { + let _ = client.search("parallel", Duration::from_secs(5)); + }); + } + }); + + for (i, client) in seekers.iter().enumerate() { + assert_eq!( + client.session_loss(), + None, + "seeker {i} should still hold its session" + ); + let hit = client + .get_search_results("parallel") + .into_iter() + .flat_map(|result| result.files) + .find(|file| file.name.contains("e2e_probe_parallel")); + assert!(hit.is_some(), "seeker {i} found nothing"); + } + + let _ = std::fs::remove_dir_all(share_dir); +} + +#[test] +fn a_peer_that_hangs_up_its_control_connection_still_delivers_the_file() { + let server = server_or_skip!(); + + // The bytes travel over their own F connection, so a peer is free to close + // the control connection first — other clients drop idle peer sockets as a + // matter of course. Treating that hangup as a failure used to fail every + // download queued with that peer, the one streaming included. + let client_port = free_port().expect("free client listen port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_fin_dl", + "pw", + client_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let mock_port = free_port().expect("free mock listen port"); + let filename = "hangup_song.mp3"; + let content: Vec = (0..2000u32).map(|i| (i % 251) as u8).collect(); + let size = content.len() as u64; + let download_dir = unique_download_dir(); + + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let cfg = MockDirectUpload { + server_addr: format!("{}:{}", server.host, server.port), + username: "e2e_finpeer".to_string(), + password: "pw".to_string(), + listen_port: mock_port, + downloader_listen_addr: format!("127.0.0.1:{client_port}"), + downloader_username: "e2e_fin_dl".to_string(), + filename: filename.to_string(), + content: content.clone(), + token: 616_161_u32, + ready: ready_tx, + close_control_first: true, + }; + let uploader = std::thread::spawn(move || { + if let Err(e) = run_mock_direct_peer(&cfg) { + eprintln!("[mock hangup peer] {e}"); + } + }); + + ready_rx + .recv_timeout(Duration::from_secs(10)) + .expect("mock peer ready"); + std::thread::sleep(Duration::from_secs(1)); + + let (_download, status_rx) = client + .download( + filename.to_string(), + "e2e_finpeer".to_string(), + size, + download_dir.display().to_string(), + ) + .expect("start download"); + + let completed = + wait_for_completion(&client, &status_rx, Duration::from_secs(20)); + let _ = uploader.join(); + + assert!( + completed, + "a closed control connection must not fail the transfer" + ); + let written = std::fs::read(download_dir.join(filename)) + .expect("downloaded file should exist"); + assert_eq!(written, content, "downloaded bytes should match the source"); + + let _ = std::fs::remove_dir_all(&download_dir); +} diff --git a/soulseek-rs/README.md b/soulseek-rs/README.md index c81f892..3cac306 100644 --- a/soulseek-rs/README.md +++ b/soulseek-rs/README.md @@ -50,7 +50,8 @@ soulseek-rs search "rjd2" --json \ ``` Exit codes: `0` success, `2` usage or config, `3` connect or login, `4` nothing -found, `5` timed out, `6` transfer failed, `1` unexpected error. +found, `5` timed out, `6` transfer failed, `7` session lost mid-command, `1` +unexpected error. The [full README](https://github.com/michel/soulseek-rs#readme) documents every command, record shape, flag, environment variable and `config.toml` key. diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index 0ddcc04..017e324 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -38,6 +38,7 @@ that form; the columns are a subset of the JSON keys. | 4 | It worked and found nothing | Widen or correct the query — retrying it verbatim will not help | | 5 | Gave up waiting | Raise `--timeout`, or `--search-timeout` for searches | | 6 | A transfer started and did not finish | The peer stalled; try another result | +| 7 | The session ended mid-command (usually another login took the username) | Nothing this run saw is reliable; retry, and give each concurrent run its own `--username` | ## Commands @@ -148,6 +149,22 @@ soulseek-rs user someuser --json `download --stdin` reads whole records, so filtering with `jq` and piping back is the normal way to be selective. +## Running several at once + +The server allows one session per account, and the later login wins: two runs +sharing a username silently kill each other's session. Give every concurrent +run its own name. + +```bash +for q in 'gary beck fold' 'marcal steady'; do + soulseek-rs search "$q" --username "$SOULSEEK_USERNAME-$RANDOM" --json & +done +wait +``` + +The listener port needs no such care: a run that finds the configured port +taken binds a free one and tells the server about that one instead. + ## Rules - **Remote paths belong to the peer.** They are backslash-separated and may @@ -159,6 +176,8 @@ is the normal way to be selective. searches. There is no unbounded form. - **Exit 4 is not a failure.** A search that finds nothing means the query was too narrow or misspelled. Widen it; use `whoami` to rule out the connection. +- **Exit 7 is not an answer.** The session ended mid-command, so nothing it saw + counts. Retry under a name no other run is using. - **Never put a password in argv** — `ps` shows it to every user on the box. Use `--password-stdin`, the OS keychain, or `SOULSEEK_PASSWORD_CMD`. - **Downloading pulls a file off a stranger's machine.** Confirm with the user diff --git a/soulseek-rs/src/cli.rs b/soulseek-rs/src/cli.rs index 86d53b0..1c61ba7 100644 --- a/soulseek-rs/src/cli.rs +++ b/soulseek-rs/src/cli.rs @@ -31,7 +31,7 @@ Examples: Exit codes: 0 success 2 usage/config 3 connect or login 4 no results - 5 timed out 6 transfer failed 1 unexpected error"; + 5 timed out 6 transfer failed 7 session lost 1 unexpected error"; #[derive(Parser, Debug, Default)] #[command( diff --git a/soulseek-rs/src/commands/mod.rs b/soulseek-rs/src/commands/mod.rs index e644b77..a28240f 100644 --- a/soulseek-rs/src/commands/mod.rs +++ b/soulseek-rs/src/commands/mod.rs @@ -17,9 +17,9 @@ use crate::cli::{Commands, MessageCommand, RoomCommand}; use crate::output::{CliError, CliResult, Exit, Out, PortmapRecord}; use crate::port_mapping::{self, PortMapper}; use soulseek_rs::types::{RoomEvent, RoomInfo}; -use soulseek_rs::{Client, ClientSettings}; +use soulseek_rs::{Client, ClientSettings, SessionLoss}; use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use std::time::{Duration, Instant}; /// Everything a one-shot command needs that does not come from its own flags. @@ -29,6 +29,47 @@ pub struct Ctx { pub download_dir: String, pub max_concurrent_downloads: usize, pub search_timeout: Duration, + /// The client of the session this command opened, once it has one, so the + /// verdict can be checked against the session that produced it. + pub session: OnceLock>, +} + +impl Ctx { + /// Re-read a "found nothing", "timed out" or "transfer failed" verdict + /// against the session that produced it. + /// + /// A session the server cut off — most often because another process + /// logged in under the same name — sees nothing at all from that moment + /// on. Reported as exit 4 it reads as "this file is not on the network", + /// and a caller acting on that gives up on files that were there. + fn checked(&self, result: CliResult) -> CliResult { + let Err(error) = result else { + return Ok(()); + }; + if !matches!( + error.exit, + Exit::NoResults | Exit::Timeout | Exit::Transfer + ) { + return Err(error); + } + match self.session.get().and_then(|client| client.session_loss()) { + Some(loss) => Err(session_lost(loss)), + None => Err(error), + } + } +} + +/// What to tell a caller whose session ended under it. +fn session_lost(loss: SessionLoss) -> CliError { + let fix = match loss { + SessionLoss::Displaced => { + "; give each concurrent run its own --username" + } + SessionLoss::Disconnected => "", + }; + CliError::session_lost(format!( + "the session ended before this command finished: {loss}{fix}" + )) } /// A logged-in client plus, while the listener is enabled, the port mapping @@ -45,21 +86,33 @@ const LOGIN_TIMEOUT: Duration = Duration::from_secs(30); impl Session { pub fn open(ctx: &Ctx) -> CliResult { - let mapper = ctx - .settings - .enable_listen - .then(|| PortMapper::spawn(ctx.settings.listen_port)); - let address = ctx.settings.server_address.to_string(); ctx.out.status(&format!("connecting to {address}…")); reachable(&address, REACH_TIMEOUT)?; - let client = login(ctx.settings.clone(), &address)?; + let client = Arc::new(login(ctx.settings.clone(), &address)?); ctx.out .status(&format!("logged in as {}", ctx.settings.username)); + let _ = ctx.session.set(Arc::clone(&client)); + + let bound = client.listen_port(); + if let Some(bound) = bound + && ctx.settings.listen_port != 0 + && bound != ctx.settings.listen_port + { + ctx.out.warn(&format!( + "listener port {} is in use; listening on {bound} instead", + ctx.settings.listen_port + )); + } + + // Ask the router for the port the listener actually holds: when the + // configured one was taken, mapping it would open a hole to whoever + // did take it. + let mapper = bound.map(PortMapper::spawn); Ok(Self { - client: Arc::new(client), + client, _mapper: mapper, }) } @@ -167,6 +220,10 @@ pub fn confirm_flush(client: &Client, timeout: Duration) -> bool { } pub fn run(ctx: &Ctx, command: Commands) -> CliResult { + ctx.checked(dispatch(ctx, command)) +} + +fn dispatch(ctx: &Ctx, command: Commands) -> CliResult { match command { Commands::Search(args) => transfer::search(ctx, &args), Commands::Download(args) => transfer::download(ctx, &args), diff --git a/soulseek-rs/src/commands/peer.rs b/soulseek-rs/src/commands/peer.rs index 894129b..3d7432b 100644 --- a/soulseek-rs/src/commands/peer.rs +++ b/soulseek-rs/src/commands/peer.rs @@ -20,11 +20,14 @@ const PRIVILEGE_TIMEOUT: Duration = Duration::from_secs(5); pub fn whoami(ctx: &Ctx) -> CliResult { let session = Session::open(ctx)?; let (folders, files) = session.client.shared_counts(); + // The bound port, not the configured one: a caller checking whether it is + // reachable needs what is true, not what was asked for. + let bound = session.client.listen_port(); ctx.out.emit(&WhoamiRecord { user: ctx.settings.username.clone(), server: ctx.settings.server_address.to_string(), - listening: ctx.settings.enable_listen, - listen_port: ctx.settings.listen_port, + listening: bound.is_some(), + listen_port: bound.unwrap_or(0), shared_folders: folders, shared_files: files, download_dir: ctx.download_dir.clone(), @@ -119,7 +122,7 @@ pub fn serve( let (folders, files) = session.client.shared_counts(); ctx.out.status(&format!( "serving {files} files in {folders} folders on port {}", - ctx.settings.listen_port + session.client.listen_port().unwrap_or_default() )); if files == 0 { ctx.out diff --git a/soulseek-rs/src/main.rs b/soulseek-rs/src/main.rs index 4267d67..213cabe 100644 --- a/soulseek-rs/src/main.rs +++ b/soulseek-rs/src/main.rs @@ -182,6 +182,7 @@ fn context( download_dir: resolved.download_dir.clone(), max_concurrent_downloads: resolved.max_concurrent_downloads, search_timeout: Duration::from_secs(resolved.search_timeout), + session: std::sync::OnceLock::new(), }) } @@ -298,11 +299,6 @@ fn run_default_tui( // Enable logger buffering BEFORE connection to prevent log artifacts soulseek_rs::utils::logger::enable_buffering(); - // Best-effort: make ourselves reachable behind a home router so - // firewalled peers can connect back. Kept alive for the session. - let _port_mapper = (!resolved.disable_listener) - .then(|| port_mapping::PortMapper::spawn(resolved.listener_port)); - let enable_listen = !resolved.disable_listener; let listen_port = resolved.listener_port; let make_settings = @@ -345,6 +341,14 @@ fn run_default_tui( persist_credentials(&outcome, config_path, file_config, &secret_store); + // Best-effort: make ourselves reachable behind a home router so firewalled + // peers can connect back. Mapped only once the listener is up, and for the + // port it really bound. Kept alive for the session. + let _port_mapper = outcome + .client + .listen_port() + .map(port_mapping::PortMapper::spawn); + let store = persist::paths::state_dir().map(persist::state::StateStore::new); diff --git a/soulseek-rs/src/output.rs b/soulseek-rs/src/output.rs index e12dd1b..0bd0d30 100644 --- a/soulseek-rs/src/output.rs +++ b/soulseek-rs/src/output.rs @@ -26,6 +26,9 @@ pub enum Exit { Timeout = 5, /// A transfer started but did not complete. Transfer = 6, + /// The server session ended mid-command, so nothing this run saw (or did + /// not see) means anything. Retry rather than believe it. + SessionLost = 7, } impl Exit { @@ -75,6 +78,10 @@ impl CliError { pub fn no_results(message: impl Into) -> Self { Self::new(Exit::NoResults, message) } + + pub fn session_lost(message: impl Into) -> Self { + Self::new(Exit::SessionLost, message) + } } impl std::fmt::Display for CliError { @@ -528,6 +535,7 @@ mod tests { assert_eq!(Exit::NoResults.code(), 4); assert_eq!(Exit::Timeout.code(), 5); assert_eq!(Exit::Transfer.code(), 6); + assert_eq!(Exit::SessionLost.code(), 7); } #[test] diff --git a/soulseek-rs/tests/cli_e2e.rs b/soulseek-rs/tests/cli_e2e.rs index 76039cd..4ff39a5 100644 --- a/soulseek-rs/tests/cli_e2e.rs +++ b/soulseek-rs/tests/cli_e2e.rs @@ -26,6 +26,7 @@ const EXIT_OK: i32 = 0; const EXIT_USAGE: i32 = 2; const EXIT_CONNECTION: i32 = 3; const EXIT_NO_RESULTS: i32 = 4; +const EXIT_SESSION_LOST: i32 = 7; /// Every environment variable the CLI reads, cleared for each run so a /// developer's shell (or a stray `.env`) cannot change what a test observes. @@ -114,7 +115,7 @@ fn help_lists_every_scriptable_command() { fn help_documents_the_exit_codes_a_script_branches_on() { let help = stdout(&run(&["--help"])); assert!(help.contains("Exit codes"), "help should list exit codes"); - for code in ["2", "3", "4", "5", "6"] { + for code in ["2", "3", "4", "5", "6", "7"] { assert!(help.contains(code), "exit code {code} should be documented"); } } @@ -552,6 +553,12 @@ impl TestServer { /// connection the *uploader* opens back to us, so a downloader without a /// listener never receives its file. fn args(&self, user: &str) -> Vec { + self.args_on(user, free_port().expect("listener port")) + } + + /// The same arguments on a chosen listener port, for the tests about what + /// happens when two runs want one port. + fn args_on(&self, user: &str, port: u16) -> Vec { vec![ "--no-config".to_string(), "--quiet".to_string(), @@ -562,7 +569,7 @@ impl TestServer { "--password".to_string(), "pw".to_string(), "--listener-port".to_string(), - free_port().expect("listener port").to_string(), + port.to_string(), ] } @@ -2800,3 +2807,183 @@ fn json_records_carry_every_extra_field_the_readme_names() { assert!(record.get(key).is_some(), "portmap should carry `{key}`"); } } + +// --------------------------------------------------------------------------- +// Several invocations at once: what an agent driving this CLI actually does. +// --------------------------------------------------------------------------- + +/// Spawn the binary as `user` with an explicit listener port, without waiting. +fn spawn_cli( + server: &TestServer, + user: &str, + port: u16, + args: &[&str], +) -> std::process::Child { + let mut all = server.args_on(user, port); + all.extend(args.iter().map(|a| (*a).to_string())); + let refs: Vec<&str> = all.iter().map(String::as_str).collect(); + command(&refs) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("the binary should start") +} + +#[test] +fn concurrent_searches_sharing_a_listener_port_all_find_the_file() { + let server = server_or_skip!(); + let share = Scratch::new("share"); + std::fs::write(share.path().join("cli_probe_multi.bin"), probe_bytes()) + .expect("share file"); + let _sharer = server.client("cli_e2e_sharer_multi", vec![share.display()]); + settle(); + + // One configured port between them, which is what a config file gives + // three agents started at the same moment. Losing the race for it used to + // cost that invocation its listener, and with it every search result. + let port = free_port().expect("shared listener port"); + // Collected on purpose: all three have to be running before any is waited + // on, which is the whole point of the test. + #[allow(clippy::needless_collect)] + let running: Vec<_> = (0..3) + .map(|i| { + spawn_cli( + &server, + &format!("cli_e2e_par_{i}"), + port, + &["--search-timeout", "10", "search", "multi"], + ) + }) + .collect(); + + for (i, child) in running.into_iter().enumerate() { + let output = child.wait_with_output().expect("wait"); + assert_eq!( + code(&output), + EXIT_OK, + "run {i} failed: {}", + stderr(&output) + ); + assert!( + records(&output) + .iter() + .any(|line| line.contains("cli_probe_multi")), + "run {i} listed nothing" + ); + } +} + +#[test] +fn a_session_taken_over_mid_search_is_not_reported_as_no_results() { + let server = server_or_skip!(); + let name = "cli_e2e_evicted"; + + // Run it loud, so its own progress line says when it is logged in: the + // eviction has to land after that, or this test would race the login it is + // trying to interrupt. + let mut args = server.args_on(name, free_port().expect("listener port")); + args.retain(|arg| arg != "--quiet"); + args.extend( + ["--search-timeout", "20", "search", "nothingmatchesthisqzx"] + .map(String::from), + ); + let refs: Vec<&str> = args.iter().map(String::as_str).collect(); + let mut searching = command(&refs) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("the binary should start"); + + let (logged_in, wait_for_login) = std::sync::mpsc::channel(); + let stderr = searching.stderr.take().expect("piped stderr"); + let progress = std::thread::spawn(move || { + let mut seen = String::new(); + for line in std::io::BufRead::lines(std::io::BufReader::new(stderr)) + .map_while(Result::ok) + { + if line.contains("logged in as") { + let _ = logged_in.send(()); + } + seen.push_str(&line); + seen.push('\n'); + } + seen + }); + wait_for_login + .recv_timeout(Duration::from_mins(1)) + .expect("the search should log in"); + + // Now take the username away, the way a second invocation under one + // account does. + let _thief = server.client(name, Vec::new()); + + let status = searching.wait().expect("wait"); + let seen = progress.join().expect("progress reader"); + assert_eq!( + status.code(), + Some(EXIT_SESSION_LOST), + "a killed session must not look like an answered query: {seen}" + ); + + let mut written = String::new(); + std::io::Read::read_to_string( + &mut searching.stdout.take().expect("piped stdout"), + &mut written, + ) + .expect("read stdout"); + assert!(written.trim().is_empty(), "stdout stays data-only"); +} + +#[test] +fn concurrent_downloads_sharing_a_listener_port_all_land() { + let server = server_or_skip!(); + let share = Scratch::new("share"); + std::fs::write(share.path().join("cli_probe_dual.bin"), probe_bytes()) + .expect("share file"); + let _sharer = server.client("cli_e2e_sharer_dual", vec![share.display()]); + settle(); + + // A transfer arrives over a connection the *uploader* opens back to us, so + // a run without a listener never receives its file however well its search + // went. Two `get` runs, one configured port between them. + let port = free_port().expect("shared listener port"); + let targets: Vec = + (0..2).map(|i| Scratch::new(&format!("dl{i}"))).collect(); + // Collected on purpose: both have to be running before either is waited + // on, which is the whole point of the test. + #[allow(clippy::needless_collect)] + let running: Vec<_> = targets + .iter() + .enumerate() + .map(|(i, target)| { + spawn_cli( + &server, + &format!("cli_e2e_dual_{i}"), + port, + &[ + "--download-dir", + &target.display(), + "--search-timeout", + "10", + "get", + "dual", + "--timeout", + "40", + ], + ) + }) + .collect(); + + for (i, (child, target)) in running.into_iter().zip(&targets).enumerate() { + let output = child.wait_with_output().expect("wait"); + assert_eq!( + code(&output), + EXIT_OK, + "run {i} failed: {}", + stderr(&output) + ); + let written = std::fs::read(target.path().join("cli_probe_dual.bin")) + .unwrap_or_else(|e| panic!("run {i} wrote no file: {e}")); + assert_eq!(written, probe_bytes(), "run {i} wrote the wrong bytes"); + } +}