fix(lib): retire a dialling actor at once and drain the socket while retired

A Retire that reached an actor still dialling was queued behind the
connection and, if that dial then failed, never applied: nothing in the
registry pointed at the actor any more, so its thread lived on. Nothing
has gone out on a dial in progress, so retiring one closes it on the
spot.

A retired actor reads only from its tick, and one socket buffer a tick
is 10 KiB a second: a real share listing would outlast the grace. Every
actor now drains up to 64 buffers a tick.

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

@@ -121,6 +121,8 @@ const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024;
 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]
@@ -228,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;
@@ -497,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();
                 }
@@ -545,6 +554,24 @@ 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.
+    fn drain(
+        reader: &mut MessageReader,
+        stream: &mut TcpStream,
+    ) -> io::Result<()> {
+        reader.read_from_socket(stream)?;
+        for _ in 1..READS_PER_TICK {
+            match reader.read_from_socket(stream) {
+                Ok(()) => {}
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
+                Err(e) => return Err(e),
+            }
+        }
+        Ok(())
+    }
+
     fn extract_and_process_messages(&mut self) {
         let username = self.peer_username();
         let mut extracted_count = 0;

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

@@ -96,52 +96,115 @@ fn tick_reaps_a_peer_idle_past_the_deadline() {
     }
 }
 
+/// 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.
+// 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_closes() {
+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(&build_shared_file_list(&[]).get_buffer())
-        .unwrap();
+    far_end.write_all(&big_listing()).unwrap();
     std::thread::sleep(Duration::from_millis(50));
     actor.tick();
+    actor.tick();
 
-    assert!(
-        matches!(rx.try_recv(), Ok(ClientOperation::BrowseResult { .. })),
-        "the listing on the wire still reaches the client"
-    );
+    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");
-    assert!(matches!(
-        rx.try_recv(),
-        Ok(ClientOperation::PeerDisconnected(7, _, None))
-    ));
+    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"
+    );
 }
 
 #[test]
-fn a_retired_actor_closed_by_the_peer_reports_a_clean_close() {
-    let (mut actor, rx, far_end) = connected_actor();
+fn a_retired_actor_reporting_an_error_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();
+    actor.disconnect_with_error(Error::from(io::ErrorKind::ConnectionReset));
 
-    assert!(actor.stream.is_none());
-    assert!(
-        matches!(
-            rx.try_recv(),
-            Ok(ClientOperation::PeerDisconnected(7, _, None))
+    match rx.try_recv() {
+        Ok(ClientOperation::PeerDisconnected(7, _, None)) => {}
+        other => panic!(
+            "a retired connection going away is not an error, got {other:?}"
         ),
-        "the peer dropping a retired connection is not an error"
-    );
+    }
+}
+
+// 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)
+    ));
 }