diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 17330f4..d46dd38 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/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(); diff --git a/soulseek-rs-lib/src/message/peer/upload_failed.rs b/soulseek-rs-lib/src/message/peer/upload_failed.rs index 1be9a4d..5007a79 100644 --- a/soulseek-rs-lib/src/message/peer/upload_failed.rs +++ b/soulseek-rs-lib/src/message/peer/upload_failed.rs @@ -10,8 +10,31 @@ impl MessageHandler for UploadFailedHandler { fn get_code(&self) -> u8 { 46 } - fn handle(&self, message: &mut Message, _sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { 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:?}"), + } } } diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index e31aac5..64c7339 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -565,6 +565,29 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result { ) } +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!();