fix(client): survive the review findings on the thread-flood fix

Review of the capacity work surfaced three real defects and two gaps:

- A peer dribbling bytes re-armed the 10s handshake read timeout on
  every read and held its accept permit forever; 128 such sockets
  wedged the accept loop for good. The handshake now has one deadline
  across all reads, plus a 4 KiB cap on buffered handshake bytes.
- A capacity refusal was logged and swallowed, leaving the refused
  peer's queued downloads stuck until the caller's own timeout. Both
  registration paths now fail those downloads so the caller gets a
  verdict, and the listener only announces PeerConnected when the
  registration actually happened.
- The transfer-claim was undone by the worker thread writing Queued
  back over it; that write is gone, so a replayed TransferResponse
  can no longer start a second writer on the same .part file.
- Offered uploads and queue waiters joined the eviction-protected
  set, and the browse protection window matches the minutes a real
  share listing takes rather than one.

Regression tests: the broker-timeout sweep failing a queued download
end to end, the replay guard observed over a real socket, and the
handshake deadline against a dribbling and an oversized peer.

soulseek-rs-lib/src/client/connection.rs

@@ -214,6 +214,7 @@ impl Client {
             ConnectionType::P => {
                 let username = peer.username;
 
+                let refused = {
                     let context = match client_context.read_safe() {
                         Ok(c) => c,
                         Err(e) => {
@@ -226,16 +227,23 @@ impl Client {
                         match registry.register_peer_protected(
                             peer_clone, stream, None, &protected,
                         ) {
-                        Ok(_) => (),
+                            Ok(_) => false,
                             Err(e) => {
-                            trace!(
+                                warn!(
-                                "Failed to spawn peer actor for {:?}: {:?}",
+                                    "[client] refusing peer connection for \
+                                     {:?}: {:?}",
                                     username, e
                                 );
+                                true
                             }
                         }
                     } else {
                         trace!("PeerRegistry not initialized");
+                        false
+                    }
+                };
+                if refused {
+                    Self::fail_queued_downloads(&client_context, &username);
                 }
             }
 

soulseek-rs-lib/src/client/mod.rs

@@ -60,7 +60,7 @@ const BROKER_CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
 
 const DEFAULT_MAX_PEERS: usize = 512;
 
-const BROWSE_PROTECT_WINDOW: Duration = Duration::from_mins(1);
+const BROWSE_PROTECT_WINDOW: Duration = Duration::from_mins(5);
 
 /// Source of non-zero correlation tokens for server-brokered connections.
 static NEXT_CONNECT_TOKEN: AtomicU32 = AtomicU32::new(1);
@@ -634,6 +634,9 @@ impl ClientContext {
             .collect();
         protected
             .extend(self.active_uploads.values().map(|u| u.username.clone()));
+        protected.extend(self.uploads.values().map(|j| j.downloader.clone()));
+        protected
+            .extend(self.upload_queue.iter().map(|q| q.downloader.clone()));
         protected.extend(self.pending_serves.keys().cloned());
         protected.extend(self.pending_peer_messages.keys().cloned());
         let now = Instant::now();

soulseek-rs-lib/src/client/tests.rs

@@ -443,3 +443,98 @@ fn protected_peers_covers_downloads_uploads_and_pending_serves() {
     context.store_browse_result("browsed".to_string(), Vec::new());
     assert!(!context.protected_peers().contains("browsed"));
 }
+
+#[test]
+fn an_expired_broker_connect_fails_the_queued_downloads() {
+    let client = Client::new("u", "p");
+    let (sender, receiver) = mpsc::channel();
+    {
+        let mut ctx = client.context.write().unwrap();
+        ctx.add_download(download(
+            "ghost",
+            "f.mp3",
+            7,
+            DownloadStatus::Queued,
+            sender,
+        ));
+        ctx.pending_connect_tokens
+            .insert(7, ("ghost".to_string(), Instant::now()));
+    }
+    let (_ops_tx, ops_rx) = mpsc::channel();
+    Client::listen_to_client_operations(
+        ops_rx,
+        client.context,
+        "u".to_string(),
+    );
+
+    let status = receiver.recv_timeout(Duration::from_secs(5));
+    assert!(
+        matches!(status, Ok(DownloadStatus::Failed(_))),
+        "the sweep must fail the queued download, got {status:?}"
+    );
+}
+
+#[test]
+fn a_replayed_transfer_response_does_not_start_a_second_transfer() {
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    listener.set_nonblocking(true).unwrap();
+    let port = u32::from(listener.local_addr().unwrap().port());
+
+    let client = Client::new("u", "p");
+    let (sender, _receiver) = mpsc::channel();
+    client.context.write().unwrap().add_download(download(
+        "peer",
+        "f.mp3",
+        9,
+        DownloadStatus::Queued,
+        sender,
+    ));
+
+    let (ops_tx, ops_rx) = mpsc::channel();
+    Client::listen_to_client_operations(
+        ops_rx,
+        client.context,
+        "u".to_string(),
+    );
+
+    let peer = Peer::new(
+        "peer".to_string(),
+        ConnectionType::F,
+        "127.0.0.1".to_string(),
+        port,
+        None,
+        0,
+        0,
+        0,
+    );
+    ops_tx
+        .send(ClientOperation::DownloadFromPeer(9, peer.clone(), true))
+        .unwrap();
+    ops_tx
+        .send(ClientOperation::DownloadFromPeer(9, peer, true))
+        .unwrap();
+
+    let deadline = Instant::now() + Duration::from_secs(5);
+    let mut connections = 0;
+    while Instant::now() < deadline {
+        match listener.accept() {
+            Ok(_) => {
+                connections += 1;
+                if connections == 1 {
+                    thread::sleep(Duration::from_millis(500));
+                }
+            }
+            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                if connections >= 1 {
+                    break;
+                }
+                thread::sleep(Duration::from_millis(20));
+            }
+            Err(_) => break,
+        }
+    }
+    assert_eq!(
+        connections, 1,
+        "a replayed TransferResponse must not dial the peer again"
+    );
+}

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

@@ -484,14 +484,6 @@ impl DownloadPeer {
             self.no_pierce
         );
 
-        if let Some(ref dl) = download {
-            let _ = dl.sender.send(DownloadStatus::Queued);
-            client_context
-                .write()
-                .map_err(|_| DownloadError::LockPoisoned)?
-                .update_download_with_status(dl.token, DownloadStatus::Queued);
-        }
-
         let mut stream = match stream {
             Some(s) => {
                 Self::apply_transfer_deadlines(&s)?;

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

@@ -3,16 +3,16 @@ use std::net::{TcpListener, TcpStream};
 use std::sync::mpsc::Sender;
 use std::sync::{Arc, RwLock};
 use std::thread;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
-use crate::client::{ClientContext, ClientOperation};
+use crate::client::{Client, ClientContext, ClientOperation};
 
 use crate::message::{Message, MessageReader};
 use crate::peer::{ConnectionType, DownloadPeer, Peer};
 use crate::types::Download;
 use crate::utils::lock::RwLockExt;
 use crate::utils::semaphore::{Permit, Semaphore};
-use crate::{DownloadStatus, debug, error, info, trace};
+use crate::{DownloadStatus, debug, error, info, trace, warn};
 
 /// How long to wait before accepting again after a failure.
 ///
@@ -28,6 +28,8 @@ const PEER_INIT_TIMEOUT: Duration = Duration::from_secs(10);
 
 const MAX_HANDSHAKES: usize = 128;
 
+const MAX_PEER_INIT_BYTES: usize = 4096;
+
 #[derive(Clone)]
 struct ConnectionContext {
     client_sender: Sender<ClientOperation>,
@@ -44,16 +46,30 @@ struct PeerInitData {
 fn read_peer_init_message(
     stream: &mut TcpStream,
     reader: &mut MessageReader,
+    deadline: Instant,
 ) -> io::Result<Message> {
     // An untrusted peer gets a bounded handshake. Without this a peer that
     // connects and stays silent parks this read forever, pinning a thread that
     // owes us a peer init.
-    stream.set_read_timeout(Some(PEER_INIT_TIMEOUT))?;
     let message = loop {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                "peer init deadline passed",
+            ));
+        }
+        stream.set_read_timeout(Some(remaining))?;
         reader.read_from_socket(stream)?;
         if let Ok(Some(msg)) = reader.extract_message() {
             break msg;
         }
+        if reader.buffer_len() > MAX_PEER_INIT_BYTES {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "peer init larger than any handshake",
+            ));
+        }
     };
 
     // Whatever this connection turns into reads on its own terms from here.
@@ -122,7 +138,7 @@ fn handle_peer_connection(
     stream: TcpStream,
     reader: MessageReader,
     context: &ConnectionContext,
-) {
+) -> bool {
     // 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
@@ -131,15 +147,16 @@ fn handle_peer_connection(
     // 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;
+        return false;
     }
     stream.set_nodelay(true).ok();
 
+    let registered = {
         let client_context = match context.client_context.read_safe() {
             Ok(c) => c,
             Err(e) => {
                 error!("[listener] handle_peer_connection lock: {}", e);
-            return;
+                return false;
             }
         };
         if let Some(ref registry) = client_context.peer_registry {
@@ -150,17 +167,24 @@ fn handle_peer_connection(
                 Some(reader),
                 &protected,
             ) {
-            Ok(_) => (),
+                Ok(_) => true,
                 Err(e) => {
-                error!(
+                    warn!(
-                    "Failed to spawn peer actor for {:?}: {:?}",
+                        "[listener] refusing peer connection for {:?}: {:?}",
                         peer.username, e
                     );
+                    false
                 }
             }
         } else {
             error!("PeerRegistry not initialized");
+            false
         }
+    };
+    if !registered {
+        Client::fail_queued_downloads(&context.client_context, &peer.username);
+    }
+    registered
 }
 
 fn handle_file_connection(
@@ -280,13 +304,13 @@ fn handle_pierce_firewall(
         0,
         0,
     );
-    handle_peer_connection(peer, stream, reader, context);
+    if handle_peer_connection(peer, stream, reader, context) {
-
         // 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(
@@ -304,7 +328,9 @@ fn handle_incoming_connection(
     let mut reader = MessageReader::new();
 
     // A peer that dials and then goes away is routine, not an error.
-    let message = match read_peer_init_message(&mut stream, &mut reader) {
+    let deadline = Instant::now() + PEER_INIT_TIMEOUT;
+    let message =
+        match read_peer_init_message(&mut stream, &mut reader, deadline) {
             Ok(message) => message,
             Err(e) => {
                 debug!(
@@ -354,14 +380,15 @@ fn handle_incoming_connection(
 
     match init_data.connection_type {
         ConnectionType::P => {
-            handle_peer_connection(peer, stream, reader, &context);
+            if handle_peer_connection(peer, stream, reader, &context) {
-            // Inbound peers don't self-announce, so nudge the client to flush
+                // Inbound peers don't self-announce, so nudge the client to
-            // anything queued for this now-connected peer — same as the
+                // flush anything queued for this now-connected peer — same as
-            // PierceFirewall path.
+                // the PierceFirewall path.
                 let _ = context
                     .client_sender
                     .send(ClientOperation::PeerConnected(init_data.username));
             }
+        }
 
         ConnectionType::F => handle_file_connection(
             peer,
@@ -444,3 +471,68 @@ impl Listen {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use std::net::TcpListener;
+
+    #[test]
+    fn a_dribbling_handshake_hits_the_total_deadline() {
+        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();
+            let _ = socket.write_all(&1000u32.to_le_bytes());
+            for _ in 0..20 {
+                if socket.write_all(&[0u8]).is_err() {
+                    return;
+                }
+                std::thread::sleep(Duration::from_millis(50));
+            }
+        });
+        let (mut stream, _) = listener.accept().unwrap();
+        let mut reader = MessageReader::new();
+
+        let started = Instant::now();
+        let result = read_peer_init_message(
+            &mut stream,
+            &mut reader,
+            started + Duration::from_millis(300),
+        );
+
+        assert!(result.is_err(), "a dribbled handshake must not succeed");
+        assert!(
+            started.elapsed() < Duration::from_secs(5),
+            "the deadline must bound the whole handshake, not each read"
+        );
+        let _ = writer.join();
+    }
+
+    #[test]
+    fn an_oversized_handshake_is_rejected() {
+        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();
+            let junk = vec![0xFFu8; MAX_PEER_INIT_BYTES * 2];
+            let _ = socket.write_all(&junk);
+            std::thread::sleep(Duration::from_millis(500));
+        });
+        let (mut stream, _) = listener.accept().unwrap();
+        let mut reader = MessageReader::new();
+
+        let result = read_peer_init_message(
+            &mut stream,
+            &mut reader,
+            Instant::now() + Duration::from_secs(5),
+        );
+
+        assert!(
+            result.is_err(),
+            "an oversized handshake must be rejected, got {result:?}"
+        );
+        let _ = writer.join();
+    }
+}