From db75a8266a95be183f99a9c7f6cafe499cad943f Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Sun, 26 Jul 2026 18:54:40 +0200 Subject: [PATCH] fix(serve): report a queued upload even when it is served between two polls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `serve` samples `uploads()` every 500ms and emits what changed, so an upload state shorter than that window never reached the log. Nothing surfaced this while every upload sat on its slot for a 500ms linger, which was exactly one poll; with the linger gone a peer can queue and be served between two samples, and "it had to wait" — the one thing the queue exists to record — silently disappeared. The pump now keeps whatever it left waiting, and `serve` drains that alongside the snapshot, so the fact survives regardless of when the next sample lands. The list is deduplicated per peer and file and capped, so a caller that never drains cannot grow it without bound. This is what made cli_e2e::serve_reports_a_queued_upload_once_the_slots_are_full fail on CI while passing locally: the assertion was real, and it was timing that decided whether the client could answer it. --- soulseek-rs-lib/src/client/mod.rs | 6 ++ soulseek-rs-lib/src/client/upload_queue.rs | 68 ++++++++++++++++++++++ soulseek-rs-lib/src/client/uploads.rs | 11 ++++ soulseek-rs/src/commands/peer.rs | 12 ++++ 4 files changed, 97 insertions(+) diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 19fd86d..ed317d1 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -344,6 +344,11 @@ pub struct ClientContext { upload_seq: u64, /// How many uploads may be in flight at once. upload_slots: usize, + /// Queued-upload states that came and went between two polls of + /// [`Client::uploads`]. A caller sampling that snapshot would otherwise + /// never see a peer that queued and was served inside one poll interval, + /// and "it waited" is exactly the fact a transfer log must not lose. + upload_events: Vec, actor_system: Arc, } impl Default for ClientContext { @@ -432,6 +437,7 @@ impl ClientContext { upload_queue: Vec::new(), upload_seq: 0, upload_slots: DEFAULT_UPLOAD_SLOTS, + upload_events: Vec::new(), downloads: DownloadStore::new(), actor_system, } diff --git a/soulseek-rs-lib/src/client/upload_queue.rs b/soulseek-rs-lib/src/client/upload_queue.rs index 70ef25e..70ddc9f 100644 --- a/soulseek-rs-lib/src/client/upload_queue.rs +++ b/soulseek-rs-lib/src/client/upload_queue.rs @@ -14,6 +14,10 @@ use super::{ClientContext, PeerRegistry, UploadJob}; use crate::types::{UploadInfo, UploadStatus}; use std::path::PathBuf; +/// Cap on remembered queued-upload states, so a caller that never drains +/// them cannot grow the list without bound. +const MAX_UPLOAD_EVENTS: usize = 256; + /// A queued upload that has just been given a slot, ready to be offered to the /// peer that asked for it. #[derive(Debug, Clone)] @@ -145,9 +149,34 @@ impl ClientContext { }, ); } + + // Whatever is still waiting had to wait. Record it, because a caller + // polling `uploads()` on an interval cannot see a peer that queued and + // was served between two polls — and with a small slot count that is + // the common case, not the rare one. + for waiting in self.queued_uploads() { + let already = self.upload_events.iter().any(|seen| { + seen.username == waiting.username + && seen.filename == waiting.filename + }); + if !already { + self.upload_events.push(waiting); + } + } + // A caller that never drains must not grow this without bound. + if self.upload_events.len() > MAX_UPLOAD_EVENTS { + let excess = self.upload_events.len() - MAX_UPLOAD_EVENTS; + self.upload_events.drain(..excess); + } + (self.peer_registry.clone(), offers) } + /// Take the queued-upload states recorded since the last call. + pub fn take_upload_events(&mut self) -> Vec { + std::mem::take(&mut self.upload_events) + } + /// Uploads occupying a slot: offered but not yet accepted, plus those /// actually streaming. /// @@ -492,4 +521,43 @@ mod context_tests { ] ); } + + // A peer that waits and is then served inside one poll interval must still + // be reportable as queued: `serve` samples on an interval, so a state that + // only ever existed between two samples would otherwise vanish from the + // transfer log entirely. + #[test] + fn a_queued_upload_is_recorded_even_after_it_leaves_the_queue() { + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "amy", "@@share\\amy.mp3"); + ask(&mut ctx, "bob", "@@share\\bob.mp3"); + + // One slot, so bob waits behind amy. + let _ = ctx.pump_uploads(&mut token); + assert!( + ctx.queued_uploads().iter().any(|u| u.username == "bob"), + "bob should be waiting while amy holds the only slot" + ); + + // Amy finishes and bob is served, so nothing is waiting any more... + ctx.uploads.clear(); + ctx.active_uploads.clear(); + let _ = ctx.pump_uploads(&mut token); + assert!( + ctx.queued_uploads().is_empty(), + "the queue should have drained" + ); + + // ...but the fact that bob waited survives for whoever asks next. + let events = ctx.take_upload_events(); + assert!( + events.iter().any(|u| u.username == "bob" + && matches!(u.status, UploadStatus::Queued(_))), + "bob's wait should still be reportable, got {events:?}" + ); + assert!( + ctx.take_upload_events().is_empty(), + "draining twice must not repeat the same events" + ); + } } diff --git a/soulseek-rs-lib/src/client/uploads.rs b/soulseek-rs-lib/src/client/uploads.rs index b0c5ede..d3b7ba8 100644 --- a/soulseek-rs-lib/src/client/uploads.rs +++ b/soulseek-rs-lib/src/client/uploads.rs @@ -49,6 +49,17 @@ impl Client { /// How many uploads run at once. Lowering it does not interrupt transfers /// already in flight; the queue simply refills more slowly. + /// Queued-upload states recorded since the last call, for a caller that + /// samples [`Client::uploads`] and would otherwise miss a peer that queued + /// and was served between two samples. + #[must_use] + pub fn take_upload_events(&self) -> Vec { + self.context + .write_safe() + .map(|mut ctx| ctx.take_upload_events()) + .unwrap_or_default() + } + pub fn set_upload_slots(&self, slots: usize) { if let Ok(mut ctx) = self.context.write_safe() { ctx.upload_slots = slots.max(1); diff --git a/soulseek-rs/src/commands/peer.rs b/soulseek-rs/src/commands/peer.rs index 3d7432b..f9b19ad 100644 --- a/soulseek-rs/src/commands/peer.rs +++ b/soulseek-rs/src/commands/peer.rs @@ -138,6 +138,18 @@ pub fn serve( let deadline = span.map(|span| Instant::now() + span); let mut reported: HashMap<(String, String), UploadRecord> = HashMap::new(); while deadline.is_none_or(|deadline| Instant::now() < deadline) { + // Queued states first, and from the client's own record rather than + // this snapshot: with few slots and small files a peer can queue and be + // served well inside one poll, and "it had to wait" is precisely what a + // transfer log exists to report. + for queued in session.client.take_upload_events() { + let record = upload_record(&queued); + let key = (record.user.clone(), record.path.clone()); + if reported.get(&key) != Some(&record) { + ctx.out.emit(&record); + reported.insert(key, record); + } + } for upload in session.client.uploads() { let record = upload_record(&upload); let key = (record.user.clone(), record.path.clone());