Merge pull request #35 from michel/issue-30-upload-failed-forward

fix(peer): forward UploadFailed to the download store

soulseek-rs-lib/src/actor/peer_actor.rs

@@ -282,8 +282,8 @@ impl PeerActor {
             PeerMessage::ProcessRead => {
                 self.process_read();
             }
-            PeerMessage::UploadFailed(username, filename) => {
-                self.handle_upload_failed(username, filename);
+            PeerMessage::UploadFailed(_, filename) => {
+                self.handle_upload_failed(filename);
             }
         }
     }
@@ -452,7 +452,8 @@ impl PeerActor {
         }
     }
 
-    fn handle_upload_failed(&self, username: String, filename: String) {
+    fn handle_upload_failed(&self, filename: String) {
+        let username = self.peer_username();
         if let Err(e) = self
             .client_channel
             .send(ClientOperation::UploadFailed(username, filename))
@@ -973,6 +974,24 @@ mod tests {
         }
     }
 
+    #[test]
+    fn an_upload_failed_message_reaches_the_client_as_this_peer() {
+        let (mut actor, rx, _far_end) = connected_actor();
+
+        actor.handle_message(PeerMessage::UploadFailed(
+            String::new(),
+            "song.mp3".to_string(),
+        ));
+
+        match rx.try_recv() {
+            Ok(ClientOperation::UploadFailed(username, filename)) => {
+                assert_eq!(username, "bob");
+                assert_eq!(filename, "song.mp3");
+            }
+            other => panic!("expected UploadFailed, got {other:?}"),
+        }
+    }
+
     #[test]
     fn tick_reaps_a_peer_idle_past_the_deadline() {
         let (mut actor, rx, _far_end) = connected_actor();

soulseek-rs-lib/src/message/peer/upload_failed.rs

@@ -10,8 +10,31 @@ impl MessageHandler<PeerMessage> for UploadFailedHandler {
     fn get_code(&self) -> u8 {
         46
     }
-    fn handle(&self, message: &mut Message, _sender: Sender<PeerMessage>) {
+    fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
         let filename = message.read_string();
         info!("Upload failed for {}", filename);
+        let _ = sender.send(PeerMessage::UploadFailed(String::new(), filename));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::message::framed;
+
+    #[test]
+    fn the_failed_filename_is_forwarded() {
+        let (tx, rx) = std::sync::mpsc::channel();
+        let mut message = framed(|m| {
+            m.write_string("@@share\\gone.mp3");
+        });
+
+        UploadFailedHandler.handle(&mut message, tx);
+        match rx.try_recv() {
+            Ok(PeerMessage::UploadFailed(_, filename)) => {
+                assert_eq!(filename, "@@share\\gone.mp3");
+            }
+            other => panic!("unexpected: {other:?}"),
+        }
     }
 }

soulseek-rs-lib/tests/e2e.rs

@@ -565,6 +565,29 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<u64> {
     )
 }
 
+fn run_refusing_uploader(
+    cfg: &MockUpload,
+    refusal: &Message,
+) -> std::io::Result<()> {
+    let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?;
+    p.set_read_timeout(Some(Duration::from_secs(10)))?;
+    p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?;
+    p.flush()?;
+    let _ = cfg.ready.send(());
+
+    loop {
+        let msg = read_framed(&mut p)?;
+        if msg.get_message_code() == 43 {
+            break;
+        }
+    }
+
+    p.write_all(&refusal.get_buffer())?;
+    p.flush()?;
+    std::thread::sleep(Duration::from_millis(500));
+    Ok(())
+}
+
 /// Open an F (file transfer) connection to `downloader_addr` and stream
 /// `content`. PeerInit(F) and the 4-byte transfer token are written together so
 /// the token lands in the listener's read buffer, where the download is looked
@@ -760,6 +783,82 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
     let _ = std::fs::remove_dir_all(&download_dir);
 }
 
+fn refusal_fails_the_download_quickly(
+    peer_username: &str,
+    build_refusal: impl FnOnce(&str) -> Message,
+) {
+    let server = server_or_skip!();
+
+    let listen_port = free_port().expect("free listen port");
+    let mut client = Client::with_settings(server.listening_settings(
+        &format!("{peer_username}_dl"),
+        "pw",
+        listen_port,
+    ));
+    client.connect().expect("connect");
+    assert!(client.login().expect("login"));
+
+    let filename = "gone_song.mp3";
+    let refusal = build_refusal(filename);
+    let (ready_tx, ready_rx) = std::sync::mpsc::channel();
+    let cfg = MockUpload {
+        listen_addr: format!("127.0.0.1:{listen_port}"),
+        peer_username: peer_username.to_string(),
+        filename: filename.to_string(),
+        content: Vec::new(),
+        token: 0,
+        ready: ready_tx,
+    };
+    let uploader = std::thread::spawn(move || {
+        if let Err(e) = run_refusing_uploader(&cfg, &refusal) {
+            eprintln!("[mock refusing uploader] {e}");
+        }
+    });
+
+    ready_rx
+        .recv_timeout(Duration::from_secs(5))
+        .expect("mock uploader P connection");
+    std::thread::sleep(Duration::from_millis(1500));
+
+    let download_dir = unique_download_dir();
+    let (_download, status_rx) = client
+        .download(
+            filename.to_string(),
+            peer_username.to_string(),
+            0,
+            download_dir.display().to_string(),
+        )
+        .expect("start download");
+
+    let deadline = Instant::now() + Duration::from_secs(10);
+    let mut failed = false;
+    while Instant::now() < deadline {
+        match status_rx.recv_timeout(Duration::from_millis(500)) {
+            Ok(DownloadStatus::Failed(_)) => {
+                failed = true;
+                break;
+            }
+            Ok(DownloadStatus::Completed) => {
+                panic!("the refused download must not complete")
+            }
+            _ => {}
+        }
+    }
+    let _ = uploader.join();
+    let _ = std::fs::remove_dir_all(&download_dir);
+
+    assert!(failed, "the peer's refusal should fail the download fast");
+}
+
+#[test]
+fn an_upload_failed_reply_fails_the_download_quickly() {
+    refusal_fails_the_download_quickly("e2e_failing_peer", |filename| {
+        let mut refusal = Message::new();
+        refusal.write_int32(46).write_string(filename);
+        refusal
+    });
+}
+
 #[test]
 fn an_interrupted_download_resumes_from_its_partial_file() {
     let server = server_or_skip!();