From 65c44cc66b7f70f4d84c2767b83ca3210b7d3cff Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Wed, 9 Sep 2026 16:27:20 +0000 Subject: [PATCH] feat(protocol): away status, excluded phrases, queue-place requests Three features the client half-had. SetStatus (28) was only ever sent as "online" at login, so there was no way to go away; excluded search phrases (160) were parsed and logged into nowhere, so every client kept spending searches on queries the server was never going to answer; and a queued download could answer a peer asking about its place but could never ask about its own. Each is now driven end to end: the away status seen by another user, a refused query on a stub server that announces phrases (soulfind keeps its filters in a database the suite cannot seed), and a peer's answer landing on the download as its queue position. --- soulseek-rs-lib/src/actor/server_actor.rs | 7 + soulseek-rs-lib/src/client/mod.rs | 31 ++ soulseek-rs-lib/src/client/operations.rs | 5 + soulseek-rs-lib/src/client/rooms.rs | 17 +- soulseek-rs-lib/src/client/search.rs | 26 ++ soulseek-rs-lib/src/client/social.rs | 33 +++ soulseek-rs-lib/src/error.rs | 6 + .../message/server/excluded_search_phrases.rs | 25 +- .../src/message/server/message_factory.rs | 18 ++ soulseek-rs-lib/tests/e2e.rs | 264 ++++++++++++++++++ 10 files changed, 429 insertions(+), 3 deletions(-) diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 6fb2636..16bd473 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -227,6 +227,8 @@ pub enum ServerMessage { CantConnectToPeer { token: u32, }, + /// Phrases the server refuses to search for (code 160). + ExcludedSearchPhrases(Vec), } pub struct ServerActor { @@ -692,6 +694,11 @@ impl ServerActor { token, }); } + ServerMessage::ExcludedSearchPhrases(phrases) => { + self.forward_to_client(ClientOperation::ExcludedSearchPhrases( + phrases, + )); + } other => { error!("[server] unroutable message: {:?}", other); } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 75dd60c..61ad87a 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -368,6 +368,8 @@ pub enum ClientOperation { CantConnectToPeer { token: u32, }, + /// Phrases the server refuses to search for (code 160). + ExcludedSearchPhrases(Vec), /// A parent candidate told us how deep it sits. `link` says which dial /// to that user is talking. ParentBranchLevel { @@ -469,6 +471,9 @@ pub struct ClientContext { privileged_users: HashSet, /// Seconds of our own privileges left (code 92), once we have asked. own_privileges: Option, + /// Phrases the server will not search for (code 160). A query carrying one + /// is refused locally rather than spent on the server. + excluded_search_phrases: Vec, /// Peers waiting for one of our upload slots. upload_queue: Vec, /// Arrival counter for the queue's first-come tie-break. @@ -575,6 +580,7 @@ impl ClientContext { wishlist_interval: None, privileged_users: HashSet::new(), own_privileges: None, + excluded_search_phrases: Vec::new(), upload_queue: Vec::new(), upload_seq: 0, upload_slots: DEFAULT_UPLOAD_SLOTS, @@ -725,6 +731,31 @@ impl ClientContext { self.user_interests.get(username).cloned() } + /// Record the phrases the server refuses to search for (code 160). + pub fn set_excluded_search_phrases(&mut self, phrases: Vec) { + self.excluded_search_phrases = phrases; + } + + /// The phrases the server refuses to search for. + #[must_use] + pub fn excluded_search_phrases(&self) -> Vec { + self.excluded_search_phrases.clone() + } + + /// The first excluded phrase `query` contains, if any. Matching is + /// case-insensitive and on substrings, which is how the server applies its + /// own filter. + #[must_use] + pub fn excluded_phrase_in(&self, query: &str) -> Option { + let haystack = query.to_lowercase(); + self.excluded_search_phrases + .iter() + .find(|phrase| { + !phrase.is_empty() && haystack.contains(&phrase.to_lowercase()) + }) + .cloned() + } + /// Forget any interests cached for `username`, so a poll after a fresh /// request cannot mistake the previous answer for this one. pub fn invalidate_user_interests(&mut self, username: &str) { diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index d53c33e..b2dee36 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -816,6 +816,11 @@ impl Client { ctx.apply_user_interests(interests); } } + ClientOperation::ExcludedSearchPhrases(phrases) => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.set_excluded_search_phrases(phrases); + } + } ClientOperation::CantConnectToPeer { token } => { // The peer we asked the server to broker gave up. The // token is the correlation we sent, so it names the diff --git a/soulseek-rs-lib/src/client/rooms.rs b/soulseek-rs-lib/src/client/rooms.rs index 6ea95d9..8fee72c 100644 --- a/soulseek-rs-lib/src/client/rooms.rs +++ b/soulseek-rs-lib/src/client/rooms.rs @@ -234,9 +234,22 @@ impl Client { pub fn browse_user(&self, username: &str) -> Result<()> { let request = crate::message::server::MessageFactory::build_get_share_file_list(); + self.context.write_safe()?.mark_browse_pending(username); + self.send_to_peer_or_queue(username, request) + } + + /// Send `request` to `username` over their control connection, or hold it + /// until one exists — asking the server for their address so it does. + /// + /// # Errors + /// [`SoulseekRs::LockPoisoned`] if the client state is poisoned. + pub(super) fn send_to_peer_or_queue( + &self, + username: &str, + request: crate::message::Message, + ) -> Result<()> { let (connected, registry) = { - let mut ctx = self.context.write_safe()?; - ctx.mark_browse_pending(username); + let ctx = self.context.read_safe()?; ( ctx.peer_registry .as_ref() diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index 9e54136..cf9ed10 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -62,6 +62,9 @@ impl Client { let Some(handle) = &self.server_handle else { return Err(SoulseekRs::NotConnected); }; + if let Some(phrase) = self.excluded_phrase_in(query) { + return Err(SoulseekRs::SearchPhraseExcluded(phrase)); + } let token = next_search_token(); self.context.write_safe()?.searches.insert( @@ -119,6 +122,9 @@ impl Client { if self.server_handle.is_none() { return Err(SoulseekRs::NotConnected); } + if let Some(phrase) = self.excluded_phrase_in(query) { + return Err(SoulseekRs::SearchPhraseExcluded(phrase)); + } let token = next_search_token(); self.context.write_safe()?.searches.insert( query.to_string(), @@ -130,6 +136,26 @@ impl Client { Ok(token) } + /// The phrases the server refuses to search for (code 160), as last + /// announced. Empty until the server has said. + #[must_use] + pub fn excluded_search_phrases(&self) -> Vec { + self.context + .read_safe() + .map(|ctx| ctx.excluded_search_phrases()) + .unwrap_or_default() + } + + /// The excluded phrase `query` carries, if any — what a search would be + /// refused for. + #[must_use] + pub fn excluded_phrase_in(&self, query: &str) -> Option { + self.context + .read_safe() + .ok() + .and_then(|ctx| ctx.excluded_phrase_in(query)) + } + /// Let responses accumulate for `timeout`, or until cancelled. /// /// Peers answer a search over their own connections whenever they get round diff --git a/soulseek-rs-lib/src/client/social.rs b/soulseek-rs-lib/src/client/social.rs index 5d402e4..488deb8 100644 --- a/soulseek-rs-lib/src/client/social.rs +++ b/soulseek-rs-lib/src/client/social.rs @@ -21,6 +21,39 @@ impl Client { self.send_server_message(MessageFactory::build_server_ping()) } + /// Set our online status (server code 28): away, or back to online. + /// + /// Other clients see it in their `GetUserStatus` and watch replies, and + /// the server uses it when it decides who to hand a search to. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn set_away(&self, away: bool) -> Result<()> { + let status = u32::from(!away) + 1; // away = 1, online = 2 + self.send_server_message(MessageFactory::build_set_status_message( + status, + )) + } + + /// Ask `username` where the file we queued with them sits (peer code 51). + /// + /// The answer arrives asynchronously and lands on the download, readable + /// through [`Client::downloads`] as its queue position. A peer we have no + /// connection to is resolved and the request delivered once it is up, the + /// same way a browse request is. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection, + /// or [`crate::SoulseekRs::LockPoisoned`] if the client state is poisoned. + pub fn request_place_in_queue( + &self, + username: &str, + filename: &str, + ) -> Result<()> { + let request = MessageFactory::build_place_in_queue_request(filename); + self.send_to_peer_or_queue(username, request) + } + /// Set our ticker in `room` — the one-line message other members see. /// Passing an empty `ticker` clears ours. /// diff --git a/soulseek-rs-lib/src/error.rs b/soulseek-rs-lib/src/error.rs index b25df46..5b3b722 100644 --- a/soulseek-rs-lib/src/error.rs +++ b/soulseek-rs-lib/src/error.rs @@ -20,6 +20,9 @@ pub enum SoulseekRs { CompressionError(String), /// A lock was poisoned by a panic in another thread LockPoisoned, + /// The query carries a phrase the server refuses to search for (code 160), + /// so sending it would spend a search on a query nothing answers. + SearchPhraseExcluded(String), } impl fmt::Display for SoulseekRs { @@ -41,6 +44,9 @@ impl fmt::Display for SoulseekRs { Self::CompressionError(msg) => { write!(f, "Compression error: {msg}") } + Self::SearchPhraseExcluded(phrase) => { + write!(f, "The server excludes searches for \"{phrase}\"") + } Self::LockPoisoned => { write!(f, "Lock poisoned by panicking thread") } diff --git a/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs b/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs index 6779bc1..2233529 100644 --- a/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs +++ b/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs @@ -12,7 +12,7 @@ impl MessageHandler for ExcludedSearchPhrasesHandler { 160 } - fn handle(&self, message: &mut Message, _sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { let item_count = message.read_int32(); let mut exluded_phrases: Vec = Vec::new(); @@ -25,6 +25,11 @@ impl MessageHandler for ExcludedSearchPhrasesHandler { exluded_phrases.push(phrase); } debug!("Excluded search phrases: {:?}", exluded_phrases); + // Kept, not just logged: the server rejects a search carrying one of + // these, so a client that forgets them spends its search allowance on + // queries that were never going to be answered. + let _ = + sender.send(ServerMessage::ExcludedSearchPhrases(exluded_phrases)); } } @@ -43,4 +48,22 @@ mod tests { }); ExcludedSearchPhrasesHandler.handle(&mut message, tx); } + + #[test] + fn the_phrases_are_reported_not_only_logged() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_int32(2); + m.write_string("banned"); + m.write_string("blocked"); + }); + + ExcludedSearchPhrasesHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::ExcludedSearchPhrases(phrases)) => { + assert_eq!(phrases, ["banned", "blocked"]); + } + other => panic!("unexpected: {other:?}"), + } + } } diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 0f7386b..cc3feb7 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -301,6 +301,15 @@ impl MessageFactory { .clone() } + /// Ask a peer where the file we queued with them sits (peer code 51). + #[must_use] + pub fn build_place_in_queue_request(filename: &str) -> Message { + Message::new() + .write_int32(51) + .write_string(filename) + .clone() + } + /// A keepalive ping (server code 32, no body). The server sends no reply; /// its only job is to keep a quiet connection from being reaped by a NAT /// or the server's own idle timeout, the way other clients ping. @@ -687,6 +696,15 @@ fn an_upload_speed_report_carries_the_rate() { assert_eq!(decoded.read_int32(), 900); } +#[test] +fn a_place_in_queue_request_names_the_file() { + let message = MessageFactory::build_place_in_queue_request("a\\b.mp3"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 51); + decoded.set_pointer(8); + assert_eq!(decoded.read_string(), "a\\b.mp3"); +} + #[test] fn a_ping_is_just_its_code() { let message = MessageFactory::build_server_ping(); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 1f9d3ee..120291e 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -4319,3 +4319,267 @@ fn a_brokered_dial_we_cannot_complete_tells_the_waiting_peer() { "the reply should quote the token the peer asked with" ); } + +// --------------------------------------------------------------------------- +// Features that were implemented but never driven end to end: the away +// status, the phrases the server refuses to search for, and asking a peer +// where our queued file sits. +// --------------------------------------------------------------------------- + +#[test] +fn going_away_is_visible_to_another_user() { + use soulseek_rs::types::UserStatus; + let server = server_or_skip!(); + + let mut alice = + Client::with_settings(server.settings("e2e_away_alice", "pw")); + let mut bob = Client::with_settings(server.settings("e2e_away_bob", "pw")); + alice.connect().expect("alice connect"); + bob.connect().expect("bob connect"); + assert!(alice.login().expect("alice login")); + assert!(bob.login().expect("bob login")); + + // Online to begin with, as the post-login handshake announces. + let status_of = |client: &Client| -> Option { + client + .user_info("e2e_away_alice") + .and_then(|info| info.presence) + .map(|presence| presence.status) + }; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline + && status_of(&bob) != Some(UserStatus::Online) + { + bob.request_user_info("e2e_away_alice").expect("ask"); + std::thread::sleep(Duration::from_millis(250)); + } + assert_eq!(status_of(&bob), Some(UserStatus::Online)); + + alice.set_away(true).expect("alice goes away"); + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline && status_of(&bob) != Some(UserStatus::Away) + { + bob.request_user_info("e2e_away_alice").expect("ask again"); + std::thread::sleep(Duration::from_millis(250)); + } + assert_eq!( + status_of(&bob), + Some(UserStatus::Away), + "the away status should reach another user" + ); + + // And coming back is visible too. + alice.set_away(false).expect("alice comes back"); + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline + && status_of(&bob) != Some(UserStatus::Online) + { + bob.request_user_info("e2e_away_alice") + .expect("ask once more"); + std::thread::sleep(Duration::from_millis(250)); + } + assert_eq!(status_of(&bob), Some(UserStatus::Online)); +} + +/// A minimal server that answers a login and then sends `after_login`. +/// +/// soulfind keeps its search filters in its database, which the suite has no +/// way to seed, so the excluded-phrase path is driven from a stub that speaks +/// just enough of the protocol to deliver code 160 on the wire. +fn stub_server_sending( + after_login: Vec, +) -> (String, std::thread::JoinHandle<()>) { + let listener = + std::net::TcpListener::bind("127.0.0.1:0").expect("bind stub server"); + let addr = listener.local_addr().expect("stub addr"); + let handle = std::thread::spawn(move || { + let Ok((mut stream, _)) = listener.accept() else { + return; + }; + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("stub read timeout"); + // Wait for the login (code 1) before answering it. + loop { + match read_framed(&mut stream) { + Ok(msg) if msg.get_message_code() == 1 => break, + Ok(_) => {} + Err(_) => return, + } + } + let mut ok = Message::new(); + ok.write_int32(1) + .write_int8(1) + .write_string("stub greeting") + .write_int32(0) + .write_string("") + .write_int8(0); + if stream.write_all(&ok.get_buffer()).is_err() { + return; + } + for message in after_login { + if stream.write_all(&message.get_buffer()).is_err() { + return; + } + } + let _ = stream.flush(); + // Hold the connection open so the client stays logged in. + std::thread::sleep(Duration::from_secs(10)); + }); + (format!("{}:{}", addr.ip(), addr.port()), handle) +} + +#[test] +fn an_excluded_phrase_is_refused_before_a_search_is_spent() { + let mut phrases = Message::new(); + phrases + .write_int32(160) + .write_int32(2) + .write_string("banned-phrase") + .write_string("blocked"); + let (addr, stub) = stub_server_sending(vec![phrases]); + let (host, port) = addr.rsplit_once(':').expect("stub addr"); + + let mut client = Client::with_settings(ClientSettings { + username: "e2e_excluded".to_string(), + password: "pw".to_string(), + server_address: PeerAddress::new( + host.to_string(), + port.parse().expect("stub port"), + ), + enable_listen: false, + listen_port: 0, + shared_directories: Vec::new(), + version: ClientVersion::default(), + }); + client.connect().expect("connect to stub"); + assert!(client.login().expect("login to stub")); + + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline + && client.excluded_search_phrases().is_empty() + { + std::thread::sleep(Duration::from_millis(100)); + } + assert_eq!( + client.excluded_search_phrases(), + ["banned-phrase", "blocked"], + "the announced phrases should be kept" + ); + + // A query carrying one is refused here rather than spent on the server, + // and the match is case-insensitive on substrings, as the server's is. + match client.search("A BANNED-PHRASE query", Duration::from_millis(100)) { + Err(soulseek_rs::SoulseekRs::SearchPhraseExcluded(phrase)) => { + assert_eq!(phrase, "banned-phrase"); + } + other => panic!("expected a refusal, got {other:?}"), + } + // A clean query is not refused. + assert!( + client + .search("something else", Duration::from_millis(100)) + .is_ok() + ); + + drop(client); + let _ = stub.join(); +} + +#[test] +fn a_queued_download_can_ask_where_it_sits() { + // We queue a file with a peer that never starts it, then ask where it + // sits (peer code 51). The peer's answer (code 44) must land on the + // download as its queue position. + let server = server_or_skip!(); + + let listen_port = free_port().expect("free listen port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_place_asker", + "pw", + listen_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let filename = "waiting_in_line.mp3"; + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let (asked_tx, asked_rx) = std::sync::mpsc::channel(); + let listen_addr = format!("127.0.0.1:{listen_port}"); + let peer = std::thread::spawn(move || -> std::io::Result<()> { + let mut p = connect_retry(&listen_addr, Duration::from_secs(5))?; + p.set_read_timeout(Some(Duration::from_secs(15)))?; + p.write_all(&peer_init_bytes("e2e_place_peer", "P", 0))?; + p.flush()?; + let _ = ready_tx.send(()); + + // Sit on the queue request, then answer the place request with 7. + loop { + let mut msg = read_framed(&mut p)?; + // A QueueUpload (43) is accepted but never started: the file + // stays in line, which is the state being asked about. + if msg.get_message_code() == 51 { + msg.set_pointer(8); + let asked = msg.read_string(); + let mut reply = Message::new(); + reply.write_int32(44).write_string(&asked).write_int32(7); + p.write_all(&reply.get_buffer())?; + p.flush()?; + let _ = asked_tx.send(asked); + break; + } + } + std::thread::sleep(Duration::from_secs(1)); + Ok(()) + }); + ready_rx + .recv_timeout(Duration::from_secs(5)) + .expect("mock peer P connection"); + std::thread::sleep(Duration::from_millis(1000)); + + let download_dir = unique_download_dir(); + let (_download, _status_rx) = client + .download( + filename.to_string(), + "e2e_place_peer".to_string(), + 10, + download_dir.display().to_string(), + ) + .expect("start download"); + + // Ask, retrying until the control connection carries it. + let deadline = Instant::now() + Duration::from_secs(10); + let mut asked = None; + while Instant::now() < deadline && asked.is_none() { + client + .request_place_in_queue("e2e_place_peer", filename) + .expect("ask for our place"); + asked = asked_rx.recv_timeout(Duration::from_millis(500)).ok(); + } + assert_eq!( + asked.as_deref(), + Some(filename), + "the peer should be asked about the file we queued" + ); + + let deadline = Instant::now() + Duration::from_secs(5); + let mut place = None; + while Instant::now() < deadline && place.is_none() { + place = client + .get_all_downloads() + .into_iter() + .find(|d| d.filename == filename) + .and_then(|d| d.queue_position); + if place.is_none() { + std::thread::sleep(Duration::from_millis(100)); + } + } + assert_eq!( + place, + Some(7), + "the peer's answer should land on the download" + ); + + let _ = peer.join(); + let _ = std::fs::remove_dir_all(download_dir); +}