From 9b87bd6a0fa3a82f42fbc0c1803f7f8534cff7cf Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Sun, 26 Jul 2026 14:20:07 +0200 Subject: [PATCH] feat!: wishlist, upload-queue privilege recognition, and search filters (#20) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the last two gaps in Soulseek's list of features an alternative client must implement to be tolerated, and adds the result filters Nicotine+ has had all along. Wishlist: server code 103 is sent and code 104's announced interval is kept rather than logged and discarded, so the re-search timer runs off the schedule the server set. `wish add|remove|list|run` plus `serve --wishlist`; wishes persist in config.toml as a TOML array so a comma inside a query survives. `wish run` prints search-shaped records, so it pipes into `download --stdin`. Privilege recognition: the code-69 user list is parsed instead of counted and dropped, and code 92 answers `whoami` as `privilege_seconds`. Uploads now go through a queue with a slot cap (`--upload-slots`, default 2) that serves privileged users first — the cap is what makes the ordering mean anything, so they are one change. `PlaceInQueueRequest` (peer code 51) is answered, where it previously had no handler at all, and a departed peer's slot is released so a stalled peer cannot wedge uploads shut. Search filters: `--exclude`, `--extension`, `--min-size`, `--max-size` beside the existing `--min-bitrate` and `--free-slots`, in one flattened arg group shared by `search`, `get` and `wish run`. Also serialises the library e2e suite behind the same gate `cli_e2e` already uses: the new queue tests hold peer sockets open on purpose, which starved concurrently-spawned soulfinds and failed unrelated download tests with `login: Timeout`. BREAKING CHANGE: `UploadStatus` gains a `Queued(u32)` variant, so an exhaustive match on it no longer compiles, and `cli::Commands` gains `Wish`. --- README.md | 20 +- soulseek-rs-lib/src/actor/peer_actor.rs | 24 +- soulseek-rs-lib/src/actor/server_actor.rs | 50 +- soulseek-rs-lib/src/client/mod.rs | 61 +- soulseek-rs-lib/src/client/operations.rs | 104 ++- soulseek-rs-lib/src/client/search.rs | 108 ++- soulseek-rs-lib/src/client/upload_queue.rs | 495 +++++++++++++ soulseek-rs-lib/src/client/uploads.rs | 100 ++- soulseek-rs-lib/src/message/peer/mod.rs | 2 + .../message/peer/place_in_queue_request.rs | 45 ++ .../src/message/server/message_factory.rs | 63 ++ soulseek-rs-lib/src/message/server/mod.rs | 2 +- .../src/message/server/privileged_users.rs | 110 ++- .../src/message/server/wish_list_interval.rs | 36 +- soulseek-rs-lib/src/types.rs | 2 + soulseek-rs-lib/tests/e2e.rs | 369 +++++++++- soulseek-rs/src/SKILL.md | 36 +- soulseek-rs/src/cli.rs | 95 ++- soulseek-rs/src/commands/mod.rs | 8 +- soulseek-rs/src/commands/peer.rs | 42 +- soulseek-rs/src/commands/settings.rs | 2 +- soulseek-rs/src/commands/transfer.rs | 251 ++++++- soulseek-rs/src/commands/wish.rs | 257 +++++++ soulseek-rs/src/main.rs | 30 +- soulseek-rs/src/output.rs | 15 + soulseek-rs/src/persist/config.rs | 103 +++ .../src/ui/panes/download_info_pane.rs | 7 +- soulseek-rs/src/ui/panes/downloads_pane.rs | 10 +- soulseek-rs/tests/cli_e2e.rs | 689 ++++++++++++++++++ 29 files changed, 2991 insertions(+), 145 deletions(-) create mode 100644 soulseek-rs-lib/src/client/upload_queue.rs create mode 100644 soulseek-rs-lib/src/message/peer/place_in_queue_request.rs create mode 100644 soulseek-rs/src/commands/wish.rs diff --git a/README.md b/README.md index d84edd4..888704a 100644 --- a/README.md +++ b/README.md @@ -25,9 +25,15 @@ does, how to install it, and every `config.toml` setting. ## Features - **Search & download**: search the network and queue downloads in the TUI, - or fetch a track in one command with `get` + or fetch a track in one command with `get`; narrow results by bitrate, size, + file type, free slots, or terms to exclude +- **Wishlist**: `wish add` something the network doesn't have today, and + `wish run` (or `serve --wishlist`) keeps looking for it on the interval the + server sets - **Sharing**: `shares add` a directory and your files show up in searches; `serve` stays online so peers can browse and download them +- **Upload queue**: a slot cap with users the server lists as privileged served + first, and peers get a truthful answer when they ask their place in line - **Browse**: list any user's shared files and download straight from the tree - **Resumable downloads**: a transfer that dies leaves a `.part` file, and re-running the download asks the peer to send only what is missing @@ -137,6 +143,8 @@ soulseek-rs search # print matching files soulseek-rs download # fetch one known file soulseek-rs download --stdin # fetch files listed on stdin soulseek-rs get # search, pick, and download +soulseek-rs wish add|remove|list # keep looking for something +soulseek-rs wish run # search every stored wish once soulseek-rs browse # list a user's shared files soulseek-rs room list # public rooms and user counts soulseek-rs room say # post to a room @@ -235,6 +243,14 @@ elif [ $? -eq 4 ]; then echo "nobody has it" fi +# Nobody has it today: keep looking, and fetch it the day someone does +soulseek-rs wish add "some rare 12 inch" +soulseek-rs wish run --json | soulseek-rs download --stdin + +# Only lossless, nothing from a bootleg folder, album-sized files only +soulseek-rs search "boards of canada" \ + --extension flac --exclude bootleg --min-size 20000000 + # Log room chat as JSON until interrupted soulseek-rs room listen lobby --follow --json >> lobby.ndjson @@ -317,7 +333,7 @@ seconds (an hour by default), or runs until interrupted with `--follow`. soulseek-rs shares add ~/Music # remembered in config.toml soulseek-rs shares status # what the network will see soulseek-rs shares reindex # after adding files on disk -soulseek-rs serve --follow +soulseek-rs serve --follow # add --upload-slots N to widen the queue ``` Uploads only show up inside a running `serve`; every other command is diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 7bf6f83..b76d732 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -2,9 +2,10 @@ use crate::actor::{Actor, ActorHandle, ConnectionState}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::peer::{ - FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueResponse, - QueueUploadHandler, SharedDirectory, SharedFileListResponseHandler, - TransferRequest, TransferResponse, UploadFailedHandler, + FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueRequest, + PlaceInQueueResponse, QueueUploadHandler, SharedDirectory, + SharedFileListResponseHandler, TransferRequest, TransferResponse, + UploadFailedHandler, }; use crate::message::server::MessageFactory; use crate::message::{Handlers, Message, MessageReader, MessageType}; @@ -39,6 +40,8 @@ pub enum PeerMessage { RequestTransfer(Download), /// A peer queued one of our shared files for download (they sent us code 43). IncomingQueueUpload(String), + /// A peer asked where their queued file sits (they sent us code 51). + IncomingPlaceInQueueRequest(String), /// A peer asked to browse our shared files (they sent us code 4). ShareListRequested, /// A peer we are browsing sent us their shared-file listing (code 5). @@ -164,6 +167,7 @@ impl PeerActor { handlers.register_handler(TransferResponse); handlers.register_handler(GetShareFileList); handlers.register_handler(UploadFailedHandler); + handlers.register_handler(PlaceInQueueRequest); handlers.register_handler(PlaceInQueueResponse); handlers.register_handler(QueueUploadHandler); handlers.register_handler(SharedFileListResponseHandler); @@ -235,6 +239,20 @@ impl PeerActor { PeerMessage::IncomingQueueUpload(filename) => { self.handle_incoming_queue_upload(filename); } + PeerMessage::IncomingPlaceInQueueRequest(filename) => { + let requester_key = self.peer_username(); + if let Err(e) = self.client_channel.send( + ClientOperation::PlaceInQueueRequested { + requester_key, + filename, + }, + ) { + error!( + "[peer_actor] forward IncomingPlaceInQueueRequest: {}", + e + ); + } + } PeerMessage::ServeUpload { token, filename, diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 0c14298..cfa239e 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -1,6 +1,7 @@ use crate::actor::{Actor, ActorHandle, ConnectionState}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; +use crate::message::server::CheckPrivilegesHandler; use crate::message::server::ConnectToPeerHandler; use crate::message::server::ExcludedSearchPhrasesHandler; use crate::message::server::FileSearchHandler; @@ -166,6 +167,20 @@ pub enum ServerMessage { token: u32, query: String, }, + /// A wishlist search (code 103): a stored query the server lets us repeat + /// once per announced interval. + WishlistSearch { + token: u32, + query: String, + }, + /// The server announced how often it will accept a wishlist search. + WishlistInterval(u32), + /// Everyone the server counts as privileged (code 69). + PrivilegedUsers(Vec), + /// Seconds of our own privileges left (code 92). + OwnPrivileges(u32), + /// Ask the server for the answer to the above. + CheckPrivileges, /// A search the server distributed to us from another user; if it matches /// our shares we reply with a FileSearchResponse. FileSearchRequest { @@ -391,7 +406,7 @@ impl ServerActor { handlers.register_handler(WishListIntervalHandler); handlers.register_handler(ParentMinSpeedHandler); handlers.register_handler(ParentSpeedRatioHandler); - handlers.register_handler(PrivilegedUsersHandler); + handlers.register_handler(CheckPrivilegesHandler); handlers.register_handler(FileSearchHandler); handlers.register_handler(GetPeerAddressHandler); handlers.register_handler(ConnectToPeerHandler); @@ -549,6 +564,39 @@ impl ServerActor { } => { self.handle_file_search_request(username, token, query); } + other => self.handle_standing_message(other), + } + } + + /// The wishlist and privilege traffic: standing searches (codes 103/104) and + /// who is privileged (codes 69/92). + /// + /// Split out only because it keeps `handle_message`'s match to a readable + /// length; there is no behaviour here beyond dispatch. + fn handle_standing_message(&mut self, message: ServerMessage) { + match message { + ServerMessage::WishlistSearch { token, query } => { + self.queue_message(MessageFactory::build_wishlist_search( + token, &query, + )); + } + ServerMessage::WishlistInterval(seconds) => { + self.forward_to_client(ClientOperation::WishlistInterval( + seconds, + )); + } + ServerMessage::PrivilegedUsers(users) => { + self.forward_to_client(ClientOperation::PrivilegedUsers(users)); + } + ServerMessage::OwnPrivileges(seconds) => { + self.forward_to_client(ClientOperation::OwnPrivileges(seconds)); + } + ServerMessage::CheckPrivileges => { + self.queue_message(MessageFactory::build_check_privileges()); + } + other => { + error!("[server] unroutable message: {:?}", other); + } } } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 003fa0b..e2b1e5a 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -22,7 +22,7 @@ use crate::{ utils::{lock::RwLockExt, md5, thread_pool::ThreadPool}, }; use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, net::TcpStream, sync::{ RwLock, @@ -35,10 +35,24 @@ use std::{ sync::{Arc, mpsc}, time::{Duration, Instant}, }; +use upload_queue::QueuedUpload; use crate::{debug, error, info, trace, warn}; const DEFAULT_LISTEN_PORT: u16 = 2234; +/// What to assume between wishlist searches until the server announces its own. +/// +/// Twelve minutes is what Soulseek servers give an unprivileged account (code +/// 104); guessing lower only gets the searches dropped. +pub const DEFAULT_WISHLIST_INTERVAL: Duration = Duration::from_mins(12); + +/// How many uploads run at once by default. +/// +/// Two is what makes the queue real: with unlimited slots there is nothing for +/// a privileged peer to jump, and a home connection split across every peer +/// that asks serves all of them badly. +pub const DEFAULT_UPLOAD_SLOTS: usize = 2; + /// How long to wait for a server-brokered (firewalled) peer to connect back /// before giving up and failing the download. Matches the direct-dial timeout. const BROKER_CONNECT_TIMEOUT: Duration = Duration::from_secs(20); @@ -248,6 +262,19 @@ pub enum ClientOperation { /// Something happened in the chat-room subsystem (list refreshed, a room /// joined/left, a message said, a member joined/left). RoomEvent(RoomEvent), + /// The server announced how many seconds must pass between wishlist + /// searches. + WishlistInterval(u32), + /// Everyone the server counts as privileged; they sort ahead of others in + /// our upload queue. + PrivilegedUsers(Vec), + /// Seconds of our own privileges left. + OwnPrivileges(u32), + /// A peer asked where their queued file sits. + PlaceInQueueRequested { + requester_key: String, + filename: String, + }, } pub struct ClientContext { pub peer_registry: Option, @@ -284,6 +311,20 @@ pub struct ClientContext { /// What the server has told us about other users, merged across the /// separate status and statistics replies. user_info: HashMap, + /// Seconds the server wants between wishlist searches (code 104), once it + /// has told us. + wishlist_interval: Option, + /// Everyone the server listed as privileged (code 69). They sort ahead of + /// other peers in [`Self::upload_queue`]. + privileged_users: HashSet, + /// Seconds of our own privileges left (code 92), once we have asked. + own_privileges: Option, + /// Peers waiting for one of our upload slots. + upload_queue: Vec, + /// Arrival counter for the queue's first-come tie-break. + upload_seq: u64, + /// How many uploads may be in flight at once. + upload_slots: usize, actor_system: Arc, } impl Default for ClientContext { @@ -580,6 +621,12 @@ impl ClientContext { room_events: Vec::new(), room_members: HashMap::new(), user_info: HashMap::new(), + wishlist_interval: None, + privileged_users: HashSet::new(), + own_privileges: None, + upload_queue: Vec::new(), + upload_seq: 0, + upload_slots: DEFAULT_UPLOAD_SLOTS, downloads: DownloadStore::new(), actor_system, } @@ -820,8 +867,9 @@ impl Client { }) } - /// Snapshot of the uploads served this session (active and finished), - /// most recent last. + /// Every upload this session knows about: the ones streaming or finished, + /// followed by the peers still waiting for a slot, in the order they will + /// be served. #[must_use] pub fn uploads(&self) -> Vec { self.context.read_safe().map_or_else( @@ -829,7 +877,7 @@ impl Client { |ctx| { let mut tokens: Vec<&u32> = ctx.active_uploads.keys().collect(); tokens.sort_unstable(); - tokens + let mut all: Vec = tokens .into_iter() .map(|token| { let upload = &ctx.active_uploads[token]; @@ -849,7 +897,9 @@ impl Client { status: upload.status.clone(), } }) - .collect() + .collect(); + all.extend(ctx.queued_uploads()); + all }, ) } @@ -919,6 +969,7 @@ mod downloads; mod operations; mod rooms; mod search; +mod upload_queue; mod uploads; #[cfg(test)] diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 65277bf..63bf1b1 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -1,10 +1,12 @@ +use super::uploads::downloader_of; use super::{ Arc, BROKER_CONNECT_TIMEOUT, Client, ClientContext, ClientOperation, ConnectionType, Download, DownloadPeer, DownloadStatus, Peer, PeerMessage, - PeerRegistry, Receiver, RwLock, RwLockExt, ServerMessage, UploadJob, - build_search_response, debug, error, info, next_connect_token, - next_upload_token, sleep, thread, trace, warn, + PeerRegistry, Receiver, RwLock, RwLockExt, ServerMessage, + build_search_response, debug, error, info, next_connect_token, sleep, + thread, trace, warn, }; +use crate::message::server::MessageFactory; impl Client { pub(crate) fn listen_to_client_operations( @@ -102,6 +104,14 @@ impl Client { None, ); } + // A peer that leaves must not keep occupying an + // upload slot: with a slot cap, a couple of + // departed peers would otherwise wedge uploads + // shut for the rest of the session. + Self::release_upload_slots( + &client_context, + &username, + ); } ClientOperation::PierceFireWall(peer) => { Self::pierce_firewall( @@ -531,6 +541,12 @@ impl Client { ), } } + ClientOperation::WishlistInterval(seconds) => { + if let Ok(mut ctx) = client_context.write_safe() + { + ctx.wishlist_interval = Some(seconds); + } + } ClientOperation::PeerConnected(username) => { // An outbound control connection just handshook. // Flush any downloads that were queued for this @@ -655,16 +671,10 @@ impl Client { requester_key, filename, } => { - // A peer queued one of our shared files. Look it - // up, mint an upload token, and offer it. - let downloader = requester_key - .strip_suffix(":direct") - .unwrap_or(&requester_key) - .to_string(); - let token = next_upload_token(); - let (registry, size) = match client_context - .write_safe() - { + // The peer served next may not be this one. + let downloader = + downloader_of(&requester_key).to_string(); + match client_context.write_safe() { Ok(mut ctx) => { let Some(file) = ctx.shares.get(&filename) @@ -677,16 +687,13 @@ impl Client { }; let size = file.size; let real_path = file.real_path.clone(); - ctx.uploads.insert( - token, - UploadJob { - downloader: downloader.clone(), - real_path, - virtual_path: filename.clone(), - size, - }, + ctx.enqueue_upload( + &requester_key, + &downloader, + &filename, + real_path, + size, ); - (ctx.peer_registry.clone(), size) } Err(e) => { error!( @@ -695,18 +702,59 @@ impl Client { ); continue; } - }; + } + Self::pump_upload_queue(&client_context); + } + ClientOperation::PlaceInQueueRequested { + requester_key, + filename, + } => { + let downloader = downloader_of(&requester_key); + let (registry, place) = + match client_context.read_safe() { + Ok(ctx) => ( + ctx.peer_registry.clone(), + ctx.place_in_queue( + downloader, &filename, + ), + ), + Err(_) => continue, + }; + // Silence would leave the peer guessing; a file + // no longer queued is being served, which is + // place 0 by the same convention Nicotine+ uses. if let Some(registry) = registry { let _ = registry.send_to_peer( &requester_key, - PeerMessage::ServeUpload { - token, - filename, - size, - }, + PeerMessage::SendMessage( + MessageFactory::build_place_in_queue_response( + &filename, + place.unwrap_or(0), + ), + ), ); } } + ClientOperation::PrivilegedUsers(users) => { + if let Ok(mut ctx) = client_context.write_safe() + { + debug!( + "[client] {} privileged users listed", + users.len() + ); + ctx.set_privileged_users(users); + } + // Someone already waiting may have just been + // outranked, but a slot that is free should + // still be filled. + Self::pump_upload_queue(&client_context); + } + ClientOperation::OwnPrivileges(seconds) => { + if let Ok(mut ctx) = client_context.write_safe() + { + ctx.own_privileges = Some(seconds); + } + } ClientOperation::StartUpload { token } => { // The peer accepted our offer: resolve their // address (from the code-9 GetPeerAddress) and diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index 63cbdef..cde3312 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -1,7 +1,7 @@ use super::{ - Arc, AtomicBool, Client, Duration, HashMap, Instant, Ordering, Result, - RwLockExt, Search, SearchResult, ServerMessage, SoulseekRs, info, md5, - sleep, + Arc, AtomicBool, Client, DEFAULT_WISHLIST_INTERVAL, Duration, HashMap, + Instant, Ordering, Result, RwLockExt, Search, SearchResult, ServerMessage, + SoulseekRs, info, md5, sleep, }; impl Client { @@ -13,53 +13,95 @@ impl Client { self.search_with_cancel(query, timeout, None) } + /// Send `query` as a wishlist search (server code 103) and return at once. + /// + /// Results accumulate under `query`, so the caller reads them back with + /// [`Client::get_search_results`] after waiting however long it wants to. + /// Starting several wishes and then waiting once is the whole point: waiting + /// per wish would cost one full search window each. + /// + /// The server rate-limits these to the interval it announced — see + /// [`Client::wishlist_interval`] — so a wish re-sent sooner than that comes + /// back empty. + /// + /// # Errors + /// [`SoulseekRs::NotConnected`] when there is no server connection. + pub fn start_wishlist_search(&self, query: &str) -> Result<()> { + self.send_search(query, true) + } + + /// How long to wait between wishlist searches: what the server announced in + /// code 104, or [`DEFAULT_WISHLIST_INTERVAL`] until it has. + #[must_use] + pub fn wishlist_interval(&self) -> Duration { + self.context + .read_safe() + .ok() + .and_then(|ctx| ctx.wishlist_interval) + .map_or(DEFAULT_WISHLIST_INTERVAL, |seconds| { + Duration::from_secs(u64::from(seconds)) + }) + } + pub fn search_with_cancel( &self, query: &str, timeout: Duration, cancel_flag: Option>, ) -> Result> { + self.send_search(query, false)?; + Self::collect_for(timeout, cancel_flag); + Ok(self.get_search_results(query)) + } + + /// Register `query` and put its search on the wire. Returns as soon as the + /// message is queued; nothing has answered yet. + fn send_search(&self, query: &str, wishlist: bool) -> Result<()> { info!("Searching for {}", query); - if let Some(handle) = &self.server_handle { - let hash = md5::md5(query); - let token = u32::from_str_radix(&hash[0..5], 16)?; - - self.context.write_safe()?.searches.insert( - query.to_string(), - Search { - token, - results: Vec::new(), - }, - ); - - let _ = handle.send(ServerMessage::FileSearch { - token, - query: query.to_string(), - }); - } else { + let Some(handle) = &self.server_handle else { return Err(SoulseekRs::NotConnected); - } + }; + let hash = md5::md5(query); + let token = u32::from_str_radix(&hash[0..5], 16)?; + self.context.write_safe()?.searches.insert( + query.to_string(), + Search { + token, + results: Vec::new(), + }, + ); + + let query = query.to_string(); + let _ = handle.send(if wishlist { + ServerMessage::WishlistSearch { token, query } + } else { + ServerMessage::FileSearch { token, query } + }); + Ok(()) + } + + /// Let responses accumulate for `timeout`, or until cancelled. + /// + /// Peers answer a search over their own connections whenever they get round + /// to it, so there is nothing to await — the window is the whole protocol. + /// Late responders are often the better sources, which is why this runs the + /// window out rather than returning on the first hit. + pub fn collect_for( + timeout: Duration, + cancel_flag: Option>, + ) { let start = Instant::now(); - loop { + while start.elapsed() < timeout { sleep(Duration::from_millis(100)); - - // Check if cancelled - if let Some(ref flag) = cancel_flag + if let Some(flag) = &cancel_flag && flag.load(Ordering::Relaxed) { info!("Search cancelled by user"); - break; - } - - // Check if timeout reached - if start.elapsed() >= timeout { - break; + return; } } - - Ok(self.get_search_results(query)) } #[must_use] diff --git a/soulseek-rs-lib/src/client/upload_queue.rs b/soulseek-rs-lib/src/client/upload_queue.rs new file mode 100644 index 0000000..70ef25e --- /dev/null +++ b/soulseek-rs-lib/src/client/upload_queue.rs @@ -0,0 +1,495 @@ +//! The queue peers wait in for one of our upload slots. +//! +//! Soulseek's rules ask an alternative client for "respect / recognition of +//! privileges": a user who has donated is meant to be served before one who +//! has not. That only means anything if there is a queue to jump — serving +//! every request the instant it arrives treats a privileged peer exactly like +//! everybody else — so the slot cap and the ordering are one feature. +//! +//! Ordering is privileged-first, then first-come. Nothing else: a fancier +//! policy (per-user fairness, small-file-first) would be a different promise +//! than the one the rules ask for. + +use super::{ClientContext, PeerRegistry, UploadJob}; +use crate::types::{UploadInfo, UploadStatus}; +use std::path::PathBuf; + +/// A queued upload that has just been given a slot, ready to be offered to the +/// peer that asked for it. +#[derive(Debug, Clone)] +pub struct UploadOffer { + pub requester_key: String, + pub token: u32, + pub virtual_path: String, + pub size: u64, +} + +/// A file a peer has asked us for and is waiting on. +#[derive(Debug, Clone)] +pub struct QueuedUpload { + /// Registry key of the asking peer's actor, which may carry a `:direct` + /// suffix — not the same string as `downloader`. + pub requester_key: String, + pub downloader: String, + pub virtual_path: String, + pub real_path: PathBuf, + pub size: u64, + /// Whether the server listed this user as privileged when they asked. + pub privileged: bool, + /// Arrival order, so the tie-break among equals is first-come. + pub seq: u64, +} + +/// What decides the order: donors first, then whoever asked first. +const fn rank(job: &QueuedUpload) -> (bool, u64) { + (!job.privileged, job.seq) +} + +/// Indices of `queue` in the order they will be served. +#[must_use] +pub fn serve_order(queue: &[QueuedUpload]) -> Vec { + let mut order: Vec = (0..queue.len()).collect(); + order.sort_by_key(|&i| rank(&queue[i])); + order +} + +/// Which entry to serve next, or `None` when nobody is waiting. +#[must_use] +pub fn next_to_serve(queue: &[QueuedUpload]) -> Option { + queue + .iter() + .enumerate() + .min_by_key(|(_, job)| rank(job)) + .map(|(index, _)| index) +} + +/// A queue position as the protocol reports it: counting from 1. +fn as_place(zero_based: usize) -> u32 { + u32::try_from(zero_based + 1).unwrap_or(u32::MAX) +} + +/// Where `filename` sits for `downloader`, counting from 1 as the protocol's +/// `PlaceInQueueResponse` does. `None` when that peer is not waiting for it, +/// which is also the answer once the file has been handed over. +#[must_use] +pub fn place_in_queue( + queue: &[QueuedUpload], + downloader: &str, + filename: &str, +) -> Option { + serve_order(queue) + .into_iter() + .position(|i| { + queue[i].downloader == downloader + && queue[i].virtual_path == filename + }) + .map(as_place) +} + +/// The queue as the client drives it: taking a request, filling slots, and +/// answering where a peer sits. +impl ClientContext { + /// Take `filename` from a peer's request and put them in the queue. The + /// caller pumps afterwards; queueing and serving are separate so a request + /// for a file we do not have never reaches the queue. + pub fn enqueue_upload( + &mut self, + requester_key: &str, + downloader: &str, + filename: &str, + real_path: std::path::PathBuf, + size: u64, + ) { + self.upload_seq += 1; + self.upload_queue.push(QueuedUpload { + requester_key: requester_key.to_string(), + downloader: downloader.to_string(), + virtual_path: filename.to_string(), + real_path, + size, + privileged: self.privileged_users.contains(downloader), + seq: self.upload_seq, + }); + } + + /// Move as many queued uploads into free slots as will fit, and return the + /// offers to send. Ordering is privileged-first, then first-come. + /// + /// The registry comes back with them so the caller can send after dropping + /// the context lock: a peer actor's mailbox is not something to hold it + /// across. + pub fn pump_uploads( + &mut self, + mut next_token: impl FnMut() -> u32, + ) -> (Option, Vec) { + let mut offers = Vec::new(); + while self.uploads_in_flight() < self.upload_slots { + let Some(index) = next_to_serve(&self.upload_queue) else { + break; + }; + let job = self.upload_queue.remove(index); + let token = next_token(); + offers.push(UploadOffer { + requester_key: job.requester_key, + token, + virtual_path: job.virtual_path.clone(), + size: job.size, + }); + self.uploads.insert( + token, + UploadJob { + downloader: job.downloader, + real_path: job.real_path, + virtual_path: job.virtual_path, + size: job.size, + }, + ); + } + (self.peer_registry.clone(), offers) + } + + /// Uploads occupying a slot: offered but not yet accepted, plus those + /// actually streaming. + /// + /// `active_uploads` keeps finished transfers so `uploads()` can report them, + /// so only the in-progress ones count — counting the whole table would let + /// a few completed transfers wedge the queue shut forever. + fn uploads_in_flight(&self) -> usize { + self.uploads.len() + + self + .active_uploads + .values() + .filter(|upload| { + matches!(upload.status, UploadStatus::InProgress) + }) + .count() + } + + /// The peers still waiting, in the order they will be served, each carrying + /// its own 1-based place. + pub(super) fn queued_uploads(&self) -> Vec { + serve_order(&self.upload_queue) + .into_iter() + .enumerate() + .map(|(place, index)| { + let job = &self.upload_queue[index]; + UploadInfo { + username: job.downloader.clone(), + filename: job.virtual_path.clone(), + size: job.size, + bytes_sent: 0, + speed_bytes_per_sec: 0.0, + status: UploadStatus::Queued(as_place(place)), + } + }) + .collect() + } + + /// Where `downloader`'s queued `filename` sits, counting from 1. `None` + /// once it has left the queue for a slot. + #[must_use] + pub fn place_in_queue( + &self, + downloader: &str, + filename: &str, + ) -> Option { + place_in_queue(&self.upload_queue, downloader, filename) + } + + /// Replace the privileged set, re-ranking anyone already waiting: the list + /// arrives at login, which can be after a peer has queued something. + pub fn set_privileged_users(&mut self, users: Vec) { + self.privileged_users = users.into_iter().collect(); + for job in &mut self.upload_queue { + job.privileged = self.privileged_users.contains(&job.downloader); + } + } + + #[must_use] + pub fn is_privileged(&self, username: &str) -> bool { + self.privileged_users.contains(username) + } + + /// Forget everything `downloader` was waiting for or had been offered, and + /// report whether that freed anything. + /// + /// An offer sits in `uploads` counting against the slot cap until the peer + /// accepts it. A peer that never will — because it has gone — would hold + /// that slot for the life of the process, so leaving is what releases it. + pub fn release_upload_slots(&mut self, downloader: &str) -> bool { + let before = self.upload_queue.len() + self.uploads.len(); + self.upload_queue.retain(|job| job.downloader != downloader); + self.uploads.retain(|_, job| job.downloader != downloader); + before != self.upload_queue.len() + self.uploads.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn queued(user: &str, privileged: bool, seq: u64) -> QueuedUpload { + QueuedUpload { + requester_key: user.to_string(), + downloader: user.to_string(), + virtual_path: format!("@@share\\{user}.mp3"), + real_path: PathBuf::from("/tmp/x.mp3"), + size: 1, + privileged, + seq, + } + } + + #[test] + fn an_empty_queue_serves_nobody() { + assert_eq!(next_to_serve(&[]), None); + } + + #[test] + fn equals_are_served_in_the_order_they_asked() { + let queue = [queued("amy", false, 1), queued("bob", false, 2)]; + assert_eq!(next_to_serve(&queue), Some(0)); + assert_eq!(serve_order(&queue), [0, 1]); + } + + #[test] + fn a_privileged_peer_is_served_before_one_that_asked_first() { + let queue = [ + queued("early_plain", false, 1), + queued("late_donor", true, 2), + ]; + assert_eq!( + next_to_serve(&queue), + Some(1), + "the donor jumps the plain user who was already waiting" + ); + } + + #[test] + fn privileged_peers_keep_first_come_order_among_themselves() { + let queue = [ + queued("donor_late", true, 9), + queued("plain", false, 2), + queued("donor_early", true, 5), + ]; + let order: Vec<&str> = serve_order(&queue) + .into_iter() + .map(|i| queue[i].downloader.as_str()) + .collect(); + assert_eq!(order, ["donor_early", "donor_late", "plain"]); + } + + #[test] + fn the_reported_place_counts_from_one() { + let queue = [queued("amy", false, 1), queued("bob", false, 2)]; + assert_eq!(place_in_queue(&queue, "amy", "@@share\\amy.mp3"), Some(1)); + assert_eq!(place_in_queue(&queue, "bob", "@@share\\bob.mp3"), Some(2)); + } + + #[test] + fn the_reported_place_reflects_being_overtaken() { + let mut queue = vec![queued("plain", false, 1)]; + assert_eq!( + place_in_queue(&queue, "plain", "@@share\\plain.mp3"), + Some(1) + ); + queue.push(queued("donor", true, 2)); + assert_eq!( + place_in_queue(&queue, "plain", "@@share\\plain.mp3"), + Some(2), + "a peer that got overtaken must be told so, not left on 1" + ); + } + + #[test] + fn a_file_nobody_queued_has_no_place() { + let queue = [queued("amy", false, 1)]; + assert_eq!(place_in_queue(&queue, "amy", "@@share\\other.mp3"), None); + assert_eq!(place_in_queue(&queue, "zoe", "@@share\\amy.mp3"), None); + } +} + +/// Tests of the queue as the client actually drives it: what a request, a +/// privileged-user list, and a finished transfer do to a real +/// [`super::ClientContext`]. +/// +/// These sit here rather than in the e2e suite because a Soulseek server only +/// ever names privileged users it knows about, and the soulfind used for e2e +/// declares nobody — so the one behaviour the rules are about would otherwise +/// go untested. +#[cfg(test)] +mod context_tests { + use crate::client::{ClientContext, DEFAULT_UPLOAD_SLOTS}; + use crate::types::UploadStatus; + use std::path::PathBuf; + + /// A context with `slots` upload slots and a token counter that hands out + /// 1, 2, 3… so an assertion can name a token. + fn context(slots: usize) -> (ClientContext, impl FnMut() -> u32) { + let mut ctx = ClientContext::new(); + ctx.upload_slots = slots; + let mut next = 0; + (ctx, move || { + next += 1; + next + }) + } + + fn ask(ctx: &mut ClientContext, user: &str, file: &str) { + ctx.enqueue_upload(user, user, file, PathBuf::from("/tmp/x"), 4096); + } + + /// Who the pump offered to, by the order it offered them. + fn offered( + ctx: &mut ClientContext, + token: &mut impl FnMut() -> u32, + ) -> Vec { + ctx.pump_uploads(token) + .1 + .into_iter() + .map(|offer| offer.requester_key) + .collect() + } + + #[test] + fn a_fresh_context_starts_with_the_default_cap() { + assert_eq!(ClientContext::new().upload_slots, DEFAULT_UPLOAD_SLOTS); + } + + #[test] + fn the_pump_fills_the_slots_and_leaves_the_rest_queued() { + let (mut ctx, mut token) = context(2); + for user in ["a", "b", "c"] { + ask(&mut ctx, user, "f.mp3"); + } + + assert_eq!(offered(&mut ctx, &mut token), ["a", "b"]); + assert_eq!( + ctx.place_in_queue("c", "f.mp3"), + Some(1), + "the third asker waits, now at the front of the queue" + ); + assert!( + offered(&mut ctx, &mut token).is_empty(), + "pumping again must not oversubscribe the slots" + ); + } + + #[test] + fn a_privileged_list_arriving_late_re_ranks_who_is_waiting() { + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "blocker", "f.mp3"); + ask(&mut ctx, "plain", "f.mp3"); + ask(&mut ctx, "donor", "f.mp3"); + assert_eq!(offered(&mut ctx, &mut token), ["blocker"]); + + // Before the server has said anything, first-come holds. + assert_eq!(ctx.place_in_queue("plain", "f.mp3"), Some(1)); + assert_eq!(ctx.place_in_queue("donor", "f.mp3"), Some(2)); + + ctx.set_privileged_users(vec!["donor".to_string()]); + + assert!(ctx.is_privileged("donor")); + assert!(!ctx.is_privileged("plain")); + assert_eq!( + ctx.place_in_queue("donor", "f.mp3"), + Some(1), + "the donor should overtake the peer that asked first" + ); + assert_eq!( + ctx.place_in_queue("plain", "f.mp3"), + Some(2), + "and the plain peer should be told they were overtaken" + ); + } + + #[test] + fn the_privileged_peer_is_the_one_served_next() { + let (mut ctx, mut token) = context(1); + ctx.set_privileged_users(vec!["donor".to_string()]); + + // Someone else already holds the only slot: a transfer in progress is + // never preempted, so the donor's turn is the *next* one. + ask(&mut ctx, "blocker", "f.mp3"); + assert_eq!(offered(&mut ctx, &mut token), ["blocker"]); + + ask(&mut ctx, "plain", "f.mp3"); + ask(&mut ctx, "donor", "f.mp3"); + + ctx.uploads.clear(); // the blocker's transfer finished + assert_eq!( + offered(&mut ctx, &mut token), + ["donor"], + "the freed slot goes to the donor, not the longer-waiting peer" + ); + } + + #[test] + fn a_finished_transfer_frees_its_slot_for_the_next_in_line() { + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "first", "f.mp3"); + ask(&mut ctx, "second", "f.mp3"); + assert_eq!(offered(&mut ctx, &mut token), ["first"]); + + ctx.uploads.clear(); + assert_eq!(offered(&mut ctx, &mut token), ["second"]); + assert_eq!(ctx.place_in_queue("second", "f.mp3"), None); + } + + #[test] + fn completed_transfers_do_not_keep_occupying_slots() { + // active_uploads keeps finished transfers so `uploads()` can report + // them; counting those would wedge the queue shut for the session. + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "first", "f.mp3"); + offered(&mut ctx, &mut token); + + let job = ctx.uploads.remove(&1).expect("the offer was recorded"); + ctx.active_uploads.insert( + 1, + crate::client::ActiveUpload { + username: job.downloader, + filename: job.virtual_path, + size: job.size, + bytes_sent: std::sync::Arc::new( + std::sync::atomic::AtomicU64::new(job.size), + ), + cancel: std::sync::Arc::new( + std::sync::atomic::AtomicBool::new(false), + ), + status: UploadStatus::Completed, + started: std::time::Instant::now(), + }, + ); + + ask(&mut ctx, "second", "f.mp3"); + assert_eq!( + offered(&mut ctx, &mut token), + ["second"], + "a completed upload must not hold its slot" + ); + } + + #[test] + fn queued_uploads_are_reported_in_the_order_they_will_be_served() { + let (mut ctx, mut token) = context(1); + ctx.set_privileged_users(vec!["donor".to_string()]); + ask(&mut ctx, "blocker", "f.mp3"); + offered(&mut ctx, &mut token); + ask(&mut ctx, "plain", "f.mp3"); + ask(&mut ctx, "donor", "f.mp3"); + + let reported: Vec<(String, UploadStatus)> = ctx + .queued_uploads() + .into_iter() + .map(|upload| (upload.username, upload.status)) + .collect(); + assert_eq!( + reported, + [ + ("donor".to_string(), UploadStatus::Queued(1)), + ("plain".to_string(), UploadStatus::Queued(2)), + ] + ); + } +} diff --git a/soulseek-rs-lib/src/client/uploads.rs b/soulseek-rs-lib/src/client/uploads.rs index 1d97daa..b0c5ede 100644 --- a/soulseek-rs-lib/src/client/uploads.rs +++ b/soulseek-rs-lib/src/client/uploads.rs @@ -1,11 +1,104 @@ use super::{ - ActiveUpload, Arc, Client, ClientContext, DownloadStatus, RwLock, - RwLockExt, collect_failed_tokens, error, thread, + ActiveUpload, Arc, Client, ClientContext, DownloadStatus, PeerMessage, + RwLock, RwLockExt, collect_failed_tokens, error, next_upload_token, thread, }; use crate::types::UploadStatus; use std::sync::atomic::{AtomicBool, AtomicU64}; +/// The peer behind a registry key, which may carry a `:direct` suffix the +/// server never sees. +pub fn downloader_of(requester_key: &str) -> &str { + requester_key + .strip_suffix(":direct") + .unwrap_or(requester_key) +} + impl Client { + /// Whether the server listed `username` as privileged. Privileged peers are + /// served before everyone else. + #[must_use] + pub fn is_privileged(&self, username: &str) -> bool { + self.context + .read_safe() + .is_ok_and(|ctx| ctx.is_privileged(username)) + } + + /// Ask the server how much of our own privilege time is left (code 92). The + /// answer arrives asynchronously; read it with + /// [`Client::own_privilege_seconds`]. + /// + /// # Errors + /// [`crate::error::SoulseekRs::NotConnected`] without a server connection. + pub fn check_privileges(&self) -> crate::error::Result<()> { + let Some(handle) = &self.server_handle else { + return Err(crate::error::SoulseekRs::NotConnected); + }; + let _ = handle.send(super::ServerMessage::CheckPrivileges); + Ok(()) + } + + /// Seconds of our own privileges left, or `None` until the server has + /// answered a [`Client::check_privileges`]. + #[must_use] + pub fn own_privilege_seconds(&self) -> Option { + self.context + .read_safe() + .ok() + .and_then(|ctx| ctx.own_privileges) + } + + /// How many uploads run at once. Lowering it does not interrupt transfers + /// already in flight; the queue simply refills more slowly. + pub fn set_upload_slots(&self, slots: usize) { + if let Ok(mut ctx) = self.context.write_safe() { + ctx.upload_slots = slots.max(1); + } + } + + /// Drop everything `username` was queued or offered, then hand their slot to + /// whoever is next. + pub(crate) fn release_upload_slots( + client_context: &Arc>, + username: &str, + ) { + let freed = client_context + .write_safe() + .is_ok_and(|mut ctx| ctx.release_upload_slots(username)); + if freed { + Self::pump_upload_queue(client_context); + } + } + + /// Fill every free upload slot from the queue and send the offers. + /// + /// Called whenever the queue or the slot count could have changed: a new + /// request, a finished transfer, or a privileged list that re-ranked who is + /// waiting. + pub(crate) fn pump_upload_queue( + client_context: &Arc>, + ) { + let (registry, offers) = match client_context.write_safe() { + Ok(mut ctx) => ctx.pump_uploads(next_upload_token), + Err(e) => { + error!("[client] pump_upload_queue write: {}", e); + return; + } + }; + let Some(registry) = registry else { + return; + }; + for offer in offers { + let _ = registry.send_to_peer( + &offer.requester_key, + PeerMessage::ServeUpload { + token: offer.token, + filename: offer.virtual_path, + size: offer.size, + }, + ); + } + } + /// Consume the upload job for `token` and stream the file to `host:port` /// on a background thread. pub(crate) fn spawn_serve( @@ -64,6 +157,9 @@ impl Client { { upload.status = status; } + // The slot this transfer held is free now, so whoever is next in + // line gets it without waiting for another request to arrive. + Self::pump_upload_queue(&context); }); } diff --git a/soulseek-rs-lib/src/message/peer/mod.rs b/soulseek-rs-lib/src/message/peer/mod.rs index 1dda0ab..60f9bfe 100644 --- a/soulseek-rs-lib/src/message/peer/mod.rs +++ b/soulseek-rs-lib/src/message/peer/mod.rs @@ -1,6 +1,7 @@ mod file_search_response; mod get_share_file_list; mod peer_init; +mod place_in_queue_request; mod place_in_queue_response; mod queue_upload; mod shared_file_list; @@ -14,6 +15,7 @@ pub use file_search_response::{ }; pub use get_share_file_list::GetShareFileList; pub use peer_init::PeerInit; +pub use place_in_queue_request::PlaceInQueueRequest; pub use place_in_queue_response::PlaceInQueueResponse; pub use queue_upload::QueueUploadHandler; pub use shared_file_list::{ diff --git a/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs b/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs new file mode 100644 index 0000000..de6485f --- /dev/null +++ b/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs @@ -0,0 +1,45 @@ +use crate::{ + message::{Message, MessageHandler}, + peer::PeerMessage, +}; +use std::sync::mpsc::Sender; + +/// `PlaceInQueueRequest` (peer code 51): a peer asking where the file they +/// queued sits in our upload queue. +/// +/// Worth answering rather than dropping: a downloader that gets no position +/// cannot tell a long queue from a client that has forgotten about it. +pub struct PlaceInQueueRequest; + +impl MessageHandler for PlaceInQueueRequest { + fn get_code(&self) -> u8 { + 51 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let filename = message.read_string(); + let _ = sender.send(PeerMessage::IncomingPlaceInQueueRequest(filename)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_asked_about_filename_is_forwarded() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = Message::new(); + message.write_raw_bytes(vec![0u8; 8]); + message.write_string("@@share\\track.mp3"); + message.set_pointer(8); + + PlaceInQueueRequest.handle(&mut message, tx); + match rx.try_recv() { + Ok(PeerMessage::IncomingPlaceInQueueRequest(filename)) => { + assert_eq!(filename, "@@share\\track.mp3"); + } + other => panic!("unexpected: {other:?}"), + } + } +} diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 08be49f..c7da605 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -87,6 +87,38 @@ impl MessageFactory { .clone() } + /// Ask the server (code 92) how much of our own privilege time is left. + #[must_use] + pub fn build_check_privileges() -> Message { + Message::new().write_int32(92).clone() + } + + /// Tell a peer where their queued file sits (peer code 44). Place counts + /// from 1. + #[must_use] + pub fn build_place_in_queue_response( + filename: &str, + place: u32, + ) -> Message { + Message::new() + .write_int32(44) + .write_string(filename) + .write_int32(place) + .clone() + } + + /// A wishlist search (server code 103). Same shape as a plain `FileSearch`, + /// but the server rate-limits it to the interval it announced in code 104 + /// and does not count it against the normal search allowance. + #[must_use] + pub fn build_wishlist_search(token: u32, query: &str) -> Message { + Message::new() + .write_int32(103) + .write_int32(token) + .write_string(query) + .clone() + } + #[must_use] pub fn build_set_status_message(status_code: u32) -> Message { Message::new() @@ -368,6 +400,37 @@ fn test_build_room_list_request() { assert_eq!(vec![64, 0, 0, 0], message.get_data()); } +#[test] +fn test_build_check_privileges() { + assert_eq!( + vec![92, 0, 0, 0], + MessageFactory::build_check_privileges().get_data() + ); +} + +#[test] +fn test_build_place_in_queue_response() { + let message = MessageFactory::build_place_in_queue_response("song.mp3", 3); + let expect: Vec = [ + 44, 0, 0, 0, // code + 8, 0, 0, 0, 115, 111, 110, 103, 46, 109, 112, 51, // "song.mp3" + 3, 0, 0, 0, // place + ] + .to_vec(); + assert_eq!(expect, message.get_data()); +} + +#[test] +fn test_build_wishlist_search() { + let message = MessageFactory::build_wishlist_search(12, "trance wax"); + let expect: Vec = [ + 103, 0, 0, 0, 12, 0, 0, 0, 10, 0, 0, 0, 116, 114, 97, 110, 99, 101, 32, + 119, 97, 120, + ] + .to_vec(); + assert_eq!(expect, message.get_data()); +} + #[test] fn test_build_file_search_message() { let message = MessageFactory::build_file_search_message(12, "trance wax"); diff --git a/soulseek-rs-lib/src/message/server/mod.rs b/soulseek-rs-lib/src/message/server/mod.rs index d5e3baf..f3fa4bc 100644 --- a/soulseek-rs-lib/src/message/server/mod.rs +++ b/soulseek-rs-lib/src/message/server/mod.rs @@ -28,7 +28,7 @@ pub use message_factory::MessageFactory; pub use message_user::MessageUser; pub use parent_min_speed::ParentMinSpeedHandler; pub use parent_speed_ratio::ParentSpeedRatioHandler; -pub use privileged_users::PrivilegedUsersHandler; +pub use privileged_users::{CheckPrivilegesHandler, PrivilegedUsersHandler}; pub use room_list::RoomListHandler; pub use say_chatroom::SayChatroomHandler; pub use user_info::{GetUserStatsHandler, GetUserStatusHandler}; diff --git a/soulseek-rs-lib/src/message/server/privileged_users.rs b/soulseek-rs-lib/src/message/server/privileged_users.rs index 9805388..8f0cc37 100644 --- a/soulseek-rs-lib/src/message/server/privileged_users.rs +++ b/soulseek-rs-lib/src/message/server/privileged_users.rs @@ -1,4 +1,3 @@ -use crate::debug; use std::sync::mpsc::Sender; use crate::{ @@ -6,6 +5,12 @@ use crate::{ message::{Message, MessageHandler}, }; +/// `PrivilegedUsers` (code 69): everyone the server counts as having donated, +/// sent once at login. +/// +/// The names matter, not the count: they decide who overtakes whom in our +/// upload queue, which is the "recognition of privileges" an alternative client +/// is expected to implement. pub struct PrivilegedUsersHandler; impl MessageHandler for PrivilegedUsersHandler { @@ -13,8 +18,105 @@ impl MessageHandler for PrivilegedUsersHandler { 69 } - fn handle(&self, message: &mut Message, _sender: Sender) { - let number = message.read_int32(); - debug!("Number of privileged users: {}", number); + fn handle(&self, message: &mut Message, sender: Sender) { + let count = message.read_int32(); + let mut users = Vec::new(); + for _ in 0..count { + // Guard against a hostile count outrunning the payload. + if message.get_pointer() + 4 > message.get_size() { + break; + } + users.push(message.read_string()); + } + let _ = sender.send(ServerMessage::PrivilegedUsers(users)); + } +} + +/// `CheckPrivileges` (code 92): how many seconds of our own privileges are +/// left. Zero means we have none. +pub struct CheckPrivilegesHandler; + +impl MessageHandler for CheckPrivilegesHandler { + fn get_code(&self) -> u8 { + 92 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let seconds = message.read_int32(); + let _ = sender.send(ServerMessage::OwnPrivileges(seconds)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn payload(build: impl FnOnce(&mut Message)) -> Message { + let mut message = Message::new(); + message.write_raw_bytes(vec![0u8; 8]); + build(&mut message); + message.set_pointer(8); + message + } + + #[test] + fn the_user_list_is_kept_not_just_counted() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = payload(|m| { + m.write_int32(2) + .write_string("donor_a") + .write_string("donor_b"); + }); + + PrivilegedUsersHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::PrivilegedUsers(users)) => { + assert_eq!(users, ["donor_a", "donor_b"]); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn an_empty_list_is_still_an_answer() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = payload(|m| { + m.write_int32(0); + }); + + PrivilegedUsersHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::PrivilegedUsers(users)) => { + assert!(users.is_empty()); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn a_hostile_count_does_not_hang() { + // count=u32::MAX with no names: the guard must return promptly rather + // than loop four billion times. + let (tx, _rx) = std::sync::mpsc::channel(); + let mut message = payload(|m| { + m.write_int32(u32::MAX); + }); + PrivilegedUsersHandler.handle(&mut message, tx); + } + + #[test] + fn our_own_remaining_privilege_time_is_reported() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = payload(|m| { + m.write_int32(86_400); + }); + + CheckPrivilegesHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::OwnPrivileges(seconds)) => { + assert_eq!(seconds, 86_400); + } + other => panic!("unexpected: {other:?}"), + } } } diff --git a/soulseek-rs-lib/src/message/server/wish_list_interval.rs b/soulseek-rs-lib/src/message/server/wish_list_interval.rs index b763149..1b57c8c 100644 --- a/soulseek-rs-lib/src/message/server/wish_list_interval.rs +++ b/soulseek-rs-lib/src/message/server/wish_list_interval.rs @@ -1,4 +1,3 @@ -use crate::debug; use std::sync::mpsc::Sender; use crate::{ @@ -6,17 +5,42 @@ use crate::{ message::{Message, MessageHandler}, }; +/// `WishlistInterval` (code 104): how often the server will accept a wishlist +/// search from us. Usually 720 seconds, or 120 for a privileged account. +/// +/// The value is the server's rate limit, not a suggestion, so it is kept rather +/// than logged: the re-search timer runs off it. pub struct WishListIntervalHandler; -// The server tells us the wishlist search interval. -// This interval is almost always 12 minutes, or 2 minutes for privileged users. impl MessageHandler for WishListIntervalHandler { fn get_code(&self) -> u8 { 104 } - fn handle(&self, message: &mut Message, _sender: Sender) { - let number = message.read_int32(); - debug!("Wishlist search interval: {} in seconds", number); + fn handle(&self, message: &mut Message, sender: Sender) { + let seconds = message.read_int32(); + let _ = sender.send(ServerMessage::WishlistInterval(seconds)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_announced_interval_is_reported_not_discarded() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = Message::new(); + message.write_raw_bytes(vec![0u8; 8]); + message.write_int32(720); + message.set_pointer(8); + + WishListIntervalHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::WishlistInterval(seconds)) => { + assert_eq!(seconds, 720); + } + other => panic!("unexpected: {other:?}"), + } } } diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index c29069d..f133b15 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -377,6 +377,8 @@ mod tests { /// Lifecycle of a file we are serving to a peer. #[derive(Debug, Clone, PartialEq, Eq)] pub enum UploadStatus { + /// Waiting for a free upload slot, at this 1-based place in the queue. + Queued(u32), InProgress, Completed, Cancelled, diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 11ffbd3..8f3567d 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -30,8 +30,20 @@ use soulseek_rs::message::server::MessageFactory; use soulseek_rs::peer::ConnectionType; use soulseek_rs::{ Client, ClientSettings, ClientVersion, DownloadStatus, PeerAddress, + UploadStatus, }; +/// Only one test at a time may drive a server. +/// +/// Each of these tests runs its own soulfind plus real clients and, for the +/// queue tests, peer sockets held open on purpose. Letting the harness start a +/// handful of those at once starves them, and a login that goes unanswered +/// fails a test for a reason that has nothing to do with the code under test: +/// seen on CI as `login: Timeout` in two download tests that the change under +/// test never went near. The CLI suite already carries this gate for the same +/// reason. +static SERVER_GATE: std::sync::Mutex<()> = std::sync::Mutex::new(()); + /// A Soulseek server to test against: either a child soulfind process we /// spawned, or an external server referenced by `SOULSEEK_TEST_SERVER`. struct TestServer { @@ -39,11 +51,18 @@ struct TestServer { port: u16, child: Option, db: Option, + _gate: std::sync::MutexGuard<'static, ()>, } impl TestServer { /// Resolve a server to test against, or `None` if the suite should skip. fn resolve() -> Option { + // A test that panicked while holding the gate poisoned it; the lock + // guards nothing but scheduling, so take it back and carry on. + let gate = SERVER_GATE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Ok(addr) = std::env::var("SOULSEEK_TEST_SERVER") { let (host, port) = split_host_port(&addr)?; wait_until_listening(&host, port, Duration::from_secs(2))?; @@ -52,13 +71,14 @@ impl TestServer { port, child: None, db: None, + _gate: gate, }); } - Self::spawn() + Self::spawn(gate) } /// Spawn a local soulfind on an ephemeral port with a throwaway database. - fn spawn() -> Option { + fn spawn(gate: std::sync::MutexGuard<'static, ()>) -> Option { let bin = soulfind_binary()?; let port = free_port()?; let db = std::env::temp_dir().join(format!("soulfind-e2e-{port}.db")); @@ -88,6 +108,7 @@ impl TestServer { port, child: Some(child), db: Some(db), + _gate: gate, }) } @@ -1818,6 +1839,350 @@ fn the_server_reports_another_users_status_and_share_counts() { let _ = std::fs::remove_dir_all(share_dir); } +// --- upload slots and privilege recognition -------------------------------- +// +// Soulseek's rules ask an alternative client to *recognise* privileges. That is +// only observable if there is a queue to jump, so these tests hold the single +// upload slot open with a peer that queues a file and then never answers the +// TransferRequest — the same thing a stalled client does, and deterministic in +// a way that racing real 4 KB transfers is not. + +/// A raw peer that logs in, dials our listener, queues `filename`, and then +/// goes quiet. Returned so the caller can keep it alive: dropping the streams +/// would release the slot it is holding. +struct QueueingPeer { + _server: TcpStream, + peer: TcpStream, +} + +impl QueueingPeer { + /// Wait for the upload offer (peer code 40) and deliberately not answer it. + /// + /// Receiving the offer is the only positive proof that this peer was given a + /// slot — an unaccepted offer never reaches `Client::uploads()`, and "has no + /// place in the queue" is also true of a peer that never asked, so polling + /// for that would pass against a completely broken pump. + fn takes_a_slot(&mut self) -> bool { + read_until_code(&mut self.peer, 40, Duration::from_secs(10)).is_some() + } +} + +fn queue_as( + server_addr: &str, + listen_addr: &str, + username: &str, + filename: &str, +) -> std::io::Result { + let server = login_raw(server_addr, username, "pw")?; + let mut peer = connect_retry(listen_addr, Duration::from_secs(5))?; + peer.set_read_timeout(Some(Duration::from_secs(10)))?; + peer.write_all(&peer_init_bytes(username, "P", 0))?; + let mut queue = Message::new(); + queue.write_int32(43).write_string(filename); + peer.write_all(&queue.get_buffer())?; + peer.flush()?; + Ok(QueueingPeer { + _server: server, + peer, + }) +} + +/// Wait for `check` to hold, polling the client's view of its own queue. +fn wait_for(mut check: impl FnMut() -> bool) -> bool { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if check() { + return true; + } + std::thread::sleep(Duration::from_millis(100)); + } + false +} + +/// Where `username` sits in `sharer`'s upload queue, read the way the CLI reads +/// it — through `uploads()` — rather than through a point query kept alive only +/// for tests. +fn place_of(sharer: &Client, username: &str, filename: &str) -> Option { + sharer.uploads().into_iter().find_map(|upload| { + match (upload.status, upload.username == username) { + (UploadStatus::Queued(place), true) + if upload.filename == filename => + { + Some(place) + } + _ => None, + } + }) +} + +/// Give `username` privileges in soulfind's own database. +/// +/// `false` means it could not be done — an external server, or no `sqlite3` on +/// this machine — which the caller treats as a skip rather than a failure. +fn grant_privileges(server: &TestServer, username: &str) -> bool { + let Some(db) = server.db.as_ref() else { + return false; + }; + // soulfind stores privileges as the unix timestamp they expire at. + let expiry = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| since.as_secs() + 86_400); + Command::new("sqlite3") + .arg(db) + .arg(format!( + "UPDATE users SET privileges = {expiry} WHERE username = \ + '{username}';" + )) + .status() + .is_ok_and(|status| status.success()) +} + +/// A share directory of its own, so the virtual paths a peer has to name are +/// predictable: soulfind-facing paths are `\`. +fn queue_share(label: &str, files: &[&str]) -> (PathBuf, String) { + // Keyed on the thread as well as the process, matching + // `unique_download_dir`: two tests are two threads, and a shared folder + // name would make their virtual paths collide. + let dir = std::env::temp_dir().join(format!( + "soulseek-e2e-queue-{label}-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("share dir"); + for name in files { + std::fs::write(dir.join(name), vec![9u8; 4096]).expect("share file"); + } + let folder = dir + .file_name() + .and_then(|n| n.to_str()) + .expect("utf-8 folder") + .to_string(); + (dir, folder) +} + +#[test] +fn a_second_request_waits_when_the_only_upload_slot_is_taken() { + let server = server_or_skip!(); + let (share, folder) = + queue_share("slots", &["blocker.mp3", "waiter.mp3", "third.mp3"]); + let server_addr = format!("{}:{}", server.host, server.port); + + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share.display().to_string()], + ..server.listening_settings("e2e_q_sharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + sharer.set_upload_slots(1); + let listen_addr = format!("127.0.0.1:{sharer_port}"); + + let blocker_file = format!("{folder}\\blocker.mp3"); + let mut blocker = + queue_as(&server_addr, &listen_addr, "e2e_q_blocker", &blocker_file) + .expect("blocker queues"); + assert!( + blocker.takes_a_slot(), + "the first request should be offered the free slot" + ); + + let waiter_file = format!("{folder}\\waiter.mp3"); + let _waiter = + queue_as(&server_addr, &listen_addr, "e2e_q_waiter", &waiter_file) + .expect("waiter queues"); + assert!( + wait_for(|| place_of(&sharer, "e2e_q_waiter", &waiter_file) == Some(1)), + "with the slot held, the next request must wait at place 1, got {:?}", + place_of(&sharer, "e2e_q_waiter", &waiter_file) + ); + + let third_file = format!("{folder}\\third.mp3"); + let _third = + queue_as(&server_addr, &listen_addr, "e2e_q_third", &third_file) + .expect("third queues"); + assert!( + wait_for(|| place_of(&sharer, "e2e_q_third", &third_file) == Some(2)), + "and the one after that at place 2, got {:?}", + place_of(&sharer, "e2e_q_third", &third_file) + ); + + // Both waiters show up as queued uploads with their places, which is what + // `serve` reports to an operator. + let queued: Vec<(String, u32)> = sharer + .uploads() + .into_iter() + .filter_map(|upload| match upload.status { + UploadStatus::Queued(place) => Some((upload.username, place)), + _ => None, + }) + .collect(); + assert_eq!( + queued, + [ + ("e2e_q_waiter".to_string(), 1), + ("e2e_q_third".to_string(), 2) + ] + ); + + let _ = std::fs::remove_dir_all(share); +} + +#[test] +fn a_privileged_peer_overtakes_one_already_waiting_for_a_slot() { + let server = server_or_skip!(); + let server_addr = format!("{}:{}", server.host, server.port); + + // The account has to exist before it can be given privileges, and the + // privileges have to exist before we log in: the server sends the + // privileged-user list once, at login. + drop( + login_raw(&server_addr, "e2e_q_donor", "pw") + .expect("the donor registers"), + ); + if !grant_privileges(&server, "e2e_q_donor") { + eprintln!( + "privilege e2e skipped: cannot write soulfind's database (needs a \ + locally spawned server and sqlite3)" + ); + return; + } + + let (share, folder) = + queue_share("privileged", &["blocker.mp3", "plain.mp3", "donor.mp3"]); + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share.display().to_string()], + ..server.listening_settings("e2e_q_psharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + sharer.set_upload_slots(1); + let listen_addr = format!("127.0.0.1:{sharer_port}"); + + // Not every soulfind build answers login with a populated code-69 list, so + // a server that names nobody is a skip rather than a failure — the + // re-ranking itself is covered by `client::upload_queue::context_tests`. + if !wait_for(|| sharer.is_privileged("e2e_q_donor")) { + eprintln!( + "privilege ordering e2e skipped: this server declared no \ + privileged users even after granting them in its database" + ); + let _ = std::fs::remove_dir_all(share); + return; + } + assert!(!sharer.is_privileged("e2e_q_plain"), "and nobody else"); + + let blocker_file = format!("{folder}\\blocker.mp3"); + let mut blocker = + queue_as(&server_addr, &listen_addr, "e2e_q_pblocker", &blocker_file) + .expect("blocker queues"); + assert!( + blocker.takes_a_slot(), + "the blocker should be offered the only slot" + ); + + // A plain user asks first and is alone in the queue… + let plain_file = format!("{folder}\\plain.mp3"); + let _plain = + queue_as(&server_addr, &listen_addr, "e2e_q_plain", &plain_file) + .expect("plain queues"); + assert!( + wait_for(|| place_of(&sharer, "e2e_q_plain", &plain_file) == Some(1)), + "got {:?}", + place_of(&sharer, "e2e_q_plain", &plain_file) + ); + + // …and then the donor asks, and goes ahead of them. + let donor_file = format!("{folder}\\donor.mp3"); + let _donor = + queue_as(&server_addr, &listen_addr, "e2e_q_donor", &donor_file) + .expect("donor queues"); + assert!( + wait_for(|| place_of(&sharer, "e2e_q_donor", &donor_file) == Some(1)), + "the privileged peer should take place 1, got {:?}", + place_of(&sharer, "e2e_q_donor", &donor_file) + ); + assert_eq!( + place_of(&sharer, "e2e_q_plain", &plain_file), + Some(2), + "and the plain user should be told they were overtaken" + ); + + let _ = std::fs::remove_dir_all(share); +} + +#[test] +fn a_peer_asking_where_it_sits_is_answered_with_its_place() { + let server = server_or_skip!(); + let (share, folder) = queue_share("place", &["blocker.mp3", "asker.mp3"]); + let server_addr = format!("{}:{}", server.host, server.port); + + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share.display().to_string()], + ..server.listening_settings("e2e_q_asharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + sharer.set_upload_slots(1); + let listen_addr = format!("127.0.0.1:{sharer_port}"); + + let blocker_file = format!("{folder}\\blocker.mp3"); + let mut blocker = + queue_as(&server_addr, &listen_addr, "e2e_q_ablocker", &blocker_file) + .expect("blocker queues"); + assert!( + blocker.takes_a_slot(), + "the blocker should be offered the only slot" + ); + + // The asker keeps its own connection so it can read the answer back. + // The asker keeps its connection so it can read the answer back. + let asker_file = format!("{folder}\\asker.mp3"); + let mut asker = + queue_as(&server_addr, &listen_addr, "e2e_q_asker", &asker_file) + .expect("asker queues"); + + assert!( + wait_for(|| place_of(&sharer, "e2e_q_asker", &asker_file) == Some(1)), + "the asker should be waiting at place 1 first" + ); + + let mut request = Message::new(); + request.write_int32(51).write_string(&asker_file); + asker.peer.write_all(&request.get_buffer()).expect("ask"); + asker.peer.flush().expect("flush"); + + let mut answer = expect_code(&mut asker.peer, 44, Duration::from_secs(10)) + .expect("a PlaceInQueueResponse should come back"); + answer.set_pointer(8); + assert_eq!(answer.read_string(), asker_file); + assert_eq!(answer.read_int32(), 1, "and it should carry the real place"); + + let _ = std::fs::remove_dir_all(share); +} + +#[test] +fn the_server_answers_how_much_privilege_time_we_have() { + let server = server_or_skip!(); + let mut client = + Client::with_settings(server.settings("e2e_priv_asker", "pw")); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + client.check_privileges().expect("ask"); + assert!( + wait_for(|| client.own_privilege_seconds().is_some()), + "the server should answer code 92" + ); + assert_eq!( + client.own_privilege_seconds(), + Some(0), + "a fresh account has no privileges, which is zero rather than unknown" + ); +} + #[test] fn a_user_who_never_logged_in_reads_as_offline() { let server = server_or_skip!(); diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index b374354..0ddcc04 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -45,9 +45,12 @@ Each entry lists the keys of the objects it prints. **`search `** — one object per matching file: `user`, `path`, `size`, `bitrate`, `duration`, `free_slot`, `slots`, `speed`. -Flags: `--min-bitrate `, `--free-slots`, `-n/--limit` (0 means all), -`--sort best|size|bitrate|speed`. Collects results for the whole -`--search-timeout` window before printing anything. +Filters: `--min-bitrate `, `--free-slots`, `--exclude ` (repeatable), +`--extension ` (repeatable, dot optional), `--min-size `, +`--max-size `. Also `-n/--limit` (0 means all) and +`--sort best|size|bitrate|speed`. Every filter given has to be satisfied. +Collects results for the whole `--search-timeout` window before printing +anything. **`download [--size N]`** or **`download --stdin`** — one object per finished transfer: `user`, `path`, `size`, `file` (the local path written). @@ -56,9 +59,20 @@ per finished transfer: `user`, `path`, `size`, `file` (the local path written). starting over, so retrying a failed download is cheap. **`get `** — search, pick, and download in one step. `--pick -best|first|all`, `-n/--limit`, `--min-bitrate`, `--free-slots`. Prefer this +best|first|all`, `-n/--limit`, plus every filter `search` accepts. Prefer this over `search` + `download` when the user just wants the file. +**`wish list`** — one object per standing search: `query`. +**`wish add `** and **`wish remove `** — change the wishlist in +the config file. `add` prints the stored `query`; `remove` prints nothing. A +repeat of an existing wish is not an error. +**`wish run`** — search every stored wish once and print `search`-shaped +objects, so the output pipes straight into `download --stdin`. `--only ` +runs one stored wish (it must already be on the list), and every `search` filter +applies. Exit 4 when no wish matched anything. Use this when the user wants +something that is not on the network right now: add it as a wish, then run the +wishlist later rather than repeating the same search. + **`browse `** — one object per file the peer shares: `user`, `directory`, `path`, `size`. @@ -68,8 +82,10 @@ be `null` when the server has not answered that half — `null` means unknown, not offline. **`whoami`** — one object: `user`, `server`, `listening`, `listen_port`, -`shared_folders`, `shared_files`, `download_dir`. The cheapest way to prove the -credentials and connection work. +`shared_folders`, `shared_files`, `download_dir`, `privilege_seconds`. The +cheapest way to prove the credentials and connection work. +`privilege_seconds` is 0 for an ordinary account and `null` only when the +server did not answer — those are different facts, so do not treat null as 0. **`room list`** — `room`, `users`. **`room users `** — `room`, `user`. @@ -81,7 +97,13 @@ credentials and connection work. **`serve`** — stay online sharing files, one object per upload change: `user`, `path`, `size`, `bytes_sent`, `speed`, `status` -(`uploading`/`completed`/`cancelled`/`failed`), `reason`. +(`queued`/`uploading`/`completed`/`cancelled`/`failed`), `reason`. A `queued` +record's `reason` is its place in the queue. `--upload-slots N` sets how many +uploads run at once (2 by default); the rest wait, with users the server lists +as privileged served first. With `--wishlist` it also re-runs the wishlist on +the interval the server sets and prints `search`-shaped objects for what it +finds, so one long-running process covers both halves of being a good +participant. **`shares list`** — one object per configured folder: `directory`, `usable`. **`shares add `** and **`shares remove `** — change the diff --git a/soulseek-rs/src/cli.rs b/soulseek-rs/src/cli.rs index e8923e5..86d53b0 100644 --- a/soulseek-rs/src/cli.rs +++ b/soulseek-rs/src/cli.rs @@ -168,6 +168,10 @@ pub enum Commands { /// List the files a user shares Browse(BrowseArgs), + /// Standing searches the server re-runs on its own schedule + #[command(subcommand)] + Wish(WishCommand), + /// Chat rooms #[command(subcommand)] Room(RoomCommand), @@ -265,6 +269,15 @@ pub struct ServeArgs { /// Stay online until interrupted, ignoring --duration #[arg(long, action = ArgAction::SetTrue)] pub follow: bool, + + /// Also re-run the wishlist on the interval the server sets, reporting + /// matches as they arrive + #[arg(long, action = ArgAction::SetTrue)] + pub wishlist: bool, + + /// How many uploads to run at once; the rest queue, privileged users first + #[arg(long, value_name = "N", value_parser = clap::value_parser!(u8).range(1..=64))] + pub upload_slots: Option, } #[derive(Args, Debug)] @@ -353,11 +366,10 @@ pub enum Pick { All, } -#[derive(Args, Debug)] -pub struct SearchArgs { - /// What to search for - pub query: String, - +/// The result restrictions every command that searches accepts, so `search`, +/// `get` and `wish run` all narrow a result set the same way. +#[derive(Args, Debug, Default, Clone)] +pub struct FilterArgs { /// Drop results below this bitrate in kbps #[arg(long, value_name = "KBPS")] pub min_bitrate: Option, @@ -366,6 +378,31 @@ pub struct SearchArgs { #[arg(long, action = ArgAction::SetTrue)] pub free_slots: bool, + /// Drop results whose path contains this term; repeat for several + #[arg(long, value_name = "TERM", action = ArgAction::Append)] + pub exclude: Vec, + + /// Keep only this file extension, with or without a dot; repeat for several + #[arg(long, value_name = "EXT", action = ArgAction::Append)] + pub extension: Vec, + + /// Drop results smaller than this many bytes + #[arg(long, value_name = "BYTES")] + pub min_size: Option, + + /// Drop results larger than this many bytes + #[arg(long, value_name = "BYTES")] + pub max_size: Option, +} + +#[derive(Args, Debug)] +pub struct SearchArgs { + /// What to search for + pub query: String, + + #[command(flatten)] + pub filter: FilterArgs, + /// Print at most this many results (0 means all) #[arg(short = 'n', long, default_value_t = 0)] pub limit: usize, @@ -417,13 +454,8 @@ pub struct GetArgs { #[arg(short = 'n', long, default_value_t = 0)] pub limit: usize, - /// Drop results below this bitrate in kbps - #[arg(long, value_name = "KBPS")] - pub min_bitrate: Option, - - /// Keep only peers advertising a free upload slot - #[arg(long, action = ArgAction::SetTrue)] - pub free_slots: bool, + #[command(flatten)] + pub filter: FilterArgs, /// Seconds to wait for each transfer before giving up #[arg( @@ -450,6 +482,45 @@ pub struct BrowseArgs { pub timeout: u64, } +#[derive(Subcommand, Debug)] +pub enum WishCommand { + /// Store a query and search for it from now on + Add { + /// What to keep looking for + query: String, + }, + + /// Stop looking for a query + Remove { + /// The wish to drop, as shown by `wish list` + query: String, + }, + + /// Print the stored queries + List, + + /// Search every stored query once and print what came back + Run(WishRunArgs), +} + +#[derive(Args, Debug)] +pub struct WishRunArgs { + /// Only run this wish instead of all of them + #[arg(long, value_name = "QUERY")] + pub only: Option, + + #[command(flatten)] + pub filter: FilterArgs, + + /// Print at most this many results per wish (0 means all) + #[arg(short = 'n', long, default_value_t = 0)] + pub limit: usize, + + /// Result order + #[arg(long, value_enum, default_value_t = SortKey::Best)] + pub sort: SortKey, +} + #[derive(Subcommand, Debug)] pub enum RoomCommand { /// List public rooms and their user counts diff --git a/soulseek-rs/src/commands/mod.rs b/soulseek-rs/src/commands/mod.rs index 11a303b..e644b77 100644 --- a/soulseek-rs/src/commands/mod.rs +++ b/soulseek-rs/src/commands/mod.rs @@ -11,6 +11,7 @@ pub mod settings; pub mod skills; pub mod social; pub mod transfer; +pub mod wish; use crate::cli::{Commands, MessageCommand, RoomCommand}; use crate::output::{CliError, CliResult, Exit, Out, PortmapRecord}; @@ -191,13 +192,16 @@ pub fn run(ctx: &Ctx, command: Commands) -> CliResult { Commands::Message(MessageCommand::Read { duration, follow }) => { social::message_read(ctx, listen_span(duration, follow)) } - Commands::Serve(args) => peer::serve(ctx, &args), Commands::User(args) => peer::user(ctx, &args), Commands::Whoami => peer::whoami(ctx), // These need no account, so main.rs answers them before asking for // credentials; the arms exist only to keep the match exhaustive. Commands::Portmap => portmap(&ctx.out, ctx.settings.listen_port), - Commands::Config(_) + // These need the config file as well as a session, so main.rs runs + // them where the store is in scope. + Commands::Serve(_) + | Commands::Wish(_) + | Commands::Config(_) | Commands::Shares(_) | Commands::Skills(_) | Commands::Completions(_) => { diff --git a/soulseek-rs/src/commands/peer.rs b/soulseek-rs/src/commands/peer.rs index 56f7abd..894129b 100644 --- a/soulseek-rs/src/commands/peer.rs +++ b/soulseek-rs/src/commands/peer.rs @@ -12,6 +12,11 @@ use std::time::{Duration, Instant}; /// How often the serve loop samples the upload table. const POLL: Duration = Duration::from_millis(500); +/// How long `whoami` waits for the server's privilege answer. It is one small +/// message on an already-open connection, so a server that has not replied by +/// now is not going to. +const PRIVILEGE_TIMEOUT: Duration = Duration::from_secs(5); + pub fn whoami(ctx: &Ctx) -> CliResult { let session = Session::open(ctx)?; let (folders, files) = session.client.shared_counts(); @@ -23,10 +28,29 @@ pub fn whoami(ctx: &Ctx) -> CliResult { shared_folders: folders, shared_files: files, download_dir: ctx.download_dir.clone(), + privilege_seconds: own_privileges(&session.client), }); Ok(()) } +/// Ask the server how much privilege time this account has left. +/// +/// A server that does not answer leaves the field null rather than reporting a +/// zero, because "no privileges" and "did not say" are different facts. +fn own_privileges(client: &soulseek_rs::Client) -> Option { + if client.check_privileges().is_err() { + return None; + } + let deadline = Instant::now() + PRIVILEGE_TIMEOUT; + while Instant::now() < deadline { + if let Some(seconds) = client.own_privilege_seconds() { + return Some(seconds); + } + std::thread::sleep(Duration::from_millis(50)); + } + None +} + pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult { let session = Session::open(ctx)?; session.client.request_user_info(&args.user).map_err(|e| { @@ -69,7 +93,13 @@ pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult { Ok(()) } -pub fn serve(ctx: &Ctx, args: &ServeArgs) -> CliResult { +/// Stay online sharing files, and — with `--wishlist` — re-run the standing +/// searches on the interval the server sets while doing so. +pub fn serve( + ctx: &Ctx, + args: &ServeArgs, + mut sweeper: Option, +) -> CliResult { if ctx.settings.shared_directories.is_empty() { return Err(CliError::usage( "nothing to share: pass --shared-dir, or add one with `shares add`", @@ -82,6 +112,10 @@ pub fn serve(ctx: &Ctx, args: &ServeArgs) -> CliResult { } let session = Session::open(ctx)?; + if let Some(slots) = args.upload_slots { + session.client.set_upload_slots(usize::from(slots)); + ctx.out.status(&format!("{slots} upload slots")); + } let (folders, files) = session.client.shared_counts(); ctx.out.status(&format!( "serving {files} files in {folders} folders on port {}", @@ -109,6 +143,9 @@ pub fn serve(ctx: &Ctx, args: &ServeArgs) -> CliResult { reported.insert(key, record); } } + if let Some(sweeper) = sweeper.as_mut() { + sweeper.tick(ctx, &session.client); + } std::thread::sleep(POLL); } Ok(()) @@ -117,6 +154,9 @@ pub fn serve(ctx: &Ctx, args: &ServeArgs) -> CliResult { /// Flatten a live upload into the record `serve` reports. fn upload_record(upload: &UploadInfo) -> UploadRecord { let (status, reason) = match &upload.status { + UploadStatus::Queued(place) => { + ("queued", Some(format!("place {place}"))) + } UploadStatus::InProgress => ("uploading", None), UploadStatus::Completed => ("completed", None), UploadStatus::Cancelled => ("cancelled", None), diff --git a/soulseek-rs/src/commands/settings.rs b/soulseek-rs/src/commands/settings.rs index 3a89b3a..672cdad 100644 --- a/soulseek-rs/src/commands/settings.rs +++ b/soulseek-rs/src/commands/settings.rs @@ -19,7 +19,7 @@ pub struct Store { } impl Store { - fn save(&self, config: &FileConfig) -> CliResult { + pub fn save(&self, config: &FileConfig) -> CliResult { let Some(path) = &self.path else { return Err(CliError::usage( "no config file to write to (running with --no-config)", diff --git a/soulseek-rs/src/commands/transfer.rs b/soulseek-rs/src/commands/transfer.rs index e34bd4c..8c27a05 100644 --- a/soulseek-rs/src/commands/transfer.rs +++ b/soulseek-rs/src/commands/transfer.rs @@ -1,7 +1,9 @@ //! Finding files and fetching them: `search`, `download`, and `get`. use super::{Ctx, Session, social}; -use crate::cli::{DownloadArgs, GetArgs, Pick, SearchArgs, SortKey}; +use crate::cli::{ + DownloadArgs, FilterArgs, GetArgs, Pick, SearchArgs, SortKey, +}; use crate::output::{ CliError, CliResult, DownloadRecord, Exit, Out, SearchRecord, }; @@ -19,25 +21,69 @@ use std::time::{Duration, Instant}; const ATTR_BITRATE: u32 = 0; const ATTR_DURATION: u32 = 1; -/// What to keep out of the raw result set. -#[derive(Debug, Clone, Copy, Default)] +/// What to keep out of the raw result set. Every field is a further +/// restriction: a candidate has to satisfy all of them. +/// +/// `exclude` and `extensions` are stored already lowercased and without leading +/// dots — see [`Filter::from`] — so matching a result costs no allocation. A +/// search returns thousands of files and every one of them is tested. +#[derive(Debug, Clone, Default)] pub struct Filter { pub min_bitrate: Option, pub free_slots_only: bool, + /// Terms that must not appear anywhere in the remote path, lowercased. + pub exclude: Vec, + /// File extensions to keep, lowercased and dotless. + pub extensions: Vec, + pub min_size: Option, + pub max_size: Option, } impl Filter { - const fn keeps(&self, candidate: &SearchRecord) -> bool { + fn keeps(&self, candidate: &SearchRecord) -> bool { if self.free_slots_only && !candidate.free_slot { return false; } - match (self.min_bitrate, candidate.bitrate) { - (Some(minimum), Some(bitrate)) => bitrate >= minimum, - // An unknown bitrate cannot satisfy a bitrate floor. - (Some(_), None) => false, - (None, _) => true, + if self.min_size.is_some_and(|min| candidate.size < min) { + return false; } + if self.max_size.is_some_and(|max| candidate.size > max) { + return false; + } + // An unknown bitrate cannot satisfy a bitrate floor. + if self + .min_bitrate + .is_some_and(|min| candidate.bitrate.is_none_or(|have| have < min)) + { + return false; + } + if self.exclude.is_empty() && self.extensions.is_empty() { + return true; + } + let path = candidate.path.to_lowercase(); + !self.exclude.iter().any(|term| path.contains(term.as_str())) + && (self.extensions.is_empty() || self.matches_extension(&path)) } + + fn matches_extension(&self, lowercase_path: &str) -> bool { + let name = lowercase_path + .rsplit(['/', '\\']) + .next() + .unwrap_or_default(); + let Some((_, actual)) = name.rsplit_once('.') else { + return false; + }; + self.extensions.iter().any(|wanted| wanted == actual) + } +} + +/// Lowercase and strip a leading dot, so the comparison in [`Filter::keeps`] is +/// a plain equality however the flag was spelled. +fn normalized(terms: &[String]) -> Vec { + terms + .iter() + .map(|term| term.trim_start_matches('.').to_lowercase()) + .collect() } /// Flatten per-peer results into individual files, dropping what `filter` @@ -112,6 +158,19 @@ pub struct Request { pub size: u64, } +impl From<&FilterArgs> for Filter { + fn from(args: &FilterArgs) -> Self { + Self { + min_bitrate: args.min_bitrate, + free_slots_only: args.free_slots, + exclude: normalized(&args.exclude), + extensions: normalized(&args.extension), + min_size: args.min_size, + max_size: args.max_size, + } + } +} + impl From<&SearchRecord> for Request { fn from(candidate: &SearchRecord) -> Self { Self { @@ -123,7 +182,7 @@ impl From<&SearchRecord> for Request { } /// Filter, rank, and cap a raw result set, failing when nothing survives. -fn ranked( +pub(super) fn ranked( results: &[SearchResult], filter: &Filter, sort: SortKey, @@ -146,10 +205,7 @@ pub fn search(ctx: &Ctx, args: &SearchArgs) -> CliResult { let results = collect(ctx, &session.client, &args.query, false)?; let found = ranked( &results, - &Filter { - min_bitrate: args.min_bitrate, - free_slots_only: args.free_slots, - }, + &Filter::from(&args.filter), args.sort, args.limit, &args.query, @@ -206,10 +262,7 @@ pub fn get(ctx: &Ctx, args: &GetArgs) -> CliResult { }; let found = ranked( &results, - &Filter { - min_bitrate: args.min_bitrate, - free_slots_only: args.free_slots, - }, + &Filter::from(&args.filter), SortKey::Best, limit, &args.query, @@ -634,6 +687,10 @@ mod tests { } } + fn paths_of(found: &[SearchRecord]) -> Vec { + found.iter().map(|c| c.path.clone()).collect() + } + fn sample() -> Vec { vec![ result("slow", 1, 100, vec![file("a\\low.mp3", 10, Some(128))]), @@ -667,28 +724,164 @@ mod tests { assert_eq!(best.speed, 900); } + /// Build a filter the way a command does, so the tests cover the flag + /// normalisation (case, leading dots) rather than assuming it. + fn filter(build: impl FnOnce(&mut FilterArgs)) -> Filter { + let mut args = FilterArgs::default(); + build(&mut args); + Filter::from(&args) + } + + fn strings(values: &[&str]) -> Vec { + values.iter().map(|v| (*v).to_string()).collect() + } + #[test] fn min_bitrate_drops_lower_and_unknown_bitrates() { - let filter = Filter { - min_bitrate: Some(256), - free_slots_only: false, - }; - let found = candidates(&sample(), &filter); - let paths: Vec<&str> = found.iter().map(|c| c.path.as_str()).collect(); - assert_eq!(paths, ["b\\best.mp3", "c\\good.mp3"]); + let filter = filter(|a| a.min_bitrate = Some(256)); + assert_eq!( + paths_of(&candidates(&sample(), &filter)), + ["b\\best.mp3", "c\\good.mp3"] + ); } #[test] fn free_slots_only_keeps_peers_advertising_a_slot() { - let filter = Filter { - min_bitrate: None, - free_slots_only: true, - }; + let filter = filter(|a| a.free_slots = true); let found = candidates(&sample(), &filter); assert!(found.iter().all(|c| c.free_slot)); assert!(found.iter().all(|c| c.user != "busy")); } + #[test] + fn exclude_drops_a_path_containing_the_term_whatever_its_case() { + let filter = filter(|a| a.exclude = strings(&["LOW"])); + let found = candidates(&sample(), &filter); + assert!(found.iter().all(|c| c.path != "a\\low.mp3")); + assert_eq!(found.len(), 3); + } + + #[test] + fn exclude_matches_any_term_not_all_of_them() { + let filter = filter(|a| a.exclude = strings(&["low", "unknown"])); + assert_eq!( + paths_of(&candidates(&sample(), &filter)), + ["b\\best.mp3", "c\\good.mp3"] + ); + } + + #[test] + fn extension_keeps_only_that_type_however_it_is_written() { + let mixed = vec![result( + "peer", + 1, + 1, + vec![ + file("a.mp3", 1, None), + file("b.FLAC", 2, None), + file("c.ogg", 3, None), + ], + )]; + for spelling in [".flac", "flac", "FLAC", ".FLAC"] { + let filter = filter(|a| a.extension = strings(&[spelling])); + assert_eq!( + paths_of(&candidates(&mixed, &filter)), + ["b.FLAC"], + "--extension {spelling} should match b.FLAC" + ); + } + } + + #[test] + fn several_extensions_are_a_union() { + let mixed = vec![result( + "peer", + 1, + 1, + vec![ + file("a.mp3", 1, None), + file("b.flac", 2, None), + file("c.ogg", 3, None), + ], + )]; + let filter = filter(|a| a.extension = strings(&["mp3", "ogg"])); + assert_eq!(paths_of(&candidates(&mixed, &filter)), ["a.mp3", "c.ogg"]); + } + + #[test] + fn an_extensionless_file_never_satisfies_an_extension_filter() { + let odd = vec![result("peer", 1, 1, vec![file("README", 1, None)])]; + let filter = filter(|a| a.extension = strings(&["mp3"])); + assert!(candidates(&odd, &filter).is_empty()); + } + + #[test] + fn a_dot_in_a_folder_name_is_not_the_files_extension() { + // Only the last path segment decides the extension, or every file under + // "Album (2004).v0" would look like a .v0 file. + let nested = vec![result( + "peer", + 1, + 1, + vec![file("Album (2004).v0\\track.mp3", 1, None)], + )]; + assert!( + candidates(&nested, &filter(|a| a.extension = strings(&["v0"]))) + .is_empty(), + "the folder's .v0 suffix must not count as the file's extension" + ); + assert_eq!( + paths_of(&candidates( + &nested, + &filter(|a| a.extension = strings(&["mp3"])) + )), + ["Album (2004).v0\\track.mp3"] + ); + } + + #[test] + fn size_bounds_are_inclusive_on_both_ends() { + let filter = filter(|a| { + a.min_size = Some(50); + a.max_size = Some(70); + }); + assert_eq!( + paths_of(&candidates(&sample(), &filter)), + ["c\\good.mp3", "c\\unknown.mp3"] + ); + } + + #[test] + fn each_size_bound_works_without_the_other() { + let floor = filter(|a| a.min_size = Some(70)); + assert_eq!( + paths_of(&candidates(&sample(), &floor)), + ["b\\best.mp3", "c\\unknown.mp3"] + ); + let ceiling = filter(|a| a.max_size = Some(10)); + assert_eq!(paths_of(&candidates(&sample(), &ceiling)), ["a\\low.mp3"]); + } + + #[test] + fn filters_combine_as_an_and() { + let filter = filter(|a| { + a.free_slots = true; + a.min_bitrate = Some(200); + a.min_size = Some(20); + a.exclude = strings(&["bad"]); + a.extension = strings(&["mp3"]); + }); + assert_eq!(paths_of(&candidates(&sample(), &filter)), ["c\\good.mp3"]); + } + + #[test] + fn a_filter_that_rejects_everything_is_the_no_results_verdict() { + let filter = filter(|a| a.min_size = Some(u64::MAX)); + let error = ranked(&sample(), &filter, SortKey::Best, 0, "q") + .expect_err("nothing survives the filter"); + assert_eq!(error.exit, Exit::NoResults); + } + #[test] fn best_prefers_a_free_slot_over_a_higher_bitrate() { let mut found = candidates(&sample(), &Filter::default()); diff --git a/soulseek-rs/src/commands/wish.rs b/soulseek-rs/src/commands/wish.rs new file mode 100644 index 0000000..a1a7eaf --- /dev/null +++ b/soulseek-rs/src/commands/wish.rs @@ -0,0 +1,257 @@ +//! Standing searches: `wish`. +//! +//! A wishlist is what makes a client keep looking for something it did not find +//! today. The server treats it as a distinct message (code 103) and rate-limits +//! it to the interval it announces at login, so the schedule is the server's to +//! set — we only decide which queries to send and what to print. + +use super::{Ctx, Session, settings::Store, transfer}; +use crate::cli::{SortKey, WishRunArgs}; +use crate::output::{CliError, CliResult, Out, WishRecord}; +use crate::persist::config::FileConfig; +use soulseek_rs::Client; +use std::time::Instant; + +pub fn add(out: &Out, store: &Store, query: &str) -> CliResult { + let mut updated = store.config.clone(); + if !updated.add_wish(query) { + if query.trim().is_empty() { + return Err(CliError::usage("a wish needs a query")); + } + out.status(&format!("already wishing for '{}'", query.trim())); + return Ok(()); + } + store.save(&updated)?; + out.emit(&WishRecord { + query: query.trim().to_string(), + }); + Ok(()) +} + +pub fn remove(out: &Out, store: &Store, query: &str) -> CliResult { + let mut updated = store.config.clone(); + if !updated.remove_wish(query) { + return Err(CliError::no_results(format!( + "'{}' is not on the wishlist", + query.trim() + ))); + } + store.save(&updated)?; + out.status(&format!("no longer wishing for '{}'", query.trim())); + Ok(()) +} + +pub fn list(out: &Out, store: &Store) -> CliResult { + let wishes = store.config.wishes(); + if wishes.is_empty() { + return Err(CliError::no_results("the wishlist is empty")); + } + for query in wishes { + out.emit(&WishRecord { query }); + } + Ok(()) +} + +pub fn run(ctx: &Ctx, store: &Store, args: &WishRunArgs) -> CliResult { + let wishes = selected(&store.config, args.only.as_deref())?; + let session = Session::open(ctx)?; + + if !sweep(ctx, &session.client, &wishes, &Sweep::from(args)) { + return Err(CliError::no_results( + "no wish matched anything on the network", + )); + } + Ok(()) +} + +/// How to narrow and order what a sweep prints — the parts of `wish run`'s flags +/// that outlive the command, so `serve --wishlist` can sweep without inventing +/// a set of command-line arguments it never received. +struct Sweep { + filter: transfer::Filter, + sort: SortKey, + limit: usize, +} + +impl Default for Sweep { + fn default() -> Self { + Self { + filter: transfer::Filter::default(), + sort: SortKey::Best, + limit: 0, + } + } +} + +impl From<&WishRunArgs> for Sweep { + fn from(args: &WishRunArgs) -> Self { + Self { + filter: transfer::Filter::from(&args.filter), + sort: args.sort, + limit: args.limit, + } + } +} + +/// Re-run the wishlist for as long as `serve` stays online, on the interval the +/// server set. Called from the serve loop rather than owning it, so uploads keep +/// being reported between sweeps. +pub struct Sweeper { + wishes: Vec, + next: Instant, +} + +impl Sweeper { + /// `None` when there is nothing to look for, so `serve` can say so once + /// instead of waking up forever with an empty list. + #[must_use] + pub fn new(wishes: Vec) -> Option { + (!wishes.is_empty()).then(|| Self { + wishes, + // The first sweep is immediate: a wishlist that waits twelve + // minutes before its first look is indistinguishable from broken. + next: Instant::now(), + }) + } + + /// Search every wish if the interval has elapsed. Does nothing otherwise, + /// so the caller can poll it as often as it likes. + /// + /// this blocks the serve loop for one search window, so uploads go + /// unreported and `--duration` overshoots by that much. Move the sweep to its + /// own thread with a channel of records if that ever matters; what mattered + /// was that it used to be one window *per wish*. + pub fn tick(&mut self, ctx: &Ctx, client: &Client) { + if Instant::now() < self.next { + return; + } + sweep(ctx, client, &self.wishes, &Sweep::default()); + let interval = client.wishlist_interval(); + self.next = Instant::now() + interval; + ctx.out + .status(&format!("next wishlist sweep in {}s", interval.as_secs())); + } +} + +/// Search every wish and print what survived the filters. Returns whether +/// anything did. +/// +/// All the searches go out before anything is waited for: peers answer whenever +/// they get round to it, so one shared window covers every wish, where waiting +/// per wish would cost a full window each. +fn sweep(ctx: &Ctx, client: &Client, wishes: &[String], how: &Sweep) -> bool { + let mut sent = Vec::new(); + for wish in wishes { + match client.start_wishlist_search(wish) { + Ok(()) => sent.push(wish), + Err(e) => ctx + .out + .warn(&format!("'{wish}' could not be searched: {e}")), + } + } + if sent.is_empty() { + return false; + } + + ctx.out.status(&format!( + "wishing for {} ({}s)…", + sent.iter() + .map(|w| format!("'{w}'")) + .collect::>() + .join(", "), + ctx.search_timeout.as_secs() + )); + Client::collect_for(ctx.search_timeout, None); + + let mut found_any = false; + for wish in sent { + let results = client.get_search_results(wish); + let ranked = + transfer::ranked(&results, &how.filter, how.sort, how.limit, wish); + let Ok(found) = ranked else { + ctx.out.status(&format!("nothing for '{wish}' this time")); + continue; + }; + found_any = true; + for candidate in &found { + ctx.out.emit(candidate); + } + } + found_any +} + +/// Which wishes a run covers. `--only` has to name a stored wish: letting a +/// typo through would quietly search for something the user never asked to +/// keep, and report it as a wishlist hit. +/// +/// Matching goes through [`FileConfig::wish`], so a query that `wish add` would +/// call a duplicate is one `--only` can select. +fn selected(config: &FileConfig, only: Option<&str>) -> CliResult> { + let Some(only) = only else { + let wishes = config.wishes(); + if wishes.is_empty() { + return Err(CliError::no_results( + "the wishlist is empty — add one with `wish add QUERY`", + )); + } + return Ok(wishes); + }; + + config + .wish(only) + .map(|wish| vec![wish.clone()]) + .ok_or_else(|| { + CliError::usage(format!( + "'{}' is not on the wishlist; `wish list` shows what is", + only.trim() + )) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::output::Exit; + + fn stored() -> FileConfig { + let mut config = FileConfig::default(); + config.add_wish("Aphex Twin"); + config.add_wish("Boards of Canada"); + config + } + + #[test] + fn without_only_every_stored_wish_runs() { + assert_eq!(selected(&stored(), None).unwrap(), stored().wishes()); + } + + #[test] + fn an_empty_wishlist_is_the_no_results_verdict() { + let error = + selected(&FileConfig::default(), None).expect_err("nothing to run"); + assert_eq!(error.exit, Exit::NoResults); + } + + #[test] + fn only_selects_one_stored_wish_ignoring_case_and_space() { + assert_eq!( + selected(&stored(), Some(" aphex twin ")).unwrap(), + ["Aphex Twin"], + "the stored spelling is what gets searched" + ); + } + + #[test] + fn only_naming_something_unstored_is_a_usage_error() { + let error = selected(&stored(), Some("plaid")) + .expect_err("a typo must not become an ad-hoc search"); + assert_eq!(error.exit, Exit::Usage); + assert!(error.message.contains("wish list"), "{}", error.message); + } + + #[test] + fn a_sweeper_needs_something_to_look_for() { + assert!(Sweeper::new(Vec::new()).is_none()); + assert!(Sweeper::new(stored().wishes()).is_some()); + } +} diff --git a/soulseek-rs/src/main.rs b/soulseek-rs/src/main.rs index b7e44c6..4267d67 100644 --- a/soulseek-rs/src/main.rs +++ b/soulseek-rs/src/main.rs @@ -8,7 +8,10 @@ mod port_mapping; mod ui; use clap::Parser; -use cli::{Cli, Commands, ConfigCommand, SharesCommand, parse_server_address}; +use cli::{ + Cli, Commands, ConfigCommand, SharesCommand, WishCommand, + parse_server_address, +}; use commands::Ctx; use output::{CliError, CliResult, Exit, Out}; use soulseek_rs::{ClientSettings, ClientVersion, PeerAddress}; @@ -93,6 +96,15 @@ fn run(mut cli: Cli, out: &Out) -> CliResult { Commands::Shares(SharesCommand::Remove { ref directory }) => { return commands::settings::shares_remove(out, &store, directory); } + Commands::Wish(WishCommand::Add { ref query }) => { + return commands::wish::add(out, &store, query); + } + Commands::Wish(WishCommand::Remove { ref query }) => { + return commands::wish::remove(out, &store, query); + } + Commands::Wish(WishCommand::List) => { + return commands::wish::list(out, &store); + } _ => {} } @@ -104,6 +116,22 @@ fn run(mut cli: Cli, out: &Out) -> CliResult { Commands::Shares(SharesCommand::Reindex) => { commands::settings::shares_reindex(&ctx) } + Commands::Wish(WishCommand::Run(ref args)) => { + commands::wish::run(&ctx, &store, args) + } + Commands::Serve(ref args) => { + let sweeper = args + .wishlist + .then(|| commands::wish::Sweeper::new(store.config.wishes())) + .flatten(); + if args.wishlist && sweeper.is_none() { + out.warn( + "--wishlist has nothing to look for; add one with \ + `wish add QUERY`", + ); + } + commands::peer::serve(&ctx, args, sweeper) + } other => commands::run(&ctx, other), } } diff --git a/soulseek-rs/src/output.rs b/soulseek-rs/src/output.rs index e97c115..e12dd1b 100644 --- a/soulseek-rs/src/output.rs +++ b/soulseek-rs/src/output.rs @@ -319,6 +319,9 @@ pub struct WhoamiRecord { pub shared_folders: u32, pub shared_files: u32, pub download_dir: String, + /// Seconds of privileges left on this account, or `null` when the server + /// did not answer in time. Zero means an ordinary account. + pub privilege_seconds: Option, } impl Record for WhoamiRecord { @@ -405,6 +408,18 @@ impl Record for ShareStatusRecord { } } +/// One standing search on the wishlist. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct WishRecord { + pub query: String, +} + +impl Record for WishRecord { + fn text(&self) -> String { + sanitize(&self.query) + } +} + /// One configuration setting. #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct ConfigRecord { diff --git a/soulseek-rs/src/persist/config.rs b/soulseek-rs/src/persist/config.rs index 6bd2fe3..7f00d11 100644 --- a/soulseek-rs/src/persist/config.rs +++ b/soulseek-rs/src/persist/config.rs @@ -23,6 +23,10 @@ pub struct FileConfig { /// Command whose stdout is the password (headless fallback, like mutt's /// `password_cmd`). Never store the password itself in the file. pub password_cmd: Option, + /// Standing searches the server lets us repeat once per wishlist interval. + /// A TOML array rather than a `config set` key, because a query may itself + /// contain a comma. + pub wishlist: Option>, } impl FileConfig { @@ -131,6 +135,54 @@ impl FileConfig { Ok(()) } + /// The standing searches, in the order they were added. + #[must_use] + pub fn wishes(&self) -> Vec { + self.wishlist.clone().unwrap_or_default() + } + + /// Add `query` to the wishlist. Returns whether it changed anything: a + /// blank query and a case-insensitive repeat are both refused, so adding + /// the same wish twice cannot fill the file with duplicates the server + /// would then be asked about twice per interval. + pub fn add_wish(&mut self, query: &str) -> bool { + let query = query.trim(); + if query.is_empty() || self.find_wish(query).is_some() { + return false; + } + self.wishlist + .get_or_insert_with(Vec::new) + .push(query.to_string()); + true + } + + /// Remove `query`, matched the same way [`Self::add_wish`] compares. + /// Returns whether it was there. + pub fn remove_wish(&mut self, query: &str) -> bool { + let Some(index) = self.find_wish(query.trim()) else { + return false; + }; + self.wishlist.get_or_insert_with(Vec::new).remove(index); + true + } + + /// The stored wish matching `query`, compared the way [`Self::add_wish`] + /// compares. Callers that need to act on a named wish go through this so + /// "the same wish" means one thing everywhere. + #[must_use] + pub fn wish(&self, query: &str) -> Option<&String> { + let index = self.find_wish(query.trim())?; + self.wishlist.as_ref()?.get(index) + } + + fn find_wish(&self, query: &str) -> Option { + let wanted = query.to_lowercase(); + self.wishlist + .as_ref()? + .iter() + .position(|wish| wish.to_lowercase() == wanted) + } + /// Load from `path`; a missing file is an empty config, a malformed file /// is an error (silently ignoring a typo'd config would be worse). pub fn load(path: &Path) -> Result { @@ -430,6 +482,7 @@ mod tests { max_concurrent_downloads: Some(2), search_timeout: Some(30), password_cmd: Some("pass show slsk".into()), + wishlist: None, }; let resolved = resolve(&bare_cli(), &file); assert_eq!(resolved.username.as_deref(), Some("alice")); @@ -577,6 +630,56 @@ mod tests { assert_eq!(FileConfig::load(&path).unwrap(), config); } + #[test] + fn a_wish_is_added_once_however_it_is_spelled() { + let mut config = FileConfig::default(); + assert!(config.add_wish("Aphex Twin")); + assert!(!config.add_wish("aphex twin"), "a repeat is not a new wish"); + assert!(!config.add_wish(" Aphex Twin "), "surrounding space too"); + assert_eq!(config.wishes(), ["Aphex Twin"]); + } + + #[test] + fn an_empty_wish_is_refused() { + let mut config = FileConfig::default(); + assert!(!config.add_wish(" ")); + assert!(config.wishes().is_empty()); + } + + #[test] + fn removing_a_wish_ignores_case_and_reports_whether_it_was_there() { + let mut config = FileConfig::default(); + config.add_wish("Boards of Canada"); + assert!(config.remove_wish("BOARDS OF CANADA")); + assert!(config.wishes().is_empty()); + assert!(!config.remove_wish("never added")); + } + + #[test] + fn wishes_keep_the_order_they_were_added_in() { + let mut config = FileConfig::default(); + for wish in ["first", "second", "third"] { + config.add_wish(wish); + } + config.remove_wish("second"); + assert_eq!(config.wishes(), ["first", "third"]); + } + + #[test] + fn a_wishlist_survives_a_save_and_load() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("config.toml"); + let mut config = FileConfig::default(); + config.add_wish("autechre, incunabula"); + config.add_wish("plaid"); + config.save(&path).unwrap(); + assert_eq!( + FileConfig::load(&path).unwrap().wishes(), + ["autechre, incunabula", "plaid"], + "a comma inside a wish must survive the round trip" + ); + } + #[test] fn unknown_keys_and_partial_files_are_tolerated() { let dir = tempfile::tempdir().unwrap(); diff --git a/soulseek-rs/src/ui/panes/download_info_pane.rs b/soulseek-rs/src/ui/panes/download_info_pane.rs index e88da88..094dbd1 100644 --- a/soulseek-rs/src/ui/panes/download_info_pane.rs +++ b/soulseek-rs/src/ui/panes/download_info_pane.rs @@ -92,6 +92,9 @@ fn build_upload_info_lines(upload: &UploadInfo) -> Vec> { lines.push(label_value("Size", &format_bytes(upload.size))); let (status_text, status_style) = match &upload.status { + UploadStatus::Queued(place) => { + (format!("Queued (#{place})"), inactive_style()) + } UploadStatus::InProgress => { ("In progress".to_string(), warning_style()) } @@ -132,7 +135,9 @@ fn build_upload_info_lines(upload: &UploadInfo) -> Vec> { ))); lines.push(Line::from(Span::styled(reason.clone(), error_style()))); } - UploadStatus::Completed | UploadStatus::Cancelled => {} + UploadStatus::Queued(_) + | UploadStatus::Completed + | UploadStatus::Cancelled => {} } lines diff --git a/soulseek-rs/src/ui/panes/downloads_pane.rs b/soulseek-rs/src/ui/panes/downloads_pane.rs index 92394cc..a8eed1d 100644 --- a/soulseek-rs/src/ui/panes/downloads_pane.rs +++ b/soulseek-rs/src/ui/panes/downloads_pane.rs @@ -1,9 +1,9 @@ use crate::models::DownloadEntry; use crate::ui::{ - GLYPH_ACTIVE, GLYPH_DONE, GLYPH_FAILED, HIGHLIGHT_SYMBOL, accent_style, - body_style, dimmed_style, download_status_glyph, error_style, format_bytes, - format_speed, header_style, inactive_style, info_style, pane_block, - pane_title, primary_style, row_highlight_style, success_style, + GLYPH_ACTIVE, GLYPH_DONE, GLYPH_FAILED, GLYPH_QUEUED, HIGHLIGHT_SYMBOL, + accent_style, body_style, dimmed_style, download_status_glyph, error_style, + format_bytes, format_speed, header_style, inactive_style, info_style, + pane_block, pane_title, primary_style, row_highlight_style, success_style, warning_style, }; use ratatui::{ @@ -119,12 +119,14 @@ pub fn render_downloads_pane( rows.extend(uploads.iter().map(|upload| { let (status_icon, status_style) = match &upload.status { + UploadStatus::Queued(_) => (GLYPH_QUEUED, warning_style()), UploadStatus::InProgress => (GLYPH_ACTIVE, accent_style()), UploadStatus::Completed => (GLYPH_DONE, success_style()), UploadStatus::Cancelled => (GLYPH_FAILED, inactive_style()), UploadStatus::Failed(_) => (GLYPH_FAILED, error_style()), }; let progress_text = match &upload.status { + UploadStatus::Queued(place) => format!("Queued #{place}"), UploadStatus::InProgress => { let percent = if upload.size > 0 { (upload.bytes_sent as f64 / upload.size as f64 * 100.0) diff --git a/soulseek-rs/tests/cli_e2e.rs b/soulseek-rs/tests/cli_e2e.rs index a687c8e..76039cd 100644 --- a/soulseek-rs/tests/cli_e2e.rs +++ b/soulseek-rs/tests/cli_e2e.rs @@ -104,6 +104,7 @@ fn help_lists_every_scriptable_command() { "portmap", "skills", "completions", + "wish", ] { assert!(help.contains(command), "--help should mention {command}"); } @@ -387,6 +388,85 @@ fn installing_completions_is_repeatable_and_reversible() { assert_eq!(action(&again), "absent"); } +// --- wishlist, the half that only touches the config file ------------------ + +#[test] +fn a_wish_is_added_listed_and_removed_through_the_config_file() { + let dir = Scratch::new("wish"); + let config = dir.path().join("config.toml"); + + let added = with_config(&config, &["wish", "add", "aphex twin"]); + assert_eq!(code(&added), EXIT_OK, "stderr: {}", stderr(&added)); + assert_eq!(records(&added), ["aphex twin"]); + + let listed = with_config(&config, &["wish", "list"]); + assert_eq!(code(&listed), EXIT_OK); + assert_eq!(records(&listed), ["aphex twin"]); + + let removed = with_config(&config, &["wish", "remove", "APHEX TWIN"]); + assert_eq!(code(&removed), EXIT_OK, "stderr: {}", stderr(&removed)); + assert!(stdout(&removed).is_empty(), "remove prints nothing"); + + let empty = with_config(&config, &["wish", "list"]); + assert_eq!(code(&empty), EXIT_NO_RESULTS); + assert!(stdout(&empty).is_empty(), "stdout must stay data-only"); +} + +#[test] +fn adding_the_same_wish_twice_is_not_an_error_and_stores_it_once() { + let dir = Scratch::new("wish"); + let config = dir.path().join("config.toml"); + + with_config(&config, &["wish", "add", "boards of canada"]); + let again = with_config(&config, &["wish", "add", "Boards Of Canada"]); + assert_eq!(code(&again), EXIT_OK, "a repeat is not a failure"); + + let listed = with_config(&config, &["wish", "list"]); + assert_eq!(records(&listed), ["boards of canada"]); +} + +#[test] +fn a_wish_holding_a_comma_survives_the_config_file() { + let dir = Scratch::new("wish"); + let config = dir.path().join("config.toml"); + + with_config(&config, &["wish", "add", "autechre, incunabula"]); + let listed = with_config(&config, &["--json", "wish", "list"]); + let stored: Vec = records(&listed) + .iter() + .filter_map(|line| serde_json::from_str(line).ok()) + .collect(); + assert_eq!(stored.len(), 1, "one wish, not two split on the comma"); + assert_eq!(stored[0]["query"], "autechre, incunabula"); +} + +#[test] +fn removing_a_wish_that_is_not_there_is_the_no_results_code() { + let dir = Scratch::new("wish"); + let config = dir.path().join("config.toml"); + + let output = with_config(&config, &["wish", "remove", "never added"]); + assert_eq!(code(&output), EXIT_NO_RESULTS); + assert!(stdout(&output).is_empty()); +} + +#[test] +fn an_empty_wish_query_is_a_usage_error() { + let dir = Scratch::new("wish"); + let config = dir.path().join("config.toml"); + + let output = with_config(&config, &["wish", "add", " "]); + assert_eq!(code(&output), EXIT_USAGE); + assert!(stdout(&output).is_empty()); +} + +#[test] +fn wish_add_without_a_config_file_to_write_to_is_a_usage_error() { + let output = run(&["--no-config", "wish", "add", "nowhere to keep this"]); + assert_eq!(code(&output), EXIT_USAGE); + assert!(stdout(&output).is_empty()); +} + // --------------------------------------------------------------------------- // Server required (soulfind) // --------------------------------------------------------------------------- @@ -625,6 +705,30 @@ fn settle() { std::thread::sleep(Duration::from_secs(1)); } +/// Like [`cli`], but keeping a config file so a command that reads or writes +/// one (the wishlist) sees it. +/// +/// `--no-config` is dropped for the obvious reason, and `--quiet` with it: +/// these commands report the schedule the *server* chose on stderr, which is +/// the only place a test can see it. +fn cli_with_config( + server: &TestServer, + user: &str, + config: &Path, + args: &[&str], +) -> Output { + let mut all: Vec = server + .args(user) + .into_iter() + .filter(|arg| arg != "--no-config" && arg != "--quiet") + .collect(); + all.push("--config".to_string()); + all.push(config.to_str().expect("utf-8 path").to_string()); + all.extend(args.iter().map(|a| (*a).to_string())); + let refs: Vec<&str> = all.iter().map(String::as_str).collect(); + run(&refs) +} + #[test] fn a_wrong_password_is_a_connection_error() { let server = server_or_skip!(); @@ -726,6 +830,470 @@ fn a_search_with_no_matches_exits_with_the_no_results_code() { assert!(stdout(&output).is_empty(), "stdout must stay data-only"); } +/// One share holding three files that differ in name, extension and size, so +/// every result filter has something to keep and something to drop. +/// +/// `flt_beta_skipme.flac` is deliberately the small one and +/// `flt_gamma_keepme.ogg` the large one, so a size bound and an extension +/// bound select different files and a combination of the two is a real test +/// rather than two spellings of the same assertion. +struct FilterFixture { + _share: Scratch, +} + +impl FilterFixture { + const SMALL: usize = 128; + const MEDIUM: usize = 4096; + const LARGE: usize = 8192; + + fn new(server: &TestServer, sharer: &str) -> (Self, Client) { + let share = Scratch::new("filter-share"); + for (name, size) in [ + ("flt_alpha_keepme.mp3", Self::MEDIUM), + ("flt_beta_skipme.flac", Self::SMALL), + ("flt_gamma_keepme.ogg", Self::LARGE), + ] { + std::fs::write(share.path().join(name), vec![7u8; size]) + .expect("share file"); + } + let client = server.client(sharer, vec![share.display()]); + settle(); + (Self { _share: share }, client) + } +} + +/// Search for the fixture as a fresh user and return the matching file names, +/// sorted so the assertion does not depend on result order. +/// +/// A new username per call matters: soulfind answers a repeat of the same +/// search from the same user with nothing. +fn filtered_names( + server: &TestServer, + seeker: &str, + filter: &[&str], +) -> (i32, Vec) { + let mut args = vec!["--search-timeout", "5", "search", "flt"]; + args.extend_from_slice(filter); + let output = cli(server, seeker, &args); + let mut names: Vec = records(&output) + .iter() + .filter_map(|line| line.split('\t').nth(3)) + .filter_map(|path| path.rsplit(['/', '\\']).next()) + .filter(|name| name.starts_with("flt_")) + .map(String::from) + .collect(); + names.sort(); + (code(&output), names) +} + +#[test] +fn exclude_drops_the_results_whose_path_contains_the_term() { + let server = server_or_skip!(); + let (_fixture, _sharer) = FilterFixture::new(&server, "cli_e2e_flt_excl"); + + let (exit, names) = + filtered_names(&server, "cli_e2e_seek_excl", &["--exclude", "skipme"]); + assert_eq!(exit, EXIT_OK); + assert_eq!(names, ["flt_alpha_keepme.mp3", "flt_gamma_keepme.ogg"]); +} + +#[test] +fn extension_keeps_one_type_and_repeats_into_a_union() { + let server = server_or_skip!(); + let (_fixture, _sharer) = FilterFixture::new(&server, "cli_e2e_flt_ext"); + + let (exit, one) = filtered_names( + &server, + "cli_e2e_seek_ext_one", + &["--extension", "mp3"], + ); + assert_eq!(exit, EXIT_OK); + assert_eq!(one, ["flt_alpha_keepme.mp3"]); + + let (exit, several) = filtered_names( + &server, + "cli_e2e_seek_ext_many", + &["--extension", ".flac", "--extension", "OGG"], + ); + assert_eq!(exit, EXIT_OK); + assert_eq!(several, ["flt_beta_skipme.flac", "flt_gamma_keepme.ogg"]); +} + +#[test] +fn size_bounds_narrow_the_result_set_from_either_end() { + let server = server_or_skip!(); + let (_fixture, _sharer) = FilterFixture::new(&server, "cli_e2e_flt_size"); + + let (exit, big) = filtered_names( + &server, + "cli_e2e_seek_size_min", + &["--min-size", &FilterFixture::MEDIUM.to_string()], + ); + assert_eq!(exit, EXIT_OK); + assert_eq!(big, ["flt_alpha_keepme.mp3", "flt_gamma_keepme.ogg"]); + + let (exit, small) = filtered_names( + &server, + "cli_e2e_seek_size_max", + &["--max-size", &FilterFixture::SMALL.to_string()], + ); + assert_eq!(exit, EXIT_OK); + assert_eq!(small, ["flt_beta_skipme.flac"]); +} + +#[test] +fn filters_given_together_all_have_to_be_satisfied() { + let server = server_or_skip!(); + let (_fixture, _sharer) = FilterFixture::new(&server, "cli_e2e_flt_comb"); + + let (exit, names) = filtered_names( + &server, + "cli_e2e_seek_comb", + &[ + "--exclude", + "skipme", + "--min-size", + &FilterFixture::LARGE.to_string(), + "--extension", + "ogg", + ], + ); + assert_eq!(exit, EXIT_OK); + assert_eq!(names, ["flt_gamma_keepme.ogg"]); +} + +#[test] +fn a_filter_that_matches_nothing_is_the_no_results_code() { + let server = server_or_skip!(); + let (_fixture, _sharer) = FilterFixture::new(&server, "cli_e2e_flt_none"); + + let output = cli( + &server, + "cli_e2e_seek_none", + &[ + "--search-timeout", + "5", + "search", + "flt", + "--extension", + "aiff", + ], + ); + assert_eq!( + code(&output), + EXIT_NO_RESULTS, + "stderr: {}", + stderr(&output) + ); + assert!(stdout(&output).is_empty(), "stdout must stay data-only"); +} + +#[test] +fn get_honours_the_same_filters_search_does() { + let server = server_or_skip!(); + let (_fixture, _sharer) = FilterFixture::new(&server, "cli_e2e_flt_get"); + let target = Scratch::new("filter-downloads"); + + let output = cli( + &server, + "cli_e2e_seek_get", + &[ + "--search-timeout", + "5", + "--download-dir", + &target.display(), + "get", + "flt", + "--extension", + "ogg", + "--timeout", + "30", + ], + ); + assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output)); + assert!( + target.path().join("flt_gamma_keepme.ogg").exists(), + "the only .ogg should be what got fetched; stdout: {}", + stdout(&output) + ); +} + +#[test] +fn wish_run_finds_a_shared_file_through_a_wishlist_search() { + let server = server_or_skip!(); + let share = Scratch::new("wish-share"); + std::fs::write(share.path().join("wisher_probe_krel.mp3"), probe_bytes()) + .expect("share file"); + let _sharer = server.client("cli_e2e_wish_sharer", vec![share.display()]); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + settle(); + + let added = with_config(&config, &["wish", "add", "krel"]); + assert_eq!(code(&added), EXIT_OK, "stderr: {}", stderr(&added)); + + let output = cli_with_config( + &server, + "cli_e2e_wish_seeker", + &config, + &["--search-timeout", "5", "wish", "run"], + ); + assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output)); + + let record = records(&output) + .into_iter() + .find(|line| line.contains("wisher_probe_krel")) + .expect("the wishlist search should find the sharer's file"); + // The same four columns `search` prints, so this pipes into + // `download --stdin` with no reshaping. + let fields: Vec<&str> = record.split('\t').collect(); + assert_eq!(fields.len(), 4, "record was {record:?}"); + assert_eq!(fields[0], "cli_e2e_wish_sharer"); + assert_eq!(fields[1], probe_bytes().len().to_string()); +} + +#[test] +fn wish_run_output_feeds_download_stdin() { + let server = server_or_skip!(); + let share = Scratch::new("wish-share"); + std::fs::write(share.path().join("wisher_probe_vosk.mp3"), probe_bytes()) + .expect("share file"); + let _sharer = server.client("cli_e2e_wish_sharer_b", vec![share.display()]); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + let target = Scratch::new("wish-downloads"); + settle(); + + with_config(&config, &["wish", "add", "vosk"]); + let found = cli_with_config( + &server, + "cli_e2e_wish_seeker_b", + &config, + &["--search-timeout", "5", "--json", "wish", "run"], + ); + assert_eq!(code(&found), EXIT_OK, "stderr: {}", stderr(&found)); + + let mut child = command(&[ + "--no-config", + "--quiet", + "--server", + &server.address(), + "--username", + "cli_e2e_wish_puller", + "--password", + "pw", + "--listener-port", + &free_port().expect("listener port").to_string(), + "--download-dir", + &target.display(), + "download", + "--stdin", + "--timeout", + "30", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("download should start"); + child + .stdin + .take() + .expect("stdin") + .write_all(stdout(&found).as_bytes()) + .expect("feed the records back"); + let pulled = child.wait_with_output().expect("download should finish"); + + assert_eq!(code(&pulled), EXIT_OK, "stderr: {}", stderr(&pulled)); + assert!( + target.path().join("wisher_probe_vosk.mp3").exists(), + "the wished-for file should have landed; stdout: {}", + stdout(&pulled) + ); +} + +#[test] +fn wish_run_with_an_empty_wishlist_is_the_no_results_code() { + let server = server_or_skip!(); + let _peer = server.client("cli_e2e_wish_quiet", Vec::new()); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + + let output = cli_with_config( + &server, + "cli_e2e_wish_none", + &config, + &["wish", "run"], + ); + assert_eq!(code(&output), EXIT_NO_RESULTS); + assert!(stdout(&output).is_empty(), "stdout must stay data-only"); +} + +#[test] +fn wish_run_only_must_name_a_stored_wish() { + let server = server_or_skip!(); + let _peer = server.client("cli_e2e_wish_quiet_b", Vec::new()); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + with_config(&config, &["wish", "add", "stored one"]); + + let unknown = cli_with_config( + &server, + "cli_e2e_wish_only_bad", + &config, + &["wish", "run", "--only", "not stored"], + ); + assert_eq!(code(&unknown), EXIT_USAGE, "a typo is not an ad-hoc search"); + assert!(stdout(&unknown).is_empty()); +} + +#[test] +fn wish_run_only_narrows_the_sweep_to_that_wish() { + let server = server_or_skip!(); + let share = Scratch::new("wish-share"); + for name in ["wisher_probe_zant.mp3", "wisher_probe_hurl.mp3"] { + std::fs::write(share.path().join(name), probe_bytes()) + .expect("share file"); + } + let _sharer = server.client("cli_e2e_wish_sharer_c", vec![share.display()]); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + settle(); + + with_config(&config, &["wish", "add", "zant"]); + with_config(&config, &["wish", "add", "hurl"]); + + let output = cli_with_config( + &server, + "cli_e2e_wish_only_ok", + &config, + &["--search-timeout", "5", "wish", "run", "--only", "ZANT"], + ); + assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output)); + + let paths = stdout(&output); + assert!(paths.contains("wisher_probe_zant"), "stdout was {paths:?}"); + assert!( + !paths.contains("wisher_probe_hurl"), + "--only must not run the other wish; stdout was {paths:?}" + ); +} + +#[test] +fn wish_run_filters_apply_to_a_wishlist_search_too() { + let server = server_or_skip!(); + let share = Scratch::new("wish-share"); + std::fs::write(share.path().join("wisher_probe_gleb.mp3"), probe_bytes()) + .expect("share file"); + std::fs::write(share.path().join("wisher_probe_gleb.flac"), probe_bytes()) + .expect("share file"); + let _sharer = server.client("cli_e2e_wish_sharer_d", vec![share.display()]); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + settle(); + + with_config(&config, &["wish", "add", "gleb"]); + let output = cli_with_config( + &server, + "cli_e2e_wish_filtered", + &config, + &[ + "--search-timeout", + "5", + "wish", + "run", + "--extension", + "flac", + ], + ); + assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output)); + let paths = stdout(&output); + assert!(paths.contains("wisher_probe_gleb.flac"), "{paths:?}"); + assert!(!paths.contains("wisher_probe_gleb.mp3"), "{paths:?}"); +} + +#[test] +fn serve_with_the_wishlist_sweeps_on_the_interval_the_server_announced() { + let server = server_or_skip!(); + let their_share = Scratch::new("wish-their-share"); + std::fs::write( + their_share.path().join("wisher_probe_yalt.mp3"), + probe_bytes(), + ) + .expect("share file"); + let _sharer = + server.client("cli_e2e_wish_sharer_e", vec![their_share.display()]); + + // A serving client needs something of its own to share. + let our_share = Scratch::new("wish-our-share"); + std::fs::write(our_share.path().join("ours.bin"), probe_bytes()) + .expect("share file"); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + settle(); + + with_config(&config, &["wish", "add", "yalt"]); + let output = cli_with_config( + &server, + "cli_e2e_wish_server", + &config, + &[ + "--search-timeout", + "5", + "--shared-dir", + &our_share.display(), + "serve", + "--wishlist", + "--duration", + "12", + ], + ); + assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output)); + assert!( + stdout(&output).contains("wisher_probe_yalt"), + "the first sweep happens at startup; stdout: {}", + stdout(&output) + ); + // Soulfind announces 12 minutes for an unprivileged account, and the + // schedule is the server's to set — so that is what must be reported. + assert!( + stderr(&output).contains("next wishlist sweep in 720s"), + "the announced interval should drive the timer; stderr: {}", + stderr(&output) + ); +} + +#[test] +fn serve_wishlist_with_nothing_wished_for_still_serves() { + let server = server_or_skip!(); + let our_share = Scratch::new("wish-our-share"); + std::fs::write(our_share.path().join("ours.bin"), probe_bytes()) + .expect("share file"); + let dir = Scratch::new("wish-config"); + let config = dir.path().join("config.toml"); + + let output = cli_with_config( + &server, + "cli_e2e_wish_server_b", + &config, + &[ + "--shared-dir", + &our_share.display(), + "serve", + "--wishlist", + "--duration", + "2", + ], + ); + assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output)); + assert!( + stderr(&output).contains("nothing to look for"), + "an empty wishlist should be said once, not serve-and-be-silent; \ + stderr: {}", + stderr(&output) + ); +} + #[test] fn browse_lists_shares_as_paths_download_understands() { let server = server_or_skip!(); @@ -1155,6 +1723,122 @@ fn whoami_confirms_the_login_and_reports_what_we_offer() { assert!(record["server"].as_str().is_some_and(|s| s.contains(':'))); } +#[test] +fn whoami_reports_how_much_privilege_time_the_account_has() { + let server = server_or_skip!(); + let _owner = server.client("cli_e2e_priv_owner", Vec::new()); + + let output = cli(&server, "cli_e2e_priv_owner", &["--json", "whoami"]); + assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output)); + + let record: serde_json::Value = + serde_json::from_str(&records(&output)[0]).expect("valid JSON"); + // Zero, not null: the server answered, and the answer is "no privileges". + // Null would mean it never said, which is a different fact. + assert_eq!( + record["privilege_seconds"], 0, + "a plain account should read as zero seconds, got {record}" + ); +} + +#[test] +fn serve_reports_a_queued_upload_once_the_slots_are_full() { + let server = server_or_skip!(); + let share = Scratch::new("queue-share"); + for name in ["q_one.bin", "q_two.bin"] { + std::fs::write(share.path().join(name), probe_bytes()) + .expect("share file"); + } + let target_a = Scratch::new("queue-dl-a"); + let target_b = Scratch::new("queue-dl-b"); + + let serving = command(&[ + "--no-config", + "--server", + &server.address(), + "--username", + "cli_e2e_queue_server", + "--password", + "pw", + "--listener-port", + &free_port().expect("listener port").to_string(), + "--shared-dir", + &share.display(), + "--json", + "serve", + "--upload-slots", + "1", + "--duration", + "18", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("serve should start"); + settle(); + settle(); + + // Two peers pull at once against a single slot, so one of them has to wait. + let mut pulls = Vec::new(); + for (user, target) in [ + ("cli_e2e_queue_a", &target_a), + ("cli_e2e_queue_b", &target_b), + ] { + pulls.push( + command(&[ + "--no-config", + "--quiet", + "--server", + &server.address(), + "--username", + user, + "--password", + "pw", + "--listener-port", + &free_port().expect("listener port").to_string(), + "--download-dir", + &target.display(), + "--search-timeout", + "5", + "get", + "q_", + "--pick", + "all", + "--timeout", + "30", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("puller should start"), + ); + } + for mut pull in pulls { + let _ = pull.wait(); + } + + let uploaded = serving.wait_with_output().expect("serve should finish"); + let statuses: Vec = stdout(&uploaded) + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter_map(|value| { + value["status"] + .as_str() + .map(std::string::ToString::to_string) + }) + .collect(); + assert!( + statuses.iter().any(|status| status == "completed"), + "at least one upload should have gone through; stdout: {}", + stdout(&uploaded) + ); + assert!( + statuses.iter().any(|status| status == "queued"), + "with one slot and two peers pulling, `serve` should report a queued \ + upload; statuses were {statuses:?}" + ); +} + #[test] fn whoami_fails_with_the_connection_code_on_a_bad_password() { let server = server_or_skip!(); @@ -1658,6 +2342,11 @@ fn the_commands_advertised_as_credential_free_run_without_an_account() { vec!["shares", "add", &shared_path], vec!["shares", "list"], vec!["shares", "remove", &shared_path], + // The wishlist is stored, not searched, until `wish run` asks the + // network — so managing it must not demand an account either. + vec!["wish", "add", "something rare"], + vec!["wish", "list"], + vec!["wish", "remove", "something rare"], vec!["skills", "install", "--dir", &skills], vec!["skills", "list", "--dir", &skills], vec!["skills", "uninstall", "--dir", &skills],