diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 74b9e21..b7aefe8 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -2,11 +2,11 @@ use crate::actor::{Actor, ActorHandle, ConnectionState}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::peer::{ - FileSearchResponse, FolderContentsRequest, GetShareFileList, PeerInit, - PlaceInQueueRequest, PlaceInQueueResponse, QueueUploadHandler, - SharedDirectory, SharedFileListResponseHandler, TransferRequest, - TransferResponse, UploadDeniedHandler, UploadFailedHandler, - UserInfoRequest, + FileSearchResponse, FolderContentsRequest, FolderContentsResponseHandler, + GetShareFileList, PeerInit, PlaceInQueueRequest, PlaceInQueueResponse, + QueueUploadHandler, SharedDirectory, SharedFileListResponseHandler, + TransferRequest, TransferResponse, UploadDeniedHandler, + UploadFailedHandler, UserInfoRequest, }; use crate::message::server::MessageFactory; use crate::message::{Handlers, Message, MessageReader, MessageType}; @@ -47,6 +47,12 @@ pub enum PeerMessage { ShareListRequested, /// A peer we are browsing sent us their shared-file listing (code 5). ShareListReceived(Vec), + /// A peer answered our `FolderContentsRequest` with one folder's listing. + FolderContentsReceived { + token: u32, + folder: String, + directories: Vec, + }, /// A peer asked what we say about ourselves (they sent us code 15). UserInfoRequested, /// A peer asked for one folder of our shares (they sent us code 36). @@ -199,6 +205,7 @@ impl PeerActor { handlers.register_handler(TransferResponse); handlers.register_handler(GetShareFileList); handlers.register_handler(FolderContentsRequest); + handlers.register_handler(FolderContentsResponseHandler); handlers.register_handler(UserInfoRequest); handlers.register_handler(UploadDeniedHandler); handlers.register_handler(UploadFailedHandler); @@ -310,6 +317,18 @@ impl PeerActor { directories, }); } + PeerMessage::FolderContentsReceived { + token, + folder, + directories, + } => { + self.forward(ClientOperation::FolderContents { + username: self.peer_username(), + token, + folder, + directories, + }); + } PeerMessage::UserInfoRequested => { self.forward(ClientOperation::UserInfoRequested { requester_key: self.peer_username(), diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index bd43d56..26b76dd 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -233,6 +233,11 @@ pub enum ServerMessage { }, /// Phrases the server refuses to search for (code 160). ExcludedSearchPhrases(Vec), + /// The upload speed a client needs before the server lets it carry + /// children (code 83). + ParentMinSpeed(u32), + /// The divisor turning that speed into a child count (code 84). + ParentSpeedRatio(u32), /// Who may enter a private room (code 133). PrivateRoomMembers { room: String, @@ -780,6 +785,14 @@ impl ServerActor { ServerMessage::CantCreateRoom { room } => { self.forward_room_event(RoomEvent::CantCreate { room }); } + ServerMessage::ParentMinSpeed(speed) => { + self.forward_to_client(ClientOperation::ParentMinSpeed(speed)); + } + ServerMessage::ParentSpeedRatio(ratio) => { + self.forward_to_client(ClientOperation::ParentSpeedRatio( + ratio, + )); + } ServerMessage::ExcludedSearchPhrases(phrases) => { self.forward_to_client(ClientOperation::ExcludedSearchPhrases( phrases, diff --git a/soulseek-rs-lib/src/client/children.rs b/soulseek-rs-lib/src/client/children.rs index e22cfa1..0a054d0 100644 --- a/soulseek-rs-lib/src/client/children.rs +++ b/soulseek-rs-lib/src/client/children.rs @@ -6,6 +6,14 @@ //! down to all of them. That is the whole duty — a child answers searches //! itself, and reports its own place to the server. //! +//! How many children to carry follows Nicotine+: the server announces the +//! upload speed a parent needs (`ParentMinSpeed`, code 83) and the divisor +//! turning speed into a child count (`ParentSpeedRatio`, code 84), and a +//! client whose recorded speed is below the minimum carries none. Until the +//! server has reported our own speed there is nothing to apply the formula to, +//! so the configured cap stands — which is how slskd runs the whole time, with +//! an operator-set limit and no speed gate. +//! //! 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. @@ -18,16 +26,39 @@ 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. +/// How many children to carry at once. Nicotine+ caps its speed-derived limit +/// here too, for the same reason: each child is a socket and a copy of the +/// network's whole search stream. pub const MAX_CHILDREN: usize = 10; +/// The child limit the server's figures imply, following Nicotine+: +/// `speed / ratio / 100`, capped, and none at all below the minimum speed the +/// server set. `None` when the server has not reported our speed yet, which +/// leaves the caller's own cap standing. +#[must_use] +pub fn limit_from_speed( + own_speed: Option, + min_speed: u32, + ratio: u32, +) -> Option { + let speed = own_speed?; + if ratio == 0 || speed < min_speed { + return Some(0); + } + Some((speed / ratio / 100) as usize) +} + /// The children hanging from us, keyed by username. pub struct Children { links: HashMap, + /// The cap in force: [`MAX_CHILDREN`], lowered to what the server's + /// figures allow once it has reported our speed. max: usize, accepting: bool, + /// Whether anything is feeding us the search stream — a parent of our own, + /// or the server. Nicotine+ refuses children without one, and rightly: + /// a child hanging from a branch that receives nothing gets nothing. + fed: bool, } impl Default for Children { @@ -43,20 +74,40 @@ impl Children { links: HashMap::new(), max: MAX_CHILDREN, accepting, + fed: false, } } - /// 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 + /// Apply the server's figures: `own_speed` is what it records for us, and + /// the other two are what it announced in codes 83 and 84. + pub fn set_limits( + &mut self, + own_speed: Option, + min_speed: u32, + ratio: u32, + ) { + self.max = limit_from_speed(own_speed, min_speed, ratio) + .map_or(MAX_CHILDREN, |limit| limit.min(MAX_CHILDREN)); + debug!("[distributed] child limit is now {}", self.max); + } + + /// Say whether something is feeding us the search stream. A client with + /// nothing to relay takes no children, and drops any it has. + pub const fn set_fed(&mut self, fed: bool) { + self.fed = fed; } - /// 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 + pub const fn is_fed(&self) -> bool { + self.fed + } + + /// Whether we would take another child right now: we serve children, we + /// have something to relay, and we are under the cap. This is what the + /// server is told in `AcceptChildren` (code 100). + #[must_use] + pub fn has_room(&self) -> bool { + self.accepting && self.fed && self.links.len() < self.max } #[must_use] @@ -76,19 +127,19 @@ impl Children { 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. + /// Take on `username` as a child. Refused when we serve no children, have + /// nothing to relay, are full, or already carry that user. pub fn accept(&mut self, username: &str, stream: TcpStream) -> bool { - if !self.accepting { + if !self.has_room() { return false; } - if self.links.len() >= self.max && !self.links.contains_key(username) { + // A second connection from a child we already carry is refused, as + // Nicotine+ does: the link we hold is the live one, and dropping it + // for an unproven second would cost that child its search stream. + if self.links.contains_key(username) { debug!( - "[distributed] refusing child {}: already carrying {}", - username, - self.links.len() + "[distributed] refusing child {}: already carried", + username ); return false; } @@ -96,6 +147,15 @@ impl Children { true } + /// Drop every child, closing their links: what a client does when it can + /// no longer feed them. + pub fn drop_all(&mut self) { + if !self.links.is_empty() { + debug!("[distributed] dropping {} children", self.links.len()); + } + self.links.clear(); + } + pub fn remove(&mut self, username: &str) { if self.links.remove(username).is_some() { debug!("[distributed] child {} is gone", username); @@ -159,7 +219,7 @@ impl Children { token: u32, query: &str, ) { - if self.links.is_empty() { + if self.is_empty() { return; } self.broadcast(&distributed::build_search(username, token, query)); @@ -181,6 +241,13 @@ mod tests { (theirs, ours) } + /// A registry that serves children and has something to relay. + fn serving() -> Children { + let mut children = Children::new(true); + children.set_fed(true); + children + } + fn read_frames(stream: &mut TcpStream, bytes: usize) -> Vec { stream .set_read_timeout(Some(std::time::Duration::from_secs(2))) @@ -194,16 +261,42 @@ mod tests { fn a_client_that_does_not_accept_children_takes_none() { let (_child, ours) = pair(); let mut children = Children::new(false); + children.set_fed(true); assert!(!children.accept("alice", ours)); assert!(children.is_empty()); assert!(!children.has_room()); } #[test] - fn children_are_taken_up_to_the_cap() { + fn a_client_nothing_feeds_takes_no_children() { + // Nicotine+ refuses children while it has neither a parent nor the + // server relaying searches: a child would hang from a branch that + // receives nothing. + let (_child, ours) = pair(); 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. + assert!(!children.has_room(), "unfed means no room"); + assert!(!children.accept("alice", ours)); + + children.set_fed(true); + let (_second, again) = pair(); + assert!(children.accept("alice", again)); + } + + #[test] + fn children_are_let_go_when_the_stream_dries_up() { + let mut children = serving(); + let (_child, ours) = pair(); + children.accept("alice", ours); + + children.set_fed(false); + children.drop_all(); + assert!(children.is_empty()); + assert!(!children.has_room()); + } + + #[test] + fn children_are_taken_up_to_the_cap() { + let mut children = serving(); let mut ends = Vec::new(); for i in 0..MAX_CHILDREN { let (child, ours) = pair(); @@ -219,18 +312,64 @@ mod tests { } #[test] - fn a_reconnecting_child_replaces_its_own_link() { - let mut children = Children::new(true); + fn a_second_link_from_a_child_we_carry_is_refused() { + // The link we hold is the live one; dropping it for an unproven + // second would cost that child its search stream. + let mut children = serving(); let (_first, ours) = pair(); assert!(children.accept("alice", ours)); let (_second, again) = pair(); - assert!(children.accept("alice", again)); + assert!(!children.accept("alice", again)); assert_eq!(children.usernames(), vec!["alice".to_string()]); } + #[test] + fn the_child_limit_follows_the_servers_figures() { + // Nicotine+: speed / ratio / 100, capped, and none at all below the + // minimum speed the server set. + assert_eq!(limit_from_speed(Some(50_000), 1_024, 50), Some(10)); + assert_eq!(limit_from_speed(Some(10_000), 1_024, 50), Some(2)); + assert_eq!( + limit_from_speed(Some(500), 1_024, 50), + Some(0), + "too slow to parent" + ); + assert_eq!( + limit_from_speed(Some(10_000), 1_024, 0), + Some(0), + "a zero ratio means the server is not allowing children" + ); + assert_eq!( + limit_from_speed(None, 1_024, 50), + None, + "an unreported speed leaves the caller's own cap standing" + ); + } + + #[test] + fn a_speed_the_server_has_not_reported_leaves_the_cap_alone() { + let mut children = serving(); + children.set_limits(None, 1_024, 50); + assert!(children.has_room(), "an unreported speed leaves us open"); + + // The server's figures win: at this speed the formula allows two. + children.set_limits(Some(10_000), 1_024, 50); + let mut ends = Vec::new(); + for i in 0..2 { + let (child, ours) = pair(); + ends.push(child); + assert!(children.accept(&format!("user{i}"), ours)); + } + assert!(!children.has_room(), "two is the limit at this speed"); + + children.set_limits(Some(100), 1_024, 50); + assert!(!children.has_room(), "a slow client carries none"); + assert_eq!(ends.len(), 2); + } + #[test] fn a_search_reaches_every_child() { - let mut children = Children::new(true); + let mut children = serving(); let (mut first, ours) = pair(); let (mut second, theirs) = pair(); children.accept("alice", ours); @@ -246,7 +385,7 @@ mod tests { #[test] fn a_child_that_hung_up_loses_its_slot() { - let mut children = Children::new(true); + let mut children = serving(); let (child, ours) = pair(); children.accept("alice", ours); drop(child); @@ -265,7 +404,7 @@ mod tests { #[test] fn a_child_is_told_where_our_branch_sits() { - let mut children = Children::new(true); + let mut children = serving(); let (mut child, ours) = pair(); children.accept("alice", ours); diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index a34e66c..12f9d43 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -264,6 +264,25 @@ impl Client { registered } + /// Tell the server we could not reach the peer it asked us to connect to, + /// quoting the token that peer is waiting on. + pub(crate) fn report_cant_connect( + client_context: &Arc>, + token: u32, + username: &str, + ) { + let sender = match client_context.read_safe() { + Ok(ctx) => ctx.server_sender.clone(), + Err(_) => return, + }; + let Some(sender) = sender else { return }; + let message = + crate::message::server::MessageFactory::build_cant_connect_to_peer( + token, username, + ); + let _ = sender.send(ServerMessage::SendMessage(message)); + } + pub(crate) fn connect_to_peer( peer: Peer, client_context: Arc>, @@ -306,7 +325,13 @@ impl Client { own_username, ); - match download_peer.download_file(client_context, None, None) { + let brokered = peer_clone.brokered; + let username = peer_clone.username; + match download_peer.download_file( + client_context.clone(), + None, + None, + ) { Ok((download, filename)) => { trace!( "[client] downloaded {} bytes {:?} ", @@ -315,6 +340,17 @@ impl Client { } Err(e) => { trace!("[client] failed to download: {}", e); + // A file connection the server asked us to make and + // we could not complete owes that peer a + // CantConnectToPeer, exactly as a control connection + // does: it is waiting on this token. + if brokered { + Self::report_cant_connect( + &client_context, + token, + &username, + ); + } } } } diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs index cfc89de..df84a2b 100644 --- a/soulseek-rs-lib/src/client/distributed.rs +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -320,8 +320,10 @@ pub fn announce_move( has_parent: bool, ) { let server = ctx.server_sender.clone(); - let accepting = ctx.children.accepting(); - announce(server.as_ref(), branch, has_parent, accepting); + // What the server needs to know is whether it may offer us to another + // peer right now, not whether we serve children in principle. + let has_room = ctx.children.has_room(); + announce(server.as_ref(), branch, has_parent, has_room); ctx.children.broadcast_stance(&branch.root, branch.level); } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index abaf670..9343e62 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -149,6 +149,11 @@ fn upload_speed( /// Build a `FileSearchResponse` for `query` against `shares`, or `None` if /// nothing matches. `own_username` is the name the searcher will download from. +// One reply's worth of state: the shares to search, who we are, and the +// figures the reply advertises. Kept as arguments rather than a struct +// because every one of them is read straight from the client context at the +// call site, and a struct would only move the same list one line up. +#[allow(clippy::too_many_arguments)] fn build_search_response( shares: &Shares, own_username: &str, @@ -157,13 +162,21 @@ fn build_search_response( free_slot: bool, speed: u32, queue_length: u32, + excluded_phrases: &[String], ) -> Option { let matches = shares.search(query); if matches.is_empty() { return None; } + // A file whose path carries a phrase the server excludes (code 160) is + // left out of the reply, the way Nicotine+ does it: the exclusion is the + // server policing what travels the search network, and answering with one + // anyway is what it is asking us not to do. let entries: Vec = matches .iter() + .filter(|f| { + !path_carries_excluded_phrase(&f.virtual_path, excluded_phrases) + }) .take(crate::types::MAX_SEARCH_REPLY_FILES) .map(|f| FileEntry { name: &f.virtual_path, @@ -171,6 +184,9 @@ fn build_search_response( attribs: &f.attributes, }) .collect(); + if entries.is_empty() { + return None; + } Some(build_file_search_response( own_username, token, @@ -181,6 +197,18 @@ fn build_search_response( )) } +/// Whether `path` contains any of the phrases the server excludes. Matching +/// is on the lowercased path, as the phrases themselves are lowercase. +fn path_carries_excluded_phrase(path: &str, excluded: &[String]) -> bool { + if excluded.is_empty() { + return false; + } + let lowered = path.to_lowercase(); + excluded + .iter() + .any(|phrase| !phrase.is_empty() && lowered.contains(phrase)) +} + #[derive(Debug, Clone)] pub struct ClientSettings { pub username: String, @@ -312,6 +340,13 @@ pub enum ClientOperation { token: u32, folder: String, }, + /// A peer answered our request for one folder of their shares. + FolderContents { + username: String, + token: u32, + folder: String, + directories: Vec, + }, /// A peer we are browsing returned their shared-file listing. BrowseResult { username: String, @@ -382,6 +417,10 @@ pub enum ClientOperation { }, /// Phrases the server refuses to search for (code 160). ExcludedSearchPhrases(Vec), + /// The minimum upload speed for carrying children (code 83). + ParentMinSpeed(u32), + /// The divisor turning that speed into a child count (code 84). + ParentSpeedRatio(u32), /// A parent candidate told us how deep it sits. `link` says which dial /// to that user is talking. ParentBranchLevel { @@ -419,6 +458,13 @@ pub struct ClientContext { leaf: distributed::Leaf, /// The children hanging from us, when the client serves any. pub(crate) children: children::Children, + /// The upload speed the server records for us, once it has told us. The + /// child limit is derived from it. + own_average_speed: Option, + /// The speed a parent needs (code 83) and the divisor turning speed into + /// a child count (code 84), as the server announced them. + parent_min_speed: u32, + parent_speed_ratio: u32, searches: HashMap, private_messages: Vec, /// Correlation tokens for server-brokered (firewalled) connections, mapping @@ -443,6 +489,8 @@ pub struct ClientContext { pending_serves: HashMap>, /// Shared-file listings received from peers we browsed. browse_results: HashMap>, + /// One-folder listings received from peers, keyed by peer and folder. + folder_contents: HashMap<(String, String), Vec>, pending_browses: HashMap, /// Latest snapshot of the public chat-room list (from `RoomList`, code 64). room_list: Vec, @@ -580,6 +628,9 @@ impl ClientContext { operations: None, leaf: distributed::Leaf::new(""), children: children::Children::default(), + own_average_speed: None, + parent_min_speed: 0, + parent_speed_ratio: 0, searches: HashMap::new(), private_messages: Vec::new(), pending_connect_tokens: HashMap::new(), @@ -592,6 +643,7 @@ impl ClientContext { active_uploads: HashMap::new(), pending_serves: HashMap::new(), browse_results: HashMap::new(), + folder_contents: HashMap::new(), pending_browses: HashMap::new(), room_list: Vec::new(), room_events: Vec::new(), @@ -723,6 +775,63 @@ impl ClientContext { self.room_events.push(event); } + /// Record what the server says our own upload speed is, and re-derive the + /// child limit from it. + pub fn set_own_average_speed(&mut self, speed: u32) { + self.own_average_speed = Some(speed); + self.apply_child_limits(); + } + + /// Record the server's parent figures (codes 83 and 84). + pub fn set_parent_min_speed(&mut self, speed: u32) { + self.parent_min_speed = speed; + self.apply_child_limits(); + } + + pub fn set_parent_speed_ratio(&mut self, ratio: u32) { + self.parent_speed_ratio = ratio; + self.apply_child_limits(); + } + + fn apply_child_limits(&mut self) { + self.children.set_limits( + self.own_average_speed, + self.parent_min_speed, + self.parent_speed_ratio, + ); + } + + /// Whether something is feeding us the distributed search stream: a + /// parent of our own, or the server relaying searches to us. Children may + /// only hang from a client that has one. + pub fn set_fed_by(&mut self, fed: bool) { + self.children.set_fed(fed); + if !fed { + self.children.drop_all(); + } + } + + /// Record one folder's listing from `username`. + pub fn store_folder_contents( + &mut self, + username: String, + folder: String, + directories: Vec, + ) { + self.folder_contents.insert((username, folder), directories); + } + + /// Remove and return the listing of `folder` from `username`, if it has + /// arrived. + pub fn take_folder_contents( + &mut self, + username: &str, + folder: &str, + ) -> Option> { + self.folder_contents + .remove(&(username.to_string(), folder.to_string())) + } + /// Who may enter the private room `room`, as last reported. #[must_use] pub fn private_room_members(&self, room: &str) -> Vec { diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 094a72d..4b31e16 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -10,6 +10,12 @@ use crate::peer::DownloadError; const CONNECT_SWEEP_INTERVAL: Duration = Duration::from_secs(1); +/// How often to ask the peers holding our queued downloads where they sit +/// (peer code 51). Five minutes is what Nicotine+ uses: often enough that a +/// queue position on screen means something, rare enough that a peer with a +/// long queue is not pestered. +const QUEUE_POSITION_INTERVAL: Duration = Duration::from_secs(300); + impl Client { pub(crate) fn listen_to_client_operations( reader: Receiver, @@ -18,6 +24,7 @@ impl Client { ) { thread::spawn(move || { let mut last_sweep = Instant::now(); + let mut last_queue_poll = Instant::now(); loop { let next = reader.recv_timeout(CONNECT_SWEEP_INTERVAL); if last_sweep.elapsed() >= CONNECT_SWEEP_INTERVAL { @@ -35,6 +42,10 @@ impl Client { } } } + if last_queue_poll.elapsed() >= QUEUE_POSITION_INTERVAL { + last_queue_poll = Instant::now(); + Self::poll_queue_positions(&client_context); + } let operation = match next { Ok(operation) => operation, Err(mpsc::RecvTimeoutError::Timeout) => continue, @@ -537,14 +548,28 @@ 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() + // Every search we receive comes from the tree: from + // our parent, or from the server acting as one. That + // is what qualifies us to carry children at all — + // Nicotine+ refuses them until something feeds it — + // and passing the search down is the whole duty. + let capacity_changed = match client_context.write_safe() { - ctx.children - .broadcast_search(&username, token, &query); + Ok(mut ctx) => { + let was_fed = ctx.children.is_fed(); + let before = ctx.children.len(); + ctx.set_fed_by(true); + ctx.children + .broadcast_search(&username, token, &query); + // A child whose socket has gone is dropped by + // that write, which frees a slot the server + // should hear about. + !was_fed || ctx.children.len() != before + } + Err(_) => false, + }; + if capacity_changed { + Self::announce_child_capacity(&client_context); } // Don't answer our own distributed search. @@ -560,6 +585,7 @@ impl Client { ctx.has_free_upload_slot(), ctx.last_upload_speed, ctx.upload_queue.len() as u32, + &ctx.excluded_search_phrases(), ), Err(e) => { error!("[client] IncomingSearch read: {}", e); @@ -754,6 +780,10 @@ impl Client { continue; }; ctx.leaf.reset(); + // A fresh session feeds us nothing until the tree (or + // the server) starts again, so the children we were + // carrying are let go rather than left starving. + ctx.set_fed_by(false); let branch = ctx.leaf.branch(); super::distributed::announce_move( &mut ctx, &branch, false, @@ -831,6 +861,9 @@ impl Client { if stream.set_nodelay(true).is_err() { continue; } + if username == own_username { + continue; // we cannot hang from ourselves + } if ctx.children.accept(&username, stream) { let branch = ctx.leaf.branch(); ctx.children.send_stance_to( @@ -842,8 +875,48 @@ impl Client { "[distributed] carrying {} children", ctx.children.len() ); + // The server stops offering us to peers once we + // are full, and starts again when one leaves. + let full = !ctx.children.has_room(); + drop(ctx); + if full { + Self::announce_child_capacity(&client_context); + } } } + ClientOperation::FolderContents { + username, + token, + folder, + directories, + } => { + trace!( + "[client] folder {} from {} (token {}): {} dirs", + folder, + username, + token, + directories.len() + ); + if let Ok(mut ctx) = client_context.write_safe() { + ctx.store_folder_contents( + username, + folder, + directories, + ); + } + } + ClientOperation::ParentMinSpeed(speed) => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.set_parent_min_speed(speed); + } + Self::announce_child_capacity(&client_context); + } + ClientOperation::ParentSpeedRatio(ratio) => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.set_parent_speed_ratio(ratio); + } + Self::announce_child_capacity(&client_context); + } ClientOperation::ExcludedSearchPhrases(phrases) => { if let Ok(mut ctx) = client_context.write_safe() { ctx.set_excluded_search_phrases(phrases); @@ -1075,6 +1148,50 @@ impl Client { }); } + /// Ask every peer holding a queued download of ours where it sits. The + /// answers arrive as `PlaceInQueueResponse` and land on the downloads. + fn poll_queue_positions(client_context: &Arc>) { + let (registry, queued) = match client_context.read_safe() { + Ok(ctx) => ( + ctx.peer_registry.clone(), + ctx.get_downloads() + .iter() + .filter(|d| matches!(d.status, DownloadStatus::Queued)) + .map(|d| (d.username.clone(), d.filename.clone())) + .collect::>(), + ), + Err(_) => return, + }; + let Some(registry) = registry else { return }; + for (username, filename) in queued { + // Only over a connection we already hold: a peer we cannot reach + // is a connection problem, not a queue question. + if !registry.contains(&username) { + continue; + } + let request = + MessageFactory::build_place_in_queue_request(&filename); + let _ = registry + .send_to_peer(&username, PeerMessage::SendMessage(request)); + } + } + + /// Tell the server whether we would take another child right now + /// (`AcceptChildren`, code 100). Sent whenever that answer changes: the + /// server only offers us to peers while it is true. + fn announce_child_capacity(client_context: &Arc>) { + let (sender, has_room) = match client_context.read_safe() { + Ok(ctx) => (ctx.server_sender.clone(), ctx.children.has_room()), + Err(_) => return, + }; + let Some(sender) = sender else { return }; + let message = + crate::message::server::MessageFactory::build_accept_children( + has_room, + ); + let _ = sender.send(ServerMessage::SendMessage(message)); + } + /// Answer a peer with the message `build` derives from the client state. fn reply_to_peer( client_context: &Arc>, diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index cf9ed10..a8a2b18 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -62,9 +62,6 @@ impl Client { let Some(handle) = &self.server_handle else { return Err(SoulseekRs::NotConnected); }; - if let Some(phrase) = self.excluded_phrase_in(query) { - return Err(SoulseekRs::SearchPhraseExcluded(phrase)); - } let token = next_search_token(); self.context.write_safe()?.searches.insert( @@ -122,9 +119,6 @@ impl Client { if self.server_handle.is_none() { return Err(SoulseekRs::NotConnected); } - if let Some(phrase) = self.excluded_phrase_in(query) { - return Err(SoulseekRs::SearchPhraseExcluded(phrase)); - } let token = next_search_token(); self.context.write_safe()?.searches.insert( query.to_string(), @@ -136,8 +130,9 @@ impl Client { Ok(token) } - /// The phrases the server refuses to search for (code 160), as last - /// announced. Empty until the server has said. + /// The phrases the server excludes from the search network (code 160), as + /// last announced. Files whose path carries one are left out of the + /// replies this client sends. Empty until the server has said. #[must_use] pub fn excluded_search_phrases(&self) -> Vec { self.context @@ -146,16 +141,6 @@ impl Client { .unwrap_or_default() } - /// The excluded phrase `query` carries, if any — what a search would be - /// refused for. - #[must_use] - pub fn excluded_phrase_in(&self, query: &str) -> Option { - self.context - .read_safe() - .ok() - .and_then(|ctx| ctx.excluded_phrase_in(query)) - } - /// Let responses accumulate for `timeout`, or until cancelled. /// /// Peers answer a search over their own connections whenever they get round diff --git a/soulseek-rs-lib/src/client/social.rs b/soulseek-rs-lib/src/client/social.rs index ef0dda6..5e3e439 100644 --- a/soulseek-rs-lib/src/client/social.rs +++ b/soulseek-rs-lib/src/client/social.rs @@ -166,6 +166,42 @@ impl Client { )) } + /// Ask `username` for the contents of one folder of their shares (peer + /// code 36) — what "download folder" in other clients asks for, without + /// pulling their whole listing. + /// + /// The answer arrives asynchronously; take it with + /// [`Client::take_folder_contents`]. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection, + /// or [`crate::SoulseekRs::LockPoisoned`] if the client state is poisoned. + pub fn request_folder_contents( + &self, + username: &str, + folder: &str, + ) -> Result<()> { + let request = crate::message::peer::build_folder_contents_request( + crate::client::next_search_token(), + folder, + ); + self.send_to_peer_or_queue(username, request) + } + + /// Remove and return the listing of `folder` from `username`, once their + /// answer to [`Client::request_folder_contents`] has arrived. + #[must_use] + pub fn take_folder_contents( + &self, + username: &str, + folder: &str, + ) -> Option> { + self.context + .write_safe() + .ok() + .and_then(|mut ctx| ctx.take_folder_contents(username, folder)) + } + /// Ask `username` where the file we queued with them sits (peer code 51). /// /// The answer arrives asynchronously and lands on the download, readable diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index d85b8f5..111a2b1 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -161,7 +161,7 @@ fn build_search_response_matches_shares_and_echoes_token() { let shares = Shares::scan(&dir).unwrap(); let response = - build_search_response(&shares, "me", 99, "xyzzy", true, 0, 0) + build_search_response(&shares, "me", 99, "xyzzy", true, 0, 0, &[]) .expect("a matching share yields a response"); let mut decoded = crate::message::Message::new_with_data(response.get_buffer()); @@ -172,7 +172,7 @@ fn build_search_response_matches_shares_and_echoes_token() { assert!(result.files.iter().any(|f| f.name.contains("probe_xyzzy"))); assert!( - build_search_response(&shares, "me", 1, "nomatch", true, 0, 0) + build_search_response(&shares, "me", 1, "nomatch", true, 0, 0, &[]) .is_none() ); let _ = std::fs::remove_dir_all(dir); @@ -1034,3 +1034,53 @@ fn losing_operatorship_keeps_the_membership_it_does_not_touch() { assert_eq!(ctx.private_room_members("club"), vec!["alice".to_string()]); assert!(ctx.private_room_operators("club").is_empty()); } + +#[test] +fn a_file_the_server_excludes_is_left_out_of_a_reply() { + // The server's excluded phrases (code 160) police what travels the search + // network: a matching file whose path carries one must not be offered, + // and a reply with nothing left is not sent at all. + let dir = std::env::temp_dir() + .join(format!("soulseek-excluded-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("Spam_Xyzzy.bin"), b"data").unwrap(); + let shares = Shares::scan(&dir).unwrap(); + + assert!( + build_search_response(&shares, "me", 1, "xyzzy", true, 0, 0, &[]) + .is_some(), + "with no exclusions the file is offered" + ); + assert!( + build_search_response( + &shares, + "me", + 1, + "xyzzy", + true, + 0, + 0, + &["spam".to_string()], + ) + .is_none(), + "an excluded phrase in the path, matched case-insensitively, \ + withholds the file — and with no files there is no reply" + ); + assert!( + build_search_response( + &shares, + "me", + 1, + "xyzzy", + true, + 0, + 0, + &["unrelated".to_string()], + ) + .is_some(), + "an exclusion the path does not carry changes nothing" + ); + + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/soulseek-rs-lib/src/error.rs b/soulseek-rs-lib/src/error.rs index 5b3b722..b25df46 100644 --- a/soulseek-rs-lib/src/error.rs +++ b/soulseek-rs-lib/src/error.rs @@ -20,9 +20,6 @@ pub enum SoulseekRs { CompressionError(String), /// A lock was poisoned by a panic in another thread LockPoisoned, - /// The query carries a phrase the server refuses to search for (code 160), - /// so sending it would spend a search on a query nothing answers. - SearchPhraseExcluded(String), } impl fmt::Display for SoulseekRs { @@ -44,9 +41,6 @@ impl fmt::Display for SoulseekRs { Self::CompressionError(msg) => { write!(f, "Compression error: {msg}") } - Self::SearchPhraseExcluded(phrase) => { - write!(f, "The server excludes searches for \"{phrase}\"") - } Self::LockPoisoned => { write!(f, "Lock poisoned by panicking thread") } diff --git a/soulseek-rs-lib/src/message/peer/folder_contents.rs b/soulseek-rs-lib/src/message/peer/folder_contents.rs index cf0f6f8..893554c 100644 --- a/soulseek-rs-lib/src/message/peer/folder_contents.rs +++ b/soulseek-rs-lib/src/message/peer/folder_contents.rs @@ -26,6 +26,37 @@ impl MessageHandler for FolderContentsRequest { } } +/// `FolderContentsResponse` (peer code 37): the reply to a folder we asked +/// for. The token is the one we quoted, so a client with two folders in +/// flight can tell the answers apart. +pub struct FolderContentsResponseHandler; +impl MessageHandler for FolderContentsResponseHandler { + fn get_code(&self) -> u32 { + 37 + } + fn handle(&self, message: &mut Message, sender: Sender) { + let Some((token, folder, directories)) = parse_folder_contents(message) + else { + return; + }; + let _ = sender.send(PeerMessage::FolderContentsReceived { + token, + folder, + directories, + }); + } +} + +/// Build a `FolderContentsRequest` (peer code 36). +#[must_use] +pub fn build_folder_contents_request(token: u32, folder: &str) -> Message { + Message::new() + .write_int32(36) + .write_int32(token) + .write_string(folder) + .clone() +} + /// Build a `FolderContentsResponse` (peer code 37). #[must_use] pub fn build_folder_contents( diff --git a/soulseek-rs-lib/src/message/peer/mod.rs b/soulseek-rs-lib/src/message/peer/mod.rs index b3bba95..1e48d17 100644 --- a/soulseek-rs-lib/src/message/peer/mod.rs +++ b/soulseek-rs-lib/src/message/peer/mod.rs @@ -17,7 +17,9 @@ pub use file_search_response::{ FileEntry, FileSearchResponse, build_file_search_response, }; pub use folder_contents::{ - FolderContentsRequest, build_folder_contents, parse_folder_contents, + FolderContentsRequest, FolderContentsResponseHandler, + build_folder_contents, build_folder_contents_request, + parse_folder_contents, }; pub use get_share_file_list::GetShareFileList; pub use peer_init::PeerInit; diff --git a/soulseek-rs-lib/src/message/server/parent_min_speed.rs b/soulseek-rs-lib/src/message/server/parent_min_speed.rs index b93ef6c..e783421 100644 --- a/soulseek-rs-lib/src/message/server/parent_min_speed.rs +++ b/soulseek-rs-lib/src/message/server/parent_min_speed.rs @@ -14,8 +14,32 @@ impl MessageHandler for ParentMinSpeedHandler { } fn handle(&self, message: &mut Message, sender: Sender) { - let _ = sender; let number = message.read_int32(); debug!("Parent min speed: {}", number); + // Kept, not just logged: with the ratio below it decides how many + // children this client may carry. + let _ = sender.send(ServerMessage::ParentMinSpeed(number)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::framed; + + #[test] + fn the_minimum_speed_is_reported_not_discarded() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_int32(1024); + }); + + ParentMinSpeedHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::ParentMinSpeed(speed)) => { + assert_eq!(speed, 1024); + } + other => panic!("unexpected: {other:?}"), + } } } diff --git a/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs b/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs index 744c6ea..e171927 100644 --- a/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs +++ b/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs @@ -14,8 +14,31 @@ impl MessageHandler for ParentSpeedRatioHandler { 84 } - fn handle(&self, message: &mut Message, _sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { let number = message.read_int32(); debug!("Parent speed ratio: {}", number); + let _ = sender.send(ServerMessage::ParentSpeedRatio(number)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::framed; + + #[test] + fn the_ratio_is_reported_not_discarded() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_int32(50); + }); + + ParentSpeedRatioHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::ParentSpeedRatio(ratio)) => { + assert_eq!(ratio, 50); + } + other => panic!("unexpected: {other:?}"), + } } } diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 64e08b0..179d22d 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -4431,7 +4431,11 @@ fn stub_server_sending( } #[test] -fn an_excluded_phrase_is_refused_before_a_search_is_spent() { +fn phrases_the_server_excludes_are_kept_for_our_replies() { + // The server tells clients which phrases are excluded from the search + // network (code 160). Nicotine+ applies them to the files it offers in a + // search reply, not to the searches it sends, and so do we — this pins + // that the announced list arrives and is kept for that use. let mut phrases = Message::new(); phrases .write_int32(160) @@ -4469,18 +4473,11 @@ fn an_excluded_phrase_is_refused_before_a_search_is_spent() { "the announced phrases should be kept" ); - // A query carrying one is refused here rather than spent on the server, - // and the match is case-insensitive on substrings, as the server's is. - match client.search("A BANNED-PHRASE query", Duration::from_millis(100)) { - Err(soulseek_rs::SoulseekRs::SearchPhraseExcluded(phrase)) => { - assert_eq!(phrase, "banned-phrase"); - } - other => panic!("expected a refusal, got {other:?}"), - } - // A clean query is not refused. + // Our own searches are not policed by them: that is the server's job, and + // no reference client refuses a query locally. assert!( client - .search("something else", Duration::from_millis(100)) + .search("a banned-phrase query", Duration::from_millis(100)) .is_ok() ); @@ -4865,6 +4862,28 @@ fn a_parent_tells_a_child_where_it_sits_and_passes_searches_down() { parent.connect().expect("parent connect"); assert!(parent.login().expect("parent login")); + // A client takes children only once something feeds it the search + // stream — here the server, which relays every search to us. Until then + // a child would hang from a branch that receives nothing, which is why + // Nicotine+ refuses one too. + let addr = format!("{}:{}", server.host, server.port); + let mut searcher = + login_raw(&addr, "e2e_par_searcher", "pw").expect("searcher login"); + let search = |searcher: &mut TcpStream, token: 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"); + }; + search(&mut searcher, 515_150); + std::thread::sleep(Duration::from_secs(2)); + let mut child = dial_as_child(listen_port, "e2e_par_child").expect("child dials"); @@ -4902,20 +4921,8 @@ fn a_parent_tells_a_child_where_it_sits_and_passes_searches_down() { // 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"); + search(&mut searcher, token); let relayed = read_distributed_until( &mut child, @@ -4938,19 +4945,24 @@ fn a_parent_tells_a_child_where_it_sits_and_passes_searches_down() { ) ); + // A second link from the same child is refused: the one we hold is live. + let mut duplicate = + dial_as_child(listen_port, "e2e_par_child").expect("child dials again"); + let told_again = + read_distributed_until(&mut duplicate, Duration::from_secs(3), |f| { + matches!(f, distributed::Distributed::BranchRoot(_)).then_some(()) + }); + assert!(told_again.is_none(), "a child we carry is not taken twice"); + assert_eq!(parent.children(), vec!["e2e_par_child".to_string()]); + drop(duplicate); + // A child that hangs up loses its place. drop(child); let deadline = Instant::now() + Duration::from_secs(20); + let mut next_token = token + 1; 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"); + search(&mut searcher, next_token); + next_token += 1; std::thread::sleep(Duration::from_millis(500)); } assert!( @@ -4984,3 +4996,55 @@ fn a_client_that_does_not_serve_children_turns_one_away() { assert!(told.is_none(), "a leaf must not take on a child"); assert!(leaf.children().is_empty()); } + +#[test] +fn one_folder_of_a_peers_shares_can_be_asked_for_by_itself() { + // "Download folder" in other clients asks for one folder (peer code 36) + // rather than pulling the peer's whole listing. + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + std::fs::create_dir_all(share_dir.join("wanted")).unwrap(); + std::fs::create_dir_all(share_dir.join("unwanted")).unwrap(); + std::fs::write(share_dir.join("wanted").join("keep.bin"), b"abcd").unwrap(); + std::fs::write(share_dir.join("unwanted").join("skip.bin"), b"efgh") + .unwrap(); + + let (_sharer, asker) = sharer_and_searcher( + &server, + &share_dir, + "e2e_folder_sharer", + "e2e_folder_asker", + ); + + // The folder is named the way the sharer advertises it: the share root's + // basename, then the subfolder, backslash-separated. + let root = share_dir.file_name().unwrap().to_string_lossy().to_string(); + let folder = format!("{root}\\wanted"); + + let deadline = Instant::now() + Duration::from_secs(20); + let mut listing = None; + while Instant::now() < deadline && listing.is_none() { + asker + .request_folder_contents("e2e_folder_sharer", &folder) + .expect("ask for one folder"); + std::thread::sleep(Duration::from_millis(500)); + listing = asker.take_folder_contents("e2e_folder_sharer", &folder); + } + let listing = listing.expect("the peer should answer with that folder"); + + assert!( + listing + .iter() + .any(|dir| dir.files.iter().any(|f| f.name.contains("keep.bin"))), + "the folder we asked for should be in the answer, got {listing:?}" + ); + assert!( + !listing + .iter() + .any(|dir| dir.files.iter().any(|f| f.name.contains("skip.bin"))), + "no other folder should come with it, got {listing:?}" + ); + + let _ = std::fs::remove_dir_all(share_dir); +}