From 1976b497929a5dd6f66d141d1951dea2285347c5 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 7 Sep 2026 23:56:33 +0200 Subject: [PATCH 1/4] feat(lib): decode distributed frames and announce a parentless stance The distributed search network carries searches from parent to child over connections whose frames have a one-byte code; the library had no reader for them. This adds the codec for Search, BranchLevel, BranchRoot and the embedded envelope, with the identifier check that keeps malformed frames from being answered, and completes the post-login stance Nicotine+ sends: HaveNoParent, BranchRoot (ourselves), BranchLevel 0 and AcceptChildren false, in place of HaveNoParent alone. --- soulseek-rs-lib/src/actor/server_actor.rs | 25 ++- .../src/actor/server_actor/tests.rs | 18 +- soulseek-rs-lib/src/message/distributed.rs | 160 ++++++++++++++++++ soulseek-rs-lib/src/message/mod.rs | 1 + .../src/message/server/message_factory.rs | 28 ++- 5 files changed, 221 insertions(+), 11 deletions(-) create mode 100644 soulseek-rs-lib/src/message/distributed.rs diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 1f2620b..c009101 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -175,32 +175,46 @@ pub struct ServerActor { shared_folder_count: u32, shared_file_count: u32, session: SessionWatch, + /// The name we logged in as, once a login has been asked for. + username: String, } /// The messages a client sends right after a successful login: its shared-file -/// counts, distributed-network opt-out, online status, and (when listening) the -/// port peers should connect to. Kept as a free function so it can be tested -/// without a live connection. +/// counts, its distributed-network stance, online status, and (when listening) +/// the port peers should connect to. Kept as a free function so it can be +/// tested without a live connection. fn post_login_messages( enable_listen: bool, listen_port: u16, shared_folders: u32, shared_files: u32, + own_username: &str, ) -> Vec { let mut messages = vec![ MessageFactory::build_shared_folders_message( shared_folders, shared_files, ), - MessageFactory::build_no_parent_message(), MessageFactory::build_set_status_message(2), ]; + messages.splice(1..1, parentless_stance(own_username)); if enable_listen { messages.push(MessageFactory::build_set_wait_port_message(listen_port)); } messages } +/// What a leaf tells the server while it has no parent: it is its own branch +/// root at level 0 and takes no children. Nicotine+ sends the same four. +fn parentless_stance(own_username: &str) -> Vec { + vec![ + MessageFactory::build_have_no_parent(true), + MessageFactory::build_branch_root(own_username), + MessageFactory::build_branch_level(0), + MessageFactory::build_accept_children(false), + ] +} + impl ServerActor { #[must_use] pub fn new( @@ -228,6 +242,7 @@ impl ServerActor { shared_folder_count, shared_file_count, session: SessionWatch::default(), + username: String::new(), } } @@ -550,6 +565,7 @@ impl ServerActor { self.listen_port, self.shared_folder_count, self.shared_file_count, + &self.username, ) { self.send_message(msg); } @@ -621,6 +637,7 @@ impl ServerActor { version: ClientVersion, response: std::sync::mpsc::Sender>, ) { + self.username.clone_from(&username); if self.stream.is_none() && !self.initiate_connection() { let _ = response.send(Err(SoulseekRs::NotConnected)); return; diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs index bf73bbe..e19ce1f 100644 --- a/soulseek-rs-lib/src/actor/server_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -193,10 +193,18 @@ fn a_successful_login_marks_the_session_live_again() { #[test] fn post_login_messages_carry_counts_and_conditional_wait_port() { - let messages = post_login_messages(true, 4321, 3, 7); + let messages = post_login_messages(true, 4321, 3, 7, "me"); let codes: Vec = messages.iter().map(code_of).collect(); - // SharedFolders, HaveNoParent, SetStatus, SetWaitPort. - assert_eq!(codes, vec![35, 71, 28, 2]); + // SharedFolders; the distributed stance HaveNoParent, BranchRoot, + // BranchLevel, AcceptChildren; SetStatus; SetWaitPort. + assert_eq!(codes, vec![35, 71, 127, 126, 100, 28, 2]); + // A leaf with no parent is its own branch root at level 0, and it + // declines children. + let mut root = messages[2].clone(); + root.set_pointer(4); + assert_eq!(root.read_string(), "me"); + assert_eq!(&messages[3].get_data()[4..8], &0u32.to_le_bytes()); + assert_eq!(messages[4].get_data()[4], 0); // The SharedFolders message (code 35) carries the real counts. let shared = messages[0].get_data(); @@ -204,7 +212,7 @@ fn post_login_messages_carry_counts_and_conditional_wait_port() { assert_eq!(u32::from_le_bytes(shared[8..12].try_into().unwrap()), 7); // Not listening omits SetWaitPort (code 2). - let no_listen = post_login_messages(false, 4321, 3, 7); + let no_listen = post_login_messages(false, 4321, 3, 7, "me"); let codes: Vec = no_listen.iter().map(code_of).collect(); - assert_eq!(codes, vec![35, 71, 28]); + assert_eq!(codes, vec![35, 71, 127, 126, 100, 28]); } diff --git a/soulseek-rs-lib/src/message/distributed.rs b/soulseek-rs-lib/src/message/distributed.rs new file mode 100644 index 0000000..64c1310 --- /dev/null +++ b/soulseek-rs-lib/src/message/distributed.rs @@ -0,0 +1,160 @@ +//! The distributed search network's wire format. +//! +//! Frames carry a one-byte code, unlike the four-byte server and peer codes, +//! and a leaf only ever reads them: 3 Search, 4 BranchLevel, 5 BranchRoot, +//! 93 EmbeddedMessage. + +use crate::message::Message; + +/// A search relayed through the tree carries this in its first field; anything +/// else is a frame we do not understand and must not answer. +const SEARCH_IDENTIFIER: u32 = 49; + +/// What a parent sends a leaf. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Distributed { + Search { + username: String, + token: u32, + query: String, + }, + BranchLevel(i32), + BranchRoot(String), +} + +/// Decode one framed distributed message (length prefix included), or `None` +/// for a ping, an unknown code or a truncated frame. +#[must_use] +pub fn parse(message: &mut Message) -> Option { + if message.get_size() < 5 { + return None; + } + message.set_pointer(5); + parse_body(message, message.get_init_code()) +} + +fn parse_body(message: &mut Message, code: u8) -> Option { + match code { + 3 => { + if message.read_int32() != SEARCH_IDENTIFIER { + return None; + } + let username = message.read_string(); + let token = message.read_int32(); + let query = message.read_string(); + (!username.is_empty() && !query.is_empty()).then_some( + Distributed::Search { + username, + token, + query, + }, + ) + } + 4 => Some(Distributed::BranchLevel(message.read_int32() as i32)), + 5 => Some(Distributed::BranchRoot(message.read_string())), + // Older branch roots wrap a search in a code-93 envelope. + 93 => { + let inner = message.read_int8(); + parse_body(message, inner) + } + _ => None, + } +} + +/// A distributed frame with the one-byte `code`: what a parent (or a test +/// standing in for one) sends down the tree. +fn framed(code: u8, body: impl FnOnce(&mut Message)) -> Message { + let mut message = Message::new(); + message.write_int8(code); + body(&mut message); + message +} + +#[must_use] +pub fn build_search(username: &str, token: u32, query: &str) -> Message { + framed(3, |m| { + m.write_int32(SEARCH_IDENTIFIER) + .write_string(username) + .write_int32(token) + .write_string(query); + }) +} + +#[must_use] +pub fn build_branch_level(level: i32) -> Message { + framed(4, |m| { + m.write_int32(level as u32); + }) +} + +#[must_use] +pub fn build_branch_root(root: &str) -> Message { + framed(5, |m| { + m.write_string(root); + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_distributed_search_carries_who_asked_and_what_for() { + let frame = build_search("seeker", 77, "aphex twin"); + let mut message = Message::new_with_data(frame.get_buffer()); + assert_eq!( + parse(&mut message), + Some(Distributed::Search { + username: "seeker".to_string(), + token: 77, + query: "aphex twin".to_string(), + }) + ); + } + + #[test] + fn a_search_with_the_wrong_identifier_is_dropped() { + let mut body = Message::new(); + body.write_int8(3) + .write_int32(7) + .write_string("seeker") + .write_int32(1) + .write_string("q"); + let mut message = Message::new_with_data(body.get_buffer()); + assert_eq!(parse(&mut message), None); + } + + #[test] + fn branch_level_and_root_frames_parse() { + let mut level = + Message::new_with_data(build_branch_level(2).get_buffer()); + assert_eq!(parse(&mut level), Some(Distributed::BranchLevel(2))); + let mut root = + Message::new_with_data(build_branch_root("rooty").get_buffer()); + assert_eq!( + parse(&mut root), + Some(Distributed::BranchRoot("rooty".to_string())) + ); + } + + #[test] + fn an_embedded_search_unwraps_to_the_search_itself() { + let inner = build_search("seeker", 5, "q").get_data(); + let mut body = Message::new(); + body.write_int8(93).write_raw_bytes(inner); + let mut message = Message::new_with_data(body.get_buffer()); + assert!(matches!( + parse(&mut message), + Some(Distributed::Search { token: 5, .. }) + )); + } + + #[test] + fn unknown_or_truncated_frames_parse_to_nothing() { + let mut ping = + Message::new_with_data(Message::new().write_int8(0).get_buffer()); + assert_eq!(parse(&mut ping), None); + let mut short = Message::new_with_data(vec![1, 0, 0, 0, 3]); + assert_eq!(parse(&mut short), None); + } +} diff --git a/soulseek-rs-lib/src/message/mod.rs b/soulseek-rs-lib/src/message/mod.rs index 28bdf53..03d70d9 100644 --- a/soulseek-rs-lib/src/message/mod.rs +++ b/soulseek-rs-lib/src/message/mod.rs @@ -1,5 +1,6 @@ pub use crate::{debug, error, info, trace, warn}; +pub mod distributed; pub mod handlers; mod message_reader; pub mod peer; diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 87fd62c..620508a 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -127,8 +127,32 @@ impl MessageFactory { .clone() } #[must_use] - pub fn build_no_parent_message() -> Message { - Message::new().write_int32(71).write_bool(true).clone() + /// Whether we lack a parent in the distributed network (code 71); the + /// server offers PossibleParents while we say so. + pub fn build_have_no_parent(no_parent: bool) -> Message { + Message::new().write_int32(71).write_bool(no_parent).clone() + } + + /// The root of the branch we hang from (code 127): ourselves while we + /// have no parent. + #[must_use] + pub fn build_branch_root(username: &str) -> Message { + Message::new() + .write_int32(127) + .write_string(username) + .clone() + } + + /// How deep in the tree we sit (code 126): 0 without a parent. + #[must_use] + pub fn build_branch_level(level: u32) -> Message { + Message::new().write_int32(126).write_int32(level).clone() + } + + /// Whether we take children in the distributed network (code 100). + #[must_use] + pub fn build_accept_children(accept: bool) -> Message { + Message::new().write_int32(100).write_bool(accept).clone() } #[must_use] pub fn build_set_wait_port_message(port: u16) -> Message { From 7d7b8fe98c212b96254d8f5632279641de000abb Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 00:16:02 +0200 Subject: [PATCH 2/4] feat(lib): join the distributed search network as a leaf The server hands every client a list of possible parents and expects it to hang from one so searches also travel peer to peer. Until now we told the server we were a lone root and left it at that, which the network tolerates but does not favour. A leaf now dials each PossibleParents candidate over a D connection, adopts the first that passes a search down after stating its branch, announces the new branch root and level to the server, and answers those searches the same way as ones from the server. Losing the parent or a ResetDistributed puts it back to a parentless stance. The leaf takes no children. The Leaf state machine is pure so its rules are unit-tested without sockets; the e2e test drives a real client from a mock parent, since soulfind never hands out parents. --- soulseek-rs-lib/src/actor/server_actor.rs | 28 +- soulseek-rs-lib/src/client/connection.rs | 1 + soulseek-rs-lib/src/client/distributed.rs | 408 ++++++++++++++++++ soulseek-rs-lib/src/client/mod.rs | 32 ++ soulseek-rs-lib/src/client/operations.rs | 79 ++++ soulseek-rs-lib/src/message/distributed.rs | 21 +- .../src/message/server/distributed.rs | 127 ++++++ soulseek-rs-lib/src/message/server/mod.rs | 4 + soulseek-rs-lib/tests/e2e.rs | 150 +++++++ 9 files changed, 840 insertions(+), 10 deletions(-) create mode 100644 soulseek-rs-lib/src/client/distributed.rs create mode 100644 soulseek-rs-lib/src/message/server/distributed.rs diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index c009101..74cdce1 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -4,6 +4,7 @@ use crate::dispatcher::MessageDispatcher; use crate::message::server::AdminMessageHandler; use crate::message::server::CheckPrivilegesHandler; use crate::message::server::ConnectToPeerHandler; +use crate::message::server::EmbeddedMessageHandler; use crate::message::server::ExcludedSearchPhrasesHandler; use crate::message::server::FileSearchHandler; use crate::message::server::GetPeerAddressHandler; @@ -14,8 +15,10 @@ use crate::message::server::MessageFactory; use crate::message::server::MessageUser; use crate::message::server::ParentMinSpeedHandler; use crate::message::server::ParentSpeedRatioHandler; +use crate::message::server::PossibleParentsHandler; use crate::message::server::PrivilegedUsersHandler; use crate::message::server::ReloggedHandler; +use crate::message::server::ResetDistributedHandler; use crate::message::server::SayChatroomHandler; use crate::message::server::UserJoinedRoomHandler; use crate::message::server::UserLeftRoomHandler; @@ -156,6 +159,11 @@ pub enum ServerMessage { room: String, username: String, }, + /// Peers the server suggests as distributed-network parents: username, + /// host, port. + PossibleParents(Vec<(String, String, u16)>), + /// The server asks us to drop our parent and start over. + ResetDistributed, } pub struct ServerActor { @@ -205,14 +213,9 @@ fn post_login_messages( } /// What a leaf tells the server while it has no parent: it is its own branch -/// root at level 0 and takes no children. Nicotine+ sends the same four. +/// root at level 0. fn parentless_stance(own_username: &str) -> Vec { - vec![ - MessageFactory::build_have_no_parent(true), - MessageFactory::build_branch_root(own_username), - MessageFactory::build_branch_level(0), - MessageFactory::build_accept_children(false), - ] + crate::message::distributed::stance(own_username, 0, false) } impl ServerActor { @@ -306,6 +309,9 @@ impl ServerActor { handlers.register_handler(LoginHandler); handlers.register_handler(ReloggedHandler); handlers.register_handler(AdminMessageHandler); + handlers.register_handler(PossibleParentsHandler); + handlers.register_handler(ResetDistributedHandler); + handlers.register_handler(EmbeddedMessageHandler); handlers.register_handler(RoomListHandler); handlers.register_handler(GetUserStatusHandler); handlers.register_handler(WatchUserHandler); @@ -517,6 +523,14 @@ impl ServerActor { /// length; there is no behaviour here beyond dispatch. fn handle_standing_message(&mut self, message: ServerMessage) { match message { + ServerMessage::PossibleParents(candidates) => { + self.forward_to_client(ClientOperation::PossibleParents( + candidates, + )); + } + ServerMessage::ResetDistributed => { + self.forward_to_client(ClientOperation::ResetDistributed); + } ServerMessage::WishlistSearch { token, query } => { self.queue_message(MessageFactory::build_wishlist_search( token, &query, diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index 8421ce9..e6d0436 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -57,6 +57,7 @@ impl Client { ctx.max_peers.clone(), ); ctx.peer_registry = Some(peer_registry); + ctx.operations = Some(sender.clone()); let listen_sender = sender.clone(); diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs new file mode 100644 index 0000000..8181050 --- /dev/null +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -0,0 +1,408 @@ +//! Our place in the distributed search network, as a leaf: which parent we +//! hang from, which candidates we are still sounding out, and what to tell the +//! server when that changes. + +use super::{Client, ClientOperation, Result, ServerMessage}; +use crate::message::Message; +use crate::message::distributed::{self, Distributed}; +use crate::message::server::MessageFactory; +use crate::peer::ConnectionType; +use crate::utils::lock::RwLockExt; +use std::collections::HashMap; +use std::io::{self, Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::Sender; +use std::time::Duration; + +/// A candidate that has not answered the dial in this long is not a parent. +const LINK_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// How often an idle link looks at its cancel flag. +const CANCEL_POLL: Duration = Duration::from_secs(1); +/// A distributed frame is a search or a branch fact; anything bigger is not +/// one of ours. +const MAX_FRAME: usize = 64 * 1024; + +/// Where we sit in the tree, as told to the server. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Branch { + pub root: String, + pub level: u32, +} + +impl Branch { + #[must_use] + pub fn parentless(own: &str) -> Self { + Self::under(own, 0) + } + + #[must_use] + pub fn under(root: &str, level: u32) -> Self { + Self { + root: root.to_string(), + level, + } + } +} + +#[derive(Default)] +struct Candidate { + level: Option, + root: Option, + cancel: Arc, +} + +/// A candidate to dial, and the flag that tells its link to stop. +pub struct Dial { + pub username: String, + pub host: String, + pub port: u16, + pub cancel: Arc, +} + +/// The leaf's view: at most one parent, adopted from the candidates the +/// server offered, the way Nicotine+ does it. Every method is a pure state +/// change so the rule is testable without a socket. +pub struct Leaf { + own: String, + parent: Option, + branch: Branch, + candidates: HashMap, +} + +impl Leaf { + #[must_use] + pub fn new(own: &str) -> Self { + Self { + own: own.to_string(), + parent: None, + branch: Branch::parentless(own), + candidates: HashMap::new(), + } + } + + #[must_use] + pub fn branch(&self) -> Branch { + self.branch.clone() + } + + #[must_use] + pub fn is_parent(&self, username: &str) -> bool { + self.parent.as_deref() == Some(username) + } + + /// New candidates: dial them all, unless we already hang from a parent. + /// Candidates still being sounded out from an earlier batch are dropped. + pub fn consider( + &mut self, + candidates: Vec<(String, String, u16)>, + ) -> Vec { + if self.parent.is_some() { + return Vec::new(); + } + self.cancel_candidates(); + candidates + .into_iter() + .map(|(username, host, port)| { + let cancel = Arc::new(AtomicBool::new(false)); + self.candidates.insert( + username.clone(), + Candidate { + cancel: cancel.clone(), + ..Candidate::default() + }, + ); + Dial { + username, + host, + port, + cancel, + } + }) + .collect() + } + + /// A branch root does not always say so separately: level 0 is its own + /// root. + pub fn branch_level(&mut self, username: &str, level: i32) { + let Some(candidate) = self.candidates.get_mut(username) else { + return; + }; + candidate.level = Some(level); + if level == 0 && candidate.root.is_none() { + candidate.root = Some(username.to_string()); + } + } + + pub fn branch_root(&mut self, username: &str, root: &str) { + if let Some(candidate) = self.candidates.get_mut(username) { + candidate.root = Some(root.to_string()); + } + } + + /// A search came down from `username`. The first one from a candidate that + /// has told us its place adopts it; the stance to announce comes back. + pub fn search_from(&mut self, username: &str) -> Option { + if self.parent.is_some() { + return None; + } + let candidate = self.candidates.get(username)?; + let (Some(level), Some(root)) = + (candidate.level, candidate.root.clone()) + else { + return None; + }; + let parent = self.candidates.remove(username)?; + self.cancel_candidates(); + self.candidates.insert(username.to_string(), parent); + self.parent = Some(username.to_string()); + self.branch = + Branch::under(&root, u32::try_from(level + 1).unwrap_or(1)); + Some(self.branch.clone()) + } + + /// The link to `username` closed. True when that was the parent, which + /// puts us back to parentless. + pub fn closed(&mut self, username: &str) -> bool { + self.candidates.remove(username); + if !self.is_parent(username) { + return false; + } + self.parent = None; + self.branch = Branch::parentless(&self.own); + true + } + + pub fn reset(&mut self) { + self.cancel_candidates(); + self.parent = None; + self.branch = Branch::parentless(&self.own); + } + + fn cancel_candidates(&mut self) { + for candidate in self.candidates.values() { + candidate.cancel.store(true, Ordering::Relaxed); + } + self.candidates.clear(); + } +} + +/// Tell the server where we sit now. +pub fn announce( + server: Option<&Sender>, + branch: &Branch, + has_parent: bool, +) { + let Some(server) = server else { + return; + }; + for message in distributed::stance(&branch.root, branch.level, has_parent) { + let _ = server.send(ServerMessage::SendMessage(message)); + } +} + +/// Dial a parent candidate on its own thread: a `D` PeerInit, then whatever it +/// sends comes back as operations until the link closes or is cancelled. +pub fn spawn_link(dial: Dial, own: String, ops: Sender) { + let _ = std::thread::Builder::new() + .name("soulseek-parent".to_string()) + .spawn(move || { + let parent = dial.username.clone(); + if let Err(e) = link(&dial, &own, &ops) + && !dial.cancel.load(Ordering::Relaxed) + { + crate::debug!("[distributed] link to {} ended: {}", parent, e); + let _ = ops.send(ClientOperation::ParentClosed { parent }); + } + }); +} + +fn link( + dial: &Dial, + own: &str, + ops: &Sender, +) -> io::Result<()> { + let addr = (dial.host.as_str(), dial.port) + .to_socket_addrs()? + .next() + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "no address") + })?; + let mut stream = TcpStream::connect_timeout(&addr, LINK_CONNECT_TIMEOUT)?; + stream.set_read_timeout(Some(CANCEL_POLL))?; + stream.write_all( + &MessageFactory::build_peer_init_message(own, ConnectionType::D, 0) + .get_buffer(), + )?; + loop { + if dial.cancel.load(Ordering::Relaxed) { + return Ok(()); + } + let Some(mut frame) = read_frame(&mut stream)? else { + continue; + }; + let parent = dial.username.clone(); + let operation = match distributed::parse(&mut frame) { + Some(Distributed::BranchLevel(level)) => { + ClientOperation::ParentBranchLevel { parent, level } + } + Some(Distributed::BranchRoot(root)) => { + ClientOperation::ParentBranchRoot { parent, root } + } + Some(Distributed::Search { + username, + token, + query, + }) => ClientOperation::ParentSearch { + parent, + username, + token, + query, + }, + None => continue, + }; + if ops.send(operation).is_err() { + return Ok(()); + } + } +} + +/// One framed message, or `None` when the read timed out with nothing sent. +fn read_frame(stream: &mut TcpStream) -> io::Result> { + let mut len = [0u8; 4]; + match stream.read_exact(&mut len) { + Ok(()) => {} + Err(e) + if matches!( + e.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + return Ok(None); + } + Err(e) => return Err(e), + } + let size = u32::from_le_bytes(len) as usize; + if size > MAX_FRAME { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "oversized frame", + )); + } + let mut data = len.to_vec(); + data.resize(4 + size, 0); + stream.read_exact(&mut data[4..])?; + Ok(Some(Message::new_with_data(data))) +} + +impl Client { + /// Offer parent candidates as the server does with PossibleParents: + /// `(username, host, port)`. The first candidate to send a search after + /// telling us its branch becomes our parent. + /// + /// # Errors + /// Returns [`crate::SoulseekRs::NotConnected`] before [`Client::connect`]. + pub fn consider_parents( + &self, + candidates: Vec<(String, String, u16)>, + ) -> Result<()> { + let ops = self + .context + .read_safe()? + .operations + .clone() + .ok_or(crate::SoulseekRs::NotConnected)?; + ops.send(ClientOperation::PossibleParents(candidates)) + .map_err(|_| crate::SoulseekRs::NotConnected) + } + + /// Where we sit in the distributed search network: the branch root's name + /// and our level, which is ourselves at 0 while we have no parent. + #[must_use] + pub fn distributed_branch(&self) -> (String, u32) { + self.context.read_safe().map_or_else( + |_| (self.username.clone(), 0), + |ctx| { + let branch = ctx.leaf.branch(); + (branch.root, branch.level) + }, + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn leaf() -> Leaf { + Leaf::new("me") + } + + #[test] + fn a_leaf_without_a_parent_is_its_own_root_at_level_zero() { + assert_eq!(leaf().branch(), Branch::parentless("me")); + } + + #[test] + fn candidates_are_dialled_only_while_there_is_no_parent() { + let mut leaf = leaf(); + let dial = leaf.consider(vec![candidate("alice"), candidate("bob")]); + assert_eq!(dial.len(), 2); + leaf.branch_level("alice", 1); + leaf.branch_root("alice", "rooty"); + assert!( + leaf.search_from("alice").is_some(), + "alice becomes the parent" + ); + assert!(leaf.consider(vec![candidate("carol")]).is_empty()); + } + + #[test] + fn the_first_search_from_a_candidate_with_branch_info_adopts_it() { + let mut leaf = leaf(); + leaf.consider(vec![candidate("alice"), candidate("bob")]); + assert_eq!(leaf.search_from("alice"), None, "no branch info yet"); + leaf.branch_level("alice", 3); + assert_eq!(leaf.search_from("alice"), None, "level without root"); + leaf.branch_root("alice", "rooty"); + assert_eq!( + leaf.search_from("alice"), + Some(Branch::under("rooty", 4)), + "level 3 parent puts us at level 4" + ); + assert_eq!(leaf.branch(), Branch::under("rooty", 4)); + assert_eq!(leaf.search_from("alice"), None, "already the parent"); + assert_eq!(leaf.search_from("bob"), None, "a loser is ignored"); + } + + #[test] + fn a_level_zero_candidate_is_its_own_root() { + let mut leaf = leaf(); + leaf.consider(vec![candidate("alice")]); + leaf.branch_level("alice", 0); + assert_eq!(leaf.search_from("alice"), Some(Branch::under("alice", 1))); + } + + #[test] + fn losing_the_parent_or_a_reset_returns_to_parentless() { + let mut leaf = leaf(); + leaf.consider(vec![candidate("alice")]); + leaf.branch_level("alice", 0); + leaf.search_from("alice"); + assert!(!leaf.closed("bob"), "a stranger closing changes nothing"); + assert!(leaf.closed("alice")); + assert_eq!(leaf.branch(), Branch::parentless("me")); + + leaf.consider(vec![candidate("alice")]); + leaf.branch_level("alice", 0); + leaf.search_from("alice"); + leaf.reset(); + assert_eq!(leaf.branch(), Branch::parentless("me")); + assert!(!leaf.consider(vec![candidate("bob")]).is_empty()); + } + + fn candidate(name: &str) -> (String, String, u16) { + (name.to_string(), "127.0.0.1".to_string(), 2234) + } +} diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index e0d71ad..9b80e54 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -334,11 +334,40 @@ pub enum ClientOperation { requester_key: String, filename: String, }, + /// Distributed-network parent candidates, from the server or a host. + PossibleParents(Vec<(String, String, u16)>), + /// The server told us to drop our parent. + ResetDistributed, + /// A parent candidate told us how deep it sits. + ParentBranchLevel { + parent: String, + level: i32, + }, + /// A parent candidate told us whose branch it is on. + ParentBranchRoot { + parent: String, + root: String, + }, + /// A search came down the tree from `parent`. + ParentSearch { + parent: String, + username: String, + token: u32, + query: String, + }, + /// The link to a parent or candidate is gone. + ParentClosed { + parent: String, + }, } pub struct ClientContext { pub peer_registry: Option, pub downloads: DownloadStore, server_sender: Option>, + /// Where anything outside the operations loop posts operations. + operations: Option>, + /// Our place in the distributed search network. + leaf: distributed::Leaf, searches: HashMap, private_messages: Vec, /// Correlation tokens for server-brokered (firewalled) connections, mapping @@ -452,6 +481,8 @@ impl ClientContext { Self { peer_registry: None, server_sender: None, + operations: None, + leaf: distributed::Leaf::new(""), searches: HashMap::new(), private_messages: Vec::new(), pending_connect_tokens: HashMap::new(), @@ -982,6 +1013,7 @@ impl Client { } mod connection; +mod distributed; mod downloads; mod operations; mod rooms; diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 099e3a5..da84f75 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -17,6 +17,9 @@ impl Client { own_username: String, ) { thread::spawn(move || { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.leaf = super::distributed::Leaf::new(&own_username); + } let mut last_sweep = Instant::now(); loop { let next = reader.recv_timeout(CONNECT_SWEEP_INTERVAL); @@ -696,6 +699,82 @@ impl Client { ); } } + ClientOperation::PossibleParents(candidates) => { + let (dials, ops) = match client_context.write_safe() { + Ok(mut ctx) => ( + ctx.leaf.consider(candidates), + ctx.operations.clone(), + ), + Err(_) => continue, + }; + let Some(ops) = ops else { continue }; + for dial in dials { + super::distributed::spawn_link( + dial, + own_username.clone(), + ops.clone(), + ); + } + } + ClientOperation::ParentBranchLevel { parent, level } => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.leaf.branch_level(&parent, level); + } + } + ClientOperation::ParentBranchRoot { parent, root } => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.leaf.branch_root(&parent, &root); + } + } + ClientOperation::ParentSearch { + parent, + username, + token, + query, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = ctx.leaf.search_from(&parent) { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); + } + if ctx.leaf.is_parent(&parent) + && let Some(ops) = &ctx.operations + { + let _ = ops.send(ClientOperation::IncomingSearch { + username, + token, + query, + }); + } + } + ClientOperation::ParentClosed { parent } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if ctx.leaf.closed(&parent) { + super::distributed::announce( + ctx.server_sender.as_ref(), + &ctx.leaf.branch(), + false, + ); + } + } + ClientOperation::ResetDistributed => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + ctx.leaf.reset(); + super::distributed::announce( + ctx.server_sender.as_ref(), + &ctx.leaf.branch(), + false, + ); + } ClientOperation::PrivilegedUsers(users) => { if let Ok(mut ctx) = client_context.write_safe() { debug!( diff --git a/soulseek-rs-lib/src/message/distributed.rs b/soulseek-rs-lib/src/message/distributed.rs index 64c1310..756219e 100644 --- a/soulseek-rs-lib/src/message/distributed.rs +++ b/soulseek-rs-lib/src/message/distributed.rs @@ -5,6 +5,7 @@ //! 93 EmbeddedMessage. use crate::message::Message; +use crate::message::server::MessageFactory; /// A search relayed through the tree carries this in its first field; anything /// else is a frame we do not understand and must not answer. @@ -30,10 +31,12 @@ pub fn parse(message: &mut Message) -> Option { return None; } message.set_pointer(5); - parse_body(message, message.get_init_code()) + parse_at(message, message.get_init_code()) } -fn parse_body(message: &mut Message, code: u8) -> Option { +/// Decode the body that follows a one-byte `code`, the pointer already past it. +#[must_use] +pub fn parse_at(message: &mut Message, code: u8) -> Option { match code { 3 => { if message.read_int32() != SEARCH_IDENTIFIER { @@ -55,7 +58,7 @@ fn parse_body(message: &mut Message, code: u8) -> Option { // Older branch roots wrap a search in a code-93 envelope. 93 => { let inner = message.read_int8(); - parse_body(message, inner) + parse_at(message, inner) } _ => None, } @@ -87,6 +90,18 @@ pub fn build_branch_level(level: i32) -> Message { }) } +/// What we tell the server about our place in the tree: the four messages +/// Nicotine+ sends together whenever it changes. +#[must_use] +pub fn stance(root: &str, level: u32, has_parent: bool) -> Vec { + vec![ + MessageFactory::build_have_no_parent(!has_parent), + MessageFactory::build_branch_root(root), + MessageFactory::build_branch_level(level), + MessageFactory::build_accept_children(false), + ] +} + #[must_use] pub fn build_branch_root(root: &str) -> Message { framed(5, |m| { diff --git a/soulseek-rs-lib/src/message/server/distributed.rs b/soulseek-rs-lib/src/message/server/distributed.rs new file mode 100644 index 0000000..fed5da9 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/distributed.rs @@ -0,0 +1,127 @@ +//! What the server says about the distributed network: parent candidates +//! (code 102), a reset (130), and a search embedded for a branch root (93). + +use crate::actor::server_actor::ServerMessage; +use crate::message::distributed::{Distributed, parse_at}; +use crate::message::{Message, MessageHandler}; +use std::sync::mpsc::Sender; + +pub struct PossibleParentsHandler; +impl MessageHandler for PossibleParentsHandler { + fn get_code(&self) -> u32 { + 102 + } + fn handle(&self, message: &mut Message, sender: Sender) { + let count = message.read_int32(); + let mut candidates = Vec::new(); + for _ in 0..count.min(10) { + if message.get_pointer() >= message.get_size() { + break; + } + let username = message.read_string(); + // The address travels as four bytes in network order. + let ip: Vec = (0..4).map(|_| message.read_int8()).collect(); + let host = format!("{}.{}.{}.{}", ip[3], ip[2], ip[1], ip[0]); + let port = message.read_int32() as u16; + candidates.push((username, host, port)); + } + let _ = sender.send(ServerMessage::PossibleParents(candidates)); + } +} + +pub struct ResetDistributedHandler; +impl MessageHandler for ResetDistributedHandler { + fn get_code(&self) -> u32 { + 130 + } + fn handle(&self, _message: &mut Message, sender: Sender) { + let _ = sender.send(ServerMessage::ResetDistributed); + } +} + +/// A search the server hands us directly because we are a branch root; it is +/// answered like any other. +pub struct EmbeddedMessageHandler; +impl MessageHandler for EmbeddedMessageHandler { + fn get_code(&self) -> u32 { + 93 + } + fn handle(&self, message: &mut Message, sender: Sender) { + let code = message.read_int8(); + if let Some(Distributed::Search { + username, + token, + query, + }) = parse_at(message, code) + { + let _ = sender.send(ServerMessage::FileSearchRequest { + username, + token, + query, + }); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc::channel; + + #[test] + fn possible_parents_carry_each_candidates_address() { + let (sender, received) = channel(); + let mut message = crate::message::framed(|m| { + m.write_int32(2) + .write_string("alice") + .write_int32(u32::from_le_bytes([1, 0, 0, 127])) + .write_int32(2234) + .write_string("bob") + .write_int32(u32::from_le_bytes([8, 8, 8, 8])) + .write_int32(2235); + }); + PossibleParentsHandler.handle(&mut message, sender); + let ServerMessage::PossibleParents(candidates) = + received.try_recv().unwrap() + else { + panic!("expected PossibleParents"); + }; + assert_eq!( + candidates, + vec![ + ("alice".to_string(), "127.0.0.1".to_string(), 2234), + ("bob".to_string(), "8.8.8.8".to_string(), 2235), + ] + ); + } + + #[test] + fn a_reset_and_an_embedded_search_become_client_events() { + let (sender, received) = channel(); + let mut reset = crate::message::framed(|_| {}); + ResetDistributedHandler.handle(&mut reset, sender.clone()); + assert!(matches!( + received.try_recv().unwrap(), + ServerMessage::ResetDistributed + )); + + let inner = crate::message::distributed::build_search("seeker", 9, "q") + .get_data(); + let mut embedded = crate::message::framed(|m| { + m.write_raw_bytes(inner); + }); + EmbeddedMessageHandler.handle(&mut embedded, sender); + let ServerMessage::FileSearchRequest { + username, + token, + query, + } = received.try_recv().unwrap() + else { + panic!("expected FileSearchRequest"); + }; + assert_eq!( + (username.as_str(), token, query.as_str()), + ("seeker", 9, "q") + ); + } +} diff --git a/soulseek-rs-lib/src/message/server/mod.rs b/soulseek-rs-lib/src/message/server/mod.rs index 3d210ab..35fcc3b 100644 --- a/soulseek-rs-lib/src/message/server/mod.rs +++ b/soulseek-rs-lib/src/message/server/mod.rs @@ -1,5 +1,6 @@ mod admin_message; mod connect_to_peer; +mod distributed; mod excluded_search_phrases; mod file_search; mod get_peer_address; @@ -22,6 +23,9 @@ mod wish_list_interval; pub use admin_message::AdminMessageHandler; pub use connect_to_peer::ConnectToPeerHandler; +pub use distributed::{ + EmbeddedMessageHandler, PossibleParentsHandler, ResetDistributedHandler, +}; pub use excluded_search_phrases::ExcludedSearchPhrasesHandler; pub use file_search::FileSearchHandler; pub use get_peer_address::GetPeerAddressHandler; diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index f6ec4ea..b0d2fe8 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -23,6 +23,7 @@ use std::sync::mpsc::{Receiver, Sender}; use std::time::{Duration, Instant}; use soulseek_rs::message::Message; +use soulseek_rs::message::distributed; use soulseek_rs::message::server::MessageFactory; use soulseek_rs::{ Client, ClientSettings, ClientVersion, ConnectionType, DownloadStatus, @@ -3616,3 +3617,152 @@ fn a_browse_listing_carries_the_files_attributes() { let _ = std::fs::remove_dir_all(share_dir); } + +/// The next inbound connection, or a timeout error: a mock that never gets +/// dialled must fail its test rather than hang it. +fn accept_within( + listener: &std::net::TcpListener, + timeout: Duration, +) -> std::io::Result { + listener.set_nonblocking(true)?; + let deadline = Instant::now() + timeout; + let stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + if Instant::now() >= deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "no inbound connection", + )); + } + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => return Err(e), + } + }; + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(Duration::from_secs(15)))?; + Ok(stream) +} + +/// The PeerInit that opens an inbound connection: who dialled, and as what. +fn peer_init_of(stream: &mut TcpStream) -> std::io::Result<(String, String)> { + let mut init = read_framed(stream)?; + assert_eq!(init.get_init_code(), 1, "expected a PeerInit"); + init.set_pointer(5); + Ok((init.read_string(), init.read_string())) +} + +// Searches also travel peer to peer, down a tree of parents and children the +// server assembles from PossibleParents. A leaf dials the candidates with a D +// connection, adopts the first that passes it a search after stating its +// branch, and answers those searches exactly like ones from the server. +// soulfind never hands out parents, so the parent here is a mock: a logged-in +// user the leaf is pointed at directly. +#[test] +fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + std::fs::create_dir_all(share_dir.join("album")).unwrap(); + std::fs::write(share_dir.join("album").join("treesearch.mp3"), b"xxxx") + .unwrap(); + + // The parent doubles as the searcher, so the leaf's answer comes back to + // the same listener. Bind all interfaces: soulfind reports the LAN address. + let parent_port = free_port().expect("parent port"); + let listener = + std::net::TcpListener::bind(("0.0.0.0", parent_port)).unwrap(); + let server_addr = format!("{}:{}", server.host, server.port); + let mut parent_srv = + login_raw(&server_addr, "e2e_parent", "pw").expect("parent logs in"); + parent_srv + .write_all( + &MessageFactory::build_set_wait_port_message(parent_port) + .get_buffer(), + ) + .unwrap(); + + let mut leaf = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.listening_settings( + "e2e_leaf", + "pw", + free_port().expect("leaf port"), + ) + }); + leaf.connect().expect("leaf connect"); + assert!(leaf.login().expect("leaf login")); + assert_eq!( + leaf.distributed_branch(), + ("e2e_leaf".to_string(), 0), + "a leaf without a parent is its own branch root" + ); + + leaf.consider_parents(vec![( + "e2e_parent".to_string(), + "127.0.0.1".to_string(), + parent_port, + )]) + .expect("consider parents"); + + let mut link = accept_within(&listener, Duration::from_secs(15)) + .expect("the leaf dials the candidate parent"); + assert_eq!( + peer_init_of(&mut link).unwrap(), + ("e2e_leaf".to_string(), "D".to_string()), + "a parent link opens with a distributed PeerInit" + ); + + // We are a branch root at level 0; the leaf hangs one level below us. + link.write_all(&distributed::build_branch_level(0).get_buffer()) + .unwrap(); + link.write_all(&distributed::build_branch_root("e2e_parent").get_buffer()) + .unwrap(); + link.write_all( + &distributed::build_search("e2e_parent", 4242, "treesearch") + .get_buffer(), + ) + .unwrap(); + + let deadline = Instant::now() + Duration::from_secs(10); + while leaf.distributed_branch().1 == 0 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + assert_eq!( + leaf.distributed_branch(), + ("e2e_parent".to_string(), 1), + "the first search from a candidate with a branch adopts it" + ); + + // The answer arrives the ordinary way: a P connection to the searcher + // carrying a FileSearchResponse for our token. + let mut p = accept_within(&listener, Duration::from_secs(20)) + .expect("the leaf dials the searcher to answer"); + assert_eq!( + peer_init_of(&mut p).unwrap(), + ("e2e_leaf".to_string(), "P".to_string()) + ); + let response = expect_code(&mut p, 9, Duration::from_secs(15)) + .expect("a FileSearchResponse for the search from the tree"); + let body = soulseek_rs::utils::zlib::inflate(&response.get_data()[8..]) + .expect("the response body inflates"); + let name_len = u32::from_le_bytes(body[..4].try_into().unwrap()) as usize; + let token_at = 4 + name_len; + assert_eq!(&body[4..token_at], b"e2e_leaf"); + assert_eq!( + u32::from_le_bytes(body[token_at..token_at + 4].try_into().unwrap()), + 4242 + ); + + // Losing the parent puts the leaf back on its own. + drop(link); + let deadline = Instant::now() + Duration::from_secs(10); + while leaf.distributed_branch().1 != 0 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + assert_eq!(leaf.distributed_branch(), ("e2e_leaf".to_string(), 0)); + + let _ = std::fs::remove_dir_all(share_dir); +} From 9d6a5aaa98147bcb8b11b7664de3306419dd18b3 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 01:06:00 +0200 Subject: [PATCH 3/4] fix(lib): harden the distributed leaf against hostile and slow parents Review of the leaf found the seams a stranger on the other end of a D connection could pull at, and a few that a slow one would. - A code-93 envelope is unwrapped once, as Nicotine+ does; a frame made of nothing but envelopes used to recurse once per byte. - A branch level is validated before it is used: negative levels leave the candidate unadoptable and the largest one no longer overflows. - The link reads through the crate's MessageReader, so a frame split across a read timeout is no longer torn or mistaken for a closed link. - Every event a link raises carries its link number, so a stale thread cannot adopt, close or move a newer link to the same user. - Later branch facts from the parent move our own place and are told to the server; a fresh login resets the leaf, whose stance is now the one place the four announcement messages come from. - Candidates from a later batch join the ones in flight instead of cancelling them, capped at ten; searches from the parent are budgeted at fifty a second; a reply queued for a searcher the server cannot place expires after a minute. - Truncated frames and half-read PossibleParents entries are dropped instead of yielding an empty root, a level of zero or a phantom candidate. The e2e now has a searcher of its own, so the answer must go to whoever asked and not to the parent that relayed it. --- soulseek-rs-lib/src/actor/server_actor.rs | 18 +- .../src/actor/server_actor/tests.rs | 18 +- soulseek-rs-lib/src/client/distributed.rs | 471 +++++++++++++----- soulseek-rs-lib/src/client/mod.rs | 40 +- soulseek-rs-lib/src/client/operations.rs | 58 ++- soulseek-rs-lib/src/client/tests.rs | 15 + soulseek-rs-lib/src/message/distributed.rs | 85 +++- .../src/message/server/distributed.rs | 59 ++- soulseek-rs-lib/tests/e2e.rs | 32 +- 9 files changed, 589 insertions(+), 207 deletions(-) diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 74cdce1..ecab497 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -183,12 +183,10 @@ pub struct ServerActor { shared_folder_count: u32, shared_file_count: u32, session: SessionWatch, - /// The name we logged in as, once a login has been asked for. - username: String, } /// The messages a client sends right after a successful login: its shared-file -/// counts, its distributed-network stance, online status, and (when listening) +/// counts, online status, and (when listening) /// the port peers should connect to. Kept as a free function so it can be /// tested without a live connection. fn post_login_messages( @@ -196,7 +194,6 @@ fn post_login_messages( listen_port: u16, shared_folders: u32, shared_files: u32, - own_username: &str, ) -> Vec { let mut messages = vec![ MessageFactory::build_shared_folders_message( @@ -205,19 +202,12 @@ fn post_login_messages( ), MessageFactory::build_set_status_message(2), ]; - messages.splice(1..1, parentless_stance(own_username)); if enable_listen { messages.push(MessageFactory::build_set_wait_port_message(listen_port)); } messages } -/// What a leaf tells the server while it has no parent: it is its own branch -/// root at level 0. -fn parentless_stance(own_username: &str) -> Vec { - crate::message::distributed::stance(own_username, 0, false) -} - impl ServerActor { #[must_use] pub fn new( @@ -245,7 +235,6 @@ impl ServerActor { shared_folder_count, shared_file_count, session: SessionWatch::default(), - username: String::new(), } } @@ -579,11 +568,13 @@ impl ServerActor { self.listen_port, self.shared_folder_count, self.shared_file_count, - &self.username, ) { self.send_message(msg); } self.session.clear(); + // The distributed stance is the leaf's to announce, and a new + // session starts without a parent. + self.forward_to_client(ClientOperation::ResetDistributed); } match self.context.write_safe() { Ok(mut ctx) => ctx.logged_in = Some(message), @@ -651,7 +642,6 @@ impl ServerActor { version: ClientVersion, response: std::sync::mpsc::Sender>, ) { - self.username.clone_from(&username); if self.stream.is_none() && !self.initiate_connection() { let _ = response.send(Err(SoulseekRs::NotConnected)); return; diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs index e19ce1f..aa99422 100644 --- a/soulseek-rs-lib/src/actor/server_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -193,18 +193,10 @@ fn a_successful_login_marks_the_session_live_again() { #[test] fn post_login_messages_carry_counts_and_conditional_wait_port() { - let messages = post_login_messages(true, 4321, 3, 7, "me"); + let messages = post_login_messages(true, 4321, 3, 7); let codes: Vec = messages.iter().map(code_of).collect(); - // SharedFolders; the distributed stance HaveNoParent, BranchRoot, - // BranchLevel, AcceptChildren; SetStatus; SetWaitPort. - assert_eq!(codes, vec![35, 71, 127, 126, 100, 28, 2]); - // A leaf with no parent is its own branch root at level 0, and it - // declines children. - let mut root = messages[2].clone(); - root.set_pointer(4); - assert_eq!(root.read_string(), "me"); - assert_eq!(&messages[3].get_data()[4..8], &0u32.to_le_bytes()); - assert_eq!(messages[4].get_data()[4], 0); + // SharedFolders; SetStatus; SetWaitPort. + assert_eq!(codes, vec![35, 28, 2]); // The SharedFolders message (code 35) carries the real counts. let shared = messages[0].get_data(); @@ -212,7 +204,7 @@ fn post_login_messages_carry_counts_and_conditional_wait_port() { assert_eq!(u32::from_le_bytes(shared[8..12].try_into().unwrap()), 7); // Not listening omits SetWaitPort (code 2). - let no_listen = post_login_messages(false, 4321, 3, 7, "me"); + let no_listen = post_login_messages(false, 4321, 3, 7); let codes: Vec = no_listen.iter().map(code_of).collect(); - assert_eq!(codes, vec![35, 71, 127, 126, 100, 28]); + assert_eq!(codes, vec![35, 28]); } diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs index 8181050..b936d40 100644 --- a/soulseek-rs-lib/src/client/distributed.rs +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -3,18 +3,18 @@ //! server when that changes. use super::{Client, ClientOperation, Result, ServerMessage}; -use crate::message::Message; +use crate::message::MessageReader; use crate::message::distributed::{self, Distributed}; use crate::message::server::MessageFactory; use crate::peer::ConnectionType; use crate::utils::lock::RwLockExt; use std::collections::HashMap; -use std::io::{self, Read, Write}; +use std::io::{self, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::Sender; -use std::time::Duration; +use std::time::{Duration, Instant}; /// A candidate that has not answered the dial in this long is not a parent. const LINK_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); @@ -23,6 +23,13 @@ const CANCEL_POLL: Duration = Duration::from_secs(1); /// A distributed frame is a search or a branch fact; anything bigger is not /// one of ours. const MAX_FRAME: usize = 64 * 1024; +/// The server offers at most ten parents at a time; more candidates than +/// that in flight is a server that keeps offering, and each one is a thread. +const MAX_CANDIDATES: usize = 10; +/// Searches admitted from the parent per second. Every search costs a scan +/// of everything we share, and the tree carries the whole network's stream. +// ponytail: a flat budget; a word index over the shares is the upgrade path. +const SEARCH_BUDGET: u32 = 50; /// Where we sit in the tree, as told to the server. #[derive(Debug, Clone, PartialEq, Eq)] @@ -46,15 +53,17 @@ impl Branch { } } -#[derive(Default)] struct Candidate { - level: Option, + link: u64, + level: Option, root: Option, cancel: Arc, } -/// A candidate to dial, and the flag that tells its link to stop. +/// A candidate to dial: the link number that tags everything it sends, and +/// the flag that tells its thread to stop. pub struct Dial { + pub link: u64, pub username: String, pub host: String, pub port: u16, @@ -63,12 +72,16 @@ pub struct Dial { /// The leaf's view: at most one parent, adopted from the candidates the /// server offered, the way Nicotine+ does it. Every method is a pure state -/// change so the rule is testable without a socket. +/// change so the rule is testable without a socket. Events carry the link +/// number they came from, so a stale thread cannot speak for a newer link +/// to the same user. pub struct Leaf { own: String, parent: Option, branch: Branch, candidates: HashMap, + next_link: u64, + searches: (Instant, u32), } impl Leaf { @@ -79,6 +92,8 @@ impl Leaf { parent: None, branch: Branch::parentless(own), candidates: HashMap::new(), + next_link: 1, + searches: (Instant::now(), 0), } } @@ -87,13 +102,15 @@ impl Leaf { self.branch.clone() } + /// Whether `link` is the live link to our parent. #[must_use] - pub fn is_parent(&self, username: &str) -> bool { + pub fn is_parent(&self, username: &str, link: u64) -> bool { self.parent.as_deref() == Some(username) + && self.has_link(username, link) } - /// New candidates: dial them all, unless we already hang from a parent. - /// Candidates still being sounded out from an earlier batch are dropped. + /// New candidates to dial: those we are not already sounding out, up to + /// the cap, and none at all while we hang from a parent. pub fn consider( &mut self, candidates: Vec<(String, String, u16)>, @@ -101,50 +118,74 @@ impl Leaf { if self.parent.is_some() { return Vec::new(); } - self.cancel_candidates(); - candidates - .into_iter() - .map(|(username, host, port)| { - let cancel = Arc::new(AtomicBool::new(false)); - self.candidates.insert( - username.clone(), - Candidate { - cancel: cancel.clone(), - ..Candidate::default() - }, - ); - Dial { - username, - host, - port, - cancel, - } - }) - .collect() + let mut dials = Vec::new(); + for (username, host, port) in candidates { + if self.candidates.contains_key(&username) + || self.candidates.len() >= MAX_CANDIDATES + { + continue; + } + let link = self.next_link; + self.next_link += 1; + let cancel = Arc::new(AtomicBool::new(false)); + self.candidates.insert( + username.clone(), + Candidate { + link, + level: None, + root: None, + cancel: cancel.clone(), + }, + ); + dials.push(Dial { + link, + username, + host, + port, + cancel, + }); + } + dials } - /// A branch root does not always say so separately: level 0 is its own - /// root. - pub fn branch_level(&mut self, username: &str, level: i32) { - let Some(candidate) = self.candidates.get_mut(username) else { - return; - }; - candidate.level = Some(level); + /// A candidate said how deep it sits. A branch root does not always say + /// so separately: level 0 is its own root. A negative level is nonsense + /// and leaves the candidate unadoptable. When it is the parent talking, + /// our own place moves with it and the new stance comes back. + pub fn branch_level( + &mut self, + username: &str, + link: u64, + level: i32, + ) -> Option { + if !self.has_link(username, link) { + return None; + } + let candidate = self.candidates.get_mut(username)?; + candidate.level = u32::try_from(level).ok(); if level == 0 && candidate.root.is_none() { candidate.root = Some(username.to_string()); } + self.moved_with_parent(username) } - pub fn branch_root(&mut self, username: &str, root: &str) { - if let Some(candidate) = self.candidates.get_mut(username) { - candidate.root = Some(root.to_string()); + pub fn branch_root( + &mut self, + username: &str, + link: u64, + root: &str, + ) -> Option { + if !self.has_link(username, link) { + return None; } + self.candidates.get_mut(username)?.root = Some(root.to_string()); + self.moved_with_parent(username) } - /// A search came down from `username`. The first one from a candidate that - /// has told us its place adopts it; the stance to announce comes back. - pub fn search_from(&mut self, username: &str) -> Option { - if self.parent.is_some() { + /// A search came down from `username`. The first one from a candidate + /// that has told us its place adopts it; the stance to announce comes back. + pub fn search_from(&mut self, username: &str, link: u64) -> Option { + if self.parent.is_some() || !self.has_link(username, link) { return None; } let candidate = self.candidates.get(username)?; @@ -157,16 +198,28 @@ impl Leaf { self.cancel_candidates(); self.candidates.insert(username.to_string(), parent); self.parent = Some(username.to_string()); - self.branch = - Branch::under(&root, u32::try_from(level + 1).unwrap_or(1)); + self.branch = Branch::under(&root, level.saturating_add(1)); Some(self.branch.clone()) } + /// Whether to answer one more search from the parent now: at most + /// `SEARCH_BUDGET` a second, the rest dropped. + pub fn admit_search(&mut self, now: Instant) -> bool { + if now.duration_since(self.searches.0) >= Duration::from_secs(1) { + self.searches = (now, 0); + } + self.searches.1 += 1; + self.searches.1 <= SEARCH_BUDGET + } + /// The link to `username` closed. True when that was the parent, which /// puts us back to parentless. - pub fn closed(&mut self, username: &str) -> bool { + pub fn closed(&mut self, username: &str, link: u64) -> bool { + if !self.has_link(username, link) { + return false; + } self.candidates.remove(username); - if !self.is_parent(username) { + if self.parent.as_deref() != Some(username) { return false; } self.parent = None; @@ -180,6 +233,28 @@ impl Leaf { self.branch = Branch::parentless(&self.own); } + fn has_link(&self, username: &str, link: u64) -> bool { + self.candidates + .get(username) + .is_some_and(|c| c.link == link) + } + + fn moved_with_parent(&mut self, username: &str) -> Option { + if self.parent.as_deref() != Some(username) { + return None; + } + let candidate = self.candidates.get(username)?; + let branch = Branch::under( + candidate.root.as_deref()?, + candidate.level?.saturating_add(1), + ); + if branch == self.branch { + return None; + } + self.branch = branch.clone(); + Some(branch) + } + fn cancel_candidates(&mut self) { for candidate in self.candidates.values() { candidate.cancel.store(true, Ordering::Relaxed); @@ -205,17 +280,31 @@ pub fn announce( /// Dial a parent candidate on its own thread: a `D` PeerInit, then whatever it /// sends comes back as operations until the link closes or is cancelled. pub fn spawn_link(dial: Dial, own: String, ops: Sender) { - let _ = std::thread::Builder::new() + let closed = ClientOperation::ParentClosed { + parent: dial.username.clone(), + link: dial.link, + }; + let thread_ops = ops.clone(); + let spawned = std::thread::Builder::new() .name("soulseek-parent".to_string()) .spawn(move || { - let parent = dial.username.clone(); - if let Err(e) = link(&dial, &own, &ops) + if let Err(e) = link(&dial, &own, &thread_ops) && !dial.cancel.load(Ordering::Relaxed) { - crate::debug!("[distributed] link to {} ended: {}", parent, e); - let _ = ops.send(ClientOperation::ParentClosed { parent }); + crate::debug!( + "[distributed] link to {} ended: {}", + dial.username, + e + ); + let _ = thread_ops.send(ClientOperation::ParentClosed { + parent: dial.username, + link: dial.link, + }); } }); + if spawned.is_err() { + let _ = ops.send(closed); + } } fn link( @@ -230,70 +319,75 @@ fn link( io::Error::new(io::ErrorKind::InvalidInput, "no address") })?; let mut stream = TcpStream::connect_timeout(&addr, LINK_CONNECT_TIMEOUT)?; + if dial.cancel.load(Ordering::Relaxed) { + return Ok(()); + } stream.set_read_timeout(Some(CANCEL_POLL))?; stream.write_all( &MessageFactory::build_peer_init_message(own, ConnectionType::D, 0) .get_buffer(), )?; + let mut reader = MessageReader::new(); loop { if dial.cancel.load(Ordering::Relaxed) { return Ok(()); } - let Some(mut frame) = read_frame(&mut stream)? else { - continue; - }; - let parent = dial.username.clone(); - let operation = match distributed::parse(&mut frame) { - Some(Distributed::BranchLevel(level)) => { - ClientOperation::ParentBranchLevel { parent, level } + match reader.read_from_socket(&mut stream) { + Ok(()) => {} + Err(e) + if matches!( + e.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + continue; } - Some(Distributed::BranchRoot(root)) => { - ClientOperation::ParentBranchRoot { parent, root } + Err(e) => return Err(e), + } + while let Some(mut frame) = reader.extract_message()? { + let Some(operation) = operation_of(dial, &mut frame) else { + continue; + }; + if ops.send(operation).is_err() { + return Ok(()); } - Some(Distributed::Search { - username, - token, - query, - }) => ClientOperation::ParentSearch { - parent, - username, - token, - query, - }, - None => continue, - }; - if ops.send(operation).is_err() { - return Ok(()); + } + if reader.buffer_len() > MAX_FRAME { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "oversized frame", + )); } } } -/// One framed message, or `None` when the read timed out with nothing sent. -fn read_frame(stream: &mut TcpStream) -> io::Result> { - let mut len = [0u8; 4]; - match stream.read_exact(&mut len) { - Ok(()) => {} - Err(e) - if matches!( - e.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ) => - { - return Ok(None); +fn operation_of( + dial: &Dial, + frame: &mut crate::message::Message, +) -> Option { + let parent = dial.username.clone(); + let link = dial.link; + Some(match distributed::parse(frame)? { + Distributed::BranchLevel(level) => ClientOperation::ParentBranchLevel { + parent, + link, + level, + }, + Distributed::BranchRoot(root) => { + ClientOperation::ParentBranchRoot { parent, link, root } } - Err(e) => return Err(e), - } - let size = u32::from_le_bytes(len) as usize; - if size > MAX_FRAME { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "oversized frame", - )); - } - let mut data = len.to_vec(); - data.resize(4 + size, 0); - stream.read_exact(&mut data[4..])?; - Ok(Some(Message::new_with_data(data))) + Distributed::Search { + username, + token, + query, + } => ClientOperation::ParentSearch { + parent, + link, + username, + token, + query, + }, + }) } impl Client { @@ -339,6 +433,18 @@ mod tests { Leaf::new("me") } + fn candidate(name: &str) -> (String, String, u16) { + (name.to_string(), "127.0.0.1".to_string(), 2234) + } + + /// Dial the candidates and return their link numbers by name. + fn dial(leaf: &mut Leaf, names: &[&str]) -> HashMap { + leaf.consider(names.iter().map(|n| candidate(n)).collect()) + .into_iter() + .map(|d| (d.username.clone(), d)) + .collect() + } + #[test] fn a_leaf_without_a_parent_is_its_own_root_at_level_zero() { assert_eq!(leaf().branch(), Branch::parentless("me")); @@ -347,12 +453,13 @@ mod tests { #[test] fn candidates_are_dialled_only_while_there_is_no_parent() { let mut leaf = leaf(); - let dial = leaf.consider(vec![candidate("alice"), candidate("bob")]); - assert_eq!(dial.len(), 2); - leaf.branch_level("alice", 1); - leaf.branch_root("alice", "rooty"); + let dials = dial(&mut leaf, &["alice", "bob"]); + assert_eq!(dials.len(), 2); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 1); + leaf.branch_root("alice", alice, "rooty"); assert!( - leaf.search_from("alice").is_some(), + leaf.search_from("alice", alice).is_some(), "alice becomes the parent" ); assert!(leaf.consider(vec![candidate("carol")]).is_empty()); @@ -361,48 +468,158 @@ mod tests { #[test] fn the_first_search_from_a_candidate_with_branch_info_adopts_it() { let mut leaf = leaf(); - leaf.consider(vec![candidate("alice"), candidate("bob")]); - assert_eq!(leaf.search_from("alice"), None, "no branch info yet"); - leaf.branch_level("alice", 3); - assert_eq!(leaf.search_from("alice"), None, "level without root"); - leaf.branch_root("alice", "rooty"); + let dials = dial(&mut leaf, &["alice", "bob"]); + let (alice, bob) = (dials["alice"].link, dials["bob"].link); + assert_eq!(leaf.search_from("alice", alice), None, "no branch yet"); + leaf.branch_level("alice", alice, 3); + assert_eq!(leaf.search_from("alice", alice), None, "level, no root"); + leaf.branch_root("alice", alice, "rooty"); assert_eq!( - leaf.search_from("alice"), + leaf.search_from("alice", alice), Some(Branch::under("rooty", 4)), "level 3 parent puts us at level 4" ); assert_eq!(leaf.branch(), Branch::under("rooty", 4)); - assert_eq!(leaf.search_from("alice"), None, "already the parent"); - assert_eq!(leaf.search_from("bob"), None, "a loser is ignored"); + assert!(leaf.is_parent("alice", alice)); + assert_eq!(leaf.search_from("alice", alice), None, "already parent"); + assert_eq!(leaf.search_from("bob", bob), None, "a loser is ignored"); + assert!( + dials["bob"].cancel.load(Ordering::Relaxed), + "the losing link is told to stop" + ); + assert!(!dials["alice"].cancel.load(Ordering::Relaxed)); } #[test] fn a_level_zero_candidate_is_its_own_root() { let mut leaf = leaf(); - leaf.consider(vec![candidate("alice")]); - leaf.branch_level("alice", 0); - assert_eq!(leaf.search_from("alice"), Some(Branch::under("alice", 1))); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_level("alice", alice, 0); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("alice", 1)) + ); + } + + #[test] + fn a_hostile_branch_level_neither_panics_nor_adopts() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_root("alice", alice, "rooty"); + leaf.branch_level("alice", alice, -1); + assert_eq!(leaf.search_from("alice", alice), None, "negative level"); + leaf.branch_level("alice", alice, i32::MAX); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("rooty", 1 << 31)) + ); + } + + #[test] + fn the_parents_later_moves_move_us_too() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); + assert_eq!(leaf.branch_level("alice", alice, 0), None, "unchanged"); + assert_eq!( + leaf.branch_root("alice", alice, "grand"), + Some(Branch::under("grand", 1)) + ); + assert_eq!( + leaf.branch_level("alice", alice, 2), + Some(Branch::under("grand", 3)) + ); + assert_eq!(leaf.branch(), Branch::under("grand", 3)); + } + + #[test] + fn a_stale_link_cannot_speak_for_a_newer_one() { + let mut leaf = leaf(); + let old = dial(&mut leaf, &["alice"])["alice"].link; + assert!(!leaf.closed("alice", old + 7), "unknown link is ignored"); + leaf.closed("alice", old); + let new = dial(&mut leaf, &["alice"])["alice"].link; + assert_ne!(old, new); + leaf.branch_level("alice", old, 0); + assert_eq!(leaf.search_from("alice", old), None, "old link ignored"); + assert!( + !leaf.closed("alice", old), + "a late close leaves the new one" + ); + leaf.branch_level("alice", new, 0); + assert!(leaf.search_from("alice", new).is_some()); + } + + #[test] + fn a_later_batch_keeps_the_candidates_in_flight_and_stays_under_the_cap() { + let mut leaf = leaf(); + let first = dial(&mut leaf, &["alice"])["alice"].link; + let again = dial(&mut leaf, &["alice", "bob"]); + assert!( + !again.contains_key("alice"), + "alice is already being dialled" + ); + assert!(again.contains_key("bob")); + assert!(leaf.has_link("alice", first)); + let names: Vec = (0..20).map(|i| format!("p{i}")).collect(); + let more = leaf + .consider(names.iter().map(|n| candidate(n)).collect::>()); + assert_eq!(more.len(), MAX_CANDIDATES - 2); } #[test] fn losing_the_parent_or_a_reset_returns_to_parentless() { let mut leaf = leaf(); - leaf.consider(vec![candidate("alice")]); - leaf.branch_level("alice", 0); - leaf.search_from("alice"); - assert!(!leaf.closed("bob"), "a stranger closing changes nothing"); - assert!(leaf.closed("alice")); + let dials = dial(&mut leaf, &["alice", "bob"]); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); + assert!(!leaf.closed("bob", dials["bob"].link), "not the parent"); + assert!(leaf.closed("alice", alice)); assert_eq!(leaf.branch(), Branch::parentless("me")); - leaf.consider(vec![candidate("alice")]); - leaf.branch_level("alice", 0); - leaf.search_from("alice"); + let dials = dial(&mut leaf, &["alice", "carol"]); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); leaf.reset(); assert_eq!(leaf.branch(), Branch::parentless("me")); + assert!(dials["alice"].cancel.load(Ordering::Relaxed)); + assert!(dials["carol"].cancel.load(Ordering::Relaxed)); assert!(!leaf.consider(vec![candidate("bob")]).is_empty()); } - fn candidate(name: &str) -> (String, String, u16) { - (name.to_string(), "127.0.0.1".to_string(), 2234) + #[test] + fn searches_beyond_the_budget_are_dropped_until_the_next_second() { + let mut leaf = leaf(); + let now = Instant::now(); + for _ in 0..SEARCH_BUDGET { + assert!(leaf.admit_search(now)); + } + assert!(!leaf.admit_search(now)); + assert!(leaf.admit_search(now + Duration::from_secs(1))); + } + + #[test] + fn announcing_tells_the_server_the_root_the_level_and_that_we_have_a_parent() + { + let (tx, rx) = std::sync::mpsc::channel(); + announce(Some(&tx), &Branch::under("rooty", 4), true); + let mut sent: Vec = Vec::new(); + while let Ok(ServerMessage::SendMessage(m)) = rx.try_recv() { + sent.push(m); + } + let codes: Vec = sent + .iter() + .map(|m| u32::from_le_bytes(m.get_data()[..4].try_into().unwrap())) + .collect(); + assert_eq!(codes, vec![71, 127, 126, 100]); + assert_eq!(sent[0].get_data()[4], 0, "HaveNoParent is false"); + let mut root = sent[1].clone(); + root.set_pointer(4); + assert_eq!(root.read_string(), "rooty"); + assert_eq!(&sent[2].get_data()[4..8], &4u32.to_le_bytes()); + assert_eq!(sent[3].get_data()[4], 0, "AcceptChildren is false"); } } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 9b80e54..160e7a6 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -60,6 +60,8 @@ pub const DEFAULT_UPLOAD_SLOTS: usize = 10; const BROKER_CONNECT_TIMEOUT: Duration = Duration::from_secs(20); const DEFAULT_MAX_PEERS: usize = 512; +/// How long a message waits for a control connection to its peer. +const PENDING_PEER_TTL: Duration = Duration::from_mins(1); const BROWSE_PROTECT_WINDOW: Duration = Duration::from_mins(5); @@ -338,19 +340,23 @@ pub enum ClientOperation { PossibleParents(Vec<(String, String, u16)>), /// The server told us to drop our parent. ResetDistributed, - /// A parent candidate told us how deep it sits. + /// A parent candidate told us how deep it sits. `link` says which dial + /// to that user is talking. ParentBranchLevel { parent: String, + link: u64, level: i32, }, /// A parent candidate told us whose branch it is on. ParentBranchRoot { parent: String, + link: u64, root: String, }, /// A search came down the tree from `parent`. ParentSearch { parent: String, + link: u64, username: String, token: u32, query: String, @@ -358,6 +364,7 @@ pub enum ClientOperation { /// The link to a parent or candidate is gone. ParentClosed { parent: String, + link: u64, }, } pub struct ClientContext { @@ -380,8 +387,10 @@ pub struct ClientContext { pub shared_directories: Vec, /// Peer listen addresses learned from GetPeerAddress responses. peer_addresses: HashMap, - /// Peer messages waiting for a control connection to that peer. - pending_peer_messages: HashMap>, + /// Peer messages waiting for a control connection to that peer, and + /// since when: a peer that never answers must not keep them forever. + pending_peer_messages: + HashMap)>, /// Uploads we have offered, keyed by our transfer token. uploads: HashMap, active_uploads: HashMap, @@ -474,6 +483,15 @@ impl ClientContext { } impl ClientContext { + /// A context for the client that logs in as `username`. + #[must_use] + pub fn for_user(username: &str) -> Self { + Self { + leaf: distributed::Leaf::new(username), + ..Self::new() + } + } + #[must_use] pub fn new() -> Self { let actor_system = Arc::new(ActorSystem::new()); @@ -725,7 +743,8 @@ impl ClientContext { ) { self.pending_peer_messages .entry(username.to_string()) - .or_default() + .or_insert_with(|| (Instant::now(), Vec::new())) + .1 .push(message); } @@ -736,9 +755,18 @@ impl ClientContext { ) -> Vec { self.pending_peer_messages .remove(username) + .map(|(_, messages)| messages) .unwrap_or_default() } + /// Drop peer messages nobody could deliver within + /// [`PENDING_PEER_TTL`]: the searcher went offline, or never existed. + pub(crate) fn expire_pending_peer_messages(&mut self, now: Instant) { + self.pending_peer_messages.retain(|_, (since, _)| { + now.duration_since(*since) < PENDING_PEER_TTL + }); + } + /// Store a shared-file listing received from browsing `username`. pub fn store_browse_result( &mut self, @@ -865,11 +893,13 @@ impl Client { listen_port: settings.listen_port, bound_port: None, address: settings.server_address, + context: Arc::new(RwLock::new(ClientContext::for_user( + &settings.username, + ))), username: settings.username, password: settings.password, version: settings.version, shared_directories: settings.shared_directories, - context: Arc::new(RwLock::new(ClientContext::new())), server_handle: None, session: SessionWatch::default(), } diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index da84f75..6d47d09 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -17,9 +17,6 @@ impl Client { own_username: String, ) { thread::spawn(move || { - if let Ok(mut ctx) = client_context.write_safe() { - ctx.leaf = super::distributed::Leaf::new(&own_username); - } let mut last_sweep = Instant::now(); loop { let next = reader.recv_timeout(CONNECT_SWEEP_INTERVAL); @@ -27,6 +24,9 @@ impl Client { last_sweep = Instant::now(); Self::sweep_expired_connects(&client_context); Self::sweep_stale_offers(&client_context); + if let Ok(mut ctx) = client_context.write_safe() { + ctx.expire_pending_peer_messages(Instant::now()); + } } let operation = match next { Ok(operation) => operation, @@ -716,18 +716,45 @@ impl Client { ); } } - ClientOperation::ParentBranchLevel { parent, level } => { - if let Ok(mut ctx) = client_context.write_safe() { - ctx.leaf.branch_level(&parent, level); + ClientOperation::ParentBranchLevel { + parent, + link, + level, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = + ctx.leaf.branch_level(&parent, link, level) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); } } - ClientOperation::ParentBranchRoot { parent, root } => { - if let Ok(mut ctx) = client_context.write_safe() { - ctx.leaf.branch_root(&parent, &root); + ClientOperation::ParentBranchRoot { + parent, + link, + root, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = + ctx.leaf.branch_root(&parent, link, &root) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); } } ClientOperation::ParentSearch { parent, + link, username, token, query, @@ -735,14 +762,17 @@ impl Client { let Ok(mut ctx) = client_context.write_safe() else { continue; }; - if let Some(branch) = ctx.leaf.search_from(&parent) { + if let Some(branch) = + ctx.leaf.search_from(&parent, link) + { super::distributed::announce( ctx.server_sender.as_ref(), &branch, true, ); } - if ctx.leaf.is_parent(&parent) + if ctx.leaf.is_parent(&parent, link) + && ctx.leaf.admit_search(Instant::now()) && let Some(ops) = &ctx.operations { let _ = ops.send(ClientOperation::IncomingSearch { @@ -752,11 +782,11 @@ impl Client { }); } } - ClientOperation::ParentClosed { parent } => { + ClientOperation::ParentClosed { parent, link } => { let Ok(mut ctx) = client_context.write_safe() else { continue; }; - if ctx.leaf.closed(&parent) { + if ctx.leaf.closed(&parent, link) { super::distributed::announce( ctx.server_sender.as_ref(), &ctx.leaf.branch(), @@ -764,6 +794,8 @@ impl Client { ); } } + // Also what a fresh login means: whatever tree we hung + // from belongs to the old session. ClientOperation::ResetDistributed => { let Ok(mut ctx) = client_context.write_safe() else { continue; diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 031edb2..1c16b42 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -821,3 +821,18 @@ fn watched_users_are_listed_in_a_stable_order() { context.add_watched_user("bob"); assert_eq!(context.watched_users(), vec!["alice", "bob", "carol"]); } + +// A reply queued for a searcher the server cannot place waits a minute, +// not forever: every such search would otherwise pin a map entry and a +// protected slot in the peer registry. +#[test] +fn a_peer_message_nobody_could_deliver_expires() { + let mut ctx = ClientContext::for_user("me"); + ctx.queue_peer_message("ghost", crate::message::Message::new()); + ctx.expire_pending_peer_messages(Instant::now()); + assert_eq!(ctx.take_peer_messages("ghost").len(), 1, "still fresh"); + + ctx.queue_peer_message("ghost", crate::message::Message::new()); + ctx.expire_pending_peer_messages(Instant::now() + PENDING_PEER_TTL); + assert!(ctx.take_peer_messages("ghost").is_empty(), "expired"); +} diff --git a/soulseek-rs-lib/src/message/distributed.rs b/soulseek-rs-lib/src/message/distributed.rs index 756219e..2ae015c 100644 --- a/soulseek-rs-lib/src/message/distributed.rs +++ b/soulseek-rs-lib/src/message/distributed.rs @@ -34,15 +34,27 @@ pub fn parse(message: &mut Message) -> Option { parse_at(message, message.get_init_code()) } -/// Decode the body that follows a one-byte `code`, the pointer already past it. +/// Decode the body that follows a one-byte `code`, the pointer already past +/// it. A code-93 envelope is unwrapped exactly once, as Nicotine+ does: an +/// envelope inside an envelope is nobody's search. #[must_use] pub fn parse_at(message: &mut Message, code: u8) -> Option { + let code = if code == 93 { + message.read_int8() + } else { + code + }; match code { 3 => { - if message.read_int32() != SEARCH_IDENTIFIER { + if remaining(message) < 4 + || message.read_int32() != SEARCH_IDENTIFIER + { return None; } let username = message.read_string(); + if remaining(message) < 4 { + return None; + } let token = message.read_int32(); let query = message.read_string(); (!username.is_empty() && !query.is_empty()).then_some( @@ -53,17 +65,20 @@ pub fn parse_at(message: &mut Message, code: u8) -> Option { }, ) } - 4 => Some(Distributed::BranchLevel(message.read_int32() as i32)), - 5 => Some(Distributed::BranchRoot(message.read_string())), - // Older branch roots wrap a search in a code-93 envelope. - 93 => { - let inner = message.read_int8(); - parse_at(message, inner) + 4 => (remaining(message) >= 4) + .then(|| Distributed::BranchLevel(message.read_int32() as i32)), + 5 => { + let root = message.read_string(); + (!root.is_empty()).then_some(Distributed::BranchRoot(root)) } _ => None, } } +const fn remaining(message: &mut Message) -> usize { + message.get_size().saturating_sub(message.get_pointer()) +} + /// A distributed frame with the one-byte `code`: what a parent (or a test /// standing in for one) sends down the tree. fn framed(code: u8, body: impl FnOnce(&mut Message)) -> Message { @@ -171,5 +186,59 @@ mod tests { assert_eq!(parse(&mut ping), None); let mut short = Message::new_with_data(vec![1, 0, 0, 0, 3]); assert_eq!(parse(&mut short), None); + let mut level = Message::new_with_data(vec![3, 0, 0, 0, 4, 1, 0]); + assert_eq!(parse(&mut level), None, "a level needs four bytes"); + let mut root = + Message::new_with_data(build_branch_root("").get_buffer()); + assert_eq!(parse(&mut root), None, "an empty root is no root"); + let mut blank = + Message::new_with_data(build_search("seeker", 1, "").get_buffer()); + assert_eq!(parse(&mut blank), None, "an empty query is no search"); + } + + // An envelope inside an envelope is unwrapped once and found empty; a + // frame that is nothing but envelopes must not recurse its way off the + // stack. + #[test] + fn a_nested_envelope_is_not_unwrapped_twice() { + let mut nested = Message::new(); + nested.write_int8(93).write_raw_bytes( + Message::new() + .write_int8(93) + .write_raw_bytes(build_search("seeker", 5, "q").get_data()) + .get_data(), + ); + let mut message = Message::new_with_data(nested.get_buffer()); + assert_eq!(parse(&mut message), None); + + let mut onions = Message::new_with_data( + Message::new() + .write_raw_bytes(vec![93u8; 64 * 1024]) + .get_buffer(), + ); + assert_eq!(parse(&mut onions), None); + } + + // The frame as the protocol documents it, byte for byte: code 3, the + // constant 49, then user, token and query. + #[test] + fn a_search_frame_matches_the_documented_layout() { + let mut bytes = vec![3u8, 49, 0, 0, 0]; + bytes.extend(2u32.to_le_bytes()); + bytes.extend(b"me"); + bytes.extend(9u32.to_le_bytes()); + bytes.extend(1u32.to_le_bytes()); + bytes.extend(b"q"); + let mut framed = (bytes.len() as u32).to_le_bytes().to_vec(); + framed.extend(&bytes); + assert_eq!(build_search("me", 9, "q").get_buffer(), framed); + assert_eq!( + parse(&mut Message::new_with_data(framed)), + Some(Distributed::Search { + username: "me".to_string(), + token: 9, + query: "q".to_string() + }) + ); } } diff --git a/soulseek-rs-lib/src/message/server/distributed.rs b/soulseek-rs-lib/src/message/server/distributed.rs index fed5da9..8dacc78 100644 --- a/soulseek-rs-lib/src/message/server/distributed.rs +++ b/soulseek-rs-lib/src/message/server/distributed.rs @@ -4,6 +4,7 @@ use crate::actor::server_actor::ServerMessage; use crate::message::distributed::{Distributed, parse_at}; use crate::message::{Message, MessageHandler}; +use std::net::Ipv4Addr; use std::sync::mpsc::Sender; pub struct PossibleParentsHandler; @@ -13,22 +14,29 @@ impl MessageHandler for PossibleParentsHandler { } fn handle(&self, message: &mut Message, sender: Sender) { let count = message.read_int32(); - let mut candidates = Vec::new(); - for _ in 0..count.min(10) { - if message.get_pointer() >= message.get_size() { - break; - } - let username = message.read_string(); - // The address travels as four bytes in network order. - let ip: Vec = (0..4).map(|_| message.read_int8()).collect(); - let host = format!("{}.{}.{}.{}", ip[3], ip[2], ip[1], ip[0]); - let port = message.read_int32() as u16; - candidates.push((username, host, port)); - } + let candidates = (0..count.min(10)) + .map_while(|_| entry(message)) + .filter_map(|(username, host, port)| { + Some((username, host, u16::try_from(port).ok()?)) + }) + .collect(); let _ = sender.send(ServerMessage::PossibleParents(candidates)); } } +/// One whole `(username, host, port)` entry, or `None` where the message +/// runs out: a truncated entry is not a parent to dial. +fn entry(message: &mut Message) -> Option<(String, String, u32)> { + let username = message.read_string(); + if username.is_empty() + || message.get_size().saturating_sub(message.get_pointer()) < 8 + { + return None; + } + let host = Ipv4Addr::from(message.read_int32()).to_string(); + Some((username, host, message.read_int32())) +} + pub struct ResetDistributedHandler; impl MessageHandler for ResetDistributedHandler { fn get_code(&self) -> u32 { @@ -95,6 +103,33 @@ mod tests { ); } + // The count says two, the bytes hold one and a half: the whole one is + // a candidate, the stub is not. A port that is not a port is skipped too. + #[test] + fn a_truncated_or_absurd_entry_is_not_a_candidate() { + let (sender, received) = channel(); + let mut message = crate::message::framed(|m| { + m.write_int32(3) + .write_string("alice") + .write_int32(u32::from_le_bytes([1, 0, 0, 127])) + .write_int32(70_000) + .write_string("bob") + .write_int32(u32::from_le_bytes([8, 8, 8, 8])) + .write_int32(2235) + .write_string("carol"); + }); + PossibleParentsHandler.handle(&mut message, sender); + let ServerMessage::PossibleParents(candidates) = + received.try_recv().unwrap() + else { + panic!("expected PossibleParents"); + }; + assert_eq!( + candidates, + vec![("bob".to_string(), "8.8.8.8".to_string(), 2235)] + ); + } + #[test] fn a_reset_and_an_embedded_search_become_client_events() { let (sender, received) = channel(); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index b0d2fe8..f4e7725 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -3669,20 +3669,21 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { std::fs::write(share_dir.join("album").join("treesearch.mp3"), b"xxxx") .unwrap(); - // The parent doubles as the searcher, so the leaf's answer comes back to - // the same listener. Bind all interfaces: soulfind reports the LAN address. - let parent_port = free_port().expect("parent port"); - let listener = - std::net::TcpListener::bind(("0.0.0.0", parent_port)).unwrap(); + // The parent and the searcher are two logged-in users with a listener + // each. Bind all interfaces: soulfind reports the LAN address. let server_addr = format!("{}:{}", server.host, server.port); - let mut parent_srv = - login_raw(&server_addr, "e2e_parent", "pw").expect("parent logs in"); - parent_srv - .write_all( - &MessageFactory::build_set_wait_port_message(parent_port) - .get_buffer(), + let raw_user = |name: &str| { + let port = free_port().expect("a port for the raw user"); + let listener = std::net::TcpListener::bind(("0.0.0.0", port)).unwrap(); + let mut srv = login_raw(&server_addr, name, "pw").expect("raw login"); + srv.write_all( + &MessageFactory::build_set_wait_port_message(port).get_buffer(), ) .unwrap(); + (listener, port, srv) + }; + let (listener, parent_port, _parent_srv) = raw_user("e2e_parent"); + let (searcher, _, _searcher_srv) = raw_user("e2e_searcher"); let mut leaf = Client::with_settings(ClientSettings { shared_directories: vec![share_dir.display().to_string()], @@ -3721,7 +3722,7 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { link.write_all(&distributed::build_branch_root("e2e_parent").get_buffer()) .unwrap(); link.write_all( - &distributed::build_search("e2e_parent", 4242, "treesearch") + &distributed::build_search("e2e_searcher", 4242, "treesearch") .get_buffer(), ) .unwrap(); @@ -3736,9 +3737,10 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { "the first search from a candidate with a branch adopts it" ); - // The answer arrives the ordinary way: a P connection to the searcher - // carrying a FileSearchResponse for our token. - let mut p = accept_within(&listener, Duration::from_secs(20)) + // The answer goes to whoever asked, not to the parent that relayed it: + // a P connection to the searcher carrying a FileSearchResponse for the + // token. + let mut p = accept_within(&searcher, Duration::from_secs(20)) .expect("the leaf dials the searcher to answer"); assert_eq!( peer_init_of(&mut p).unwrap(), From 85f26134eda8434be7647de3d521d98c7c1c8a63 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 01:25:02 +0200 Subject: [PATCH 4/4] fix(lib): free a silent parent's slot and pace the moves we report A candidate that accepted the dial and then said nothing kept its slot for the session, and ten of those blocked every later batch; a link now closes after a minute of silence. A parent that flaps its level made us tell the server four messages per flap; moves are now told at most once a second, the latest one winning. A parent that reports level zero is its own root from then on, whatever it said before. The wait for a searcher's control connection runs from the last message queued for it, not the first. A successful login is pinned to reset the leaf. --- .../src/actor/server_actor/tests.rs | 29 ++++++- soulseek-rs-lib/src/client/distributed.rs | 75 +++++++++++++++++-- soulseek-rs-lib/src/client/mod.rs | 12 +-- soulseek-rs-lib/src/client/operations.rs | 9 +++ 4 files changed, 111 insertions(+), 14 deletions(-) diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs index aa99422..8faba13 100644 --- a/soulseek-rs-lib/src/actor/server_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -36,15 +36,22 @@ fn the_server_socket_keeps_itself_alive() { } fn parked_actor(port: u16) -> ServerActor { - let (tx, _rx) = std::sync::mpsc::channel(); - ServerActor::new( + parked_actor_with_client(port).0 +} + +fn parked_actor_with_client( + port: u16, +) -> (ServerActor, std::sync::mpsc::Receiver) { + let (tx, rx) = std::sync::mpsc::channel(); + let actor = ServerActor::new( PeerAddress::new("127.0.0.1".to_string(), port), tx, 0, false, 0, 0, - ) + ); + (actor, rx) } #[test] @@ -191,6 +198,22 @@ fn a_successful_login_marks_the_session_live_again() { assert_eq!(actor.session.loss(), None); } +// A new session starts without a parent: the leaf drops whatever tree it +// hung from and announces its parentless stance from there. +#[test] +fn a_successful_login_resets_the_distributed_leaf() { + let (mut actor, client) = parked_actor_with_client(1); + + actor.handle_login_status(true); + + assert!(matches!( + client.try_recv(), + Ok(ClientOperation::ResetDistributed) + )); + actor.handle_login_status(false); + assert!(client.try_recv().is_err(), "a failed login resets nothing"); +} + #[test] fn post_login_messages_carry_counts_and_conditional_wait_port() { let messages = post_login_messages(true, 4321, 3, 7); diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs index b936d40..013ecf9 100644 --- a/soulseek-rs-lib/src/client/distributed.rs +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -20,6 +20,13 @@ use std::time::{Duration, Instant}; const LINK_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); /// How often an idle link looks at its cancel flag. const CANCEL_POLL: Duration = Duration::from_secs(1); +/// A parent relays the network's whole search stream; one that says nothing +/// for this long is gone, and holding its slot would block every later +/// candidate. +const LINK_IDLE: Duration = Duration::from_mins(1); +/// Moves of our branch are told to the server at most this often; a parent +/// that flaps its level must not make us flood the server. +const MOVE_INTERVAL: Duration = Duration::from_secs(1); /// A distributed frame is a search or a branch fact; anything bigger is not /// one of ours. const MAX_FRAME: usize = 64 * 1024; @@ -82,6 +89,10 @@ pub struct Leaf { candidates: HashMap, next_link: u64, searches: (Instant, u32), + /// When we last told the server we moved, and whether a move since then + /// is still waiting to be told. + moved: Option, + move_pending: bool, } impl Leaf { @@ -94,6 +105,8 @@ impl Leaf { candidates: HashMap::new(), next_link: 1, searches: (Instant::now(), 0), + moved: None, + move_pending: false, } } @@ -163,7 +176,7 @@ impl Leaf { } let candidate = self.candidates.get_mut(username)?; candidate.level = u32::try_from(level).ok(); - if level == 0 && candidate.root.is_none() { + if level == 0 { candidate.root = Some(username.to_string()); } self.moved_with_parent(username) @@ -251,8 +264,24 @@ impl Leaf { if branch == self.branch { return None; } - self.branch = branch.clone(); - Some(branch) + self.branch = branch; + self.move_pending = true; + self.due_announcement(Instant::now()) + } + + /// The branch to tell the server about, if a move is waiting and the + /// last one was told at least [`MOVE_INTERVAL`] ago. + pub fn due_announcement(&mut self, now: Instant) -> Option { + if !self.move_pending + || self + .moved + .is_some_and(|at| now.duration_since(at) < MOVE_INTERVAL) + { + return None; + } + self.move_pending = false; + self.moved = Some(now); + Some(self.branch.clone()) } fn cancel_candidates(&mut self) { @@ -328,18 +357,25 @@ fn link( .get_buffer(), )?; let mut reader = MessageReader::new(); + let mut heard = Instant::now(); loop { if dial.cancel.load(Ordering::Relaxed) { return Ok(()); } match reader.read_from_socket(&mut stream) { - Ok(()) => {} + Ok(()) => heard = Instant::now(), Err(e) if matches!( e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut ) => { + if heard.elapsed() > LINK_IDLE { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "silent parent", + )); + } continue; } Err(e) => return Err(e), @@ -516,7 +552,7 @@ mod tests { } #[test] - fn the_parents_later_moves_move_us_too() { + fn the_parents_later_moves_move_us_too_but_are_told_once_a_second() { let mut leaf = leaf(); let alice = dial(&mut leaf, &["alice"])["alice"].link; leaf.branch_level("alice", alice, 0); @@ -528,9 +564,36 @@ mod tests { ); assert_eq!( leaf.branch_level("alice", alice, 2), + None, + "a second move within the second waits" + ); + assert_eq!(leaf.branch(), Branch::under("grand", 3), "but it took"); + assert_eq!(leaf.due_announcement(Instant::now()), None); + assert_eq!( + leaf.due_announcement(Instant::now() + MOVE_INTERVAL), Some(Branch::under("grand", 3)) ); - assert_eq!(leaf.branch(), Branch::under("grand", 3)); + assert_eq!( + leaf.due_announcement(Instant::now() + MOVE_INTERVAL), + None, + "told once" + ); + } + + #[test] + fn a_parent_that_becomes_a_root_is_the_root_from_then_on() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_root("alice", alice, "grand"); + leaf.branch_level("alice", alice, 5); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("grand", 6)) + ); + assert_eq!( + leaf.branch_level("alice", alice, 0), + Some(Branch::under("alice", 1)) + ); } #[test] diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 160e7a6..5dc3a34 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -388,7 +388,8 @@ pub struct ClientContext { /// Peer listen addresses learned from GetPeerAddress responses. peer_addresses: HashMap, /// Peer messages waiting for a control connection to that peer, and - /// since when: a peer that never answers must not keep them forever. + /// when the last was queued: a peer that never answers must not keep + /// them forever. pending_peer_messages: HashMap)>, /// Uploads we have offered, keyed by our transfer token. @@ -741,11 +742,12 @@ impl ClientContext { username: &str, message: crate::message::Message, ) { - self.pending_peer_messages + let entry = self + .pending_peer_messages .entry(username.to_string()) - .or_insert_with(|| (Instant::now(), Vec::new())) - .1 - .push(message); + .or_insert_with(|| (Instant::now(), Vec::new())); + entry.0 = Instant::now(); + entry.1.push(message); } /// Remove and return the messages queued for `username`. diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 6d47d09..bedf5a3 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -26,6 +26,15 @@ impl Client { Self::sweep_stale_offers(&client_context); if let Ok(mut ctx) = client_context.write_safe() { ctx.expire_pending_peer_messages(Instant::now()); + if let Some(branch) = + ctx.leaf.due_announcement(Instant::now()) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); + } } } let operation = match next {