diff --git a/README.md b/README.md index e014a39..7af9a3d 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,13 @@ soulseek-rs daemon status # who it is, what it shares, who is attached soulseek-rs daemon stop ``` +Every window is a view of the same session, not a session of its own. Open a +second TUI and it shows the searches and transfers already running, including +ones the first started; queue something in either and both see it. The status +bar names the daemon it is attached to, so it is obvious which you are looking +at. Closing a window leaves its downloads running, because they belong to the +daemon. + Everything that touches the network goes through it. `config`, `skills`, `completions` and `portmap` stay local, because they are about this machine rather than a session. `shares add`/`remove` do both: they write the config diff --git a/docs/openrpc.json b/docs/openrpc.json index 12d5037..0cd8934 100644 --- a/docs/openrpc.json +++ b/docs/openrpc.json @@ -787,6 +787,47 @@ ], "type": "object" }, + "SearchSummary": { + "description": "One search as the daemon sees it, for a client that did not run it.", + "properties": { + "files": { + "description": "Files found so far, across every peer that has answered.", + "format": "uint", + "minimum": 0, + "type": "integer" + }, + "query": { + "type": "string" + }, + "started_secs_ago": { + "description": "How long ago the query went out. A client compares this with its own\nsearch window to decide whether the search is still collecting.", + "format": "uint64", + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "query", + "files", + "started_secs_ago" + ], + "type": "object" + }, + "Searches": { + "description": "Every search this session has run, newest last. Shared by all clients, so\na second one attaching sees what the first has been looking for.", + "properties": { + "searches": { + "items": { + "$ref": "#/components/schemas/SearchSummary" + }, + "type": "array" + } + }, + "required": [ + "searches" + ], + "type": "object" + }, "Seconds": { "properties": { "seconds": { @@ -1285,6 +1326,38 @@ }, "summary": "Everything that has answered a search so far." }, + { + "name": "search.list", + "paramStructure": "by-name", + "params": [], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/Searches" + } + }, + "summary": "Every search this session has run, shared by all clients." + }, + { + "name": "search.forget", + "paramStructure": "by-name", + "params": [ + { + "name": "query", + "required": true, + "schema": { + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/Ack" + } + }, + "summary": "Drop a search and its results, for every client at once." + }, { "name": "search.wishlist", "paramStructure": "by-name", diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index cde3312..01b16c1 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -137,6 +137,18 @@ impl Client { }) } + /// Drop a search and everything it collected. + /// + /// Returns whether there was one to drop. A client that stays up for days + /// would otherwise hold every result set it has ever seen, and a caller + /// that has dismissed a search has no other way to say so. + #[must_use] + pub fn forget_search(&self, search_key: &str) -> bool { + self.context + .write_safe() + .is_ok_and(|mut ctx| ctx.searches.remove(search_key).is_some()) + } + #[must_use] pub fn get_all_searches(&self) -> HashMap { self.context diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index bb252c1..2058960 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -159,6 +159,11 @@ While it runs, every other command on this machine uses its session automatically, with no flags, no credentials, and no second login. `--bind ADDR` also accepts connections from other machines, which need the token. +State lives in the daemon, so every client reads the same thing: `search` sees +searches any client ran, the transfer queue is shared, and `message read` +starts from the conversation the daemon already collected rather than only +what arrives while it runs. + **`daemon status`** — one object: `user`, `server`, `version`, `protocol`, `listening`, `listen_port`, `shared_folders`, `shared_files`, `download_dir`, `session_loss`, `clients`, `uptime_secs`. Exit 3 when nothing is running. @@ -217,21 +222,42 @@ soulseek-rs user someuser --json `download --stdin` reads whole records, so filtering with `jq` and piping back is the normal way to be selective. -## Running several at once +## Several tracks at once -With a daemon running this needs no thought. They all share its one session: +Asked for a list of tracks (an album, a playlist, "these ten songs"), **give +each track its own subagent**, and run **at most ten at a time**. One subagent +searches for its track, judges the candidates, downloads the one it picked, and +reports back. + +**Report as you go, not only at the end.** Each subagent tells the main agent +when its search has found candidates and again when its download completes, so +progress is visible while the rest are still working. A batch that reports only +once everything is finished looks indistinguishable from a batch that is stuck. +One short line each is enough: what it searched for, what it picked and why, +where the file landed, or which of the exit codes it hit and what that means +for that track. + +Do this rather than writing a shell script that loops. Picking a file is a +judgement call every time: bitrate against file size, a peer with a free slot +against a faster one, the right pressing against a live bootleg with the same +name. A loop cannot make that call, so it either takes the first hit or needs +you to encode rules you would rather apply case by case. A subagent per track +also means one failure is one track, not a dead pipeline, and each comes back +with a sentence you can pass on. + +Ten at once, not more. Every peer you queue behind is someone's home machine, +and the searches all share the daemon's one session. ```bash -for q in 'gary beck fold' 'marcal steady'; do - soulseek-rs search "$q" --json & -done -wait +# What each subagent runs for its own track: +soulseek-rs search 'gary beck fold' --json # look +soulseek-rs get 'gary beck fold' --json # or search, pick and fetch in one ``` -Without one, the same loop is a bug: the runs share a username, so each login -displaces the last and they kill each other's session (exit 7). Giving each run -its own `--username` avoids the collision by registering a fresh account on the -server for every search, which is worse. Do not do it. Start the daemon. +This works because the daemon holds the session: ten subagents at once are ten +commands sharing one login. Without a daemon they would displace each +other (exit 7), and giving each its own `--username` would register a throwaway +account per track, which is worse. Start the daemon first. The listener port needs no care either way: a run that finds the configured port taken binds a free one and tells the server about that one instead. diff --git a/soulseek-rs/src/api.rs b/soulseek-rs/src/api.rs index d54330f..b660b62 100644 --- a/soulseek-rs/src/api.rs +++ b/soulseek-rs/src/api.rs @@ -23,6 +23,23 @@ use std::sync::atomic::AtomicBool; use std::sync::mpsc::Receiver; use std::time::Duration; +/// What a session knows about one search it has run. +/// +/// `started_secs_ago` is what lets a second window tell a search that is still +/// collecting from one that has finished: the collecting window is the +/// client's, so without the daemon saying when a search began, every window +/// but the one that started it would show the result as final immediately. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SessionSearch { + pub query: String, + /// Files found so far, across every peer that has answered. + pub files: usize, + /// How long ago the search went out, or `None` when the session cannot + /// say — which is the local case, where the window that ran it already + /// knows. + pub started_secs_ago: Option, +} + /// Everything the command layer and the TUI ask of a logged-in session. /// /// Two rules keep the remote implementation honest: @@ -36,6 +53,11 @@ use std::time::Duration; /// local mirror. pub trait SessionApi: Send + Sync { fn username(&self) -> String; + /// The daemon this session is borrowed from, for the UI to name. `None` + /// when the session belongs to this process. + fn daemon_endpoint(&self) -> Option { + None + } fn listen_port(&self) -> Option; fn session_loss(&self) -> Option; @@ -56,6 +78,14 @@ pub trait SessionApi: Send + Sync { cancel: Option>, ) -> Result>; fn get_search_results(&self, key: &str) -> Vec; + /// Every search the session knows about. + /// + /// Daemon-side this is shared, so a second client sees what the first has + /// been searching for rather than starting from a blank list. + fn all_searches(&self) -> Vec; + /// 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; 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<()>; @@ -152,6 +182,23 @@ impl SessionApi for Client { Self::get_search_results(self, key) } + fn forget_search(&self, query: &str) -> bool { + Self::forget_search(self, query) + } + + fn all_searches(&self) -> Vec { + Self::get_all_searches(self) + .into_iter() + .map(|(query, search)| SessionSearch { + query, + files: search.results.iter().map(|r| r.files.len()).sum(), + // A local window started every search it can see, so it + // already knows which are still running. + started_secs_ago: None, + }) + .collect() + } + fn get_search_results_count(&self, key: &str) -> usize { Self::get_search_results_count(self, key) } diff --git a/soulseek-rs/src/commands/social.rs b/soulseek-rs/src/commands/social.rs index 168c92d..c348866 100644 --- a/soulseek-rs/src/commands/social.rs +++ b/soulseek-rs/src/commands/social.rs @@ -261,6 +261,22 @@ pub fn message_send(ctx: &Ctx, user: &str, message: &str) -> CliResult { pub fn message_read(ctx: &Ctx, span: Option) -> CliResult { let session = Session::open(ctx)?; + // What the session already holds comes first. A daemon has been collecting + // the whole time it has been up, including while nothing was reading, so + // starting from the next arrival would silently skip everything said so + // far. A session of our own knows nothing yet, so this is empty and the + // command behaves as it always did. + for message in session.client.message_history() { + if message.outgoing { + continue; + } + ctx.out.emit(&PrivateMessageRecord { + timestamp: u32::try_from(message.at).unwrap_or(0), + user: message.peer, + message: message.text, + }); + } + stream(&ctx.out, "reading private messages", span, || { for message in session.client.take_private_messages() { ctx.out.emit(&PrivateMessageRecord { diff --git a/soulseek-rs/src/daemon/mod.rs b/soulseek-rs/src/daemon/mod.rs index 6937f4c..7287ef1 100644 --- a/soulseek-rs/src/daemon/mod.rs +++ b/soulseek-rs/src/daemon/mod.rs @@ -228,6 +228,7 @@ pub fn run( stop: Arc::clone(&stop), open: Arc::new(std::sync::atomic::AtomicUsize::new(0)), rooms: Mutex::new(rooms), + searches: Mutex::new(std::collections::HashMap::new()), }); let drainer = { diff --git a/soulseek-rs/src/daemon/openrpc.rs b/soulseek-rs/src/daemon/openrpc.rs index b60c5db..80f05ae 100644 --- a/soulseek-rs/src/daemon/openrpc.rs +++ b/soulseek-rs/src/daemon/openrpc.rs @@ -12,9 +12,9 @@ use super::proto::{ Ack, AuthParams, AuthResult, BrowseEvent, DaemonStatus, DirectoriesParams, DownloadStartParams, DownloadStarted, DownloadStatusEvent, Downloads, Event, IntervalSeconds, Members, MessageParams, Messages, Method, - QueryParams, RoomEventDto, RoomRef, SayParams, SearchResults, Seconds, - SessionLossEvent, SharesStatus, SlotsParams, TransferRef, UploadInfoDto, - Uploads, UserMessageDto, UserRef, UserResult, + QueryParams, RoomEventDto, RoomRef, SayParams, SearchResults, Searches, + Seconds, SessionLossEvent, SharesStatus, SlotsParams, TransferRef, + UploadInfoDto, Uploads, UserMessageDto, UserRef, UserResult, }; use schemars::{JsonSchema, SchemaGenerator, generate::SchemaSettings}; use serde_json::{Value, json}; @@ -41,7 +41,8 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value { Method::Auth => schema_of::(generator), Method::SearchStart | Method::SearchResults - | Method::SearchWishlist => schema_of::(generator), + | Method::SearchWishlist + | Method::SearchForget => schema_of::(generator), Method::DownloadStart => schema_of::(generator), Method::DownloadPause | Method::DownloadResume @@ -68,6 +69,7 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value { | Method::PrivilegesOwn | Method::RoomListRequest | Method::MessageHistory + | Method::SearchList | Method::SharesStatusOf | Method::SharesReindex => nothing(), } @@ -80,6 +82,7 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value { Method::DaemonStatus => schema_of::(generator), Method::RpcDiscover => json!({ "type": "object" }), Method::SearchResults => schema_of::(generator), + Method::SearchList => schema_of::(generator), Method::SearchWishlistInterval => { schema_of::(generator) } @@ -110,6 +113,7 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value { | Method::RoomSay | Method::MessageSend | Method::BrowseUser + | Method::SearchForget | Method::UserRequest => schema_of::(generator), } } @@ -131,6 +135,12 @@ fn summary(method: Method) -> &'static str { Method::SearchResults => { "Everything that has answered a search so far." } + Method::SearchList => { + "Every search this session has run, shared by all clients." + } + Method::SearchForget => { + "Drop a search and its results, for every client at once." + } Method::SearchWishlist => { "Send a query as a wishlist search (server code 103)." } diff --git a/soulseek-rs/src/daemon/proto/method.rs b/soulseek-rs/src/daemon/proto/method.rs index 860f917..56a708e 100644 --- a/soulseek-rs/src/daemon/proto/method.rs +++ b/soulseek-rs/src/daemon/proto/method.rs @@ -81,6 +81,26 @@ pub struct DirectoriesParams { pub directories: Vec, } +/// Every search this session has run, newest last. Shared by all clients, so +/// a second one attaching sees what the first has been looking for. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] +pub struct Searches { + pub searches: Vec, +} + +/// One search as the daemon sees it, for a client that did not run it. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] +pub struct SearchSummary { + pub query: String, + /// Files found so far, across every peer that has answered. + pub files: usize, + /// How long ago the query went out. A client compares this with its own + /// search window to decide whether the search is still collecting. + pub started_secs_ago: u64, +} + #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] #[cfg_attr(test, derive(schemars::JsonSchema))] pub struct SearchResults { diff --git a/soulseek-rs/src/daemon/proto/mod.rs b/soulseek-rs/src/daemon/proto/mod.rs index bbe76e8..bd464db 100644 --- a/soulseek-rs/src/daemon/proto/mod.rs +++ b/soulseek-rs/src/daemon/proto/mod.rs @@ -80,6 +80,8 @@ wire_vocabulary! { RpcDiscover => "rpc.discover", SearchStart => "search.start", SearchResults => "search.results", + SearchList => "search.list", + SearchForget => "search.forget", SearchWishlist => "search.wishlist", SearchWishlistInterval => "search.wishlist_interval", DownloadStart => "download.start", diff --git a/soulseek-rs/src/daemon/server/mod.rs b/soulseek-rs/src/daemon/server/mod.rs index 18e11e3..19674e3 100644 --- a/soulseek-rs/src/daemon/server/mod.rs +++ b/soulseek-rs/src/daemon/server/mod.rs @@ -11,9 +11,9 @@ use super::proto::{ DaemonStatus, DirectoriesParams, DownloadDto, DownloadStartParams, DownloadStarted, Downloads, IntervalSeconds, Members, MessageParams, Messages, Method, OPENRPC, PROTOCOL_VERSION, QueryParams, RoomRef, - RpcError, SayParams, SearchResultDto, SearchResults, Seconds, SharesStatus, - SlotsParams, TransferRef, UploadInfoDto, Uploads, UserInfoDto, UserRef, - UserResult, + RpcError, SayParams, SearchResultDto, SearchResults, SearchSummary, + Searches, Seconds, SharesStatus, SlotsParams, TransferRef, UploadInfoDto, + Uploads, UserInfoDto, UserRef, UserResult, }; use crate::api::SessionApi; use crate::output::Exit; @@ -41,6 +41,10 @@ pub struct Daemon { pub open: Arc, /// Rooms this session has joined, so a restart can rejoin them. pub rooms: Mutex>, + /// When each search went out. The window a search collects for belongs to + /// the client, so a second client can only tell a running search from a + /// finished one if the daemon remembers when it started. + pub searches: Mutex>, } impl Daemon { @@ -84,6 +88,7 @@ impl Daemon { format!("search failed: {e}"), ) })?; + self.record_search_start(&query.query); ok(Ack::OK) } Method::SearchResults => { @@ -97,6 +102,25 @@ impl Daemon { .collect(), }) } + Method::SearchList => ok(Searches { + searches: self + .session + .all_searches() + .into_iter() + .map(|search| SearchSummary { + files: search.files, + started_secs_ago: self.search_age(&search.query), + query: search.query, + }) + .collect(), + }), + Method::SearchForget => { + let query: QueryParams = parse(params)?; + self.forget_search_start(&query.query); + ok(Ack { + ok: self.session.forget_search(&query.query), + }) + } Method::SearchWishlist => { let query: QueryParams = parse(params)?; self.session.start_wishlist_search(&query.query).map_err( @@ -361,6 +385,30 @@ impl Daemon { } } + /// How long ago a search went out, or a long time when we never saw it + /// start — a search restored or begun before this daemon did. + fn search_age(&self, query: &str) -> u64 { + self.searches + .lock() + .ok() + .and_then(|searches| { + searches.get(query).map(|at| at.elapsed().as_secs()) + }) + .unwrap_or(u64::MAX) + } + + fn record_search_start(&self, query: &str) { + if let Ok(mut searches) = self.searches.lock() { + searches.insert(query.to_string(), Instant::now()); + } + } + + fn forget_search_start(&self, query: &str) { + if let Ok(mut searches) = self.searches.lock() { + searches.remove(query); + } + } + fn identify(&self) -> AuthResult { AuthResult { protocol: PROTOCOL_VERSION, @@ -536,6 +584,7 @@ mod tests { stop: Arc::new(AtomicBool::new(false)), open: Arc::new(std::sync::atomic::AtomicUsize::new(0)), rooms: Mutex::new(Vec::new()), + searches: Mutex::new(std::collections::HashMap::new()), } } @@ -570,6 +619,12 @@ mod tests { ) -> Vec { Vec::new() } + fn all_searches(&self) -> Vec { + Vec::new() + } + fn forget_search(&self, _query: &str) -> bool { + false + } fn get_search_results_count(&self, _key: &str) -> usize { 0 } diff --git a/soulseek-rs/src/models/app_state.rs b/soulseek-rs/src/models/app_state.rs index 1879c7c..0256b47 100644 --- a/soulseek-rs/src/models/app_state.rs +++ b/soulseek-rs/src/models/app_state.rs @@ -16,6 +16,13 @@ pub struct SearchEntry { pub query: String, pub status: SearchStatus, pub results: Vec, + /// What the session says this search has found, for a window that has not + /// fetched the results itself. Fetching every search every frame is what + /// a shared cache would otherwise cost. + pub known_files: usize, + /// Whether this window started the search. Its own worker knows when it + /// finishes; for the rest the session is the authority. + pub owned: bool, pub start_time: Instant, #[allow(dead_code)] pub cancel_flag: Arc, diff --git a/soulseek-rs/src/persist/snapshot.rs b/soulseek-rs/src/persist/snapshot.rs index 69bf92e..b3b4452 100644 --- a/soulseek-rs/src/persist/snapshot.rs +++ b/soulseek-rs/src/persist/snapshot.rs @@ -107,6 +107,8 @@ pub fn restore_searches(state: &mut AppState, queries: &[String]) { query: query.clone(), status: SearchStatus::Completed, results: Vec::new(), + known_files: 0, + owned: false, start_time: std::time::Instant::now(), cancel_flag: std::sync::Arc::new( std::sync::atomic::AtomicBool::new(false), @@ -147,6 +149,8 @@ mod tests { query: query.into(), status: SearchStatus::Active, results: Vec::new(), + known_files: 0, + owned: false, start_time: Instant::now(), cancel_flag: Arc::new(AtomicBool::new(false)), } diff --git a/soulseek-rs/src/remote/mod.rs b/soulseek-rs/src/remote/mod.rs index 645ae33..656f517 100644 --- a/soulseek-rs/src/remote/mod.rs +++ b/soulseek-rs/src/remote/mod.rs @@ -7,14 +7,14 @@ mod mirror; -use crate::api::SessionApi; +use crate::api::{SessionApi, SessionSearch}; use crate::daemon::proto::{ Ack, AuthParams, AuthResult, CODE_APPLICATION, ChatMessageDto, DaemonStatus, DirectoriesParams, DownloadDto, DownloadStartParams, DownloadStarted, Downloads, Event, IntervalSeconds, Members, MessageParams, Messages, Method, PROTOCOL_VERSION, QueryParams, Request, Response, - RoomRef, RpcError, SayParams, SearchResults, Seconds, SharesStatus, - SlotsParams, TransferRef, Uploads, UserRef, UserResult, + RoomRef, RpcError, SayParams, SearchResults, Searches, Seconds, + SharesStatus, SlotsParams, TransferRef, Uploads, UserRef, UserResult, }; use mirror::Mirror; use serde::de::DeserializeOwned; @@ -147,6 +147,8 @@ pub struct RemoteSession { mirror: Arc, next_id: AtomicU64, username: String, + /// Where this session was reached, so the UI can say so. + endpoint: String, /// Set when the reader thread stops, so a call fails immediately rather /// than waiting out its timeout against a socket nobody is reading. closed: Arc, @@ -185,6 +187,7 @@ impl RemoteSession { mirror, next_id: AtomicU64::new(1), username: String::new(), + endpoint: endpoint.to_string(), closed: hung_up, }; @@ -467,6 +470,10 @@ impl SessionApi for RemoteSession { self.username.clone() } + fn daemon_endpoint(&self) -> Option { + Some(self.endpoint.clone()) + } + fn listen_port(&self) -> Option { self.status().and_then(|status| status.listen_port) } @@ -510,6 +517,31 @@ impl SessionApi for RemoteSession { .unwrap_or_default() } + fn all_searches(&self) -> Vec { + self.request::<_, Searches>(Method::SearchList, ()) + .map(|listed| { + listed + .searches + .into_iter() + .map(|search| SessionSearch { + query: search.query, + files: search.files, + started_secs_ago: Some(search.started_secs_ago), + }) + .collect() + }) + .unwrap_or_default() + } + + fn forget_search(&self, query: &str) -> bool { + self.ask( + Method::SearchForget, + QueryParams { + query: query.to_string(), + }, + ) + } + fn get_search_results_count(&self, key: &str) -> usize { self.get_search_results(key).len() } diff --git a/soulseek-rs/src/ui/downloads.rs b/soulseek-rs/src/ui/downloads.rs index cd2fa7e..0747f76 100644 --- a/soulseek-rs/src/ui/downloads.rs +++ b/soulseek-rs/src/ui/downloads.rs @@ -42,11 +42,18 @@ fn active_progress(downloads: &[DownloadEntry]) -> Option<(u64, u64, f64)> { } /// Renders download statistics in a reusable way +/// The status bar. +/// +/// `daemon` names the session's owner when this window is attached to one. It +/// belongs in the title because a shared session behaves differently from a +/// private one: another window can add transfers, and closing this one leaves +/// them running. pub fn render_download_stats( frame: &mut Frame, area: ratatui::layout::Rect, downloads: &[DownloadEntry], active_count: usize, + daemon: Option<&str>, ) { let completed = downloads .iter() @@ -72,7 +79,11 @@ pub fn render_download_stats( let active = active_progress(downloads); - let block = pane_block(false).title(plain_title("Status", false)); + let title = match daemon { + Some(endpoint) => format!("Status · daemon {endpoint}"), + None => "Status".to_string(), + }; + let block = pane_block(false).title(plain_title(&title, false)); let inner_area = block.inner(area); frame.render_widget(block, area); diff --git a/soulseek-rs/src/ui/main_tui/downloads.rs b/soulseek-rs/src/ui/main_tui/downloads.rs index 1180e7f..9fb3722 100644 --- a/soulseek-rs/src/ui/main_tui/downloads.rs +++ b/soulseek-rs/src/ui/main_tui/downloads.rs @@ -59,6 +59,22 @@ impl MainTui { /// Remove all completed / failed / timed-out downloads from the list. pub(super) fn clear_finished_downloads(&mut self) { + // Forget them in the session too. The list is a view of what the + // session holds, so dropping them only here would last until the next + // poll put them straight back. + for entry in &self.state.downloads { + if matches!( + entry.download.status, + DownloadStatus::Completed + | DownloadStatus::Failed(_) + | DownloadStatus::TimedOut + ) { + self.client.remove_download( + &entry.download.username, + &entry.download.filename, + ); + } + } self.state.downloads.retain(|entry| { !matches!( entry.download.status, @@ -218,13 +234,30 @@ impl MainTui { pub(super) fn update_downloads(&mut self) { if let Some(ref receiver) = self.state.downloads_receiver_channel { while let Ok((download, download_receiver)) = receiver.try_recv() { - self.state.downloads.push(DownloadEntry { - download, - receiver: Some(download_receiver), + // A transfer of ours can already be listed: the session + // registers it the moment it is queued, so a frame between + // that and the worker handing it over here will have picked it + // up from the session. Take over that row rather than adding a + // second one for the same file. + let existing = self.state.downloads.iter_mut().find(|entry| { + entry.download.username == download.username + && entry.download.filename == download.filename }); + match existing { + Some(entry) => { + entry.download = download; + entry.receiver = Some(download_receiver); + } + None => self.state.downloads.push(DownloadEntry { + download, + receiver: Some(download_receiver), + }), + } } } + self.sync_downloads_with_session(); + self.state.active_downloads_count = 0; for download_entry in &mut self.state.downloads { if let Some(ref receiver) = download_entry.receiver { @@ -241,4 +274,49 @@ impl MainTui { } } } + + /// Make the list show the session's transfers, not just this window's. + /// + /// The queue belongs to the session, so against a daemon it includes what + /// another client queued and what was still running before this window + /// opened. Transfers this window started keep their own status channel, + /// which reports sooner than a poll; everything else is refreshed from the + /// session each tick. + fn sync_downloads_with_session(&mut self) { + // Only worth doing when the session is shared. A window with a + // session of its own started every transfer in it, so there is + // nothing to learn — and the list also holds completed transfers + // restored from disk that the session has never heard of, which this + // reconciliation would take to mean "cancelled elsewhere" and delete. + if self.client.daemon_endpoint().is_none() { + return; + } + + let known = self.client.get_all_downloads(); + + for download in &known { + let existing = self.state.downloads.iter_mut().find(|entry| { + entry.download.username == download.username + && entry.download.filename == download.filename + }); + match existing { + // Ours: the channel is more current than this snapshot. + Some(entry) if entry.receiver.is_some() => {} + Some(entry) => entry.download.status = download.status.clone(), + None => self.state.downloads.push(DownloadEntry { + download: download.clone(), + receiver: None, + }), + } + } + + // Anything the session has forgotten is gone for every window. + self.state.downloads.retain(|entry| { + entry.receiver.is_some() + || known.iter().any(|download| { + download.username == entry.download.username + && download.filename == entry.download.filename + }) + }); + } } diff --git a/soulseek-rs/src/ui/main_tui/mod.rs b/soulseek-rs/src/ui/main_tui/mod.rs index 386c5a4..01e8aa2 100644 --- a/soulseek-rs/src/ui/main_tui/mod.rs +++ b/soulseek-rs/src/ui/main_tui/mod.rs @@ -269,15 +269,40 @@ pub fn launch_main_tui( mod tests { use super::*; use crate::daemon::proto::ChatMessageDto; - use crate::models::MessageDirection; + use crate::models::{MessageDirection, SearchStatus}; - /// A session that knows a conversation and nothing else — what attaching - /// to a daemon looks like from the TUI's side. - struct TalkativeSession(Vec); + /// A stand-in for a daemon: it already holds a conversation, a transfer + /// queue, and a search cache, none of which this window created. + #[derive(Default)] + struct TalkativeSession { + history: Vec, + downloads: Vec, + searches: std::sync::Mutex>, + /// Whether this session is a daemon's. Local sessions do not sync. + shared: bool, + } + + fn queued(username: &str, filename: &str) -> soulseek_rs::types::Download { + let (sender, _) = std::sync::mpsc::channel(); + soulseek_rs::types::Download { + username: username.to_string(), + filename: filename.to_string(), + token: 1, + size: 4096, + download_directory: "/tmp".to_string(), + status: soulseek_rs::DownloadStatus::Queued, + sender, + queue_position: None, + metadata: soulseek_rs::types::DownloadMetadata::default(), + } + } impl SessionApi for TalkativeSession { fn message_history(&self) -> Vec { - self.0.clone() + self.history.clone() + } + fn daemon_endpoint(&self) -> Option { + self.shared.then(|| "test-daemon".to_string()) } fn username(&self) -> String { "tester".to_string() @@ -302,6 +327,15 @@ mod tests { ) -> Vec { Vec::new() } + fn all_searches(&self) -> Vec { + self.searches.lock().expect("not poisoned").clone() + } + fn forget_search(&self, query: &str) -> bool { + let mut searches = self.searches.lock().expect("not poisoned"); + let before = searches.len(); + searches.retain(|search| search.query != query); + searches.len() != before + } fn get_search_results_count(&self, _key: &str) -> usize { 0 } @@ -346,7 +380,7 @@ mod tests { Err(soulseek_rs::SoulseekRs::NotConnected) } fn get_all_downloads(&self) -> Vec { - Vec::new() + self.downloads.clone() } fn pause_download(&self, _u: &str, _f: &str) -> bool { false @@ -441,8 +475,36 @@ mod tests { } fn tui(history: Vec) -> MainTui { + with_session(TalkativeSession { + history, + shared: true, + ..TalkativeSession::default() + }) + } + + fn shared(searches: Vec) -> TalkativeSession { + TalkativeSession { + searches: std::sync::Mutex::new(searches), + shared: true, + ..TalkativeSession::default() + } + } + + fn search_of( + query: &str, + files: usize, + age: u64, + ) -> crate::api::SessionSearch { + crate::api::SessionSearch { + query: query.to_string(), + files, + started_secs_ago: Some(age), + } + } + + fn with_session(session: TalkativeSession) -> MainTui { MainTui::new( - Arc::new(TalkativeSession(history)), + Arc::new(session), "/tmp".to_string(), 1, Duration::from_secs(1), @@ -481,6 +543,167 @@ mod tests { assert_eq!(tui.state.chat_peers(), ["bob"]); } + #[test] + fn a_window_shows_transfers_it_did_not_start() { + // Two windows on one daemon are two views of one queue. A transfer + // queued in the other one has to appear here, or closing a window + // would look like losing its downloads. + let mut tui = with_session(TalkativeSession { + downloads: vec![queued("bob", "@@x\\theirs.mp3")], + shared: true, + ..TalkativeSession::default() + }); + assert!( + tui.state.downloads.is_empty(), + "nothing until the first poll" + ); + + tui.update_downloads(); + + assert_eq!(tui.state.downloads.len(), 1); + assert_eq!(tui.state.downloads[0].download.filename, "@@x\\theirs.mp3"); + assert!( + tui.state.downloads[0].receiver.is_none(), + "someone else's transfer has no channel of ours" + ); + + // Polling again must not double it up. + tui.update_downloads(); + assert_eq!(tui.state.downloads.len(), 1); + } + + #[test] + fn a_transfer_the_session_forgets_leaves_every_window() { + let mut tui = with_session(TalkativeSession { + downloads: vec![queued("bob", "gone.mp3")], + shared: true, + ..TalkativeSession::default() + }); + tui.update_downloads(); + assert_eq!(tui.state.downloads.len(), 1); + + // The session no longer has it — cancelled from the other window. + tui.client = Arc::new(TalkativeSession { + shared: true, + ..TalkativeSession::default() + }); + tui.update_downloads(); + assert!( + tui.state.downloads.is_empty(), + "a queue is shared in both directions" + ); + } + + #[test] + fn a_window_shows_searches_run_in_another() { + let mut tui = + with_session(shared(vec![search_of("their query", 7, 9_999)])); + tui.update_search_results(); + + assert_eq!(tui.state.searches.len(), 1); + assert_eq!(tui.state.searches[0].query, "their query"); + + // And it is not added a second time on the next frame. + tui.update_search_results(); + assert_eq!(tui.state.searches.len(), 1); + } + + #[test] + fn a_search_another_window_is_still_running_shows_as_active() { + // It had said "Done" the moment it appeared, because an imported + // search carried no notion of still collecting — so the other window + // looked finished while results were still arriving. + let mut tui = + with_session(shared(vec![search_of("still going", 3, 0)])); + tui.update_search_results(); + + assert_eq!(tui.state.searches[0].status, SearchStatus::Active); + assert_eq!( + tui.state.searches[0].known_files, 3, + "and its count comes across before the results do" + ); + } + + #[test] + fn a_search_that_has_run_its_window_shows_as_done() { + let mut tui = + with_session(shared(vec![search_of("finished", 12, 9_999)])); + tui.update_search_results(); + assert_eq!(tui.state.searches[0].status, SearchStatus::Completed); + } + + #[test] + fn removing_a_search_removes_it_from_the_session() { + // Removing it only from this window would last until the next sync, + // which would put it straight back — and in a shared session it would + // come back for every other window too. + let mut tui = + with_session(shared(vec![search_of("unwanted", 1, 9_999)])); + tui.update_search_results(); + assert_eq!(tui.state.searches.len(), 1); + + tui.remove_search_at_index(0); + tui.update_search_results(); + + assert!(tui.state.searches.is_empty(), "and it stays gone"); + } + + #[test] + fn clearing_every_search_clears_them_in_the_session_too() { + let mut tui = with_session(shared(vec![ + search_of("one", 1, 9_999), + search_of("two", 2, 9_999), + ])); + tui.update_search_results(); + assert_eq!(tui.state.searches.len(), 2); + + tui.clear_all_searches(); + tui.update_search_results(); + + assert!( + tui.state.searches.is_empty(), + "clearing has to reach the session, or every row comes back" + ); + } + + #[test] + fn a_search_dismissed_in_another_window_leaves_this_one() { + let session = shared(vec![search_of("theirs", 4, 9_999)]); + let mut tui = with_session(session); + tui.update_search_results(); + assert_eq!(tui.state.searches.len(), 1); + + // The other window dismissed it. + tui.client.forget_search("theirs"); + tui.update_search_results(); + + assert!(tui.state.searches.is_empty()); + } + + #[test] + fn a_window_with_its_own_session_is_left_alone() { + // Nothing to share, so nothing is synced — and in particular the + // completed transfers restored from disk, which the session has never + // heard of, are not mistaken for cancelled and deleted. + let mut tui = with_session(TalkativeSession { + downloads: Vec::new(), + shared: false, + ..TalkativeSession::default() + }); + tui.state.downloads.push(crate::models::DownloadEntry { + download: queued("bob", "from-disk.mp3"), + receiver: None, + }); + + tui.update_downloads(); + + assert_eq!( + tui.state.downloads.len(), + 1, + "a local window's history must survive the tick" + ); + } + #[test] fn a_local_session_starts_from_its_own_snapshot_instead() { // Locally the session knows no history and the state file is the whole diff --git a/soulseek-rs/src/ui/main_tui/render.rs b/soulseek-rs/src/ui/main_tui/render.rs index f0f7779..5e108fc 100644 --- a/soulseek-rs/src/ui/main_tui/render.rs +++ b/soulseek-rs/src/ui/main_tui/render.rs @@ -41,11 +41,13 @@ impl MainTui { constraints.push(Constraint::Length(3)); // Shortcuts let main_chunks = Layout::vertical(constraints).split(frame.area()); + let daemon = self.client.daemon_endpoint(); render_download_stats( frame, main_chunks[0], &self.state.downloads, self.state.active_downloads_count, + daemon.as_deref(), ); self.render_content(frame, main_chunks[1]); diff --git a/soulseek-rs/src/ui/main_tui/search.rs b/soulseek-rs/src/ui/main_tui/search.rs index 9d9bee7..5a443f7 100644 --- a/soulseek-rs/src/ui/main_tui/search.rs +++ b/soulseek-rs/src/ui/main_tui/search.rs @@ -26,6 +26,13 @@ impl MainTui { // Check if we're removing the currently active search let was_active_search = self.state.selected_search_index == Some(index); + // Drop it from the session too. The list is a view of what the session + // holds, so removing it only here would last until the next sync put + // it straight back — and against a daemon it would come back for every + // other window as well. + if let Some(search) = self.state.searches.get(index) { + self.client.forget_search(&search.query); + } self.state.searches.remove(index); if let Some(current_idx) = self.state.selected_search_index { @@ -64,7 +71,19 @@ impl MainTui { .store(true, std::sync::atomic::Ordering::Relaxed); } - // Clear searches + // And drop them from the session, or the next sync brings every one + // of them straight back — the same reason removing a single search + // has to say so. + let queries: Vec = self + .state + .searches + .iter() + .map(|search| search.query.clone()) + .collect(); + for query in queries { + self.client.forget_search(&query); + } + self.state.searches.clear(); self.state.searches_table_state.select(None); self.state.selected_search_index = None; @@ -173,6 +192,8 @@ impl MainTui { query: query.clone(), status: SearchStatus::Active, results: Vec::new(), + known_files: 0, + owned: true, start_time: Instant::now(), cancel_flag: cancel_flag.clone(), }; @@ -213,17 +234,98 @@ impl MainTui { }); } + /// Show the searches the session knows about, not just this window's. + /// + /// 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) { + // 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. + if self.client.daemon_endpoint().is_none() { + return; + } + + let timeout = self.search_timeout.as_secs(); + for known in self.client.all_searches() { + let running = known + .started_secs_ago + .is_some_and(|elapsed| elapsed < timeout); + + if let Some(existing) = self + .state + .searches + .iter_mut() + .find(|search| search.query == known.query) + { + // Ours keeps the status its own worker thread sets; the rest + // follow the session, so a search finishing in one window + // stops saying "Active" in the others. + if known.started_secs_ago.is_some() && !existing.owned { + existing.status = if running { + SearchStatus::Active + } else { + SearchStatus::Completed + }; + existing.known_files = known.files; + } + continue; + } + + self.state.searches.push(SearchEntry { + query: known.query, + status: if running { + SearchStatus::Active + } else { + SearchStatus::Completed + }, + results: Vec::new(), + known_files: known.files, + owned: false, + start_time: std::time::Instant::now(), + cancel_flag: std::sync::Arc::new( + std::sync::atomic::AtomicBool::new(false), + ), + }); + } + + // 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)); + } + pub(super) fn update_search_results(&mut self) { + 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. let all_results: Vec<(usize, Vec<_>)> = self .state .searches .iter() .enumerate() + .filter(|(idx, search)| { + search.status == SearchStatus::Active + || selected_search_index == Some(*idx) + }) .map(|(idx, s)| (idx, s.query.clone())) .filter_map(|(idx, query)| { self.client diff --git a/soulseek-rs/src/ui/panes/searches_pane.rs b/soulseek-rs/src/ui/panes/searches_pane.rs index 22e525b..a0aaeaf 100644 --- a/soulseek-rs/src/ui/panes/searches_pane.rs +++ b/soulseek-rs/src/ui/panes/searches_pane.rs @@ -35,7 +35,14 @@ pub fn render_searches_pane( } }; - let results_count = search.results.len(); + // What this window has fetched, or what the session says it will + // find when it does. A search another window ran has a count long + // before this one has pulled the results across. + let results_count = if search.results.is_empty() { + search.known_files + } else { + search.results.len() + }; let results_text = if results_count == 0 && search.status == SearchStatus::Active { diff --git a/soulseek-rs/tests/daemon_protocol_e2e.rs b/soulseek-rs/tests/daemon_protocol_e2e.rs index 32982f3..eb586aa 100644 --- a/soulseek-rs/tests/daemon_protocol_e2e.rs +++ b/soulseek-rs/tests/daemon_protocol_e2e.rs @@ -667,3 +667,129 @@ fn rapid_connect_and_drop_does_not_wear_the_daemon_out() { "abandoned connections must not still be counted: {status}" ); } + +/// What one client does, the other sees. +/// +/// Two TUIs on one daemon are two views of a single session, not two sessions +/// that happen to share a login. A search run in one appears in the other's +/// list, and a transfer queued in one appears in the other's queue — which is +/// what makes closing one window harmless. +#[test] +fn one_clients_work_shows_up_in_another() { + let daemon = daemon_or_skip!(); + let mut first = daemon.connect_tcp().expect("connect"); + first.authenticate(Some(&daemon.token())).expect("first in"); + let mut second = daemon.connect_tcp().expect("connect"); + second + .authenticate(Some(&daemon.token())) + .expect("second in"); + + // Nothing searched yet, so neither sees anything. + let queries = |connection: &mut Connection, id: u32| -> Vec { + let reply = connection + .call(&format!( + r#"{{"jsonrpc":"2.0","id":{id},"method":"search.list"}}"# + )) + .expect("an answer"); + let parsed: serde_json::Value = + serde_json::from_str(&reply).expect("valid JSON"); + parsed["result"]["searches"] + .as_array() + .expect("searches is an array") + .iter() + .filter_map(|search| search["query"].as_str().map(String::from)) + .collect() + }; + assert!(!queries(&mut second, 1).contains(&"shared_probe".to_string())); + + // The first client searches. + first + .call( + r#"{"jsonrpc":"2.0","id":2,"method":"search.start","params":{"query":"shared_probe"}}"#, + ) + .expect("an answer"); + + // The second client sees that search without having run it. + assert!( + queries(&mut second, 3).contains(&"shared_probe".to_string()), + "a search in one window must appear in the other" + ); + + // And its results are the same set, because there is one search cache. + let results = |connection: &mut Connection, id: u32| -> usize { + let reply = connection + .call(&format!( + r#"{{"jsonrpc":"2.0","id":{id},"method":"search.results","params":{{"query":"shared_probe"}}}}"# + )) + .expect("an answer"); + let parsed: serde_json::Value = + serde_json::from_str(&reply).expect("valid JSON"); + parsed["result"]["results"] + .as_array() + .expect("results is an array") + .len() + }; + assert_eq!( + results(&mut first, 4), + results(&mut second, 5), + "both windows read one search cache" + ); +} + +/// A client that attaches later starts from the session's state. +/// +/// This is what makes a second window useful at all: it opens showing what is +/// already going on rather than a blank slate that fills only with what it +/// does itself. +#[test] +fn a_client_that_attaches_later_sees_what_is_already_there() { + let daemon = daemon_or_skip!(); + + // An early client does some work. + let mut early = daemon.connect_tcp().expect("connect"); + early.authenticate(Some(&daemon.token())).expect("in"); + early + .call( + r#"{"jsonrpc":"2.0","id":1,"method":"search.start","params":{"query":"earlier_probe"}}"#, + ) + .expect("an answer"); + early + .call( + r#"{"jsonrpc":"2.0","id":2,"method":"room.join","params":{"room":"cli_e2e_late_room"}}"#, + ) + .expect("an answer"); + + // A window opened afterwards is told about it. + let mut late = daemon.connect_tcp().expect("connect"); + late.authenticate(Some(&daemon.token())).expect("in"); + + let reply = late + .call(r#"{"jsonrpc":"2.0","id":3,"method":"search.list"}"#) + .expect("an answer"); + let parsed: serde_json::Value = + serde_json::from_str(&reply).expect("valid JSON"); + let mut queries = parsed["result"]["searches"] + .as_array() + .expect("searches is an array") + .iter() + .filter_map(|search| search["query"].as_str()); + assert!( + queries.any(|query| query == "earlier_probe"), + "a window that opens later must see the searches already run: {reply}" + ); + + // The transfer queue is the session's, so both read the same one. + for (connection, id) in [(&mut early, 4), (&mut late, 5)] { + let reply = connection + .call(&format!( + r#"{{"jsonrpc":"2.0","id":{id},"method":"download.list"}}"# + )) + .expect("an answer"); + let parsed: serde_json::Value = + serde_json::from_str(&reply).expect("valid JSON"); + assert!( + parsed["result"]["downloads"].is_array(), + "both windows read one queue: {reply}" + ); + } +} diff --git a/website/src/pages/docs.tsx b/website/src/pages/docs.tsx index e2adfe7..1083f73 100644 --- a/website/src/pages/docs.tsx +++ b/website/src/pages/docs.tsx @@ -250,6 +250,12 @@ const Daemon = () => ( open, so there is no address to pass and no token to handle. Without one running, everything behaves exactly as it did before.

+

+ Every window is a view of the same session. Open a second TUI and it shows the + searches and transfers already running; queue something in either and both see + it. The status bar names the daemon, and closing a window leaves its downloads + going, because they belong to the daemon rather than the window. +

Why bother: the server permits a single session per account. Two one-off commands at once either cut each other off, or push you into a throwaway