Merge pull request #27 from michel/slow-tui-daemon

fix: keep an attached TUI fast after a search

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;
                             }
                         }

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()
+    }
 }

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");

soulseek-rs-lib/src/types.rs

@@ -59,6 +59,36 @@ pub struct Search {
     pub results: Vec<SearchResult>,
 }
 
+/// 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<Self> {
         let pointer = message.get_pointer();

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<Vec<SearchResult>>;
     fn start_wishlist_search(&self, query: &str) -> Result<()>;
@@ -198,11 +200,15 @@ impl SessionApi for Client {
     }
 
     fn all_searches(&self) -> Vec<SessionSearch> {
-        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<Vec<SearchResult>> {

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<Vec<SearchResult>> {

soulseek-rs/src/ui/main_tui/mod.rs

@@ -274,6 +274,14 @@ mod tests {
         history: Vec<ChatMessageDto>,
         downloads: Vec<soulseek_rs::types::Download>,
         searches: std::sync::Mutex<Vec<crate::api::SessionSearch>>,
+        /// What `try_get_search_results` hands back to the window.
+        results: std::sync::Mutex<Vec<soulseek_rs::SearchResult>>,
+        /// 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<crate::api::SessionSearch> {
+            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<Vec<soulseek_rs::SearchResult>> {
-            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<TalkativeSession>) -> 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<TalkativeSession>, 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

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<String> = 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,8 +377,12 @@ impl MainTui {
                         self.state.results_filtered_indices = indices;
                     }
                 }
+            }
+        }
 
-                // Mark as completed after timeout
+        // 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
             {
@@ -380,7 +391,6 @@ impl MainTui {
         }
     }
 }
-}
 
 /// Filter `items` by a case-insensitive substring match on filename or
 /// username. Returns the matching items alongside their indices in the original