diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index e3ca79a..3cf03ef 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -61,6 +61,9 @@ pub struct PeerActor { /// replaced actor is later stopped — which would otherwise alias and evict /// a newer actor registered under the same username. disconnect_reported: bool, + /// Registry-assigned unique id, echoed in terminal notifications so the + /// client evicts only this actor and never a newer namesake. + id: u64, } impl PeerActor { @@ -71,6 +74,7 @@ impl PeerActor { reader: Option, client_channel: Sender, own_username: String, + id: u64, ) -> Self { let outbound = stream.is_none(); let connection_state = if stream.is_some() { @@ -93,6 +97,7 @@ impl PeerActor { outbound, established: false, disconnect_reported: false, + id, } } @@ -449,9 +454,13 @@ impl PeerActor { // client can fall back to server-brokered connect. Anything else is a // normal disconnect. let op = if self.outbound && !self.established { - ClientOperation::PeerConnectFailed(username) + ClientOperation::PeerConnectFailed(self.id, username) } else { - ClientOperation::PeerDisconnected(username, Some(error.into())) + ClientOperation::PeerDisconnected( + self.id, + username, + Some(error.into()), + ) }; if let Err(e) = self.client_channel.send(op) { @@ -471,7 +480,7 @@ impl PeerActor { if let Err(e) = self .client_channel - .send(ClientOperation::PeerDisconnected(username, None)) + .send(ClientOperation::PeerDisconnected(self.id, username, None)) { error!("Failed to send disconnect notification: {}", e); } diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index e9cf70f..7c7338d 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -8,11 +8,20 @@ use crate::{debug, error}; use std::collections::HashMap; use std::net::TcpStream; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc::Sender; use std::sync::{Arc, Mutex}; +/// Source of unique per-actor ids so terminal-outcome eviction can be made +/// identity-aware (a replaced actor must not evict its replacement). +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)>; + pub struct PeerRegistry { - peers: Arc>>>, + peers: Arc>, actor_system: Arc, client_channel: Sender, own_username: String, @@ -40,6 +49,7 @@ impl PeerRegistry { reader: Option, ) -> Result, String> { let username = peer.username.clone(); + let id = NEXT_PEER_ID.fetch_add(1, Ordering::Relaxed); let actor = PeerActor::new( peer, @@ -47,6 +57,7 @@ impl PeerRegistry { reader, self.client_channel.clone(), self.own_username.clone(), + id, ); let handle = @@ -59,9 +70,11 @@ impl PeerRegistry { .lock_safe() .map_err(|e| format!("peer registry lock poisoned: {e}"))?; // Stop any actor already registered under this username so it does not - // become an orphan pinning a pool worker forever (and so its later - // shutdown cannot remove this new connection from the registry). - if let Some(old_handle) = peers.insert(username.clone(), handle.clone()) + // 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())) { let _ = old_handle.stop(); debug!( @@ -76,7 +89,7 @@ impl PeerRegistry { #[must_use] pub fn get_peer(&self, username: &str) -> Option> { match self.peers.lock_safe() { - Ok(peers) => peers.get(username).cloned(), + Ok(peers) => peers.get(username).map(|(_, handle)| handle.clone()), Err(e) => { error!("[peer_registry] get_peer: {}", e); None @@ -96,13 +109,40 @@ impl PeerRegistry { return None; } }; - let handle = peers.remove(username); + let removed = peers.remove(username); - if handle.is_some() { + if removed.is_some() { debug!("[peer_registry] Removed peer actor for {}", username); } - handle + removed.map(|(_, handle)| handle) + } + + /// Remove and return the actor for `username` only if it is still the actor + /// with `id`. A stale (replaced) actor's terminal notification therefore + /// cannot evict the newer actor that now occupies the slot. + #[must_use] + pub fn remove_peer_if( + &self, + username: &str, + id: u64, + ) -> Option> { + let mut peers = match self.peers.lock_safe() { + Ok(p) => p, + Err(e) => { + error!("[peer_registry] remove_peer_if: {}", e); + return None; + } + }; + 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 + ); + return removed; + } + None } #[must_use] @@ -169,3 +209,52 @@ impl Clone for PeerRegistry { } } } + +#[cfg(test)] +mod tests { + use super::PeerRegistry; + use crate::actor::ActorSystem; + use crate::peer::{ConnectionType, Peer}; + use crate::utils::thread_pool::ThreadPool; + use std::net::{TcpListener, TcpStream}; + use std::sync::Arc; + + #[test] + fn remove_peer_if_respects_actor_identity() { + let pool = Arc::new(ThreadPool::new(2)); + let system = Arc::new(ActorSystem::new(pool)); + let (tx, _rx) = std::sync::mpsc::channel(); + let registry = PeerRegistry::new(system, tx, "me".to_string()); + + // A real loopback connection makes the actor inbound (no dial-out); + // non-blocking so it can process Stop promptly on teardown. + 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 _server_side = listener.accept().unwrap().0; + + let peer = Peer::new( + "bob".to_string(), + ConnectionType::P, + "127.0.0.1".to_string(), + u32::from(addr.port()), + None, + 0, + 0, + 0, + ); + registry.register_peer(peer, Some(stream), None).unwrap(); + assert!(registry.contains("bob")); + + // A stale / wrong id must not evict the live actor. + assert!(registry.remove_peer_if("bob", u64::MAX).is_none()); + assert!(registry.contains("bob")); + + // Unconditional removal still works (and stops the actor). + let handle = registry.remove_peer("bob"); + assert!(handle.is_some()); + let _ = handle.unwrap().stop(); + assert!(!registry.contains("bob")); + } +} diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index 620d4f0..5fd48cf 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -89,7 +89,7 @@ pub enum ClientOperation { NewPeer(NewPeer), ConnectToPeer(Peer), SearchResult(SearchResult), - PeerDisconnected(String, Option), + PeerDisconnected(u64, String, Option), PierceFireWall(Peer), DownloadFromPeer(u32, Peer, bool), UpdateDownloadTokens(Transfer, String), @@ -111,8 +111,8 @@ pub enum ClientOperation { PeerConnected(String), /// A direct outbound connection to this peer failed before it was /// established — the peer is likely firewalled, so fall back to asking the - /// server to broker the connection. - PeerConnectFailed(String), + /// server to broker the connection. Carries the reporting actor's id. + PeerConnectFailed(u64, String), } pub struct ClientContext { pub peer_registry: Option, @@ -934,6 +934,7 @@ impl Client { } } ClientOperation::PeerDisconnected( + id, username, error, ) => { @@ -941,7 +942,9 @@ impl Client { // below acquires a write lock on the same // RwLock, which would self-deadlock the entire // client ops loop if this read guard were still - // held on this thread. + // held on this thread. Evict only if this exact + // actor still occupies the slot, so a replaced + // actor's shutdown can't remove its successor. { let context = match client_context .read_safe() @@ -957,8 +960,8 @@ impl Client { }; if let Some(ref registry) = context.peer_registry - && let Some(handle) = - registry.remove_peer(&username) + && let Some(handle) = registry + .remove_peer_if(&username, id) { let _ = handle.stop(); } @@ -1373,7 +1376,10 @@ impl Client { } } } - ClientOperation::PeerConnectFailed(username) => { + ClientOperation::PeerConnectFailed( + id, + username, + ) => { // Direct connect failed: ask the server to // broker it. Register a correlation token, then // send ConnectToPeer so the (firewalled) peer @@ -1388,11 +1394,14 @@ impl Client { // longer shadows the brokered reconnect // (a stale registry entry would make // later downloads queue into a dead, - // streamless actor and hang). - if let Some(handle) = - ctx.peer_registry.as_ref().and_then( - |r| r.remove_peer(&username), - ) + // streamless actor and hang). Identity- + // aware so a newer namesake is untouched. + if let Some(handle) = ctx + .peer_registry + .as_ref() + .and_then(|r| { + r.remove_peer_if(&username, id) + }) { let _ = handle.stop(); } @@ -1422,10 +1431,10 @@ impl Client { .send(ServerMessage::SendMessage(msg)); // Bound the brokered attempt: if no PierceFirewall - // arrives, fail the peer's queued downloads (so - // the caller's Receiver unblocks) and reclaim the - // token. A successful pierce consumes the token - // first, making this a no-op. + // consumes the token, fail the peer's queued + // downloads (so the caller's Receiver unblocks) + // and reclaim the token. A successful pierce + // takes the token first, making this a no-op. let timeout_ctx = client_context.clone(); let timeout_user = username.clone(); thread::spawn(move || { @@ -1437,21 +1446,6 @@ impl Client { .is_some() }); if still_pending { - if let Some(handle) = timeout_ctx - .read_safe() - .ok() - .and_then(|c| { - c.peer_registry - .as_ref() - .and_then(|r| { - r.remove_peer( - &timeout_user, - ) - }) - }) - { - let _ = handle.stop(); - } Self::fail_queued_downloads( &timeout_ctx, &timeout_user,