From ea168605b53a7bcbcd7bebe02cd39bec595f9742 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 00:35:11 +0200 Subject: [PATCH 1/5] fix(lib): keep a replaced peer connection open for the reply in flight Two connections to the same peer race whenever both sides dial at once: we open one to send a browse, the peer opens one to deliver a search reply. The registry keeps one actor per username, and until now stopped the older one on the spot, which closed the socket carrying our browse before the listing could arrive. Under load the stress gate lost up to two thirds of its browses that way. A replaced actor is now retired instead: it sends nothing new, keeps reading for thirty seconds so a reply already on the wire still reaches the client, and then stops itself. The peer closing a retired connection is expected and reported as a clean close, so it cannot fail transfers running on the connection that took over. --- soulseek-rs-lib/src/actor/peer_actor.rs | 87 +++++++++++++++++++++- soulseek-rs-lib/src/actor/peer_registry.rs | 10 +-- 2 files changed, 91 insertions(+), 6 deletions(-) diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index faf21e2..d0f2c8c 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -61,6 +61,9 @@ pub enum PeerMessage { size: u64, }, ProcessRead, + /// A newer connection to this peer took over. Deliver what is already + /// on the wire, then close. + Retire, } pub struct PeerActor { @@ -101,6 +104,10 @@ pub struct PeerActor { /// control connection, and each one holds a descriptor and a thread for /// the life of the process. last_activity: Instant, + /// Set once a newer connection to the same peer replaced this one. The + /// stream stays open until then so a reply already in flight, a browse + /// listing say, still arrives; past it the actor closes and stops. + retire_deadline: Option, } /// Cap on queued outbound bytes. A peer that stops reading must not be able to @@ -112,6 +119,8 @@ const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024; /// simply reconnects — every serious client reaps idle peers this way. Five /// minutes comfortably outlives a search's chatter and queue updates. const IDLE_DISCONNECT: Duration = Duration::from_mins(5); +/// How long a replaced connection keeps listening for replies in flight. +const RETIRE_GRACE: Duration = Duration::from_secs(30); impl PeerActor { #[must_use] @@ -148,6 +157,7 @@ impl PeerActor { serving_tokens: std::collections::HashSet::new(), pending_write: Vec::new(), last_activity: Instant::now(), + retire_deadline: None, } } @@ -229,6 +239,9 @@ impl PeerActor { PeerMessage::SendMessage(message) => { self.send_message(message); } + PeerMessage::Retire => { + self.retire_deadline = Some(Instant::now() + RETIRE_GRACE); + } PeerMessage::FileSearchResult(file_search) => { self.handle_file_search_result(file_search); } @@ -664,6 +677,12 @@ impl PeerActor { } fn disconnect_with_error(&mut self, error: Error) { + // A retired connection is expected to go: the peer closing it is not + // evidence the peer is gone, and must not fail its transfers on the + // connection that replaced this one. + if self.retire_deadline.is_some() { + return self.disconnect(); + } let username = self.peer_username(); // Include the cause: a failed outbound dial reports nothing else, so // without it an unreachable peer is indistinguishable from a clean @@ -919,9 +938,24 @@ impl Actor for PeerActor { ); self.disconnect(); } + if self.stream.is_some() + && self + .retire_deadline + .is_some_and(|d| d <= Instant::now()) + { + self.disconnect(); + } + } + } + // Nothing in the registry points at a retired actor any more, so + // it is the one to stop itself. + ConnectionState::Disconnected => { + if self.retire_deadline.is_some() + && let Some(handle) = &self.self_handle + { + let _ = handle.stop(); } } - ConnectionState::Disconnected => {} } } } @@ -929,6 +963,7 @@ impl Actor for PeerActor { #[cfg(test)] mod tests { use super::*; + use crate::message::peer::build_shared_file_list; use crate::peer::{ConnectionType, Peer}; use std::net::{TcpListener, TcpStream}; use std::sync::mpsc::Receiver; @@ -1024,4 +1059,54 @@ mod tests { other => panic!("expected a clean PeerDisconnected, got {other:?}"), } } + + // A connection replaced by a newer one to the same peer may still carry + // a reply already on the wire: it is retired, not cut. + #[test] + fn a_retired_actor_delivers_what_is_on_the_wire_then_closes() { + use std::io::Write; + let (mut actor, rx, mut far_end) = connected_actor(); + + actor.handle_message(PeerMessage::Retire); + far_end + .write_all(&build_shared_file_list(&[]).get_buffer()) + .unwrap(); + std::thread::sleep(Duration::from_millis(50)); + actor.tick(); + + assert!( + matches!(rx.try_recv(), Ok(ClientOperation::BrowseResult { .. })), + "the listing on the wire still reaches the client" + ); + assert!(actor.stream.is_some(), "retiring is not yet closing"); + + actor.retire_deadline = + Some(Instant::now().checked_sub(Duration::from_secs(1)).unwrap()); + actor.tick(); + + assert!(actor.stream.is_none(), "past the grace the stream closes"); + assert!(matches!( + rx.try_recv(), + Ok(ClientOperation::PeerDisconnected(7, _, None)) + )); + } + + #[test] + fn a_retired_actor_closed_by_the_peer_reports_a_clean_close() { + let (mut actor, rx, far_end) = connected_actor(); + + actor.handle_message(PeerMessage::Retire); + drop(far_end); + std::thread::sleep(Duration::from_millis(50)); + actor.tick(); + + assert!(actor.stream.is_none()); + assert!( + matches!( + rx.try_recv(), + Ok(ClientOperation::PeerDisconnected(7, _, None)) + ), + "the peer dropping a retired connection is not an error" + ); + } } diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index 0ff9230..f4b7007 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -126,14 +126,14 @@ impl PeerRegistry { actor.set_self_handle(handle); }) .map_err(|e| format!("failed to spawn peer actor thread: {e}"))?; - // Stop any actor already registered under this username so it does not - // become an orphan pinning a pool worker forever. Eviction on the - // replaced actor's later shutdown is identity-aware (keyed on its id), - // so stopping it here cannot evict this new connection. + // Retire any actor already registered under this username rather than + // stop it: a reply to a request we sent on it may still be on the + // wire. It stops itself once that grace is over, and the eviction its + // shutdown triggers is keyed on its id, so it cannot evict this one. if let Some((_, old_handle)) = peers.insert(username.clone(), (id, handle.clone())) { - let _ = old_handle.stop(); + let _ = old_handle.send(PeerMessage::Retire); debug!( "[peer_registry] Replaced existing peer actor for {}", username From 3f1494a4f940ba69c17ccc9d7b26afb4032e9a2e Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 00:35:11 +0200 Subject: [PATCH 2/5] fix(stress): serve a browse that lands while an offer awaits its answer The mock seeder drained every message but the TransferResponse while it waited for the client to allow an upload, so a browse arriving in that window was silently dropped and counted against the client. --- soulseek-rs-lib/examples/stress.rs | 42 +++++++++++++++++++----------- 1 file changed, 27 insertions(+), 15 deletions(-) diff --git a/soulseek-rs-lib/examples/stress.rs b/soulseek-rs-lib/examples/stress.rs index 5c0d30e..56b4321 100644 --- a/soulseek-rs-lib/examples/stress.rs +++ b/soulseek-rs-lib/examples/stress.rs @@ -496,18 +496,7 @@ impl SeederState { Err(_) => return, }; match msg.get_message_code() { - // GetShareFileList: reply with everything we hold. - 4 => { - let dir = SharedDirectory { - name: format!("stress\\{}", self.name), - files: self.files.clone(), - }; - let list = - build_shared_file_list(std::slice::from_ref(&dir)); - if !self.reply(&mut p, shared, &list.get_buffer()) { - return; - } - } + 4 if !self.send_listing(&mut p, shared) => return, // QueueUpload: offer the transfer, then stream it once allowed. 43 => { msg.set_pointer(8); @@ -527,7 +516,7 @@ impl SeederState { if !self.reply(&mut p, shared, &tr.get_buffer()) { return; } - if !self.wait_for_allow(&mut p, token) { + if !self.wait_for_allow(&mut p, shared, token) { return; } let content = self.content.clone(); @@ -543,8 +532,25 @@ impl SeederState { } } - /// Wait for the client's `TransferResponse` allowing `token`. - fn wait_for_allow(&self, p: &mut TcpStream, token: u32) -> bool { + /// Everything we hold, the answer to a GetShareFileList. + fn send_listing(&self, p: &mut TcpStream, shared: bool) -> bool { + let dir = SharedDirectory { + name: format!("stress\\{}", self.name), + files: self.files.clone(), + }; + let list = build_shared_file_list(std::slice::from_ref(&dir)); + self.reply(p, shared, &list.get_buffer()) + } + + /// Wait for the client's `TransferResponse` allowing `token`. A browse + /// that arrives meanwhile is answered: a busy client accepts an offer + /// late, and a browse it sent in between is not lost work on its side. + fn wait_for_allow( + &self, + p: &mut TcpStream, + shared: bool, + token: u32, + ) -> bool { let deadline = Instant::now() + Duration::from_secs(30); while Instant::now() < deadline && !self.stop.load(Ordering::Relaxed) { match read_framed(p) { @@ -554,6 +560,12 @@ impl SeederState { return true; } } + Ok(msg) + if msg.get_message_code() == 4 + && !self.send_listing(p, shared) => + { + return false; + } Ok(_) => {} Err(ref e) if is_timeout(e) => {} Err(_) => return false, From 2d4a6fdd5184e123a1b160103157cff2277bd2ce Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 00:36:03 +0200 Subject: [PATCH 3/5] refactor(lib): move the peer actor's tests to their own file --- soulseek-rs-lib/src/actor/peer_actor.rs | 150 +----------------- soulseek-rs-lib/src/actor/peer_actor/tests.rs | 147 +++++++++++++++++ 2 files changed, 148 insertions(+), 149 deletions(-) create mode 100644 soulseek-rs-lib/src/actor/peer_actor/tests.rs diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index d0f2c8c..241fbec 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -961,152 +961,4 @@ impl Actor for PeerActor { } #[cfg(test)] -mod tests { - use super::*; - use crate::message::peer::build_shared_file_list; - use crate::peer::{ConnectionType, Peer}; - use std::net::{TcpListener, TcpStream}; - use std::sync::mpsc::Receiver; - - /// A connected inbound actor over a real loopback socket, plus the far - /// end (kept alive so reads see silence, not EOF). - fn connected_actor() -> (PeerActor, Receiver, TcpStream) { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); - let addr = listener.local_addr().unwrap(); - let stream = TcpStream::connect(addr).unwrap(); - stream.set_nonblocking(true).unwrap(); - let (far_end, _) = listener.accept().unwrap(); - - let (mut actor, rx) = make_actor(Some(stream)); - actor.on_start(); - (actor, rx, far_end) - } - - fn make_actor( - stream: Option, - ) -> (PeerActor, Receiver) { - let (tx, rx) = std::sync::mpsc::channel(); - let peer = Peer::new( - "bob".to_string(), - ConnectionType::P, - "127.0.0.1".to_string(), - 0, - None, - 0, - 0, - 0, - ); - let actor = PeerActor::new(peer, stream, None, tx, "me".to_string(), 7); - (actor, rx) - } - - #[test] - fn a_timed_out_connect_parks_the_actor_in_disconnected() { - let (mut actor, rx) = make_actor(None); - actor.connection_state = ConnectionState::Connecting { - since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(), - }; - - actor.tick(); - - assert!( - matches!(actor.connection_state, ConnectionState::Disconnected), - "a timed-out connect must leave Connecting" - ); - match rx.try_recv() { - Ok(ClientOperation::PeerConnectFailed(7, username)) => { - assert_eq!(username, "bob"); - } - other => panic!("expected PeerConnectFailed, got {other:?}"), - } - } - - #[test] - fn an_upload_failed_message_reaches_the_client_as_this_peer() { - let (mut actor, rx, _far_end) = connected_actor(); - - actor.handle_message(PeerMessage::UploadFailed( - String::new(), - "song.mp3".to_string(), - )); - - match rx.try_recv() { - Ok(ClientOperation::UploadFailed(username, filename)) => { - assert_eq!(username, "bob"); - assert_eq!(filename, "song.mp3"); - } - other => panic!("expected UploadFailed, got {other:?}"), - } - } - - #[test] - fn tick_reaps_a_peer_idle_past_the_deadline() { - let (mut actor, rx, _far_end) = connected_actor(); - - actor.tick(); - assert!(actor.stream.is_some(), "a fresh connection stays open"); - - actor.last_activity = Instant::now() - .checked_sub(IDLE_DISCONNECT + Duration::from_secs(1)) - .unwrap(); - actor.tick(); - - assert!(actor.stream.is_none(), "an idle stream must be closed"); - match rx.try_recv() { - Ok(ClientOperation::PeerDisconnected(7, username, None)) => { - assert_eq!(username, "bob"); - } - other => panic!("expected a clean PeerDisconnected, got {other:?}"), - } - } - - // A connection replaced by a newer one to the same peer may still carry - // a reply already on the wire: it is retired, not cut. - #[test] - fn a_retired_actor_delivers_what_is_on_the_wire_then_closes() { - use std::io::Write; - let (mut actor, rx, mut far_end) = connected_actor(); - - actor.handle_message(PeerMessage::Retire); - far_end - .write_all(&build_shared_file_list(&[]).get_buffer()) - .unwrap(); - std::thread::sleep(Duration::from_millis(50)); - actor.tick(); - - assert!( - matches!(rx.try_recv(), Ok(ClientOperation::BrowseResult { .. })), - "the listing on the wire still reaches the client" - ); - assert!(actor.stream.is_some(), "retiring is not yet closing"); - - actor.retire_deadline = - Some(Instant::now().checked_sub(Duration::from_secs(1)).unwrap()); - actor.tick(); - - assert!(actor.stream.is_none(), "past the grace the stream closes"); - assert!(matches!( - rx.try_recv(), - Ok(ClientOperation::PeerDisconnected(7, _, None)) - )); - } - - #[test] - fn a_retired_actor_closed_by_the_peer_reports_a_clean_close() { - let (mut actor, rx, far_end) = connected_actor(); - - actor.handle_message(PeerMessage::Retire); - drop(far_end); - std::thread::sleep(Duration::from_millis(50)); - actor.tick(); - - assert!(actor.stream.is_none()); - assert!( - matches!( - rx.try_recv(), - Ok(ClientOperation::PeerDisconnected(7, _, None)) - ), - "the peer dropping a retired connection is not an error" - ); - } -} +mod tests; diff --git a/soulseek-rs-lib/src/actor/peer_actor/tests.rs b/soulseek-rs-lib/src/actor/peer_actor/tests.rs new file mode 100644 index 0000000..63c10c2 --- /dev/null +++ b/soulseek-rs-lib/src/actor/peer_actor/tests.rs @@ -0,0 +1,147 @@ +use super::*; +use crate::message::peer::build_shared_file_list; +use crate::peer::{ConnectionType, Peer}; +use std::net::{TcpListener, TcpStream}; +use std::sync::mpsc::Receiver; + +/// A connected inbound actor over a real loopback socket, plus the far +/// end (kept alive so reads see silence, not EOF). +fn connected_actor() -> (PeerActor, Receiver, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let stream = TcpStream::connect(addr).unwrap(); + stream.set_nonblocking(true).unwrap(); + let (far_end, _) = listener.accept().unwrap(); + + let (mut actor, rx) = make_actor(Some(stream)); + actor.on_start(); + (actor, rx, far_end) +} + +fn make_actor( + stream: Option, +) -> (PeerActor, Receiver) { + let (tx, rx) = std::sync::mpsc::channel(); + let peer = Peer::new( + "bob".to_string(), + ConnectionType::P, + "127.0.0.1".to_string(), + 0, + None, + 0, + 0, + 0, + ); + let actor = PeerActor::new(peer, stream, None, tx, "me".to_string(), 7); + (actor, rx) +} + +#[test] +fn a_timed_out_connect_parks_the_actor_in_disconnected() { + let (mut actor, rx) = make_actor(None); + actor.connection_state = ConnectionState::Connecting { + since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(), + }; + + actor.tick(); + + assert!( + matches!(actor.connection_state, ConnectionState::Disconnected), + "a timed-out connect must leave Connecting" + ); + match rx.try_recv() { + Ok(ClientOperation::PeerConnectFailed(7, username)) => { + assert_eq!(username, "bob"); + } + other => panic!("expected PeerConnectFailed, got {other:?}"), + } +} + +#[test] +fn an_upload_failed_message_reaches_the_client_as_this_peer() { + let (mut actor, rx, _far_end) = connected_actor(); + + actor.handle_message(PeerMessage::UploadFailed( + String::new(), + "song.mp3".to_string(), + )); + + match rx.try_recv() { + Ok(ClientOperation::UploadFailed(username, filename)) => { + assert_eq!(username, "bob"); + assert_eq!(filename, "song.mp3"); + } + other => panic!("expected UploadFailed, got {other:?}"), + } +} + +#[test] +fn tick_reaps_a_peer_idle_past_the_deadline() { + let (mut actor, rx, _far_end) = connected_actor(); + + actor.tick(); + assert!(actor.stream.is_some(), "a fresh connection stays open"); + + actor.last_activity = Instant::now() + .checked_sub(IDLE_DISCONNECT + Duration::from_secs(1)) + .unwrap(); + actor.tick(); + + assert!(actor.stream.is_none(), "an idle stream must be closed"); + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, username, None)) => { + assert_eq!(username, "bob"); + } + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } +} + +// A connection replaced by a newer one to the same peer may still carry +// a reply already on the wire: it is retired, not cut. +#[test] +fn a_retired_actor_delivers_what_is_on_the_wire_then_closes() { + use std::io::Write; + let (mut actor, rx, mut far_end) = connected_actor(); + + actor.handle_message(PeerMessage::Retire); + far_end + .write_all(&build_shared_file_list(&[]).get_buffer()) + .unwrap(); + std::thread::sleep(Duration::from_millis(50)); + actor.tick(); + + assert!( + matches!(rx.try_recv(), Ok(ClientOperation::BrowseResult { .. })), + "the listing on the wire still reaches the client" + ); + assert!(actor.stream.is_some(), "retiring is not yet closing"); + + actor.retire_deadline = + Some(Instant::now().checked_sub(Duration::from_secs(1)).unwrap()); + actor.tick(); + + assert!(actor.stream.is_none(), "past the grace the stream closes"); + assert!(matches!( + rx.try_recv(), + Ok(ClientOperation::PeerDisconnected(7, _, None)) + )); +} + +#[test] +fn a_retired_actor_closed_by_the_peer_reports_a_clean_close() { + let (mut actor, rx, far_end) = connected_actor(); + + actor.handle_message(PeerMessage::Retire); + drop(far_end); + std::thread::sleep(Duration::from_millis(50)); + actor.tick(); + + assert!(actor.stream.is_none()); + assert!( + matches!( + rx.try_recv(), + Ok(ClientOperation::PeerDisconnected(7, _, None)) + ), + "the peer dropping a retired connection is not an error" + ); +} From 2ab7902d1fcb3664a8e3814b224102c617a49d5c Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 01:00:21 +0200 Subject: [PATCH 4/5] fix(lib): retire a dialling actor at once and drain the socket while retired A Retire that reached an actor still dialling was queued behind the connection and, if that dial then failed, never applied: nothing in the registry pointed at the actor any more, so its thread lived on. Nothing has gone out on a dial in progress, so retiring one closes it on the spot. A retired actor reads only from its tick, and one socket buffer a tick is 10 KiB a second: a real share listing would outlast the grace. Every actor now drains up to 64 buffers a tick. --- soulseek-rs-lib/src/actor/peer_actor.rs | 29 ++++- soulseek-rs-lib/src/actor/peer_actor/tests.rs | 115 ++++++++++++++---- 2 files changed, 117 insertions(+), 27 deletions(-) diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 241fbec..147bc1f 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -121,6 +121,8 @@ const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024; const IDLE_DISCONNECT: Duration = Duration::from_mins(5); /// How long a replaced connection keeps listening for replies in flight. const RETIRE_GRACE: Duration = Duration::from_secs(30); +/// Socket buffers read per tick: 64 KiB, against a 100 ms tick. +const READS_PER_TICK: usize = 64; impl PeerActor { #[must_use] @@ -228,6 +230,13 @@ impl PeerActor { if matches!(self.connection_state, ConnectionState::Connecting { .. }) { match &msg { PeerMessage::SetUsername(_) | PeerMessage::ProcessRead => {} + // Nothing has gone out on a dial still in progress, so there + // is nothing to wait for. + PeerMessage::Retire => { + self.retire_deadline = Some(Instant::now()); + self.disconnect(); + return; + } _ => { self.queued_messages.push(msg); return; @@ -497,7 +506,7 @@ impl PeerActor { return; }; - match self.reader.read_from_socket(stream) { + match Self::drain(&mut self.reader, stream) { Ok(()) => { self.last_activity = Instant::now(); } @@ -545,6 +554,24 @@ impl PeerActor { self.extract_and_process_messages(); } + /// Read what the socket holds, up to `READS_PER_TICK` buffers, so a big + /// listing lands in a few ticks rather than a few minutes; the cap keeps + /// a peer that never stops sending from starving the mailbox. + fn drain( + reader: &mut MessageReader, + stream: &mut TcpStream, + ) -> io::Result<()> { + reader.read_from_socket(stream)?; + for _ in 1..READS_PER_TICK { + match reader.read_from_socket(stream) { + Ok(()) => {} + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break, + Err(e) => return Err(e), + } + } + Ok(()) + } + fn extract_and_process_messages(&mut self) { let username = self.peer_username(); let mut extracted_count = 0; diff --git a/soulseek-rs-lib/src/actor/peer_actor/tests.rs b/soulseek-rs-lib/src/actor/peer_actor/tests.rs index 63c10c2..ff0c9f2 100644 --- a/soulseek-rs-lib/src/actor/peer_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/peer_actor/tests.rs @@ -96,52 +96,115 @@ fn tick_reaps_a_peer_idle_past_the_deadline() { } } +/// A listing well past one socket buffer, so delivery within a couple of +/// ticks proves the actor drains the socket rather than sipping 1 KiB a tick. +fn big_listing() -> Vec { + let files = (0..400u64) + .map(|i| { + ( + format!("track-{i:03}-{}.flac", "x".repeat(i as usize % 17)), + i, + ) + }) + .collect(); + let dir = crate::message::peer::SharedDirectory { + name: "album".to_string(), + files, + }; + let bytes = build_shared_file_list(std::slice::from_ref(&dir)).get_buffer(); + assert!(bytes.len() > 4096, "the listing must span several reads"); + bytes +} + +fn self_handle( + actor: &mut PeerActor, +) -> Receiver> { + let (tx, rx) = std::sync::mpsc::channel(); + actor.set_self_handle(ActorHandle { sender: tx }); + rx +} + // A connection replaced by a newer one to the same peer may still carry -// a reply already on the wire: it is retired, not cut. +// a reply already on the wire: it is retired, not cut, and once nothing +// more can come it is the one to stop itself. #[test] -fn a_retired_actor_delivers_what_is_on_the_wire_then_closes() { +fn a_retired_actor_delivers_what_is_on_the_wire_then_stops_itself() { use std::io::Write; let (mut actor, rx, mut far_end) = connected_actor(); + let stop_rx = self_handle(&mut actor); actor.handle_message(PeerMessage::Retire); - far_end - .write_all(&build_shared_file_list(&[]).get_buffer()) - .unwrap(); + far_end.write_all(&big_listing()).unwrap(); std::thread::sleep(Duration::from_millis(50)); actor.tick(); + actor.tick(); - assert!( - matches!(rx.try_recv(), Ok(ClientOperation::BrowseResult { .. })), - "the listing on the wire still reaches the client" - ); + match rx.try_recv() { + Ok(ClientOperation::BrowseResult { username, .. }) => { + assert_eq!(username, "bob"); + } + other => { + panic!("expected the listing to reach the client, got {other:?}") + } + } assert!(actor.stream.is_some(), "retiring is not yet closing"); + assert!(stop_rx.try_recv().is_err(), "not stopped while listening"); actor.retire_deadline = Some(Instant::now().checked_sub(Duration::from_secs(1)).unwrap()); actor.tick(); - assert!(actor.stream.is_none(), "past the grace the stream closes"); - assert!(matches!( - rx.try_recv(), - Ok(ClientOperation::PeerDisconnected(7, _, None)) - )); + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } + + actor.tick(); + assert!( + matches!(stop_rx.try_recv(), Ok(crate::actor::ActorMessage::Stop)), + "a closed retired actor stops itself" + ); } #[test] -fn a_retired_actor_closed_by_the_peer_reports_a_clean_close() { - let (mut actor, rx, far_end) = connected_actor(); +fn a_retired_actor_reporting_an_error_reports_a_clean_close() { + let (mut actor, rx, _far_end) = connected_actor(); actor.handle_message(PeerMessage::Retire); - drop(far_end); - std::thread::sleep(Duration::from_millis(50)); - actor.tick(); + actor.disconnect_with_error(Error::from(io::ErrorKind::ConnectionReset)); - assert!(actor.stream.is_none()); - assert!( - matches!( - rx.try_recv(), - Ok(ClientOperation::PeerDisconnected(7, _, None)) + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!( + "a retired connection going away is not an error, got {other:?}" ), - "the peer dropping a retired connection is not an error" - ); + } +} + +// Retiring an actor whose dial has not completed: nothing went out, so +// there is nothing to wait for, and it must not linger in the queue that +// only a successful connection drains. +#[test] +fn retiring_a_connecting_actor_closes_it_at_once() { + let (mut actor, rx) = make_actor(None); + let stop_rx = self_handle(&mut actor); + actor.connection_state = ConnectionState::Connecting { + since: Instant::now(), + }; + + actor.handle_message(PeerMessage::Retire); + + assert!(matches!( + actor.connection_state, + ConnectionState::Disconnected + )); + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } + actor.tick(); + assert!(matches!( + stop_rx.try_recv(), + Ok(crate::actor::ActorMessage::Stop) + )); } From bcf71dabec15ebace3706550f9668bb31dff4986 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 01:23:19 +0200 Subject: [PATCH 5/5] fix(lib): parse what a peer sent before acting on its close Draining several buffers a tick meant a reply and the close after it could land in one call, and the close was acted on first: the reply sat unparsed in the reader while the actor disconnected. Only the first read of a drain reports an error now; the close is seen again on the next tick, after the reply has been handled. --- soulseek-rs-lib/src/actor/peer_actor.rs | 11 ++++---- soulseek-rs-lib/src/actor/peer_actor/tests.rs | 26 +++++++++++++++++++ 2 files changed, 32 insertions(+), 5 deletions(-) diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 147bc1f..bf63e3a 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -556,17 +556,18 @@ impl PeerActor { /// Read what the socket holds, up to `READS_PER_TICK` buffers, so a big /// listing lands in a few ticks rather than a few minutes; the cap keeps - /// a peer that never stops sending from starving the mailbox. + /// a peer that never stops sending from starving the mailbox. Only the + /// first read reports an error: a peer that sends its reply and hangs up + /// has still sent the reply, which must be parsed before the close is + /// acted on, and the socket says so again on the next call. fn drain( reader: &mut MessageReader, stream: &mut TcpStream, ) -> io::Result<()> { reader.read_from_socket(stream)?; for _ in 1..READS_PER_TICK { - match reader.read_from_socket(stream) { - Ok(()) => {} - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break, - Err(e) => return Err(e), + if reader.read_from_socket(stream).is_err() { + break; } } Ok(()) diff --git a/soulseek-rs-lib/src/actor/peer_actor/tests.rs b/soulseek-rs-lib/src/actor/peer_actor/tests.rs index ff0c9f2..be62ec0 100644 --- a/soulseek-rs-lib/src/actor/peer_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/peer_actor/tests.rs @@ -166,6 +166,32 @@ fn a_retired_actor_delivers_what_is_on_the_wire_then_stops_itself() { ); } +// The peer that replaced the connection considers the old one done too: it +// writes its reply and hangs up in the same breath. The reply still counts. +#[test] +fn a_reply_sent_right_before_the_peer_hangs_up_is_still_delivered() { + use std::io::Write; + let (mut actor, rx, mut far_end) = connected_actor(); + + actor.handle_message(PeerMessage::Retire); + far_end.write_all(&big_listing()).unwrap(); + drop(far_end); + std::thread::sleep(Duration::from_millis(50)); + actor.tick(); + actor.tick(); + + match rx.try_recv() { + Ok(ClientOperation::BrowseResult { username, .. }) => { + assert_eq!(username, "bob"); + } + other => panic!("expected the listing before the close, got {other:?}"), + } + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } +} + #[test] fn a_retired_actor_reporting_an_error_reports_a_clean_close() { let (mut actor, rx, _far_end) = connected_actor();