From 543e9ee865ea8994860d493aeeb616b93d8bbd43 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 17 Aug 2026 09:22:03 +0200 Subject: [PATCH 1/4] fix(peer): wait for the transfer token before the download lookup The listener looked the download up immediately after the PeerInit(F), but the 4-byte transfer token routinely arrives in a separate TCP segment, so every firewalled peer download failed with "Download info missing". Wait for the token within the handshake deadline and give up on the connection when it never arrives. Also quiets the dispatcher's unhandled-message-code log to debug; the server routinely sends codes the library doesn't implement. Fixes #41 --- soulseek-rs-lib/src/dispatcher.rs | 4 +- soulseek-rs-lib/src/peer/listen.rs | 97 +++++++++++++++++++++++++++--- soulseek-rs-lib/tests/e2e.rs | 45 ++++++++++---- 3 files changed, 124 insertions(+), 22 deletions(-) diff --git a/soulseek-rs-lib/src/dispatcher.rs b/soulseek-rs-lib/src/dispatcher.rs index a034bd9..9c50d80 100644 --- a/soulseek-rs-lib/src/dispatcher.rs +++ b/soulseek-rs-lib/src/dispatcher.rs @@ -1,7 +1,7 @@ use crate::message::{Message, handlers::Handlers}; use std::sync::mpsc::Sender; -use crate::warn; +use crate::debug; pub struct MessageDispatcher { owner_name: String, @@ -30,7 +30,7 @@ impl MessageDispatcher { message.set_pointer(8); handler.handle(message, self.sender.clone()); } else { - warn!( + debug!( "[{}:dispatcher] No handler found for message code: {code}", self.owner_name ); diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 55e8d25..3536be2 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -96,12 +96,32 @@ fn parse_peer_init_message(mut message: Message) -> Option { }) } +fn wait_for_transfer_token( + stream: &mut TcpStream, + reader: &mut MessageReader, + deadline: Instant, +) -> io::Result<()> { + while reader.buffer_len() < 4 { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "transfer token deadline passed", + )); + } + stream.set_read_timeout(Some(remaining))?; + reader.read_from_socket(stream)?; + } + stream.set_read_timeout(None)?; + Ok(()) +} + fn extract_download_from_buffer( reader: &mut MessageReader, client_context: &Arc>, username: &str, peer_ip: &str, - peer_port: u16, + peer_port: u32, ) -> Option { if reader.buffer_len() == 0 { return None; @@ -161,24 +181,32 @@ fn handle_peer_connection( fn handle_file_connection( peer: Peer, - stream: TcpStream, + mut stream: TcpStream, mut reader: MessageReader, token: u32, context: &ConnectionContext, - peer_ip: &str, - peer_port: u16, + deadline: Instant, ) { trace!( "[client] DownloadFromPeer token: {} peer: {:?}", token, peer ); + if let Err(e) = wait_for_transfer_token(&mut stream, &mut reader, deadline) + { + debug!( + "[listener:{}:{}] no transfer token arrived: {e}", + peer.host, peer.port + ); + return; + } + let download = extract_download_from_buffer( &mut reader, &context.client_context, &peer.username, - peer_ip, - peer_port, + &peer.host, + peer.port, ); let failure_token = download.as_ref().map(|d| d.token); @@ -368,8 +396,7 @@ fn handle_incoming_connection( reader, init_data.token, &context, - &peer_ip, - peer_port, + deadline, ), ConnectionType::D => { debug!( @@ -482,6 +509,60 @@ mod tests { let _ = writer.join(); } + #[test] + fn the_transfer_token_is_waited_for_across_segments() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let writer = std::thread::spawn(move || { + let mut socket = TcpStream::connect(addr).unwrap(); + std::thread::sleep(Duration::from_millis(100)); + socket.write_all(&42u32.to_le_bytes()).unwrap(); + std::thread::sleep(Duration::from_millis(200)); + }); + let (mut stream, _) = listener.accept().unwrap(); + let mut reader = MessageReader::new(); + + let result = wait_for_transfer_token( + &mut stream, + &mut reader, + Instant::now() + Duration::from_secs(5), + ); + + assert!( + result.is_ok(), + "token in a later segment must be waited for" + ); + assert!(reader.buffer_len() >= 4, "token bytes must be buffered"); + let _ = writer.join(); + } + + #[test] + fn waiting_for_a_transfer_token_stops_at_the_deadline() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let writer = std::thread::spawn(move || { + let socket = TcpStream::connect(addr).unwrap(); + std::thread::sleep(Duration::from_millis(500)); + drop(socket); + }); + let (mut stream, _) = listener.accept().unwrap(); + let mut reader = MessageReader::new(); + + let started = Instant::now(); + let result = wait_for_transfer_token( + &mut stream, + &mut reader, + started + Duration::from_millis(200), + ); + + assert!(result.is_err(), "a silent peer must not park the thread"); + assert!( + started.elapsed() < Duration::from_secs(5), + "the deadline must bound the wait" + ); + let _ = writer.join(); + } + #[test] fn an_oversized_handshake_is_rejected() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index ab48b13..39c3433 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -478,6 +478,7 @@ struct MockUpload { content: Vec, token: u32, ready: Sender<()>, + token_delay: Duration, } /// Build a `PeerInit` (peer code 1) frame: `[len][1][username][conn_type][token]`. @@ -562,6 +563,7 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result { &cfg.peer_username, cfg.token, &cfg.content, + cfg.token_delay, ) } @@ -589,10 +591,8 @@ fn run_refusing_uploader( } /// Open an F (file transfer) connection to `downloader_addr` and stream -/// `content`. PeerInit(F) and the 4-byte transfer token are written together so -/// the token lands in the listener's read buffer, where the download is looked -/// up by token; the downloader then sends an 8-byte START_DOWNLOAD offset before -/// we send the bytes. +/// `content`; the downloader sends an 8-byte START_DOWNLOAD offset before we +/// send the bytes. /// /// Like a real peer, only the bytes past that offset are sent. Returns the /// offset the downloader asked for. @@ -601,12 +601,19 @@ fn serve_file_over_f( username: &str, token: u32, content: &[u8], + token_delay: Duration, ) -> std::io::Result { let mut f = connect_retry(downloader_addr, Duration::from_secs(5))?; f.set_read_timeout(Some(Duration::from_secs(10)))?; - let mut init = peer_init_bytes(username, "F", token); - init.extend_from_slice(&token.to_le_bytes()); - f.write_all(&init)?; + if token_delay.is_zero() { + let mut init = peer_init_bytes(username, "F", token); + init.extend_from_slice(&token.to_le_bytes()); + f.write_all(&init)?; + } else { + f.write_all(&peer_init_bytes(username, "F", token))?; + std::thread::sleep(token_delay); + f.write_all(&token.to_le_bytes())?; + } f.flush()?; let mut start = [0u8; 8]; @@ -717,14 +724,13 @@ fn unique_download_dir() -> PathBuf { dir } -#[test] -fn a_file_downloads_from_a_peer_over_p_and_f_connections() { +fn file_downloads_over_p_and_f(peer_username: &str, token_delay: Duration) { let server = server_or_skip!(); // Downloader: connected to the server with its peer listener enabled. let listen_port = free_port().expect("free listen port"); let mut client = Client::with_settings(server.listening_settings( - "e2e_downloader", + &format!("{peer_username}_dl"), "pw", listen_port, )); @@ -741,11 +747,12 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() { let (ready_tx, ready_rx) = std::sync::mpsc::channel(); let cfg = MockUpload { listen_addr: format!("127.0.0.1:{listen_port}"), - peer_username: "e2e_mockpeer".to_string(), + peer_username: peer_username.to_string(), filename: filename.to_string(), content: content.clone(), token, ready: ready_tx, + token_delay, }; let uploader = std::thread::spawn(move || { if let Err(e) = run_mock_uploader(&cfg) { @@ -764,7 +771,7 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() { let (_download, status_rx) = client .download( filename.to_string(), - "e2e_mockpeer".to_string(), + peer_username.to_string(), size, download_dir.display().to_string(), ) @@ -783,6 +790,16 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() { let _ = std::fs::remove_dir_all(&download_dir); } +#[test] +fn a_file_downloads_from_a_peer_over_p_and_f_connections() { + file_downloads_over_p_and_f("e2e_mockpeer", Duration::ZERO); +} + +#[test] +fn a_file_downloads_when_the_transfer_token_arrives_late() { + file_downloads_over_p_and_f("e2e_latepeer", Duration::from_millis(300)); +} + fn refusal_fails_the_download_quickly( peer_username: &str, build_refusal: impl FnOnce(&str) -> Message, @@ -808,6 +825,7 @@ fn refusal_fails_the_download_quickly( content: Vec::new(), token: 0, ready: ready_tx, + token_delay: Duration::ZERO, }; let uploader = std::thread::spawn(move || { if let Err(e) = run_refusing_uploader(&cfg, &refusal) { @@ -905,6 +923,7 @@ fn an_interrupted_download_resumes_from_its_partial_file() { content: content.clone(), token: 424_243_u32, ready: ready_tx, + token_delay: Duration::ZERO, }; let (offset_tx, offset_rx) = std::sync::mpsc::channel(); let uploader = std::thread::spawn(move || match run_mock_uploader(&cfg) { @@ -1122,6 +1141,7 @@ fn run_mock_direct_peer(cfg: &MockDirectUpload) -> std::io::Result<()> { &cfg.username, cfg.token, &cfg.content, + Duration::ZERO, ) .map(|_| ()) } @@ -1269,6 +1289,7 @@ fn run_mock_firewalled_peer(cfg: &MockFirewalledUpload) -> std::io::Result<()> { &cfg.username, cfg.token, &cfg.content, + Duration::ZERO, ) .map(|_| ()) } From ac06cb9a5abd51b5db90d4f0ff923340bc679b91 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 17 Aug 2026 09:41:33 +0200 Subject: [PATCH 2/4] feat(lib): reserve client minor versions 1 (unidentified) and 2 (reference client) Document the minor-version registry on ClientVersion: within the soulseek-rs major (176), minor 1 stays the library default for unidentified consumers and minor 2 is now the reference CLI/TUI client. Downstream clients reserve their own minor via issue #12. --- soulseek-rs-lib/src/client/mod.rs | 5 +++-- soulseek-rs-lib/src/types.rs | 18 +++++++++++++++--- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index dd9cf09..b0fd64e 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -171,8 +171,9 @@ pub struct ClientSettings { /// Empty means nothing is shared. pub shared_directories: Vec, /// The version reported to the server on login. Defaults to the - /// soulseek-rs major version with minor version 1; clients and bots built - /// on this library should pick their own minor version. + /// soulseek-rs major version with minor version 1 ("unidentified"); + /// clients and bots built on this library should reserve their own minor + /// version (see [`ClientVersion`]). pub version: ClientVersion, } diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index fafdf8e..4ad4c7d 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -15,6 +15,18 @@ pub struct File { /// Major versions identify client projects on the Soulseek network and are /// reserved per project for the project's lifetime: /// . +/// +/// Within the soulseek-rs major, minor versions identify the individual +/// clients built on this library. Reserved minors: +/// +/// | Minor | Client | +/// |-------|-----------------------------------------------| +/// | 1 | unidentified (library default) | +/// | 2 | soulseek-rs reference client (CLI/TUI) | +/// +/// Building on this library? Keep [`ClientVersion::MAJOR`] and pick an +/// unused minor for your client by opening an issue: +/// . #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct ClientVersion { pub major: u32, @@ -23,14 +35,14 @@ pub struct ClientVersion { impl ClientVersion { /// The major version reserved for the soulseek-rs project. Clients and - /// bots built on this library keep this major and pick their own non-zero - /// minor version. + /// bots built on this library keep this major and pick their own + /// reserved minor version (see the table on [`ClientVersion`]). pub const MAJOR: u32 = 176; /// The version the soulseek-rs reference client (CLI/TUI) logs in with. pub const REFERENCE_CLIENT: Self = Self { major: Self::MAJOR, - minor: 100, + minor: 2, }; } From 1611b4baf3815185a6cd70270dbd82df73ca0179 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 17 Aug 2026 09:43:45 +0200 Subject: [PATCH 3/4] docs: list projects built on soulseek-rs-lib Add a 'Projects using soulseek-rs-lib' section to the README and a matching section on the website community page, starting with seakarr. Both point new projects at issue #12 to reserve a client minor version. --- README.md | 9 +++++++++ website/src/lib/links.ts | 2 ++ website/src/pages/community.tsx | 25 +++++++++++++++++++++++++ 3 files changed, 36 insertions(+) diff --git a/README.md b/README.md index ae21a75..b0d0816 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,15 @@ implementation for anyone building their own client, and **soulseek-rs**, the client built on it. The library stays lean on dependencies and has none today; the client takes them freely. +### Projects using soulseek-rs-lib + +- [seakarr](https://github.com/binhex/seakarr) — automated Soulseek music + downloader with library quality upgrading. + +Built something on the library? Open a PR to add it here, and reserve a client +minor version in [#12](https://github.com/michel/soulseek-rs/issues/12) so your +project is identifiable on the network. + ## Installation Homebrew ships a prebuilt binary, so no Rust toolchain: diff --git a/website/src/lib/links.ts b/website/src/lib/links.ts index 226d1f3..00019df 100644 --- a/website/src/lib/links.ts +++ b/website/src/lib/links.ts @@ -13,6 +13,8 @@ export const LINKS = { slskd: 'https://github.com/slskd/slskd', slsknet: 'https://www.slsknet.org', seeleseek: 'https://seeleseek.net', + seakarr: 'https://github.com/binhex/seakarr', + minorVersions: 'https://github.com/michel/soulseek-rs/issues/12', } as const export const VERSION = __APP_VERSION__ diff --git a/website/src/pages/community.tsx b/website/src/pages/community.tsx index f93eef2..22f4ff4 100644 --- a/website/src/pages/community.tsx +++ b/website/src/pages/community.tsx @@ -259,6 +259,31 @@ export const Community = () => { +
+ + The protocol crate is there for anyone building their own client, bot, or + automation on Soulseek. These projects do. + + + +

