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); +}