From 352816cc1e5f09a85387342e70c4b078efca303e Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 7 Sep 2026 15:11:21 +0200 Subject: [PATCH] fix(lib): advertise real slot, speed and queue figures in search replies Every reply to a peer's search claimed a free upload slot, no speed and an empty queue, whatever the sharer was doing, and searchers rank sources on exactly those fields. A reply now says whether a slot is free, the bytes per second of the last completed upload, and how many uploads are waiting. --- soulseek-rs-lib/examples/stress.rs | 2 +- soulseek-rs-lib/src/client/mod.rs | 12 ++++- soulseek-rs-lib/src/client/operations.rs | 3 ++ soulseek-rs-lib/src/client/tests.rs | 10 ++-- soulseek-rs-lib/src/client/upload_queue.rs | 6 ++- soulseek-rs-lib/src/client/uploads.rs | 9 +++- .../src/message/peer/file_search_response.rs | 23 +++++++- soulseek-rs-lib/src/peer/upload_peer.rs | 9 ++-- soulseek-rs-lib/tests/e2e.rs | 53 +++++++++++++++++++ 9 files changed, 114 insertions(+), 13 deletions(-) diff --git a/soulseek-rs-lib/examples/stress.rs b/soulseek-rs-lib/examples/stress.rs index ca13759..b73aaf2 100644 --- a/soulseek-rs-lib/examples/stress.rs +++ b/soulseek-rs-lib/examples/stress.rs @@ -415,7 +415,7 @@ impl SeederState { }) .collect(); let response = - build_file_search_response(&self.name, token, &entries, 1, 0); + build_file_search_response(&self.name, token, &entries, 1, 0, 0); let Ok(mut guard) = self.conn.lock() else { return; diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 1abf26e..ab2baa7 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -150,6 +150,9 @@ fn build_search_response( own_username: &str, token: u32, query: &str, + free_slot: bool, + speed: u32, + queue_length: u32, ) -> Option { let matches = shares.search(query); if matches.is_empty() { @@ -168,8 +171,9 @@ fn build_search_response( own_username, token, &entries, - 1, - 0, + u8::from(free_slot), + speed, + queue_length, )) } @@ -376,6 +380,9 @@ pub struct ClientContext { upload_seq: u64, /// How many uploads may be in flight at once. upload_slots: usize, + /// Bytes per second of the last completed upload, advertised in search + /// replies; zero until one has finished. + last_upload_speed: u32, /// 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, @@ -458,6 +465,7 @@ impl ClientContext { upload_queue: Vec::new(), upload_seq: 0, upload_slots: DEFAULT_UPLOAD_SLOTS, + last_upload_speed: 0, upload_events: Vec::new(), downloads: DownloadStore::new(), actor_system, diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index bcd2cc7..467ae19 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -593,6 +593,9 @@ impl Client { &own_username, token, &query, + ctx.has_free_upload_slot(), + ctx.last_upload_speed, + ctx.upload_queue.len() as u32, ), Err(e) => { error!("[client] IncomingSearch read: {}", e); diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 73e892a..031edb2 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -160,8 +160,9 @@ fn build_search_response_matches_shares_and_echoes_token() { std::fs::write(dir.join("probe_xyzzy.bin"), b"data").unwrap(); let shares = Shares::scan(&dir).unwrap(); - let response = build_search_response(&shares, "me", 99, "xyzzy") - .expect("a matching share yields a response"); + let response = + build_search_response(&shares, "me", 99, "xyzzy", true, 0, 0) + .expect("a matching share yields a response"); let mut decoded = crate::message::Message::new_with_data(response.get_buffer()); decoded.set_pointer(8); @@ -170,7 +171,10 @@ fn build_search_response_matches_shares_and_echoes_token() { assert_eq!(result.token, 99); assert!(result.files.iter().any(|f| f.name.contains("probe_xyzzy"))); - assert!(build_search_response(&shares, "me", 1, "nomatch").is_none()); + assert!( + build_search_response(&shares, "me", 1, "nomatch", true, 0, 0) + .is_none() + ); let _ = std::fs::remove_dir_all(dir); } diff --git a/soulseek-rs-lib/src/client/upload_queue.rs b/soulseek-rs-lib/src/client/upload_queue.rs index 9bc03b9..474c860 100644 --- a/soulseek-rs-lib/src/client/upload_queue.rs +++ b/soulseek-rs-lib/src/client/upload_queue.rs @@ -123,7 +123,7 @@ impl ClientContext { mut next_token: impl FnMut() -> u32, ) -> (Option, Vec) { let mut offers = Vec::new(); - while self.uploads_in_flight() < self.upload_slots { + while self.has_free_upload_slot() { let Some(index) = next_to_serve(&self.upload_queue) else { break; }; @@ -190,6 +190,10 @@ impl ClientContext { .count() } + pub(super) fn has_free_upload_slot(&self) -> bool { + self.uploads_in_flight() < self.upload_slots + } + /// The peers still waiting, in the order they will be served, each carrying /// its own 1-based place. pub(super) fn queued_uploads(&self) -> Vec { diff --git a/soulseek-rs-lib/src/client/uploads.rs b/soulseek-rs-lib/src/client/uploads.rs index cd42654..1907691 100644 --- a/soulseek-rs-lib/src/client/uploads.rs +++ b/soulseek-rs-lib/src/client/uploads.rs @@ -145,8 +145,9 @@ impl Client { &bytes_sent, &cancel, ); + let streamed = result.as_ref().ok().copied(); let status = match &result { - Ok(()) => UploadStatus::Completed, + Ok(_) => UploadStatus::Completed, Err(e) if e.kind() == std::io::ErrorKind::Interrupted => { UploadStatus::Cancelled } @@ -158,7 +159,13 @@ impl Client { if let Ok(mut ctx) = context.write_safe() && let Some(upload) = ctx.active_uploads.get_mut(&token) { + let secs = upload.started.elapsed().as_secs_f64(); upload.status = status; + if let Some(bytes) = streamed + && secs > 0.0 + { + ctx.last_upload_speed = (bytes as f64 / secs) as u32; + } } // The slot this transfer held is free now, so whoever is next in // line gets it without waiting for another request to arrive. diff --git a/soulseek-rs-lib/src/message/peer/file_search_response.rs b/soulseek-rs-lib/src/message/peer/file_search_response.rs index 4923c00..90bb704 100644 --- a/soulseek-rs-lib/src/message/peer/file_search_response.rs +++ b/soulseek-rs-lib/src/message/peer/file_search_response.rs @@ -22,6 +22,7 @@ pub fn build_file_search_response( files: &[FileEntry], slots: u8, speed: u32, + queue_length: u32, ) -> Message { let mut payload = Message::new(); payload @@ -39,7 +40,10 @@ pub fn build_file_search_response( payload.write_int32(code).write_int32(value); } } - payload.write_int8(slots).write_int32(speed).write_int32(0); // free upload slots / queue length (well-formed trailer) + payload + .write_int8(slots) + .write_int32(speed) + .write_int32(queue_length); let compressed = compress_stored(&payload.get_data()); Message::new() @@ -112,7 +116,7 @@ fn build_file_search_response_roundtrips_through_the_decoder() { attribs: &[], }, ]; - let message = build_file_search_response("e2e_sharer", 42, &files, 1, 0); + let message = build_file_search_response("e2e_sharer", 42, &files, 1, 0, 0); // Decode via the exact production decoder used for real peer responses: // the dispatcher positions the pointer at 8 (past length + code). @@ -132,3 +136,18 @@ fn build_file_search_response_roundtrips_through_the_decoder() { assert!(result.files[1].attribs.is_empty()); assert_eq!(result.slots, 1); } + +#[test] +fn the_trailer_carries_slots_speed_and_queue_length() { + let message = build_file_search_response("me", 1, &[], 0, 900, 7); + let mut body = Message::new_with_data( + crate::utils::zlib::deflate(&message.get_data()[4..]).unwrap(), + ); + body.read_string(); + body.read_int32(); + assert_eq!(body.read_int32(), 0, "no files"); + assert_eq!( + (body.read_int8(), body.read_int32(), body.read_int32()), + (0, 900, 7) + ); +} diff --git a/soulseek-rs-lib/src/peer/upload_peer.rs b/soulseek-rs-lib/src/peer/upload_peer.rs index b102ba3..57138d4 100644 --- a/soulseek-rs-lib/src/peer/upload_peer.rs +++ b/soulseek-rs-lib/src/peer/upload_peer.rs @@ -19,7 +19,8 @@ use crate::trace; /// START_DOWNLOAD offset before we stream the file. /// /// `bytes_sent` is updated as the transfer progresses, and setting `cancel` -/// aborts the stream with an [`io::ErrorKind::Interrupted`] error. +/// aborts the stream with an [`io::ErrorKind::Interrupted`] error. Returns the +/// bytes streamed by this call, which a resumed transfer's prefix is not. /// /// # Errors /// Returns any I/O error opening the file or talking to the peer. @@ -31,7 +32,7 @@ pub fn serve_file( path: &Path, bytes_sent: &AtomicU64, cancel: &AtomicBool, -) -> io::Result<()> { +) -> io::Result { let mut file = File::open(path)?; let socket = format!("{host}:{port}") @@ -78,6 +79,7 @@ pub fn serve_file( } let mut buffer = vec![0u8; 64 * 1024]; + let mut streamed = 0u64; loop { if cancel.load(Ordering::Relaxed) { return Err(io::Error::new( @@ -91,6 +93,7 @@ pub fn serve_file( } stream.write_all(&buffer[..read])?; bytes_sent.fetch_add(read as u64, Ordering::Relaxed); + streamed += read as u64; } stream.flush()?; @@ -100,7 +103,7 @@ pub fn serve_file( // a slot cap is time no other peer can use. stream.shutdown(std::net::Shutdown::Write).ok(); trace!("[upload] served {} to {}:{}", path.display(), host, port); - Ok(()) + Ok(streamed) } #[cfg(test)] diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 821db49..8760cfd 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -1761,6 +1761,15 @@ fn two_real_clients_search_and_download() { assert_eq!(upload.username, "e2e_leecher"); assert_eq!(upload.bytes_sent, size); + // The next reply advertises what that upload measured. + let _ = leecher.search(query, Duration::from_secs(1)); + let reply = reply_from(&leecher, query, "e2e_sharer") + .expect("the sharer answers again"); + assert!( + reply.speed > 0, + "a finished upload sets the advertised speed" + ); + let _ = std::fs::remove_dir_all(share_dir); let _ = std::fs::remove_dir_all(download_dir); } @@ -3127,3 +3136,47 @@ fn two_live_searches_never_share_a_token() { let _ = std::fs::remove_dir_all(share_dir); } + +// Searchers pick sources by the free-slot flag in a reply, and every reply +// said "free" whatever the queue looked like. +#[test] +fn a_search_reply_reports_no_free_slot_while_the_only_slot_is_taken() { + let server = server_or_skip!(); + let (share, folder) = + queue_share("honest", &["blocker.mp3", "probe_honest.bin"]); + let (sharer, searcher) = sharer_and_searcher( + &server, + &share, + "e2e_honest_sharer", + "e2e_honest_searcher", + ); + sharer.set_upload_slots(1); + + let query = "probe_honest"; + let _ = searcher.search(query, Duration::from_secs(1)); + let reply = reply_from(&searcher, query, "e2e_honest_sharer") + .expect("the sharer answers"); + assert_eq!(reply.slots, 1, "nothing is uploading yet"); + + let server_addr = format!("{}:{}", server.host, server.port); + let listen_addr = + format!("127.0.0.1:{}", sharer.listen_port().expect("listening")); + let mut blocker = queue_as( + &server_addr, + &listen_addr, + "e2e_honest_blocker", + &format!("{folder}\\blocker.mp3"), + ) + .expect("blocker queues"); + assert!( + blocker.takes_a_slot(), + "the blocker is offered the only slot" + ); + + let _ = searcher.search(query, Duration::from_secs(1)); + let reply = reply_from(&searcher, query, "e2e_honest_sharer") + .expect("the sharer answers again"); + assert_eq!(reply.slots, 0, "the only slot is taken"); + + let _ = std::fs::remove_dir_all(share); +}