diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index d4098d9..e0d71ad 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -107,6 +107,8 @@ struct UploadJob { real_path: std::path::PathBuf, virtual_path: String, size: u64, + /// When the offer went out; `None` once the peer has answered it. + offered: Option, } /// Live bookkeeping for an upload being served (or recently finished). diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 1bc662c..7eb0e25 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -23,6 +23,7 @@ impl Client { if last_sweep.elapsed() >= CONNECT_SWEEP_INTERVAL { last_sweep = Instant::now(); Self::sweep_expired_connects(&client_context); + Self::sweep_stale_offers(&client_context); } let operation = match next { Ok(operation) => operation, @@ -718,9 +719,10 @@ impl Client { // address (from the code-9 GetPeerAddress) and // stream the file, or queue until it resolves. let (job_addr, downloader) = match client_context - .read_safe() + .write_safe() { - Ok(ctx) => { + Ok(mut ctx) => { + ctx.mark_offer_answered(token); let Some(job) = ctx.uploads.get(&token) else { continue; }; @@ -875,6 +877,15 @@ impl Client { } } + fn sweep_stale_offers(client_context: &Arc>) { + let freed = client_context + .write_safe() + .is_ok_and(|mut ctx| ctx.expire_stale_offers(Instant::now())); + if freed { + Self::pump_upload_queue(client_context); + } + } + fn sweep_expired_connects(client_context: &Arc>) { let expired = match client_context.write_safe() { Ok(mut ctx) => ctx.take_expired_connects(Instant::now()), diff --git a/soulseek-rs-lib/src/client/upload_queue.rs b/soulseek-rs-lib/src/client/upload_queue.rs index 474c860..4c59f85 100644 --- a/soulseek-rs-lib/src/client/upload_queue.rs +++ b/soulseek-rs-lib/src/client/upload_queue.rs @@ -10,7 +10,7 @@ //! policy (per-user fairness, small-file-first) would be a different promise //! than the one the rules ask for. -use super::{ClientContext, PeerRegistry, UploadJob}; +use super::{ClientContext, Duration, Instant, PeerRegistry, UploadJob}; use crate::types::{UploadInfo, UploadStatus}; use std::path::PathBuf; @@ -18,6 +18,11 @@ use std::path::PathBuf; /// them cannot grow the list without bound. const MAX_UPLOAD_EVENTS: usize = 256; +/// How long an offered slot waits for the peer's TransferResponse. A peer that +/// never answers, having gone away or hung, would otherwise hold the slot for +/// the life of the process; with two slots, two of them shut uploads down. +const OFFER_TIMEOUT: Duration = Duration::from_secs(30); + /// A queued upload that has just been given a slot, ready to be offered to the /// peer that asked for it. #[derive(Debug, Clone)] @@ -142,6 +147,7 @@ impl ClientContext { real_path: job.real_path, virtual_path: job.virtual_path, size: job.size, + offered: Some(Instant::now()), }, ); } @@ -239,6 +245,46 @@ impl ClientContext { self.privileged_users.contains(username) } + /// Drop offers older than [`OFFER_TIMEOUT`], recording each as a failed + /// upload so an operator sees why the peer never got its file. Returns + /// whether anything was dropped, which is the caller's cue to pump again. + pub fn expire_stale_offers(&mut self, now: Instant) -> bool { + let stale: Vec = self + .uploads + .iter() + .filter(|(_, job)| { + job.offered.is_some_and(|sent| { + now.duration_since(sent) >= OFFER_TIMEOUT + }) + }) + .map(|(&token, _)| token) + .collect(); + for token in &stale { + let Some(job) = self.uploads.remove(token) else { + continue; + }; + self.upload_events.push(UploadInfo { + username: job.downloader, + filename: job.virtual_path, + size: job.size, + bytes_sent: 0, + speed_bytes_per_sec: 0.0, + status: UploadStatus::Failed( + "no answer to the transfer offer".to_string(), + ), + }); + } + !stale.is_empty() + } + + /// The peer accepted the offer for `token`; it may still take a while to + /// resolve their address and connect, and that wait must not expire it. + pub fn mark_offer_answered(&mut self, token: u32) { + if let Some(job) = self.uploads.get_mut(&token) { + job.offered = None; + } + } + /// Forget everything `downloader` was waiting for or had been offered, and /// report whether that freed anything. /// @@ -347,6 +393,7 @@ mod tests { /// go untested. #[cfg(test)] mod context_tests { + use super::OFFER_TIMEOUT; use crate::client::{ClientContext, DEFAULT_UPLOAD_SLOTS}; use crate::types::UploadStatus; use std::path::PathBuf; @@ -464,6 +511,38 @@ mod context_tests { assert_eq!(ctx.place_in_queue("second", "f.mp3"), None); } + #[test] + fn an_offer_nobody_answers_frees_its_slot_after_the_timeout() { + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "gone", "f.mp3"); + ask(&mut ctx, "next", "f.mp3"); + assert_eq!(offered(&mut ctx, &mut token), ["gone"]); + + let now = std::time::Instant::now(); + assert!(!ctx.expire_stale_offers(now), "a fresh offer stays"); + assert!( + ctx.expire_stale_offers(now + OFFER_TIMEOUT), + "an unanswered offer expires" + ); + assert_eq!(offered(&mut ctx, &mut token), ["next"]); + assert!(ctx.take_upload_events().iter().any(|event| { + event.username == "gone" + && matches!(event.status, UploadStatus::Failed(_)) + })); + } + + #[test] + fn an_accepted_offer_waiting_on_an_address_does_not_expire() { + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "slow", "f.mp3"); + offered(&mut ctx, &mut token); + + ctx.mark_offer_answered(1); + let later = std::time::Instant::now() + OFFER_TIMEOUT * 2; + assert!(!ctx.expire_stale_offers(later), "an answered offer stays"); + assert!(ctx.uploads.contains_key(&1)); + } + #[test] fn completed_transfers_do_not_keep_occupying_slots() { // active_uploads keeps finished transfers so `uploads()` can report diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 15f4495..1dc23be 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -2312,7 +2312,12 @@ impl QueueingPeer { /// place in the queue" is also true of a peer that never asked, so polling /// for that would pass against a completely broken pump. fn takes_a_slot(&mut self) -> bool { - read_until_code(&mut self.peer, 40, Duration::from_secs(10)).is_some() + self.takes_a_slot_within(Duration::from_secs(10)) + } + + fn takes_a_slot_within(&mut self, timeout: Duration) -> bool { + let _ = self.peer.set_read_timeout(Some(timeout)); + read_until_code(&mut self.peer, 40, timeout).is_some() } } @@ -3478,3 +3483,52 @@ fn a_server_announcement_arrives_as_a_message_from_the_server() { .expect("bob should hear the announcement"); assert_eq!(announcement.message(), "going down at nine"); } + +// A peer that is offered a slot and never answers must not keep it: with +// two slots by default, two such peers shut uploads down for the session. +#[test] +fn an_unanswered_upload_offer_frees_its_slot() { + let server = server_or_skip!(); + let (share, folder) = queue_share("expiry", &["silent.mp3", "patient.mp3"]); + let server_addr = format!("{}:{}", server.host, server.port); + + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share.display().to_string()], + ..server.listening_settings("e2e_expiry_sharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + sharer.set_upload_slots(1); + let listen_addr = format!("127.0.0.1:{sharer_port}"); + + let mut silent = queue_as( + &server_addr, + &listen_addr, + "e2e_expiry_silent", + &format!("{folder}\\silent.mp3"), + ) + .expect("silent peer queues"); + assert!(silent.takes_a_slot(), "the silent peer is offered the slot"); + + let mut patient = queue_as( + &server_addr, + &listen_addr, + "e2e_expiry_patient", + &format!("{folder}\\patient.mp3"), + ) + .expect("patient peer queues"); + assert!( + patient.takes_a_slot_within(Duration::from_mins(1)), + "the slot must come free once the silent peer's offer expires" + ); + assert!( + sharer.take_upload_events().iter().any(|event| { + event.username == "e2e_expiry_silent" + && matches!(event.status, UploadStatus::Failed(_)) + }), + "the operator sees the unanswered offer fail" + ); + + let _ = std::fs::remove_dir_all(share); +}