feat: firewalled peer connections via server-brokered connect

Implement the FIREWALLED connection mode: when a direct outbound connection to
a peer fails (peer unreachable), fall back to asking the server to broker it.

- Add MessageFactory::build_connect_to_peer (server code 18).
- A pre-established outbound failure now emits ClientOperation::PeerConnectFailed
  instead of a normal disconnect; the client registers a correlation token and
  sends ConnectToPeer so the peer connects back.
- The listener accepts the peer's PierceFirewall (peer code 0) connect-back,
  matches the pending token, registers it as the P control connection, and
  fires PeerConnected to flush the queued download.

Fix a latent aliasing bug this surfaced: a peer actor could emit two terminal
notifications (a connect-failure, then a disconnect when later replaced and
stopped), and the second removed the *newer* actor registered under the same
username, killing the brokered connection mid-transfer. An actor now reports a
terminal outcome at most once (disconnect_reported guard).

Verified end to end against soulfind: the mock advertises an unlistened port so
the direct connect is refused, the server brokers the connection, the mock
pierces back, and the file transfers — bytes match. A companion test
(soulfind_brokers_connect_to_peer_between_users) pins the server behaviour the
mode relies on.

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

@@ -53,6 +53,14 @@ pub struct PeerActor {
     /// True when we initiated this connection (no stream supplied at
     /// construction), so we must send a `PeerInit` once connected.
     outbound: bool,
+    /// Set once the connection is up. A pre-established failure on an outbound
+    /// connection means the peer is unreachable (likely firewalled).
+    established: bool,
+    /// Set once this actor has reported a terminal outcome (disconnect or
+    /// connect-failure). Guards against a second notification — e.g. when a
+    /// replaced actor is later stopped — which would otherwise alias and evict
+    /// a newer actor registered under the same username.
+    disconnect_reported: bool,
 }
 
 impl PeerActor {
@@ -83,6 +91,8 @@ impl PeerActor {
             queued_messages: Vec::new(),
             own_username,
             outbound,
+            established: false,
+            disconnect_reported: false,
         }
     }
 
