fix(messages): acknowledge stored private messages, and disconnect cleanly

Two bugs an e2e test found the moment one existed for this path.

A private message the server held while we were offline arrives with
new_message false, and the handler only acknowledged the ones flagged
new — so exactly the messages that need an ack never got one and the
server re-delivered them at every login, forever. Every delivery is now
acknowledged.

A dropped Client also left its server connection open, so the session
lived on: mail addressed to us went to a socket nobody was reading and a
later login collided with our own ghost. Client::disconnect closes the
server and peer connections, and Drop calls it.

Also covers two features that were never driven end to end: an offline
message delivered once and not again, and a wishlist search answered like
any other.

soulseek-rs-lib/src/actor/peer_registry.rs

@@ -202,6 +202,24 @@ impl PeerRegistry {
         None
     }
 
+    /// Stop every peer actor and empty the registry, closing the connections
+    /// they hold. Used when the client shuts down.
+    pub fn stop_all(&self) {
+        let handles = match self.peers.lock_safe() {
+            Ok(mut peers) => peers
+                .drain()
+                .map(|(_, (_, handle))| handle)
+                .collect::<Vec<_>>(),
+            Err(e) => {
+                error!("[peer_registry] stop_all: {}", e);
+                return;
+            }
+        };
+        for handle in handles {
+            let _ = handle.stop();
+        }
+    }
+
     #[must_use]
     pub fn contains(&self, username: &str) -> bool {
         match self.peers.lock_safe() {

soulseek-rs-lib/src/client/connection.rs

@@ -133,6 +133,30 @@ impl Client {
         Ok(())
     }
 
+    /// Close the server connection and every peer connection.
+    ///
+    /// The server sees the socket close and marks us offline, which is what
+    /// releases the username: a client that simply goes out of scope without
+    /// this leaves a session the server still believes in, so mail addressed
+    /// to us is delivered to a connection nobody is reading and later logins
+    /// collide with our own ghost.
+    ///
+    /// Called automatically when the client is dropped. The peer listener's
+    /// bound port, if one was opened, is released with the process.
+    pub fn disconnect(&mut self) {
+        if let Some(handle) = self.server_handle.take() {
+            let _ = handle.stop();
+        }
+        let registry = self
+            .context
+            .read_safe()
+            .ok()
+            .and_then(|ctx| ctx.peer_registry.clone());
+        if let Some(registry) = registry {
+            registry.stop_all();
+        }
+    }
+
     /// Log in and wait for the server's verdict.
     ///
     /// The wait is bounded: the actor only answers once it has a working

soulseek-rs-lib/src/client/mod.rs

@@ -1163,6 +1163,12 @@ pub struct Client {
     session: SessionWatch,
 }
 
+impl Drop for Client {
+    fn drop(&mut self) {
+        self.disconnect();
+    }
+}
+
 impl Client {
     pub fn new(
         username: impl Into<String>,

soulseek-rs-lib/src/client/tests.rs

@@ -551,7 +551,7 @@ fn an_expired_broker_connect_fails_the_queued_downloads() {
     let (_ops_tx, ops_rx) = mpsc::channel();
     Client::listen_to_client_operations(
         ops_rx,
-        client.context,
+        client.context.clone(),
         "u".to_string(),
     );
 
@@ -582,7 +582,7 @@ fn a_replayed_transfer_response_does_not_start_a_second_transfer() {
     let (ops_tx, ops_rx) = mpsc::channel();
     Client::listen_to_client_operations(
         ops_rx,
-        client.context,
+        client.context.clone(),
         "u".to_string(),
     );
 
@@ -648,7 +648,7 @@ fn a_cancelled_download_is_not_started_when_the_peer_allows_it() {
     let (ops_tx, ops_rx) = mpsc::channel();
     Client::listen_to_client_operations(
         ops_rx,
-        client.context,
+        client.context.clone(),
         "u".to_string(),
     );
     let peer = Peer::new(

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

@@ -28,16 +28,73 @@ impl MessageHandler<ServerMessage> for MessageUser {
 
         info!("[MessageUser] User message received:{:?}", user_message);
 
-        // Acknowledge freshly delivered messages so the server does not keep
-        // re-sending them on every reconnect.
-        if new_message {
+        // Acknowledge every message, not only the ones flagged new. The flag
+        // marks a message delivered as it was sent; one the server stored
+        // while we were offline arrives with it false, and those are exactly
+        // the messages the server keeps re-sending at every login until they
+        // are acknowledged.
         let _ = sender.send(ServerMessage::SendMessage(
             MessageFactory::build_message_acked(id),
         ));
-        }
 
         // Surface the message to the client so it can be read via the API.
         let _ =
             sender.send(ServerMessage::PrivateMessageReceived(user_message));
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::message::framed;
+
+    fn deliver(new_message: bool) -> Vec<ServerMessage> {
+        let (tx, rx) = std::sync::mpsc::channel();
+        let mut message = framed(|m| {
+            m.write_int32(7);
+            m.write_int32(1_700_000_000);
+            m.write_string("alice");
+            m.write_string("hello");
+            m.write_int8(u8::from(new_message));
+        });
+        MessageUser.handle(&mut message, tx);
+        rx.try_iter().collect()
+    }
+
+    fn acked_id(messages: &[ServerMessage]) -> Option<u32> {
+        messages.iter().find_map(|message| match message {
+            ServerMessage::SendMessage(m) => {
+                let mut decoded = Message::new_with_data(m.get_buffer());
+                (decoded.get_message_code() == 23).then(|| {
+                    decoded.set_pointer(8);
+                    decoded.read_int32()
+                })
+            }
+            _ => None,
+        })
+    }
+
+    #[test]
+    fn a_freshly_delivered_message_is_acknowledged() {
+        let sent = deliver(true);
+        assert_eq!(acked_id(&sent), Some(7));
+    }
+
+    #[test]
+    fn a_message_stored_while_we_were_offline_is_acknowledged_too() {
+        // Stored messages arrive with new_message false; without an ack the
+        // server re-delivers them at every login, forever.
+        let sent = deliver(false);
+        assert_eq!(acked_id(&sent), Some(7));
+    }
+
+    #[test]
+    fn the_message_itself_is_surfaced_to_the_client() {
+        let sent = deliver(false);
+        assert!(sent.iter().any(|message| matches!(
+            message,
+            ServerMessage::PrivateMessageReceived(m)
+                if m.message() == "hello" && m.username() == "alice"
+        )));
+    }
+}

soulseek-rs-lib/tests/e2e.rs

@@ -4708,3 +4708,103 @@ fn a_private_room_someone_else_owns_cannot_be_taken() {
         "a refused room must not appear as one we belong to"
     );
 }
+
+#[test]
+fn an_acknowledged_offline_message_is_not_delivered_twice() {
+    // Private messages sent to an offline user are stored by the server and
+    // handed over at the next login; the client acknowledges each one
+    // (MessageAcked, code 23) so the server drops it. Without that ack the
+    // same message arrives again on every login, which is what this pins.
+    let server = server_or_skip!();
+
+    // Register the recipient so the server will hold mail for them.
+    let mut recipient =
+        Client::with_settings(server.settings("e2e_ack_recipient", "pw"));
+    recipient.connect().expect("recipient connect");
+    assert!(recipient.login().expect("recipient login"));
+    drop(recipient);
+    // The dropped session's socket must actually be gone before the mail is
+    // sent, or the server delivers it live to a connection nobody is reading.
+    std::thread::sleep(Duration::from_secs(2));
+
+    let mut sender =
+        Client::with_settings(server.settings("e2e_ack_sender", "pw"));
+    sender.connect().expect("sender connect");
+    assert!(sender.login().expect("sender login"));
+    let body = "left while you were out";
+    sender
+        .send_private_message("e2e_ack_recipient", body)
+        .expect("send to an offline user");
+    std::thread::sleep(Duration::from_millis(500));
+
+    let received_once = |label: &str| -> bool {
+        let mut client =
+            Client::with_settings(server.settings("e2e_ack_recipient", "pw"));
+        client.connect().unwrap_or_else(|e| panic!("{label}: {e}"));
+        assert!(client.login().expect("recipient login"), "{label}");
+        let deadline = Instant::now() + Duration::from_secs(3);
+        let mut got = false;
+        while Instant::now() < deadline && !got {
+            got = client
+                .take_private_messages()
+                .iter()
+                .any(|m| m.message() == body);
+            if !got {
+                std::thread::sleep(Duration::from_millis(100));
+            }
+        }
+        std::thread::sleep(Duration::from_millis(500));
+        got
+    };
+
+    assert!(
+        received_once("first login"),
+        "the stored message should arrive at the next login"
+    );
+    assert!(
+        !received_once("second login"),
+        "an acknowledged message must not be delivered again"
+    );
+}
+
+#[test]
+fn a_wishlist_search_is_answered_like_any_other() {
+    let server = server_or_skip!();
+
+    let share_dir = unique_download_dir();
+    let content: Vec<u8> = (0..1024u32).map(|i| (i % 251) as u8).collect();
+    std::fs::write(share_dir.join("e2e_wishlist_talisman.bin"), &content)
+        .unwrap();
+
+    let (_sharer, searcher) = sharer_and_searcher(
+        &server,
+        &share_dir,
+        "e2e_wish_sharer",
+        "e2e_wish_seeker",
+    );
+
+    // A wish returns at once; its results accumulate under the query.
+    let query = "talisman";
+    searcher
+        .start_wishlist_search(query)
+        .expect("start a wishlist search");
+
+    let reply = reply_from(&searcher, query, "e2e_wish_sharer")
+        .expect("a wishlist search should be answered like a plain one");
+    assert!(
+        reply
+            .files
+            .iter()
+            .any(|f| f.name.contains("e2e_wishlist_talisman")),
+        "the reply should carry the matching file, got {:?}",
+        reply.files
+    );
+
+    // The server also announces how often it will accept one.
+    assert!(
+        searcher.wishlist_interval() > Duration::ZERO,
+        "the wishlist interval should be a real wait"
+    );
+
+    let _ = std::fs::remove_dir_all(share_dir);
+}