fix: survive busy sessions without exhausting file descriptors

The peer listener died with "Too many open files (os error 24)" on any
long-lived non-daemon session: only the daemon raised RLIMIT_NOFILE, and
peer sockets were unbounded — no idle reaping, no deadlines on accepted
transfer streams, and one peer could hold two connections under split
registry identities.

- raise the open-file limit at the top of run(), for every entry point,
  and warn from Client::connect when the soft limit is below 1024
- reap peer control connections idle for five minutes; a clean
  disconnect no longer discards the peer's queued uploads
- give listener-supplied download streams and upload serving the same
  read/write deadlines dialled sockets already had
- register inbound peers under their plain username so the registry
  holds one connection per peer, and announce them like pierced ones
- hold the registry lock across actor spawn + insert and surface
  thread-spawn failure, so an instantly-dead peer cannot leave a
  zombie entry

Cargo.lock

@@ -2569,6 +2569,9 @@ dependencies = [
 [[package]]
 name = "soulseek-rs-lib"
 version = "13.0.0"
+dependencies = [
+ "libc",
+]
 
 [[package]]
 name = "stable_deref_trait"

soulseek-rs-lib/Cargo.toml

@@ -18,3 +18,6 @@ path = "src/lib.rs"
 workspace = true
 
 [dependencies]
+
+[target.'cfg(unix)'.dependencies]
+libc = "0.2.189"

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

@@ -87,9 +87,34 @@ impl ActorSystem {
     /// The callback receives the actor handle before `on_start` is called
     pub fn spawn_with_handle<A: Actor, F>(
         &self,
-        mut actor: A,
+        actor: A,
         init: F,
     ) -> ActorHandle<A::Message>
+    where
+        F: FnOnce(&mut A, ActorHandle<A::Message>) + Send + 'static,
+    {
+        match self.try_spawn_with_handle(actor, init) {
+            Ok(handle) => handle,
+            Err(e) => {
+                error!("[actor_system] failed to spawn actor thread: {}", e);
+                // A handle nobody serves: sends fail, which callers of the
+                // infallible spawn already tolerate.
+                let (sender, _) = channel();
+                ActorHandle { sender }
+            }
+        }
+    }
+
+    /// Like [`Self::spawn_with_handle`], but hands a thread-spawn failure back
+    /// to the caller instead of returning a handle no actor will ever serve.
+    /// Storing such a dead handle is how a registry entry becomes a zombie
+    /// that claims its username for the rest of the process.
+    #[allow(clippy::unused_self)] // method for symmetry with the spawn family
+    pub(crate) fn try_spawn_with_handle<A: Actor, F>(
+        &self,
+        mut actor: A,
+        init: F,
+    ) -> std::io::Result<ActorHandle<A::Message>>
     where
         F: FnOnce(&mut A, ActorHandle<A::Message>) + Send + 'static,
     {
@@ -97,7 +122,7 @@ impl ActorSystem {
         let handle = ActorHandle { sender };
         let handle_for_init = handle.clone();
 
-        let spawned = thread::Builder::new()
+        thread::Builder::new()
             .name("soulseek-actor".to_string())
             .stack_size(ACTOR_STACK_SIZE)
             .spawn(move || {
@@ -105,13 +130,9 @@ impl ActorSystem {
                 actor.on_start();
                 Self::run_actor_loop(&mut actor, receiver);
                 actor.on_stop();
-            });
+            })?;
 
-        if let Err(e) = spawned {
+        Ok(handle)
-            error!("[actor_system] failed to spawn actor thread: {}", e);
-        }
-
-        handle
     }
 
     fn run_actor_loop<A: Actor>(

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

@@ -88,12 +88,23 @@ pub struct PeerActor {
     /// larger than the send buffer — a full share listing is easily megabytes —
     /// only partly fits; the rest drains on later ticks.
     pending_write: Vec<u8>,
+    /// When bytes last moved on this connection, in either direction. A peer
+    /// that goes quiet is eventually reaped: nothing else ever closes an idle
+    /// control connection, and each one holds a descriptor and a thread for
+    /// the life of the process.
+    last_activity: Instant,
 }
 
 /// Cap on queued outbound bytes. A peer that stops reading must not be able to
 /// grow this without bound; dropping that connection is the right answer.
 const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024;
 
+/// How long a connected peer may stay silent before its control connection is
+/// closed. Transfers run on their own sockets, and a peer that wants us again
+/// 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);
+
 impl PeerActor {
     #[must_use]
     pub fn new(
@@ -128,6 +139,7 @@ impl PeerActor {
             id,
             serving_tokens: std::collections::HashSet::new(),
             pending_write: Vec::new(),
+            last_activity: Instant::now(),
         }
     }
 
@@ -460,7 +472,9 @@ impl PeerActor {
             };
 
             match self.reader.read_from_socket(stream) {
-                Ok(()) => {}
+                Ok(()) => {
+                    self.last_activity = Instant::now();
+                }
                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
                 Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
                     if let Ok(peer_lock) = self.peer.read_safe() {
@@ -609,6 +623,9 @@ impl PeerActor {
             }
         };
         self.pending_write.drain(..written);
+        if written > 0 {
+            self.last_activity = Instant::now();
+        }
 
         if let Err(e) = outcome {
             error!(
@@ -876,9 +893,75 @@ impl Actor for PeerActor {
                     self.flush_pending_write();
                     self.process_read();
                     self.flush_pending_write();
+
+                    if self.stream.is_some()
+                        && self.last_activity.elapsed() > IDLE_DISCONNECT
+                    {
+                        debug!(
+                            "[peer:{}] idle for {}s, closing",
+                            self.peer_username(),
+                            IDLE_DISCONNECT.as_secs()
+                        );
+                        self.disconnect();
+                    }
                 }
             }
             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 (tx, rx) = std::sync::mpsc::channel();
+        let peer = Peer::new(
+            "bob".to_string(),
+            ConnectionType::P,
+            "127.0.0.1".to_string(),
+            u32::from(addr.port()),
+            None,
+            0,
+            0,
+            0,
+        );
+        let mut actor =
+            PeerActor::new(peer, Some(stream), None, tx, "me".to_string(), 7);
+        actor.on_start();
+        (actor, rx, far_end)
+    }
+
+    #[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:?}"),
+        }
+    }
+}

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

@@ -60,15 +60,23 @@ impl PeerRegistry {
             id,
         );
 
-        let handle =
+        // Take the map lock before the actor exists: a peer that dies
-            self.actor_system.spawn_with_handle(actor, |actor, handle| {
+        // instantly (refused dial, immediate hangup) reports its terminal
-                actor.set_self_handle(handle);
+        // outcome to the client loop, whose eviction takes this same lock.
-            });
+        // With the insert racing the spawn, eviction could run first, find
-
+        // nothing, and the entry inserted afterwards became a permanent
+        // zombie claiming the username.
         let mut peers = self
             .peers
             .lock_safe()
             .map_err(|e| format!("peer registry lock poisoned: {e}"))?;
+
+        let handle = self
+            .actor_system
+            .try_spawn_with_handle(actor, |actor, handle| {
+                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),
@@ -233,4 +241,51 @@ mod tests {
         let _ = handle.unwrap().stop();
         assert!(!registry.contains("bob"));
     }
+
+    // A dial that is refused reports its terminal outcome almost instantly —
+    // possibly while register_peer is still between spawn and insert. The
+    // registry holds its lock across both, so the eviction that follows must
+    // always find the entry; a miss left a permanent zombie claiming the
+    // username.
+    #[test]
+    fn refused_dial_does_not_leave_a_zombie_entry() {
+        use crate::client::ClientOperation;
+
+        let system = Arc::new(ActorSystem::new());
+        let (tx, rx) = std::sync::mpsc::channel();
+        let registry = PeerRegistry::new(system, tx, "me".to_string());
+
+        // A port with nothing listening behind it: bind, learn it, drop it.
+        let port = {
+            let probe = TcpListener::bind("127.0.0.1:0").unwrap();
+            probe.local_addr().unwrap().port()
+        };
+
+        let peer = Peer::new(
+            "ghost".to_string(),
+            ConnectionType::P,
+            "127.0.0.1".to_string(),
+            u32::from(port),
+            None,
+            0,
+            0,
+            0,
+        );
+        registry.register_peer(peer, None, None).unwrap();
+
+        // Play the client ops loop: take the terminal outcome, evict by id.
+        match rx.recv_timeout(std::time::Duration::from_secs(10)) {
+            Ok(ClientOperation::PeerConnectFailed(id, username)) => {
+                assert_eq!(username, "ghost");
+                if let Some(handle) = registry.remove_peer_if(&username, id) {
+                    let _ = handle.stop();
+                }
+            }
+            other => panic!("expected PeerConnectFailed, got {other:?}"),
+        }
+        assert!(
+            !registry.contains("ghost"),
+            "a refused dial must not leave a registry entry"
+        );
+    }
 }

soulseek-rs-lib/src/client/connection.rs

@@ -2,7 +2,7 @@ use super::{
     Arc, Client, ClientContext, ClientOperation, ConnectionType, DownloadPeer,
     DownloadStatus, Listen, Peer, PeerRegistry, Receiver, Result, RwLock,
     RwLockExt, Sender, ServerActor, ServerMessage, Shares, SoulseekRs,
-    TcpStream, error, info, mpsc, thread, trace,
+    TcpStream, error, info, mpsc, thread, trace, warn,
 };
 
 /// Ceiling on the wait for a login verdict. Generous enough for a slow server
@@ -10,8 +10,40 @@ use super::{
 const LOGIN_RESPONSE_TIMEOUT: std::time::Duration =
     std::time::Duration::from_secs(45);
 
+/// Soft open-file limit below which a session should expect "too many open
+/// files": every peer holds a socket, and one busy search talks to hundreds
+/// of them.
+const COMFORTABLE_OPEN_FILES: u64 = 1024;
+
+/// Say up front when the process cannot hold a busy session's sockets, so an
+/// operator reads a warning with a number in it instead of an EMFILE storm an
+/// hour later. The embedding process owns its rlimits; the library only looks.
+#[cfg(unix)]
+fn warn_if_open_file_limit_is_low() {
+    // SAFETY: getrlimit only fills the struct it is handed.
+    let limit = unsafe {
+        let mut limit = std::mem::zeroed::<libc::rlimit>();
+        if libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) != 0 {
+            return;
+        }
+        limit.rlim_cur
+    };
+    if limit < COMFORTABLE_OPEN_FILES {
+        warn!(
+            "only {limit} open files allowed; a busy session wants \
+             {COMFORTABLE_OPEN_FILES} or more — expect \"too many open \
+             files\" under load"
+        );
+    }
+}
+
+#[cfg(not(unix))]
+fn warn_if_open_file_limit_is_low() {}
+
 impl Client {
     pub fn connect(&mut self) -> Result<()> {
+        warn_if_open_file_limit_is_low();
+
         let (sender, message_reader): (
             Sender<ClientOperation>,
             Receiver<ClientOperation>,

soulseek-rs-lib/src/client/mod.rs

@@ -248,7 +248,7 @@ pub enum ClientOperation {
         query: String,
     },
     /// A peer queued one of our shared files; `requester_key` is the registry
-    /// key of the peer actor (may carry a `:direct` suffix).
+    /// key of the peer actor — the peer's username.
     QueueUpload {
         requester_key: String,
         filename: String,

soulseek-rs-lib/src/client/operations.rs

@@ -1,4 +1,3 @@
-use super::uploads::downloader_of;
 use super::{
     Arc, BROKER_CONNECT_TIMEOUT, Client, ClientContext, ClientOperation,
     ConnectionType, Download, DownloadPeer, DownloadStatus, Peer, PeerMessage,
@@ -75,6 +74,14 @@ impl Client {
                                 let _ = handle.stop();
                             }
                         }
+                        // Only an error is evidence the peer is gone. A clean
+                        // close — our idle reaper, or a remote client tidying
+                        // an idle socket while it waits in our queue — must
+                        // not throw away everything that peer has queued: the
+                        // connection comes back, the queue cannot. A departed
+                        // peer wedging uploads shut is still covered: its
+                        // erroring transfer fails within the socket deadlines
+                        // and the error branch here frees its slots.
                         if let Some(error) = error {
                             warn!(
                                 "[client] Peer {} disconnected with error: {:?}",
@@ -85,12 +92,11 @@ impl Client {
                                 &username,
                                 None,
                             );
+                            Self::release_upload_slots(
+                                &client_context,
+                                &username,
+                            );
                         }
-                        // A peer that leaves must not keep occupying an
-                        // upload slot: with a slot cap, a couple of
-                        // departed peers would otherwise wedge uploads
-                        // shut for the rest of the session.
-                        Self::release_upload_slots(&client_context, &username);
                     }
                     ClientOperation::DownloadFromPeer(token, peer, allowed) => {
                         let maybe_download = match client_context.read_safe() {
@@ -440,10 +446,11 @@ impl Client {
                         }
                     }
                     ClientOperation::PeerConnected(username) => {
-                        // An outbound control connection just handshook.
+                        // A control connection just came up — one we dialled,
-                        // Flush any downloads that were queued for this
+                        // or an inbound one the listener registered. Flush any
-                        // peer while we were still connecting. Collect
+                        // downloads that were queued for this peer while it
-                        // under a read guard, then act without it held.
+                        // was unreachable. Collect under a read guard, then
+                        // act without it held.
                         let (registry, files): (
                             Option<PeerRegistry>,
                             Vec<String>,
@@ -547,8 +554,6 @@ impl Client {
                         filename,
                     } => {
                         // The peer served next may not be this one.
-                        let downloader =
-                            downloader_of(&requester_key).to_string();
                         match client_context.write_safe() {
                             Ok(mut ctx) => {
                                 let Some(file) = ctx.shares.get(&filename)
@@ -563,7 +568,6 @@ impl Client {
                                 let real_path = file.real_path.clone();
                                 ctx.enqueue_upload(
                                     &requester_key,
-                                    &downloader,
                                     &filename,
                                     real_path,
                                     size,
@@ -580,12 +584,11 @@ impl Client {
                         requester_key,
                         filename,
                     } => {
-                        let downloader = downloader_of(&requester_key);
                         let (registry, place) = match client_context.read_safe()
                         {
                             Ok(ctx) => (
                                 ctx.peer_registry.clone(),
-                                ctx.place_in_queue(downloader, &filename),
+                                ctx.place_in_queue(&requester_key, &filename),
                             ),
                             Err(_) => continue,
                         };

soulseek-rs-lib/src/client/tests.rs

@@ -327,3 +327,71 @@ fn upload_speed_is_reported_only_while_running() {
         "unexpected rate {rate}"
     );
 }
+
+// A clean disconnect — the idle reaper, or a remote client tidying an idle
+// socket while it waits in our queue — must keep the peer's queued uploads;
+// only an error disconnect is evidence the peer is gone.
+#[test]
+fn a_clean_disconnect_keeps_queued_uploads_an_error_drops_them() {
+    let client = Client::new("test-user", "test-password");
+    let (ops, ops_rx) = mpsc::channel();
+    Client::listen_to_client_operations(
+        ops_rx,
+        client.context.clone(),
+        "me".to_string(),
+    );
+
+    client.context.write().unwrap().enqueue_upload(
+        "amy",
+        "@@share\\f.mp3",
+        std::path::PathBuf::from("/tmp/f.mp3"),
+        4096,
+    );
+
+    ops.send(ClientOperation::PeerDisconnected(
+        1,
+        "amy".to_string(),
+        None,
+    ))
+    .unwrap();
+    // The loop is serial, so once this fence op is visible the disconnect
+    // before it has been handled.
+    ops.send(ClientOperation::OwnPrivileges(7)).unwrap();
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while client.own_privilege_seconds() != Some(7) {
+        assert!(Instant::now() < deadline, "ops loop never caught up");
+        thread::sleep(Duration::from_millis(10));
+    }
+    assert_eq!(
+        client
+            .context
+            .read()
+            .unwrap()
+            .place_in_queue("amy", "@@share\\f.mp3"),
+        Some(1),
+        "a clean disconnect must keep the queue"
+    );
+
+    ops.send(ClientOperation::PeerDisconnected(
+        1,
+        "amy".to_string(),
+        Some(crate::error::SoulseekRs::NotConnected),
+    ))
+    .unwrap();
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        let place = client
+            .context
+            .read()
+            .unwrap()
+            .place_in_queue("amy", "@@share\\f.mp3");
+        if place.is_none() {
+            break;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "an error disconnect must drop the queue"
+        );
+        thread::sleep(Duration::from_millis(10));
+    }
+}

soulseek-rs-lib/src/client/upload_queue.rs

@@ -31,9 +31,7 @@ pub struct UploadOffer {
 /// A file a peer has asked us for and is waiting on.
 #[derive(Debug, Clone)]
 pub struct QueuedUpload {
-    /// Registry key of the asking peer's actor, which may carry a `:direct`
+    /// The asking peer's username — also the registry key of its actor.
-    /// suffix — not the same string as `downloader`.
-    pub requester_key: String,
     pub downloader: String,
     pub virtual_path: String,
     pub real_path: PathBuf,
@@ -98,7 +96,6 @@ impl ClientContext {
     /// for a file we do not have never reaches the queue.
     pub fn enqueue_upload(
         &mut self,
-        requester_key: &str,
         downloader: &str,
         filename: &str,
         real_path: std::path::PathBuf,
@@ -106,7 +103,6 @@ impl ClientContext {
     ) {
         self.upload_seq += 1;
         self.upload_queue.push(QueuedUpload {
-            requester_key: requester_key.to_string(),
             downloader: downloader.to_string(),
             virtual_path: filename.to_string(),
             real_path,
@@ -134,7 +130,7 @@ impl ClientContext {
             let job = self.upload_queue.remove(index);
             let token = next_token();
             offers.push(UploadOffer {
-                requester_key: job.requester_key,
+                requester_key: job.downloader.clone(),
                 token,
                 virtual_path: job.virtual_path.clone(),
                 size: job.size,
@@ -259,7 +255,6 @@ mod tests {
 
     fn queued(user: &str, privileged: bool, seq: u64) -> QueuedUpload {
         QueuedUpload {
-            requester_key: user.to_string(),
             downloader: user.to_string(),
             virtual_path: format!("@@share\\{user}.mp3"),
             real_path: PathBuf::from("/tmp/x.mp3"),
@@ -365,7 +360,7 @@ mod context_tests {
     }
 
     fn ask(ctx: &mut ClientContext, user: &str, file: &str) {
-        ctx.enqueue_upload(user, user, file, PathBuf::from("/tmp/x"), 4096);
+        ctx.enqueue_upload(user, file, PathBuf::from("/tmp/x"), 4096);
     }
 
     /// Who the pump offered to, by the order it offered them.

soulseek-rs-lib/src/client/uploads.rs

@@ -5,14 +5,6 @@ use super::{
 use crate::types::UploadStatus;
 use std::sync::atomic::{AtomicBool, AtomicU64};
 
-/// The peer behind a registry key, which may carry a `:direct` suffix the
-/// server never sees.
-pub fn downloader_of(requester_key: &str) -> &str {
-    requester_key
-        .strip_suffix(":direct")
-        .unwrap_or(requester_key)
-}
-
 impl Client {
     /// Whether the server listed `username` as privileged. Privileged peers are
     /// served before everyone else.

soulseek-rs-lib/src/peer/download_peer.rs

@@ -231,6 +231,19 @@ impl DownloadPeer {
         )
         .map_err(DownloadError::ConnectionFailed)?;
 
+        Self::apply_transfer_deadlines(&stream)?;
+
+        Ok(stream)
+    }
+
+    /// Bound every stall on a transfer socket. This must cover streams the
+    /// listener accepted as well as ones we dialled: an accepted F connection
+    /// used to arrive here with no read timeout at all, so a peer that sent a
+    /// token and went silent parked the transfer thread — holding the socket
+    /// and the open `.part` file — for the rest of the session.
+    fn apply_transfer_deadlines(
+        stream: &TcpStream,
+    ) -> Result<(), DownloadError> {
         stream
             .set_read_timeout(Some(Duration::from_secs(30)))
             .map_err(DownloadError::ConnectionFailed)?;
@@ -240,8 +253,7 @@ impl DownloadPeer {
         stream
             .set_nodelay(true)
             .map_err(DownloadError::ConnectionFailed)?;
-
+        Ok(())
-        Ok(stream)
     }
 
     fn perform_handshake(
@@ -481,7 +493,10 @@ impl DownloadPeer {
         }
 
         let mut stream = match stream {
-            Some(s) => s,
+            Some(s) => {
+                Self::apply_transfer_deadlines(&s)?;
+                s
+            }
             None => self.establish_connection()?,
         };
 
@@ -622,6 +637,23 @@ mod tests {
         let _ = std::fs::remove_dir_all(dir);
     }
 
+    // Streams the listener accepted arrive with no deadlines; without these a
+    // peer that sends a token and goes silent parks the transfer thread —
+    // socket and open .part file included — for the rest of the session.
+    #[test]
+    fn supplied_streams_get_transfer_deadlines() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let stream =
+            std::net::TcpStream::connect(listener.local_addr().unwrap())
+                .unwrap();
+
+        DownloadPeer::apply_transfer_deadlines(&stream).unwrap();
+
+        let secs = std::time::Duration::from_secs;
+        assert_eq!(stream.read_timeout().unwrap(), Some(secs(30)));
+        assert_eq!(stream.write_timeout().unwrap(), Some(secs(5)));
+    }
+
     #[test]
     fn test_establish_connection_invalid_address() {
         let download_peer = DownloadPeer::new(

soulseek-rs-lib/src/peer/listen.rs

@@ -178,7 +178,7 @@ fn handle_file_connection(
     let failure_token = download.as_ref().map(|d| d.token);
 
     let download_peer = DownloadPeer::new(
-        format!("{}:direct", peer.username),
+        peer.username.clone(),
         peer.host.clone(),
         peer.port,
         token,
@@ -325,8 +325,13 @@ fn handle_incoming_connection(
         init_data.username, init_data.connection_type, init_data.token
     );
 
+    // Register under the plain username, the same key outbound connections
+    // use. A suffixed key ("user:direct") put the same peer in the registry
+    // twice — two sockets, two threads — and made every "already connected?"
+    // check miss, so an inbound searcher still got a second, outbound
+    // connection dialled at them.
     let peer = Peer::new(
-        format!("{}:direct", init_data.username),
+        init_data.username.clone(),
         init_data.connection_type.clone(),
         peer_ip.clone(),
         peer_port.into(),
@@ -339,6 +344,12 @@ fn handle_incoming_connection(
     match init_data.connection_type {
         ConnectionType::P => {
             handle_peer_connection(peer, stream, reader, &context);
+            // Inbound peers don't self-announce, so nudge the client to flush
+            // anything queued for this now-connected peer — same as the
+            // PierceFirewall path.
+            let _ = context
+                .client_sender
+                .send(ClientOperation::PeerConnected(init_data.username));
         }
 
         ConnectionType::F => handle_file_connection(

soulseek-rs-lib/src/peer/upload_peer.rs

@@ -42,6 +42,17 @@ pub fn serve_file(
         })?;
     let mut stream =
         TcpStream::connect_timeout(&socket, Duration::from_secs(20))?;
+    // A downloader that connects and then goes silent must cost seconds, not
+    // the rest of the session: without deadlines a wedged `read_exact` (or a
+    // zero-window `write_all`) holds this thread, the socket, the open file,
+    // and — because the upload stays InProgress — an upload slot, forever.
+    // Thirty seconds of total silence is a dead transfer, not a slow one: even
+    // a slow reader drains some buffer within that. The write deadline is
+    // deliberately looser than the download side's 5s: bulk data flows this
+    // way, and a congested but living peer may hold the send buffer full far
+    // longer than a peer draining our tiny control frames ever would.
+    stream.set_read_timeout(Some(Duration::from_secs(30)))?;
+    stream.set_write_timeout(Some(Duration::from_secs(30)))?;
     stream.set_nodelay(true).ok();
 
     // PeerInit(F) + the 4-byte token in a single write so they coalesce.

soulseek-rs-lib/tests/e2e.rs

@@ -734,14 +734,14 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
         .recv_timeout(Duration::from_secs(5))
         .expect("mock uploader P connection");
 
-    // The listener registers an incoming peer under "<username>:direct"; give
+    // The listener registers an incoming peer under its plain username; give
     // that registration a moment to complete before queuing the download.
     std::thread::sleep(Duration::from_millis(1500));
 
     let (_download, status_rx) = client
         .download(
             filename.to_string(),
-            "e2e_mockpeer:direct".to_string(),
+            "e2e_mockpeer".to_string(),
             size,
             download_dir.display().to_string(),
         )
@@ -811,7 +811,7 @@ fn an_interrupted_download_resumes_from_its_partial_file() {
     let (_download, status_rx) = client
         .download(
             filename.to_string(),
-            "e2e_resume_peer:direct".to_string(),
+            "e2e_resume_peer".to_string(),
             size,
             download_dir.display().to_string(),
         )

soulseek-rs/src/daemon/mod.rs

@@ -44,15 +44,6 @@ pub const TOKEN_NAME: &str = "daemon.token";
 /// loses little, rare enough not to churn the disk on a busy queue.
 const SAVE_INTERVAL: Duration = Duration::from_secs(5);
 
-/// Open files the daemon asks the OS for at startup.
-///
-/// The client holds a socket per peer and will talk to a thousand of them
-/// during a busy search, plus transfers, the share index, and control
-/// connections. A shell hands its children a soft limit of 256 on macOS, which
-/// a single search blows through — and a one-shot command never noticed
-/// because it exited before it mattered. A resident process has to ask.
-const WANTED_OPEN_FILES: u64 = 8192;
-
 #[must_use]
 pub fn socket_path() -> Option<PathBuf> {
     crate::persist::paths::state_dir().map(|dir| dir.join(SOCKET_NAME))
@@ -161,21 +152,6 @@ pub fn run(
     let token = load_or_create_token(&state_dir.join(TOKEN_NAME))
         .map_err(|e| CliError::usage(format!("cannot read the token: {e}")))?;
 
-    // Before anything opens a socket: running out of file descriptors mid
-    // search leaves the client unable to accept peers *or* control
-    // connections, and the only symptom is everything quietly failing.
-    match raise_open_file_limit() {
-        Ok(limit) if limit < WANTED_OPEN_FILES => ctx.out.warn(&format!(
-            "only {limit} open files allowed; a busy search wants \
-             {WANTED_OPEN_FILES}. Raise it with `ulimit -n {WANTED_OPEN_FILES}` \
-             before starting the daemon."
-        )),
-        Ok(_) => {}
-        Err(e) => ctx
-            .out
-            .warn(&format!("could not check the open-file limit: {e}")),
-    }
-
     // Bind before logging in: a second daemon should fail in a millisecond,
     // not after a full login handshake against the server.
     let mut listeners = Vec::new();
@@ -272,59 +248,6 @@ pub fn run(
     Ok(())
 }
 
-/// Ask the OS for enough open files to run a busy session, and report what we
-/// actually got.
-///
-/// Raising our own soft limit is what a service does: the limit a shell
-/// inherits is sized for a shell. The hard limit is the ceiling, and asking
-/// past it fails outright rather than clamping, so this asks for the smaller
-/// of what we want and what we are allowed.
-#[cfg(unix)]
-fn raise_open_file_limit() -> std::io::Result<u64> {
-    // SAFETY: every call takes a pointer to a `rlimit` this function owns, and
-    // none of them retains it.
-    unsafe {
-        let mut limit = std::mem::zeroed::<libc::rlimit>();
-        if libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) != 0 {
-            return Err(std::io::Error::last_os_error());
-        }
-        if limit.rlim_cur >= WANTED_OPEN_FILES {
-            return Ok(limit.rlim_cur);
-        }
-
-        let previous = limit.rlim_cur;
-        let ceiling = if limit.rlim_max == libc::RLIM_INFINITY {
-            WANTED_OPEN_FILES
-        } else {
-            limit.rlim_max.min(WANTED_OPEN_FILES)
-        };
-
-        // macOS rejects an unlimited *hard* limit for open files, so raising
-        // the soft one while leaving `rlim_max` as it came back fails with
-        // EINVAL. Naming both is what it will accept.
-        for candidate in [
-            libc::rlimit {
-                rlim_cur: ceiling,
-                rlim_max: limit.rlim_max,
-            },
-            libc::rlimit {
-                rlim_cur: ceiling,
-                rlim_max: ceiling,
-            },
-        ] {
-            if libc::setrlimit(libc::RLIMIT_NOFILE, &raw const candidate) == 0 {
-                return Ok(ceiling);
-            }
-        }
-        Ok(previous)
-    }
-}
-
-#[cfg(not(unix))]
-const fn raise_open_file_limit() -> std::io::Result<u64> {
-    Ok(WANTED_OPEN_FILES)
-}
-
 /// Say what is now possible, once, on stderr.
 ///
 /// The token goes here rather than only into a file because the moment you

soulseek-rs/src/run.rs

@@ -16,6 +16,23 @@ use std::time::Duration;
 use std::{env, io};
 
 pub fn run(mut cli: Cli, out: &Out) -> CliResult {
+    // Before anything opens a socket: every entry point that logs in — the
+    // TUI, `serve`, one-shot commands, the daemon — holds a socket per peer,
+    // and the soft limit a shell hands its children (256 on macOS) is gone
+    // after one busy search. The raise lived in the daemon first; the other
+    // entry points run the same peer machinery and crashed the same way.
+    match raise_open_file_limit() {
+        Ok(limit) if limit < WANTED_OPEN_FILES => out.warn(&format!(
+            "only {limit} open files allowed; a busy search wants \
+             {WANTED_OPEN_FILES}. Raise it with `ulimit -n \
+             {WANTED_OPEN_FILES}` first."
+        )),
+        Ok(_) => {}
+        Err(e) => {
+            out.warn(&format!("could not check the open-file limit: {e}"));
+        }
+    }
+
     let config_path = config_path(&cli);
     let file_config = match &config_path {
         Some(path) => crate::persist::config::FileConfig::load(path)
@@ -557,3 +574,64 @@ fn persist_credentials(
         eprintln!("⚠️  Could not store password in keychain: {e}");
     }
 }
+
+/// Open files a session asks the OS for at startup.
+///
+/// The client holds a socket per peer and will talk to a thousand of them
+/// during a busy search, plus transfers, the share index, and control
+/// connections. A shell hands its children a soft limit of 256 on macOS,
+/// which a single search blows through.
+const WANTED_OPEN_FILES: u64 = 8192;
+
+/// Ask the OS for enough open files to run a busy session, and report what we
+/// actually got.
+///
+/// Raising our own soft limit is what a service does: the limit a shell
+/// inherits is sized for a shell. The hard limit is the ceiling, and asking
+/// past it fails outright rather than clamping, so this asks for the smaller
+/// of what we want and what we are allowed.
+#[cfg(unix)]
+fn raise_open_file_limit() -> std::io::Result<u64> {
+    // SAFETY: every call takes a pointer to a `rlimit` this function owns, and
+    // none of them retains it.
+    unsafe {
+        let mut limit = std::mem::zeroed::<libc::rlimit>();
+        if libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) != 0 {
+            return Err(std::io::Error::last_os_error());
+        }
+        if limit.rlim_cur >= WANTED_OPEN_FILES {
+            return Ok(limit.rlim_cur);
+        }
+
+        let previous = limit.rlim_cur;
+        let ceiling = if limit.rlim_max == libc::RLIM_INFINITY {
+            WANTED_OPEN_FILES
+        } else {
+            limit.rlim_max.min(WANTED_OPEN_FILES)
+        };
+
+        // macOS rejects an unlimited *hard* limit for open files, so raising
+        // the soft one while leaving `rlim_max` as it came back fails with
+        // EINVAL. Naming both is what it will accept.
+        for candidate in [
+            libc::rlimit {
+                rlim_cur: ceiling,
+                rlim_max: limit.rlim_max,
+            },
+            libc::rlimit {
+                rlim_cur: ceiling,
+                rlim_max: ceiling,
+            },
+        ] {
+            if libc::setrlimit(libc::RLIMIT_NOFILE, &raw const candidate) == 0 {
+                return Ok(ceiling);
+            }
+        }
+        Ok(previous)
+    }
+}
+
+#[cfg(not(unix))]
+const fn raise_open_file_limit() -> std::io::Result<u64> {
+    Ok(WANTED_OPEN_FILES)
+}