feat: direct peer connections — download() connects to unconnected peers

Implement the DIRECT connection-initiation mode so a download can target a
peer the client is not already connected to:

- Fix the outbound peer handshake. on_connection_established sent
  build_watch_user() (a server WatchUser message, wrongly framed with
  get_data()) for outbound connections; it now sends a proper length-prefixed
  PeerInit (peer code 1) with our own username. own_username is plumbed in
  via PeerRegistry -> PeerActor; inbound peers (outbound=false) are unchanged.
- download() now auto-connects: if there is no control connection to the peer
  it sends GetPeerAddress and defers the QueueUpload until the outbound stream
  handshakes, signalled by a new ClientOperation::PeerConnected. A never-
  connected client still resolves the download to Failed (unit test).
- Add Client::connect_peer(username) to open/pre-warm a control connection.

Verified end to end against soulfind: a_file_downloads_from_a_peer_via_direct_connection
logs a mock peer into the server, advertises its port via SetWaitPort, and the
client resolves the address, dials it, and downloads the file — bytes match.
The F-serve leg is shared with the passive-download mock.

soulseek-rs-lib/src/actor/peer_actor.rs

@@ -49,6 +49,10 @@ pub struct PeerActor {
     dispatcher: Option<MessageDispatcher<PeerMessage>>,
     dispatcher_receiver: Option<Receiver<PeerMessage>>,
     queued_messages: Vec<PeerMessage>,
+    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<TcpStream>,
         reader: Option<MessageReader>,
         client_channel: Sender<ClientOperation>,
+        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() {
+        let (username, connection_type, token) = match self.peer.read_safe() {
-            Ok(peer) => (peer.username.clone(), peer.token),
+            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 {
+        // Connections we initiated must announce themselves with a PeerInit
-            let handshake_msg = MessageFactory::build_watch_user(token);
+        // (peer code 1); inbound peers already sent us theirs, so we stay
-            if let Err(e) = stream.write_all(&handshake_msg.get_data()) {
+        // 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();
         }

soulseek-rs-lib/src/actor/peer_registry.rs

@@ -15,6 +15,7 @@ pub struct PeerRegistry {
     peers: Arc<Mutex<HashMap<String, ActorHandle<PeerMessage>>>>,
     actor_system: Arc<ActorSystem>,
     client_channel: Sender<ClientOperation>,
+    own_username: String,
 }
 
 impl PeerRegistry {
@@ -22,11 +23,13 @@ impl PeerRegistry {
     pub fn new(
         actor_system: Arc<ActorSystem>,
         client_channel: Sender<ClientOperation>,
+        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<ActorHandle<PeerMessage>, String> {
         let username = peer.username.clone();
 
-        let actor =
+        let actor = PeerActor::new(
-            PeerActor::new(peer, stream, reader, self.client_channel.clone());
+            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(),
         }
     }
 }

soulseek-rs-lib/src/client.rs

@@ -97,6 +97,7 @@ pub enum ClientOperation {
     },
     SetServerSender(Sender<ServerMessage>),
     PrivateMessageReceived(UserMessage),
+    PeerConnected(String),
 }
 pub struct ClientContext {
     pub peer_registry: Option<PeerRegistry>,
@@ -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 =
+        let peer_registry = PeerRegistry::new(
-            PeerRegistry::new(ctx.actor_system.clone(), sender.clone());
+            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<UserMessage> {
@@ -684,18 +725,29 @@ impl Client {
         let mut context = self.context.write_safe()?;
         context.add_download(download.clone());
 
-        let download_initiated =
+        // If we already have a control connection to this peer, queue the
-            if let Some(ref registry) = context.peer_registry {
+        // upload immediately. Otherwise open one directly (server GetPeerAddress
-                registry
+        // → outbound PeerInit → PeerConnected → the queued upload is flushed).
-                    .queue_upload(&username, download.filename.clone())
+        let peer_registered = context
-                    .is_ok()
+            .peer_registry
-            } else {
+            .as_ref()
-                false
+            .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<PeerRegistry>,
+                                    Vec<String>,
+                                ) = 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) => {

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<u8> = [
+        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");

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
+    // 5. Open the F (file) connection to the downloader's listener and stream.
-    //    single write so the token lands in the listener's read buffer, where
+    serve_file_over_f(
-    //    the download is looked up by token.
+        &cfg.listen_addr,
-    let mut f = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?;
+        &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);
+    let mut init = peer_init_bytes(username, "F", token);
-    init.extend_from_slice(&cfg.token.to_le_bytes());
+    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<u8>,
+    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<u8> = (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);
+}