Merge pull request #79 from michel/lib/offer-expiry

fix(lib): let an unanswered upload offer expire

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<Instant>,
 }
 
 /// Live bookkeeping for an upload being served (or recently finished).

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<RwLock<ClientContext>>) {
+        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<RwLock<ClientContext>>) {
         let expired = match client_context.write_safe() {
             Ok(mut ctx) => ctx.take_expired_connects(Instant::now()),

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<u32> = 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

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);
+}