From aa61a1a434c656fdd2bbdb5cdcafdf314f167542 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Sun, 6 Sep 2026 21:40:45 +0200 Subject: [PATCH] feat: cancel a download from the TUI, the CLI, and the daemon Closes #54. A download can now be cancelled whether it is queued, in progress, or paused. Cancelled is a terminal status: the store refuses to overwrite it, the status channel relays nothing after it, the partial file is deleted, and the transfer thread releases its socket at its next one-second poll instead of the 30s read deadline. A queued download that is cancelled declines the peer's later offer with TransferResponse "Cancelled", and an inbound file connection for it is refused. A fresh download of the same file evicts the dead row so it is not shadowed, and a cancelled transfer is not re-queued after a restart. Surfaces: Client::cancel_download, download.cancel over RPC, `transfer cancel ` on the CLI (which also stops an upload by that name), and x in the transfers pane, where d dismisses the row. --- README.md | 7 +- soulseek-rs-lib/examples/stress.rs | 1 + soulseek-rs-lib/src/client/downloads.rs | 19 ++ soulseek-rs-lib/src/client/operations.rs | 22 +- soulseek-rs-lib/src/client/tests.rs | 65 +++++ soulseek-rs-lib/src/download_store.rs | 119 ++++++++- .../src/message/server/message_factory.rs | 22 ++ soulseek-rs-lib/src/peer/download_peer.rs | 189 ++++++++++++-- soulseek-rs-lib/src/peer/listen.rs | 21 +- soulseek-rs-lib/src/peer/mod.rs | 3 +- soulseek-rs-lib/src/types.rs | 21 +- soulseek-rs-lib/tests/e2e.rs | 238 ++++++++++++++++++ soulseek-rs/docs/openrpc.json | 39 +++ soulseek-rs/src/SKILL.md | 5 + soulseek-rs/src/api.rs | 5 + soulseek-rs/src/cli.rs | 16 ++ soulseek-rs/src/commands/mod.rs | 5 +- soulseek-rs/src/commands/transfer.rs | 16 ++ soulseek-rs/src/daemon/mod.rs | 11 +- soulseek-rs/src/daemon/openrpc.rs | 5 + soulseek-rs/src/daemon/proto/dto.rs | 3 + soulseek-rs/src/daemon/proto/mod.rs | 2 + soulseek-rs/src/daemon/server/mod.rs | 12 + soulseek-rs/src/persist/snapshot.rs | 35 ++- soulseek-rs/src/persist/state.rs | 17 ++ soulseek-rs/src/remote/mirror.rs | 7 +- soulseek-rs/src/remote/mod.rs | 4 + soulseek-rs/src/ui/main_tui/downloads.rs | 78 +++--- soulseek-rs/src/ui/main_tui/input.rs | 2 +- soulseek-rs/src/ui/main_tui/mod.rs | 113 ++++++++- soulseek-rs/src/ui/main_tui/render.rs | 1 + .../src/ui/panes/download_info_pane.rs | 7 +- soulseek-rs/src/ui/panes/downloads_pane.rs | 5 +- soulseek-rs/src/ui/panes/mod.rs | 4 +- soulseek-rs/src/ui/styles.rs | 2 + soulseek-rs/tests/cli_e2e.rs | 74 ++++++ soulseek-rs/tests/daemon_protocol_e2e.rs | 42 ++++ website/src/pages/docs.tsx | 4 +- 38 files changed, 1139 insertions(+), 102 deletions(-) diff --git a/README.md b/README.md index b0d0816..0621669 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,8 @@ does, how to install it, and every `config.toml` setting. - **Browse**: list any user's shared files and download straight from the tree - **Resumable downloads**: a dead transfer leaves a `.part`, and re-running it asks the peer for only what is missing +- **Cancel a transfer**: `x` in the TUI or `transfer cancel` from a script stops + a download, queued or in progress, and deletes its `.part`; uploads too - **Chat rooms**: list, join, and talk in public rooms, several open at once - **Private messages**: send and receive, with an inbox in the TUI - **Firewalled peers**: browsing and downloads fall back to server-brokered @@ -251,6 +253,7 @@ 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 transfer cancel # stop a download or upload 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 @@ -290,9 +293,9 @@ account of its own can still ask whether one is running. `shares status` and `shares reindex` are not among them, because reporting what the network sees means logging in and scanning the folders. -Six commands perform an action and print nothing, answering with their exit +Seven commands perform an action and print nothing, answering with their exit code alone: `room say`, `message send`, `shares add`, `shares remove`, -`wish remove` and `daemon stop`. `room listen`, `message read` and `serve` +`wish remove`, `transfer cancel` and `daemon stop`. `room listen`, `message read` and `serve` stream records until `--duration` seconds pass or, with `--follow`, until interrupted. The rest print records and exit. diff --git a/soulseek-rs-lib/examples/stress.rs b/soulseek-rs-lib/examples/stress.rs index d82e939..e51eb19 100644 --- a/soulseek-rs-lib/examples/stress.rs +++ b/soulseek-rs-lib/examples/stress.rs @@ -1197,6 +1197,7 @@ fn main() { soulseek_rs::DownloadStatus::Queued => "Queued", soulseek_rs::DownloadStatus::InProgress { .. } => "InProgress", soulseek_rs::DownloadStatus::Completed => "Completed", + soulseek_rs::DownloadStatus::Cancelled => "Cancelled", soulseek_rs::DownloadStatus::Failed(_) => "Failed", soulseek_rs::DownloadStatus::TimedOut => "TimedOut", soulseek_rs::DownloadStatus::Paused { .. } => "Paused", diff --git a/soulseek-rs-lib/src/client/downloads.rs b/soulseek-rs-lib/src/client/downloads.rs index 076d4d6..28c8177 100644 --- a/soulseek-rs-lib/src/client/downloads.rs +++ b/soulseek-rs-lib/src/client/downloads.rs @@ -3,6 +3,7 @@ use super::{ Receiver, Result, RwLock, RwLockExt, Sender, error, info, mpsc, next_download_token, }; +use crate::peer::part_path_of; impl Client { #[must_use] @@ -35,6 +36,24 @@ impl Client { } } + #[must_use] + pub fn cancel_download(&self, username: &str, filename: &str) -> bool { + let cancelled = match self.context.write_safe() { + Ok(mut ctx) => ctx.downloads.cancel_by_file(username, filename), + Err(e) => { + error!("[client] cancel_download: {}", e); + None + } + }; + let Some(download) = cancelled else { + return false; + }; + if let Some(part) = part_path_of(&download) { + let _ = std::fs::remove_file(part); + } + true + } + #[must_use] pub fn remove_queued_download( &self, diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index a2d332b..bcd2cc7 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -6,6 +6,7 @@ use super::{ thread, trace, warn, }; use crate::message::server::MessageFactory; +use crate::peer::DownloadError; const CONNECT_SWEEP_INTERVAL: Duration = Duration::from_secs(1); @@ -148,6 +149,7 @@ impl Client { download.status, DownloadStatus::InProgress { .. } | DownloadStatus::Completed + | DownloadStatus::Cancelled ) { debug!( "[client] transfer token {} already claimed; \ @@ -210,6 +212,7 @@ impl Client { download.size, filename ); } + Err(DownloadError::Cancelled) => {} Err(e) => { let reason = Some(e.to_string()); let _ = download.sender.send( @@ -363,7 +366,14 @@ impl Client { } }); - if let Some((old_token, download)) = download_to_update + let cancelled = download_to_update + .as_ref() + .is_some_and(|(_, d)| { + matches!(d.status, DownloadStatus::Cancelled) + }); + if !cancelled + && let Some((old_token, download)) = + download_to_update { trace!( "[client] UpdateDownloadTokens found {old_token}, transfer: {:?}", @@ -386,10 +396,16 @@ impl Client { let registry = context.peer_registry.clone(); drop(context); if let Some(registry) = registry { - let response = + let response = if cancelled { + MessageFactory::build_transfer_denial_message( + transfer.token, + "Cancelled", + ) + } else { MessageFactory::build_transfer_response_message( transfer, - ); + ) + }; let _ = registry.send_to_peer( &username, PeerMessage::SendMessage(response), diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 46a424b..73e892a 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -623,6 +623,71 @@ fn a_replayed_transfer_response_does_not_start_a_second_transfer() { ); } +#[test] +fn a_cancelled_download_is_not_started_when_the_peer_allows_it() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let port = u32::from(listener.local_addr().unwrap().port()); + + let client = Client::new("u", "p"); + let (sender, _receiver) = mpsc::channel(); + client.context.write().unwrap().add_download(download( + "peer", + "f.mp3", + 9, + DownloadStatus::Cancelled, + sender, + )); + + let (ops_tx, ops_rx) = mpsc::channel(); + Client::listen_to_client_operations( + ops_rx, + client.context, + "u".to_string(), + ); + let peer = Peer::new( + "peer".to_string(), + ConnectionType::F, + "127.0.0.1".to_string(), + port, + None, + 0, + 0, + 0, + ); + ops_tx + .send(ClientOperation::DownloadFromPeer(9, peer, true)) + .unwrap(); + + thread::sleep(Duration::from_millis(500)); + assert!( + matches!(listener.accept(), Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock), + "a cancelled download must not dial the peer" + ); +} + +#[test] +fn cancel_download_marks_the_store_and_answers_whether_it_matched() { + let client = Client::new("u", "p"); + let (sender, receiver) = mpsc::channel(); + client.context.write().unwrap().add_download(download( + "peer", + "f.mp3", + 9, + DownloadStatus::Queued, + sender, + )); + + assert!(client.cancel_download("peer", "f.mp3")); + assert!(!client.cancel_download("peer", "f.mp3")); + assert!(!client.cancel_download("peer", "other.mp3")); + assert!(matches!(receiver.try_recv(), Ok(DownloadStatus::Cancelled))); + assert!(matches!( + client.get_all_downloads()[0].status, + DownloadStatus::Cancelled + )); +} + #[test] fn room_member_stats_are_returned_sorted_and_scoped_to_their_room() { use crate::types::{RoomUserStats, UserStatus}; diff --git a/soulseek-rs-lib/src/download_store.rs b/soulseek-rs-lib/src/download_store.rs index 532a0a3..3cd9699 100644 --- a/soulseek-rs-lib/src/download_store.rs +++ b/soulseek-rs-lib/src/download_store.rs @@ -12,6 +12,11 @@ impl DownloadStore { } pub fn add(&mut self, download: Download) { + self.downloads.retain(|d| { + !(d.is_finished() + && d.username == download.username + && d.filename == download.filename) + }); self.downloads.push(download); } @@ -48,10 +53,33 @@ impl DownloadStore { &self.downloads } - pub fn update_status(&mut self, token: u32, status: DownloadStatus) { - if let Some(download) = self.get_by_token_mut(token) { - download.status = status; + pub fn update_status( + &mut self, + token: u32, + status: DownloadStatus, + ) -> bool { + let Some(download) = self.get_by_token_mut(token) else { + return false; + }; + if matches!(download.status, DownloadStatus::Cancelled) { + return false; } + download.status = status; + true + } + + pub fn cancel_by_file( + &mut self, + username: &str, + filename: &str, + ) -> Option { + let download = self.get_by_file_mut(username, filename)?; + if download.is_finished() { + return None; + } + download.status = DownloadStatus::Cancelled; + let _ = download.sender.send(DownloadStatus::Cancelled); + Some(download.clone()) } pub fn update_queue_position( @@ -218,6 +246,91 @@ mod tests { assert!(!store.update_queue_position("other", "song.mp3", 1)); } + #[test] + fn cancel_by_file_flips_an_unfinished_download_and_tells_its_channel() { + let mut store = DownloadStore::new(); + let (tx, rx) = mpsc::channel(); + let mut queued = make_download(1, DownloadStatus::Queued); + queued.sender = tx; + store.add(queued); + store.add(make_download( + 2, + DownloadStatus::InProgress { + bytes_downloaded: 5, + total_bytes: 100, + speed_bytes_per_sec: 1.0, + }, + )); + store.add(make_download( + 3, + DownloadStatus::Paused { + bytes_downloaded: 5, + total_bytes: 100, + }, + )); + store.add(make_download(4, DownloadStatus::Completed)); + store.add(make_download(5, DownloadStatus::Failed(None))); + + assert!(store.cancel_by_file("peer", "file-1.mp3").is_some()); + assert!(matches!(rx.try_recv(), Ok(DownloadStatus::Cancelled))); + assert!(store.cancel_by_file("peer", "file-2.mp3").is_some()); + assert!(store.cancel_by_file("peer", "file-3.mp3").is_some()); + assert!(store.cancel_by_file("peer", "file-4.mp3").is_none()); + assert!(store.cancel_by_file("peer", "file-5.mp3").is_none()); + assert!(store.cancel_by_file("peer", "file-9.mp3").is_none()); + for token in 1..=3 { + let download = store.get_by_token(token).unwrap(); + assert!(matches!(download.status, DownloadStatus::Cancelled)); + assert!(download.is_finished()); + } + } + + #[test] + fn a_cancelled_download_stays_cancelled() { + let mut store = DownloadStore::new(); + store.add(make_download(1, DownloadStatus::Queued)); + assert!(store.cancel_by_file("peer", "file-1.mp3").is_some()); + assert!(!store.update_status( + 1, + DownloadStatus::InProgress { + bytes_downloaded: 0, + total_bytes: 100, + speed_bytes_per_sec: 0.0, + }, + )); + assert!( + !store + .update_status(1, DownloadStatus::Failed(Some("late".into()))) + ); + assert!(matches!( + store.get_by_token(1).unwrap().status, + DownloadStatus::Cancelled + )); + assert!(store.cancel_by_file("peer", "file-1.mp3").is_none()); + assert!(!store.resume_by_file("peer", "file-1.mp3")); + assert!(!store.pause_by_file("peer", "file-1.mp3")); + } + + #[test] + fn add_evicts_a_finished_entry_for_the_same_file() { + let mut store = DownloadStore::new(); + store.add(make_download(1, DownloadStatus::Cancelled)); + store.add(make_download(2, DownloadStatus::Queued)); + let mut fresh = make_download(3, DownloadStatus::Queued); + fresh.filename = "file-1.mp3".to_string(); + store.add(fresh); + let mut again = make_download(4, DownloadStatus::Queued); + again.filename = "file-2.mp3".to_string(); + store.add(again); + + assert_eq!(store.tokens(), vec![2, 3, 4], "only the dead row goes"); + assert!(store.cancel_by_file("peer", "file-1.mp3").is_some()); + assert!(matches!( + store.get_by_token(3).unwrap().status, + DownloadStatus::Cancelled + )); + } + #[test] fn pause_then_resume_in_progress_download() { let mut store = DownloadStore::new(); diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 4ce3256..2cf6793 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -249,6 +249,16 @@ impl MessageFactory { .write_bool(true) .clone() } + #[must_use] + pub fn build_transfer_denial_message(token: u32, reason: &str) -> Message { + Message::new() + .write_int32(41) + .write_int32(token) + .write_bool(false) + .write_string(reason) + .clone() + } + #[must_use] pub fn build_pierce_firewall_message(token: u32) -> Message { Message::new() @@ -321,6 +331,18 @@ fn test_build_upload_transfer_request() { assert_eq!(transfer.size, 4096); } +#[test] +fn a_transfer_denial_carries_the_token_and_the_reason() { + let message = + MessageFactory::build_transfer_denial_message(555, "Cancelled"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code_u32(), 41); + decoded.set_pointer(8); + assert_eq!(decoded.read_int32(), 555); + assert!(!decoded.read_bool()); + assert_eq!(decoded.read_string(), "Cancelled"); +} + #[test] fn test_build_peer_init_message() { use crate::peer::ConnectionType; diff --git a/soulseek-rs-lib/src/peer/download_peer.rs b/soulseek-rs-lib/src/peer/download_peer.rs index accc2b0..b29b4a5 100644 --- a/soulseek-rs-lib/src/peer/download_peer.rs +++ b/soulseek-rs-lib/src/peer/download_peer.rs @@ -14,6 +14,8 @@ use crate::types::{Download, DownloadStatus}; use crate::utils::path::{PART_SUFFIX, expand_tilde}; const READ_BUFFER_SIZE: usize = 8192; +const CANCEL_POLL: Duration = Duration::from_secs(1); +const STALL_DEADLINE: Duration = Duration::from_secs(30); const PROGRESS_UPDATE_CHUNKS: usize = 15; // ~120KB (15 * 8192 bytes) #[derive(Debug)] @@ -29,6 +31,7 @@ pub enum DownloadError { PathResolutionError(String), InvalidTokenBytes, LockPoisoned, + Cancelled, IncompleteDownload { received: usize, expected: usize }, } @@ -56,6 +59,7 @@ impl std::fmt::Display for DownloadError { write!(f, "Invalid token bytes received") } Self::LockPoisoned => write!(f, "Lock poisoned"), + Self::Cancelled => write!(f, "Download cancelled"), Self::IncompleteDownload { received, expected } => write!( f, "Incomplete download: received {received} of {expected} bytes" @@ -73,6 +77,16 @@ fn extract_filename_from_path(full_path: &str) -> &str { .unwrap_or(full_path) } +fn part_path(final_path: &str) -> String { + format!("{final_path}{PART_SUFFIX}") +} + +pub fn part_path_of(download: &Download) -> Option { + resolve_download_path(download) + .ok() + .map(|path| part_path(&path)) +} + /// Where a download's bytes end up: its configured directory (falling back /// to the parent if it names a file) joined with the remote basename. /// @@ -121,7 +135,7 @@ struct PartFile { impl PartFile { fn open(download: &Download) -> Result { let final_path = resolve_download_path(download)?; - let path = PathBuf::from(format!("{final_path}{PART_SUFFIX}")); + let path = PathBuf::from(part_path(&final_path)); if let Some(parent) = path.parent() { fs::create_dir_all(parent) @@ -164,6 +178,11 @@ impl PartFile { self.written >= self.expected } + fn discard(self) { + drop(self.file); + let _ = fs::remove_file(part_path(&self.final_path)); + } + /// A short transfer is a failure, and the `.part` stays put so the next /// attempt can resume it. fn finish(self) -> Result { @@ -174,11 +193,8 @@ impl PartFile { }); } drop(self.file); - fs::rename( - format!("{}{PART_SUFFIX}", self.final_path), - &self.final_path, - ) - .map_err(DownloadError::FileWriteError)?; + fs::rename(part_path(&self.final_path), &self.final_path) + .map_err(DownloadError::FileWriteError)?; Ok(self.final_path) } } @@ -245,7 +261,7 @@ impl DownloadPeer { stream: &TcpStream, ) -> Result<(), DownloadError> { stream - .set_read_timeout(Some(Duration::from_secs(30))) + .set_read_timeout(Some(STALL_DEADLINE)) .map_err(DownloadError::ConnectionFailed)?; stream .set_write_timeout(Some(Duration::from_secs(5))) @@ -289,6 +305,7 @@ impl DownloadPeer { stream: &mut TcpStream, client_context: &Arc>, ) -> Result<(Download, PartFile), DownloadError> { + Self::wait_while_paused(client_context, &download)?; let part = PartFile::open(&download)?; stream .write_all(&part.written.to_le_bytes()) @@ -337,6 +354,10 @@ impl DownloadPeer { let mut read_buffer = [0u8; READ_BUFFER_SIZE]; let mut chunk_counter = 0usize; let mut last_update_time = Instant::now(); + let mut last_data = Instant::now(); + stream + .set_read_timeout(Some(CANCEL_POLL)) + .map_err(DownloadError::ConnectionFailed)?; trace!( "[download_peer:{}] Starting to read data from peer", @@ -349,8 +370,19 @@ impl DownloadPeer { }; loop { - if let Some((ref dl, _)) = transfer { - Self::wait_while_paused(client_context, dl)?; + if let Some((dl, _)) = &transfer { + match Self::wait_while_paused(client_context, dl) { + Ok(true) => last_data = Instant::now(), + Ok(false) => {} + Err(e) => { + if matches!(e, DownloadError::Cancelled) + && let Some((_, part)) = transfer.take() + { + part.discard(); + } + return Err(e); + } + } } match stream.read(&mut read_buffer) { @@ -362,6 +394,7 @@ impl DownloadPeer { break; } Ok(bytes_read) => { + last_data = Instant::now(); let data = &read_buffer[..bytes_read]; if transfer.is_none() && !self.no_pierce { @@ -404,6 +437,11 @@ impl DownloadPeer { break; } } + Err(e) + if matches!( + e.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) && last_data.elapsed() < STALL_DEADLINE => {} Err(e) => { return Err(DownloadError::StreamReadError(e)); } @@ -444,16 +482,21 @@ impl DownloadPeer { download: &Download, status: DownloadStatus, ) { - let _ = download.sender.send(status.clone()); - if let Ok(mut context) = client_context.write() { - context.update_download_with_status(download.token, status); + let applied = client_context.write().is_ok_and(|mut context| { + context + .downloads + .update_status(download.token, status.clone()) + }); + if applied { + let _ = download.sender.send(status); } } fn wait_while_paused( client_context: &Arc>, download: &Download, - ) -> Result<(), DownloadError> { + ) -> Result { + let mut waited = false; loop { let status = client_context .read() @@ -462,10 +505,14 @@ impl DownloadPeer { .map(|download| download.status.clone()) .ok_or(DownloadError::TokenNotFound(download.token))?; + if matches!(status, DownloadStatus::Cancelled) { + return Err(DownloadError::Cancelled); + } if !matches!(status, DownloadStatus::Paused { .. }) { - return Ok(()); + return Ok(waited); } + waited = true; thread::sleep(Duration::from_millis(200)); } } @@ -514,8 +561,9 @@ mod tests { use super::{ DownloadError, DownloadPeer, PartFile, extract_filename_from_path, }; + use crate::client::ClientContext; use crate::types::{Download, DownloadMetadata, DownloadStatus}; - use std::sync::mpsc; + use std::sync::{Arc, RwLock, mpsc}; fn scratch_dir(name: &str) -> std::path::PathBuf { let dir = std::env::temp_dir() @@ -539,6 +587,117 @@ mod tests { } } + #[test] + fn a_discarded_part_file_is_removed_from_disk() { + let dir = scratch_dir("discard"); + let mut part = PartFile::open(&download_into(&dir, 8)).unwrap(); + part.write(b"data").unwrap(); + let path = dir.join("song.mp3.part"); + assert!(path.exists()); + + part.discard(); + + assert!(!path.exists()); + assert!(!dir.join("song.mp3").exists()); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn a_pause_is_reported_so_the_stall_budget_restarts() { + let dir = scratch_dir("paused"); + let mut download = download_into(&dir, 8); + download.status = DownloadStatus::Paused { + bytes_downloaded: 0, + total_bytes: 8, + }; + let context = Arc::new(RwLock::new(ClientContext::new())); + context.write().unwrap().add_download(download.clone()); + let resumer = { + let context = context.clone(); + std::thread::spawn(move || { + std::thread::sleep(std::time::Duration::from_millis(300)); + assert!( + context + .write() + .unwrap() + .downloads + .resume_by_file("peer", "song.mp3") + ); + }) + }; + + assert!(matches!( + DownloadPeer::wait_while_paused(&context, &download), + Ok(true) + )); + assert!(matches!( + DownloadPeer::wait_while_paused(&context, &download), + Ok(false) + )); + let _ = resumer.join(); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn a_cancelled_download_stops_the_transfer_loop() { + let dir = scratch_dir("cancelled"); + let mut download = download_into(&dir, 8); + download.status = DownloadStatus::Cancelled; + let context = Arc::new(RwLock::new(ClientContext::new())); + context.write().unwrap().add_download(download.clone()); + + assert!(matches!( + DownloadPeer::wait_while_paused(&context, &download), + Err(DownloadError::Cancelled) + )); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn a_status_after_cancel_is_not_relayed_on_the_channel() { + let dir = scratch_dir("relay"); + let (tx, rx) = mpsc::channel(); + let mut download = download_into(&dir, 8); + download.sender = tx; + download.status = DownloadStatus::Cancelled; + let context = Arc::new(RwLock::new(ClientContext::new())); + context.write().unwrap().add_download(download.clone()); + + DownloadPeer::send_download_status( + &context, + &download, + DownloadStatus::InProgress { + bytes_downloaded: 4, + total_bytes: 8, + speed_bytes_per_sec: 1.0, + }, + ); + + assert!(rx.try_recv().is_err(), "a cancelled row hears nothing more"); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn start_transfer_refuses_a_cancelled_download() { + let dir = scratch_dir("refuse"); + let mut download = download_into(&dir, 8); + download.status = DownloadStatus::Cancelled; + let context = Arc::new(RwLock::new(ClientContext::new())); + context.write().unwrap().add_download(download.clone()); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut stream = + std::net::TcpStream::connect(listener.local_addr().unwrap()) + .unwrap(); + let _peer = listener.accept().unwrap(); + + let outcome = + DownloadPeer::start_transfer(download, &mut stream, &context); + + assert!(matches!(outcome, Err(DownloadError::Cancelled))); + assert!(!dir.join("song.mp3.part").exists(), "no .part is opened"); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn a_directory_that_does_not_exist_yet_is_created_not_dodged() { let dir = scratch_dir("fresh").join("Soulseek"); diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 18cf6c4..a546d9b 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -8,7 +8,7 @@ use std::time::{Duration, Instant}; use crate::client::{Client, ClientContext, ClientOperation}; use crate::message::{Message, MessageReader}; -use crate::peer::{ConnectionType, DownloadPeer, Peer}; +use crate::peer::{ConnectionType, DownloadError, DownloadPeer, Peer}; use crate::types::Download; use crate::utils::lock::RwLockExt; use crate::utils::semaphore::{Permit, Semaphore}; @@ -140,6 +140,7 @@ fn wait_for_download_registration( DownloadStatus::InProgress { .. } | DownloadStatus::Paused { .. } | DownloadStatus::Completed + | DownloadStatus::Cancelled ) }) .cloned(), @@ -255,6 +256,7 @@ fn handle_file_connection( download.size, filename ); } + Err(DownloadError::Cancelled) => {} Err(e) => { error!( "Failed to download file from {}:{} (token: {}) - Error: {}", @@ -508,6 +510,23 @@ mod tests { } } + #[test] + fn a_cancelled_download_is_not_claimable() { + let client_context = Arc::new(RwLock::new(ClientContext::new())); + let mut download = registered_download(7); + download.status = DownloadStatus::Cancelled; + client_context.write().unwrap().add_download(download); + + let claimed = wait_for_download_registration( + &client_context, + 7, + "peer", + Instant::now() + Duration::from_millis(200), + ); + + assert!(claimed.is_none(), "a cancelled transfer must not restart"); + } + #[test] fn a_download_registered_during_the_wait_is_found() { let client_context = Arc::new(RwLock::new(ClientContext::new())); diff --git a/soulseek-rs-lib/src/peer/mod.rs b/soulseek-rs-lib/src/peer/mod.rs index eb80909..65d0323 100644 --- a/soulseek-rs-lib/src/peer/mod.rs +++ b/soulseek-rs-lib/src/peer/mod.rs @@ -6,7 +6,8 @@ pub mod upload_peer; pub use crate::actor::peer_actor::{PeerActor, PeerMessage}; pub use crate::actor::peer_registry::PeerRegistry; -pub use download_peer::DownloadPeer; +pub(crate) use download_peer::part_path_of; +pub use download_peer::{DownloadError, DownloadPeer}; use crate::message::Message; use core::fmt; diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index 4410f3a..7687b5a 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -186,12 +186,7 @@ pub struct Download { impl Download { #[must_use] pub const fn is_finished(&self) -> bool { - matches!( - self.status, - DownloadStatus::Completed - | DownloadStatus::Failed(_) - | DownloadStatus::TimedOut - ) + self.status.is_terminal() } #[must_use] @@ -234,11 +229,25 @@ pub enum DownloadStatus { total_bytes: u64, }, Completed, + Cancelled, /// Failed, optionally with a human-readable reason. Failed(Option), TimedOut, } +impl DownloadStatus { + #[must_use] + pub const fn is_terminal(&self) -> bool { + matches!( + self, + Self::Completed + | Self::Cancelled + | Self::Failed(_) + | Self::TimedOut + ) + } +} + /// Whether a user is reachable, as the server reports it. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum UserStatus { diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index aac3bd6..c5ba362 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -600,6 +600,76 @@ fn run_mock_uploader_f_first(cfg: &MockUpload) -> std::io::Result { .map_err(|_| std::io::Error::other("F connection thread panicked"))? } +fn run_mock_slow_uploader( + cfg: &MockUpload, + stall_after: Option, +) -> std::io::Result { + let mut p = open_control_and_await_queue(cfg)?; + p.write_all(&transfer_request(cfg).get_buffer())?; + p.flush()?; + expect_code(&mut p, 41, Duration::from_secs(10))?; + + let mut f = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?; + f.set_read_timeout(Some(Duration::from_secs(10)))?; + f.set_write_timeout(Some(Duration::from_secs(5)))?; + let mut init = peer_init_bytes(&cfg.peer_username, "F", cfg.token); + init.extend_from_slice(&cfg.token.to_le_bytes()); + f.write_all(&init)?; + let mut start = [0u8; 8]; + f.read_exact(&mut start)?; + + let mut sent = 0; + for (n, chunk) in cfg.content.chunks(4096).enumerate() { + if stall_after == Some(n) { + f.set_read_timeout(Some(Duration::from_secs(15)))?; + match f.read(&mut [0u8; 1]) { + Ok(0) => { + return Err(std::io::Error::other( + "the downloader hung up", + )); + } + Ok(_) => {} + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::WouldBlock + | std::io::ErrorKind::TimedOut + ) => {} + Err(e) => return Err(e), + } + } + f.write_all(chunk)?; + f.flush()?; + sent += chunk.len(); + std::thread::sleep(Duration::from_millis(20)); + } + Ok(sent) +} + +fn run_mock_uploader_offering_after( + cfg: &MockUpload, + queued: &Sender<()>, + go: &Receiver<()>, +) -> std::io::Result<(bool, String)> { + let mut p = open_control_and_await_queue(cfg)?; + let _ = queued.send(()); + go.recv() + .map_err(|_| std::io::Error::other("the test never said go"))?; + + p.write_all(&transfer_request(cfg).get_buffer())?; + p.flush()?; + let mut msg = expect_code(&mut p, 41, Duration::from_secs(10))?; + msg.set_pointer(8); + let _token = msg.read_int32(); + let allowed = msg.read_bool(); + let reason = if allowed { + String::new() + } else { + msg.read_string() + }; + Ok((allowed, reason)) +} + fn run_refusing_uploader( cfg: &MockUpload, refusal: &Message, @@ -843,6 +913,174 @@ fn a_download_completes_when_the_f_connection_beats_token_registration() { ); } +fn cancel_mid_transfer(peer_username: &str, stall_after: Option) { + let server = server_or_skip!(); + + let listen_port = free_port().expect("free listen port"); + let mut client = Client::with_settings(server.listening_settings( + &format!("{peer_username}_dl"), + "pw", + listen_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let filename = "cancel_me.mp3"; + let content: Vec = (0..400_000u32).map(|i| (i % 251) as u8).collect(); + let download_dir = unique_download_dir(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let cfg = MockUpload { + listen_addr: format!("127.0.0.1:{listen_port}"), + peer_username: peer_username.to_string(), + filename: filename.to_string(), + content: content.clone(), + token: 424_250_u32, + ready: ready_tx, + token_delay: Duration::ZERO, + }; + let uploader = + std::thread::spawn(move || run_mock_slow_uploader(&cfg, stall_after)); + ready_rx + .recv_timeout(Duration::from_secs(5)) + .expect("mock uploader P connection"); + std::thread::sleep(Duration::from_millis(1500)); + + let (_download, status_rx) = client + .download( + filename.to_string(), + peer_username.to_string(), + content.len() as u64, + download_dir.display().to_string(), + ) + .expect("start download"); + + let part = download_dir.join(format!("{filename}.part")); + assert!( + wait_for(|| part.exists()), + "the transfer should be streaming" + ); + if stall_after.is_some() { + std::thread::sleep(Duration::from_secs(2)); + } + let cancelled_at = Instant::now(); + assert!(client.cancel_download(peer_username, filename)); + + let deadline = Instant::now() + Duration::from_secs(10); + let mut cancelled = false; + while Instant::now() < deadline { + match status_rx.recv_timeout(Duration::from_millis(500)) { + Ok(DownloadStatus::Cancelled) => { + cancelled = true; + break; + } + Ok(DownloadStatus::Completed | DownloadStatus::Failed(_)) => break, + _ => {} + } + } + assert!(cancelled, "the status channel should report the cancel"); + assert!(wait_for(|| !part.exists()), "the partial file should go"); + assert!(!download_dir.join(filename).exists()); + assert!(client.get_all_downloads().iter().any(|d| { + d.filename == filename && matches!(d.status, DownloadStatus::Cancelled) + })); + + let outcome = uploader.join().expect("mock uploader thread"); + assert!( + outcome.is_err(), + "the uploader should see its connection drop, got {outcome:?}" + ); + assert!( + cancelled_at.elapsed() < Duration::from_secs(8), + "the socket should be released within seconds, not at the read \ + deadline" + ); + assert!( + status_rx.try_recv().is_err(), + "nothing follows Cancelled on the channel" + ); + let _ = std::fs::remove_dir_all(&download_dir); +} + +#[test] +fn a_cancelled_download_stops_and_drops_its_partial_file() { + cancel_mid_transfer("e2e_cancel_peer", None); +} + +#[test] +fn a_cancelled_stalled_download_is_released_promptly() { + cancel_mid_transfer("e2e_stalled_peer", Some(2)); +} + +#[test] +fn a_cancelled_queued_download_declines_the_peers_offer() { + let server = server_or_skip!(); + + let listen_port = free_port().expect("free listen port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_cancel_queued_dl", + "pw", + listen_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let filename = "never_starts.mp3"; + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let (queued_tx, queued_rx) = std::sync::mpsc::channel(); + let (go_tx, go_rx) = std::sync::mpsc::channel(); + let cfg = MockUpload { + listen_addr: format!("127.0.0.1:{listen_port}"), + peer_username: "e2e_cancel_queued_peer".to_string(), + filename: filename.to_string(), + content: vec![0; 10], + token: 424_251_u32, + ready: ready_tx, + token_delay: Duration::ZERO, + }; + let uploader = std::thread::spawn(move || { + run_mock_uploader_offering_after(&cfg, &queued_tx, &go_rx) + }); + ready_rx + .recv_timeout(Duration::from_secs(5)) + .expect("mock uploader P connection"); + std::thread::sleep(Duration::from_millis(1500)); + + let download_dir = unique_download_dir(); + let (_download, status_rx) = client + .download( + filename.to_string(), + "e2e_cancel_queued_peer".to_string(), + 10, + download_dir.display().to_string(), + ) + .expect("start download"); + queued_rx + .recv_timeout(Duration::from_secs(10)) + .expect("the peer should receive our queue request"); + + assert!(client.cancel_download("e2e_cancel_queued_peer", filename)); + assert!(matches!( + status_rx.recv_timeout(Duration::from_secs(2)), + Ok(DownloadStatus::Cancelled) + )); + + go_tx.send(()).expect("mock uploader waiting"); + let (allowed, reason) = uploader + .join() + .expect("mock uploader thread") + .expect("the peer should get an answer to its offer"); + assert!( + !allowed, + "a cancelled download must not accept the transfer" + ); + assert_eq!(reason, "Cancelled"); + assert!(client.get_all_downloads().iter().any(|d| { + d.filename == filename && matches!(d.status, DownloadStatus::Cancelled) + })); + assert!(!download_dir.join(format!("{filename}.part")).exists()); + let _ = std::fs::remove_dir_all(&download_dir); +} + fn refusal_fails_the_download_quickly( peer_username: &str, build_refusal: impl FnOnce(&str) -> Message, diff --git a/soulseek-rs/docs/openrpc.json b/soulseek-rs/docs/openrpc.json index e28e771..365ebe3 100644 --- a/soulseek-rs/docs/openrpc.json +++ b/soulseek-rs/docs/openrpc.json @@ -416,6 +416,18 @@ ], "type": "object" }, + { + "properties": { + "state": { + "const": "cancelled", + "type": "string" + } + }, + "required": [ + "state" + ], + "type": "object" + }, { "properties": { "reason": { @@ -1648,6 +1660,33 @@ }, "summary": "Drop a transfer that has not started." }, + { + "name": "download.cancel", + "paramStructure": "by-name", + "params": [ + { + "name": "filename", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "username", + "required": true, + "schema": { + "type": "string" + } + } + ], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/Ack" + } + }, + "summary": "Stop a download, queued or in progress; its partial file is deleted." + }, { "name": "download.set_dir", "paramStructure": "by-name", diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index af488ac..220d3e0 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -108,6 +108,11 @@ per finished transfer: `user`, `path`, `size`, `file` (the local path written). `.part` behind; re-running the same command resumes from it rather than starting over, so retrying a failed download is cheap. +**`transfer cancel `** — stop a download (queued or in progress; +its `.part` is deleted) or an upload being served. Prints nothing; exit 0 +is the confirmation, exit 4 means nothing by that name was in flight. Only +useful against a daemon, since a one-shot `download` ends with its own run. + **`get `** — search, pick, and download in one step. `--pick best|first|all`, `-n/--limit`, plus every filter `search` accepts. Use it when any copy will do. It ranks on free slot, bitrate and peer speed, so it cannot diff --git a/soulseek-rs/src/api.rs b/soulseek-rs/src/api.rs index b9001bd..5c242f5 100644 --- a/soulseek-rs/src/api.rs +++ b/soulseek-rs/src/api.rs @@ -116,6 +116,7 @@ pub trait SessionApi: Send + Sync { fn resume_download(&self, username: &str, filename: &str) -> bool; fn remove_queued_download(&self, username: &str, filename: &str) -> bool; fn remove_download(&self, username: &str, filename: &str) -> bool; + fn cancel_download(&self, username: &str, filename: &str) -> bool; fn uploads(&self) -> Vec; fn take_upload_events(&self) -> Vec; @@ -301,6 +302,10 @@ impl SessionApi for Client { Self::remove_download(self, username, filename) } + fn cancel_download(&self, username: &str, filename: &str) -> bool { + Self::cancel_download(self, username, filename) + } + fn uploads(&self) -> Vec { Self::uploads(self) } diff --git a/soulseek-rs/src/cli.rs b/soulseek-rs/src/cli.rs index e7ae4d6..7b96954 100644 --- a/soulseek-rs/src/cli.rs +++ b/soulseek-rs/src/cli.rs @@ -180,6 +180,10 @@ pub enum Commands { /// Download files by user and remote path Download(DownloadArgs), + /// Transfers in flight + #[command(subcommand)] + Transfer(TransferCommand), + /// Search, pick automatically, and download in one step Get(GetArgs), @@ -575,6 +579,18 @@ pub struct WishRunArgs { pub sort: SortKey, } +#[derive(Subcommand, Debug)] +pub enum TransferCommand { + /// Stop a download or upload; a cancelled download's partial file is + /// deleted + Cancel { + /// The other side of the transfer + user: String, + /// Remote path exactly as printed by `search`, `browse`, or `serve` + path: String, + }, +} + /// Watching only lasts as long as the session doing it, so these are aimed at /// a running daemon: it stays connected and keeps receiving the status pushes. #[derive(Subcommand, Debug)] diff --git a/soulseek-rs/src/commands/mod.rs b/soulseek-rs/src/commands/mod.rs index f678aae..27df78a 100644 --- a/soulseek-rs/src/commands/mod.rs +++ b/soulseek-rs/src/commands/mod.rs @@ -15,7 +15,7 @@ pub mod transfer; pub mod wish; use crate::api::SessionApi; -use crate::cli::{Commands, MessageCommand, RoomCommand}; +use crate::cli::{Commands, MessageCommand, RoomCommand, TransferCommand}; use crate::output::{CliError, CliResult, Exit, Out, PortmapRecord}; use crate::port_mapping::{self, PortMapper}; use crate::remote::{Endpoint, RemoteSession}; @@ -286,6 +286,9 @@ fn dispatch(ctx: &Ctx, command: Commands) -> CliResult { match command { Commands::Search(args) => transfer::search(ctx, &args), Commands::Download(args) => transfer::download(ctx, &args), + Commands::Transfer(TransferCommand::Cancel { user, path }) => { + transfer::cancel(ctx, &user, &path) + } Commands::Get(args) => transfer::get(ctx, &args), Commands::Browse(args) => social::browse(ctx, &args), Commands::Room(RoomCommand::List { timeout }) => { diff --git a/soulseek-rs/src/commands/transfer.rs b/soulseek-rs/src/commands/transfer.rs index 2a46b5d..bd37b5b 100644 --- a/soulseek-rs/src/commands/transfer.rs +++ b/soulseek-rs/src/commands/transfer.rs @@ -553,6 +553,9 @@ fn download_one( reason.unwrap_or_else(|| "the transfer failed".to_string()), )); } + Ok(DownloadStatus::Cancelled) => { + return Err(CliError::transfer("the transfer was cancelled")); + } Ok(DownloadStatus::Queued) => { out.status(&format!( "queued behind {}'s uploads", @@ -585,6 +588,19 @@ fn download_one( } } +pub fn cancel(ctx: &Ctx, user: &str, path: &str) -> CliResult { + let session = Session::open(ctx)?; + if session.client.cancel_download(user, path) + | session.client.cancel_upload(user, path) + { + ctx.out.status(&format!("cancelled {path} ({user})")); + return Ok(()); + } + Err(CliError::no_results(format!( + "no download or upload of {path} with {user}" + ))) +} + /// Parse `search`/`browse` output back into download requests. /// /// Both shapes this program emits are accepted: a JSON object per line, or diff --git a/soulseek-rs/src/daemon/mod.rs b/soulseek-rs/src/daemon/mod.rs index 046da2f..0575f39 100644 --- a/soulseek-rs/src/daemon/mod.rs +++ b/soulseek-rs/src/daemon/mod.rs @@ -469,16 +469,7 @@ fn save( let downloads: Vec = session .get_all_downloads() .iter() - .map(|download| PersistedDownload { - username: download.username.clone(), - filename: download.filename.clone(), - size: download.size, - download_directory: download.download_directory.clone(), - completed: matches!( - download.status, - soulseek_rs::DownloadStatus::Completed - ), - }) + .filter_map(PersistedDownload::capture) .collect(); let _ = store.save_downloads(&downloads); diff --git a/soulseek-rs/src/daemon/openrpc.rs b/soulseek-rs/src/daemon/openrpc.rs index f742340..59e19ee 100644 --- a/soulseek-rs/src/daemon/openrpc.rs +++ b/soulseek-rs/src/daemon/openrpc.rs @@ -44,6 +44,7 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value { | Method::DownloadResume | Method::DownloadRemove | Method::DownloadRemoveQueued + | Method::DownloadCancel | Method::UploadCancel => schema_of::(generator), Method::DownloadSetDir => schema_of::(generator), Method::UploadSlots => schema_of::(generator), @@ -108,6 +109,7 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value { | Method::DownloadResume | Method::DownloadRemove | Method::DownloadRemoveQueued + | Method::DownloadCancel | Method::DownloadSetDir | Method::UploadCancel | Method::UploadSlots @@ -162,6 +164,9 @@ fn summary(method: Method) -> &'static str { Method::DownloadResume => "Resume a paused transfer.", Method::DownloadRemove => "Forget a transfer entirely.", Method::DownloadRemoveQueued => "Drop a transfer that has not started.", + Method::DownloadCancel => { + "Stop a download, queued or in progress; its partial file is deleted." + } Method::DownloadSetDir => { "Land future transfers in a different folder on the daemon's host." } diff --git a/soulseek-rs/src/daemon/proto/dto.rs b/soulseek-rs/src/daemon/proto/dto.rs index c7ad6aa..304ecfb 100644 --- a/soulseek-rs/src/daemon/proto/dto.rs +++ b/soulseek-rs/src/daemon/proto/dto.rs @@ -111,6 +111,7 @@ pub enum DownloadStatusDto { total_bytes: u64, }, Completed, + Cancelled, Failed { reason: Option, }, @@ -138,6 +139,7 @@ impl From<&DownloadStatus> for DownloadStatusDto { total_bytes: *total_bytes, }, DownloadStatus::Completed => Self::Completed, + DownloadStatus::Cancelled => Self::Cancelled, DownloadStatus::Failed(reason) => Self::Failed { reason: reason.clone(), }, @@ -167,6 +169,7 @@ impl From for DownloadStatus { total_bytes, }, DownloadStatusDto::Completed => Self::Completed, + DownloadStatusDto::Cancelled => Self::Cancelled, DownloadStatusDto::Failed { reason } => Self::Failed(reason), DownloadStatusDto::TimedOut => Self::TimedOut, } diff --git a/soulseek-rs/src/daemon/proto/mod.rs b/soulseek-rs/src/daemon/proto/mod.rs index e705562..7a3fea7 100644 --- a/soulseek-rs/src/daemon/proto/mod.rs +++ b/soulseek-rs/src/daemon/proto/mod.rs @@ -90,6 +90,7 @@ wire_vocabulary! { DownloadResume => "download.resume", DownloadRemove => "download.remove", DownloadRemoveQueued => "download.remove_queued", + DownloadCancel => "download.cancel", DownloadSetDir => "download.set_dir", UploadList => "upload.list", UploadCancel => "upload.cancel", @@ -194,6 +195,7 @@ mod tests { total_bytes: 100, }, DownloadStatus::Completed, + DownloadStatus::Cancelled, DownloadStatus::Failed(Some("gone".into())), DownloadStatus::Failed(None), DownloadStatus::TimedOut, diff --git a/soulseek-rs/src/daemon/server/mod.rs b/soulseek-rs/src/daemon/server/mod.rs index 5694a59..37bd4e2 100644 --- a/soulseek-rs/src/daemon/server/mod.rs +++ b/soulseek-rs/src/daemon/server/mod.rs @@ -184,6 +184,15 @@ impl Daemon { ), }) } + Method::DownloadCancel => { + let transfer: TransferRef = parse(params)?; + ok(Ack { + ok: self.session.cancel_download( + &transfer.username, + &transfer.filename, + ), + }) + } Method::UploadList => ok(Uploads { uploads: self @@ -846,6 +855,9 @@ mod tests { fn cancel_upload(&self, _username: &str, _filename: &str) -> bool { false } + fn cancel_download(&self, _username: &str, _filename: &str) -> bool { + false + } fn set_upload_slots(&self, _slots: usize) {} fn check_privileges(&self) -> soulseek_rs::Result<()> { Ok(()) diff --git a/soulseek-rs/src/persist/snapshot.rs b/soulseek-rs/src/persist/snapshot.rs index b3b4452..554f3c8 100644 --- a/soulseek-rs/src/persist/snapshot.rs +++ b/soulseek-rs/src/persist/snapshot.rs @@ -6,7 +6,6 @@ use super::state::{PersistedDownload, PersistedMessage}; use crate::models::{ AppState, ChatMessage, MessageDirection, SearchEntry, SearchStatus, }; -use soulseek_rs::DownloadStatus; /// How much private-message history survives a restart. /// a flat cap on the whole log rather than per conversation; make @@ -34,16 +33,7 @@ impl Snapshot { let downloads = state .downloads .iter() - .map(|entry| PersistedDownload { - username: entry.download.username.clone(), - filename: entry.download.filename.clone(), - size: entry.download.size, - download_directory: entry.download.download_directory.clone(), - completed: matches!( - entry.download.status, - DownloadStatus::Completed - ), - }) + .filter_map(|entry| PersistedDownload::capture(&entry.download)) .collect(); let mut queries: Vec = Vec::new(); @@ -121,6 +111,7 @@ pub fn restore_searches(state: &mut AppState, queries: &[String]) { mod tests { use super::*; use crate::models::DownloadEntry; + use soulseek_rs::DownloadStatus; use soulseek_rs::types::{Download, DownloadMetadata}; use std::sync::atomic::AtomicBool; use std::sync::{Arc, mpsc}; @@ -188,6 +179,28 @@ mod tests { assert_eq!(snapshot.downloads[0].download_directory, "/music"); } + #[test] + fn a_cancelled_download_is_not_captured() { + let mut state = AppState::new(); + state + .downloads + .push(download("gone.mp3", DownloadStatus::Cancelled)); + state + .downloads + .push(download("queued.mp3", DownloadStatus::Queued)); + + let snapshot = Snapshot::capture(&state); + assert_eq!( + snapshot + .downloads + .iter() + .map(|d| d.filename.as_str()) + .collect::>(), + vec!["queued.mp3"], + "a cancel must not come back as a retry after a restart" + ); + } + #[test] fn capture_takes_queries_in_order_without_duplicates() { let mut state = AppState::new(); diff --git a/soulseek-rs/src/persist/state.rs b/soulseek-rs/src/persist/state.rs index 91836ef..ffb6c54 100644 --- a/soulseek-rs/src/persist/state.rs +++ b/soulseek-rs/src/persist/state.rs @@ -9,6 +9,7 @@ use color_eyre::Result; use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; +use soulseek_rs::{DownloadStatus, types::Download}; use std::path::{Path, PathBuf}; /// The shape every file is written in today. Bump it — and convert the older @@ -24,6 +25,22 @@ pub struct PersistedDownload { pub completed: bool, } +impl PersistedDownload { + #[must_use] + pub fn capture(download: &Download) -> Option { + if matches!(download.status, DownloadStatus::Cancelled) { + return None; + } + Some(Self { + username: download.username.clone(), + filename: download.filename.clone(), + size: download.size, + download_directory: download.download_directory.clone(), + completed: matches!(download.status, DownloadStatus::Completed), + }) + } +} + /// One line of private-message history. #[derive(Debug, Clone, PartialEq, Eq, Serialize, serde::Deserialize)] pub struct PersistedMessage { diff --git a/soulseek-rs/src/remote/mirror.rs b/soulseek-rs/src/remote/mirror.rs index 1f995f9..10472f4 100644 --- a/soulseek-rs/src/remote/mirror.rs +++ b/soulseek-rs/src/remote/mirror.rs @@ -106,12 +106,7 @@ impl Mirror { fn route(buffers: &mut Buffers, update: DownloadStatusEvent) { let key = (update.username, update.filename); let status = DownloadStatus::from(update.status); - let terminal = matches!( - status, - DownloadStatus::Completed - | DownloadStatus::Failed(_) - | DownloadStatus::TimedOut - ); + let terminal = status.is_terminal(); let Some(sender) = buffers.downloads.get(&key) else { return; }; diff --git a/soulseek-rs/src/remote/mod.rs b/soulseek-rs/src/remote/mod.rs index 2d32f66..435d4b0 100644 --- a/soulseek-rs/src/remote/mod.rs +++ b/soulseek-rs/src/remote/mod.rs @@ -639,6 +639,10 @@ impl SessionApi for RemoteSession { self.ask(Method::DownloadRemove, Self::transfer(username, filename)) } + fn cancel_download(&self, username: &str, filename: &str) -> bool { + self.ask(Method::DownloadCancel, Self::transfer(username, filename)) + } + fn uploads(&self) -> Vec { self.request::<_, Uploads>(Method::UploadList, ()) .map(|listed| listed.uploads.into_iter().map(Into::into).collect()) diff --git a/soulseek-rs/src/ui/main_tui/downloads.rs b/soulseek-rs/src/ui/main_tui/downloads.rs index 5fefec1..620dcde 100644 --- a/soulseek-rs/src/ui/main_tui/downloads.rs +++ b/soulseek-rs/src/ui/main_tui/downloads.rs @@ -1,18 +1,10 @@ use super::MainTui; use crate::models::{DownloadEntry, FileDisplayData}; +use crate::ui::panes::{SelectedTransfer, selected_transfer}; +use soulseek_rs::types::UploadStatus; use soulseek_rs::{DownloadStatus, types::Download}; use std::{sync::mpsc, thread}; -/// Done, given up, or ran out of time: the row is all that is left. -const fn is_finished(download: &Download) -> bool { - matches!( - download.status, - DownloadStatus::Completed - | DownloadStatus::Failed(_) - | DownloadStatus::TimedOut - ) -} - impl MainTui { /// Channel for queued downloads, created on first use. pub(super) fn downloads_sender( @@ -73,7 +65,7 @@ impl MainTui { // session holds, so dropping them only here would last until the next // poll put them straight back. for entry in &self.state.downloads { - if is_finished(&entry.download) { + if entry.download.is_finished() { self.client.remove_download( &entry.download.username, &entry.download.filename, @@ -82,7 +74,7 @@ impl MainTui { } self.state .downloads - .retain(|entry| !is_finished(&entry.download)); + .retain(|entry| !entry.download.is_finished()); let len = self.state.downloads.len(); if len == 0 { self.state.downloads_table_state.select(None); @@ -139,7 +131,7 @@ impl MainTui { // A finished transfer may exist only in this window (restored from // an earlier run), so the session's "nothing to forget" is not a // reason to keep the row. - DownloadStatus::Completed => { + DownloadStatus::Completed | DownloadStatus::Cancelled => { self.client .remove_download(&download.username, &download.filename); true @@ -214,29 +206,43 @@ impl MainTui { self.state.results_selected_indices.clear(); } - /// Cancel the selected transfer when it is an upload row (uploads are - /// listed after the downloads in the shared pane). - pub(super) fn cancel_selected_upload(&self) { - let Some(index) = self.state.downloads_table_state.selected() else { - return; - }; - let Some(upload) = self - .state - .uploads - .get(index.wrapping_sub(self.state.downloads.len())) - else { - return; - }; - if upload.status == soulseek_rs::types::UploadStatus::InProgress - && !self - .client - .cancel_upload(&upload.username, &upload.filename) - { - soulseek_rs::warn!( - "No in-progress upload of {} to {} to cancel", - upload.filename, - upload.username - ); + pub(super) fn cancel_selected_transfer(&self) { + let selected = selected_transfer( + self.state.downloads_table_state.selected(), + &self.state.downloads, + &self.state.uploads, + ); + match selected { + Some(SelectedTransfer::Download(entry)) + if !entry.download.is_finished() => + { + let download = &entry.download; + if !self + .client + .cancel_download(&download.username, &download.filename) + { + soulseek_rs::warn!( + "No download of {} from {} to cancel", + download.filename, + download.username + ); + } + } + Some(SelectedTransfer::Upload(upload)) + if upload.status == UploadStatus::InProgress => + { + let cancelled = self + .client + .cancel_upload(&upload.username, &upload.filename); + if !cancelled { + soulseek_rs::warn!( + "No in-progress upload of {} to {} to cancel", + upload.filename, + upload.username + ); + } + } + _ => {} } } diff --git a/soulseek-rs/src/ui/main_tui/input.rs b/soulseek-rs/src/ui/main_tui/input.rs index 4c320b4..4c5bcce 100644 --- a/soulseek-rs/src/ui/main_tui/input.rs +++ b/soulseek-rs/src/ui/main_tui/input.rs @@ -422,7 +422,7 @@ impl MainTui { cycle(&mut self.state.downloads_table_state, rows, true); } KeyCode::Char('x') => { - self.cancel_selected_upload(); + self.cancel_selected_transfer(); } KeyCode::Char('p') => { self.toggle_selected_download_pause(); diff --git a/soulseek-rs/src/ui/main_tui/mod.rs b/soulseek-rs/src/ui/main_tui/mod.rs index 61ccb06..ec02f20 100644 --- a/soulseek-rs/src/ui/main_tui/mod.rs +++ b/soulseek-rs/src/ui/main_tui/mod.rs @@ -289,6 +289,8 @@ mod tests { download_dir_set: std::sync::Mutex>, /// What `set_shared_directories` was last asked for, if anything. shares_set: std::sync::Mutex>>, + download_cancelled: std::sync::Mutex>, + upload_cancelled: std::sync::Mutex>, } fn queued(username: &str, filename: &str) -> soulseek_rs::types::Download { @@ -432,8 +434,15 @@ mod tests { fn take_upload_events(&self) -> Vec { Vec::new() } - fn cancel_upload(&self, _u: &str, _f: &str) -> bool { - false + fn cancel_upload(&self, u: &str, f: &str) -> bool { + *self.upload_cancelled.lock().expect("not poisoned") = + Some((u.to_string(), f.to_string())); + true + } + fn cancel_download(&self, u: &str, f: &str) -> bool { + *self.download_cancelled.lock().expect("not poisoned") = + Some((u.to_string(), f.to_string())); + true } fn set_upload_slots(&self, _slots: usize) {} fn check_privileges(&self) -> soulseek_rs::Result<()> { @@ -558,6 +567,106 @@ mod tests { } } + #[test] + fn x_cancels_the_highlighted_download() { + let session = Arc::new(TalkativeSession::default()); + let mut tui = attach(session.clone()); + tui.state.downloads.push(crate::models::DownloadEntry { + download: at_status( + "bob", + "song.mp3", + soulseek_rs::DownloadStatus::InProgress { + bytes_downloaded: 1, + total_bytes: 4096, + speed_bytes_per_sec: 1.0, + }, + ), + receiver: None, + }); + tui.state.downloads_table_state.select(Some(0)); + tui.state.focused_pane = FocusedPane::Downloads; + + tui.handle_key_event(key('x')); + + assert_eq!( + *session.download_cancelled.lock().expect("not poisoned"), + Some(("bob".to_string(), "song.mp3".to_string())) + ); + assert_eq!( + *session.upload_cancelled.lock().expect("not poisoned"), + None + ); + } + + #[test] + fn d_clears_a_cancelled_row() { + let mut tui = with_session(TalkativeSession::default()); + tui.state.downloads.push(crate::models::DownloadEntry { + download: at_status( + "bob", + "gone.mp3", + soulseek_rs::DownloadStatus::Cancelled, + ), + receiver: None, + }); + tui.state.downloads_table_state.select(Some(0)); + tui.state.focused_pane = FocusedPane::Downloads; + + tui.handle_key_event(key('d')); + + assert!(tui.state.downloads.is_empty(), "the row goes"); + } + + #[test] + fn x_on_a_finished_download_asks_nothing() { + let session = Arc::new(TalkativeSession::default()); + let mut tui = attach(session.clone()); + tui.state.downloads.push(crate::models::DownloadEntry { + download: completed("bob", "done.mp3"), + receiver: None, + }); + tui.state.downloads_table_state.select(Some(0)); + tui.state.focused_pane = FocusedPane::Downloads; + + tui.handle_key_event(key('x')); + + assert_eq!( + *session.download_cancelled.lock().expect("not poisoned"), + None + ); + } + + #[test] + fn x_on_an_upload_row_still_cancels_the_upload() { + let session = Arc::new(TalkativeSession::default()); + let mut tui = attach(session.clone()); + tui.state.downloads.push(crate::models::DownloadEntry { + download: queued("bob", "song.mp3"), + receiver: None, + }); + tui.state.uploads.push(soulseek_rs::types::UploadInfo { + username: "alice".to_string(), + filename: "served.mp3".to_string(), + size: 10, + bytes_sent: 1, + speed_bytes_per_sec: 1.0, + status: soulseek_rs::types::UploadStatus::InProgress, + }); + tui.state.downloads_table_state.select(Some(1)); + tui.state.focused_pane = FocusedPane::Downloads; + + tui.handle_key_event(key('x')); + + assert_eq!( + *session.upload_cancelled.lock().expect("not poisoned"), + Some(("alice".to_string(), "served.mp3".to_string())) + ); + assert_eq!( + *session.download_cancelled.lock().expect("not poisoned"), + None + ); + } + fn with_session(session: TalkativeSession) -> MainTui { attach(Arc::new(session)) } diff --git a/soulseek-rs/src/ui/main_tui/render.rs b/soulseek-rs/src/ui/main_tui/render.rs index 9e7c42e..4a1f25c 100644 --- a/soulseek-rs/src/ui/main_tui/render.rs +++ b/soulseek-rs/src/ui/main_tui/render.rs @@ -392,6 +392,7 @@ impl MainTui { FocusedPane::Downloads => { vec![ ("p", "pause/resume"), + ("x", "cancel"), ("r", "retry failed"), ("d", "delete queued/done"), ("c", "clear finished"), diff --git a/soulseek-rs/src/ui/panes/download_info_pane.rs b/soulseek-rs/src/ui/panes/download_info_pane.rs index 408a9f1..616504d 100644 --- a/soulseek-rs/src/ui/panes/download_info_pane.rs +++ b/soulseek-rs/src/ui/panes/download_info_pane.rs @@ -177,6 +177,9 @@ fn build_info_lines( } DownloadStatus::Paused { .. } => ("Paused".to_string(), info_style()), DownloadStatus::Completed => ("Completed".to_string(), success_style()), + DownloadStatus::Cancelled => { + ("Cancelled".to_string(), inactive_style()) + } DownloadStatus::Failed(_) => ("Failed".to_string(), error_style()), DownloadStatus::TimedOut => ("Timed out".to_string(), error_style()), }; @@ -271,7 +274,9 @@ fn build_info_lines( ))); } } - DownloadStatus::Completed | DownloadStatus::TimedOut => {} + DownloadStatus::Completed + | DownloadStatus::Cancelled + | DownloadStatus::TimedOut => {} } lines diff --git a/soulseek-rs/src/ui/panes/downloads_pane.rs b/soulseek-rs/src/ui/panes/downloads_pane.rs index 920d7fa..493b6e8 100644 --- a/soulseek-rs/src/ui/panes/downloads_pane.rs +++ b/soulseek-rs/src/ui/panes/downloads_pane.rs @@ -81,6 +81,7 @@ pub fn render_downloads_pane( ) } DownloadStatus::Completed => "Completed".to_string(), + DownloadStatus::Cancelled => "Cancelled".to_string(), DownloadStatus::Failed(_) => "Failed".to_string(), DownloadStatus::TimedOut => "Timed out".to_string(), }; @@ -97,7 +98,9 @@ pub fn render_downloads_pane( DownloadStatus::Failed(_) | DownloadStatus::TimedOut => { error_style() } - DownloadStatus::Queued => inactive_style(), + DownloadStatus::Queued | DownloadStatus::Cancelled => { + inactive_style() + } _ => primary_style(), }; diff --git a/soulseek-rs/src/ui/panes/mod.rs b/soulseek-rs/src/ui/panes/mod.rs index 32af3e8..2e6c678 100644 --- a/soulseek-rs/src/ui/panes/mod.rs +++ b/soulseek-rs/src/ui/panes/mod.rs @@ -8,7 +8,9 @@ mod searches_pane; pub use browse_pane::render_browse_pane; pub use chat_pane::render_chat_pane; -pub use download_info_pane::{render_download_info_pane, selected_transfer}; +pub use download_info_pane::{ + SelectedTransfer, render_download_info_pane, selected_transfer, +}; pub use downloads_pane::render_downloads_pane; pub use results_pane::{ResultsPaneParams, render_results_pane}; pub use rooms_pane::render_rooms_pane; diff --git a/soulseek-rs/src/ui/styles.rs b/soulseek-rs/src/ui/styles.rs index 503866e..2df9ab5 100644 --- a/soulseek-rs/src/ui/styles.rs +++ b/soulseek-rs/src/ui/styles.rs @@ -152,6 +152,7 @@ pub fn download_status_glyph(status: &DownloadStatus) -> (&'static str, Style) { DownloadStatus::InProgress { .. } => (GLYPH_ACTIVE, accent_style()), DownloadStatus::Paused { .. } => (GLYPH_PAUSED, info_style()), DownloadStatus::Completed => (GLYPH_DONE, success_style()), + DownloadStatus::Cancelled => (GLYPH_FAILED, inactive_style()), DownloadStatus::Failed(_) => (GLYPH_FAILED, error_style()), DownloadStatus::TimedOut => (GLYPH_TIMED_OUT, error_style()), } @@ -237,6 +238,7 @@ mod tests { (DownloadStatus::Completed, GLYPH_DONE, PHOSPHOR), (DownloadStatus::Failed(None), GLYPH_FAILED, ALARM), (DownloadStatus::TimedOut, GLYPH_TIMED_OUT, ALARM), + (DownloadStatus::Cancelled, GLYPH_FAILED, COLOR_INACTIVE), ] { let (got_glyph, style) = download_status_glyph(&status); assert_eq!(got_glyph, glyph); diff --git a/soulseek-rs/tests/cli_e2e.rs b/soulseek-rs/tests/cli_e2e.rs index d295c5f..3086a6c 100644 --- a/soulseek-rs/tests/cli_e2e.rs +++ b/soulseek-rs/tests/cli_e2e.rs @@ -27,6 +27,7 @@ const EXIT_OK: i32 = 0; const EXIT_USAGE: i32 = 2; const EXIT_CONNECTION: i32 = 3; const EXIT_NO_RESULTS: i32 = 4; +const EXIT_TRANSFER: i32 = 6; const EXIT_SESSION_LOST: i32 = 7; /// Build a command for the binary, isolated from the developer's environment @@ -1554,6 +1555,79 @@ fn browse_lists_shares_as_paths_download_understands() { assert_eq!(fields[1], probe_bytes().len().to_string()); } +#[test] +fn transfer_cancel_without_a_matching_transfer_is_the_no_results_verdict() { + let server = server_or_skip!(); + let output = cli( + &server, + "cli_e2e_canceller", + &["transfer", "cancel", "nobody", "nothing.mp3"], + ); + assert_eq!( + code(&output), + EXIT_NO_RESULTS, + "stderr: {}", + stderr(&output) + ); + assert!( + stderr(&output).contains("nothing.mp3"), + "the verdict names the transfer: {}", + stderr(&output) + ); +} + +#[test] +fn transfer_cancel_stops_a_download_another_client_is_waiting_on() { + if mode() != Mode::Daemon { + println!("skipped: a one-shot download owns its own session"); + return; + } + let server = server_or_skip!(); + let mut all = server.args("cli_e2e_cancel_waiter"); + all.extend( + [ + "download", + "e2e_ghost_peer", + "@@ghost\\never.mp3", + "--size", + "10", + "--timeout", + "60", + ] + .iter() + .map(|a| (*a).to_string()), + ); + let refs: Vec<&str> = all.iter().map(String::as_str).collect(); + let waiting = command(&refs) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("the binary should start"); + + let deadline = Instant::now() + Duration::from_secs(15); + let mut cancelled = None; + while Instant::now() < deadline { + let output = cli( + &server, + "cli_e2e_cancel_waiter", + &["transfer", "cancel", "e2e_ghost_peer", "@@ghost\\never.mp3"], + ); + if code(&output) == EXIT_OK { + cancelled = Some(output); + break; + } + std::thread::sleep(Duration::from_millis(500)); + } + assert!( + cancelled.is_some(), + "the queued transfer should be cancellable" + ); + + let waited = waiting.wait_with_output().expect("wait"); + assert_eq!(code(&waited), EXIT_TRANSFER, "stderr: {}", stderr(&waited)); + assert!(stderr(&waited).contains("cancelled"), "{}", stderr(&waited)); +} + #[test] fn download_fetches_a_file_and_prints_where_it_landed() { let server = server_or_skip!(); diff --git a/soulseek-rs/tests/daemon_protocol_e2e.rs b/soulseek-rs/tests/daemon_protocol_e2e.rs index 49b1ec0..88e7bca 100644 --- a/soulseek-rs/tests/daemon_protocol_e2e.rs +++ b/soulseek-rs/tests/daemon_protocol_e2e.rs @@ -905,6 +905,48 @@ fn a_wishlist_search_is_reported_as_just_started() { ); } +#[test] +fn a_queued_download_is_cancelled_and_listed_as_such() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("tcp"); + connection + .authenticate(Some(&daemon.token())) + .expect("an answer"); + let started = connection + .call( + r#"{"jsonrpc":"2.0","id":2,"method":"download.start","params":{"username":"e2e_ghost_peer","filename":"@@ghost\\never.mp3","size":10}}"#, + ) + .expect("an answer"); + assert!(!started.contains("error"), "got {started}"); + let reply = connection + .call( + r#"{"jsonrpc":"2.0","id":3,"method":"download.cancel","params":{"username":"e2e_ghost_peer","filename":"@@ghost\\never.mp3"}}"#, + ) + .expect("an answer"); + assert!(reply.contains(r#""ok":true"#), "got {reply}"); + let listed = connection + .call( + r#"{"jsonrpc":"2.0","id":4,"method":"download.list","params":{}}"#, + ) + .expect("an answer"); + assert!(listed.contains(r#""state":"cancelled""#), "got {listed}"); +} + +#[test] +fn cancelling_an_unknown_download_is_answered_with_a_plain_no() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("tcp"); + connection + .authenticate(Some(&daemon.token())) + .expect("an answer"); + let reply = connection + .call( + r#"{"jsonrpc":"2.0","id":2,"method":"download.cancel","params":{"username":"nobody","filename":"nothing.mp3"}}"#, + ) + .expect("an answer"); + assert!(reply.contains(r#""ok":false"#), "got {reply}"); +} + /// The upload-slot limit is retunable on a live session. #[test] fn the_upload_slot_limit_can_be_retuned_live() { diff --git a/website/src/pages/docs.tsx b/website/src/pages/docs.tsx index a81aa33..6141138 100644 --- a/website/src/pages/docs.tsx +++ b/website/src/pages/docs.tsx @@ -53,7 +53,7 @@ const PANES: readonly { num?: number; title: string; body: string }[] = [ { num: 3, title: 'Downloads / Uploads', - body: 'Live transfers in both directions with progress and speed: queued, active, complete, failed.', + body: 'Live transfers in both directions with progress and speed: queued, active, complete, cancelled, failed. x cancels the selected transfer.', }, { title: 'Info', @@ -255,6 +255,8 @@ const Daemon = () => ( { t: 'cm', text: "# opt out; open this run's own session" }, { t: 'cmd', text: 'soulseek-rs' }, { t: 'cm', text: '# the TUI attaches to the same session' }, + { t: 'cmd', text: 'soulseek-rs transfer cancel ""' }, + { t: 'cm', text: '# stop a download the daemon is running' }, { t: 'cmd', text: 'soulseek-rs daemon status' }, { t: 'cmd', text: 'soulseek-rs daemon stop' }, ]}