From 6f32a4e0902f1d860e80051abdc6e7f84058d8af Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Wed, 9 Sep 2026 18:38:34 +0000 Subject: [PATCH] fix: review findings on the protocol work Correctness: - A child that stopped reading could freeze the whole client: the relay writes under the context lock and the socket was blocking, so one stalled peer took every download, upload and API call with it. Child writes now have a deadline, and a child that hits it loses its slot. - Relaying searches to children was gated by the leaf's answer budget, so a parent forwarded a fraction of the stream while advertising itself as healthy. The budget bounds our own share scan, which is what it was for; the tree's stream goes down whole. - A dropped client kept its listener port for the life of the process, so a retried login advertised an ephemeral port nobody was told about. The listener is now stopped and the operations thread let go. - Excluded phrases were compared against a lowercased path without being lowercased themselves, so a capitalised phrase never matched. They are normalised where they arrive. - Losing our own operatorship wiped the room's whole operator roster, though we are still a member who can see it. - A brokered ConnectToPeer of type D was logged and dropped; the peer is now told, since we never dial out to adopt a child. - A peer request made with no server connection returned Ok and parked forever; it says NotConnected, as its docs always claimed. - An empty interest was refused locally but still sent, leaving the server's list and ours permanently apart. Simplification: - Deleted the excluded-phrase helper nothing called, and the field doc that promised behaviour it never had. - The four roster handlers and four standing handlers were eight constructors carrying constants their own message code already implies: one newtype each, 100 lines less. - One place registers a search; the phrase list is borrowed, not cloned per relayed search. --- soulseek-rs-lib/src/actor/server_actor.rs | 19 ++- soulseek-rs-lib/src/client/connection.rs | 53 ++++++-- soulseek-rs-lib/src/client/mod.rs | 75 ++++++----- soulseek-rs-lib/src/client/operations.rs | 69 +++++++--- soulseek-rs-lib/src/client/rooms.rs | 12 +- soulseek-rs-lib/src/client/search.rs | 11 +- soulseek-rs-lib/src/client/social.rs | 24 +++- soulseek-rs-lib/src/client/tests.rs | 54 +++++++- .../src/message/server/private_rooms.rs | 121 +++--------------- soulseek-rs-lib/src/peer/listen.rs | 6 + soulseek-rs-lib/tests/e2e.rs | 44 +++++++ 11 files changed, 294 insertions(+), 194 deletions(-) diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index ec0e3c3..0e5e4f2 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -427,16 +427,14 @@ impl ServerActor { handlers.register_handler(CantConnectToPeerHandler); handlers.register_handler(RoomMembersHandler); handlers.register_handler(RoomOperatorsHandler); - handlers.register_handler(RoomRosterChangeHandler::member_added()); - handlers.register_handler(RoomRosterChangeHandler::member_removed()); - handlers.register_handler(RoomRosterChangeHandler::operator_added()); - handlers.register_handler(RoomRosterChangeHandler::operator_removed()); - handlers.register_handler(OwnRoomStandingHandler::membership_granted()); - handlers.register_handler(OwnRoomStandingHandler::membership_revoked()); - handlers - .register_handler(OwnRoomStandingHandler::operatorship_granted()); - handlers - .register_handler(OwnRoomStandingHandler::operatorship_revoked()); + // Members added/removed (134/135), operators added/removed (143/144). + for code in [134, 135, 143, 144] { + handlers.register_handler(RoomRosterChangeHandler(code)); + } + // Our own membership (139/140) and operatorship (145/146). + for code in [139, 140, 145, 146] { + handlers.register_handler(OwnRoomStandingHandler(code)); + } handlers.register_handler(CantCreateRoomHandler); handlers.register_handler(RoomTickersHandler); handlers.register_handler(RoomTickerAddedHandler); @@ -954,6 +952,7 @@ impl ServerActor { ) { if let Err(e) = self.client_channel.send(ClientOperation::IncomingSearch { + from_parent: false, username, token, query, diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index 12f9d43..01bffb7 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -1,9 +1,10 @@ use super::{ - Arc, Client, ClientContext, ClientOperation, ConnectionType, DownloadPeer, - Listen, Peer, PeerRegistry, Receiver, Result, RwLock, RwLockExt, Sender, - ServerActor, ServerMessage, Shares, SoulseekRs, TcpStream, error, info, - mpsc, thread, trace, warn, + Arc, AtomicBool, Client, ClientContext, ClientOperation, ConnectionType, + DownloadPeer, Listen, Peer, PeerRegistry, Receiver, Result, RwLock, + RwLockExt, Sender, ServerActor, ServerMessage, Shares, SoulseekRs, + TcpStream, error, info, mpsc, thread, trace, warn, }; +use std::sync::atomic::Ordering; /// Ceiling on the wait for a login verdict. Generous enough for a slow server /// and a retrying connect, short enough that an unattended caller ends. @@ -96,6 +97,7 @@ impl Client { } else { None }; + self.listener_stopped = Arc::new(AtomicBool::new(false)); let mut server_actor = ServerActor::new( self.address.clone(), @@ -118,9 +120,16 @@ impl Client { let client_sender = listen_sender; let context = self.context.clone(); let own_username = self.username.clone(); + let stopped = self.listener_stopped.clone(); thread::spawn(move || { - Listen::serve(&listener, client_sender, context, own_username); + Listen::serve( + &listener, + client_sender, + context, + own_username, + stopped, + ); }); } @@ -147,14 +156,26 @@ impl Client { 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()); + let registry = match self.context.write_safe() { + Ok(mut ctx) => { + // Dropping our end of the operations channel is what lets its + // thread finish once the peers and the listener have gone. + ctx.operations = None; + ctx.peer_registry.clone() + } + Err(_) => None, + }; if let Some(registry) = registry { registry.stop_all(); } + // The listener thread is parked in accept(), so it only notices the + // flag when a connection arrives: dial it once ourselves. Without + // this the port stays bound for the life of the process, and the next + // client falls back to an ephemeral port nobody was told about. + self.listener_stopped.store(true, Ordering::Relaxed); + if let Some(port) = self.bound_port.take() { + let _ = TcpStream::connect(("127.0.0.1", port)); + } } /// Log in and wait for the server's verdict. @@ -355,7 +376,17 @@ impl Client { } } ConnectionType::D => { - error!("ConnectionType::D not implemented"); + // ponytail: we never dial out to adopt a child — children + // reach us through the listener. Tell the peer rather than + // leave it waiting; wire the outbound half if firewalled + // children turn out to matter. + if let (true, Some(token)) = (peer.brokered, peer.token) { + Self::report_cant_connect( + &client_context, + token, + &peer.username, + ); + } } } } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index c6ab69c..a9ea8d7 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -315,6 +315,10 @@ pub enum ClientOperation { username: String, token: u32, query: String, + /// True when it came down the tree from our parent, which has + /// already passed it to our children and is subject to the answer + /// budget. A search the server relayed is neither. + from_parent: bool, }, /// A peer queued one of our shared files; `requester_key` is the registry /// key of the peer actor — the peer's username. @@ -555,8 +559,8 @@ 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. + /// Phrases the server excludes from the search network (code 160). Files + /// whose path carries one are left out of the replies we send. excluded_search_phrases: Vec, /// Peers waiting for one of our upload slots. upload_queue: Vec, @@ -779,15 +783,12 @@ impl ClientContext { members, granted, } => { - // Our own standing is rendered from the same rosters, so a - // revocation must drop the room rather than leave a roster - // we are no longer part of looking current. - if !granted { - // Losing membership takes the operator roster with it: - // a room we cannot enter has no roster worth showing. - if *members { - self.private_room_members.remove(room); - } + // Losing membership takes both rosters with it: a room we + // cannot enter has none worth showing. Losing operatorship + // leaves us a member who still sees them, and the server + // narrates that roster change separately (code 144). + if !granted && *members { + self.private_room_members.remove(room); self.private_room_operators.remove(room); } } @@ -802,29 +803,45 @@ impl ClientContext { /// Interests are matched by the server case-insensitively, and Nicotine+ /// lowercases them before sending; do the same so two spellings of one /// interest cannot both be held. - pub fn add_own_interest(&mut self, item: &str, liked: bool) -> String { + /// Returns the stored spelling, or `None` for an empty item — which is + /// no interest, and must not be sent to the server either. + pub fn add_own_interest( + &mut self, + item: &str, + liked: bool, + ) -> Option { let item = item.trim().to_lowercase(); + if item.is_empty() { + return None; + } let list = if liked { &mut self.own_interests.likes } else { &mut self.own_interests.hates }; - if !item.is_empty() && !list.contains(&item) { + if !list.contains(&item) { list.push(item.clone()); } - item + Some(item) } - /// Forget one of our own interests. - pub fn remove_own_interest(&mut self, item: &str, liked: bool) -> String { + /// Forget one of our own interests. `None` for an empty item, as above. + pub fn remove_own_interest( + &mut self, + item: &str, + liked: bool, + ) -> Option { let item = item.trim().to_lowercase(); + if item.is_empty() { + return None; + } let list = if liked { &mut self.own_interests.likes } else { &mut self.own_interests.hates }; list.retain(|held| held != &item); - item + Some(item) } /// What we like and hate, as last set. @@ -1042,9 +1059,11 @@ impl ClientContext { self.user_interests.get(username).cloned() } - /// Record the phrases the server refuses to search for (code 160). + /// Record the phrases the server excludes (code 160), lowercased here so + /// the per-file check is a plain `contains` on a lowercased path. pub fn set_excluded_search_phrases(&mut self, phrases: Vec) { - self.excluded_search_phrases = phrases; + self.excluded_search_phrases = + phrases.iter().map(|p| p.to_lowercase()).collect(); } /// The phrases the server refuses to search for. @@ -1053,20 +1072,6 @@ impl ClientContext { 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) { @@ -1383,6 +1388,9 @@ pub struct Client { server_handle: Option>, context: Arc>, session: SessionWatch, + /// Tells the peer listener to stop, so a disconnected client releases the + /// port it bound. + listener_stopped: Arc, } impl Drop for Client { @@ -1416,6 +1424,7 @@ impl Client { shared_directories: settings.shared_directories, server_handle: None, session: SessionWatch::default(), + listener_stopped: Arc::new(AtomicBool::new(false)), } } diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 57b6072..7781709 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -16,6 +16,11 @@ const CONNECT_SWEEP_INTERVAL: Duration = Duration::from_secs(1); /// long queue is not pestered. const QUEUE_POSITION_INTERVAL: Duration = Duration::from_secs(300); +/// How long a write to a child may block. The relay runs under the client +/// context lock, so an unbounded write would freeze every other operation +/// behind one child that stopped reading. +const CHILD_WRITE_TIMEOUT: Duration = Duration::from_secs(5); + impl Client { pub(crate) fn listen_to_client_operations( reader: Receiver, @@ -582,30 +587,48 @@ impl Client { username, token, query, + from_parent, } => { // Every search we receive comes from the tree: from // our parent, or from the server acting as one. That // is what qualifies us to carry children at all — // Nicotine+ refuses them until something feeds it — // and passing the search down is the whole duty. - let capacity_changed = match client_context.write_safe() + let (capacity_changed, answer) = match client_context + .write_safe() { Ok(mut ctx) => { let was_fed = ctx.children.is_fed(); let before = ctx.children.len(); ctx.set_fed_by(true); - ctx.children - .broadcast_search(&username, token, &query); + if !from_parent { + // A parent's search was passed down where + // it arrived; this is the server's. + ctx.children.broadcast_search( + &username, token, &query, + ); + } + // Answering costs a scan of everything we + // share, so the budget applies to searches + // from the tree. The server's own are few. + let answer = !from_parent + || ctx.leaf.admit_search(Instant::now()); // A child whose socket has gone is dropped by // that write, which frees a slot the server // should hear about. - !was_fed || ctx.children.len() != before + ( + !was_fed || ctx.children.len() != before, + answer, + ) } - Err(_) => false, + Err(_) => (false, false), }; if capacity_changed { Self::announce_child_capacity(&client_context); } + if !answer { + continue; + } // Don't answer our own distributed search. if username == own_username { @@ -620,7 +643,7 @@ impl Client { ctx.has_free_upload_slot(), ctx.last_upload_speed, ctx.upload_queue.len() as u32, - &ctx.excluded_search_phrases(), + &ctx.excluded_search_phrases, ), Err(e) => { error!("[client] IncomingSearch read: {}", e); @@ -786,15 +809,22 @@ impl Client { &mut ctx, &branch, true, ); } - if ctx.leaf.is_parent(&parent, link) - && ctx.leaf.admit_search(Instant::now()) - && let Some(ops) = &ctx.operations - { - let _ = ops.send(ClientOperation::IncomingSearch { - username, - token, - query, - }); + if ctx.leaf.is_parent(&parent, link) { + // The tree's stream is not ours to throttle: every + // search goes down to our children. The budget + // below bounds only the share scan we do for our + // own reply. + ctx.children + .broadcast_search(&username, token, &query); + if let Some(ops) = &ctx.operations { + let _ = + ops.send(ClientOperation::IncomingSearch { + username, + token, + query, + from_parent: true, + }); + } } } ClientOperation::ParentClosed { parent, link } => { @@ -893,7 +923,14 @@ impl Client { if !ctx.children.has_room() { continue; } - if stream.set_nodelay(true).is_err() { + // A child that stops reading must not wedge us: the + // relay writes under the context lock, so the write + // is bounded and a child that hits it loses its slot. + if stream.set_nodelay(true).is_err() + || stream + .set_write_timeout(Some(CHILD_WRITE_TIMEOUT)) + .is_err() + { continue; } if username == own_username { diff --git a/soulseek-rs-lib/src/client/rooms.rs b/soulseek-rs-lib/src/client/rooms.rs index 8fee72c..d11da0c 100644 --- a/soulseek-rs-lib/src/client/rooms.rs +++ b/soulseek-rs-lib/src/client/rooms.rs @@ -263,13 +263,17 @@ impl Client { .send_to_peer(username, PeerMessage::SendMessage(request)); } } else { + // Without a server there is nobody to resolve the peer's address, + // so the request would sit in the queue forever: say so instead. + let handle = self + .server_handle + .as_ref() + .ok_or(SoulseekRs::NotConnected)?; self.context .write_safe()? .queue_peer_message(username, request); - if let Some(handle) = &self.server_handle { - let _ = handle - .send(ServerMessage::GetPeerAddress(username.to_string())); - } + let _ = handle + .send(ServerMessage::GetPeerAddress(username.to_string())); } Ok(()) } diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index a8a2b18..3b44d20 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -59,19 +59,10 @@ impl Client { fn send_search(&self, query: &str, wishlist: bool) -> Result<()> { info!("Searching for {}", query); + let token = self.register_search(query)?; let Some(handle) = &self.server_handle else { return Err(SoulseekRs::NotConnected); }; - let token = next_search_token(); - - self.context.write_safe()?.searches.insert( - query.to_string(), - Search { - token, - results: Vec::new(), - }, - ); - let query = query.to_string(); let _ = handle.send(if wishlist { ServerMessage::WishlistSearch { token, query } diff --git a/soulseek-rs-lib/src/client/social.rs b/soulseek-rs-lib/src/client/social.rs index 10d235c..89b1322 100644 --- a/soulseek-rs-lib/src/client/social.rs +++ b/soulseek-rs-lib/src/client/social.rs @@ -301,7 +301,11 @@ impl Client { /// # Errors /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. pub fn add_interest(&self, item: &str) -> Result<()> { - let item = self.context.write_safe()?.add_own_interest(item, true); + let Some(item) = + self.context.write_safe()?.add_own_interest(item, true) + else { + return Ok(()); // nothing to like or unlike + }; self.send_server_message(MessageFactory::build_add_thing_i_like(&item)) } @@ -310,7 +314,11 @@ impl Client { /// # Errors /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. pub fn remove_interest(&self, item: &str) -> Result<()> { - let item = self.context.write_safe()?.remove_own_interest(item, true); + let Some(item) = + self.context.write_safe()?.remove_own_interest(item, true) + else { + return Ok(()); // nothing to like or unlike + }; self.send_server_message(MessageFactory::build_remove_thing_i_like( &item, )) @@ -321,7 +329,11 @@ impl Client { /// # Errors /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. pub fn add_dislike(&self, item: &str) -> Result<()> { - let item = self.context.write_safe()?.add_own_interest(item, false); + let Some(item) = + self.context.write_safe()?.add_own_interest(item, false) + else { + return Ok(()); // nothing to like or unlike + }; self.send_server_message(MessageFactory::build_add_thing_i_hate(&item)) } @@ -330,7 +342,11 @@ impl Client { /// # Errors /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. pub fn remove_dislike(&self, item: &str) -> Result<()> { - let item = self.context.write_safe()?.remove_own_interest(item, false); + let Some(item) = + self.context.write_safe()?.remove_own_interest(item, false) + else { + return Ok(()); // nothing to like or unlike + }; self.send_server_message(MessageFactory::build_remove_thing_i_hate( &item, )) diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 7506e66..6be8609 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -1015,7 +1015,10 @@ fn revoked_membership_drops_the_room_we_can_no_longer_see() { } #[test] -fn losing_operatorship_keeps_the_membership_it_does_not_touch() { +fn losing_operatorship_keeps_the_roster_we_can_still_see() { + // Demotion does not blind us: we are still a member, and the room's other + // operators are still ours to show. The server narrates our own removal + // from that roster separately (code 144). let mut ctx = ClientContext::new(); ctx.apply_room_event(RoomEvent::PrivateMembers { room: "club".into(), @@ -1023,7 +1026,7 @@ fn losing_operatorship_keeps_the_membership_it_does_not_touch() { }); ctx.apply_room_event(RoomEvent::PrivateOperators { room: "club".into(), - users: vec!["alice".into()], + users: vec!["alice".into(), "bob".into()], }); ctx.apply_room_event(RoomEvent::OwnStandingChanged { @@ -1032,7 +1035,23 @@ fn losing_operatorship_keeps_the_membership_it_does_not_touch() { granted: false, }); assert_eq!(ctx.private_room_members("club"), vec!["alice".to_string()]); - assert!(ctx.private_room_operators("club").is_empty()); + assert_eq!( + ctx.private_room_operators("club"), + vec!["alice".to_string(), "bob".to_string()], + "the other operators are still there to show" + ); + + ctx.apply_room_event(RoomEvent::PrivateRosterChanged { + room: "club".into(), + username: "alice".into(), + members: false, + added: false, + }); + assert_eq!( + ctx.private_room_operators("club"), + vec!["bob".to_string()], + "only the demotion the server narrates removes us" + ); } #[test] @@ -1082,6 +1101,25 @@ fn a_file_the_server_excludes_is_left_out_of_a_reply() { "an exclusion the path does not carry changes nothing" ); + // The phrases are lowercased where they arrive, so a server that sends + // one capitalised still matches a path. + let mut ctx = ClientContext::new(); + ctx.set_excluded_search_phrases(vec!["SPAM".to_string()]); + assert!( + build_search_response( + &shares, + "me", + 1, + "xyzzy", + true, + 0, + 0, + &ctx.excluded_search_phrases(), + ) + .is_none(), + "a capitalised phrase from the server still withholds the file" + ); + let _ = std::fs::remove_dir_all(&dir); } @@ -1091,7 +1129,15 @@ fn our_own_interests_are_kept_lowercased_and_deduplicated() { // the session ends, so they are held here in one spelling to be sent // again next login. let mut ctx = ClientContext::new(); - assert_eq!(ctx.add_own_interest(" Krautrock ", true), "krautrock"); + assert_eq!( + ctx.add_own_interest(" Krautrock ", true).as_deref(), + Some("krautrock") + ); + assert_eq!( + ctx.add_own_interest(" ", true), + None, + "an empty interest is not stored, and is not sent either" + ); ctx.add_own_interest("KRAUTROCK", true); ctx.add_own_interest("Muzak", false); diff --git a/soulseek-rs-lib/src/message/server/private_rooms.rs b/soulseek-rs-lib/src/message/server/private_rooms.rs index 18d8aa6..7936014 100644 --- a/soulseek-rs-lib/src/message/server/private_rooms.rs +++ b/soulseek-rs-lib/src/message/server/private_rooms.rs @@ -57,55 +57,14 @@ impl MessageHandler for RoomOperatorsHandler { } } -/// One user added to (`added`) or removed from a private room's roster: -/// members (134/135) or operators (143/144). -pub struct RoomRosterChangeHandler { - code: u32, - members: bool, - added: bool, -} - -impl RoomRosterChangeHandler { - #[must_use] - pub const fn member_added() -> Self { - Self { - code: 134, - members: true, - added: true, - } - } - - #[must_use] - pub const fn member_removed() -> Self { - Self { - code: 135, - members: true, - added: false, - } - } - - #[must_use] - pub const fn operator_added() -> Self { - Self { - code: 143, - members: false, - added: true, - } - } - - #[must_use] - pub const fn operator_removed() -> Self { - Self { - code: 144, - members: false, - added: false, - } - } -} +/// One user added to or removed from a private room's roster, by the code it +/// arrived under: members added (134) or removed (135), operators added (143) +/// or removed (144). +pub struct RoomRosterChangeHandler(pub u32); impl MessageHandler for RoomRosterChangeHandler { fn get_code(&self) -> u32 { - self.code + self.0 } fn handle(&self, message: &mut Message, sender: Sender) { @@ -114,69 +73,27 @@ impl MessageHandler for RoomRosterChangeHandler { let _ = sender.send(ServerMessage::PrivateRoomRosterChanged { room, username, - members: self.members, - added: self.added, + members: matches!(self.0, 134 | 135), + added: matches!(self.0, 134 | 143), }); } } /// Our own standing in a private room changed: membership granted (139) or /// revoked (140), operatorship granted (145) or revoked (146). -pub struct OwnRoomStandingHandler { - code: u32, - members: bool, - granted: bool, -} - -impl OwnRoomStandingHandler { - #[must_use] - pub const fn membership_granted() -> Self { - Self { - code: 139, - members: true, - granted: true, - } - } - - #[must_use] - pub const fn membership_revoked() -> Self { - Self { - code: 140, - members: true, - granted: false, - } - } - - #[must_use] - pub const fn operatorship_granted() -> Self { - Self { - code: 145, - members: false, - granted: true, - } - } - - #[must_use] - pub const fn operatorship_revoked() -> Self { - Self { - code: 146, - members: false, - granted: false, - } - } -} +pub struct OwnRoomStandingHandler(pub u32); impl MessageHandler for OwnRoomStandingHandler { fn get_code(&self) -> u32 { - self.code + self.0 } fn handle(&self, message: &mut Message, sender: Sender) { let room = message.read_string(); let _ = sender.send(ServerMessage::OwnRoomStandingChanged { room, - members: self.members, - granted: self.granted, + members: matches!(self.0, 139 | 140), + granted: matches!(self.0, 139 | 145), }); } } @@ -243,10 +160,10 @@ mod tests { #[test] fn each_roster_change_reports_which_roster_and_which_way() { for (handler, members, added) in [ - (RoomRosterChangeHandler::member_added(), true, true), - (RoomRosterChangeHandler::member_removed(), true, false), - (RoomRosterChangeHandler::operator_added(), false, true), - (RoomRosterChangeHandler::operator_removed(), false, false), + (RoomRosterChangeHandler(134), true, true), + (RoomRosterChangeHandler(135), true, false), + (RoomRosterChangeHandler(143), false, true), + (RoomRosterChangeHandler(144), false, false), ] { let (tx, rx) = std::sync::mpsc::channel(); let mut message = framed(|m| { @@ -274,10 +191,10 @@ mod tests { #[test] fn our_own_standing_reports_the_room_and_the_change() { for (handler, members, granted) in [ - (OwnRoomStandingHandler::membership_granted(), true, true), - (OwnRoomStandingHandler::membership_revoked(), true, false), - (OwnRoomStandingHandler::operatorship_granted(), false, true), - (OwnRoomStandingHandler::operatorship_revoked(), false, false), + (OwnRoomStandingHandler(139), true, true), + (OwnRoomStandingHandler(140), true, false), + (OwnRoomStandingHandler(145), false, true), + (OwnRoomStandingHandler(146), false, false), ] { let (tx, rx) = std::sync::mpsc::channel(); let mut message = framed(|m| { diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 2726a8a..beb1f64 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -1,5 +1,6 @@ use std::io; use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::Sender; use std::sync::{Arc, RwLock}; use std::thread; @@ -422,6 +423,7 @@ impl Listen { client_sender: Sender, client_context: Arc>, own_username: String, + stopped: Arc, ) { info!("[listener] listening on {:?}", listener.local_addr()); @@ -433,6 +435,10 @@ impl Listen { let handshakes = Arc::new(Semaphore::new(MAX_HANDSHAKES)); for stream in listener.incoming() { + if stopped.load(Ordering::Relaxed) { + debug!("[listener] stopping, releasing the port"); + return; + } let Ok(stream) = stream else { error!( "[listener] Failed to accept connection: {}", diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index aa72f6b..800f3b6 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -5452,3 +5452,47 @@ fn privileges_can_be_handed_to_another_user() { "the gifted privileges should show up as time of our own" ); } + +#[test] +fn a_disconnected_client_gives_its_listener_port_back() { + // A client that goes out of scope must release the port it bound: the + // next one has to be able to advertise the port it was configured with, + // not fall back to an ephemeral one nobody was told about. No server is + // needed — binding the listener happens in connect(), before the dial. + let port = free_port().expect("a port to hold"); + let settings = |port: u16| ClientSettings { + username: "e2e_port_release".to_string(), + password: "pw".to_string(), + server_address: PeerAddress::new("127.0.0.1".to_string(), 1), + enable_listen: true, + listen_port: port, + shared_directories: Vec::new(), + accept_children: false, + version: ClientVersion::default(), + }; + + let mut first = Client::with_settings(settings(port)); + first.connect().expect("bind the listener"); + assert_eq!(first.listen_port(), Some(port)); + drop(first); + + // The listener thread notices the stop the moment it is woken, which the + // disconnect does by dialling it; give it a beat to unwind. + let deadline = Instant::now() + Duration::from_secs(5); + let mut second_port = None; + while Instant::now() < deadline && second_port != Some(port) { + let mut second = Client::with_settings(settings(port)); + second.connect().expect("bind again"); + second_port = second.listen_port(); + if second_port == Some(port) { + break; + } + drop(second); + std::thread::sleep(Duration::from_millis(200)); + } + assert_eq!( + second_port, + Some(port), + "the port should be free again once the first client is dropped" + ); +}