dumbgoy/soulseek-rs-modded
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(protocol): folder requests, and heuristics matched to other clients

Browse Source 
Reading Nicotine+ and slskd against what was just written turned up three
places where this client had the message right and the behaviour wrong.

Excluded search phrases (code 160) are not a filter on the searches we
send — the server already drops those. Nicotine+ applies them to the files
it offers, leaving out any whose path carries an excluded phrase, and a
reply with nothing left is not sent. Ours now does the same, and the
local refusal of outgoing searches is gone.

Serving children now follows the same rules as Nicotine+: the limit comes
from the server's ParentMinSpeed and ParentSpeedRatio against the upload
speed the server records for us (capped at ten, none at all below the
minimum), children are refused until something feeds us the search stream
and let go when it stops, a second link from a child we already carry is
refused, and AcceptChildren is re-sent whenever that answer changes. Where
the server has never reported our speed there is nothing to apply the
formula to, so the configured cap stands, which is how slskd runs.

Queued downloads now ask where they sit every five minutes, as Nicotine+
does, instead of only when a caller asks; a brokered file connection we
cannot complete reports CantConnectToPeer like a control connection does;
and a peer's shares can be asked for one folder at a time (peer code 36
and its reply, 37), which is what "download folder" is built on.
This commit is contained in:
Michel de Graaf
2026-09-09 17:01:19 +00:00
parent bf3e3ec4ad
commit 0630b548b9
16 changed files with 751 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
soulseek-rs-lib/src/actor/peer_actor.rs
+24 -5
View File
@@ -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<SharedDirectory>),
/// A peer answered our `FolderContentsRequest` with one folder's listing.
FolderContentsReceived {
token: u32,
folder: String,
directories: Vec<SharedDirectory>,
},
/// 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(),
soulseek-rs-lib/src/actor/server_actor.rs
+13
View File
@@ -233,6 +233,11 @@ pub enum ServerMessage {
},
/// Phrases the server refuses to search for (code 160).
ExcludedSearchPhrases(Vec<String>),
/// 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,
soulseek-rs-lib/src/client/children.rs
+169 -30
View File
@@ -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<u32>,
min_speed: u32,
ratio: u32,
) -> Option<usize> {
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<String, TcpStream>,
/// 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<u32>,
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<u8> {
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);
soulseek-rs-lib/src/client/connection.rs
+37 -1
View File
@@ -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<RwLock<ClientContext>>,
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<RwLock<ClientContext>>,
@@ -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,
);
}
}
}
}
soulseek-rs-lib/src/client/distributed.rs
+4 -2
View File
@@ -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);
}
soulseek-rs-lib/src/client/mod.rs
+109
View File
@@ -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<crate::message::Message> {
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<FileEntry> = 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<SharedDirectory>,
},
/// 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<String>),
/// 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<u32>,
/// 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<String, Search>,
private_messages: Vec<UserMessage>,
/// Correlation tokens for server-brokered (firewalled) connections, mapping
@@ -443,6 +489,8 @@ pub struct ClientContext {
pending_serves: HashMap<String, Vec<u32>>,
/// Shared-file listings received from peers we browsed.
browse_results: HashMap<String, Vec<SharedDirectory>>,
/// One-folder listings received from peers, keyed by peer and folder.
folder_contents: HashMap<(String, String), Vec<SharedDirectory>>,
pending_browses: HashMap<String, Instant>,
/// Latest snapshot of the public chat-room list (from `RoomList`, code 64).
room_list: Vec<RoomInfo>,
@@ -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<SharedDirectory>,
) {
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<Vec<SharedDirectory>> {
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<String> {
soulseek-rs-lib/src/client/operations.rs
+124 -7
View File
@@ -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<ClientOperation>,
@@ -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<RwLock<ClientContext>>) {
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::<Vec<_>>(),
),
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<RwLock<ClientContext>>) {
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<RwLock<ClientContext>>,
soulseek-rs-lib/src/client/search.rs
+3 -18
View File
@@ -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<String> {
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<String> {
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
soulseek-rs-lib/src/client/social.rs
+36
View File
@@ -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<Vec<crate::SharedDirectory>> {
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
soulseek-rs-lib/src/client/tests.rs
+52 -2
View File
@@ -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);
}
soulseek-rs-lib/src/error.rs
-6
View File
@@ -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")
}
soulseek-rs-lib/src/message/peer/folder_contents.rs
+31
View File
@@ -26,6 +26,37 @@ impl MessageHandler<PeerMessage> 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<PeerMessage> for FolderContentsResponseHandler {
fn get_code(&self) -> u32 {
37
}
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
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(
soulseek-rs-lib/src/message/peer/mod.rs
+3 -1
View File
@@ -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;
soulseek-rs-lib/src/message/server/parent_min_speed.rs
+25 -1
View File
@@ -14,8 +14,32 @@ impl MessageHandler<ServerMessage> for ParentMinSpeedHandler {
}
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
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:?}"),
}
}
}
soulseek-rs-lib/src/message/server/parent_speed_ratio.rs
+24 -1
View File
@@ -14,8 +14,31 @@ impl MessageHandler<ServerMessage> for ParentSpeedRatioHandler {
84
}
fn handle(&self, message: &mut Message, _sender: Sender<ServerMessage>) {
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
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:?}"),
}
}
}
soulseek-rs-lib/tests/e2e.rs
+97 -33
View File
@@ -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);
}