+ seakarr — automated Soulseek music + downloader with library quality upgrading: it scans your collection, finds + albums below your bitrate bar, and fetches better copies. +

+

+ Built something on soulseek-rs-lib? Open a pull request to get + listed, and{' '} + reserve a client minor version{' '} + so your project is identifiable on the network. +

+ + +
+
Date: Mon, 17 Aug 2026 10:23:21 +0200 Subject: [PATCH 4/4] fix(peer): survive download token races on the listener path A P-connection drop failed every download for its user, including ones actively transferring on their own healthy F connection: collect_failed_tokens now only fails queued downloads. And an F connection could look its token up before the client-operations thread had registered it: the listener now waits for the registration, and claims it only for the matching peer and only when not already claimed. Fixes #42 --- soulseek-rs-lib/src/download_store.rs | 60 +++++++- soulseek-rs-lib/src/peer/listen.rs | 199 +++++++++++++++++++++----- soulseek-rs-lib/tests/e2e.rs | 97 +++++++++---- 3 files changed, 289 insertions(+), 67 deletions(-) diff --git a/soulseek-rs-lib/src/download_store.rs b/soulseek-rs-lib/src/download_store.rs index c07f89b..532a0a3 100644 --- a/soulseek-rs-lib/src/download_store.rs +++ b/soulseek-rs-lib/src/download_store.rs @@ -142,8 +142,8 @@ impl DownloadStore { } } -/// Returns the tokens of downloads matching `username` (and optionally a -/// `filename`) after notifying their senders of `Failed`. +/// Returns the tokens of queued downloads matching `username` (and optionally +/// a `filename`) after notifying their senders of `Failed`. /// /// Caller is responsible for then calling `update_status` and `remove` for /// each token, typically under a write lock. @@ -159,6 +159,7 @@ pub fn collect_failed_tokens( .filter(|d| { d.username == username && filename.is_none_or(|f| d.filename == *f) }) + .filter(|d| matches!(d.status, DownloadStatus::Queued)) .map(|d| { let _ = d.sender.send(DownloadStatus::Failed(Some( "The upload failed on the other side".to_string(), @@ -305,6 +306,61 @@ mod tests { assert!(!store.remove_by_file("peer", "song.mp3"), "idempotent"); } + #[test] + fn collect_failed_tokens_collects_only_queued_downloads() { + let mut store = DownloadStore::new(); + let (tx_active, rx_active) = mpsc::channel(); + let (tx_paused, rx_paused) = mpsc::channel(); + let (tx_completed, rx_completed) = mpsc::channel(); + let (tx_queued, _rx_queued) = mpsc::channel(); + + let mut active = make_download( + 1, + DownloadStatus::InProgress { + bytes_downloaded: 25, + total_bytes: 100, + speed_bytes_per_sec: 10.0, + }, + ); + active.sender = tx_active; + + let mut paused = make_download( + 2, + DownloadStatus::Paused { + bytes_downloaded: 25, + total_bytes: 100, + }, + ); + paused.sender = tx_paused; + + let mut completed = make_download(3, DownloadStatus::Completed); + completed.sender = tx_completed; + + let mut queued = make_download(4, DownloadStatus::Queued); + queued.sender = tx_queued; + + store.add(active); + store.add(paused); + store.add(completed); + store.add(queued); + + let tokens = collect_failed_tokens(&store, "peer", None); + + assert_eq!(tokens, vec![4], "only the queued download may fail"); + assert!( + rx_active.try_recv().is_err(), + "an in-progress transfer must not be notified of failure" + ); + assert!( + rx_paused.try_recv().is_err(), + "a paused transfer must not be notified of failure" + ); + assert!( + rx_completed.try_recv().is_err(), + "a completed download must not be notified of failure" + ); + } + #[test] fn collect_failed_tokens_notifies_and_lists_matching() { let mut store = DownloadStore::new(); diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 3536be2..18cf6c4 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -26,6 +26,8 @@ const PEER_INIT_MESSAGE_CODE: u8 = 1; /// How long an accepted peer gets to send its peer-init handshake. const PEER_INIT_TIMEOUT: Duration = Duration::from_secs(10); +const TOKEN_REGISTRATION_TIMEOUT: Duration = Duration::from_secs(10); + const MAX_HANDSHAKES: usize = 128; const MAX_PEER_INIT_BYTES: usize = 4096; @@ -116,41 +118,52 @@ fn wait_for_transfer_token( Ok(()) } -fn extract_download_from_buffer( - reader: &mut MessageReader, - client_context: &Arc>, - username: &str, - peer_ip: &str, - peer_port: u32, -) -> Option { - if reader.buffer_len() == 0 { - return None; - } +fn take_transfer_token(reader: &mut MessageReader) -> Option { let buffer = reader.get_buffer(); - let token = u32::from_le_bytes(buffer.get(0..4)?.try_into().ok()?); - trace!( - "[listener:{}] got transfer_token: {} from data chunk", - username, token - ); + Some(u32::from_le_bytes(buffer.get(0..4)?.try_into().ok()?)) +} - let context = match client_context.read_safe() { - Ok(c) => c, - Err(e) => { - error!("[listener] client context lock: {}", e); +fn wait_for_download_registration( + client_context: &Arc>, + token: u32, + username: &str, + deadline: Instant, +) -> Option { + loop { + let download = match client_context.read_safe() { + Ok(context) => context + .get_download_by_token(token) + .filter(|d| { + d.username == username + && !matches!( + d.status, + DownloadStatus::InProgress { .. } + | DownloadStatus::Paused { .. } + | DownloadStatus::Completed + ) + }) + .cloned(), + Err(e) => { + error!("[listener] client context lock: {}", e); + return None; + } + }; + if download.is_some() { + return download; + } + if Instant::now() >= deadline { + if let Ok(context) = client_context.read_safe() { + trace!( + "[listener:{}] no claimable download for token {}, known tokens: {:?}", + username, + token, + context.get_download_tokens() + ); + } return None; } - }; - let download = context.get_download_by_token(token).cloned(); - - if download.is_none() { - let download_tokens = context.get_download_tokens(); - trace!( - "[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}", - token, download_tokens - ); + thread::sleep(Duration::from_millis(50)); } - - download } fn handle_peer_connection( @@ -201,13 +214,15 @@ fn handle_file_connection( return; } - let download = extract_download_from_buffer( - &mut reader, - &context.client_context, - &peer.username, - &peer.host, - peer.port, - ); + let download = + take_transfer_token(&mut reader).and_then(|transfer_token| { + wait_for_download_registration( + &context.client_context, + transfer_token, + &peer.username, + Instant::now() + TOKEN_REGISTRATION_TIMEOUT, + ) + }); let failure_token = download.as_ref().map(|d| d.token); let download_peer = DownloadPeer::new( @@ -474,8 +489,116 @@ impl Listen { #[cfg(test)] mod tests { use super::*; + use crate::types::DownloadMetadata; use std::io::Write; use std::net::TcpListener; + use std::sync::mpsc; + + fn registered_download(token: u32) -> Download { + Download { + username: "peer".to_string(), + filename: "song.mp3".to_string(), + token, + size: 100, + download_directory: std::env::temp_dir().display().to_string(), + status: DownloadStatus::Queued, + sender: mpsc::channel().0, + queue_position: None, + metadata: DownloadMetadata::default(), + } + } + + #[test] + fn a_download_registered_during_the_wait_is_found() { + let client_context = Arc::new(RwLock::new(ClientContext::new())); + let registering = { + let client_context = client_context.clone(); + std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(100)); + client_context + .write() + .unwrap() + .add_download(registered_download(7)); + }) + }; + + let download = wait_for_download_registration( + &client_context, + 7, + "peer", + Instant::now() + Duration::from_secs(5), + ); + + assert!( + download.is_some(), + "a token registered during the wait must be found" + ); + let _ = registering.join(); + } + + #[test] + fn an_unregistered_token_gives_up_at_the_deadline() { + let client_context = Arc::new(RwLock::new(ClientContext::new())); + + let started = Instant::now(); + let download = wait_for_download_registration( + &client_context, + 7, + "peer", + started + Duration::from_millis(200), + ); + + assert!(download.is_none(), "an unknown token must not be invented"); + assert!( + started.elapsed() < Duration::from_secs(2), + "the deadline must bound the wait" + ); + } + + #[test] + fn a_download_for_another_peer_is_not_claimed() { + let client_context = Arc::new(RwLock::new(ClientContext::new())); + client_context + .write() + .unwrap() + .add_download(registered_download(7)); + + let download = wait_for_download_registration( + &client_context, + 7, + "impostor", + Instant::now() + Duration::from_millis(200), + ); + + assert!( + download.is_none(), + "a token must only match the peer it was registered for" + ); + } + + #[test] + fn an_already_claimed_download_is_not_reclaimed() { + let client_context = Arc::new(RwLock::new(ClientContext::new())); + let mut claimed = registered_download(7); + claimed.status = DownloadStatus::InProgress { + bytes_downloaded: 25, + total_bytes: 100, + speed_bytes_per_sec: 10.0, + }; + client_context.write().unwrap().add_download(claimed); + + let download = wait_for_download_registration( + &client_context, + 7, + "peer", + Instant::now() + Duration::from_millis(200), + ); + + assert!( + download.is_none(), + "a transfer already in progress must not be claimed again" + ); + } #[test] fn a_dribbling_handshake_hits_the_total_deadline() { @@ -557,7 +680,7 @@ mod tests { assert!(result.is_err(), "a silent peer must not park the thread"); assert!( - started.elapsed() < Duration::from_secs(5), + started.elapsed() < Duration::from_secs(2), "the deadline must bound the wait" ); let _ = writer.join(); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 39c3433..072bc21 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -520,36 +520,43 @@ fn connect_retry(addr: &str, timeout: Duration) -> std::io::Result { } } -fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result { - // 1. P (control) connection: register ourselves with the downloader. +fn open_control_and_await_queue( + cfg: &MockUpload, +) -> std::io::Result { let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?; p.set_read_timeout(Some(Duration::from_secs(10)))?; p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?; p.flush()?; let _ = cfg.ready.send(()); - // 2. Wait for the downloader's QueueUpload (peer code 43). loop { - let mut msg = read_framed(&mut p)?; + let msg = read_framed(&mut p)?; if msg.get_message_code() == 43 { - msg.set_pointer(8); - let _requested = msg.read_string(); break; } } + Ok(p) +} - // 3. Offer the transfer with a TransferRequest (peer code 40). The size here - // becomes the download's expected size, so it must match the content. +/// Build a TransferRequest (peer code 40). The size becomes the download's +/// expected size, so it must match the content. +fn transfer_request(cfg: &MockUpload) -> Message { let mut tr = Message::new(); tr.write_int32(40) .write_int32(1) // direction: upload .write_int32(cfg.token) .write_string(&cfg.filename) .write_int64(cfg.content.len() as u64); - p.write_all(&tr.get_buffer())?; + tr +} + +fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result { + let mut p = open_control_and_await_queue(cfg)?; + + p.write_all(&transfer_request(cfg).get_buffer())?; p.flush()?; - // 4. Wait for the downloader to allow it (TransferResponse, peer code 41). + // Wait for the downloader to allow it (TransferResponse, peer code 41). loop { let msg = read_framed(&mut p)?; if msg.get_message_code() == 41 { @@ -557,7 +564,6 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result { } } - // 5. Open the F (file) connection to the downloader's listener and stream. serve_file_over_f( &cfg.listen_addr, &cfg.peer_username, @@ -567,22 +573,38 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result { ) } +fn run_mock_uploader_f_first(cfg: &MockUpload) -> std::io::Result { + let mut p = open_control_and_await_queue(cfg)?; + + let listen_addr = cfg.listen_addr.clone(); + let peer_username = cfg.peer_username.clone(); + let token = cfg.token; + let content = cfg.content.clone(); + let token_delay = cfg.token_delay; + let f = std::thread::spawn(move || { + serve_file_over_f( + &listen_addr, + &peer_username, + token, + &content, + token_delay, + ) + }); + + std::thread::sleep(Duration::from_millis(300)); + + p.write_all(&transfer_request(cfg).get_buffer())?; + p.flush()?; + + f.join() + .map_err(|_| std::io::Error::other("F connection thread panicked"))? +} + fn run_refusing_uploader( cfg: &MockUpload, refusal: &Message, ) -> std::io::Result<()> { - let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?; - p.set_read_timeout(Some(Duration::from_secs(10)))?; - p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?; - p.flush()?; - let _ = cfg.ready.send(()); - - loop { - let msg = read_framed(&mut p)?; - if msg.get_message_code() == 43 { - break; - } - } + let mut p = open_control_and_await_queue(cfg)?; p.write_all(&refusal.get_buffer())?; p.flush()?; @@ -724,7 +746,11 @@ fn unique_download_dir() -> PathBuf { dir } -fn file_downloads_over_p_and_f(peer_username: &str, token_delay: Duration) { +fn file_downloads_over_p_and_f( + peer_username: &str, + token_delay: Duration, + run_uploader: fn(&MockUpload) -> std::io::Result, +) { let server = server_or_skip!(); // Downloader: connected to the server with its peer listener enabled. @@ -755,7 +781,7 @@ fn file_downloads_over_p_and_f(peer_username: &str, token_delay: Duration) { token_delay, }; let uploader = std::thread::spawn(move || { - if let Err(e) = run_mock_uploader(&cfg) { + if let Err(e) = run_uploader(&cfg) { eprintln!("[mock uploader] {e}"); } }); @@ -792,12 +818,29 @@ fn file_downloads_over_p_and_f(peer_username: &str, token_delay: Duration) { #[test] fn a_file_downloads_from_a_peer_over_p_and_f_connections() { - file_downloads_over_p_and_f("e2e_mockpeer", Duration::ZERO); + file_downloads_over_p_and_f( + "e2e_mockpeer", + Duration::ZERO, + run_mock_uploader, + ); } #[test] fn a_file_downloads_when_the_transfer_token_arrives_late() { - file_downloads_over_p_and_f("e2e_latepeer", Duration::from_millis(300)); + file_downloads_over_p_and_f( + "e2e_latepeer", + Duration::from_millis(300), + run_mock_uploader, + ); +} + +#[test] +fn a_download_completes_when_the_f_connection_beats_token_registration() { + file_downloads_over_p_and_f( + "e2e_earlypeer", + Duration::ZERO, + run_mock_uploader_f_first, + ); } fn refusal_fails_the_download_quickly(