From 492b96176670f30325b713be91b154288482aec6 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Thu, 13 Aug 2026 14:10:20 +0200 Subject: [PATCH] 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 | 48 +++--- soulseek-rs-lib/src/client/mod.rs | 5 +- soulseek-rs-lib/src/client/tests.rs | 95 +++++++++++ soulseek-rs-lib/src/peer/download_peer.rs | 8 - soulseek-rs-lib/src/peer/listen.rs | 196 ++++++++++++++++------ 5 files changed, 271 insertions(+), 81 deletions(-) diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index bc4a6a4..c9219f9 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -214,28 +214,36 @@ impl Client { ConnectionType::P => { let username = peer.username; - let context = match client_context.read_safe() { - Ok(c) => c, - Err(e) => { - error!("[client] connect_to_peer read: {}", e); - return; + let refused = { + let context = match client_context.read_safe() { + Ok(c) => c, + Err(e) => { + error!("[client] connect_to_peer read: {}", e); + return; + } + }; + if let Some(ref registry) = context.peer_registry { + let protected = context.protected_peers(); + match registry.register_peer_protected( + peer_clone, stream, None, &protected, + ) { + Ok(_) => false, + Err(e) => { + warn!( + "[client] refusing peer connection for \ + {:?}: {:?}", + username, e + ); + true + } + } + } else { + trace!("PeerRegistry not initialized"); + false } }; - if let Some(ref registry) = context.peer_registry { - let protected = context.protected_peers(); - match registry.register_peer_protected( - peer_clone, stream, None, &protected, - ) { - Ok(_) => (), - Err(e) => { - trace!( - "Failed to spawn peer actor for {:?}: {:?}", - username, e - ); - } - } - } else { - trace!("PeerRegistry not initialized"); + if refused { + Self::fail_queued_downloads(&client_context, &username); } } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index b935ee9..dd9cf09 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/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(); diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 9b5efcb..23e8ca4 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/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" + ); +} diff --git a/soulseek-rs-lib/src/peer/download_peer.rs b/soulseek-rs-lib/src/peer/download_peer.rs index 3ea609e..accc2b0 100644 --- a/soulseek-rs-lib/src/peer/download_peer.rs +++ b/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)?; diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 739e3a0..070df8f 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/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, @@ -44,16 +46,30 @@ struct PeerInitData { fn read_peer_init_message( stream: &mut TcpStream, reader: &mut MessageReader, + deadline: Instant, ) -> io::Result { // 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,36 +147,44 @@ 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 client_context = match context.client_context.read_safe() { - Ok(c) => c, - Err(e) => { - error!("[listener] handle_peer_connection lock: {}", e); - return; + let registered = { + let client_context = match context.client_context.read_safe() { + Ok(c) => c, + Err(e) => { + error!("[listener] handle_peer_connection lock: {}", e); + return false; + } + }; + if let Some(ref registry) = client_context.peer_registry { + let protected = client_context.protected_peers(); + match registry.register_peer_protected( + peer.clone(), + Some(stream), + Some(reader), + &protected, + ) { + Ok(_) => true, + Err(e) => { + warn!( + "[listener] refusing peer connection for {:?}: {:?}", + peer.username, e + ); + false + } + } + } else { + error!("PeerRegistry not initialized"); + false } }; - if let Some(ref registry) = client_context.peer_registry { - let protected = client_context.protected_peers(); - match registry.register_peer_protected( - peer.clone(), - Some(stream), - Some(reader), - &protected, - ) { - Ok(_) => (), - Err(e) => { - error!( - "Failed to spawn peer actor for {:?}: {:?}", - peer.username, e - ); - } - } - } else { - error!("PeerRegistry not initialized"); + 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); - - // 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)); + 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,15 +328,17 @@ 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) { - Ok(message) => message, - Err(e) => { - debug!( - "[listener:{peer_ip}:{peer_port}] no peer init message: {e}" - ); - return; - } - }; + 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!( + "[listener:{peer_ip}:{peer_port}] no peer init message: {e}" + ); + return; + } + }; drop(permit); // A firewalled peer brokered through the server connects back with a @@ -354,13 +380,14 @@ fn handle_incoming_connection( match init_data.connection_type { ConnectionType::P => { - handle_peer_connection(peer, stream, reader, &context); - // Inbound peers don't self-announce, so nudge the client to flush - // anything queued for this now-connected peer — same as the - // PierceFirewall path. - let _ = context - .client_sender - .send(ClientOperation::PeerConnected(init_data.username)); + if handle_peer_connection(peer, stream, reader, &context) { + // Inbound peers don't self-announce, so nudge the client to + // flush anything queued for this now-connected peer — same as + // the PierceFirewall path. + let _ = context + .client_sender + .send(ClientOperation::PeerConnected(init_data.username)); + } } ConnectionType::F => handle_file_connection( @@ -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(); + } +}