From 393f69e478262c1569767f805e5cef4117bbfec2 Mon Sep 17 00:00:00 2001 From: michel Date: Sun, 19 Jul 2026 18:46:06 +0200 Subject: [PATCH] feat(shares): serve queued files to peers over F connections Implement the upload path so a peer can download our shared files: - fix the mis-coded PlaceInQueueResponse (43 -> 44) and add a real QueueUpload handler (code 43) -> PeerMessage::IncomingQueueUpload; - the peer actor forwards it as ClientOperation::QueueUpload; the client looks the file up in its shares, mints an upload token (high half of the token space, disjoint from md5 download tokens), and offers it via ServeUpload -> build_upload_transfer_request (peer code 40, direction=upload); - a TransferResponse for one of our serving tokens is handled by a surgical early-return that emits StartUpload, leaving the download path untouched; - StartUpload resolves the downloader's address (GetPeerAddress, cached) and streams the file via a new peer::upload_peer::serve_file over an F connection. Tested: build_upload_transfer_request decodes via the production Transfer parser; serve_file streams a file correctly over a loopback F connection. --- soulseek-rs-lib/src/actor/peer_actor.rs | 51 ++++- soulseek-rs-lib/src/client.rs | 187 +++++++++++++++++- soulseek-rs-lib/src/message/peer/mod.rs | 2 + .../message/peer/place_in_queue_response.rs | 2 +- .../src/message/peer/queue_upload.rs | 19 ++ .../src/message/server/message_factory.rs | 32 +++ soulseek-rs-lib/src/peer/mod.rs | 1 + soulseek-rs-lib/src/peer/upload_peer.rs | 109 ++++++++++ 8 files changed, 396 insertions(+), 7 deletions(-) create mode 100644 soulseek-rs-lib/src/message/peer/queue_upload.rs create mode 100644 soulseek-rs-lib/src/peer/upload_peer.rs diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 3cf03ef..95717a7 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -3,7 +3,7 @@ use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::peer::{ FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueResponse, - TransferRequest, TransferResponse, UploadFailedHandler, + QueueUploadHandler, TransferRequest, TransferResponse, UploadFailedHandler, }; use crate::message::server::MessageFactory; use crate::message::{Handlers, Message, MessageReader, MessageType}; @@ -36,6 +36,14 @@ pub enum PeerMessage { SetUsername(String), QueueUpload(String), RequestTransfer(Download), + /// A peer queued one of our shared files for download (they sent us code 43). + IncomingQueueUpload(String), + /// Offer the queued file to that peer: send an upload TransferRequest. + ServeUpload { + token: u32, + filename: String, + size: u64, + }, ProcessRead, } @@ -64,6 +72,9 @@ pub struct PeerActor { /// Registry-assigned unique id, echoed in terminal notifications so the /// client evicts only this actor and never a newer namesake. id: u64, + /// Transfer tokens for uploads we are serving to this peer. A TransferResponse + /// for one of these is our upload being accepted, not a download offer. + serving_tokens: std::collections::HashSet, } impl PeerActor { @@ -98,6 +109,7 @@ impl PeerActor { established: false, disconnect_reported: false, id, + serving_tokens: std::collections::HashSet::new(), } } @@ -138,6 +150,7 @@ impl PeerActor { handlers.register_handler(GetShareFileList); handlers.register_handler(UploadFailedHandler); handlers.register_handler(PlaceInQueueResponse); + handlers.register_handler(QueueUploadHandler); handlers.register_handler(PeerInit); self.dispatcher = Some(MessageDispatcher::new( @@ -234,6 +247,18 @@ impl PeerActor { username, token, allowed ); + // If this token is one of our uploads, the peer just accepted + // our offer — start streaming. This leaves the download path + // (every other token) byte-for-byte unchanged. + if self.serving_tokens.remove(&token) { + if allowed { + let _ = self + .client_channel + .send(ClientOperation::StartUpload { token }); + } + return; + } + if allowed { debug!( "[peer:{}] Transfer allowed, ready to connect with token {:}", @@ -298,6 +323,30 @@ impl PeerActor { MessageFactory::build_queue_upload_message(&filename); self.send_message(message); } + PeerMessage::IncomingQueueUpload(filename) => { + // A peer wants to download one of our shared files. Ask the + // client (which owns the shares) to prepare the upload. + let requester_key = self.peer_username(); + if let Err(e) = + self.client_channel.send(ClientOperation::QueueUpload { + requester_key, + filename, + }) + { + error!("[peer_actor] forward IncomingQueueUpload: {}", e); + } + } + PeerMessage::ServeUpload { + token, + filename, + size, + } => { + self.serving_tokens.insert(token); + let message = MessageFactory::build_upload_transfer_request( + &filename, token, size, + ); + self.send_message(message); + } PeerMessage::RequestTransfer(download) => { let message = MessageFactory::build_transfer_request_message( &download.filename, diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index b953719..16d009f 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -47,6 +47,20 @@ fn next_connect_token() -> u32 { NEXT_CONNECT_TOKEN.fetch_add(1, Ordering::Relaxed).max(1) } +/// Upload tokens are minted in the high half of the space so they never collide +/// with download tokens (md5-derived, always < 2^20). +static NEXT_UPLOAD_TOKEN: AtomicU32 = AtomicU32::new(0x8000_0000); + +fn next_upload_token() -> u32 { + NEXT_UPLOAD_TOKEN.fetch_add(1, Ordering::Relaxed) +} + +/// A file we have agreed to serve to a peer, awaiting their TransferResponse. +struct UploadJob { + downloader: String, + real_path: std::path::PathBuf, +} + /// Build a `FileSearchResponse` for `query` against `shares`, or `None` if /// nothing matches. `own_username` is the name the searcher will download from. fn build_search_response( @@ -149,6 +163,16 @@ pub enum ClientOperation { token: u32, query: String, }, + /// A peer queued one of our shared files; `requester_key` is the registry + /// key of the peer actor (may carry a `:direct` suffix). + QueueUpload { + requester_key: String, + filename: String, + }, + /// The peer accepted our upload offer for `token`; start streaming. + StartUpload { + token: u32, + }, /// A direct outbound connection to this peer failed before it was /// established — the peer is likely firewalled, so fall back to asking the /// server to broker the connection. Carries the reporting actor's id. @@ -170,6 +194,10 @@ pub struct ClientContext { peer_addresses: HashMap, /// Peer messages waiting for a control connection to that peer. pending_peer_messages: HashMap>, + /// Uploads we have offered, keyed by our transfer token. + uploads: HashMap, + /// Upload tokens waiting for the downloader's address to be resolved. + pending_serves: HashMap>, actor_system: Arc, } impl Default for ClientContext { @@ -457,6 +485,8 @@ impl ClientContext { shares: Arc::new(Shares::empty()), peer_addresses: HashMap::new(), pending_peer_messages: HashMap::new(), + uploads: HashMap::new(), + pending_serves: HashMap::new(), downloads: DownloadStore::new(), actor_system, } @@ -987,6 +1017,33 @@ impl Client { } } + /// Consume the upload job for `token` and stream the file to `host:port` + /// on a background thread. + fn spawn_serve( + client_context: &Arc>, + own_username: &str, + token: u32, + host: String, + port: u32, + ) { + let Ok(mut ctx) = client_context.write_safe() else { + return; + }; + let Some(job) = ctx.uploads.remove(&token) else { + return; + }; + drop(ctx); + let own = own_username.to_string(); + let real_path = job.real_path; + thread::spawn(move || { + if let Err(e) = crate::peer::upload_peer::serve_file( + &host, port, &own, token, &real_path, + ) { + error!("[client] serve {}: {}", real_path.display(), e); + } + }); + } + fn process_failed_uploads( client_context: Arc>, username: &str, @@ -1317,11 +1374,27 @@ impl Client { obfuscated_port ); - // Cache the address for the serve/search paths. - if let Ok(mut ctx) = client_context.write_safe() - { - ctx.cache_peer_address( - &username, + // Cache the address for the serve/search paths, + // and collect any uploads waiting on it. + let waiting_serves = + match client_context.write_safe() { + Ok(mut ctx) => { + ctx.cache_peer_address( + &username, + host.clone(), + port, + ); + ctx.pending_serves + .remove(&username) + .unwrap_or_default() + } + Err(_) => Vec::new(), + }; + for token in waiting_serves { + Self::spawn_serve( + &client_context, + &own_username, + token, host.clone(), port, ); @@ -1607,6 +1680,110 @@ impl Client { } } } + ClientOperation::QueueUpload { + requester_key, + filename, + } => { + // A peer queued one of our shared files. Look it + // up, mint an upload token, and offer it. + let downloader = requester_key + .strip_suffix(":direct") + .unwrap_or(&requester_key) + .to_string(); + let token = next_upload_token(); + let (registry, size) = match client_context + .write_safe() + { + Ok(mut ctx) => { + let Some(file) = + ctx.shares.get(&filename) + else { + debug!( + "[client] QueueUpload for unknown file {}", + filename + ); + continue; + }; + let size = file.size; + let real_path = file.real_path.clone(); + ctx.uploads.insert( + token, + UploadJob { + downloader: downloader.clone(), + real_path, + }, + ); + (ctx.peer_registry.clone(), size) + } + Err(e) => { + error!( + "[client] QueueUpload write: {}", + e + ); + continue; + } + }; + if let Some(registry) = registry { + let _ = registry.send_to_peer( + &requester_key, + PeerMessage::ServeUpload { + token, + filename, + size, + }, + ); + } + } + ClientOperation::StartUpload { token } => { + // The peer accepted our offer: resolve their + // address (from the code-9 GetPeerAddress) and + // stream the file, or queue until it resolves. + let (job_addr, downloader) = + match client_context.read_safe() { + Ok(ctx) => { + let Some(job) = + ctx.uploads.get(&token) + else { + continue; + }; + ( + ctx.peer_address( + &job.downloader, + ), + job.downloader.clone(), + ) + } + Err(_) => continue, + }; + if let Some((host, port)) = job_addr { + Self::spawn_serve( + &client_context, + &own_username, + token, + host, + port, + ); + } else { + if let Ok(mut ctx) = + client_context.write_safe() + { + ctx.pending_serves + .entry(downloader.clone()) + .or_default() + .push(token); + } + if let Ok(ctx) = client_context.read_safe() + && let Some(sender) = + ctx.server_sender.clone() + { + let _ = sender.send( + ServerMessage::GetPeerAddress( + downloader, + ), + ); + } + } + } ClientOperation::PeerConnectFailed( id, username, diff --git a/soulseek-rs-lib/src/message/peer/mod.rs b/soulseek-rs-lib/src/message/peer/mod.rs index 36b4559..1f7eef8 100644 --- a/soulseek-rs-lib/src/message/peer/mod.rs +++ b/soulseek-rs-lib/src/message/peer/mod.rs @@ -2,6 +2,7 @@ mod file_search_response; mod get_share_file_list; mod peer_init; mod place_in_queue_response; +mod queue_upload; mod transfer_request; mod transfer_response; mod upload_failed; @@ -13,6 +14,7 @@ pub use file_search_response::{ pub use get_share_file_list::GetShareFileList; pub use peer_init::PeerInit; pub use place_in_queue_response::PlaceInQueueResponse; +pub use queue_upload::QueueUploadHandler; pub use transfer_request::TransferRequest; pub use transfer_response::TransferResponse; pub use upload_failed::UploadFailedHandler; diff --git a/soulseek-rs-lib/src/message/peer/place_in_queue_response.rs b/soulseek-rs-lib/src/message/peer/place_in_queue_response.rs index 0d4a38e..ab9c4cf 100644 --- a/soulseek-rs-lib/src/message/peer/place_in_queue_response.rs +++ b/soulseek-rs-lib/src/message/peer/place_in_queue_response.rs @@ -8,7 +8,7 @@ pub struct PlaceInQueueResponse; impl MessageHandler for PlaceInQueueResponse { fn get_code(&self) -> u8 { - 43 + 44 } fn handle(&self, message: &mut Message, sender: Sender) { diff --git a/soulseek-rs-lib/src/message/peer/queue_upload.rs b/soulseek-rs-lib/src/message/peer/queue_upload.rs new file mode 100644 index 0000000..a88241b --- /dev/null +++ b/soulseek-rs-lib/src/message/peer/queue_upload.rs @@ -0,0 +1,19 @@ +use crate::{ + message::{Message, MessageHandler}, + peer::PeerMessage, +}; +use std::sync::mpsc::Sender; + +/// A peer asking to download one of our shared files (peer code 43). +pub struct QueueUploadHandler; + +impl MessageHandler for QueueUploadHandler { + fn get_code(&self) -> u8 { + 43 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let filename = message.read_string(); + let _ = sender.send(PeerMessage::IncomingQueueUpload(filename)); + } +} diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 1ee490d..e11cc0e 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -142,6 +142,23 @@ impl MessageFactory { .write_string(filename) .clone() } + /// A TransferRequest (peer code 40) initiating an *upload*: we offer a file + /// to a peer who queued it, quoting our transfer token and its size. + #[must_use] + pub fn build_upload_transfer_request( + filename: &str, + token: u32, + size: u64, + ) -> Message { + Message::new() + .write_int32(40) + .write_int32(1) // direction: upload + .write_int32(token) + .write_string(filename) + .write_int64(size) + .clone() + } + #[must_use] pub fn build_transfer_response_message(transfer: Transfer) -> Message { Message::new() @@ -199,6 +216,21 @@ fn test_build_login_message() { assert_eq!(expect, message.get_data()); } +#[test] +fn test_build_upload_transfer_request() { + use crate::types::Transfer; + let message = + MessageFactory::build_upload_transfer_request("song.mp3", 555, 4096); + // Decode via the production Transfer parser (dispatcher starts at offset 8). + let mut decoded = Message::new_with_data(message.get_buffer()); + decoded.set_pointer(8); + let transfer = Transfer::new_from_message(&mut decoded); + assert_eq!(transfer.direction, 1); // upload + assert_eq!(transfer.token, 555); + assert_eq!(transfer.filename, "song.mp3"); + assert_eq!(transfer.size, 4096); +} + #[test] fn test_build_peer_init_message() { use crate::peer::ConnectionType; diff --git a/soulseek-rs-lib/src/peer/mod.rs b/soulseek-rs-lib/src/peer/mod.rs index 7006a0b..7d861cd 100644 --- a/soulseek-rs-lib/src/peer/mod.rs +++ b/soulseek-rs-lib/src/peer/mod.rs @@ -1,5 +1,6 @@ mod download_peer; pub mod listen; +pub mod upload_peer; // Export actor types pub use crate::actor::peer_actor::{PeerActor, PeerMessage}; diff --git a/soulseek-rs-lib/src/peer/upload_peer.rs b/soulseek-rs-lib/src/peer/upload_peer.rs new file mode 100644 index 0000000..b0aecd3 --- /dev/null +++ b/soulseek-rs-lib/src/peer/upload_peer.rs @@ -0,0 +1,109 @@ +//! 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::net::{TcpStream, ToSocketAddrs}; +use std::path::Path; +use std::time::Duration; + +use crate::message::server::MessageFactory; +use crate::peer::ConnectionType; +use crate::trace; + +/// Connect to the downloader's file listener and stream `path`'s bytes. +/// +/// We announce ourselves with a `PeerInit(F)` immediately followed by the raw +/// transfer token (so it lands in the downloader's read buffer, where the +/// download is matched by token), then the downloader sends an 8-byte +/// START_DOWNLOAD offset before we stream the file. +/// +/// # Errors +/// Returns any I/O error opening the file or talking to the peer. +pub fn serve_file( + host: &str, + port: u32, + own_username: &str, + token: u32, + path: &Path, +) -> io::Result<()> { + let mut file = File::open(path)?; + + let socket = format!("{host}:{port}") + .to_socket_addrs()? + .next() + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "no address") + })?; + let mut stream = + TcpStream::connect_timeout(&socket, Duration::from_secs(20))?; + stream.set_nodelay(true).ok(); + + // PeerInit(F) + the 4-byte token in a single write so they coalesce. + let mut init = MessageFactory::build_peer_init_message( + own_username, + ConnectionType::F, + token, + ) + .get_buffer(); + init.extend_from_slice(&token.to_le_bytes()); + stream.write_all(&init)?; + stream.flush()?; + + // The downloader replies with an 8-byte START_DOWNLOAD offset first. + let mut offset = [0u8; 8]; + stream.read_exact(&mut offset)?; + + io::copy(&mut file, &mut stream)?; + stream.flush()?; + + // Linger so the downloader drains everything before the socket closes. + std::thread::sleep(Duration::from_millis(500)); + trace!("[upload] served {} to {}:{}", path.display(), host, port); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::serve_file; + use std::io::{Read, Write}; + use std::net::TcpListener; + + #[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 listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = u32::from(listener.local_addr().unwrap().port()); + + let uploader = std::thread::spawn(move || { + serve_file("127.0.0.1", port, "me", 777, &path) + }); + + // 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 received = Vec::new(); + stream.read_to_end(&mut received).unwrap(); + assert_eq!(received, content); + + let _ = uploader.join(); + let _ = std::fs::remove_dir_all(dir); + } +}