Merge pull request #86 from michel/ci/stress-serve-browses-while-offering

fix(lib): keep a replaced peer connection open for the reply in flight

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,

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<Instant>,
 }
 
 /// Cap on queued outbound bytes. A peer that stops reading must not be able to
@@ -112,6 +119,10 @@ 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);
+/// Socket buffers read per tick: 64 KiB, against a 100 ms tick.
+const READS_PER_TICK: usize = 64;
 
 impl PeerActor {
     #[must_use]
@@ -148,6 +159,7 @@ impl PeerActor {
             serving_tokens: std::collections::HashSet::new(),
             pending_write: Vec::new(),
             last_activity: Instant::now(),
+            retire_deadline: None,
         }
     }
 
@@ -218,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;
@@ -229,6 +248,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);
             }
@@ -484,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();
                 }
@@ -532,6 +554,25 @@ 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. 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 {
+            if reader.read_from_socket(stream).is_err() {
+                break;
+            }
+        }
+        Ok(())
+    }
+
     fn extract_and_process_messages(&mut self) {
         let username = self.peer_username();
         let mut extracted_count = 0;
@@ -664,6 +705,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,109 +966,27 @@ 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 => {}
         }
     }
 }
 
 #[cfg(test)]
-mod tests {
-    use super::*;
-    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<ClientOperation>, 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<TcpStream>,
-    ) -> (PeerActor, Receiver<ClientOperation>) {
-        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:?}"),
-        }
-    }
-}
+mod tests;

soulseek-rs-lib/src/actor/peer_actor/tests.rs

@@ -0,0 +1,236 @@
+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<ClientOperation>, 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<TcpStream>,
+) -> (PeerActor, Receiver<ClientOperation>) {
+    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 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<u8> {
+    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<crate::actor::ActorMessage<PeerMessage>> {
+    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, 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_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(&big_listing()).unwrap();
+    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 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");
+    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"
+    );
+}
+
+// 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();
+
+    actor.handle_message(PeerMessage::Retire);
+    actor.disconnect_with_error(Error::from(io::ErrorKind::ConnectionReset));
+
+    match rx.try_recv() {
+        Ok(ClientOperation::PeerDisconnected(7, _, None)) => {}
+        other => panic!(
+            "a retired connection going away is not an error, got {other:?}"
+        ),
+    }
+}
+
+// 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)
+    ));
+}

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