diff --git a/Cargo.lock b/Cargo.lock index 1195697..0688f05 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2569,6 +2569,9 @@ dependencies = [ [[package]] name = "soulseek-rs-lib" version = "13.0.0" +dependencies = [ + "libc", +] [[package]] name = "stable_deref_trait" diff --git a/soulseek-rs-lib/Cargo.toml b/soulseek-rs-lib/Cargo.toml index 30e2b37..34c8298 100644 --- a/soulseek-rs-lib/Cargo.toml +++ b/soulseek-rs-lib/Cargo.toml @@ -18,3 +18,6 @@ path = "src/lib.rs" workspace = true [dependencies] + +[target.'cfg(unix)'.dependencies] +libc = "0.2.189" diff --git a/soulseek-rs-lib/src/actor/mod.rs b/soulseek-rs-lib/src/actor/mod.rs index 45122a1..d9eb252 100644 --- a/soulseek-rs-lib/src/actor/mod.rs +++ b/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( &self, - mut actor: A, + actor: A, init: F, ) -> ActorHandle + where + F: FnOnce(&mut A, ActorHandle) + 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( + &self, + mut actor: A, + init: F, + ) -> std::io::Result> where F: FnOnce(&mut A, ActorHandle) + 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 { - error!("[actor_system] failed to spawn actor thread: {}", e); - } - - handle + Ok(handle) } fn run_actor_loop( diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 825bb4f..ad8710f 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/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, + /// 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, 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:?}"), + } + } +} diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index 60e136b..7eec579 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -60,15 +60,23 @@ impl PeerRegistry { id, ); - let handle = - self.actor_system.spawn_with_handle(actor, |actor, handle| { - actor.set_self_handle(handle); - }); - + // Take the map lock before the actor exists: a peer that dies + // instantly (refused dial, immediate hangup) reports its terminal + // 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" + ); + } } diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index cfa722d..da2bd69 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/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::(); + 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, Receiver, diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 6f5de79..f473408 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/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, diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index e3303b3..7bdc715 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/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. - // Flush any downloads that were queued for this - // peer while we were still connecting. Collect - // under a read guard, then act without it held. + // A control connection just came up — one we dialled, + // or an inbound one the listener registered. Flush any + // downloads that were queued for this peer while it + // was unreachable. Collect under a read guard, then + // act without it held. let (registry, files): ( Option, Vec, @@ -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, }; diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index f95dad7..9179497 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/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)); + } +} diff --git a/soulseek-rs-lib/src/client/upload_queue.rs b/soulseek-rs-lib/src/client/upload_queue.rs index 70ddc9f..9bc03b9 100644 --- a/soulseek-rs-lib/src/client/upload_queue.rs +++ b/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` - /// suffix — not the same string as `downloader`. - pub requester_key: String, + /// The asking peer's username — also the registry key of its actor. 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. diff --git a/soulseek-rs-lib/src/client/uploads.rs b/soulseek-rs-lib/src/client/uploads.rs index 3b3cc2c..cd42654 100644 --- a/soulseek-rs-lib/src/client/uploads.rs +++ b/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. diff --git a/soulseek-rs-lib/src/peer/download_peer.rs b/soulseek-rs-lib/src/peer/download_peer.rs index 6828743..3ea609e 100644 --- a/soulseek-rs-lib/src/peer/download_peer.rs +++ b/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(stream) + Ok(()) } 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( diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index e647ba1..ed77a24 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/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( diff --git a/soulseek-rs-lib/src/peer/upload_peer.rs b/soulseek-rs-lib/src/peer/upload_peer.rs index 33921eb..b102ba3 100644 --- a/soulseek-rs-lib/src/peer/upload_peer.rs +++ b/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. diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index a445c88..e31aac5 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/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 ":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(), ) diff --git a/soulseek-rs/src/daemon/mod.rs b/soulseek-rs/src/daemon/mod.rs index 6a9641a..ad9584e 100644 --- a/soulseek-rs/src/daemon/mod.rs +++ b/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 { 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 { - // 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::(); - 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 { - 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 diff --git a/soulseek-rs/src/run.rs b/soulseek-rs/src/run.rs index 8727af3..af56d40 100644 --- a/soulseek-rs/src/run.rs +++ b/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 { + // 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::(); + 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 { + Ok(WANTED_OPEN_FILES) +}