diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 66eb84c..a03a79a 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -49,6 +49,10 @@ pub struct PeerActor { dispatcher: Option>, dispatcher_receiver: Option>, queued_messages: Vec, + own_username: String, + /// True when we initiated this connection (no stream supplied at + /// construction), so we must send a `PeerInit` once connected. + outbound: bool, } impl PeerActor { @@ -58,7 +62,9 @@ impl PeerActor { stream: Option, reader: Option, client_channel: Sender, + own_username: String, ) -> Self { + let outbound = stream.is_none(); let connection_state = if stream.is_some() { ConnectionState::Connected } else { @@ -75,6 +81,8 @@ impl PeerActor { dispatcher: None, dispatcher_receiver: None, queued_messages: Vec::new(), + own_username, + outbound, } } @@ -529,8 +537,12 @@ impl PeerActor { } fn on_connection_established(&mut self) { - let (username, token) = match self.peer.read_safe() { - Ok(peer) => (peer.username.clone(), peer.token), + let (username, connection_type, token) = match self.peer.read_safe() { + Ok(peer) => ( + peer.username.clone(), + peer.connection_type.clone(), + peer.token, + ), Err(e) => { error!( "[peer_actor] on_connection_established peer lock: {}", @@ -544,11 +556,19 @@ impl PeerActor { return; }; - if let Some(token) = token { - let handshake_msg = MessageFactory::build_watch_user(token); - if let Err(e) = stream.write_all(&handshake_msg.get_data()) { + // Connections we initiated must announce themselves with a PeerInit + // (peer code 1); inbound peers already sent us theirs, so we stay + // silent for them. `get_buffer` prepends the length prefix the peer's + // MessageReader expects. + if self.outbound { + let init = MessageFactory::build_peer_init_message( + &self.own_username, + connection_type, + token.unwrap_or(0), + ); + if let Err(e) = stream.write_all(&init.get_buffer()) { error!( - "[peer:{}] Failed to send watch user handshake: {}", + "[peer:{}] Failed to send PeerInit handshake: {}", username, e ); self.disconnect_with_error(e); @@ -563,6 +583,14 @@ impl PeerActor { self.handle_message(msg); } + // Tell the client the control connection is live so any downloads + // queued for this peer can now be requested over a handshaken stream. + if self.outbound { + let _ = self + .client_channel + .send(ClientOperation::PeerConnected(username)); + } + if let Some(ref handle) = self.self_handle { handle.send(PeerMessage::ProcessRead).ok(); } diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index ad72102..e9cf70f 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -15,6 +15,7 @@ pub struct PeerRegistry { peers: Arc>>>, actor_system: Arc, client_channel: Sender, + own_username: String, } impl PeerRegistry { @@ -22,11 +23,13 @@ impl PeerRegistry { pub fn new( actor_system: Arc, client_channel: Sender, + own_username: String, ) -> Self { Self { peers: Arc::new(Mutex::new(HashMap::new())), actor_system, client_channel, + own_username, } } @@ -38,8 +41,13 @@ impl PeerRegistry { ) -> Result, String> { let username = peer.username.clone(); - let actor = - PeerActor::new(peer, stream, reader, self.client_channel.clone()); + let actor = PeerActor::new( + peer, + stream, + reader, + self.client_channel.clone(), + self.own_username.clone(), + ); let handle = self.actor_system.spawn_with_handle(actor, |actor, handle| { @@ -157,6 +165,7 @@ impl Clone for PeerRegistry { peers: self.peers.clone(), actor_system: self.actor_system.clone(), client_channel: self.client_channel.clone(), + own_username: self.own_username.clone(), } } } diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index 30e3286..83099f3 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -97,6 +97,7 @@ pub enum ClientOperation { }, SetServerSender(Sender), PrivateMessageReceived(UserMessage), + PeerConnected(String), } pub struct ClientContext { pub peer_registry: Option, @@ -254,6 +255,26 @@ fn test_client_pause_and_resume_download() { )); } +#[test] +fn download_without_a_connection_resolves_failed() { + // A client that never connected has no server handle and no peer registry, + // so it cannot open a connection to the peer: the download must resolve to + // Failed rather than hang Queued forever. + let client = Client::new("test-user", "test-password"); + let (_download, receiver) = client + .download( + "song.mp3".to_string(), + "peer".to_string(), + 100, + "test".to_string(), + ) + .expect("download() should return a handle"); + assert!(matches!( + receiver.recv_timeout(Duration::from_secs(1)), + Ok(DownloadStatus::Failed) + )); +} + #[test] fn test_client_removes_only_queued_downloads() { let client = Client::new("test-user", "test-password"); @@ -375,8 +396,11 @@ impl Client { let mut ctx = self.context.write_safe()?; ctx.sender = Some(sender.clone()); - let peer_registry = - PeerRegistry::new(ctx.actor_system.clone(), sender.clone()); + let peer_registry = PeerRegistry::new( + ctx.actor_system.clone(), + sender.clone(), + self.username.clone(), + ); ctx.peer_registry = Some(peer_registry); let listen_sender = sender.clone(); @@ -461,6 +485,23 @@ impl Client { Ok(()) } + /// Ask the server for a peer's address and open a direct control + /// connection to it. Downloads queued for that peer are sent automatically + /// once the connection is established. + /// + /// # Errors + /// Returns [`SoulseekRs::NotConnected`] if the client is not connected. + pub fn connect_peer(&self, username: &str) -> Result<()> { + let handle = self + .server_handle + .as_ref() + .ok_or(SoulseekRs::NotConnected)?; + handle + .send(ServerMessage::GetPeerAddress(username.to_string())) + .map_err(|_| SoulseekRs::NotConnected)?; + Ok(()) + } + /// Remove and return all private messages received since the last call. #[must_use] pub fn take_private_messages(&self) -> Vec { @@ -684,18 +725,29 @@ impl Client { let mut context = self.context.write_safe()?; context.add_download(download.clone()); - let download_initiated = - if let Some(ref registry) = context.peer_registry { - registry - .queue_upload(&username, download.filename.clone()) - .is_ok() - } else { - false - }; + // If we already have a control connection to this peer, queue the + // upload immediately. Otherwise open one directly (server GetPeerAddress + // → outbound PeerInit → PeerConnected → the queued upload is flushed). + let peer_registered = context + .peer_registry + .as_ref() + .is_some_and(|r| r.contains(&username)); + let queued_now = peer_registered + && context.peer_registry.as_ref().is_some_and(|r| { + r.queue_upload(&username, download.filename.clone()).is_ok() + }); drop(context); - if !download_initiated { + let failed = if peer_registered { + !queued_now + } else { + // No existing connection: initiate one. Only a genuinely + // unconnected client (no server handle) fails outright here. + self.connect_peer(&username).is_err() + }; + + if failed { let _ = download.sender.send(DownloadStatus::Failed); self.context .write_safe()? @@ -1187,6 +1239,44 @@ impl Client { e ), }, + ClientOperation::PeerConnected(username) => { + // An outbound control connection just handshook. + // Flush any downloads that were queued for this + // peer while we were still connecting. Collect + // under a read guard, then act without it held. + let (registry, files): ( + Option, + Vec, + ) = match client_context.read_safe() { + Ok(ctx) => ( + ctx.peer_registry.clone(), + ctx.get_downloads() + .iter() + .filter(|d| { + d.username == username + && matches!( + d.status, + DownloadStatus::Queued + ) + }) + .map(|d| d.filename.clone()) + .collect(), + ), + Err(e) => { + error!( + "[client] PeerConnected read: {}", + e + ); + continue; + } + }; + if let Some(registry) = registry { + for filename in files { + let _ = registry + .queue_upload(&username, filename); + } + } + } } } Err(e) => { diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 9e041d8..d8748ed 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -142,7 +142,6 @@ impl MessageFactory { .clone() } - #[allow(dead_code)] #[must_use] pub fn build_peer_init_message( own_username: &str, @@ -184,6 +183,22 @@ fn test_build_login_message() { assert_eq!(expect, message.get_data()); } +#[test] +fn test_build_peer_init_message() { + use crate::peer::ConnectionType; + let message = + MessageFactory::build_peer_init_message("bob", ConnectionType::P, 7); + // [1][len=3]"bob"[len=1]"P"[token=7] — no length prefix in get_data() + let expect: Vec = [ + 1, // PeerInit code (int8) + 3, 0, 0, 0, 98, 111, 98, // username "bob" + 1, 0, 0, 0, 80, // connection type "P" + 7, 0, 0, 0, // token + ] + .to_vec(); + assert_eq!(expect, message.get_data()); +} + #[test] fn test_build_message_user() { let message = MessageFactory::build_message_user("bob", "hi"); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index f4e2abd..ef9c2e0 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -26,6 +26,7 @@ use std::sync::mpsc::Sender; use std::time::{Duration, Instant}; use soulseek_rs::message::Message; +use soulseek_rs::message::server::MessageFactory; use soulseek_rs::{Client, ClientSettings, DownloadStatus, PeerAddress}; /// A Soulseek server to test against: either a child soulfind process we @@ -449,21 +450,36 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<()> { } } - // 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))?; + // 5. Open the F (file) connection to the downloader's listener and stream. + serve_file_over_f( + &cfg.listen_addr, + &cfg.peer_username, + cfg.token, + &cfg.content, + ) +} + +/// 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. +fn serve_file_over_f( + downloader_addr: &str, + username: &str, + token: u32, + content: &[u8], +) -> 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(&cfg.peer_username, "F", cfg.token); - init.extend_from_slice(&cfg.token.to_le_bytes()); + let mut init = peer_init_bytes(username, "F", token); + init.extend_from_slice(&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.write_all(content)?; f.flush()?; // Keep the connection open briefly so the reader drains everything. @@ -567,3 +583,203 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() { let _ = std::fs::remove_dir_all(&download_dir); } + +// --------------------------------------------------------------------------- +// Direct-connection download: the client initiates the peer connection. +// +// Here the mock is a server-registered peer. It logs in to soulfind and +// advertises a listen port (SetWaitPort), so when our client asks the server +// for the peer's address (GetPeerAddress) and dials it directly, the mock +// accepts that inbound P connection and then serves the file. This exercises +// the outbound PeerInit handshake and the auto-connecting download() path. +// --------------------------------------------------------------------------- + +struct MockDirectUpload { + server_addr: String, + username: String, + password: String, + listen_port: u16, + downloader_listen_addr: String, + downloader_username: String, + filename: String, + content: Vec, + token: u32, + ready: Sender<()>, +} + +fn run_mock_direct_peer(cfg: &MockDirectUpload) -> std::io::Result<()> { + // 1. Log in to the server so it knows this user is online. + let mut srv = connect_retry(&cfg.server_addr, Duration::from_secs(5))?; + srv.set_read_timeout(Some(Duration::from_secs(10)))?; + srv.write_all( + &MessageFactory::build_login_message(&cfg.username, &cfg.password) + .get_buffer(), + )?; + srv.flush()?; + loop { + let msg = read_framed(&mut srv)?; + if msg.get_message_code() == 1 { + break; // login response + } + } + + // 2. Bind the peer listener, then advertise its port so the server can hand + // our address to the downloader. Bind all interfaces: soulfind reports + // the host's LAN address (not 127.0.0.1), and the downloader dials that. + let listener = std::net::TcpListener::bind(("0.0.0.0", cfg.listen_port))?; + srv.write_all( + &MessageFactory::build_set_wait_port_message(cfg.listen_port) + .get_buffer(), + )?; + srv.flush()?; + let _ = cfg.ready.send(()); + + // 3. Accept the downloader's inbound P (control) connection and validate its + // PeerInit (peer code 1, int8 code so the fields start at offset 5). The + // accept is bounded so a misrouted connection fails the test instead of + // hanging it. + listener.set_nonblocking(true)?; + let accept_deadline = Instant::now() + Duration::from_secs(15); + let (mut p, _addr) = loop { + match listener.accept() { + Ok(pair) => break pair, + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + if Instant::now() >= accept_deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "no inbound P connection from the downloader", + )); + } + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => return Err(e), + } + }; + p.set_nonblocking(false)?; + p.set_read_timeout(Some(Duration::from_secs(10)))?; + let mut init = read_framed(&mut p)?; + assert_eq!(init.get_message_code(), 1, "expected inbound PeerInit"); + init.set_pointer(5); + assert_eq!(init.read_string(), cfg.downloader_username, "PeerInit user"); + assert_eq!(init.read_string(), "P", "PeerInit connection type"); + + // 4. Negotiate the transfer exactly as in the passive path. + loop { + let mut msg = read_framed(&mut p)?; + if msg.get_message_code() == 43 { + msg.set_pointer(8); + let _requested = msg.read_string(); + break; + } + } + let mut tr = Message::new(); + tr.write_int32(40) + .write_int32(1) + .write_int32(cfg.token) + .write_string(&cfg.filename) + .write_int64(cfg.content.len() as u64); + p.write_all(&tr.get_buffer())?; + p.flush()?; + loop { + let msg = read_framed(&mut p)?; + if msg.get_message_code() == 41 { + break; // TransferResponse + } + } + + // 5. Stream the bytes over an F connection to the downloader's listener. + // `srv` stays in scope so the peer remains online for the whole transfer. + serve_file_over_f( + &cfg.downloader_listen_addr, + &cfg.username, + cfg.token, + &cfg.content, + ) +} + +#[test] +fn a_file_downloads_from_a_peer_via_direct_connection() { + let server = server_or_skip!(); + + // Downloader with a listener enabled (needed for the F leg). + let client_port = free_port().expect("free client listen port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_direct_dl", + "pw", + client_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let mock_port = free_port().expect("free mock listen port"); + let filename = "direct_song.mp3"; + let content: Vec = (0..2000u32).map(|i| (i % 251) as u8).collect(); + let size = content.len() as u64; + let token = 515_151_u32; + let download_dir = unique_download_dir(); + + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let cfg = MockDirectUpload { + server_addr: format!("{}:{}", server.host, server.port), + username: "e2e_directpeer".to_string(), + password: "pw".to_string(), + listen_port: mock_port, + downloader_listen_addr: format!("127.0.0.1:{client_port}"), + downloader_username: "e2e_direct_dl".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_direct_peer(&cfg) { + eprintln!("[mock direct peer] {e}"); + } + }); + + ready_rx + .recv_timeout(Duration::from_secs(10)) + .expect("mock direct peer ready"); + // Let the server finish processing SetWaitPort before we resolve the address. + std::thread::sleep(Duration::from_millis(1000)); + + // Target the plain username so download() takes the direct-connect path. + let (_download, status_rx) = client + .download( + filename.to_string(), + "e2e_directpeer".to_string(), + size, + download_dir.display().to_string(), + ) + .expect("start download"); + + 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 direct 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); +}