From 06cd63873609511278af0a391ba5dff03f3d3d5e Mon Sep 17 00:00:00 2001 From: michel Date: Sun, 19 Jul 2026 13:58:05 +0200 Subject: [PATCH] fix: drive inbound peer connections non-blocking so downloads don't stall MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A peer actor multiplexes socket reads with its mailbox on one thread: tick() reads the socket while outgoing messages (e.g. a queued QueueUpload that starts a download) are delivered through the mailbox between ticks. Inbound connections accepted by the listener were left in blocking mode, so once such a peer went idle tick() parked inside read() and never drained the mailbox — a download queued against that peer never sent its QueueUpload and hung. Outbound connections already set non-blocking; do the same for inbound P connections. Covered by a new server-optional e2e test that downloads a file end to end from an in-process mock peer over real P (control) and F (transfer) connections, asserting the bytes written match the source. The mock speaks the library's own public Message wire format, so no dependencies are added. --- soulseek-rs-lib/src/peer/listen.rs | 12 ++ soulseek-rs-lib/tests/e2e.rs | 221 ++++++++++++++++++++++++++++- 2 files changed, 232 insertions(+), 1 deletion(-) diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index bfdefec..254ad95 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -115,6 +115,18 @@ fn handle_peer_connection( _peer_ip: &str, _peer_port: u16, ) { + // The peer actor multiplexes socket reads with its mailbox on a single + // thread: `tick()` reads the socket, but outgoing messages (e.g. a queued + // QueueUpload for a download) are delivered through the mailbox between + // ticks. A blocking socket would park `tick()` inside `read` whenever the + // peer is idle, starving the mailbox and stalling downloads. Match the + // outbound path and drive this connection non-blocking. + if let Err(e) = stream.set_nonblocking(true) { + error!("[listener] failed to set peer stream non-blocking: {}", e); + return; + } + stream.set_nodelay(true).ok(); + let client_context = match context.client_context.read_safe() { Ok(c) => c, Err(e) => { diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 0f01896..f4e2abd 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -18,12 +18,15 @@ #![allow(clippy::doc_markdown)] +use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::path::PathBuf; use std::process::{Child, Command, Stdio}; +use std::sync::mpsc::Sender; use std::time::{Duration, Instant}; -use soulseek_rs::{Client, ClientSettings, PeerAddress}; +use soulseek_rs::message::Message; +use soulseek_rs::{Client, ClientSettings, DownloadStatus, PeerAddress}; /// A Soulseek server to test against: either a child soulfind process we /// spawned, or an external server referenced by `SOULSEEK_TEST_SERVER`. @@ -348,3 +351,219 @@ fn two_clients_can_be_logged_in_together() { assert!(alice.login().expect("alice login")); assert!(bob.login().expect("bob login")); } + +// --------------------------------------------------------------------------- +// Peer-to-peer download coverage. +// +// The client cannot serve files (the upload side is not implemented), so a real +// download is exercised with a minimal in-process "mock uploader" that speaks +// the peer protocol using the library's own public `Message` wire format. It +// drives the same path a real peer would after a search: a `P` control +// connection to our listener, a `QueueUpload` → `TransferRequest` → +// `TransferResponse` negotiation, then an `F` connection that streams the bytes. +// --------------------------------------------------------------------------- + +/// Configuration for a one-shot mock uploader. +struct MockUpload { + listen_addr: String, + peer_username: String, + filename: String, + content: Vec, + token: u32, + ready: Sender<()>, +} + +/// Build a `PeerInit` (peer code 1) frame: `[len][1][username][conn_type][token]`. +fn peer_init_bytes(username: &str, conn_type: &str, token: u32) -> Vec { + let mut m = Message::new(); + m.write_int8(1) + .write_string(username) + .write_string(conn_type) + .write_int32(token); + m.get_buffer() +} + +/// Read one length-prefixed peer message (`[len:4 LE][payload]`) from a blocking +/// stream. The returned `Message` keeps the length prefix, so `get_message_code` +/// and `set_pointer(8)` behave exactly as they do inside the library. +fn read_framed(stream: &mut TcpStream) -> std::io::Result { + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_le_bytes(len_buf) as usize; + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload)?; + let mut data = len_buf.to_vec(); + data.extend_from_slice(&payload); + Ok(Message::new_with_data(data)) +} + +fn connect_retry(addr: &str, timeout: Duration) -> std::io::Result { + let deadline = Instant::now() + timeout; + loop { + match TcpStream::connect(addr) { + Ok(stream) => return Ok(stream), + Err(e) => { + if Instant::now() >= deadline { + return Err(e); + } + std::thread::sleep(Duration::from_millis(50)); + } + } + } +} + +fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<()> { + // 1. P (control) connection: register ourselves with the downloader. + 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)?; + if msg.get_message_code() == 43 { + msg.set_pointer(8); + let _requested = msg.read_string(); + break; + } + } + + // 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. + 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())?; + p.flush()?; + + // 4. Wait for the downloader to allow it (TransferResponse, peer code 41). + loop { + let msg = read_framed(&mut p)?; + if msg.get_message_code() == 41 { + break; + } + } + + // 5. F (file) connection. PeerInit(F) and the 4-byte transfer token go in a + // single write so the token lands in the listener's read buffer, where + // the download is looked up by token. + let mut f = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?; + f.set_read_timeout(Some(Duration::from_secs(10)))?; + 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)?; + f.flush()?; + + // 6. The downloader replies with an 8-byte START_DOWNLOAD offset, then we + // stream the file bytes. + let mut start = [0u8; 8]; + f.read_exact(&mut start)?; + f.write_all(&cfg.content)?; + f.flush()?; + + // Keep the connection open briefly so the reader drains everything. + std::thread::sleep(Duration::from_millis(500)); + Ok(()) +} + +fn unique_download_dir() -> PathBuf { + let dir = std::env::temp_dir().join(format!( + "soulseek-e2e-dl-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::create_dir_all(&dir); + dir +} + +#[test] +fn a_file_downloads_from_a_peer_over_p_and_f_connections() { + 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", + "pw", + listen_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let filename = "mock_song.mp3"; + let content: Vec = (0..2000u32).map(|i| (i % 251) as u8).collect(); + let size = content.len() as u64; + let token = 424_242_u32; + let download_dir = unique_download_dir(); + + // Start the mock uploader; it signals once its P connection is established. + 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(), + filename: filename.to_string(), + content: content.clone(), + token, + ready: ready_tx, + }; + let uploader = std::thread::spawn(move || { + if let Err(e) = run_mock_uploader(&cfg) { + eprintln!("[mock uploader] {e}"); + } + }); + + ready_rx + .recv_timeout(Duration::from_secs(5)) + .expect("mock uploader P connection"); + + // The listener registers an incoming peer under ":direct"; give + // that registration a moment to complete before queuing the download. + std::thread::sleep(Duration::from_millis(1500)); + + let (_download, status_rx) = client + .download( + filename.to_string(), + "e2e_mockpeer:direct".to_string(), + size, + download_dir.display().to_string(), + ) + .expect("start download"); + + // Wait for completion via the per-download status channel. + let mut completed = false; + let deadline = Instant::now() + Duration::from_secs(20); + while Instant::now() < deadline { + match status_rx.recv_timeout(Duration::from_millis(500)) { + Ok(DownloadStatus::Completed) => { + completed = true; + break; + } + Ok(DownloadStatus::Failed | DownloadStatus::TimedOut) => { + break; + } + _ => {} + } + if client + .get_all_downloads() + .iter() + .any(|d| matches!(d.status, DownloadStatus::Completed)) + { + completed = true; + break; + } + } + let _ = uploader.join(); + + assert!(completed, "the download should reach Completed"); + + let written = std::fs::read(download_dir.join(filename)) + .expect("downloaded file should exist"); + assert_eq!(written, content, "downloaded bytes should match the source"); + + let _ = std::fs::remove_dir_all(&download_dir); +}