From ea25325e682bd0c315dff6e8d700f5856a3bdc45 Mon Sep 17 00:00:00 2001 From: michel Date: Sun, 19 Jul 2026 17:47:45 +0200 Subject: [PATCH] fix: reap failed outbound peer actors and bound the brokered connect Two related lifecycle defects in the direct/firewalled download path: - On a failed direct connect, ClientOperation::PeerConnectFailed left the dead outbound actor registered and running. The actor pinned a thread-pool worker forever (its loop only exits on Stop), so a handful of unreachable-peer downloads could exhaust the fixed pool and wedge the whole client; and a later download() to the same peer saw registry.contains()==true and queued into the dead, streamless actor, hanging silently. The arm now removes and stops the dead actor before brokering, mirroring PeerDisconnected. - The server-brokered fallback had no timeout, so a peer that never connects back (offline / double-firewalled / unroutable address) left the download Queued forever and its Receiver blocked. Arm a bounded timeout: if no PierceFirewall consumes the correlation token within BROKER_CONNECT_TIMEOUT, fail the peer's queued downloads (on the channel and in the store) and reclaim the token. A successful pierce consumes the token first, so the timeout is a no-op on the happy path. Add a unit test for the queued-download failure helper; the direct and firewalled e2e tests still pass (pierce arrives well within the timeout). --- soulseek-rs-lib/src/client.rs | 135 +++++++++++++++++++++++++++++++--- 1 file changed, 126 insertions(+), 9 deletions(-) diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index 5822848..620d4f0 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -31,6 +31,10 @@ use std::{ use crate::{debug, error, info, trace, warn}; const DEFAULT_LISTEN_PORT: u16 = 2234; +/// How long to wait for a server-brokered (firewalled) peer to connect back +/// before giving up and failing the download. Matches the direct-dial timeout. +const BROKER_CONNECT_TIMEOUT: Duration = Duration::from_secs(20); + /// Source of non-zero correlation tokens for server-brokered connections. static NEXT_CONNECT_TOKEN: AtomicU32 = AtomicU32::new(1); @@ -289,6 +293,39 @@ fn download_without_a_connection_resolves_failed() { )); } +#[test] +fn fail_queued_downloads_notifies_receiver_and_store() { + // When a brokered connect times out, every Queued download for the peer + // must resolve to Failed both on its channel and in the store. + let client = Client::new("u", "p"); + let (sender, receiver) = mpsc::channel(); + client.context.write().unwrap().add_download(Download { + username: "peer".to_string(), + filename: "f.mp3".to_string(), + token: 7, + size: 10, + download_directory: "d".to_string(), + status: DownloadStatus::Queued, + sender, + queue_position: None, + metadata: DownloadMetadata::default(), + }); + + Client::fail_queued_downloads(&client.context, "peer"); + + assert!(matches!(receiver.try_recv(), Ok(DownloadStatus::Failed))); + assert!(matches!( + client + .context + .read() + .unwrap() + .get_download_by_token(7) + .unwrap() + .status, + DownloadStatus::Failed + )); +} + #[test] fn test_client_removes_only_queued_downloads() { let client = Client::new("test-user", "test-password"); @@ -783,6 +820,34 @@ impl Client { Ok((download, download_receiver)) } + /// Fail every still-`Queued` download for `username`, both on the caller's + /// status channel (so a blocked `Receiver` unblocks) and in the store. + fn fail_queued_downloads( + client_context: &Arc>, + username: &str, + ) { + let mut context = match client_context.write_safe() { + Ok(c) => c, + Err(e) => { + error!("[client] fail_queued_downloads write: {}", e); + return; + } + }; + let doomed: Vec<(u32, Sender)> = context + .get_downloads() + .iter() + .filter(|d| { + d.username == username + && matches!(d.status, DownloadStatus::Queued) + }) + .map(|d| (d.token, d.sender.clone())) + .collect(); + for (token, sender) in doomed { + let _ = sender.send(DownloadStatus::Failed); + context.update_download_with_status(token, DownloadStatus::Failed); + } + } + fn process_failed_uploads( client_context: Arc>, username: &str, @@ -1318,6 +1383,19 @@ impl Client { .write_safe() { Ok(mut ctx) => { + // Reap the dead outbound actor so it + // stops pinning a pool worker and no + // longer shadows the brokered reconnect + // (a stale registry entry would make + // later downloads queue into a dead, + // streamless actor and hang). + if let Some(handle) = + ctx.peer_registry.as_ref().and_then( + |r| r.remove_peer(&username), + ) + { + let _ = handle.stop(); + } ctx.add_pending_connect( token, username.clone(), @@ -1332,15 +1410,54 @@ impl Client { continue; } }; - if let Some(sender) = server_sender { - let msg = crate::message::server::MessageFactory::build_connect_to_peer( - token, - &username, - ConnectionType::P, - ); - let _ = sender - .send(ServerMessage::SendMessage(msg)); - } + let Some(sender) = server_sender else { + continue; + }; + let msg = crate::message::server::MessageFactory::build_connect_to_peer( + token, + &username, + ConnectionType::P, + ); + let _ = sender + .send(ServerMessage::SendMessage(msg)); + + // Bound the brokered attempt: if no PierceFirewall + // arrives, fail the peer's queued downloads (so + // the caller's Receiver unblocks) and reclaim the + // token. A successful pierce consumes the token + // first, making this a no-op. + let timeout_ctx = client_context.clone(); + let timeout_user = username.clone(); + thread::spawn(move || { + sleep(BROKER_CONNECT_TIMEOUT); + let still_pending = timeout_ctx + .write_safe() + .is_ok_and(|mut c| { + c.take_pending_connect(token) + .is_some() + }); + if still_pending { + if let Some(handle) = timeout_ctx + .read_safe() + .ok() + .and_then(|c| { + c.peer_registry + .as_ref() + .and_then(|r| { + r.remove_peer( + &timeout_user, + ) + }) + }) + { + let _ = handle.stop(); + } + Self::fail_queued_downloads( + &timeout_ctx, + &timeout_user, + ); + } + }); } } }