diff --git a/README.md b/README.md index 5460ede..d84edd4 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,8 @@ does, how to install it, and every `config.toml` setting. - **Sharing**: `shares add` a directory and your files show up in searches; `serve` stays online so peers can browse and download them - **Browse**: list any user's shared files and download straight from the tree +- **Resumable downloads**: a transfer that dies leaves a `.part` file, and + re-running the download asks the peer to send only what is missing - **Chat rooms**: list, join, and talk in public rooms, several open at once - **Private messages**: send and receive messages, with an inbox in the TUI - **Firewalled peers**: downloads and browsing fall back to server-brokered @@ -327,6 +329,12 @@ there yet. Downloads run under a deadline (`--timeout`, 300s by default) and `--max-concurrent-downloads` at a time, so an unattended run always ends. +An interrupted download is picked up where it stopped. Bytes stream into +`.part` as they arrive and it is renamed only once the file is whole, so +re-running the same `download` asks the peer to start at the offset already on +disk. Partial files are never offered to other peers, and deleting the `.part` +starts the transfer over. + ### Configuration Every setting can come from a flag, an environment variable, or the config diff --git a/soulseek-rs-lib/src/peer/download_peer.rs b/soulseek-rs-lib/src/peer/download_peer.rs index 7212c60..be4805b 100644 --- a/soulseek-rs-lib/src/peer/download_peer.rs +++ b/soulseek-rs-lib/src/peer/download_peer.rs @@ -1,8 +1,8 @@ -use std::fs; +use std::fs::{self, File, OpenOptions}; use std::io::{self, Read, Write}; use std::net::TcpStream; use std::net::ToSocketAddrs; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::sync::{Arc, RwLock}; use std::thread; use std::time::{Duration, Instant}; @@ -11,10 +11,8 @@ use crate::client::ClientContext; use crate::message::server::MessageFactory; use crate::trace; use crate::types::{Download, DownloadStatus}; -use crate::utils::path::expand_tilde; +use crate::utils::path::{PART_SUFFIX, expand_tilde}; -const START_DOWNLOAD: [u8; 8] = - [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]; const READ_BUFFER_SIZE: usize = 8192; const PROGRESS_UPDATE_CHUNKS: usize = 15; // ~120KB (15 * 8192 bytes) @@ -91,34 +89,114 @@ impl FileManager { let filename_only = Self::extract_filename_from_path(filename); output_directory.join(filename_only) } -} -struct StreamProcessor { - total_bytes: usize, - received: bool, - buffer: Vec, -} + /// Where a download's bytes end up: its configured directory (falling back + /// to the parent if it names a file) joined with the remote basename. + fn resolve_download_path( + download: &Download, + ) -> Result { + let mut expanded_path = expand_tilde(&download.download_directory); -impl StreamProcessor { - const fn new() -> Self { - Self { - total_bytes: 0, - received: false, - buffer: Vec::new(), + if !expanded_path.is_dir() { + expanded_path = expanded_path + .parent() + .ok_or_else(|| { + DownloadError::PathResolutionError(format!( + "Cannot resolve parent directory for: {}", + expanded_path.display() + )) + })? + .to_path_buf(); } - } - fn process_data_chunk(&mut self, data: &[u8]) { - self.buffer.extend_from_slice(data); - self.total_bytes += data.len(); - } + let final_path = Self::create_download_path_from_filename( + expanded_path, + &download.filename, + ); - const fn should_continue(&self, expected_size: Option) -> bool { - if let Some(size) = expected_size { - self.total_bytes < size + final_path + .to_str() + .ok_or_else(|| { + DownloadError::PathResolutionError(format!( + "Path contains invalid UTF-8: {}", + final_path.display() + )) + }) + .map(String::from) + } +} + +/// The `.part` file a transfer streams into, so an interrupted transfer leaves +/// behind what it managed to fetch and the next attempt can pick up from there. +struct PartFile { + file: File, + final_path: String, + written: u64, + expected: u64, +} + +impl PartFile { + fn open(download: &Download) -> Result { + let final_path = FileManager::resolve_download_path(download)?; + let path = PathBuf::from(format!("{final_path}{PART_SUFFIX}")); + + if let Some(parent) = path.parent() { + fs::create_dir_all(parent) + .map_err(DownloadError::FileWriteError)?; + } + + // A `.part` that already meets or exceeds the expected size is left over + // from a different (or corrupted) transfer, not a resume point. + let on_disk = fs::metadata(&path).map_or(0, |m| m.len()); + let written = if on_disk < download.size { on_disk } else { 0 }; + + let file = if written > 0 { + OpenOptions::new().append(true).open(&path) } else { - true + File::create(&path) } + .map_err(DownloadError::FileWriteError)?; + + Ok(Self { + file, + final_path, + written, + expected: download.size, + }) + } + + /// Peers can coalesce trailing bytes into the last chunk, so overshoot past + /// the expected size is dropped rather than written. + fn write(&mut self, data: &[u8]) -> Result<(), DownloadError> { + let remaining = (self.expected - self.written) as usize; + let data = &data[..data.len().min(remaining)]; + self.file + .write_all(data) + .map_err(DownloadError::FileWriteError)?; + self.written += data.len() as u64; + Ok(()) + } + + const fn is_complete(&self) -> bool { + self.written >= self.expected + } + + /// A short transfer is a failure, and the `.part` stays put so the next + /// attempt can resume it. + fn finish(self) -> Result { + if !self.is_complete() { + return Err(DownloadError::IncompleteDownload { + received: self.written as usize, + expected: self.expected as usize, + }); + } + drop(self.file); + fs::rename( + format!("{}{PART_SUFFIX}", self.final_path), + &self.final_path, + ) + .map_err(DownloadError::FileWriteError)?; + Ok(self.final_path) } } @@ -209,12 +287,27 @@ impl DownloadPeer { Ok(()) } - fn handle_pierce_firewall_response( + /// Soulseek's START_DOWNLOAD is an 8-byte little-endian offset, so anything + /// already in the `.part` file is skipped by asking the peer to start past it. + fn start_transfer( + download: Download, + stream: &mut TcpStream, + client_context: &Arc>, + ) -> Result<(Download, PartFile), DownloadError> { + let part = PartFile::open(&download)?; + stream + .write_all(&part.written.to_le_bytes()) + .map_err(DownloadError::StreamWriteError)?; + Self::report_progress(client_context, &download, part.written, 0.0); + Ok((download, part)) + } + + fn begin_pierced_download( &self, data: &[u8], stream: &mut TcpStream, client_context: &Arc>, - ) -> Result { + ) -> Result<(Download, PartFile), DownloadError> { let token_bytes = data.get(0..4).ok_or(DownloadError::InvalidTokenBytes)?; let token_array: [u8; 4] = token_bytes @@ -227,10 +320,6 @@ impl DownloadPeer { self.username, token_u32 ); - stream - .write_all(&START_DOWNLOAD) - .map_err(DownloadError::StreamWriteError)?; - let client_guard = client_context .read() .map_err(|_| DownloadError::LockPoisoned)?; @@ -238,92 +327,67 @@ impl DownloadPeer { client_guard.get_download_by_token(token_u32).cloned(); drop(client_guard); - download_info.ok_or(DownloadError::TokenNotFound(token_u32)) + let download = + download_info.ok_or(DownloadError::TokenNotFound(token_u32))?; + + Self::start_transfer(download, stream, client_context) } fn read_download_stream( &self, stream: &mut TcpStream, client_context: &Arc>, - mut download: Option, - ) -> Result<(Vec, Download), DownloadError> { - let mut processor = StreamProcessor::new(); - let mut read_buffer = [1u8; READ_BUFFER_SIZE]; - let mut chunk_counter = 0; - let start_time = Instant::now(); - let mut last_update_time = start_time; + download: Option, + ) -> Result<(Download, String), DownloadError> { + let mut read_buffer = [0u8; READ_BUFFER_SIZE]; + let mut chunk_counter = 0usize; + let mut last_update_time = Instant::now(); trace!( "[download_peer:{}] Starting to read data from peer", self.username ); - if download.is_some() { - stream - .write_all(&START_DOWNLOAD) - .map_err(DownloadError::StreamWriteError)?; - if let Some(ref dl) = download { - Self::send_download_status( - client_context, - dl, - DownloadStatus::InProgress { - bytes_downloaded: 0, - total_bytes: dl.size, - speed_bytes_per_sec: 0.0, - }, - ); - } - } + let mut transfer = match download { + Some(dl) => Some(Self::start_transfer(dl, stream, client_context)?), + None => None, + }; loop { - if let Some(ref dl) = download { + if let Some((ref dl, _)) = transfer { Self::wait_while_paused(client_context, dl)?; } match stream.read(&mut read_buffer) { Ok(0) => { trace!( - "[download_peer:{}] connection closed by peer. bytes read: {}", - self.username, processor.total_bytes + "[download_peer:{}] connection closed by peer", + self.username ); break; } Ok(bytes_read) => { let data = &read_buffer[..bytes_read]; - if !self.no_pierce && !processor.received { - let new_download = self - .handle_pierce_firewall_response( - data, - stream, - client_context, - )?; - trace!( - "[download_peer:{}] got download info for token: {} - filename: {}", - self.username, self.token, new_download.filename - ); - download = Some(new_download); - processor.received = true; - if let Some(ref dl) = download { - Self::send_download_status( - client_context, - dl, - DownloadStatus::InProgress { - bytes_downloaded: 0, - total_bytes: dl.size, - speed_bytes_per_sec: 0.0, - }, - ); - } + if transfer.is_none() && !self.no_pierce { + transfer = Some(self.begin_pierced_download( + data, + stream, + client_context, + )?); continue; } - processor.process_data_chunk(data); + let Some((dl, part)) = transfer.as_mut() else { + return Err(DownloadError::DownloadInfoMissing( + self.token, + )); + }; + + part.write(data)?; chunk_counter += 1; - if let Some(ref dl) = download - && chunk_counter % PROGRESS_UPDATE_CHUNKS == 0 - { + if chunk_counter.is_multiple_of(PROGRESS_UPDATE_CHUNKS) { let elapsed = last_update_time.elapsed().as_secs_f64(); let bytes_since_last_update = PROGRESS_UPDATE_CHUNKS * READ_BUFFER_SIZE; @@ -332,23 +396,16 @@ impl DownloadPeer { } else { 0.0 }; - - let status = DownloadStatus::InProgress { - bytes_downloaded: processor.total_bytes as u64, - total_bytes: dl.size, - speed_bytes_per_sec: speed, - }; - Self::send_download_status(client_context, dl, status); - + Self::report_progress( + client_context, + dl, + part.written, + speed, + ); last_update_time = Instant::now(); } - let expected_size = download - .as_ref() - .ok_or(DownloadError::DownloadInfoMissing(self.token))? - .size as usize; - - if !processor.should_continue(Some(expected_size)) { + if part.is_complete() { break; } } @@ -363,15 +420,28 @@ impl DownloadPeer { self.username ); - let download = - download.ok_or(DownloadError::DownloadInfoMissing(self.token))?; + let Some((download, part)) = transfer else { + return Err(DownloadError::DownloadInfoMissing(self.token)); + }; - let buffer = Self::finalize_download_buffer( - processor.buffer, - download.size as usize, - )?; + Ok((download, part.finish()?)) + } - Ok((buffer, download)) + fn report_progress( + client_context: &Arc>, + download: &Download, + bytes_downloaded: u64, + speed_bytes_per_sec: f64, + ) { + Self::send_download_status( + client_context, + download, + DownloadStatus::InProgress { + bytes_downloaded, + total_bytes: download.size, + speed_bytes_per_sec, + }, + ); } fn send_download_status( @@ -405,73 +475,6 @@ impl DownloadPeer { } } - fn resolve_download_path( - download: &Download, - ) -> Result { - let download_directory = &download.download_directory; - let mut expanded_path = expand_tilde(download_directory); - - if !expanded_path.is_dir() { - expanded_path = expanded_path - .parent() - .ok_or_else(|| { - DownloadError::PathResolutionError(format!( - "Cannot resolve parent directory for: {}", - expanded_path.display() - )) - })? - .to_path_buf(); - } - - let final_path = FileManager::create_download_path_from_filename( - expanded_path, - &download.filename, - ); - - final_path - .to_str() - .ok_or_else(|| { - DownloadError::PathResolutionError(format!( - "Path contains invalid UTF-8: {}", - final_path.display() - )) - }) - .map(String::from) - } - - /// Validate and normalize a fully-read download buffer against the size the - /// peer promised in the transfer request. A peer that closes the connection - /// early yields fewer bytes than expected — that must be reported as a - /// failure, not silently saved as a truncated "completed" file. A peer that - /// sends trailing bytes past the expected size has them trimmed off. - fn finalize_download_buffer( - mut buffer: Vec, - expected_size: usize, - ) -> Result, DownloadError> { - if buffer.len() < expected_size { - return Err(DownloadError::IncompleteDownload { - received: buffer.len(), - expected: expected_size, - }); - } - buffer.truncate(expected_size); - Ok(buffer) - } - - fn save_downloaded_file( - path: &str, - data: &[u8], - ) -> Result<(), DownloadError> { - if let Some(parent) = Path::new(path).parent() { - fs::create_dir_all(parent) - .map_err(DownloadError::FileWriteError)?; - } - - fs::write(path, data).map_err(DownloadError::FileWriteError)?; - - Ok(()) - } - pub fn download_file( self, client_context: Arc>, @@ -504,17 +507,12 @@ impl DownloadPeer { self.perform_handshake(&mut stream)?; trace!("[download_peer:{}] handshake completed", self.username); - let (buffer, download) = + let (download, final_path) = self.read_download_stream(&mut stream, &client_context, download)?; - let final_path = Self::resolve_download_path(&download)?; - Self::save_downloaded_file(&final_path, &buffer)?; - trace!( "[download_peer:{}] download completed successfully: {} bytes, saved to: {}", - self.username, - buffer.len(), - final_path + self.username, download.size, final_path ); Ok((download, final_path)) @@ -523,36 +521,103 @@ impl DownloadPeer { #[cfg(test)] mod tests { - use super::{DownloadError, DownloadPeer, FileManager}; + use super::{DownloadError, DownloadPeer, FileManager, PartFile}; + use crate::types::{Download, DownloadMetadata, DownloadStatus}; + use std::sync::mpsc; + + fn scratch_dir(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir() + .join(format!("soulseek-part-{}-{name}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + fn download_into(dir: &std::path::Path, size: u64) -> Download { + Download { + username: "peer".to_string(), + filename: "song.mp3".to_string(), + token: 1, + size, + download_directory: dir.display().to_string(), + status: DownloadStatus::Queued, + sender: mpsc::channel().0, + queue_position: None, + metadata: DownloadMetadata::default(), + } + } #[test] - fn finalize_rejects_truncated_download() { - // Peer closed early: 5 of 10 promised bytes. Must be a failure so the - // partial file is never reported as Completed. - let result = - DownloadPeer::finalize_download_buffer(vec![1, 2, 3, 4, 5], 10); + fn a_partial_file_sets_the_resume_offset() { + let dir = scratch_dir("resume"); + std::fs::write(dir.join("song.mp3.part"), b"0123").unwrap(); + + let mut part = PartFile::open(&download_into(&dir, 10)).unwrap(); + assert_eq!(part.written, 4, "resume from what is already on disk"); + + part.write(b"456789").unwrap(); + let final_path = part.finish().unwrap(); + + assert_eq!(std::fs::read(&final_path).unwrap(), b"0123456789"); + assert!( + !dir.join("song.mp3.part").exists(), + "the .part is renamed, not left behind" + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn an_oversized_partial_file_restarts_from_zero() { + // Left over from a different transfer — not a valid resume point. + let dir = scratch_dir("stale"); + std::fs::write(dir.join("song.mp3.part"), vec![9u8; 40]).unwrap(); + + let mut part = PartFile::open(&download_into(&dir, 10)).unwrap(); + assert_eq!(part.written, 0); + + part.write(&(0..10).collect::>()).unwrap(); + let final_path = part.finish().unwrap(); + assert_eq!( + std::fs::read(final_path).unwrap(), + (0..10).collect::>() + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn a_truncated_transfer_fails_and_keeps_the_partial_file() { + let dir = scratch_dir("truncated"); + let mut part = PartFile::open(&download_into(&dir, 10)).unwrap(); + part.write(b"01234").unwrap(); + assert!(matches!( - result, + part.finish(), Err(DownloadError::IncompleteDownload { received: 5, expected: 10 }) )); + assert_eq!( + std::fs::read(dir.join("song.mp3.part")).unwrap(), + b"01234", + "the partial stays on disk for the next attempt" + ); + assert!(!dir.join("song.mp3").exists()); + let _ = std::fs::remove_dir_all(dir); } #[test] - fn finalize_trims_overshoot_to_expected_size() { - // Peer sent 12 bytes for a 10-byte file (trailing bytes coalesced in). - let result = - DownloadPeer::finalize_download_buffer((0..12).collect(), 10); - assert_eq!(result.unwrap(), (0..10).collect::>()); - } + fn overshoot_past_the_expected_size_is_dropped() { + let dir = scratch_dir("overshoot"); + let mut part = PartFile::open(&download_into(&dir, 10)).unwrap(); + part.write(&(0..12).collect::>()).unwrap(); - #[test] - fn finalize_accepts_exact_size() { - let result = - DownloadPeer::finalize_download_buffer((0..10).collect(), 10); - assert_eq!(result.unwrap(), (0..10).collect::>()); + assert!(part.is_complete()); + assert_eq!( + std::fs::read(part.finish().unwrap()).unwrap(), + (0..10).collect::>() + ); + let _ = std::fs::remove_dir_all(dir); } #[test] diff --git a/soulseek-rs-lib/src/peer/upload_peer.rs b/soulseek-rs-lib/src/peer/upload_peer.rs index 8ed3359..34c16ed 100644 --- a/soulseek-rs-lib/src/peer/upload_peer.rs +++ b/soulseek-rs-lib/src/peer/upload_peer.rs @@ -1,7 +1,7 @@ //! Serving a shared file to a peer over an F (file transfer) connection. use std::fs::File; -use std::io::{self, Read, Write}; +use std::io::{self, Read, Seek, SeekFrom, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::path::Path; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; @@ -55,9 +55,16 @@ pub fn serve_file( stream.write_all(&init)?; stream.flush()?; - // The downloader replies with an 8-byte START_DOWNLOAD offset first. + // The downloader replies with an 8-byte START_DOWNLOAD offset first. A + // non-zero offset means it is resuming and already holds that prefix, so + // seek past it and count it as sent to keep progress absolute. let mut offset = [0u8; 8]; stream.read_exact(&mut offset)?; + let offset = u64::from_le_bytes(offset); + if offset > 0 { + file.seek(SeekFrom::Start(offset))?; + bytes_sent.fetch_add(offset, Ordering::Relaxed); + } let mut buffer = vec![0u8; 64 * 1024]; loop { @@ -86,24 +93,50 @@ pub fn serve_file( mod tests { use super::serve_file; use std::io::{Read, Write}; - use std::net::TcpListener; + use std::net::{TcpListener, TcpStream}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + /// Play the downloader: accept the F connection, drain the PeerInit frame + /// and the raw token, then reply with the 8-byte START_DOWNLOAD offset. + /// Returns the connected stream and the token the uploader announced. + fn accept_and_send_offset( + listener: &TcpListener, + offset: u64, + ) -> (TcpStream, u32) { + let (mut stream, _) = listener.accept().unwrap(); + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).unwrap(); + let mut payload = vec![0u8; u32::from_le_bytes(len_buf) as usize]; + stream.read_exact(&mut payload).unwrap(); + let mut token = [0u8; 4]; + stream.read_exact(&mut token).unwrap(); + + stream.write_all(&offset.to_le_bytes()).unwrap(); + stream.flush().unwrap(); + (stream, u32::from_le_bytes(token)) + } + + fn scratch_file(name: &str, content: &[u8]) -> std::path::PathBuf { + let dir = std::env::temp_dir() + .join(format!("soulseek-upload-{name}-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("probe.bin"); + std::fs::write(&path, content).unwrap(); + path + } + #[test] fn serve_file_streams_the_file_over_an_f_connection() { let content: Vec = (0..4096u32).map(|i| (i % 251) as u8).collect(); - let dir = std::env::temp_dir() - .join(format!("soulseek-upload-{}", std::process::id())); - std::fs::create_dir_all(&dir).unwrap(); - let path = dir.join("probe.bin"); - std::fs::write(&path, &content).unwrap(); + let path = scratch_file("stream", &content); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let port = u32::from(listener.local_addr().unwrap().port()); let bytes_sent = Arc::new(AtomicU64::new(0)); let sent_counter = bytes_sent.clone(); + let dir = path.parent().unwrap().to_path_buf(); let uploader = std::thread::spawn(move || { serve_file( "127.0.0.1", @@ -116,20 +149,8 @@ mod tests { ) }); - // Downloader side: accept, read PeerInit(F) frame + raw token, send the - // 8-byte START_DOWNLOAD offset, then read the streamed bytes. - let (mut stream, _) = listener.accept().unwrap(); - let mut len_buf = [0u8; 4]; - stream.read_exact(&mut len_buf).unwrap(); - let len = u32::from_le_bytes(len_buf) as usize; - let mut payload = vec![0u8; len]; - stream.read_exact(&mut payload).unwrap(); - let mut token = [0u8; 4]; - stream.read_exact(&mut token).unwrap(); - assert_eq!(u32::from_le_bytes(token), 777); - - stream.write_all(&[0u8; 8]).unwrap(); - stream.flush().unwrap(); + let (mut stream, token) = accept_and_send_offset(&listener, 0); + assert_eq!(token, 777); let mut received = Vec::new(); stream.read_to_end(&mut received).unwrap(); @@ -140,13 +161,49 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + #[test] + fn serve_file_resumes_from_a_non_zero_offset() { + let content: Vec = (0..4096u32).map(|i| (i % 251) as u8).collect(); + let path = scratch_file("resume", &content); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = u32::from(listener.local_addr().unwrap().port()); + + let bytes_sent = Arc::new(AtomicU64::new(0)); + let sent_counter = bytes_sent.clone(); + let dir = path.parent().unwrap().to_path_buf(); + let uploader = std::thread::spawn(move || { + serve_file( + "127.0.0.1", + port, + "me", + 779, + &path, + &sent_counter, + &AtomicBool::new(false), + ) + }); + + // The downloader already holds the first 1000 bytes. + let (mut stream, _) = accept_and_send_offset(&listener, 1000); + + let mut received = Vec::new(); + stream.read_to_end(&mut received).unwrap(); + assert_eq!(received, content[1000..], "only the missing tail is sent"); + + let _ = uploader.join(); + assert_eq!( + bytes_sent.load(Ordering::Relaxed), + 4096, + "progress counts the resumed prefix" + ); + let _ = std::fs::remove_dir_all(dir); + } + #[test] fn serve_file_stops_when_cancelled() { - let dir = std::env::temp_dir() - .join(format!("soulseek-upload-cancel-{}", std::process::id())); - std::fs::create_dir_all(&dir).unwrap(); - let path = dir.join("big.bin"); - std::fs::write(&path, vec![7u8; 1024 * 1024]).unwrap(); + let path = scratch_file("cancel", &vec![7u8; 1024 * 1024]); + let dir = path.parent().unwrap().to_path_buf(); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let port = u32::from(listener.local_addr().unwrap().port()); @@ -167,15 +224,7 @@ mod tests { ) }); - let (mut stream, _) = listener.accept().unwrap(); - let mut len_buf = [0u8; 4]; - stream.read_exact(&mut len_buf).unwrap(); - let mut payload = vec![0u8; u32::from_le_bytes(len_buf) as usize]; - stream.read_exact(&mut payload).unwrap(); - let mut token = [0u8; 4]; - stream.read_exact(&mut token).unwrap(); - stream.write_all(&[0u8; 8]).unwrap(); - stream.flush().unwrap(); + let (mut stream, _) = accept_and_send_offset(&listener, 0); let mut received = Vec::new(); let _ = stream.read_to_end(&mut received); diff --git a/soulseek-rs-lib/src/shares/mod.rs b/soulseek-rs-lib/src/shares/mod.rs index 5053e67..1314dd6 100644 --- a/soulseek-rs-lib/src/shares/mod.rs +++ b/soulseek-rs-lib/src/shares/mod.rs @@ -198,6 +198,14 @@ fn scan_root( if meta.is_dir() { stack.push(path); } else if meta.is_file() { + // The download directory is usually shared too; an in-progress + // `.part` file is not something to offer other peers. + if path + .to_string_lossy() + .ends_with(crate::utils::path::PART_SUFFIX) + { + continue; + } let virtual_path = virtual_path_for(root_name, root, &path); files.push(SharedFile { virtual_path, @@ -253,6 +261,18 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[test] + fn in_progress_downloads_are_not_shared() { + let root = temp_tree(); + std::fs::write(root.join("half.mp3.part"), b"dd").unwrap(); + + let shares = Shares::scan(&root).unwrap(); + assert_eq!(shares.file_count(), 3, ".part files stay out of the index"); + let base = root.file_name().unwrap().to_string_lossy().into_owned(); + assert!(shares.get(&format!("{base}\\half.mp3.part")).is_none()); + let _ = std::fs::remove_dir_all(root); + } + #[test] fn nested_files_use_backslash_virtual_paths() { let root = temp_tree(); diff --git a/soulseek-rs-lib/src/utils/path.rs b/soulseek-rs-lib/src/utils/path.rs index b1d261d..80d0462 100644 --- a/soulseek-rs-lib/src/utils/path.rs +++ b/soulseek-rs-lib/src/utils/path.rs @@ -1,5 +1,11 @@ use std::path::PathBuf; +/// Extension of the file a download streams into while it is still running. +/// +/// Downloads usually land in a shared directory, so the transfer code that +/// creates these and the share scanner that must skip them agree here. +pub const PART_SUFFIX: &str = ".part"; + #[must_use] pub fn expand_tilde(path: &str) -> PathBuf { if let Some(stripped) = path.strip_prefix('~') diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 9f6c4d9..11ffbd3 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -22,7 +22,7 @@ use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::path::PathBuf; use std::process::{Child, Command, Stdio}; -use std::sync::mpsc::Sender; +use std::sync::mpsc::{Receiver, Sender}; use std::time::{Duration, Instant}; use soulseek_rs::message::Message; @@ -520,7 +520,7 @@ fn connect_retry(addr: &str, timeout: Duration) -> std::io::Result { } } -fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<()> { +fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result { // 1. P (control) connection: register ourselves with the downloader. let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?; p.set_read_timeout(Some(Duration::from_secs(10)))?; @@ -571,12 +571,15 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<()> { /// the token lands in the listener's read buffer, where the download is looked /// up by token; the downloader then sends an 8-byte START_DOWNLOAD offset before /// we send the bytes. +/// +/// Like a real peer, only the bytes past that offset are sent. Returns the +/// offset the downloader asked for. fn serve_file_over_f( downloader_addr: &str, username: &str, token: u32, content: &[u8], -) -> std::io::Result<()> { +) -> std::io::Result { let mut f = connect_retry(downloader_addr, Duration::from_secs(5))?; f.set_read_timeout(Some(Duration::from_secs(10)))?; let mut init = peer_init_bytes(username, "F", token); @@ -586,12 +589,13 @@ fn serve_file_over_f( let mut start = [0u8; 8]; f.read_exact(&mut start)?; - f.write_all(content)?; + let offset = u64::from_le_bytes(start); + f.write_all(&content[offset as usize..])?; f.flush()?; // Keep the connection open briefly so the reader drains everything. std::thread::sleep(Duration::from_millis(500)); - Ok(()) + Ok(offset) } /// Log a raw socket in to the server and drain up to the login response, @@ -653,6 +657,34 @@ fn expect_code( }) } +/// Wait for a download to reach `Completed`, watching both its status channel +/// and the client's download list (whichever reports first). Returns false if it +/// fails, times out, or the deadline passes. +fn wait_for_completion( + client: &Client, + status_rx: &Receiver, + timeout: Duration, +) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + match status_rx.recv_timeout(Duration::from_millis(500)) { + Ok(DownloadStatus::Completed) => return true, + Ok(DownloadStatus::Failed(_) | DownloadStatus::TimedOut) => { + return false; + } + _ => {} + } + if client + .get_all_downloads() + .iter() + .any(|d| matches!(d.status, DownloadStatus::Completed)) + { + return true; + } + } + false +} + fn unique_download_dir() -> PathBuf { let dir = std::env::temp_dir().join(format!( "soulseek-e2e-dl-{}-{:?}", @@ -716,29 +748,8 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() { ) .expect("start download"); - // Wait for completion via the per-download status channel. - let mut completed = false; - let deadline = Instant::now() + Duration::from_secs(20); - while Instant::now() < deadline { - match status_rx.recv_timeout(Duration::from_millis(500)) { - Ok(DownloadStatus::Completed) => { - completed = true; - break; - } - Ok(DownloadStatus::Failed(_) | DownloadStatus::TimedOut) => { - break; - } - _ => {} - } - if client - .get_all_downloads() - .iter() - .any(|d| matches!(d.status, DownloadStatus::Completed)) - { - completed = true; - break; - } - } + let completed = + wait_for_completion(&client, &status_rx, Duration::from_secs(20)); let _ = uploader.join(); assert!(completed, "the download should reach Completed"); @@ -750,6 +761,87 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() { let _ = std::fs::remove_dir_all(&download_dir); } +#[test] +fn an_interrupted_download_resumes_from_its_partial_file() { + let server = server_or_skip!(); + + let listen_port = free_port().expect("free listen port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_resume_dl", + "pw", + listen_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let filename = "resumed_song.mp3"; + let content: Vec = (0..2000u32).map(|i| (i % 251) as u8).collect(); + let size = content.len() as u64; + let already_have = 800usize; + let download_dir = unique_download_dir(); + + // Stand in for a transfer that died after 800 of 2000 bytes. + std::fs::write( + download_dir.join(format!("{filename}.part")), + &content[..already_have], + ) + .expect("seed the partial file"); + + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let cfg = MockUpload { + listen_addr: format!("127.0.0.1:{listen_port}"), + peer_username: "e2e_resume_peer".to_string(), + filename: filename.to_string(), + content: content.clone(), + token: 424_243_u32, + ready: ready_tx, + }; + let (offset_tx, offset_rx) = std::sync::mpsc::channel(); + let uploader = std::thread::spawn(move || match run_mock_uploader(&cfg) { + Ok(offset) => { + let _ = offset_tx.send(offset); + } + Err(e) => eprintln!("[mock uploader] {e}"), + }); + + ready_rx + .recv_timeout(Duration::from_secs(5)) + .expect("mock uploader P connection"); + std::thread::sleep(Duration::from_millis(1500)); + + let (_download, status_rx) = client + .download( + filename.to_string(), + "e2e_resume_peer:direct".to_string(), + size, + download_dir.display().to_string(), + ) + .expect("start download"); + + let completed = + wait_for_completion(&client, &status_rx, Duration::from_secs(20)); + let _ = uploader.join(); + + assert!(completed, "the resumed download should reach Completed"); + assert_eq!( + offset_rx.recv_timeout(Duration::from_secs(1)), + Ok(already_have as u64), + "the peer should be asked to start past the bytes we already had" + ); + assert_eq!( + std::fs::read(download_dir.join(filename)) + .expect("downloaded file should exist"), + content, + "the resumed prefix and the fetched tail should form the whole file" + ); + assert!( + !download_dir.join(format!("{filename}.part")).exists(), + "the .part should be renamed away once complete" + ); + + let _ = std::fs::remove_dir_all(&download_dir); +} + #[test] fn soulfind_brokers_connect_to_peer_between_users() { // The firewalled download mode relies on the server forwarding a @@ -911,6 +1003,7 @@ fn run_mock_direct_peer(cfg: &MockDirectUpload) -> std::io::Result<()> { cfg.token, &cfg.content, ) + .map(|_| ()) } #[test] @@ -969,26 +1062,8 @@ fn a_file_downloads_from_a_peer_via_direct_connection() { ) .expect("start download"); - let mut completed = false; - let deadline = Instant::now() + Duration::from_secs(20); - while Instant::now() < deadline { - match status_rx.recv_timeout(Duration::from_millis(500)) { - Ok(DownloadStatus::Completed) => { - completed = true; - break; - } - Ok(DownloadStatus::Failed(_) | DownloadStatus::TimedOut) => break, - _ => {} - } - if client - .get_all_downloads() - .iter() - .any(|d| matches!(d.status, DownloadStatus::Completed)) - { - completed = true; - break; - } - } + let completed = + wait_for_completion(&client, &status_rx, Duration::from_secs(20)); let _ = uploader.join(); assert!(completed, "the direct download should reach Completed"); @@ -1074,6 +1149,7 @@ fn run_mock_firewalled_peer(cfg: &MockFirewalledUpload) -> std::io::Result<()> { cfg.token, &cfg.content, ) + .map(|_| ()) } #[test] @@ -1128,26 +1204,8 @@ fn a_file_downloads_from_a_firewalled_peer_via_server_broker() { ) .expect("start download"); - let mut completed = false; - let deadline = Instant::now() + Duration::from_secs(25); - while Instant::now() < deadline { - match status_rx.recv_timeout(Duration::from_millis(500)) { - Ok(DownloadStatus::Completed) => { - completed = true; - break; - } - Ok(DownloadStatus::Failed(_) | DownloadStatus::TimedOut) => break, - _ => {} - } - if client - .get_all_downloads() - .iter() - .any(|d| matches!(d.status, DownloadStatus::Completed)) - { - completed = true; - break; - } - } + let completed = + wait_for_completion(&client, &status_rx, Duration::from_secs(25)); let _ = uploader.join(); assert!(completed, "the firewalled download should reach Completed"); @@ -1229,26 +1287,8 @@ fn two_real_clients_search_and_download() { ) .expect("start download"); - let mut completed = false; - let deadline = Instant::now() + Duration::from_secs(25); - while Instant::now() < deadline { - match status_rx.recv_timeout(Duration::from_millis(500)) { - Ok(DownloadStatus::Completed) => { - completed = true; - break; - } - Ok(DownloadStatus::Failed(_) | DownloadStatus::TimedOut) => break, - _ => {} - } - if leecher - .get_all_downloads() - .iter() - .any(|d| matches!(d.status, DownloadStatus::Completed)) - { - completed = true; - break; - } - } + let completed = + wait_for_completion(&leecher, &status_rx, Duration::from_secs(25)); assert!(completed, "the download should complete"); // The virtual path is backslash-separated; the saved file uses the basename. diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index 6da4d96..719eebb 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -51,7 +51,9 @@ Flags: `--min-bitrate `, `--free-slots`, `-n/--limit` (0 means all), **`download [--size N]`** or **`download --stdin`** — one object per finished transfer: `user`, `path`, `size`, `file` (the local path written). -`--stdin` consumes `search --json` output as-is. +`--stdin` consumes `search --json` output as-is. A transfer that fails leaves a +`.part` behind; re-running the same command resumes from it rather than +starting over, so retrying a failed download is cheap. **`get `** — search, pick, and download in one step. `--pick best|first|all`, `-n/--limit`, `--min-bitrate`, `--free-slots`. Prefer this