From ea168605b53a7bcbcd7bebe02cd39bec595f9742 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 00:35:11 +0200 Subject: [PATCH] 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