fix(serve): report a queued upload even when it is served between two polls

`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.
This commit is contained in:
Michel de Graaf
2026-07-26 18:54:40 +02:00
parent 7435d48b32
commit db75a8266a
4 changed files with 97 additions and 0 deletions
+6
View File
@@ -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<crate::types::UploadInfo>,
actor_system: Arc<ActorSystem>,
}
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,
}
@@ -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<UploadInfo> {
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"
);
}
}
+11
View File
@@ -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<crate::types::UploadInfo> {
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);
+12
View File
@@ -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());