fix: identity-aware peer eviction so a replaced actor can't evict its successor

When a healthy peer actor was replaced under the same username (both sides
connecting at once, or two overlapping downloads to the same not-yet-connected
peer), its graceful shutdown still emitted PeerDisconnected, and the client
evicted by username — removing and stopping the NEWER actor now occupying the
slot, tearing down the live connection and stalling that peer's downloads.

Give each peer actor a unique id (assigned by the registry) and echo it in
PeerDisconnected/PeerConnectFailed. The client now evicts via
remove_peer_if(username, id), which removes the entry only when the reporting
actor still owns the slot, so a stale actor's shutdown is a no-op. The brokered
timeout no longer removes an actor by username either, avoiding the same
aliasing. Add a registry unit test for the identity guard.

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<MessageReader>,
         client_channel: Sender<ClientOperation>,
         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);
         }

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<String, (u64, ActorHandle<PeerMessage>)>;
+
 pub struct PeerRegistry {
-    peers: Arc<Mutex<HashMap<String, ActorHandle<PeerMessage>>>>,
+    peers: Arc<Mutex<PeerMap>>,
     actor_system: Arc<ActorSystem>,
     client_channel: Sender<ClientOperation>,
     own_username: String,
@@ -40,6 +49,7 @@ impl PeerRegistry {
         reader: Option<MessageReader>,
     ) -> Result<ActorHandle<PeerMessage>, 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
+        // become an orphan pinning a pool worker forever. Eviction on the
-        // shutdown cannot remove this new connection from the registry).
+        // replaced actor's later shutdown is identity-aware (keyed on its id),
-        if let Some(old_handle) = peers.insert(username.clone(), handle.clone())
+        // 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<ActorHandle<PeerMessage>> {
         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<ActorHandle<PeerMessage>> {
+        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"));
+    }
+}

soulseek-rs-lib/src/client.rs

@@ -89,7 +89,7 @@ pub enum ClientOperation {
     NewPeer(NewPeer),
     ConnectToPeer(Peer),
     SearchResult(SearchResult),
-    PeerDisconnected(String, Option<SoulseekRs>),
+    PeerDisconnected(u64, String, Option<SoulseekRs>),
     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.
+    /// server to broker the connection. Carries the reporting actor's id.
-    PeerConnectFailed(String),
+    PeerConnectFailed(u64, String),
 }
 pub struct ClientContext {
     pub peer_registry: Option<PeerRegistry>,
@@ -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) =
+                                        && let Some(handle) = registry
-                                            registry.remove_peer(&username)
+                                            .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).
+                                        // streamless actor and hang). Identity-
-                                        if let Some(handle) =
+                                        // aware so a newer namesake is untouched.
-                                            ctx.peer_registry.as_ref().and_then(
+                                        if let Some(handle) = ctx
-                                                |r| r.remove_peer(&username),
+                                            .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
+                                // consumes the token, fail the peer's queued
-                                // the caller's Receiver unblocks) and reclaim the
+                                // downloads (so the caller's Receiver unblocks)
-                                // token. A successful pierce consumes the token
+                                // and reclaim the token. A successful pierce
-                                // first, making this a no-op.
+                                // 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,