diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index e3303b3..014a0df 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -44,7 +44,7 @@ impl Client { // Find the search with matching token for search in context.searches.values_mut() { if search.token == result_token { - search.results.push(search_result); + search.accept(search_result); break; } } diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index 01b16c1..ebcac80 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -156,4 +156,33 @@ impl Client { .map(|ctx| ctx.searches.clone()) .unwrap_or_default() } + + /// Every registered search and how many files it has collected, without + /// copying the result sets. + /// + /// This is the accessor to poll. [`Client::get_all_searches`] clones every + /// result of every search, which for a popular query is megabytes per + /// call — a caller that only wants to know what exists and whether + /// anything new arrived reads the counts instead. + #[must_use] + pub fn search_file_counts(&self) -> Vec<(String, usize)> { + self.context + .read_safe() + .map(|ctx| { + ctx.searches + .iter() + .map(|(query, search)| { + ( + query.clone(), + search + .results + .iter() + .map(|result| result.files.len()) + .sum(), + ) + }) + .collect() + }) + .unwrap_or_default() + } } diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index f95dad7..ae4d94c 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -174,6 +174,90 @@ fn build_search_response_matches_shares_and_echoes_token() { let _ = std::fs::remove_dir_all(dir); } +fn peer_files(count: usize) -> SearchResult { + SearchResult { + token: 1, + files: (0..count) + .map(|i| crate::types::File { + username: "bob".to_string(), + name: format!("song-{i}.mp3"), + size: 1, + attribs: std::collections::HashMap::new(), + }) + .collect(), + slots: 1, + speed: 0, + username: "bob".to_string(), + } +} + +#[test] +fn search_file_counts_cover_every_search_without_the_results() { + // The daemon polls these counts many times a second; they must come from + // a walk of the cache, never a copy of it. + let client = Client::new("u", "p"); + { + let mut context = client.context.write().unwrap(); + context.searches.insert( + "aphex twin".to_string(), + Search { + token: 1, + results: vec![peer_files(2), peer_files(1)], + }, + ); + context.searches.insert( + "nothing yet".to_string(), + Search { + token: 2, + results: Vec::new(), + }, + ); + } + + let mut counts = client.search_file_counts(); + counts.sort(); + assert_eq!( + counts, + [ + ("aphex twin".to_string(), 3), + ("nothing yet".to_string(), 0) + ] + ); +} + +#[test] +fn a_search_stops_collecting_once_it_has_enough_responses() { + // A popular query on the live network draws answers for minutes — + // nearly a million files and gigabytes of memory for one search. + // Surplus responders are dropped, not archived. + let mut search = Search { + token: 1, + results: Vec::new(), + }; + for _ in 0..(crate::types::MAX_SEARCH_RESPONSES + 50) { + search.accept(peer_files(1)); + } + assert_eq!(search.results.len(), crate::types::MAX_SEARCH_RESPONSES); +} + +#[test] +fn a_flood_of_files_fills_a_search_before_the_response_cap() { + // A handful of whales with huge matching collections must not add up + // to an unbounded set just because the responses are few. + let mut search = Search { + token: 1, + results: Vec::new(), + }; + for _ in 0..10 { + search.accept(peer_files(crate::types::MAX_SEARCH_FILES / 2)); + } + assert_eq!( + search.results.len(), + 2, + "two half-cap responses fill the search; the rest are dropped" + ); +} + #[test] fn test_client_removes_only_queued_downloads() { let client = Client::new("test-user", "test-password"); diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index 81b06aa..fafdf8e 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -59,6 +59,36 @@ pub struct Search { pub results: Vec, } +/// Peer responses a search keeps before it stops collecting. +/// +/// Peers answer a popular query for minutes after the window closes; left +/// unbounded, one live-network search grew to nearly a million files and +/// gigabytes of memory, and every consumer of the results paid for the +/// whole set. A result set is a shortlist to pick from, not an archive of +/// everyone who answered. +pub const MAX_SEARCH_RESPONSES: usize = 500; + +/// Files a search keeps across all responses, so a few peers with huge +/// matching collections cannot blow past the response cap. +pub const MAX_SEARCH_FILES: usize = 10_000; + +impl Search { + /// Store one peer's response, unless this search already holds enough. + /// + /// The last accepted response may carry the total past + /// [`MAX_SEARCH_FILES`]; responses are kept whole rather than truncated. + pub fn accept(&mut self, result: SearchResult) { + // ponytail: counts files linearly per call — bounded by the response + // cap; a running total would mean a new pub field on Search. + let files: usize = + self.results.iter().map(|result| result.files.len()).sum(); + if self.results.len() < MAX_SEARCH_RESPONSES && files < MAX_SEARCH_FILES + { + self.results.push(result); + } + } +} + impl SearchResult { pub fn new_from_message(message: &mut Message) -> Result { let pointer = message.get_pointer(); diff --git a/soulseek-rs/src/api.rs b/soulseek-rs/src/api.rs index f3cc854..4006a84 100644 --- a/soulseek-rs/src/api.rs +++ b/soulseek-rs/src/api.rs @@ -86,6 +86,8 @@ pub trait SessionApi: Send + Sync { /// Drop a search from the session, so every window loses it rather than /// only the one that dismissed it. fn forget_search(&self, query: &str) -> bool; + /// How many files a search has collected so far — a count cheap enough + /// to poll, never a copy of the results. fn get_search_results_count(&self, key: &str) -> usize; fn try_get_search_results(&self, key: &str) -> Option>; fn start_wishlist_search(&self, query: &str) -> Result<()>; @@ -198,11 +200,15 @@ impl SessionApi for Client { } fn all_searches(&self) -> Vec { - Self::get_all_searches(self) + // Counts, not copies: the daemon answers this for every attached + // client many times a second, and cloning every result set to sum + // its lengths held the context lock against the thread storing new + // results. + Self::search_file_counts(self) .into_iter() - .map(|(query, search)| SessionSearch { + .map(|(query, files)| SessionSearch { query, - files: search.results.iter().map(|r| r.files.len()).sum(), + files, // A local window started every search it can see, so it // already knows which are still running. started_secs_ago: None, @@ -211,7 +217,12 @@ impl SessionApi for Client { } fn get_search_results_count(&self, key: &str) -> usize { - Self::get_search_results_count(self, key) + // Files, agreeing with the remote session — the library's own + // accessor of this name counts per-peer responses instead. + Self::search_file_counts(self) + .into_iter() + .find(|(query, _)| query == key) + .map_or(0, |(_, files)| files) } fn try_get_search_results(&self, key: &str) -> Option> { diff --git a/soulseek-rs/src/remote/mod.rs b/soulseek-rs/src/remote/mod.rs index 970812b..e236d05 100644 --- a/soulseek-rs/src/remote/mod.rs +++ b/soulseek-rs/src/remote/mod.rs @@ -544,7 +544,12 @@ impl SessionApi for RemoteSession { } fn get_search_results_count(&self, key: &str) -> usize { - self.get_search_results(key).len() + // The summary's file count, not the full set: callers poll this to + // ask whether anything has arrived, and the set can be megabytes. + self.all_searches() + .into_iter() + .find(|search| search.query == key) + .map_or(0, |search| search.files) } fn try_get_search_results(&self, key: &str) -> Option> { diff --git a/soulseek-rs/src/ui/main_tui/mod.rs b/soulseek-rs/src/ui/main_tui/mod.rs index 478208e..aa4fb1a 100644 --- a/soulseek-rs/src/ui/main_tui/mod.rs +++ b/soulseek-rs/src/ui/main_tui/mod.rs @@ -274,6 +274,14 @@ mod tests { history: Vec, downloads: Vec, searches: std::sync::Mutex>, + /// What `try_get_search_results` hands back to the window. + results: std::sync::Mutex>, + /// How many times a window asked for the search list. + all_searches_calls: std::sync::atomic::AtomicUsize, + /// How many times a window pulled a full result set across. Against a + /// daemon each pull is the whole set over the socket, so the tests + /// count them. + result_fetches: std::sync::atomic::AtomicUsize, /// Whether this session is a daemon's. Local sessions do not sync. shared: bool, /// What `set_download_directory` was last asked for, if anything. @@ -328,6 +336,8 @@ mod tests { Vec::new() } fn all_searches(&self) -> Vec { + self.all_searches_calls + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); self.searches.lock().expect("not poisoned").clone() } fn forget_search(&self, query: &str) -> bool { @@ -343,7 +353,9 @@ mod tests { &self, _key: &str, ) -> Option> { - None + self.result_fetches + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + Some(self.results.lock().expect("not poisoned").clone()) } fn start_wishlist_search( &self, @@ -512,8 +524,12 @@ mod tests { } fn with_session(session: TalkativeSession) -> MainTui { + attach(Arc::new(session)) + } + + fn attach(session: Arc) -> MainTui { MainTui::new( - Arc::new(session), + session, "/tmp".to_string(), Duration::from_secs(1), None, @@ -689,6 +705,130 @@ mod tests { assert!(tui.state.searches.is_empty()); } + fn files_from_bob(names: &[&str]) -> soulseek_rs::SearchResult { + soulseek_rs::SearchResult { + token: 0, + files: names + .iter() + .map(|name| soulseek_rs::types::File { + username: "bob".to_string(), + name: (*name).to_string(), + size: 42, + attribs: std::collections::HashMap::new(), + }) + .collect(), + slots: 1, + speed: 0, + username: "bob".to_string(), + } + } + + #[test] + fn a_tick_asks_the_session_for_its_searches_once() { + // Against a daemon every call is a round trip, and the daemon answers + // it by walking its whole search cache — twice per frame was half the + // cost of attaching at all. + let session = Arc::new(shared(vec![search_of("q", 0, 9_999)])); + let mut tui = attach(session.clone()); + + tui.update_search_results(); + + assert_eq!( + session + .all_searches_calls + .load(std::sync::atomic::Ordering::Relaxed), + 1, + "one sync means one listing, not one per use of it" + ); + } + + /// An attached window that selected a two-file search and has already + /// ticked past its one necessary fetch. + fn window_that_pulled_two_files() -> (Arc, MainTui) { + let session = Arc::new(TalkativeSession { + results: std::sync::Mutex::new(vec![files_from_bob(&[ + "@@x\\a.mp3", + "@@x\\b.mp3", + ])]), + ..shared(vec![search_of("aphex twin", 2, 9_999)]) + }); + let mut tui = attach(session.clone()); + tui.state.selected_search_index = Some(0); + tui.update_search_results(); + tui.update_search_results(); + (session, tui) + } + + #[test] + fn a_selected_search_with_nothing_new_is_not_pulled_across_again() { + // The result set only accumulates, so an unchanged count means a + // fetch would hand back what this window already holds — against a + // daemon that is the entire result set over the socket, every frame, + // forever. + let (session, tui) = window_that_pulled_two_files(); + + assert_eq!(tui.state.searches[0].results.len(), 2); + assert_eq!( + session + .result_fetches + .load(std::sync::atomic::Ordering::Relaxed), + 1, + "the set crosses once; after that the unchanged count spares it" + ); + } + + #[test] + fn fresh_results_still_reach_a_window_that_skipped_a_fetch() { + let (session, mut tui) = window_that_pulled_two_files(); + + // A late responder lands in the session. + *session.searches.lock().expect("not poisoned") = + vec![search_of("aphex twin", 3, 9_999)]; + *session.results.lock().expect("not poisoned") = + vec![files_from_bob(&["@@x\\a.mp3", "@@x\\b.mp3", "@@x\\c.mp3"])]; + tui.update_search_results(); + + assert_eq!( + tui.state.searches[0].results.len(), + 3, + "a changed count must bring the new results across" + ); + } + + #[test] + fn a_local_window_fetches_results_without_counting_on_the_session() { + // Locally nothing maintains known_files, so the count gate must stay + // out of the way: a search whose count still reads zero has to get + // its results anyway. + let session = Arc::new(TalkativeSession { + results: std::sync::Mutex::new(vec![files_from_bob(&[ + "@@x\\a.mp3", + "@@x\\b.mp3", + ])]), + shared: false, + ..TalkativeSession::default() + }); + let mut tui = attach(session); + tui.state.searches.push(crate::models::SearchEntry { + query: "mine".to_string(), + status: SearchStatus::Active, + results: Vec::new(), + known_files: 0, + owned: true, + start_time: std::time::Instant::now(), + cancel_flag: Arc::new(std::sync::atomic::AtomicBool::new(false)), + }); + + tui.update_search_results(); + + assert_eq!( + tui.state.searches[0].results.len(), + 2, + "a zero count matching zero held rows must not read as \ + nothing-to-fetch on a local session" + ); + } + #[test] fn a_window_with_its_own_session_is_left_alone() { // Nothing to share, so nothing is synced — and in particular the diff --git a/soulseek-rs/src/ui/main_tui/search.rs b/soulseek-rs/src/ui/main_tui/search.rs index 31bcc06..ca53f74 100644 --- a/soulseek-rs/src/ui/main_tui/search.rs +++ b/soulseek-rs/src/ui/main_tui/search.rs @@ -231,23 +231,23 @@ impl MainTui { } /// Show the searches the session knows about, not just this window's. + /// Returns whether it synced, which is only the case against a daemon. /// /// Against a daemon the search cache is shared, so a window that opens /// later sees what the others have looked for, and a query run in one /// appears in the other. A search is still collecting if it went out less /// than one search window ago — the window belongs to whichever client /// started it, so the age is the only thing another window can judge by. - fn sync_searches_with_session(&mut self) { + fn sync_searches_with_session(&mut self) -> bool { // Only worth doing when the session is shared: a window with its own - // session ran every search in the list and already knows their state, - // so asking would cost a full copy of every result set per frame and - // tell us nothing. + // session ran every search in the list and already knows their state. if self.client.daemon_endpoint().is_none() { - return; + return false; } let timeout = self.search_timeout.as_secs(); - for known in self.client.all_searches() { + let known_searches = self.client.all_searches(); + for known in &known_searches { let running = known .started_secs_ago .is_some_and(|elapsed| elapsed < timeout); @@ -267,13 +267,15 @@ impl MainTui { } else { SearchStatus::Completed }; - existing.known_files = known.files; } + // The count is every entry's change detector: as long as it + // matches what this window holds, there is nothing to fetch. + existing.known_files = known.files; continue; } self.state.searches.push(SearchEntry { - query: known.query, + query: known.query.clone(), status: if running { SearchStatus::Active } else { @@ -291,36 +293,41 @@ impl MainTui { // Dismissed elsewhere means dismissed here. Ours stay: this window is // the authority on a search it started. - let live: Vec = self - .client - .all_searches() - .into_iter() - .map(|search| search.query) - .collect(); - self.state - .searches - .retain(|search| search.owned || live.contains(&search.query)); + self.state.searches.retain(|search| { + search.owned + || known_searches + .iter() + .any(|known| known.query == search.query) + }); + true } pub(super) fn update_search_results(&mut self) { - self.sync_searches_with_session(); + // Only a synced session reports counts, so only there can an + // unchanged count spare the fetch. Locally `known_files` is never + // maintained and skipping on it would starve every search. + let counted = self.sync_searches_with_session(); let timeout = self.search_timeout; let selected_search_index = self.state.selected_search_index; // Fetch all results in one go (single lock acquisition per query) // Use try_get_search_results to avoid blocking the UI thread - // Only what the user can see or what is still arriving. A shared - // cache can hold every search every window has ever run, and fetching - // all of them each frame is a round trip apiece against a daemon. - // The rest show the count the session reported. + // Only what the user can see or what is still arriving, and against + // a daemon only when the session's count says something actually + // changed: results accumulate, so an unchanged count means a fetch + // would carry the entire set across the socket to learn nothing. + // A re-searched query resets the session's set; only a refill that + // lands back on exactly the held total within one tick could leave + // stale rows, and that window was already showing them. let all_results: Vec<(usize, Vec<_>)> = self .state .searches .iter() .enumerate() .filter(|(idx, search)| { - search.status == SearchStatus::Active - || selected_search_index == Some(*idx) + (search.status == SearchStatus::Active + || selected_search_index == Some(*idx)) + && (!counted || search.known_files != search.results.len()) }) .map(|(idx, s)| (idx, s.query.clone())) .filter_map(|(idx, query)| { @@ -370,13 +377,16 @@ impl MainTui { self.state.results_filtered_indices = indices; } } + } + } - // Mark as completed after timeout - if search.status == SearchStatus::Active - && search.start_time.elapsed() > timeout - { - search.status = SearchStatus::Completed; - } + // Outside the fetch loop: a tick that had nothing to fetch must + // still finish a run-out search. + for search in &mut self.state.searches { + if search.status == SearchStatus::Active + && search.start_time.elapsed() > timeout + { + search.status = SearchStatus::Completed; } } }