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

@@ -227,6 +227,8 @@ pub enum ServerMessage {
     CantConnectToPeer {
         token: u32,
     },
+    /// Phrases the server refuses to search for (code 160).
+    ExcludedSearchPhrases(Vec<String>),
 }
 
 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);
             }

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<String>),
     /// 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<String>,
     /// Seconds of our own privileges left (code 92), once we have asked.
     own_privileges: Option<u32>,
+    /// 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<String>,
     /// Peers waiting for one of our upload slots.
     upload_queue: Vec<QueuedUpload>,
     /// 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<String>) {
+        self.excluded_search_phrases = phrases;
+    }
+
+    /// The phrases the server refuses to search for.
+    #[must_use]
+    pub fn excluded_search_phrases(&self) -> Vec<String> {
+        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<String> {
+        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) {

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

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()?;
+            let ctx = self.context.read_safe()?;
-            ctx.mark_browse_pending(username);
             (
                 ctx.peer_registry
                     .as_ref()

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<String> {
+        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<String> {
+        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

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.
     ///

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")
             }

soulseek-rs-lib/src/message/server/excluded_search_phrases.rs

@@ -12,7 +12,7 @@ impl MessageHandler<ServerMessage> for ExcludedSearchPhrasesHandler {
         160
     }
 
-    fn handle(&self, message: &mut Message, _sender: Sender<ServerMessage>) {
+    fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
         let item_count = message.read_int32();
 
         let mut exluded_phrases: Vec<String> = Vec::new();
@@ -25,6 +25,11 @@ impl MessageHandler<ServerMessage> 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:?}"),
+        }
+    }
 }

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();

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<UserStatus> {
+        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<Message>,
+) -> (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);
+}