From 2a90619a64388369a6a9c7117e0036d15a03baef Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Thu, 13 Aug 2026 15:28:36 +0200 Subject: [PATCH] refactor(client): dedupe peer registration and order eviction by id The registry stored a registration Instant only to pick the eviction victim, but the actor id is already monotonic per registration and refreshed at the same moments, so it orders identically. The register-refuse-fail sequence was also copy-pasted between the ops loop and the listener; both now call one register_peer_or_fail. --- soulseek-rs-lib/src/actor/peer_registry.rs | 26 +++---- soulseek-rs-lib/src/client/connection.rs | 79 +++++++++++++--------- soulseek-rs-lib/src/peer/listen.rs | 42 ++---------- 3 files changed, 62 insertions(+), 85 deletions(-) diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index 3819830..0ff9230 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -11,7 +11,6 @@ use std::net::TcpStream; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc::Sender; use std::sync::{Arc, Mutex}; -use std::time::Instant; /// Source of unique per-actor ids so terminal-outcome eviction can be made /// identity-aware (a replaced actor must not evict its replacement). @@ -19,7 +18,7 @@ static NEXT_PEER_ID: AtomicU64 = AtomicU64::new(1); /// Registered peers keyed by username, each stored with the unique id of the /// actor currently occupying the slot. -type PeerMap = HashMap, Instant)>; +type PeerMap = HashMap)>; pub struct PeerRegistry { peers: Arc>, @@ -104,7 +103,7 @@ impl PeerRegistry { let victim = peers .iter() .filter(|(name, _)| !protected.contains(name.as_str())) - .min_by_key(|(_, (_, _, registered))| *registered) + .min_by_key(|(_, (id, _))| *id) .map(|(name, _)| name.clone()); let Some(victim) = victim else { return Err(format!( @@ -112,7 +111,7 @@ impl PeerRegistry { is busy; refusing {username}" )); }; - if let Some((_, handle, _)) = peers.remove(&victim) { + if let Some((_, handle)) = peers.remove(&victim) { let _ = handle.stop(); debug!( "[peer_registry] evicted idle peer {} to admit {}", @@ -131,8 +130,8 @@ impl PeerRegistry { // 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. - if let Some((_, old_handle, _)) = - peers.insert(username.clone(), (id, handle.clone(), Instant::now())) + if let Some((_, old_handle)) = + peers.insert(username.clone(), (id, handle.clone())) { let _ = old_handle.stop(); debug!( @@ -147,9 +146,7 @@ impl PeerRegistry { #[must_use] pub fn get_peer(&self, username: &str) -> Option> { match self.peers.lock_safe() { - Ok(peers) => { - peers.get(username).map(|(_, handle, _)| handle.clone()) - } + Ok(peers) => peers.get(username).map(|(_, handle)| handle.clone()), Err(e) => { error!("[peer_registry] get_peer: {}", e); None @@ -175,7 +172,7 @@ impl PeerRegistry { debug!("[peer_registry] Removed peer actor for {}", username); } - removed.map(|(_, handle, _)| handle) + removed.map(|(_, handle)| handle) } /// Remove and return the actor for `username` only if it is still the actor @@ -194,11 +191,8 @@ impl PeerRegistry { return None; } }; - if peers - .get(username) - .is_some_and(|(stored, _, _)| *stored == id) - { - let removed = peers.remove(username).map(|(_, handle, _)| handle); + if peers.get(username).is_some_and(|(stored, _)| *stored == id) { + let removed = peers.remove(username).map(|(_, handle)| handle); debug!( "[peer_registry] Removed peer actor {} for {}", id, username @@ -261,7 +255,6 @@ mod tests { use std::net::{TcpListener, TcpStream}; use std::sync::Arc; use std::sync::atomic::AtomicUsize; - use std::time::Duration; fn loopback_peer(name: &str) -> (Peer, TcpStream, TcpStream) { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -306,7 +299,6 @@ mod tests { for name in ["a", "b", "c"] { let (peer, stream, _far_end) = loopback_peer(name); registry.register_peer(peer, Some(stream), None).unwrap(); - std::thread::sleep(Duration::from_millis(5)); } assert!(!registry.contains("a"), "the oldest peer must be evicted"); diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index c9219f9..8421ce9 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -199,6 +199,46 @@ impl Client { } } + pub(crate) fn register_peer_or_fail( + client_context: &Arc>, + peer: Peer, + stream: Option, + reader: Option, + ) -> bool { + let username = peer.username.clone(); + let registered = { + let context = match client_context.read_safe() { + Ok(c) => c, + Err(e) => { + error!("[client] register peer read: {}", e); + return false; + } + }; + if let Some(ref registry) = context.peer_registry { + let protected = context.protected_peers(); + match registry + .register_peer_protected(peer, stream, reader, &protected) + { + Ok(_) => true, + Err(e) => { + warn!( + "[client] refusing peer connection for {:?}: {:?}", + username, e + ); + false + } + } + } else { + trace!("PeerRegistry not initialized"); + false + } + }; + if !registered { + Self::fail_queued_downloads(client_context, &username); + } + registered + } + pub(crate) fn connect_to_peer( peer: Peer, client_context: Arc>, @@ -212,39 +252,12 @@ impl Client { ); match peer.connection_type { ConnectionType::P => { - let username = peer.username; - - 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 refused { - Self::fail_queued_downloads(&client_context, &username); - } + Self::register_peer_or_fail( + &client_context, + peer_clone, + stream, + None, + ); } ConnectionType::F => { diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 070df8f..55e8d25 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -12,7 +12,7 @@ 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, warn}; +use crate::{DownloadStatus, debug, error, info, trace}; /// How long to wait before accepting again after a failure. /// @@ -151,40 +151,12 @@ fn handle_peer_connection( } 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 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 !registered { - Client::fail_queued_downloads(&context.client_context, &peer.username); - } - registered + Client::register_peer_or_fail( + &context.client_context, + peer, + Some(stream), + Some(reader), + ) } fn handle_file_connection(