@@ -429,9 +439,22 @@ impl PeerActor {
 
         self.stream.take();
 
-        if let Err(e) = self.client_channel.send(
+        if self.disconnect_reported {
-            ClientOperation::PeerDisconnected(username, Some(error.into())),
+            return;
-        ) {
+        }
+        self.disconnect_reported = true;
+
+        // A direct outbound connection that never established means the peer is
+        // unreachable (likely firewalled): signal a connect failure so the
+        // client can fall back to server-brokered connect. Anything else is a
+        // normal disconnect.
+        let op = if self.outbound && !self.established {
+            ClientOperation::PeerConnectFailed(username)
+        } else {
+            ClientOperation::PeerDisconnected(username, Some(error.into()))
+        };
+
+        if let Err(e) = self.client_channel.send(op) {
             error!("Failed to send disconnect notification: {}", e);
         }
     }
@@ -441,6 +464,11 @@ impl PeerActor {
 
         self.stream.take();
 
+        if self.disconnect_reported {
+            return;
+        }
+        self.disconnect_reported = true;
+
         if let Err(e) = self
             .client_channel
             .send(ClientOperation::PeerDisconnected(username, None))
@@ -552,6 +580,8 @@ impl PeerActor {
             }
         };
 
+        self.established = true;
+
         let Some(ref mut stream) = self.stream else {
             return;
         };

soulseek-rs-lib/src/client.rs

@@ -18,7 +18,7 @@ use std::{
     net::TcpStream,
     sync::{
         RwLock,
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicBool, AtomicU32, Ordering},
         mpsc::{Receiver, Sender},
     },
     thread::{self, sleep},
@@ -31,6 +31,13 @@ use std::{
 use crate::{debug, error, info, trace, warn};
 const DEFAULT_LISTEN_PORT: u16 = 2234;
 
+/// Source of non-zero correlation tokens for server-brokered connections.
+static NEXT_CONNECT_TOKEN: AtomicU32 = AtomicU32::new(1);
+
+fn next_connect_token() -> u32 {
+    NEXT_CONNECT_TOKEN.fetch_add(1, Ordering::Relaxed).max(1)
+}
+
 #[derive(Debug, Clone)]
 pub struct ClientSettings {
     pub username: String,
@@ -98,6 +105,10 @@ pub enum ClientOperation {
     SetServerSender(Sender<ServerMessage>),
     PrivateMessageReceived(UserMessage),
     PeerConnected(String),
+    /// A direct outbound connection to this peer failed before it was
+    /// established — the peer is likely firewalled, so fall back to asking the
+    /// server to broker the connection.
+    PeerConnectFailed(String),
 }
 pub struct ClientContext {
     pub peer_registry: Option<PeerRegistry>,
@@ -106,6 +117,9 @@ pub struct ClientContext {
     server_sender: Option<Sender<ServerMessage>>,
     searches: HashMap<String, Search>,
     private_messages: Vec<UserMessage>,
+    /// Correlation tokens for server-brokered (firewalled) connections, mapping
+    /// a token we sent in a ConnectToPeer to the peer we expect back.
+    pending_connect_tokens: HashMap<u32, String>,
     actor_system: Arc<ActorSystem>,
 }
 impl Default for ClientContext {
@@ -333,11 +347,23 @@ impl ClientContext {
             server_sender: None,
             searches: HashMap::new(),
             private_messages: Vec::new(),
+            pending_connect_tokens: HashMap::new(),
             downloads: DownloadStore::new(),
             actor_system,
         }
     }
 
+    /// Remember that a server-brokered connection to `username` is pending under
+    /// `token`; the peer will quote it back in a PierceFirewall.
+    pub fn add_pending_connect(&mut self, token: u32, username: String) {
+        self.pending_connect_tokens.insert(token, username);
+    }
+
+    /// Resolve and consume the peer expected for a brokered connection `token`.
+    pub fn take_pending_connect(&mut self, token: u32) -> Option<String> {
+        self.pending_connect_tokens.remove(&token)
+    }
+
     /// Record a private message received from another user.
     pub fn push_private_message(&mut self, message: UserMessage) {
         self.private_messages.push(message);
@@ -1277,6 +1303,40 @@ impl Client {
                                     }
                                 }
                             }
+                            ClientOperation::PeerConnectFailed(username) => {
+                                // Direct connect failed: ask the server to
+                                // broker it. Register a correlation token, then
+                                // send ConnectToPeer so the (firewalled) peer
+                                // connects back to our listener quoting it.
+                                let token = next_connect_token();
+                                let server_sender = match client_context
+                                    .write_safe()
+                                {
+                                    Ok(mut ctx) => {
+                                        ctx.add_pending_connect(
+                                            token,
+                                            username.clone(),
+                                        );
+                                        ctx.server_sender.clone()
+                                    }
+                                    Err(e) => {
+                                        error!(
+                                            "[client] PeerConnectFailed write: {}",
+                                            e
+                                        );
+                                        continue;
+                                    }
+                                };
+                                if let Some(sender) = server_sender {
+                                    let msg = crate::message::server::MessageFactory::build_connect_to_peer(
+                                        token,
+                                        &username,
+                                        ConnectionType::P,
+                                    );
+                                    let _ = sender
+                                        .send(ServerMessage::SendMessage(msg));
+                                }
+                            }
                         }
                     }
                     Err(e) => {

soulseek-rs-lib/src/message/server/message_factory.rs

@@ -80,6 +80,22 @@ impl MessageFactory {
         Message::new().write_int32(23).write_int32(id).clone()
     }
 
+    /// Ask the server (code 18) to broker a connection to a firewalled peer:
+    /// the server tells that peer to connect back to us, quoting `token`.
+    #[must_use]
+    pub fn build_connect_to_peer(
+        token: u32,
+        username: &str,
+        connection_type: ConnectionType,
+    ) -> Message {
+        Message::new()
+            .write_int32(18)
+            .write_int32(token)
+            .write_string(username)
+            .write_string(&connection_type.to_string())
+            .clone()
+    }
+
     #[must_use]
     pub fn build_set_status_message(status_code: u32) -> Message {
         Message::new()

soulseek-rs-lib/src/peer/listen.rs

@@ -16,7 +16,6 @@ const PEER_INIT_MESSAGE_CODE: u8 = 1;
 
 #[derive(Clone)]
 struct ConnectionContext {
-    #[allow(dead_code)]
     client_sender: Sender<ClientOperation>,
     client_context: Arc<RwLock<ClientContext>>,
     own_username: String,
@@ -227,6 +226,54 @@ fn handle_file_connection(
     }
 }
 
+/// Handle a `PierceFirewall` (peer init code 0): a peer we asked the server to
+/// broker (because a direct connection failed) connecting back to us. It quotes
+/// the correlation token we registered; we register the stream as its P control
+/// connection and tell the client the connection is live.
+fn handle_pierce_firewall(
+    mut message: Message,
+    stream: TcpStream,
+    reader: MessageReader,
+    context: &ConnectionContext,
+    peer_ip: &str,
+    peer_port: u16,
+) {
+    message.set_pointer(5); // skip length prefix (4) + int8 code (1)
+    let token = message.read_int32();
+
+    let username = match context.client_context.write_safe() {
+        Ok(mut ctx) => ctx.take_pending_connect(token),
+        Err(e) => {
+            error!("[listener] pierce firewall lock: {}", e);
+            return;
+        }
+    };
+    let Some(username) = username else {
+        debug!(
+            "[listener:{peer_ip}:{peer_port}] PierceFirewall token {token} is not pending; ignoring"
+        );
+        return;
+    };
+
+    let peer = Peer::new(
+        username.clone(),
+        ConnectionType::P,
+        peer_ip.to_string(),
+        peer_port.into(),
+        None,
+        0,
+        0,
+        0,
+    );
+    handle_peer_connection(peer, stream, reader, context, peer_ip, peer_port);
+
+    // Inbound peers don't self-announce, so nudge the client to flush any
+    // downloads queued for this now-connected peer.
+    let _ = context
+        .client_sender
+        .send(ClientOperation::PeerConnected(username));
+}
+
 fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
     let Ok(peer_addr) = stream.peer_addr() else {
         error!("[listener] failed to get peer address");
@@ -245,6 +292,15 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
         return;
     };
 
+    // A firewalled peer brokered through the server connects back with a
+    // PierceFirewall (code 0) instead of a PeerInit (code 1).
+    if message.get_message_code() == 0 {
+        handle_pierce_firewall(
+            message, stream, reader, &context, &peer_ip, peer_port,
+        );
+        return;
+    }
+
     let Some(init_data) = parse_peer_init_message(message) else {
         error!(
             "[listener:{peer_ip}:{peer_port}] Invalid or unknown peer init message"

soulseek-rs-lib/tests/e2e.rs

@@ -27,6 +27,7 @@ use std::time::{Duration, Instant};
 
 use soulseek_rs::message::Message;
 use soulseek_rs::message::server::MessageFactory;
+use soulseek_rs::peer::ConnectionType;
 use soulseek_rs::{Client, ClientSettings, DownloadStatus, PeerAddress};
 
 /// A Soulseek server to test against: either a child soulfind process we
@@ -487,6 +488,60 @@ fn serve_file_over_f(
     Ok(())
 }
 
+/// Log a raw socket in to the server and drain up to the login response,
+/// returning the still-open stream (the user stays online while it lives).
+fn login_raw(
+    server_addr: &str,
+    username: &str,
+    password: &str,
+) -> std::io::Result<TcpStream> {
+    let mut srv = connect_retry(server_addr, Duration::from_secs(5))?;
+    srv.set_read_timeout(Some(Duration::from_secs(10)))?;
+    srv.write_all(
+        &MessageFactory::build_login_message(username, password).get_buffer(),
+    )?;
+    srv.flush()?;
+    loop {
+        let msg = read_framed(&mut srv)?;
+        if msg.get_message_code() == 1 {
+            break;
+        }
+    }
+    Ok(srv)
+}
+
+/// Read framed messages from `stream` until one has `code`, or the deadline
+/// passes. Returns the matching message.
+fn read_until_code(
+    stream: &mut TcpStream,
+    code: u8,
+    timeout: Duration,
+) -> Option<Message> {
+    let deadline = Instant::now() + timeout;
+    while Instant::now() < deadline {
+        match read_framed(stream) {
+            Ok(msg) if msg.get_message_code() == code => return Some(msg),
+            Ok(_) => {}
+            Err(_) => return None,
+        }
+    }
+    None
+}
+
+/// Like [`read_until_code`] but turns a miss into an `io::Error` for `?`.
+fn expect_code(
+    stream: &mut TcpStream,
+    code: u8,
+    timeout: Duration,
+) -> std::io::Result<Message> {
+    read_until_code(stream, code, timeout).ok_or_else(|| {
+        std::io::Error::new(
+            std::io::ErrorKind::TimedOut,
+            format!("timed out waiting for message code {code}"),
+        )
+    })
+}
+
 fn unique_download_dir() -> PathBuf {
     let dir = std::env::temp_dir().join(format!(
         "soulseek-e2e-dl-{}-{:?}",
@@ -584,6 +639,52 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
     let _ = std::fs::remove_dir_all(&download_dir);
 }
 
+#[test]
+fn soulfind_brokers_connect_to_peer_between_users() {
+    // The firewalled download mode relies on the server forwarding a
+    // ConnectToPeer request to the target user. Confirm soulfind does this: a
+    // requester (with a wait port so the server knows its address) asks the
+    // server to broker a connection to an online target, and the target must
+    // receive a forwarded ConnectToPeer (server code 18) naming the requester.
+    let server = server_or_skip!();
+    let addr = format!("{}:{}", server.host, server.port);
+
+    let mut target =
+        login_raw(&addr, "e2e_broker_target", "pw").expect("target login");
+
+    let mut requester =
+        login_raw(&addr, "e2e_broker_req", "pw").expect("requester login");
+    let req_port = free_port().expect("free port");
+    requester
+        .write_all(
+            &MessageFactory::build_set_wait_port_message(req_port).get_buffer(),
+        )
+        .expect("set wait port");
+    requester.flush().expect("flush wait port");
+
+    let token = 987_654_u32;
+    requester
+        .write_all(
+            &MessageFactory::build_connect_to_peer(
+                token,
+                "e2e_broker_target",
+                ConnectionType::P,
+            )
+            .get_buffer(),
+        )
+        .expect("send ConnectToPeer");
+    requester.flush().expect("flush ConnectToPeer");
+
+    let mut brokered = read_until_code(&mut target, 18, Duration::from_secs(5))
+        .expect("target should receive a brokered ConnectToPeer");
+    brokered.set_pointer(8);
+    assert_eq!(
+        brokered.read_string(),
+        "e2e_broker_req",
+        "the brokered message should name the requester"
+    );
+}
+
 // ---------------------------------------------------------------------------
 // Direct-connection download: the client initiates the peer connection.
 //
@@ -783,3 +884,162 @@ fn a_file_downloads_from_a_peer_via_direct_connection() {
 
     let _ = std::fs::remove_dir_all(&download_dir);
 }
+
+// ---------------------------------------------------------------------------
+// Firewalled download: the peer is unreachable directly, so the connection is
+// brokered through the server.
+//
+// The mock advertises a port nobody listens on, so the downloader's direct
+// connection fails. The client then asks the server to broker the connection
+// (ConnectToPeer); the mock, reading its server stream, sees the forwarded
+// request and connects back to the downloader with a PierceFirewall. That
+// pierced connection becomes the P control channel, and the file is served.
+// ---------------------------------------------------------------------------
+
+struct MockFirewalledUpload {
+    server_addr: String,
+    username: String,
+    password: String,
+    bogus_port: u16,
+    downloader_listen_addr: String,
+    filename: String,
+    content: Vec<u8>,
+    token: u32,
+    ready: Sender<()>,
+}
+
+fn run_mock_firewalled_peer(cfg: &MockFirewalledUpload) -> std::io::Result<()> {
+    // 1. Log in and advertise a port that nobody listens on, so the downloader's
+    //    direct connection is refused and it falls back to server brokering.
+    let mut srv = login_raw(&cfg.server_addr, &cfg.username, &cfg.password)?;
+    srv.write_all(
+        &MessageFactory::build_set_wait_port_message(cfg.bogus_port)
+            .get_buffer(),
+    )?;
+    srv.flush()?;
+    let _ = cfg.ready.send(());
+
+    // 2. Wait for the server-brokered ConnectToPeer (server code 18) and read
+    //    the correlation token (after username, type, ip and port).
+    let mut ctp = expect_code(&mut srv, 18, Duration::from_secs(15))?;
+    ctp.set_pointer(8);
+    let _who = ctp.read_string();
+    let _conn_type = ctp.read_string();
+    let _ip = ctp.read_int32();
+    let _port = ctp.read_int32();
+    let connect_token = ctp.read_int32();
+
+    // 3. Connect back to the downloader with a PierceFirewall (peer code 0);
+    //    this becomes the P control connection.
+    let mut p =
+        connect_retry(&cfg.downloader_listen_addr, Duration::from_secs(5))?;
+    p.set_read_timeout(Some(Duration::from_secs(10)))?;
+    p.write_all(
+        &MessageFactory::build_pierce_firewall_message(connect_token)
+            .get_buffer(),
+    )?;
+    p.flush()?;
+
+    // 4. Negotiate the transfer over the pierced connection.
+    let _queue = expect_code(&mut p, 43, Duration::from_secs(10))?;
+    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()?;
+    let _response = expect_code(&mut p, 41, Duration::from_secs(10))?;
+
+    // 5. Serve the bytes over an F connection to the downloader's listener.
+    serve_file_over_f(
+        &cfg.downloader_listen_addr,
+        &cfg.username,
+        cfg.token,
+        &cfg.content,
+    )
+}
+
+#[test]
+fn a_file_downloads_from_a_firewalled_peer_via_server_broker() {
+    let server = server_or_skip!();
+
+    let client_port = free_port().expect("free client listen port");
+    let mut client = Client::with_settings(server.listening_settings(
+        "e2e_fw_dl",
+        "pw",
+        client_port,
+    ));
+    client.connect().expect("connect");
+    assert!(client.login().expect("login"));
+
+    let bogus_port = free_port().expect("bogus port"); // advertised, unlistened
+    let filename = "firewalled_song.mp3";
+    let content: Vec<u8> = (0..2000u32).map(|i| (i % 251) as u8).collect();
+    let size = content.len() as u64;
+    let token = 606_060_u32;
+    let download_dir = unique_download_dir();
+
+    let (ready_tx, ready_rx) = std::sync::mpsc::channel();
+    let cfg = MockFirewalledUpload {
+        server_addr: format!("{}:{}", server.host, server.port),
+        username: "e2e_fw_peer".to_string(),
+        password: "pw".to_string(),
+        bogus_port,
+        downloader_listen_addr: format!("127.0.0.1:{client_port}"),
+        filename: filename.to_string(),
+        content: content.clone(),
+        token,
+        ready: ready_tx,
+    };
+    let uploader = std::thread::spawn(move || {
+        if let Err(e) = run_mock_firewalled_peer(&cfg) {
+            eprintln!("[mock firewalled peer] {e}");
+        }
+    });
+
+    ready_rx
+        .recv_timeout(Duration::from_secs(10))
+        .expect("mock firewalled peer ready");
+    std::thread::sleep(Duration::from_millis(1000));
+
+    let (_download, status_rx) = client
+        .download(
+            filename.to_string(),
+            "e2e_fw_peer".to_string(),
+            size,
+            download_dir.display().to_string(),
+        )
+        .expect("start download");
+
+    let mut completed = false;
+    let deadline = Instant::now() + Duration::from_secs(25);
+    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 firewalled 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);
+}