From 460031b532e4b2b322c0a3a9d8806e7f3443a620 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Wed, 9 Sep 2026 16:47:55 +0000 Subject: [PATCH] feat(distributed): serve children in the search tree MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The client read the distributed network but never carried it: a peer dialling with a D connection was logged and dropped, and AcceptChildren was hardcoded false, so this client took from the tree without ever relaying a search for anyone else. With accept_children set, peers hang from us: each is told where our branch sits when it connects and again when the branch moves, every search that reaches us is passed down, and a child that goes away loses its slot. Ten at a time, matching Nicotine+, since each costs a socket and a copy of the network's whole search stream — which is also why it is opt-in, through the new accept_children config key (ignored without a listener, since a child would have nothing to dial). --- soulseek-rs-lib/examples/stress.rs | 1 + soulseek-rs-lib/src/client/children.rs | 278 +++++++++++++++++++++ soulseek-rs-lib/src/client/distributed.rs | 23 +- soulseek-rs-lib/src/client/mod.rs | 36 ++- soulseek-rs-lib/src/client/operations.rs | 76 ++++-- soulseek-rs-lib/src/message/distributed.rs | 9 +- soulseek-rs-lib/src/peer/listen.rs | 10 +- soulseek-rs-lib/tests/e2e.rs | 176 +++++++++++++ soulseek-rs-lib/tests/resume_e2e.rs | 1 + soulseek-rs/src/persist/config.rs | 71 +++++- soulseek-rs/src/run.rs | 3 + soulseek-rs/tests/common/mod.rs | 1 + 12 files changed, 648 insertions(+), 37 deletions(-) create mode 100644 soulseek-rs-lib/src/client/children.rs diff --git a/soulseek-rs-lib/examples/stress.rs b/soulseek-rs-lib/examples/stress.rs index 6952ce1..81c6a8f 100644 --- a/soulseek-rs-lib/examples/stress.rs +++ b/soulseek-rs-lib/examples/stress.rs @@ -1026,6 +1026,7 @@ fn main() { enable_listen: true, listen_port: client_port, shared_directories: vec![share_dir.display().to_string()], + accept_children: false, version: ClientVersion::default(), }); client.connect().expect("client connect"); diff --git a/soulseek-rs-lib/src/client/children.rs b/soulseek-rs-lib/src/client/children.rs new file mode 100644 index 0000000..e22cfa1 --- /dev/null +++ b/soulseek-rs-lib/src/client/children.rs @@ -0,0 +1,278 @@ +//! Serving children: the other half of the distributed search network. +//! +//! A leaf only reads the tree; a parent also carries it. When the client is +//! configured to accept children, peers dial us with a `D` connection, we tell +//! each one where our branch sits, and every search that reaches us is passed +//! down to all of them. That is the whole duty — a child answers searches +//! itself, and reports its own place to the server. +//! +//! Accepting children is opt-in ([`crate::ClientSettings::accept_children`]): +//! it costs a socket and the network's whole search stream per child, which is +//! not something a client should take on without being asked to. + +use std::collections::HashMap; +use std::io::Write; +use std::net::TcpStream; + +use crate::message::Message; +use crate::message::distributed; +use crate::{debug, trace}; + +/// How many children to carry at once. Nicotine+ offers ten; the cost is one +/// socket and one copy of every search each, so the cap is what keeps a +/// parent's outbound traffic bounded. +pub const MAX_CHILDREN: usize = 10; + +/// The children hanging from us, keyed by username. +pub struct Children { + links: HashMap, + max: usize, + accepting: bool, +} + +impl Default for Children { + fn default() -> Self { + Self::new(false) + } +} + +impl Children { + #[must_use] + pub fn new(accepting: bool) -> Self { + Self { + links: HashMap::new(), + max: MAX_CHILDREN, + accepting, + } + } + + /// Whether we take children at all, and still have room for one. + #[must_use] + pub fn has_room(&self) -> bool { + self.accepting && self.links.len() < self.max + } + + /// Whether the client accepts children at all, regardless of how full it + /// is — what the server is told in `AcceptChildren` (code 100). + #[must_use] + pub const fn accepting(&self) -> bool { + self.accepting + } + + #[must_use] + pub fn len(&self) -> usize { + self.links.len() + } + + #[must_use] + pub fn is_empty(&self) -> bool { + self.links.is_empty() + } + + #[must_use] + pub fn usernames(&self) -> Vec { + let mut names: Vec = self.links.keys().cloned().collect(); + names.sort(); + names + } + + /// Take on `username` as a child. Refused when we are not accepting or + /// are full; a second connection from a child we already carry replaces + /// the first, since the peer only reconnects when it thinks the old one + /// is gone. + pub fn accept(&mut self, username: &str, stream: TcpStream) -> bool { + if !self.accepting { + return false; + } + if self.links.len() >= self.max && !self.links.contains_key(username) { + debug!( + "[distributed] refusing child {}: already carrying {}", + username, + self.links.len() + ); + return false; + } + self.links.insert(username.to_string(), stream); + true + } + + pub fn remove(&mut self, username: &str) { + if self.links.remove(username).is_some() { + debug!("[distributed] child {} is gone", username); + } + } + + /// Send one frame to `username`, dropping the child if the write fails. + pub fn send_to(&mut self, username: &str, message: &Message) { + let failed = match self.links.get_mut(username) { + Some(stream) => stream.write_all(&message.get_buffer()).is_err(), + None => return, + }; + if failed { + self.remove(username); + } + } + + /// Send one frame to every child, dropping those whose write fails: a + /// child that has gone away must not keep its slot. + pub fn broadcast(&mut self, message: &Message) { + let bytes = message.get_buffer(); + let mut dead = Vec::new(); + for (username, stream) in &mut self.links { + if stream.write_all(&bytes).is_err() { + dead.push(username.clone()); + } + } + for username in dead { + self.remove(&username); + } + } + + /// Tell `username` where our branch sits — what a child needs before it + /// can report its own place to the server. + pub fn send_stance_to(&mut self, username: &str, root: &str, level: u32) { + trace!( + "[distributed] telling {} we sit at {}/{}", + username, root, level + ); + self.send_to(username, &distributed::build_branch_root(root)); + self.send_to( + username, + &distributed::build_branch_level( + i32::try_from(level).unwrap_or(i32::MAX), + ), + ); + } + + /// Tell every child our branch moved. + pub fn broadcast_stance(&mut self, root: &str, level: u32) { + self.broadcast(&distributed::build_branch_root(root)); + self.broadcast(&distributed::build_branch_level( + i32::try_from(level).unwrap_or(i32::MAX), + )); + } + + /// Pass a search down the tree. + pub fn broadcast_search( + &mut self, + username: &str, + token: u32, + query: &str, + ) { + if self.links.is_empty() { + return; + } + self.broadcast(&distributed::build_search(username, token, query)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Read; + use std::net::{TcpListener, TcpStream}; + + /// A connected socket pair: the child's end and ours. + fn pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let addr = listener.local_addr().expect("addr"); + let ours = TcpStream::connect(addr).expect("connect"); + let (theirs, _) = listener.accept().expect("accept"); + (theirs, ours) + } + + fn read_frames(stream: &mut TcpStream, bytes: usize) -> Vec { + stream + .set_read_timeout(Some(std::time::Duration::from_secs(2))) + .expect("timeout"); + let mut buf = vec![0u8; bytes]; + stream.read_exact(&mut buf).expect("read"); + buf + } + + #[test] + fn a_client_that_does_not_accept_children_takes_none() { + let (_child, ours) = pair(); + let mut children = Children::new(false); + assert!(!children.accept("alice", ours)); + assert!(children.is_empty()); + assert!(!children.has_room()); + } + + #[test] + fn children_are_taken_up_to_the_cap() { + let mut children = Children::new(true); + // The child ends are kept alive so the links stay up; the parent + // drops a child whose socket closes. + let mut ends = Vec::new(); + for i in 0..MAX_CHILDREN { + let (child, ours) = pair(); + ends.push(child); + assert!(children.accept(&format!("user{i}"), ours)); + } + assert_eq!(ends.len(), MAX_CHILDREN, "every link is still held"); + assert!(!children.has_room(), "a full parent has no room"); + + let (_late, ours) = pair(); + assert!(!children.accept("one_too_many", ours)); + assert_eq!(children.len(), MAX_CHILDREN); + } + + #[test] + fn a_reconnecting_child_replaces_its_own_link() { + let mut children = Children::new(true); + let (_first, ours) = pair(); + assert!(children.accept("alice", ours)); + let (_second, again) = pair(); + assert!(children.accept("alice", again)); + assert_eq!(children.usernames(), vec!["alice".to_string()]); + } + + #[test] + fn a_search_reaches_every_child() { + let mut children = Children::new(true); + let (mut first, ours) = pair(); + let (mut second, theirs) = pair(); + children.accept("alice", ours); + children.accept("bob", theirs); + + children.broadcast_search("seeker", 42, "jazz"); + + let expected = + distributed::build_search("seeker", 42, "jazz").get_buffer(); + assert_eq!(read_frames(&mut first, expected.len()), expected); + assert_eq!(read_frames(&mut second, expected.len()), expected); + } + + #[test] + fn a_child_that_hung_up_loses_its_slot() { + let mut children = Children::new(true); + let (child, ours) = pair(); + children.accept("alice", ours); + drop(child); + + // The first write after the peer is gone may still be buffered; the + // second is the one that fails. Either way the child must not keep + // its slot forever. + for _ in 0..50 { + children.broadcast_search("seeker", 1, "q"); + if children.is_empty() { + break; + } + } + assert!(children.is_empty(), "a dead child is dropped"); + } + + #[test] + fn a_child_is_told_where_our_branch_sits() { + let mut children = Children::new(true); + let (mut child, ours) = pair(); + children.accept("alice", ours); + + children.send_stance_to("alice", "rooty", 3); + + let mut expected = distributed::build_branch_root("rooty").get_buffer(); + expected.extend(distributed::build_branch_level(3).get_buffer()); + assert_eq!(read_frames(&mut child, expected.len()), expected); + } +} diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs index 013ecf9..cfc89de 100644 --- a/soulseek-rs-lib/src/client/distributed.rs +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -297,15 +297,34 @@ pub fn announce( server: Option<&Sender>, branch: &Branch, has_parent: bool, + accept_children: bool, ) { let Some(server) = server else { return; }; - for message in distributed::stance(&branch.root, branch.level, has_parent) { + for message in distributed::stance( + &branch.root, + branch.level, + has_parent, + accept_children, + ) { let _ = server.send(ServerMessage::SendMessage(message)); } } +/// Tell the server where we sit, and tell our children too: a branch that +/// moved leaves every child reporting a stale place until they hear it. +pub fn announce_move( + ctx: &mut super::ClientContext, + branch: &Branch, + has_parent: bool, +) { + let server = ctx.server_sender.clone(); + let accepting = ctx.children.accepting(); + announce(server.as_ref(), branch, has_parent, accepting); + ctx.children.broadcast_stance(&branch.root, branch.level); +} + /// 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) { @@ -668,7 +687,7 @@ mod tests { 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); + announce(Some(&tx), &Branch::under("rooty", 4), true, false); let mut sent: Vec = Vec::new(); while let Ok(ServerMessage::SendMessage(m)) = rx.try_recv() { sent.push(m); diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index b108b2b..abaf670 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -191,6 +191,11 @@ pub struct ClientSettings { /// Directories whose files are shared with (uploaded to) other peers. /// Empty means nothing is shared. pub shared_directories: Vec, + /// Whether to serve children in the distributed search network: peers + /// hang from us with a `D` connection and we pass every search we receive + /// down to them. Off by default — it costs a socket and the network's + /// whole search stream per child. + pub accept_children: bool, /// The version reported to the server on login. Defaults to the /// soulseek-rs major version with minor version 1 ("unidentified"); /// clients built on this library should reserve their own minor @@ -223,6 +228,7 @@ impl Default for ClientSettings { enable_listen: true, listen_port: DEFAULT_LISTEN_PORT, shared_directories: Vec::new(), + accept_children: false, version: ClientVersion::default(), } } @@ -343,6 +349,12 @@ pub enum ClientOperation { PossibleParents(Vec<(String, String, u16)>), /// The server told us to drop our parent. ResetDistributed, + /// A peer dialled us with a `D` connection, asking to hang from us in the + /// distributed search network. + ChildConnected { + username: String, + stream: std::net::TcpStream, + }, /// Recommendations from the server: either from our own interests /// (`global` false, code 54) or server-wide (`global` true, code 56). Recommendations { @@ -405,6 +417,8 @@ pub struct ClientContext { operations: Option>, /// Our place in the distributed search network. leaf: distributed::Leaf, + /// The children hanging from us, when the client serves any. + pub(crate) children: children::Children, searches: HashMap, private_messages: Vec, /// Correlation tokens for server-brokered (firewalled) connections, mapping @@ -565,6 +579,7 @@ impl ClientContext { server_sender: None, operations: None, leaf: distributed::Leaf::new(""), + children: children::Children::default(), searches: HashMap::new(), private_messages: Vec::new(), pending_connect_tokens: HashMap::new(), @@ -1180,14 +1195,14 @@ impl Client { #[must_use] pub fn with_settings(settings: ClientSettings) -> Self { logger::init(); + let mut context = ClientContext::for_user(&settings.username); + context.children = children::Children::new(settings.accept_children); Self { enable_listen: settings.enable_listen, listen_port: settings.listen_port, bound_port: None, address: settings.server_address, - context: Arc::new(RwLock::new(ClientContext::for_user( - &settings.username, - ))), + context: Arc::new(RwLock::new(context)), username: settings.username, password: settings.password, version: settings.version, @@ -1197,6 +1212,20 @@ impl Client { } } + /// The children currently hanging from us in the distributed search + /// network, by username. Empty unless the client was built with + /// [`ClientSettings::accept_children`]. + #[must_use] + pub fn children(&self) -> Vec { + match self.context.read_safe() { + Ok(ctx) => ctx.children.usernames(), + Err(e) => { + error!("[client] children: {}", e); + Vec::new() + } + } + } + /// The username we log in as, for attributing our own messages. #[must_use] pub fn username(&self) -> &str { @@ -1334,6 +1363,7 @@ impl Client { } } +mod children; mod connection; mod distributed; mod downloads; diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index b2dee36..094a72d 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -29,10 +29,8 @@ impl Client { if let Some(branch) = ctx.leaf.due_announcement(Instant::now()) { - super::distributed::announce( - ctx.server_sender.as_ref(), - &branch, - true, + super::distributed::announce_move( + &mut ctx, &branch, true, ); } } @@ -539,6 +537,16 @@ impl Client { token, query, } => { + // Pass it down the tree first: carrying the search + // stream is a parent's whole duty, and it is owed + // even for a search we ourselves cannot answer. + if let Ok(mut ctx) = client_context.write_safe() + && !ctx.children.is_empty() + { + ctx.children + .broadcast_search(&username, token, &query); + } + // Don't answer our own distributed search. if username == own_username { continue; @@ -679,10 +687,8 @@ impl Client { if let Some(branch) = ctx.leaf.branch_level(&parent, link, level) { - super::distributed::announce( - ctx.server_sender.as_ref(), - &branch, - true, + super::distributed::announce_move( + &mut ctx, &branch, true, ); } } @@ -697,10 +703,8 @@ impl Client { if let Some(branch) = ctx.leaf.branch_root(&parent, link, &root) { - super::distributed::announce( - ctx.server_sender.as_ref(), - &branch, - true, + super::distributed::announce_move( + &mut ctx, &branch, true, ); } } @@ -717,10 +721,8 @@ impl Client { if let Some(branch) = ctx.leaf.search_from(&parent, link) { - super::distributed::announce( - ctx.server_sender.as_ref(), - &branch, - true, + super::distributed::announce_move( + &mut ctx, &branch, true, ); } if ctx.leaf.is_parent(&parent, link) @@ -739,10 +741,9 @@ impl Client { continue; }; if ctx.leaf.closed(&parent, link) { - super::distributed::announce( - ctx.server_sender.as_ref(), - &ctx.leaf.branch(), - false, + let branch = ctx.leaf.branch(); + super::distributed::announce_move( + &mut ctx, &branch, false, ); } } @@ -753,10 +754,9 @@ impl Client { continue; }; ctx.leaf.reset(); - super::distributed::announce( - ctx.server_sender.as_ref(), - &ctx.leaf.branch(), - false, + let branch = ctx.leaf.branch(); + super::distributed::announce_move( + &mut ctx, &branch, false, ); } ClientOperation::PrivilegedUsers(users) => { @@ -816,6 +816,34 @@ impl Client { ctx.apply_user_interests(interests); } } + ClientOperation::ChildConnected { username, stream } => { + // A peer wants to hang from us. Take it on if this + // client serves children and has room, then tell it + // where our branch sits so it can report its own + // place; otherwise the socket is dropped here and the + // peer looks for another parent. + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if !ctx.children.has_room() { + continue; + } + if stream.set_nodelay(true).is_err() { + continue; + } + if ctx.children.accept(&username, stream) { + let branch = ctx.leaf.branch(); + ctx.children.send_stance_to( + &username, + &branch.root, + branch.level, + ); + debug!( + "[distributed] carrying {} children", + ctx.children.len() + ); + } + } ClientOperation::ExcludedSearchPhrases(phrases) => { if let Ok(mut ctx) = client_context.write_safe() { ctx.set_excluded_search_phrases(phrases); diff --git a/soulseek-rs-lib/src/message/distributed.rs b/soulseek-rs-lib/src/message/distributed.rs index 2ae015c..9917ce6 100644 --- a/soulseek-rs-lib/src/message/distributed.rs +++ b/soulseek-rs-lib/src/message/distributed.rs @@ -108,12 +108,17 @@ 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 { +pub fn stance( + root: &str, + level: u32, + has_parent: bool, + accept_children: 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), + MessageFactory::build_accept_children(accept_children), ] } diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index eff0b7b..2726a8a 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -378,9 +378,17 @@ fn handle_incoming_connection( deadline, ), ConnectionType::D => { + // A child asking to hang from us. The client decides whether it + // serves children at all; here we only hand the socket over. debug!( - "[listener:{peer_ip}:{peer_port}] connection type is D, not supported yet, closing connection. " + "[listener:{peer_ip}:{peer_port}] distributed child {} offered", + init_data.username ); + let _ = + context.client_sender.send(ClientOperation::ChildConnected { + username: init_data.username, + stream, + }); } } } diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index f8f2a91..64e08b0 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -118,6 +118,7 @@ impl TestServer { enable_listen: false, listen_port: 0, shared_directories: Vec::new(), + accept_children: false, version: ClientVersion::default(), } } @@ -4450,6 +4451,7 @@ fn an_excluded_phrase_is_refused_before_a_search_is_spent() { enable_listen: false, listen_port: 0, shared_directories: Vec::new(), + accept_children: false, version: ClientVersion::default(), }); client.connect().expect("connect to stub"); @@ -4808,3 +4810,177 @@ fn a_wishlist_search_is_answered_like_any_other() { let _ = std::fs::remove_dir_all(share_dir); } + +// --------------------------------------------------------------------------- +// The other half of the tree: serving children. +// +// A client that accepts children takes `D` connections from peers, tells each +// where our branch sits, and passes every search it receives down to them. +// --------------------------------------------------------------------------- + +/// Read distributed frames from a child link until `pick` matches or time runs +/// out. +fn read_distributed_until( + stream: &mut TcpStream, + timeout: Duration, + mut pick: impl FnMut(&distributed::Distributed) -> Option, +) -> Option { + stream + .set_read_timeout(Some(Duration::from_millis(500))) + .ok()?; + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + // read_framed already returns the frame with its length prefix, + // which is exactly what the parser expects. + let Ok(mut frame) = read_framed(stream) else { + continue; + }; + if let Some(parsed) = distributed::parse(&mut frame) + && let Some(found) = pick(&parsed) + { + return Some(found); + } + } + None +} + +/// Dial `port` as a distributed child called `username`. +fn dial_as_child(port: u16, username: &str) -> std::io::Result { + let mut child = + connect_retry(&format!("127.0.0.1:{port}"), Duration::from_secs(5))?; + child.write_all(&peer_init_bytes(username, "D", 0))?; + child.flush()?; + Ok(child) +} + +#[test] +fn a_parent_tells_a_child_where_it_sits_and_passes_searches_down() { + let server = server_or_skip!(); + + let listen_port = free_port().expect("free listen port"); + let mut parent = Client::with_settings(ClientSettings { + accept_children: true, + ..server.listening_settings("e2e_par_parent", "pw", listen_port) + }); + parent.connect().expect("parent connect"); + assert!(parent.login().expect("parent login")); + + let mut child = + dial_as_child(listen_port, "e2e_par_child").expect("child dials"); + + // The child is told our branch before anything else: without it, it has + // nothing to report to the server as its own place. + let root = read_distributed_until( + &mut child, + Duration::from_secs(10), + |f| match f { + distributed::Distributed::BranchRoot(root) => Some(root.clone()), + _ => None, + }, + ) + .expect("a child should be told our branch root"); + assert_eq!( + root, "e2e_par_parent", + "a parentless client is its own root" + ); + let level = read_distributed_until( + &mut child, + Duration::from_secs(10), + |f| match f { + distributed::Distributed::BranchLevel(level) => Some(*level), + _ => None, + }, + ) + .expect("a child should be told our branch level"); + assert_eq!(level, 0); + + let deadline = Instant::now() + Duration::from_secs(5); + while parent.children().is_empty() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + assert_eq!(parent.children(), vec!["e2e_par_child".to_string()]); + + // A search the server hands us must reach the child, whether or not we + // can answer it ourselves — carrying the stream is the parent's duty. + let addr = format!("{}:{}", server.host, server.port); + let mut searcher = + login_raw(&addr, "e2e_par_searcher", "pw").expect("searcher login"); + let token = 515_151_u32; + searcher + .write_all( + &MessageFactory::build_file_search_message( + token, + "e2e_child_relay_probe", + ) + .get_buffer(), + ) + .expect("send a search"); + searcher.flush().expect("flush the search"); + + let relayed = read_distributed_until( + &mut child, + Duration::from_secs(20), + |f| match f { + distributed::Distributed::Search { + username, + token: got, + query, + } if *got == token => Some((username.clone(), query.clone())), + _ => None, + }, + ) + .expect("the search should be passed down to the child"); + assert_eq!( + relayed, + ( + "e2e_par_searcher".to_string(), + "e2e_child_relay_probe".to_string() + ) + ); + + // A child that hangs up loses its place. + drop(child); + let deadline = Instant::now() + Duration::from_secs(20); + while !parent.children().is_empty() && Instant::now() < deadline { + searcher + .write_all( + &MessageFactory::build_file_search_message( + token + 1, + "e2e_child_relay_probe", + ) + .get_buffer(), + ) + .expect("send another search"); + std::thread::sleep(Duration::from_millis(500)); + } + assert!( + parent.children().is_empty(), + "a child that went away should not keep its slot" + ); +} + +#[test] +fn a_client_that_does_not_serve_children_turns_one_away() { + let server = server_or_skip!(); + + // The default: a leaf. A peer that dials with a D connection is dropped, + // and never told a branch it could hang from. + let listen_port = free_port().expect("free listen port"); + let mut leaf = Client::with_settings(server.listening_settings( + "e2e_leafonly", + "pw", + listen_port, + )); + leaf.connect().expect("leaf connect"); + assert!(leaf.login().expect("leaf login")); + + let mut child = + dial_as_child(listen_port, "e2e_leafonly_child").expect("child dials"); + + let told = + read_distributed_until(&mut child, Duration::from_secs(3), |f| { + matches!(f, distributed::Distributed::BranchRoot(_)).then_some(()) + }); + assert!(told.is_none(), "a leaf must not take on a child"); + assert!(leaf.children().is_empty()); +} diff --git a/soulseek-rs-lib/tests/resume_e2e.rs b/soulseek-rs-lib/tests/resume_e2e.rs index 2f5cf55..7271c25 100644 --- a/soulseek-rs-lib/tests/resume_e2e.rs +++ b/soulseek-rs-lib/tests/resume_e2e.rs @@ -67,6 +67,7 @@ impl Subject { enable_listen: true, listen_port: 0, shared_directories: Vec::new(), + accept_children: false, version: ClientVersion::default(), }); client.connect().expect("connect subject"); diff --git a/soulseek-rs/src/persist/config.rs b/soulseek-rs/src/persist/config.rs index 3781647..6dd6f8c 100644 --- a/soulseek-rs/src/persist/config.rs +++ b/soulseek-rs/src/persist/config.rs @@ -12,6 +12,11 @@ pub struct FileConfig { pub server: Option, pub listener_port: Option, pub disable_listener: Option, + /// Serve children in the distributed search network: other peers hang + /// from this client and every search it receives is passed down to them. + /// Off by default — it costs a socket and the network's whole search + /// stream per child, and it needs the listener. + pub accept_children: Option, pub download_dir: Option, /// Single shared folder (also what `--shared-dir` sets). Prefer /// `shared_dirs` for multiple; both may be combined. @@ -46,6 +51,7 @@ impl FileConfig { "server", "listener_port", "disable_listener", + "accept_children", "download_dir", "shared_dirs", "max_concurrent_downloads", @@ -64,6 +70,7 @@ impl FileConfig { "server" => self.server.clone(), "listener_port" => self.listener_port.map(|v| v.to_string()), "disable_listener" => self.disable_listener.map(|v| v.to_string()), + "accept_children" => self.accept_children.map(|v| v.to_string()), "download_dir" => self.download_dir.clone(), "shared_dirs" => { let dirs = self.shared_dirs.clone().unwrap_or_default(); @@ -127,6 +134,22 @@ impl FileConfig { }) }; } + "accept_children" => { + self.accept_children = if clear { + None + } else { + Some(match value { + "true" | "1" | "yes" => true, + "false" | "0" | "no" => false, + other => { + return Err(format!( + "accept_children wants true or false, got \ + '{other}'" + )); + } + }) + }; + } "shared_dirs" => { self.shared_dirs = (!clear).then(|| { value @@ -251,6 +274,7 @@ pub struct Resolved { pub server: String, pub listener_port: u16, pub disable_listener: bool, + pub accept_children: bool, pub download_dir: String, pub shared_dirs: Vec, pub max_concurrent_downloads: usize, @@ -283,6 +307,11 @@ pub fn resolve(cli: &crate::cli::Cli, file: &FileConfig) -> Resolved { .clone() .or_else(|| file.download_dir.clone()) .unwrap_or_else(super::paths::default_download_dir); + let disable_listener = if cli.listener { + false + } else { + cli.no_listener || file.disable_listener.unwrap_or(false) + }; Resolved { username: cli.username.clone().or_else(|| file.username.clone()), server: cli @@ -294,11 +323,11 @@ pub fn resolve(cli: &crate::cli::Cli, file: &FileConfig) -> Resolved { .listener_port .or(file.listener_port) .unwrap_or(DEFAULT_LISTENER_PORT), - disable_listener: if cli.listener { - false - } else { - cli.no_listener || file.disable_listener.unwrap_or(false) - }, + disable_listener, + // A client with no listener has nothing for a child to dial, so the + // setting only takes effect alongside one. + accept_children: file.accept_children.unwrap_or(false) + && !disable_listener, download_dir: download_dir.clone(), shared_dirs: resolve_shared_dirs(cli, file, &download_dir), max_concurrent_downloads: cli @@ -496,6 +525,37 @@ mod tests { assert!(!resolve(&cli, &FileConfig::default()).disable_listener); } + #[test] + fn accepting_children_needs_a_listener() { + // Serving children means peers dial us; with the listener off there + // is nothing to dial, so the setting must not survive resolution. + let file = FileConfig { + accept_children: Some(true), + ..FileConfig::default() + }; + assert!(resolve(&bare_cli(), &file).accept_children); + + let off = FileConfig { + disable_listener: Some(true), + ..file.clone() + }; + assert!(!resolve(&bare_cli(), &off).accept_children); + + let mut cli = bare_cli(); + cli.no_listener = true; + assert!(!resolve(&cli, &file).accept_children); + } + + #[test] + fn accept_children_round_trips_through_get_and_set() { + let mut file = FileConfig::default(); + file.set("accept_children", "true").expect("set"); + assert_eq!(file.get("accept_children").as_deref(), Some("true")); + assert!(file.set("accept_children", "maybe").is_err()); + file.set("accept_children", "").expect("clear"); + assert_eq!(file.get("accept_children"), None); + } + #[test] fn file_values_override_defaults() { let file = FileConfig { @@ -503,6 +563,7 @@ mod tests { server: Some("localhost:2242".into()), listener_port: Some(4321), disable_listener: Some(true), + accept_children: None, download_dir: Some("/music".into()), shared_dir: Some("/shared".into()), shared_dirs: None, diff --git a/soulseek-rs/src/run.rs b/soulseek-rs/src/run.rs index af56d40..11c2c20 100644 --- a/soulseek-rs/src/run.rs +++ b/soulseek-rs/src/run.rs @@ -259,6 +259,7 @@ fn context( enable_listen: !resolved.disable_listener, listen_port: resolved.listener_port, shared_directories: shared_directories(resolved, out), + accept_children: resolved.accept_children, version: ClientVersion::REFERENCE_CLIENT, }, download_dir, @@ -432,6 +433,7 @@ fn run_default_tui( let enable_listen = !resolved.disable_listener; let listen_port = resolved.listener_port; + let accept_children = resolved.accept_children; let make_settings = move |username: String, password: String| ClientSettings { username, @@ -440,6 +442,7 @@ fn run_default_tui( enable_listen, listen_port, shared_directories: shared_directories.clone(), + accept_children, version: ClientVersion::REFERENCE_CLIENT, }; diff --git a/soulseek-rs/tests/common/mod.rs b/soulseek-rs/tests/common/mod.rs index 223131c..03ada03 100644 --- a/soulseek-rs/tests/common/mod.rs +++ b/soulseek-rs/tests/common/mod.rs @@ -137,6 +137,7 @@ pub fn login(host: &str, port: u16, user: &str, shares: Vec) -> Client { enable_listen: true, listen_port: free_port().expect("peer port"), shared_directories: shares, + accept_children: false, version: soulseek_rs::ClientVersion::default(), }); client.connect().expect("peer connect");