From e8a48c6b064fdbfe28901fb0e61cbb5e8675df7c Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Sun, 26 Jul 2026 14:52:31 +0200 Subject: [PATCH] fix!: make concurrent invocations work instead of silently seeing nothing (#22) Running several one-shot commands at once returned "no results" for queries that had plenty. Four defects, each of which made a dead session look like an empty network: - The peer listener bound its port inside a detached thread and panicked on EADDRINUSE, unwinding that thread alone. The process carried on with no listener, so no peer could deliver a search response. `Listen::bind` now takes the socket before login and falls back to an ephemeral port when the configured one is held. - SetWaitPort advertised the configured port whether or not it was bound, so peers dialled a sibling process that discarded responses to tickets it never issued. The port the listener really holds is what gets advertised, mapped through UPnP, and reported by `whoami` and `serve`. - A zero-byte read is EOF on a TCP socket, but MessageReader reported it as success, so a server that closed cleanly left the actor polling a dead socket. It is now an error, and Relogged (code 41) has a handler. The peer actor classifies that error: a control connection that was established and then closed is a clean disconnect, because peers drop idle sockets while the bytes move over their own connection, and reporting an error there fails every download queued with that peer including the one streaming. A connection that never established keeps the error path, which is what the server-brokered fallback reads to know a peer is unreachable. - A session the server cut off - usually because another process logged in under the same name - exited 4, documented as "it worked and found nothing". Callers concluded the file was unavailable. Sessions now record why they ended and those verdicts become exit 7. The server allows one session per account, so concurrency still needs a distinct --username per run; SKILL.md documents that. The listener port needs no such care any more. Six e2e tests against soulfind cover it, each of which fails without the fix: a taken listen port, a displaced session, three clients searching at once under one configured port, the two CLI equivalents, and a peer that hangs up its control connection while a transfer is still to come. BREAKING CHANGE: `Listen::start` is replaced by `Listen::bind` and `Listen::serve`, so the caller owns the socket and can read the port it actually bound. `Client::listen_port` and `Client::session_loss` are new. --- README.md | 5 + soulseek-rs-lib/src/actor/peer_actor.rs | 19 ++ soulseek-rs-lib/src/actor/server_actor.rs | 35 ++- soulseek-rs-lib/src/client/connection.rs | 26 +- soulseek-rs-lib/src/client/mod.rs | 25 +- soulseek-rs-lib/src/lib.rs | 4 +- soulseek-rs-lib/src/message/message_reader.rs | 36 ++- soulseek-rs-lib/src/message/server/mod.rs | 2 + .../src/message/server/relogged.rs | 18 ++ soulseek-rs-lib/src/peer/listen.rs | 43 +-- soulseek-rs-lib/src/types.rs | 59 ++++ soulseek-rs-lib/tests/e2e.rs | 262 +++++++++++++++++- soulseek-rs/README.md | 3 +- soulseek-rs/src/SKILL.md | 19 ++ soulseek-rs/src/cli.rs | 2 +- soulseek-rs/src/commands/mod.rs | 75 ++++- soulseek-rs/src/commands/peer.rs | 9 +- soulseek-rs/src/main.rs | 14 +- soulseek-rs/src/output.rs | 8 + soulseek-rs/tests/cli_e2e.rs | 191 ++++++++++++- 20 files changed, 801 insertions(+), 54 deletions(-) create mode 100644 soulseek-rs-lib/src/message/server/relogged.rs 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"); + } +}