use std::fs::{self, File, OpenOptions}; use std::io::{self, Read, Write}; use std::net::TcpStream; use std::net::ToSocketAddrs; use std::path::PathBuf; use std::sync::{Arc, RwLock}; use std::thread; use std::time::{Duration, Instant}; use crate::client::ClientContext; use crate::message::server::MessageFactory; use crate::trace; use crate::types::{Download, DownloadStatus}; use crate::utils::path::{PART_SUFFIX, expand_tilde}; /// Read in 64 KiB slices: an eighth of the syscalls of the 8 KiB reads before, /// and what a fast peer fills per read. slskd reads 256 KiB, Nicotine+ ~50 KiB. const READ_BUFFER_SIZE: usize = 64 * 1024; const CANCEL_POLL: Duration = Duration::from_secs(1); const STALL_DEADLINE: Duration = Duration::from_secs(30); /// Bytes between progress reports. Counting bytes rather than reads keeps /// the report volume flat however big a read is, and gives a slow peer's /// rate a window wide enough to mean something. const PROGRESS_INTERVAL_BYTES: usize = 128 * 1024; #[derive(Debug)] pub enum DownloadError { ConnectionFailed(io::Error), InvalidAddress(String), HandshakeFailed(io::Error), StreamReadError(io::Error), StreamWriteError(io::Error), TokenNotFound(u32), DownloadInfoMissing(u32), FileWriteError(io::Error), PathResolutionError(String), InvalidTokenBytes, LockPoisoned, Cancelled, IncompleteDownload { received: usize, expected: usize }, } impl std::fmt::Display for DownloadError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::ConnectionFailed(e) => write!(f, "Connection failed: {e}"), Self::InvalidAddress(addr) => { write!(f, "Invalid address: {addr}") } Self::HandshakeFailed(e) => write!(f, "Handshake failed: {e}"), Self::StreamReadError(e) => write!(f, "Stream read error: {e}"), Self::StreamWriteError(e) => write!(f, "Stream write error: {e}"), Self::TokenNotFound(token) => { write!(f, "Token not found: {token}") } Self::DownloadInfoMissing(token) => { write!(f, "Download info missing for token: {token}") } Self::FileWriteError(e) => write!(f, "File write error: {e}"), Self::PathResolutionError(msg) => { write!(f, "Path resolution error: {msg}") } Self::InvalidTokenBytes => { write!(f, "Invalid token bytes received") } Self::LockPoisoned => write!(f, "Lock poisoned"), Self::Cancelled => write!(f, "Download cancelled"), Self::IncompleteDownload { received, expected } => write!( f, "Incomplete download: received {received} of {expected} bytes" ), } } } impl std::error::Error for DownloadError {} fn extract_filename_from_path(full_path: &str) -> &str { full_path .split(['/', '\\']) .next_back() .unwrap_or(full_path) } fn part_path(final_path: &str) -> String { format!("{final_path}{PART_SUFFIX}") } pub fn part_path_of(download: &Download) -> Option { resolve_download_path(download) .ok() .map(|path| part_path(&path)) } /// 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. /// /// A directory that does not exist yet stays the target — `PartFile::open` /// creates it. Treating it as "not a directory" and dodging to the parent /// made a daemon whose folder was never created download into `~/Downloads` /// while reporting `~/Downloads/Soulseek`. fn resolve_download_path(download: &Download) -> Result { let mut expanded_path = expand_tilde(&download.download_directory); if expanded_path.is_file() { 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 = expanded_path.join(extract_filename_from_path(&download.filename)); 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 = resolve_download_path(download)?; let path = PathBuf::from(part_path(&final_path)); 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 { 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 } fn discard(self) { drop(self.file); let _ = fs::remove_file(part_path(&self.final_path)); } /// 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(part_path(&self.final_path), &self.final_path) .map_err(DownloadError::FileWriteError)?; Ok(self.final_path) } } pub struct DownloadPeer { username: String, host: String, port: u32, #[allow(dead_code)] own_username: String, token: u32, no_pierce: bool, } impl DownloadPeer { #[must_use] pub const fn new( username: String, host: String, port: u32, token: u32, no_pierce: bool, own_username: String, ) -> Self { Self { username, host, port, own_username, token, no_pierce, } } fn establish_connection(&self) -> Result { let socket_address = format!("{}:{}", self.host, self.port) .to_socket_addrs() .map_err(DownloadError::ConnectionFailed)? .next() .ok_or_else(|| { DownloadError::InvalidAddress(format!( "{}:{}", self.host, self.port )) })?; let stream = TcpStream::connect_timeout( &socket_address, Duration::from_secs(20), ) .map_err(DownloadError::ConnectionFailed)?; Self::apply_transfer_deadlines(&stream)?; Ok(stream) } /// Bound every stall on a transfer socket. This must cover streams the /// listener accepted as well as ones we dialled: an accepted F connection /// used to arrive here with no read timeout at all, so a peer that sent a /// token and went silent parked the transfer thread — holding the socket /// and the open `.part` file — for the rest of the session. fn apply_transfer_deadlines( stream: &TcpStream, ) -> Result<(), DownloadError> { stream .set_read_timeout(Some(STALL_DEADLINE)) .map_err(DownloadError::ConnectionFailed)?; stream .set_write_timeout(Some(Duration::from_secs(5))) .map_err(DownloadError::ConnectionFailed)?; stream .set_nodelay(true) .map_err(DownloadError::ConnectionFailed)?; Ok(()) } fn perform_handshake( &self, stream: &mut TcpStream, ) -> Result<(), DownloadError> { trace!( "[download_peer:{}] performing handshake no_pierce: {}", self.username, self.no_pierce ); if !self.no_pierce { let message = MessageFactory::build_pierce_firewall_message(self.token); stream .write_all(&message.get_buffer()) .map_err(DownloadError::HandshakeFailed)?; trace!( "[download_peer:{}] sending pierce firewall message token: {}: {:?}", self.username, self.token, &message.get_buffer() ); stream.flush().map_err(DownloadError::HandshakeFailed)?; } Ok(()) } /// 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> { Self::wait_while_paused(client_context, &download)?; 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<(Download, PartFile), DownloadError> { let token_bytes = data.get(0..4).ok_or(DownloadError::InvalidTokenBytes)?; let token_array: [u8; 4] = token_bytes .try_into() .map_err(|_| DownloadError::InvalidTokenBytes)?; let token_u32 = u32::from_le_bytes(token_array); trace!( "[download_peer:{}] got token: {} from data chunk", self.username, token_u32 ); let client_guard = client_context .read() .map_err(|_| DownloadError::LockPoisoned)?; let download_info = client_guard.get_download_by_token(token_u32).cloned(); drop(client_guard); 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>, download: Option, ) -> Result<(Download, String), DownloadError> { let mut read_buffer = vec![0u8; READ_BUFFER_SIZE]; let mut bytes_since_last_update = 0usize; let mut last_update_time = Instant::now(); let mut last_data = Instant::now(); stream .set_read_timeout(Some(CANCEL_POLL)) .map_err(DownloadError::ConnectionFailed)?; trace!( "[download_peer:{}] Starting to read data from peer", self.username ); let mut transfer = match download { Some(dl) => Some(Self::start_transfer(dl, stream, client_context)?), None => None, }; loop { if let Some((dl, _)) = &transfer { match Self::wait_while_paused(client_context, dl) { Ok(true) => last_data = Instant::now(), Ok(false) => {} Err(e) => { if matches!(e, DownloadError::Cancelled) && let Some((_, part)) = transfer.take() { part.discard(); } return Err(e); } } } match stream.read(&mut read_buffer) { Ok(0) => { trace!( "[download_peer:{}] connection closed by peer", self.username ); break; } Ok(bytes_read) => { last_data = Instant::now(); let data = &read_buffer[..bytes_read]; if transfer.is_none() && !self.no_pierce { transfer = Some(self.begin_pierced_download( data, stream, client_context, )?); continue; } let Some((dl, part)) = transfer.as_mut() else { return Err(DownloadError::DownloadInfoMissing( self.token, )); }; part.write(data)?; bytes_since_last_update += bytes_read; if bytes_since_last_update >= PROGRESS_INTERVAL_BYTES { let elapsed = last_update_time.elapsed().as_secs_f64(); let speed = if elapsed > 0.0 { bytes_since_last_update as f64 / elapsed } else { 0.0 }; Self::report_progress( client_context, dl, part.written, speed, ); bytes_since_last_update = 0; last_update_time = Instant::now(); } if part.is_complete() { break; } } Err(e) if matches!( e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut ) && last_data.elapsed() < STALL_DEADLINE => {} Err(e) => { return Err(DownloadError::StreamReadError(e)); } } } trace!( "[download_peer:{}] finished reading data from peer", self.username ); let Some((download, part)) = transfer else { return Err(DownloadError::DownloadInfoMissing(self.token)); }; Ok((download, part.finish()?)) } 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( client_context: &Arc>, download: &Download, status: DownloadStatus, ) { let applied = client_context.write().is_ok_and(|mut context| { context .downloads .update_status(download.token, status.clone()) }); if applied { let _ = download.sender.send(status); } } fn wait_while_paused( client_context: &Arc>, download: &Download, ) -> Result { let mut waited = false; loop { let status = client_context .read() .map_err(|_| DownloadError::LockPoisoned)? .get_download_by_token(download.token) .map(|download| download.status.clone()) .ok_or(DownloadError::TokenNotFound(download.token))?; if matches!(status, DownloadStatus::Cancelled) { return Err(DownloadError::Cancelled); } if !matches!(status, DownloadStatus::Paused { .. }) { return Ok(waited); } waited = true; thread::sleep(Duration::from_millis(200)); } } pub fn download_file( self, client_context: Arc>, download: Option, stream: Option, ) -> Result<(Download, String), DownloadError> { trace!( "[download_peer:{}] download_file: download is present?: {:?}, stream is present?: {:?}, no_pierce: {}", self.username, download.is_some(), stream.is_some(), self.no_pierce ); let mut stream = match stream { Some(s) => { Self::apply_transfer_deadlines(&s)?; s } None => self.establish_connection()?, }; trace!("[download_peer:{}] connected", self.username); self.perform_handshake(&mut stream)?; trace!("[download_peer:{}] handshake completed", self.username); let (download, final_path) = self.read_download_stream(&mut stream, &client_context, download)?; trace!( "[download_peer:{}] download completed successfully: {} bytes, saved to: {}", self.username, download.size, final_path ); Ok((download, final_path)) } } #[cfg(test)] mod tests { use super::{ DownloadError, DownloadPeer, PartFile, extract_filename_from_path, }; use crate::client::ClientContext; use crate::types::{Download, DownloadMetadata, DownloadStatus}; use std::sync::{Arc, RwLock, 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 a_discarded_part_file_is_removed_from_disk() { let dir = scratch_dir("discard"); let mut part = PartFile::open(&download_into(&dir, 8)).unwrap(); part.write(b"data").unwrap(); let path = dir.join("song.mp3.part"); assert!(path.exists()); part.discard(); assert!(!path.exists()); assert!(!dir.join("song.mp3").exists()); let _ = std::fs::remove_dir_all(&dir); } #[test] fn a_pause_is_reported_so_the_stall_budget_restarts() { let dir = scratch_dir("paused"); let mut download = download_into(&dir, 8); download.status = DownloadStatus::Paused { bytes_downloaded: 0, total_bytes: 8, }; let context = Arc::new(RwLock::new(ClientContext::new())); context.write().unwrap().add_download(download.clone()); let resumer = { let context = context.clone(); std::thread::spawn(move || { std::thread::sleep(std::time::Duration::from_millis(300)); assert!( context .write() .unwrap() .downloads .resume_by_file("peer", "song.mp3") ); }) }; assert!(matches!( DownloadPeer::wait_while_paused(&context, &download), Ok(true) )); assert!(matches!( DownloadPeer::wait_while_paused(&context, &download), Ok(false) )); let _ = resumer.join(); let _ = std::fs::remove_dir_all(&dir); } #[test] fn a_cancelled_download_stops_the_transfer_loop() { let dir = scratch_dir("cancelled"); let mut download = download_into(&dir, 8); download.status = DownloadStatus::Cancelled; let context = Arc::new(RwLock::new(ClientContext::new())); context.write().unwrap().add_download(download.clone()); assert!(matches!( DownloadPeer::wait_while_paused(&context, &download), Err(DownloadError::Cancelled) )); let _ = std::fs::remove_dir_all(&dir); } #[test] fn a_status_after_cancel_is_not_relayed_on_the_channel() { let dir = scratch_dir("relay"); let (tx, rx) = mpsc::channel(); let mut download = download_into(&dir, 8); download.sender = tx; download.status = DownloadStatus::Cancelled; let context = Arc::new(RwLock::new(ClientContext::new())); context.write().unwrap().add_download(download.clone()); DownloadPeer::send_download_status( &context, &download, DownloadStatus::InProgress { bytes_downloaded: 4, total_bytes: 8, speed_bytes_per_sec: 1.0, }, ); assert!(rx.try_recv().is_err(), "a cancelled row hears nothing more"); let _ = std::fs::remove_dir_all(&dir); } #[test] fn start_transfer_refuses_a_cancelled_download() { let dir = scratch_dir("refuse"); let mut download = download_into(&dir, 8); download.status = DownloadStatus::Cancelled; let context = Arc::new(RwLock::new(ClientContext::new())); context.write().unwrap().add_download(download.clone()); let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let mut stream = std::net::TcpStream::connect(listener.local_addr().unwrap()) .unwrap(); let _peer = listener.accept().unwrap(); let outcome = DownloadPeer::start_transfer(download, &mut stream, &context); assert!(matches!(outcome, Err(DownloadError::Cancelled))); assert!(!dir.join("song.mp3.part").exists(), "no .part is opened"); let _ = std::fs::remove_dir_all(&dir); } #[test] fn a_directory_that_does_not_exist_yet_is_created_not_dodged() { let dir = scratch_dir("fresh").join("Soulseek"); let mut part = PartFile::open(&download_into(&dir, 4)).unwrap(); part.write(b"data").unwrap(); let final_path = part.finish().unwrap(); assert_eq!( std::path::Path::new(&final_path).parent(), Some(dir.as_path()), "the bytes belong in the configured directory, not its parent" ); assert_eq!(std::fs::read(&final_path).unwrap(), b"data"); let _ = std::fs::remove_dir_all(dir.parent().unwrap()); } #[test] 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!( 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 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(); assert!(part.is_complete()); assert_eq!( std::fs::read(part.finish().unwrap()).unwrap(), (0..10).collect::>() ); let _ = std::fs::remove_dir_all(dir); } // Streams the listener accepted arrive with no deadlines; without these a // peer that sends a token and goes silent parks the transfer thread — // socket and open .part file included — for the rest of the session. #[test] fn supplied_streams_get_transfer_deadlines() { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let stream = std::net::TcpStream::connect(listener.local_addr().unwrap()) .unwrap(); DownloadPeer::apply_transfer_deadlines(&stream).unwrap(); let secs = std::time::Duration::from_secs; assert_eq!(stream.read_timeout().unwrap(), Some(secs(30))); assert_eq!(stream.write_timeout().unwrap(), Some(secs(5))); } #[test] fn test_establish_connection_invalid_address() { let download_peer = DownloadPeer::new( "user".to_string(), "invalid-host".to_string(), 9999, 123, false, "own_user".to_string(), ); let result = download_peer.establish_connection(); assert!(result.is_err()); } #[test] fn test_extract_filename_from_path() { assert_eq!(extract_filename_from_path("/path/to/file.mp3"), "file.mp3"); assert_eq!( extract_filename_from_path("C:\\path\\to\\file.mp3"), "file.mp3" ); assert_eq!( extract_filename_from_path( "@@bhfrv\\Soulseek Downloads\\complete\\Beatport Top Deep House (2021)\\michel test file.mp3" ), "michel test file.mp3" ); assert_eq!(extract_filename_from_path("file.mp3"), "file.mp3"); } }