diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..a16944a --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,43 @@ +# Contributing + +Issues and pull requests are both welcome, and so is just saying what's missing. +No contributor agreement, no template. + +## Pull requests + +Branch off `develop` and target `develop`. `master` only moves through a release +PR. + +Write the subject as a +[conventional commit](https://www.conventionalcommits.org), like +`fix(lib): route on the full four-byte message code`. release-plz reads these to +pick the next version and write the changelog. One change per pull request; +small ones land fastest. + +Before you push: + +```bash +cargo fmt --all +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test +``` + +CI runs those on Linux, macOS and Windows, plus an end-to-end suite against a +real [soulfind](https://github.com/soulfind-dev/soulfind) server. The e2e suites +skip when no server is around, so `cargo test` stays green without one. +[Development](./README.md#development) has the setup if you want to run them. + +AI-assisted changes are fine, and no need to say so. Keep the trailers out of +the commit message: no `Co-Authored-By:` for a model, no session links. + +## Issues + +The command you ran, what you expected, and what happened is plenty. +`RUST_LOG=trace` output helps. Don't worry about polishing it. + +Built something on `soulseek-rs-lib`? Open a PR adding it to the README, and +reserve a client minor version in +[#12](https://github.com/michel/soulseek-rs/issues/12) so it's identifiable on +the network. + +Contributions are under MIT, like the rest of the project. diff --git a/README.md b/README.md index cfea6a3..3628402 100644 --- a/README.md +++ b/README.md @@ -646,6 +646,12 @@ soulfind instead of skipping. points `SOULFIND_BIN` at it, and runs both the library and CLI e2e suites with `SOULSEEK_E2E_REQUIRED=1`, so a missing server fails instead of skipping. +## Contributing + +Issues and pull requests both welcome. Branch off `develop`, keep the change +small, and see [CONTRIBUTING.md](./CONTRIBUTING.md) for the commit convention +and the checks CI runs. + ## License MIT. See [LICENSE](./LICENSE). diff --git a/soulseek-rs-lib/examples/stress.rs b/soulseek-rs-lib/examples/stress.rs index e51eb19..a14f750 100644 --- a/soulseek-rs-lib/examples/stress.rs +++ b/soulseek-rs-lib/examples/stress.rs @@ -316,7 +316,9 @@ impl Seeder { Metrics::bump(&self.metrics.mocks_offline); return; }; - let Ok(listener) = TcpListener::bind(("127.0.0.1", port)) else { + // All interfaces: soulfind reports the host's LAN address for a + // loopback login, and the client dials that. + let Ok(listener) = TcpListener::bind(("0.0.0.0", port)) else { Metrics::bump(&self.metrics.mocks_offline); return; }; @@ -415,7 +417,7 @@ impl SeederState { }) .collect(); let response = - build_file_search_response(&self.name, token, &entries, 1, 0); + build_file_search_response(&self.name, token, &entries, 1, 0, 0); let Ok(mut guard) = self.conn.lock() else { return; @@ -481,7 +483,7 @@ impl SeederState { let Ok(init) = read_framed(&mut p) else { return; }; - if init.get_message_code() != 1 { + if init.get_init_code() != 1 { return; } } @@ -630,7 +632,9 @@ impl Leecher { Metrics::bump(&self.metrics.mocks_offline); return; }; - let Ok(listener) = TcpListener::bind(("127.0.0.1", port)) else { + // All interfaces: soulfind reports the host's LAN address for a + // loopback login, and the client dials that. + let Ok(listener) = TcpListener::bind(("0.0.0.0", port)) else { Metrics::bump(&self.metrics.mocks_offline); return; }; @@ -735,7 +739,7 @@ fn accept_file_connection( let Ok(mut init) = read_framed(&mut stream) else { continue; }; - if init.get_message_code() != 1 { + if init.get_init_code() != 1 { continue; } init.set_pointer(5); diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 821af9f..faf21e2 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -2,10 +2,11 @@ use crate::actor::{Actor, ActorHandle, ConnectionState}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::peer::{ - FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueRequest, - PlaceInQueueResponse, QueueUploadHandler, SharedDirectory, - SharedFileListResponseHandler, TransferRequest, TransferResponse, - UploadDeniedHandler, UploadFailedHandler, + FileSearchResponse, FolderContentsRequest, GetShareFileList, PeerInit, + PlaceInQueueRequest, PlaceInQueueResponse, QueueUploadHandler, + SharedDirectory, SharedFileListResponseHandler, TransferRequest, + TransferResponse, UploadDeniedHandler, UploadFailedHandler, + UserInfoRequest, }; use crate::message::server::MessageFactory; use crate::message::{Handlers, Message, MessageReader, MessageType}; @@ -46,6 +47,13 @@ pub enum PeerMessage { ShareListRequested, /// A peer we are browsing sent us their shared-file listing (code 5). ShareListReceived(Vec), + /// A peer asked what we say about ourselves (they sent us code 15). + UserInfoRequested, + /// A peer asked for one folder of our shares (they sent us code 36). + FolderContentsRequested { + token: u32, + folder: String, + }, /// Offer the queued file to that peer: send an upload TransferRequest. ServeUpload { token: u32, @@ -178,6 +186,8 @@ impl PeerActor { handlers.register_handler(TransferRequest); handlers.register_handler(TransferResponse); handlers.register_handler(GetShareFileList); + handlers.register_handler(FolderContentsRequest); + handlers.register_handler(UserInfoRequest); handlers.register_handler(UploadDeniedHandler); handlers.register_handler(UploadFailedHandler); handlers.register_handler(PlaceInQueueRequest); @@ -268,10 +278,27 @@ impl PeerActor { self.handle_serve_upload(token, filename, size); } PeerMessage::ShareListRequested => { - self.handle_share_list_requested(); + self.forward(ClientOperation::ShareListRequested { + requester_key: self.peer_username(), + }); } PeerMessage::ShareListReceived(directories) => { - self.handle_share_list_received(directories); + self.forward(ClientOperation::BrowseResult { + username: self.peer_username(), + directories, + }); + } + PeerMessage::UserInfoRequested => { + self.forward(ClientOperation::UserInfoRequested { + requester_key: self.peer_username(), + }); + } + PeerMessage::FolderContentsRequested { token, folder } => { + self.forward(ClientOperation::FolderContentsRequested { + requester_key: self.peer_username(), + token, + folder, + }); } PeerMessage::RequestTransfer(download) => { let message = MessageFactory::build_transfer_request_message( @@ -431,25 +458,9 @@ impl PeerActor { self.send_message(message); } - fn handle_share_list_requested(&self) { - let requester_key = self.peer_username(); - if let Err(e) = self - .client_channel - .send(ClientOperation::ShareListRequested { requester_key }) - { - error!("[peer_actor] forward ShareListRequested: {}", e); - } - } - - fn handle_share_list_received(&self, directories: Vec) { - let username = self.peer_username(); - if let Err(e) = - self.client_channel.send(ClientOperation::BrowseResult { - username, - directories, - }) - { - error!("[peer_actor] forward BrowseResult: {}", e); + fn forward(&self, operation: ClientOperation) { + if let Err(e) = self.client_channel.send(operation) { + error!("[peer_actor] forward to client: {}", e); } } @@ -535,7 +546,7 @@ impl PeerActor { message .get_message_name( MessageType::Peer, - u32::from(message.get_message_code()) + message.get_message_code() ) .map_err(|e| e.to_string()) ); diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 01b48d2..1f2620b 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -1,6 +1,7 @@ use crate::actor::{Actor, ActorHandle, ConnectionState}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; +use crate::message::server::AdminMessageHandler; use crate::message::server::CheckPrivilegesHandler; use crate::message::server::ConnectToPeerHandler; use crate::message::server::ExcludedSearchPhrasesHandler; @@ -41,6 +42,9 @@ use std::time::{Duration, Instant}; use crate::{SoulseekRs, debug, error, trace, warn}; +mod types; +pub use types::{Context, PeerAddress, UserMessage}; + /// Ceiling on the wait for the server's login verdict. A loaded server can /// take seconds to answer, so this stays inside the caller's own 45s bound /// rather than undercutting it. @@ -48,103 +52,6 @@ const LOGIN_VERDICT_TIMEOUT: Duration = Duration::from_secs(30); const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); -#[derive(Debug, Clone)] -pub struct PeerAddress { - host: String, - port: u16, -} - -impl PeerAddress { - #[must_use] - pub const fn new(host: String, port: u16) -> Self { - Self { host, port } - } - - #[must_use] - pub fn get_host(&self) -> &str { - &self.host - } - - #[must_use] - pub const fn get_port(&self) -> u16 { - self.port - } -} - -impl std::fmt::Display for PeerAddress { - fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { - write!(f, "{}:{}", self.host, self.port) - } -} - -#[derive(Debug, Default)] -pub struct Context { - pub logged_in: Option, -} - -impl Context { - #[must_use] - pub fn new() -> Self { - Self::default() - } -} -#[derive(Debug, Clone)] -pub struct UserMessage { - id: u32, - timestamp: u32, - username: String, - message: String, - new_message: bool, -} -impl UserMessage { - #[must_use] - pub const fn new( - id: u32, - timestamp: u32, - username: String, - message: String, - new_message: bool, - ) -> Self { - Self { - id, - timestamp, - username, - message, - new_message, - } - } - /// The server-assigned id of this message (used to acknowledge it). - #[must_use] - pub const fn id(&self) -> u32 { - self.id - } - - /// Unix timestamp the server recorded for this message. - #[must_use] - pub const fn timestamp(&self) -> u32 { - self.timestamp - } - - /// The username of the sender. - #[must_use] - pub fn username(&self) -> &str { - &self.username - } - - /// The message body. - #[must_use] - pub fn message(&self) -> &str { - &self.message - } - - /// Whether the server flagged this as freshly delivered (as opposed to a - /// message replayed because it was queued while the recipient was offline). - #[must_use] - pub const fn is_new(&self) -> bool { - self.new_message - } -} - /// The server actor's mailbox. Marked non-exhaustive: each protocol message /// the client learns adds a variant, and that must not break callers. #[derive(Debug, Clone)] @@ -330,7 +237,7 @@ impl ServerActor { } fn initiate_connection(&mut self) -> bool { - let stream = (self.address.host.as_str(), self.address.port) + let stream = (self.address.get_host(), self.address.get_port()) .to_socket_addrs() .ok() .and_then(|mut addrs| { @@ -350,6 +257,9 @@ impl ServerActor { return false; } stream.set_nodelay(true).ok(); + if let Err(e) = crate::utils::keepalive::set_keepalive(&stream) { + warn!("[server] keepalive not set: {}", e); + } self.stream = Some(stream); self.connection_state = ConnectionState::Connecting { @@ -380,6 +290,7 @@ impl ServerActor { handlers.register_handler(LoginHandler); handlers.register_handler(ReloggedHandler); + handlers.register_handler(AdminMessageHandler); handlers.register_handler(RoomListHandler); handlers.register_handler(GetUserStatusHandler); handlers.register_handler(WatchUserHandler); @@ -474,6 +385,36 @@ impl ServerActor { ServerMessage::PrivateMessageReceived(user_message) => { self.handle_private_message_received(user_message); } + ServerMessage::ProcessRead => { + self.process_read(); + } + ServerMessage::Login { + username, + password, + version, + response, + } => { + self.handle_login(username, password, version, response); + } + ServerMessage::FileSearch { token, query } => { + self.file_search(token, &query); + } + ServerMessage::FileSearchRequest { + username, + token, + query, + } => { + self.handle_file_search_request(username, token, query); + } + other => self.handle_social_message(other), + } + } + + /// Rooms and users (codes 7, 5, 36, 13-17, 64): forwarded as room events + /// and user status or stats. Split out, like the standing traffic below, + /// only to keep `handle_message`'s match readable. + fn handle_social_message(&mut self, message: ServerMessage) { + match message { ServerMessage::RoomListReceived(rooms) => { self.forward_room_event(RoomEvent::List(rooms)); } @@ -550,27 +491,6 @@ impl ServerActor { ServerMessage::RoomUserLeft { room, username } => { self.forward_room_event(RoomEvent::UserLeft { room, username }); } - ServerMessage::ProcessRead => { - self.process_read(); - } - ServerMessage::Login { - username, - password, - version, - response, - } => { - self.handle_login(username, password, version, response); - } - ServerMessage::FileSearch { token, query } => { - self.file_search(token, &query); - } - ServerMessage::FileSearchRequest { - username, - token, - query, - } => { - self.handle_file_search_request(username, token, query); - } other => self.handle_standing_message(other), } } @@ -803,7 +723,7 @@ impl ServerActor { message .get_message_name( MessageType::Server, - u32::from(message.get_message_code()) + message.get_message_code() ) .map_err(|e| e.to_string()) ); @@ -998,205 +918,4 @@ impl Actor for ServerActor { } #[cfg(test)] -mod tests { - use super::*; - - fn code_of(message: &Message) -> u32 { - u32::from_le_bytes(message.get_data()[0..4].try_into().unwrap()) - } - - #[test] - fn a_timed_out_connect_parks_the_actor_in_disconnected() { - let mut actor = parked_actor(1); - actor.connection_state = ConnectionState::Connecting { - since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(), - }; - - actor.tick(); - - assert!( - matches!(actor.connection_state, ConnectionState::Disconnected), - "a timed-out connect must leave Connecting" - ); - } - - fn parked_actor(port: u16) -> ServerActor { - let (tx, _rx) = std::sync::mpsc::channel(); - ServerActor::new( - PeerAddress::new("127.0.0.1".to_string(), port), - tx, - 0, - false, - 0, - 0, - ) - } - - #[test] - fn dropping_the_connection_forgets_the_dispatcher_and_the_verdict() { - let mut actor = parked_actor(1); - actor.initialize_dispatcher(); - actor.connection_state = ConnectionState::Connected; - actor.handle_login_status(true); - - actor.disconnect_with_error(); - - assert_eq!(actor.session.loss(), Some(SessionLoss::Disconnected)); - assert!( - actor.dispatcher_sender.is_none(), - "a login queued now must wait in the buffer, not in a channel \ - the next connection replaces" - ); - assert_eq!( - actor.context.read().unwrap().logged_in, - None, - "the old verdict must not answer the next login" - ); - } - - #[test] - fn disconnect_clears_a_partial_frame_so_the_next_session_reframes_clean() { - use std::io::Write; - - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let addr = listener.local_addr().unwrap(); - let client = TcpStream::connect(addr).unwrap(); - let (mut server_side, _) = listener.accept().unwrap(); - let partial_frame = [10u8, 0, 0, 0, 1, 2, 3]; - server_side.write_all(&partial_frame).unwrap(); - server_side.flush().unwrap(); - client.set_nonblocking(true).unwrap(); - - let mut actor = parked_actor(addr.port()); - actor.stream = Some(client); - actor.connection_state = ConnectionState::Connected; - for _ in 0..50 { - actor.process_read(); - if actor.reader.buffer_len() > 0 { - break; - } - std::thread::sleep(Duration::from_millis(10)); - } - assert!( - actor.reader.buffer_len() > 0, - "an incomplete frame should be waiting for its rest" - ); - - actor.disconnect(); - - assert_eq!( - actor.reader.buffer_len(), - 0, - "a stale partial frame carried across a reconnect reads the next \ - session's bytes as a length prefix and wedges recovery" - ); - } - - #[test] - fn a_login_on_a_parked_actor_dials_again_and_queues_the_login() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let mut actor = parked_actor(listener.local_addr().unwrap().port()); - actor.session.record(SessionLoss::Disconnected); - let (response, _verdict) = std::sync::mpsc::channel(); - - actor.handle_login( - "u".into(), - "p".into(), - ClientVersion::default(), - response, - ); - - assert!( - matches!( - actor.connection_state, - ConnectionState::Connecting { .. } - ), - "a parked actor must dial again on login" - ); - assert_eq!( - actor.queued_messages.len(), - 1, - "the login waits for the connection to come up" - ); - } - - #[test] - fn a_parked_actor_handles_login_rather_than_queueing_it() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let mut actor = parked_actor(listener.local_addr().unwrap().port()); - actor.session.record(SessionLoss::Disconnected); - let (response, _verdict) = std::sync::mpsc::channel(); - - actor.handle_message(ServerMessage::Login { - username: "u".into(), - password: "p".into(), - version: ClientVersion::default(), - response, - }); - - assert!( - matches!( - actor.connection_state, - ConnectionState::Connecting { .. } - ), - "a parked actor must act on a login, not park it in the queue" - ); - assert!( - actor - .queued_messages - .iter() - .all(|m| !matches!(m, ServerMessage::Login { .. })), - "the login must not be sitting unhandled in the queue" - ); - } - - #[test] - fn a_login_with_nobody_listening_is_refused_at_once() { - let mut actor = parked_actor(1); - let (response, verdict) = std::sync::mpsc::channel(); - - actor.handle_login( - "u".into(), - "p".into(), - ClientVersion::default(), - response, - ); - - assert!( - matches!( - verdict.recv_timeout(Duration::from_secs(5)), - Ok(Err(SoulseekRs::NotConnected)) - ), - "a refused dial must not make the caller wait out the verdict \ - timeout" - ); - } - - #[test] - fn a_successful_login_marks_the_session_live_again() { - let mut actor = parked_actor(1); - actor.session.record(SessionLoss::Disconnected); - - actor.handle_login_status(true); - - assert_eq!(actor.session.loss(), None); - } - - #[test] - fn post_login_messages_carry_counts_and_conditional_wait_port() { - let messages = post_login_messages(true, 4321, 3, 7); - let codes: Vec = messages.iter().map(code_of).collect(); - // SharedFolders, HaveNoParent, SetStatus, SetWaitPort. - assert_eq!(codes, vec![35, 71, 28, 2]); - - // The SharedFolders message (code 35) carries the real counts. - let shared = messages[0].get_data(); - assert_eq!(u32::from_le_bytes(shared[4..8].try_into().unwrap()), 3); - assert_eq!(u32::from_le_bytes(shared[8..12].try_into().unwrap()), 7); - - // Not listening omits SetWaitPort (code 2). - let no_listen = post_login_messages(false, 4321, 3, 7); - let codes: Vec = no_listen.iter().map(code_of).collect(); - assert_eq!(codes, vec![35, 71, 28]); - } -} +mod tests; diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs new file mode 100644 index 0000000..bf73bbe --- /dev/null +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -0,0 +1,210 @@ +use super::*; + +fn code_of(message: &Message) -> u32 { + u32::from_le_bytes(message.get_data()[0..4].try_into().unwrap()) +} + +#[test] +fn a_timed_out_connect_parks_the_actor_in_disconnected() { + let mut actor = parked_actor(1); + actor.connection_state = ConnectionState::Connecting { + since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(), + }; + + actor.tick(); + + assert!( + matches!(actor.connection_state, ConnectionState::Disconnected), + "a timed-out connect must leave Connecting" + ); +} + +#[cfg(unix)] +#[test] +fn the_server_socket_keeps_itself_alive() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut actor = parked_actor(listener.local_addr().unwrap().port()); + + assert!(actor.initiate_connection()); + + assert!( + crate::utils::keepalive::keepalive_enabled( + actor.stream.as_ref().unwrap() + ), + "an idle NAT mapping must be kept alive and a dead path noticed" + ); +} + +fn parked_actor(port: u16) -> ServerActor { + let (tx, _rx) = std::sync::mpsc::channel(); + ServerActor::new( + PeerAddress::new("127.0.0.1".to_string(), port), + tx, + 0, + false, + 0, + 0, + ) +} + +#[test] +fn dropping_the_connection_forgets_the_dispatcher_and_the_verdict() { + let mut actor = parked_actor(1); + actor.initialize_dispatcher(); + actor.connection_state = ConnectionState::Connected; + actor.handle_login_status(true); + + actor.disconnect_with_error(); + + assert_eq!(actor.session.loss(), Some(SessionLoss::Disconnected)); + assert!( + actor.dispatcher_sender.is_none(), + "a login queued now must wait in the buffer, not in a channel \ + the next connection replaces" + ); + assert_eq!( + actor.context.read().unwrap().logged_in, + None, + "the old verdict must not answer the next login" + ); +} + +#[test] +fn disconnect_clears_a_partial_frame_so_the_next_session_reframes_clean() { + use std::io::Write; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let client = TcpStream::connect(addr).unwrap(); + let (mut server_side, _) = listener.accept().unwrap(); + let partial_frame = [10u8, 0, 0, 0, 1, 2, 3]; + server_side.write_all(&partial_frame).unwrap(); + server_side.flush().unwrap(); + client.set_nonblocking(true).unwrap(); + + let mut actor = parked_actor(addr.port()); + actor.stream = Some(client); + actor.connection_state = ConnectionState::Connected; + for _ in 0..50 { + actor.process_read(); + if actor.reader.buffer_len() > 0 { + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + actor.reader.buffer_len() > 0, + "an incomplete frame should be waiting for its rest" + ); + + actor.disconnect(); + + assert_eq!( + actor.reader.buffer_len(), + 0, + "a stale partial frame carried across a reconnect reads the next \ + session's bytes as a length prefix and wedges recovery" + ); +} + +#[test] +fn a_login_on_a_parked_actor_dials_again_and_queues_the_login() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut actor = parked_actor(listener.local_addr().unwrap().port()); + actor.session.record(SessionLoss::Disconnected); + let (response, _verdict) = std::sync::mpsc::channel(); + + actor.handle_login( + "u".into(), + "p".into(), + ClientVersion::default(), + response, + ); + + assert!( + matches!(actor.connection_state, ConnectionState::Connecting { .. }), + "a parked actor must dial again on login" + ); + assert_eq!( + actor.queued_messages.len(), + 1, + "the login waits for the connection to come up" + ); +} + +#[test] +fn a_parked_actor_handles_login_rather_than_queueing_it() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let mut actor = parked_actor(listener.local_addr().unwrap().port()); + actor.session.record(SessionLoss::Disconnected); + let (response, _verdict) = std::sync::mpsc::channel(); + + actor.handle_message(ServerMessage::Login { + username: "u".into(), + password: "p".into(), + version: ClientVersion::default(), + response, + }); + + assert!( + matches!(actor.connection_state, ConnectionState::Connecting { .. }), + "a parked actor must act on a login, not park it in the queue" + ); + assert!( + actor + .queued_messages + .iter() + .all(|m| !matches!(m, ServerMessage::Login { .. })), + "the login must not be sitting unhandled in the queue" + ); +} + +#[test] +fn a_login_with_nobody_listening_is_refused_at_once() { + let mut actor = parked_actor(1); + let (response, verdict) = std::sync::mpsc::channel(); + + actor.handle_login( + "u".into(), + "p".into(), + ClientVersion::default(), + response, + ); + + assert!( + matches!( + verdict.recv_timeout(Duration::from_secs(5)), + Ok(Err(SoulseekRs::NotConnected)) + ), + "a refused dial must not make the caller wait out the verdict \ + timeout" + ); +} + +#[test] +fn a_successful_login_marks_the_session_live_again() { + let mut actor = parked_actor(1); + actor.session.record(SessionLoss::Disconnected); + + actor.handle_login_status(true); + + assert_eq!(actor.session.loss(), None); +} + +#[test] +fn post_login_messages_carry_counts_and_conditional_wait_port() { + let messages = post_login_messages(true, 4321, 3, 7); + let codes: Vec = messages.iter().map(code_of).collect(); + // SharedFolders, HaveNoParent, SetStatus, SetWaitPort. + assert_eq!(codes, vec![35, 71, 28, 2]); + + // The SharedFolders message (code 35) carries the real counts. + let shared = messages[0].get_data(); + assert_eq!(u32::from_le_bytes(shared[4..8].try_into().unwrap()), 3); + assert_eq!(u32::from_le_bytes(shared[8..12].try_into().unwrap()), 7); + + // Not listening omits SetWaitPort (code 2). + let no_listen = post_login_messages(false, 4321, 3, 7); + let codes: Vec = no_listen.iter().map(code_of).collect(); + assert_eq!(codes, vec![35, 71, 28]); +} diff --git a/soulseek-rs-lib/src/actor/server_actor/types.rs b/soulseek-rs-lib/src/actor/server_actor/types.rs new file mode 100644 index 0000000..51667c8 --- /dev/null +++ b/soulseek-rs-lib/src/actor/server_actor/types.rs @@ -0,0 +1,99 @@ +//! The values the server actor hands out: where a peer listens, a private +//! message, and the login context. + +#[derive(Debug, Clone)] +pub struct PeerAddress { + host: String, + port: u16, +} + +impl PeerAddress { + #[must_use] + pub const fn new(host: String, port: u16) -> Self { + Self { host, port } + } + + #[must_use] + pub fn get_host(&self) -> &str { + &self.host + } + + #[must_use] + pub const fn get_port(&self) -> u16 { + self.port + } +} + +impl std::fmt::Display for PeerAddress { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + write!(f, "{}:{}", self.host, self.port) + } +} + +#[derive(Debug, Default)] +pub struct Context { + pub logged_in: Option, +} + +impl Context { + #[must_use] + pub fn new() -> Self { + Self::default() + } +} +#[derive(Debug, Clone)] +pub struct UserMessage { + id: u32, + timestamp: u32, + username: String, + message: String, + new_message: bool, +} +impl UserMessage { + #[must_use] + pub const fn new( + id: u32, + timestamp: u32, + username: String, + message: String, + new_message: bool, + ) -> Self { + Self { + id, + timestamp, + username, + message, + new_message, + } + } + /// The server-assigned id of this message (used to acknowledge it). + #[must_use] + pub const fn id(&self) -> u32 { + self.id + } + + /// Unix timestamp the server recorded for this message. + #[must_use] + pub const fn timestamp(&self) -> u32 { + self.timestamp + } + + /// The username of the sender. + #[must_use] + pub fn username(&self) -> &str { + &self.username + } + + /// The message body. + #[must_use] + pub fn message(&self) -> &str { + &self.message + } + + /// Whether the server flagged this as freshly delivered (as opposed to a + /// message replayed because it was queued while the recipient was offline). + #[must_use] + pub const fn is_new(&self) -> bool { + self.new_message + } +} diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 3584fbc..d4098d9 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -17,7 +17,7 @@ use crate::{ peer::{ConnectionType, DownloadPeer, Peer, PeerMessage, listen::Listen}, shares::Shares, types::{Download, Search, SearchResult}, - utils::{lock::RwLockExt, md5}, + utils::lock::RwLockExt, }; use std::{ collections::{HashMap, HashSet}, @@ -90,6 +90,17 @@ fn next_download_token() -> u32 { NEXT_DOWNLOAD_TOKEN.fetch_add(1, Ordering::Relaxed) % 0x8000_0000 } +/// Source of search tokens. +/// +/// A counter, not a hash of the query: a peer's answer is routed to the first +/// search holding its token, so two queries sharing one would pour results +/// into each other. +static NEXT_SEARCH_TOKEN: AtomicU32 = AtomicU32::new(1); + +fn next_search_token() -> u32 { + NEXT_SEARCH_TOKEN.fetch_add(1, Ordering::Relaxed) +} + /// A file we have agreed to serve to a peer, awaiting their TransferResponse. struct UploadJob { downloader: String, @@ -139,6 +150,9 @@ fn build_search_response( own_username: &str, token: u32, query: &str, + free_slot: bool, + speed: u32, + queue_length: u32, ) -> Option { let matches = shares.search(query); if matches.is_empty() { @@ -146,6 +160,7 @@ fn build_search_response( } let entries: Vec = matches .iter() + .take(crate::types::MAX_SEARCH_REPLY_FILES) .map(|f| FileEntry { name: &f.virtual_path, size: f.size, @@ -156,8 +171,9 @@ fn build_search_response( own_username, token, &entries, - 1, - 0, + u8::from(free_slot), + speed, + queue_length, )) } @@ -276,6 +292,16 @@ pub enum ClientOperation { ShareListRequested { requester_key: String, }, + /// A peer asked what we say about ourselves; send a UserInfoResponse. + UserInfoRequested { + requester_key: String, + }, + /// A peer asked for one folder of our shares; send a FolderContentsResponse. + FolderContentsRequested { + requester_key: String, + token: u32, + folder: String, + }, /// A peer we are browsing returned their shared-file listing. BrowseResult { username: String, @@ -364,6 +390,9 @@ pub struct ClientContext { upload_seq: u64, /// How many uploads may be in flight at once. upload_slots: usize, + /// Bytes per second of the last completed upload, advertised in search + /// replies; zero until one has finished. + last_upload_speed: u32, /// Queued-upload states that came and went between two polls of /// [`Client::uploads`]. A caller sampling that snapshot would otherwise /// never see a peer that queued and was served inside one poll interval, @@ -446,6 +475,7 @@ impl ClientContext { upload_queue: Vec::new(), upload_seq: 0, upload_slots: DEFAULT_UPLOAD_SLOTS, + last_upload_speed: 0, upload_events: Vec::new(), downloads: DownloadStore::new(), actor_system, diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index bcd2cc7..1bc662c 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -593,6 +593,9 @@ impl Client { &own_username, token, &query, + ctx.has_free_upload_slot(), + ctx.last_upload_speed, + ctx.upload_queue.len() as u32, ), Err(e) => { error!("[client] IncomingSearch read: {}", e); @@ -753,35 +756,49 @@ impl Client { } } ClientOperation::ShareListRequested { requester_key } => { - // Reply with our full shared-file listing. - let (registry, message) = match client_context - .read_safe() - { - Ok(ctx) => { - let dirs = ctx - .shares - .directories() - .into_iter() - .map(|(name, files)| { - crate::message::peer::SharedDirectory { - name, - files, - } - }) - .collect::>(); - ( - ctx.peer_registry.clone(), - crate::message::peer::build_shared_file_list(&dirs), + Self::reply_to_peer( + &client_context, + &requester_key, + |ctx| { + crate::message::peer::build_shared_file_list( + &shared_directories(ctx), ) - } - Err(_) => continue, - }; - if let Some(registry) = registry { - let _ = registry.send_to_peer( - &requester_key, - PeerMessage::SendMessage(message), - ); - } + }, + ); + } + ClientOperation::UserInfoRequested { requester_key } => { + Self::reply_to_peer( + &client_context, + &requester_key, + |ctx| { + crate::message::peer::build_user_info( + ctx.upload_slots as u32, + ctx.upload_queue.len() as u32, + ctx.has_free_upload_slot(), + ) + }, + ); + } + ClientOperation::FolderContentsRequested { + requester_key, + token, + folder, + } => { + // ponytail: walks the whole share for one folder, as a + // browse does; index by directory if it ever shows up. + Self::reply_to_peer( + &client_context, + &requester_key, + |ctx| { + let dirs: Vec<_> = shared_directories(ctx) + .into_iter() + .filter(|dir| dir.name == folder) + .collect(); + crate::message::peer::build_folder_contents( + token, &folder, &dirs, + ) + }, + ); } ClientOperation::BrowseResult { username, @@ -842,6 +859,22 @@ impl Client { }); } + /// Answer a peer with the message `build` derives from the client state. + fn reply_to_peer( + client_context: &Arc>, + requester_key: &str, + build: impl FnOnce(&ClientContext) -> crate::message::Message, + ) { + let (registry, message) = match client_context.read_safe() { + Ok(ctx) => (ctx.peer_registry.clone(), build(&ctx)), + Err(_) => return, + }; + if let Some(registry) = registry { + let _ = registry + .send_to_peer(requester_key, PeerMessage::SendMessage(message)); + } + } + fn sweep_expired_connects(client_context: &Arc>) { let expired = match client_context.write_safe() { Ok(mut ctx) => ctx.take_expired_connects(Instant::now()), @@ -852,3 +885,17 @@ impl Client { } } } + +/// Our shares in the wire form a browse or folder reply carries. +fn shared_directories( + ctx: &ClientContext, +) -> Vec { + ctx.shares + .directories() + .into_iter() + .map(|(name, files)| crate::message::peer::SharedDirectory { + name, + files, + }) + .collect() +} diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index ebcac80..215deb5 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -1,7 +1,7 @@ use super::{ Arc, AtomicBool, Client, DEFAULT_WISHLIST_INTERVAL, Duration, HashMap, Instant, Ordering, Result, RwLockExt, Search, SearchResult, ServerMessage, - SoulseekRs, info, md5, sleep, + SoulseekRs, info, next_search_token, sleep, }; impl Client { @@ -62,8 +62,7 @@ impl Client { let Some(handle) = &self.server_handle else { return Err(SoulseekRs::NotConnected); }; - let hash = md5::md5(query); - let token = u32::from_str_radix(&hash[0..5], 16)?; + let token = next_search_token(); self.context.write_safe()?.searches.insert( query.to_string(), diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 73e892a..031edb2 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -160,8 +160,9 @@ fn build_search_response_matches_shares_and_echoes_token() { std::fs::write(dir.join("probe_xyzzy.bin"), b"data").unwrap(); let shares = Shares::scan(&dir).unwrap(); - let response = build_search_response(&shares, "me", 99, "xyzzy") - .expect("a matching share yields a response"); + let response = + build_search_response(&shares, "me", 99, "xyzzy", true, 0, 0) + .expect("a matching share yields a response"); let mut decoded = crate::message::Message::new_with_data(response.get_buffer()); decoded.set_pointer(8); @@ -170,7 +171,10 @@ fn build_search_response_matches_shares_and_echoes_token() { assert_eq!(result.token, 99); assert!(result.files.iter().any(|f| f.name.contains("probe_xyzzy"))); - assert!(build_search_response(&shares, "me", 1, "nomatch").is_none()); + assert!( + build_search_response(&shares, "me", 1, "nomatch", true, 0, 0) + .is_none() + ); let _ = std::fs::remove_dir_all(dir); } diff --git a/soulseek-rs-lib/src/client/upload_queue.rs b/soulseek-rs-lib/src/client/upload_queue.rs index 9bc03b9..474c860 100644 --- a/soulseek-rs-lib/src/client/upload_queue.rs +++ b/soulseek-rs-lib/src/client/upload_queue.rs @@ -123,7 +123,7 @@ impl ClientContext { mut next_token: impl FnMut() -> u32, ) -> (Option, Vec) { let mut offers = Vec::new(); - while self.uploads_in_flight() < self.upload_slots { + while self.has_free_upload_slot() { let Some(index) = next_to_serve(&self.upload_queue) else { break; }; @@ -190,6 +190,10 @@ impl ClientContext { .count() } + pub(super) fn has_free_upload_slot(&self) -> bool { + self.uploads_in_flight() < self.upload_slots + } + /// The peers still waiting, in the order they will be served, each carrying /// its own 1-based place. pub(super) fn queued_uploads(&self) -> Vec { diff --git a/soulseek-rs-lib/src/client/uploads.rs b/soulseek-rs-lib/src/client/uploads.rs index cd42654..860a5f9 100644 --- a/soulseek-rs-lib/src/client/uploads.rs +++ b/soulseek-rs-lib/src/client/uploads.rs @@ -1,7 +1,9 @@ use super::{ ActiveUpload, Arc, Client, ClientContext, DownloadStatus, PeerMessage, - RwLock, RwLockExt, collect_failed_tokens, error, next_upload_token, thread, + RwLock, RwLockExt, ServerMessage, collect_failed_tokens, error, + next_upload_token, thread, }; +use crate::message::server::MessageFactory; use crate::types::UploadStatus; use std::sync::atomic::{AtomicBool, AtomicU64}; @@ -145,8 +147,9 @@ impl Client { &bytes_sent, &cancel, ); + let streamed = result.as_ref().ok().copied(); let status = match &result { - Ok(()) => UploadStatus::Completed, + Ok(_) => UploadStatus::Completed, Err(e) if e.kind() == std::io::ErrorKind::Interrupted => { UploadStatus::Cancelled } @@ -155,10 +158,26 @@ impl Client { UploadStatus::Failed(e.to_string()) } }; + let mut report = None; if let Ok(mut ctx) = context.write_safe() && let Some(upload) = ctx.active_uploads.get_mut(&token) { + let secs = upload.started.elapsed().as_secs_f64(); upload.status = status; + if let Some(bytes) = streamed + && secs > 0.0 + { + let speed = (bytes as f64 / secs) as u32; + ctx.last_upload_speed = speed; + report = ctx.server_sender.clone().map(|s| (s, speed)); + } + } + // The server folds each finished upload's rate into the average + // speed it shows other users for us. + if let Some((server, speed)) = report { + let _ = server.send(ServerMessage::SendMessage( + MessageFactory::build_send_upload_speed(speed), + )); } // The slot this transfer held is free now, so whoever is next in // line gets it without waiting for another request to arrive. diff --git a/soulseek-rs-lib/src/dispatcher.rs b/soulseek-rs-lib/src/dispatcher.rs index 9c50d80..916771c 100644 --- a/soulseek-rs-lib/src/dispatcher.rs +++ b/soulseek-rs-lib/src/dispatcher.rs @@ -37,3 +37,40 @@ impl MessageDispatcher { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::MessageHandler; + use std::sync::mpsc::channel; + + struct Echo(u32); + + impl MessageHandler for Echo { + fn get_code(&self) -> u32 { + self.0 + } + fn handle(&self, _: &mut Message, sender: Sender) { + let _ = sender.send(self.0); + } + } + + fn framed(code: u32) -> Message { + Message::new_with_data(Message::new().write_int32(code).get_buffer()) + } + + #[test] + fn a_code_above_255_is_not_confused_with_its_low_byte() { + let (sender, received) = channel(); + let mut handlers = Handlers::new(); + handlers.register_handler(Echo(1)); + handlers.register_handler(Echo(1001)); + let dispatcher = + MessageDispatcher::new("test".to_string(), sender, handlers); + + dispatcher.dispatch(&mut framed(257)); + dispatcher.dispatch(&mut framed(1001)); + + assert_eq!(received.try_iter().collect::>(), [1001]); + } +} diff --git a/soulseek-rs-lib/src/message/handlers.rs b/soulseek-rs-lib/src/message/handlers.rs index 03fc764..1eb8ee5 100644 --- a/soulseek-rs-lib/src/message/handlers.rs +++ b/soulseek-rs-lib/src/message/handlers.rs @@ -4,11 +4,11 @@ use crate::message::Message; use std::sync::mpsc::Sender; pub trait MessageHandler: Send { - fn get_code(&self) -> u8; + fn get_code(&self) -> u32; fn handle(&self, message: &mut Message, sender: Sender); } pub struct Handlers { - handlers: HashMap + Send>>, + handlers: HashMap + Send>>, } impl Default for Handlers { @@ -35,7 +35,7 @@ impl Handlers { #[must_use] pub fn get_handler( &self, - code: u8, + code: u32, ) -> Option<&(dyn MessageHandler + Send)> { self.handlers.get(&code).map(|v| &**v) } diff --git a/soulseek-rs-lib/src/message/mod.rs b/soulseek-rs-lib/src/message/mod.rs index df62c8f..28bdf53 100644 --- a/soulseek-rs-lib/src/message/mod.rs +++ b/soulseek-rs-lib/src/message/mod.rs @@ -51,21 +51,19 @@ impl Message { } } - #[allow(dead_code)] #[must_use] - pub fn get_message_code_u32(&self) -> u32 { + pub fn get_message_code(&self) -> u32 { if self.data.len() < 8 { return 0; } self.data[4..8].try_into().map_or(0, u32::from_le_bytes) } + /// The one-byte code of a peer init frame (PierceFirewall 0, PeerInit 1), + /// the only messages whose code is not four bytes wide. #[must_use] - pub fn get_message_code(&self) -> u8 { - if self.data.len() <= 4 { - return 0; - } - self.data[4] + pub fn get_init_code(&self) -> u8 { + self.data.get(4).copied().unwrap_or(0) } #[must_use] @@ -254,6 +252,7 @@ impl Message { 100 => Ok("AcceptChildren"), 102 => Ok("PossibleParents"), 104 => Ok("WishlistInterval"), + 121 => Ok("SendUploadSpeed"), 160 => Ok("ExcludedSearchPhrases"), 1001 => Ok("CantConnectToPeer"), _ => Err(Error(format!("Unknown server message code: {code}"))), @@ -388,10 +387,19 @@ fn read_string_with_missing_length_prefix_does_not_panic() { #[test] fn get_message_code_on_short_message_does_not_panic() { // A zero-length frame drains to exactly the 4-byte length prefix, leaving - // no room for a message code at data[4]. + // no room for a code after it. let msg = Message::new_with_data(vec![0, 0, 0, 0]); assert_eq!(msg.get_message_code(), 0); - assert_eq!(msg.get_message_code_u32(), 0); + assert_eq!(msg.get_init_code(), 0); +} + +#[test] +fn an_init_frame_reads_its_one_byte_code() { + let init = Message::new_with_data( + Message::new().write_int8(1).write_string("ab").get_buffer(), + ); + assert_eq!(init.get_init_code(), 1); + assert_ne!(init.get_message_code(), 1); } #[test] diff --git a/soulseek-rs-lib/src/message/peer/file_search_response.rs b/soulseek-rs-lib/src/message/peer/file_search_response.rs index f6367e4..10a8134 100644 --- a/soulseek-rs-lib/src/message/peer/file_search_response.rs +++ b/soulseek-rs-lib/src/message/peer/file_search_response.rs @@ -1,7 +1,7 @@ use crate::message::{Message, MessageHandler}; use crate::peer::PeerMessage; use crate::types::SearchResult; -use crate::utils::zlib::compress_stored; +use crate::utils::zlib::deflate; use std::sync::mpsc::Sender; /// A borrowed view of one file to advertise in a search response, kept @@ -22,6 +22,7 @@ pub fn build_file_search_response( files: &[FileEntry], slots: u8, speed: u32, + queue_length: u32, ) -> Message { let mut payload = Message::new(); payload @@ -39,9 +40,12 @@ pub fn build_file_search_response( payload.write_int32(code).write_int32(value); } } - payload.write_int8(slots).write_int32(speed).write_int32(0); // free upload slots / queue length (well-formed trailer) + payload + .write_int8(slots) + .write_int32(speed) + .write_int32(queue_length); - let compressed = compress_stored(&payload.get_data()); + let compressed = deflate(&payload.get_data()); Message::new() .write_int32(9) .write_raw_bytes(compressed) @@ -50,7 +54,7 @@ pub fn build_file_search_response( pub struct FileSearchResponse; impl MessageHandler for FileSearchResponse { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 9 } fn handle(&self, message: &mut Message, sender: Sender) { @@ -112,7 +116,7 @@ fn build_file_search_response_roundtrips_through_the_decoder() { attribs: &[], }, ]; - let message = build_file_search_response("e2e_sharer", 42, &files, 1, 0); + let message = build_file_search_response("e2e_sharer", 42, &files, 1, 0, 0); // Decode via the exact production decoder used for real peer responses: // the dispatcher positions the pointer at 8 (past length + code). @@ -132,3 +136,18 @@ fn build_file_search_response_roundtrips_through_the_decoder() { assert!(result.files[1].attribs.is_empty()); assert_eq!(result.slots, 1); } + +#[test] +fn the_trailer_carries_slots_speed_and_queue_length() { + let message = build_file_search_response("me", 1, &[], 0, 900, 7); + let mut body = Message::new_with_data( + crate::utils::zlib::inflate(&message.get_data()[4..]).unwrap(), + ); + body.read_string(); + body.read_int32(); + assert_eq!(body.read_int32(), 0, "no files"); + assert_eq!( + (body.read_int8(), body.read_int32(), body.read_int32()), + (0, 900, 7) + ); +} diff --git a/soulseek-rs-lib/src/message/peer/folder_contents.rs b/soulseek-rs-lib/src/message/peer/folder_contents.rs new file mode 100644 index 0000000..93f7077 --- /dev/null +++ b/soulseek-rs-lib/src/message/peer/folder_contents.rs @@ -0,0 +1,68 @@ +//! `FolderContentsRequest` (peer code 36) and its reply (code 37): one folder +//! of a share in the listing form code 5 uses, which is what "download +//! folder" in other clients asks for. + +use crate::message::peer::SharedDirectory; +use crate::message::peer::shared_file_list::{ + decompress_body, read_directories, write_directories, +}; +use crate::message::{Message, MessageHandler}; +use crate::peer::PeerMessage; +use crate::utils::zlib::deflate; +use std::sync::mpsc::Sender; + +pub struct FolderContentsRequest; +impl MessageHandler for FolderContentsRequest { + fn get_code(&self) -> u32 { + 36 + } + fn handle(&self, message: &mut Message, sender: Sender) { + let token = message.read_int32(); + let folder = message.read_string(); + let _ = + sender.send(PeerMessage::FolderContentsRequested { token, folder }); + } +} + +/// Build a `FolderContentsResponse` (peer code 37). +#[must_use] +pub fn build_folder_contents( + token: u32, + folder: &str, + dirs: &[SharedDirectory], +) -> Message { + let mut payload = Message::new(); + payload.write_int32(token).write_string(folder); + write_directories(&mut payload, dirs); + Message::new() + .write_int32(37) + .write_raw_bytes(deflate(&payload.get_data())) + .clone() +} + +/// Parse a `FolderContentsResponse` positioned at its compressed blob, or +/// `None` when it does not inflate. +#[must_use] +pub fn parse_folder_contents( + message: &mut Message, +) -> Option<(u32, String, Vec)> { + let mut body = decompress_body(message)?; + let token = body.read_int32(); + let folder = body.read_string(); + Some((token, folder, read_directories(&mut body))) +} + +#[test] +fn a_folder_listing_round_trips() { + let dirs = vec![SharedDirectory { + name: "music\\album".to_string(), + files: vec![("one.flac".to_string(), 40), ("two.flac".to_string(), 50)], + }]; + let built = build_folder_contents(7, "music\\album", &dirs); + let mut message = Message::new_with_data(built.get_data()); + message.set_pointer(4); + assert_eq!( + parse_folder_contents(&mut message), + Some((7, "music\\album".to_string(), dirs)) + ); +} diff --git a/soulseek-rs-lib/src/message/peer/get_share_file_list.rs b/soulseek-rs-lib/src/message/peer/get_share_file_list.rs index d650be3..cd8c5de 100644 --- a/soulseek-rs-lib/src/message/peer/get_share_file_list.rs +++ b/soulseek-rs-lib/src/message/peer/get_share_file_list.rs @@ -8,7 +8,7 @@ use std::sync::mpsc::Sender; /// owns the shares) builds the real SharedFileListResponse in reply. pub struct GetShareFileList; impl MessageHandler for GetShareFileList { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 4 } fn handle(&self, _message: &mut Message, sender: Sender) { diff --git a/soulseek-rs-lib/src/message/peer/mod.rs b/soulseek-rs-lib/src/message/peer/mod.rs index 0f245d9..4c37b1a 100644 --- a/soulseek-rs-lib/src/message/peer/mod.rs +++ b/soulseek-rs-lib/src/message/peer/mod.rs @@ -1,4 +1,5 @@ mod file_search_response; +mod folder_contents; mod get_share_file_list; mod peer_init; mod place_in_queue_request; @@ -9,11 +10,15 @@ mod transfer_request; mod transfer_response; mod upload_denied; mod upload_failed; +mod user_info; // Re-export handlers pub use file_search_response::{ FileEntry, FileSearchResponse, build_file_search_response, }; +pub use folder_contents::{ + FolderContentsRequest, build_folder_contents, parse_folder_contents, +}; pub use get_share_file_list::GetShareFileList; pub use peer_init::PeerInit; pub use place_in_queue_request::PlaceInQueueRequest; @@ -27,3 +32,4 @@ pub use transfer_request::TransferRequest; pub use transfer_response::TransferResponse; pub use upload_denied::UploadDeniedHandler; pub use upload_failed::UploadFailedHandler; +pub use user_info::{UserInfoRequest, build_user_info}; diff --git a/soulseek-rs-lib/src/message/peer/peer_init.rs b/soulseek-rs-lib/src/message/peer/peer_init.rs index 7960098..3c7dbf8 100644 --- a/soulseek-rs-lib/src/message/peer/peer_init.rs +++ b/soulseek-rs-lib/src/message/peer/peer_init.rs @@ -8,7 +8,7 @@ use crate::{ pub struct PeerInit; impl MessageHandler for PeerInit { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 1 } diff --git a/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs b/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs index 9dc0224..2bbf775 100644 --- a/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs +++ b/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs @@ -12,7 +12,7 @@ use std::sync::mpsc::Sender; pub struct PlaceInQueueRequest; impl MessageHandler for PlaceInQueueRequest { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 51 } 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 9a372c5..447a77b 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 @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct PlaceInQueueResponse; impl MessageHandler for PlaceInQueueResponse { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 44 } diff --git a/soulseek-rs-lib/src/message/peer/queue_upload.rs b/soulseek-rs-lib/src/message/peer/queue_upload.rs index a88241b..a4c2143 100644 --- a/soulseek-rs-lib/src/message/peer/queue_upload.rs +++ b/soulseek-rs-lib/src/message/peer/queue_upload.rs @@ -8,7 +8,7 @@ use std::sync::mpsc::Sender; pub struct QueueUploadHandler; impl MessageHandler for QueueUploadHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 43 } diff --git a/soulseek-rs-lib/src/message/peer/shared_file_list.rs b/soulseek-rs-lib/src/message/peer/shared_file_list.rs index 4136b7c..6632797 100644 --- a/soulseek-rs-lib/src/message/peer/shared_file_list.rs +++ b/soulseek-rs-lib/src/message/peer/shared_file_list.rs @@ -4,7 +4,7 @@ use crate::message::{Message, MessageHandler}; use crate::peer::PeerMessage; -use crate::utils::zlib::{compress_stored, deflate}; +use crate::utils::zlib::{deflate, inflate}; use std::sync::mpsc::Sender; /// One shared directory and the files directly in it (basename + size). @@ -17,7 +17,7 @@ pub struct SharedDirectory { /// Receives a peer's `SharedFileListResponse` (peer code 5) when browsing them. pub struct SharedFileListResponseHandler; impl MessageHandler for SharedFileListResponseHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 5 } fn handle(&self, message: &mut Message, sender: Sender) { @@ -30,6 +30,29 @@ impl MessageHandler for SharedFileListResponseHandler { #[must_use] pub fn build_shared_file_list(dirs: &[SharedDirectory]) -> Message { let mut payload = Message::new(); + write_directories(&mut payload, dirs); + payload.write_int32(0); // unknown + payload.write_int32(0); // number of private directories + + let compressed = deflate(&payload.get_data()); + Message::new() + .write_int32(5) + .write_raw_bytes(compressed) + .clone() +} + +/// Parse the (zlib-compressed) `SharedFileListResponse` payload. `message` must +/// be positioned at the compressed blob (the dispatcher sets pointer 8). +/// +/// Returns an empty listing if the payload is malformed. +#[must_use] +pub fn parse_shared_file_list(message: &mut Message) -> Vec { + decompress_body(message) + .map_or_else(Vec::new, |mut body| read_directories(&mut body)) +} + +/// Write a directory listing in the form both code 5 and code 37 carry. +pub fn write_directories(payload: &mut Message, dirs: &[SharedDirectory]) { payload.write_int32(dirs.len() as u32); for dir in dirs { payload @@ -44,35 +67,15 @@ pub fn build_shared_file_list(dirs: &[SharedDirectory]) -> Message { .write_int32(0); // attribute count } } - payload.write_int32(0); // unknown - payload.write_int32(0); // number of private directories - - let compressed = compress_stored(&payload.get_data()); - Message::new() - .write_int32(5) - .write_raw_bytes(compressed) - .clone() } -/// Parse the (zlib-compressed) `SharedFileListResponse` payload. `message` must -/// be positioned at the compressed blob (the dispatcher sets pointer 8). -/// -/// Returns an empty listing if the payload is malformed. -#[must_use] -pub fn parse_shared_file_list(message: &mut Message) -> Vec { - let pointer = message.get_pointer(); - let size = message.get_size(); - let compressed = message.get_slice(pointer, size); - let Ok(data) = deflate(&compressed) else { - return Vec::new(); - }; - - let mut body = Message::new_with_data(data); +/// Read back what [`write_directories`] wrote, stopping early when a hostile +/// count outruns the payload so a bogus length cannot spin into a huge +/// allocation loop. +pub fn read_directories(body: &mut Message) -> Vec { let dir_count = body.read_int32(); let mut dirs = Vec::new(); for _ in 0..dir_count { - // Stop if a hostile count outruns the (decompressed) payload, so a - // bogus length can't spin us into a huge allocation loop. if body.get_pointer() >= body.get_size() { break; } @@ -104,12 +107,20 @@ pub fn parse_shared_file_list(message: &mut Message) -> Vec { dirs } +/// Inflate a compressed peer payload positioned at the blob. +pub fn decompress_body(message: &mut Message) -> Option { + let pointer = message.get_pointer(); + let size = message.get_size(); + inflate(&message.get_slice(pointer, size)) + .ok() + .map(Message::new_with_data) +} + #[test] fn hostile_dir_count_does_not_hang() { // A compressed body claiming ~4 billion directories with no data must // parse to empty promptly rather than looping into an OOM. - let compressed = - crate::utils::zlib::compress_stored(&u32::MAX.to_le_bytes()); + let compressed = crate::utils::zlib::deflate(&u32::MAX.to_le_bytes()); let mut message = crate::message::framed(|m| { m.write_raw_bytes(compressed); }); diff --git a/soulseek-rs-lib/src/message/peer/transfer_request.rs b/soulseek-rs-lib/src/message/peer/transfer_request.rs index f431fe0..e0ed13d 100644 --- a/soulseek-rs-lib/src/message/peer/transfer_request.rs +++ b/soulseek-rs-lib/src/message/peer/transfer_request.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct TransferRequest; impl MessageHandler for TransferRequest { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 40 } fn handle(&self, message: &mut Message, sender: Sender) { diff --git a/soulseek-rs-lib/src/message/peer/transfer_response.rs b/soulseek-rs-lib/src/message/peer/transfer_response.rs index 699b492..ad13685 100644 --- a/soulseek-rs-lib/src/message/peer/transfer_response.rs +++ b/soulseek-rs-lib/src/message/peer/transfer_response.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct TransferResponse; impl MessageHandler for TransferResponse { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 41 } diff --git a/soulseek-rs-lib/src/message/peer/upload_denied.rs b/soulseek-rs-lib/src/message/peer/upload_denied.rs index 3ee5d94..761a9e2 100644 --- a/soulseek-rs-lib/src/message/peer/upload_denied.rs +++ b/soulseek-rs-lib/src/message/peer/upload_denied.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct UploadDeniedHandler; impl MessageHandler for UploadDeniedHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 50 } fn handle(&self, message: &mut Message, sender: Sender) { diff --git a/soulseek-rs-lib/src/message/peer/upload_failed.rs b/soulseek-rs-lib/src/message/peer/upload_failed.rs index 5007a79..b53f9f2 100644 --- a/soulseek-rs-lib/src/message/peer/upload_failed.rs +++ b/soulseek-rs-lib/src/message/peer/upload_failed.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct UploadFailedHandler; impl MessageHandler for UploadFailedHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 46 } fn handle(&self, message: &mut Message, sender: Sender) { diff --git a/soulseek-rs-lib/src/message/peer/user_info.rs b/soulseek-rs-lib/src/message/peer/user_info.rs new file mode 100644 index 0000000..0887ad8 --- /dev/null +++ b/soulseek-rs-lib/src/message/peer/user_info.rs @@ -0,0 +1,33 @@ +//! `UserInfoRequest` (peer code 15) and its reply (code 16): what a peer says +//! about itself when another user opens its profile. + +use crate::message::{Message, MessageHandler}; +use crate::peer::PeerMessage; +use std::sync::mpsc::Sender; + +pub struct UserInfoRequest; +impl MessageHandler for UserInfoRequest { + fn get_code(&self) -> u32 { + 15 + } + fn handle(&self, _message: &mut Message, sender: Sender) { + let _ = sender.send(PeerMessage::UserInfoRequested); + } +} + +/// Build a `UserInfoResponse` (peer code 16) with no description or picture. +#[must_use] +pub fn build_user_info( + upload_slots: u32, + queue_size: u32, + slots_free: bool, +) -> Message { + Message::new() + .write_int32(16) + .write_string("") + .write_bool(false) + .write_int32(upload_slots) + .write_int32(queue_size) + .write_bool(slots_free) + .clone() +} diff --git a/soulseek-rs-lib/src/message/server/admin_message.rs b/soulseek-rs-lib/src/message/server/admin_message.rs new file mode 100644 index 0000000..e043e47 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/admin_message.rs @@ -0,0 +1,30 @@ +//! `AdminMessage` (server code 66): an announcement from the server to +//! everyone, surfaced as a private message from the user "server". + +use crate::actor::server_actor::{ServerMessage, UserMessage}; +use crate::message::{Message, MessageHandler}; +use std::sync::mpsc::Sender; +use std::time::{SystemTime, UNIX_EPOCH}; + +pub struct AdminMessageHandler; + +impl MessageHandler for AdminMessageHandler { + fn get_code(&self) -> u32 { + 66 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |since| since.as_secs() as u32); + let announcement = UserMessage::new( + 0, + now, + "server".to_string(), + message.read_string(), + false, + ); + let _ = + sender.send(ServerMessage::PrivateMessageReceived(announcement)); + } +} diff --git a/soulseek-rs-lib/src/message/server/connect_to_peer.rs b/soulseek-rs-lib/src/message/server/connect_to_peer.rs index d0a930d..1ffd706 100644 --- a/soulseek-rs-lib/src/message/server/connect_to_peer.rs +++ b/soulseek-rs-lib/src/message/server/connect_to_peer.rs @@ -5,7 +5,7 @@ use std::sync::mpsc::Sender; pub struct ConnectToPeerHandler; impl MessageHandler for ConnectToPeerHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 18 } fn handle(&self, message: &mut Message, sender: Sender) { diff --git a/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs b/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs index 2b7d3c5..6779bc1 100644 --- a/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs +++ b/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs @@ -8,7 +8,7 @@ use std::sync::mpsc::Sender; pub struct ExcludedSearchPhrasesHandler; impl MessageHandler for ExcludedSearchPhrasesHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 160 } diff --git a/soulseek-rs-lib/src/message/server/file_search.rs b/soulseek-rs-lib/src/message/server/file_search.rs index 509f538..c8def58 100644 --- a/soulseek-rs-lib/src/message/server/file_search.rs +++ b/soulseek-rs-lib/src/message/server/file_search.rs @@ -9,7 +9,7 @@ use crate::{ pub struct FileSearchHandler; impl MessageHandler for FileSearchHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 26 } fn handle(&self, message: &mut Message, sender: Sender) { diff --git a/soulseek-rs-lib/src/message/server/get_peer_address.rs b/soulseek-rs-lib/src/message/server/get_peer_address.rs index 7ea77a4..576d426 100644 --- a/soulseek-rs-lib/src/message/server/get_peer_address.rs +++ b/soulseek-rs-lib/src/message/server/get_peer_address.rs @@ -5,7 +5,7 @@ use std::sync::mpsc::Sender; pub struct GetPeerAddressHandler; impl MessageHandler for GetPeerAddressHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 3 } diff --git a/soulseek-rs-lib/src/message/server/join_room.rs b/soulseek-rs-lib/src/message/server/join_room.rs index 273e10c..b3462c7 100644 --- a/soulseek-rs-lib/src/message/server/join_room.rs +++ b/soulseek-rs-lib/src/message/server/join_room.rs @@ -8,7 +8,7 @@ use std::sync::mpsc::Sender; pub struct JoinRoomHandler; impl MessageHandler for JoinRoomHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 14 } diff --git a/soulseek-rs-lib/src/message/server/leave_room.rs b/soulseek-rs-lib/src/message/server/leave_room.rs index 5612ac5..583987e 100644 --- a/soulseek-rs-lib/src/message/server/leave_room.rs +++ b/soulseek-rs-lib/src/message/server/leave_room.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct LeaveRoomHandler; impl MessageHandler for LeaveRoomHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 15 } diff --git a/soulseek-rs-lib/src/message/server/login.rs b/soulseek-rs-lib/src/message/server/login.rs index 0248503..80dde9b 100644 --- a/soulseek-rs-lib/src/message/server/login.rs +++ b/soulseek-rs-lib/src/message/server/login.rs @@ -8,7 +8,7 @@ use crate::message::MessageHandler; pub struct LoginHandler; impl MessageHandler for LoginHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 1 } diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 2cf6793..87fd62c 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -143,6 +143,16 @@ impl MessageFactory { Message::new().write_int32(7).write_string(username).clone() } + /// Tell the server (code 121) what a finished upload averaged, in bytes + /// per second; it folds that into the speed other users see for us. + #[must_use] + pub fn build_send_upload_speed(bytes_per_second: u32) -> Message { + Message::new() + .write_int32(121) + .write_int32(bytes_per_second) + .clone() + } + /// Ask the server (code 36) for a user's share statistics. #[must_use] pub fn build_get_user_stats(username: &str) -> Message { @@ -336,7 +346,7 @@ fn a_transfer_denial_carries_the_token_and_the_reason() { let message = MessageFactory::build_transfer_denial_message(555, "Cancelled"); let mut decoded = Message::new_with_data(message.get_buffer()); - assert_eq!(decoded.get_message_code_u32(), 41); + assert_eq!(decoded.get_message_code(), 41); decoded.set_pointer(8); assert_eq!(decoded.read_int32(), 555); assert!(!decoded.read_bool()); @@ -474,3 +484,12 @@ fn test_build_unwatch_user() { let expect: Vec = [6, 0, 0, 0, 3, 0, 0, 0, b'b', b'o', b'b'].to_vec(); assert_eq!(expect, message.get_data()); } + +#[test] +fn an_upload_speed_report_carries_the_rate() { + let message = MessageFactory::build_send_upload_speed(900); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 121); + decoded.set_pointer(8); + assert_eq!(decoded.read_int32(), 900); +} diff --git a/soulseek-rs-lib/src/message/server/message_user.rs b/soulseek-rs-lib/src/message/server/message_user.rs index 02f80a5..0520197 100644 --- a/soulseek-rs-lib/src/message/server/message_user.rs +++ b/soulseek-rs-lib/src/message/server/message_user.rs @@ -8,7 +8,7 @@ use std::sync::mpsc::Sender; pub struct MessageUser; impl MessageHandler for MessageUser { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 22 } diff --git a/soulseek-rs-lib/src/message/server/mod.rs b/soulseek-rs-lib/src/message/server/mod.rs index b2d93b7..3d210ab 100644 --- a/soulseek-rs-lib/src/message/server/mod.rs +++ b/soulseek-rs-lib/src/message/server/mod.rs @@ -1,3 +1,4 @@ +mod admin_message; mod connect_to_peer; mod excluded_search_phrases; mod file_search; @@ -19,6 +20,7 @@ mod user_left_room; mod watch_user; mod wish_list_interval; +pub use admin_message::AdminMessageHandler; pub use connect_to_peer::ConnectToPeerHandler; pub use excluded_search_phrases::ExcludedSearchPhrasesHandler; pub use file_search::FileSearchHandler; diff --git a/soulseek-rs-lib/src/message/server/parent_min_speed.rs b/soulseek-rs-lib/src/message/server/parent_min_speed.rs index e0aa010..b93ef6c 100644 --- a/soulseek-rs-lib/src/message/server/parent_min_speed.rs +++ b/soulseek-rs-lib/src/message/server/parent_min_speed.rs @@ -9,7 +9,7 @@ use crate::{ pub struct ParentMinSpeedHandler; impl MessageHandler for ParentMinSpeedHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 83 } diff --git a/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs b/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs index 1888f61..744c6ea 100644 --- a/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs +++ b/soulseek-rs-lib/src/message/server/parent_speed_ratio.rs @@ -10,7 +10,7 @@ pub struct ParentSpeedRatioHandler; // The server sends us a speed ratio determining the number of children we can have in the distributed network. The maximum number of children is our upload speed divided by the speed ratio. impl MessageHandler for ParentSpeedRatioHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 84 } diff --git a/soulseek-rs-lib/src/message/server/privileged_users.rs b/soulseek-rs-lib/src/message/server/privileged_users.rs index 2dc042c..7f78ccc 100644 --- a/soulseek-rs-lib/src/message/server/privileged_users.rs +++ b/soulseek-rs-lib/src/message/server/privileged_users.rs @@ -14,7 +14,7 @@ use crate::{ pub struct PrivilegedUsersHandler; impl MessageHandler for PrivilegedUsersHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 69 } @@ -37,7 +37,7 @@ impl MessageHandler for PrivilegedUsersHandler { pub struct CheckPrivilegesHandler; impl MessageHandler for CheckPrivilegesHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 92 } diff --git a/soulseek-rs-lib/src/message/server/relogged.rs b/soulseek-rs-lib/src/message/server/relogged.rs index 84ce7a7..1d15361 100644 --- a/soulseek-rs-lib/src/message/server/relogged.rs +++ b/soulseek-rs-lib/src/message/server/relogged.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct ReloggedHandler; impl MessageHandler for ReloggedHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 41 } diff --git a/soulseek-rs-lib/src/message/server/room_list.rs b/soulseek-rs-lib/src/message/server/room_list.rs index 8596606..710a7ab 100644 --- a/soulseek-rs-lib/src/message/server/room_list.rs +++ b/soulseek-rs-lib/src/message/server/room_list.rs @@ -8,7 +8,7 @@ use std::sync::mpsc::Sender; pub struct RoomListHandler; impl MessageHandler for RoomListHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 64 } diff --git a/soulseek-rs-lib/src/message/server/say_chatroom.rs b/soulseek-rs-lib/src/message/server/say_chatroom.rs index b95e9f3..87e30b7 100644 --- a/soulseek-rs-lib/src/message/server/say_chatroom.rs +++ b/soulseek-rs-lib/src/message/server/say_chatroom.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct SayChatroomHandler; impl MessageHandler for SayChatroomHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 13 } diff --git a/soulseek-rs-lib/src/message/server/user_info.rs b/soulseek-rs-lib/src/message/server/user_info.rs index cd8b7a9..0a0cb7d 100644 --- a/soulseek-rs-lib/src/message/server/user_info.rs +++ b/soulseek-rs-lib/src/message/server/user_info.rs @@ -12,7 +12,7 @@ use std::sync::mpsc::Sender; pub struct GetUserStatusHandler; impl MessageHandler for GetUserStatusHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 7 } @@ -33,7 +33,7 @@ impl MessageHandler for GetUserStatusHandler { pub struct GetUserStatsHandler; impl MessageHandler for GetUserStatsHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 36 } diff --git a/soulseek-rs-lib/src/message/server/user_joined_room.rs b/soulseek-rs-lib/src/message/server/user_joined_room.rs index c7c785b..2682b53 100644 --- a/soulseek-rs-lib/src/message/server/user_joined_room.rs +++ b/soulseek-rs-lib/src/message/server/user_joined_room.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct UserJoinedRoomHandler; impl MessageHandler for UserJoinedRoomHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 16 } diff --git a/soulseek-rs-lib/src/message/server/user_left_room.rs b/soulseek-rs-lib/src/message/server/user_left_room.rs index c5ff89b..9438163 100644 --- a/soulseek-rs-lib/src/message/server/user_left_room.rs +++ b/soulseek-rs-lib/src/message/server/user_left_room.rs @@ -7,7 +7,7 @@ use std::sync::mpsc::Sender; pub struct UserLeftRoomHandler; impl MessageHandler for UserLeftRoomHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 17 } diff --git a/soulseek-rs-lib/src/message/server/watch_user.rs b/soulseek-rs-lib/src/message/server/watch_user.rs index 1499189..8fd478f 100644 --- a/soulseek-rs-lib/src/message/server/watch_user.rs +++ b/soulseek-rs-lib/src/message/server/watch_user.rs @@ -21,7 +21,7 @@ use std::sync::mpsc::Sender; pub struct WatchUserHandler; impl MessageHandler for WatchUserHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 5 } diff --git a/soulseek-rs-lib/src/message/server/wish_list_interval.rs b/soulseek-rs-lib/src/message/server/wish_list_interval.rs index 2d5b06e..ce8f141 100644 --- a/soulseek-rs-lib/src/message/server/wish_list_interval.rs +++ b/soulseek-rs-lib/src/message/server/wish_list_interval.rs @@ -13,7 +13,7 @@ use crate::{ pub struct WishListIntervalHandler; impl MessageHandler for WishListIntervalHandler { - fn get_code(&self) -> u8 { + fn get_code(&self) -> u32 { 104 } diff --git a/soulseek-rs-lib/src/peer/download_peer.rs b/soulseek-rs-lib/src/peer/download_peer.rs index b29b4a5..b941d32 100644 --- a/soulseek-rs-lib/src/peer/download_peer.rs +++ b/soulseek-rs-lib/src/peer/download_peer.rs @@ -13,10 +13,15 @@ use crate::trace; use crate::types::{Download, DownloadStatus}; use crate::utils::path::{PART_SUFFIX, expand_tilde}; -const READ_BUFFER_SIZE: usize = 8192; +/// 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); -const PROGRESS_UPDATE_CHUNKS: usize = 15; // ~120KB (15 * 8192 bytes) +/// 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 { @@ -351,8 +356,8 @@ impl DownloadPeer { client_context: &Arc>, download: Option, ) -> Result<(Download, String), DownloadError> { - let mut read_buffer = [0u8; READ_BUFFER_SIZE]; - let mut chunk_counter = 0usize; + 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 @@ -413,12 +418,10 @@ impl DownloadPeer { }; part.write(data)?; - chunk_counter += 1; + bytes_since_last_update += bytes_read; - if chunk_counter.is_multiple_of(PROGRESS_UPDATE_CHUNKS) { + if bytes_since_last_update >= PROGRESS_INTERVAL_BYTES { let elapsed = last_update_time.elapsed().as_secs_f64(); - let bytes_since_last_update = - PROGRESS_UPDATE_CHUNKS * READ_BUFFER_SIZE; let speed = if elapsed > 0.0 { bytes_since_last_update as f64 / elapsed } else { @@ -430,6 +433,7 @@ impl DownloadPeer { part.written, speed, ); + bytes_since_last_update = 0; last_update_time = Instant::now(); } diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index a546d9b..20e747d 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -360,7 +360,7 @@ fn handle_incoming_connection( // A firewalled peer brokered through the server connects back with a // PierceFirewall (code 0) instead of a PeerInit (code 1). - if message.get_message_code() == 0 { + if message.get_init_code() == 0 { handle_pierce_firewall( message, stream, reader, &context, &peer_ip, peer_port, ); diff --git a/soulseek-rs-lib/src/peer/mod.rs b/soulseek-rs-lib/src/peer/mod.rs index 65d0323..586320d 100644 --- a/soulseek-rs-lib/src/peer/mod.rs +++ b/soulseek-rs-lib/src/peer/mod.rs @@ -191,7 +191,7 @@ fn test_new_from_message2() { let mut message = Message::new_with_data(data); message.set_pointer(8); - println!("code: {}", message.get_message_code_u32()); + println!("code: {}", message.get_message_code()); let peer = Peer::new_from_message(&mut message).unwrap(); diff --git a/soulseek-rs-lib/src/peer/upload_peer.rs b/soulseek-rs-lib/src/peer/upload_peer.rs index b102ba3..57138d4 100644 --- a/soulseek-rs-lib/src/peer/upload_peer.rs +++ b/soulseek-rs-lib/src/peer/upload_peer.rs @@ -19,7 +19,8 @@ use crate::trace; /// START_DOWNLOAD offset before we stream the file. /// /// `bytes_sent` is updated as the transfer progresses, and setting `cancel` -/// aborts the stream with an [`io::ErrorKind::Interrupted`] error. +/// aborts the stream with an [`io::ErrorKind::Interrupted`] error. Returns the +/// bytes streamed by this call, which a resumed transfer's prefix is not. /// /// # Errors /// Returns any I/O error opening the file or talking to the peer. @@ -31,7 +32,7 @@ pub fn serve_file( path: &Path, bytes_sent: &AtomicU64, cancel: &AtomicBool, -) -> io::Result<()> { +) -> io::Result { let mut file = File::open(path)?; let socket = format!("{host}:{port}") @@ -78,6 +79,7 @@ pub fn serve_file( } let mut buffer = vec![0u8; 64 * 1024]; + let mut streamed = 0u64; loop { if cancel.load(Ordering::Relaxed) { return Err(io::Error::new( @@ -91,6 +93,7 @@ pub fn serve_file( } stream.write_all(&buffer[..read])?; bytes_sent.fetch_add(read as u64, Ordering::Relaxed); + streamed += read as u64; } stream.flush()?; @@ -100,7 +103,7 @@ pub fn serve_file( // a slot cap is time no other peer can use. stream.shutdown(std::net::Shutdown::Write).ok(); trace!("[upload] served {} to {}:{}", path.display(), host, port); - Ok(()) + Ok(streamed) } #[cfg(test)] diff --git a/soulseek-rs-lib/src/shares/mod.rs b/soulseek-rs-lib/src/shares/mod.rs index 49fc9ad..341ec89 100644 --- a/soulseek-rs-lib/src/shares/mod.rs +++ b/soulseek-rs-lib/src/shares/mod.rs @@ -13,6 +13,9 @@ use std::path::{Path, PathBuf}; pub struct SharedFile { /// Backslash-separated path exposed to peers, e.g. `music\album\song.mp3`. pub virtual_path: String, + /// `virtual_path` lowercased once at scan time, so answering a search does + /// not lowercase every shared path again. + search_key: String, /// The real filesystem path used to serve the bytes. pub real_path: PathBuf, pub size: u64, @@ -106,10 +109,7 @@ impl Shares { } self.files .iter() - .filter(|f| { - let haystack = f.virtual_path.to_lowercase(); - terms.iter().all(|t| haystack.contains(t.as_str())) - }) + .filter(|f| terms.iter().all(|t| f.search_key.contains(t.as_str()))) .collect() } @@ -200,6 +200,7 @@ fn scan_root( } let virtual_path = virtual_path_for(root_name, root, &path); files.push(SharedFile { + search_key: virtual_path.to_lowercase(), virtual_path, real_path: path, size: meta.len(), @@ -278,6 +279,20 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[test] + fn a_mixed_case_path_matches_a_lowercase_query() { + let root = std::env::temp_dir() + .join(format!("soulseek-shares-case-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(root.join("Live Set")).unwrap(); + std::fs::write(root.join("Live Set").join("Opening TRACK.flac"), b"x") + .unwrap(); + let shares = Shares::scan(&root).unwrap(); + + assert_eq!(shares.search("live opening track").len(), 1); + let _ = std::fs::remove_dir_all(root); + } + #[test] fn search_is_case_insensitive_and_multi_term_and() { let root = temp_tree(); diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index 7687b5a..1f07ea5 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -1,6 +1,6 @@ use std::{collections::HashMap, sync::mpsc::Sender}; -use crate::{error::Result, message::Message, utils::zlib::deflate}; +use crate::{error::Result, message::Message, utils::zlib::inflate}; #[derive(Debug, Clone)] #[allow(dead_code)] @@ -84,6 +84,14 @@ pub const MAX_SEARCH_RESPONSES: usize = 500; /// matching collections cannot blow past the response cap. pub const MAX_SEARCH_FILES: usize = 10_000; +/// Files one reply to a peer's search carries at most. +/// +/// A broad query against a big share would otherwise hand the whole index +/// to whoever typed it, on every such search that arrives. Nicotine+ sends +/// 300 by default and slskd 500; a searcher with that many sources from one +/// peer has plenty. +pub const MAX_SEARCH_REPLY_FILES: usize = 300; + impl Search { /// Store one peer's response, unless this search already holds enough. /// @@ -106,7 +114,7 @@ impl SearchResult { let pointer = message.get_pointer(); let size = message.get_size(); let data: Vec = message.get_slice(pointer, size); - let deflated = deflate(&data)?; + let deflated = inflate(&data)?; let mut message = Message::new_with_data(deflated); let username = message.read_string(); @@ -415,7 +423,7 @@ mod tests { body.extend_from_slice(&0u32.to_le_bytes()); // username "" (len 0) body.extend_from_slice(&7u32.to_le_bytes()); // token body.extend_from_slice(&u32::MAX.to_le_bytes()); // n_files (hostile) - let compressed = crate::utils::zlib::compress_stored(&body); + let compressed = crate::utils::zlib::deflate(&body); let mut message = Message::new_with_data(compressed); let result = SearchResult::new_from_message(&mut message) .expect("hostile count should parse, not error"); diff --git a/soulseek-rs-lib/src/utils/keepalive.rs b/soulseek-rs-lib/src/utils/keepalive.rs new file mode 100644 index 0000000..b7c4358 --- /dev/null +++ b/soulseek-rs-lib/src/utils/keepalive.rs @@ -0,0 +1,99 @@ +//! TCP keepalive for the server connection. +//! +//! A NAT that drops an idle mapping leaves the socket looking connected while +//! the server can no longer reach us, and nothing ever errors. Probes every +//! ten seconds keep the mapping alive and turn a dead path into a read error +//! within about half a minute, which is what Nicotine+ does. The server +//! stopped answering ServerPing (code 32) long ago, so that is no substitute. + +use std::io; +use std::net::TcpStream; + +#[cfg(unix)] +pub fn set_keepalive(stream: &TcpStream) -> io::Result<()> { + use std::os::unix::io::AsRawFd; + let fd = stream.as_raw_fd(); + set_option(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE, 1)?; + #[cfg(any( + target_vendor = "apple", + target_os = "linux", + target_os = "android", + target_os = "freebsd", + target_os = "netbsd" + ))] + { + set_option(fd, libc::IPPROTO_TCP, IDLE_SECONDS_OPTION, 10)?; + set_option(fd, libc::IPPROTO_TCP, libc::TCP_KEEPINTVL, 2)?; + set_option(fd, libc::IPPROTO_TCP, libc::TCP_KEEPCNT, 10)?; + } + Ok(()) +} + +#[cfg(target_vendor = "apple")] +const IDLE_SECONDS_OPTION: libc::c_int = libc::TCP_KEEPALIVE; +#[cfg(any( + target_os = "linux", + target_os = "android", + target_os = "freebsd", + target_os = "netbsd" +))] +const IDLE_SECONDS_OPTION: libc::c_int = libc::TCP_KEEPIDLE; + +#[cfg(unix)] +fn set_option( + fd: libc::c_int, + level: libc::c_int, + name: libc::c_int, + value: libc::c_int, +) -> io::Result<()> { + // SAFETY: `value` outlives the call and its exact size travels with it. + let rc = unsafe { + libc::setsockopt( + fd, + level, + name, + (&raw const value).cast(), + std::mem::size_of::() as libc::socklen_t, + ) + }; + if rc == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } +} + +// ponytail: Windows wants WSAIoctl(SIO_KEEPALIVE_VALS); wire it when the +// daemon ships there. +#[cfg(not(unix))] +pub fn set_keepalive(_stream: &TcpStream) -> io::Result<()> { + Ok(()) +} + +#[cfg(all(test, unix))] +pub fn keepalive_enabled(stream: &TcpStream) -> bool { + use std::os::unix::io::AsRawFd; + let mut value: libc::c_int = 0; + let mut len = std::mem::size_of::() as libc::socklen_t; + // SAFETY: `value` and `len` outlive the call and `len` bounds the write. + let rc = unsafe { + libc::getsockopt( + stream.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_KEEPALIVE, + (&raw mut value).cast(), + &raw mut len, + ) + }; + rc == 0 && value != 0 +} + +#[cfg(all(test, unix))] +#[test] +fn keepalive_is_off_until_asked_for() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let stream = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + assert!(!keepalive_enabled(&stream)); + set_keepalive(&stream).unwrap(); + assert!(keepalive_enabled(&stream)); +} diff --git a/soulseek-rs-lib/src/utils/logger.rs b/soulseek-rs-lib/src/utils/logger.rs index 208dd73..ea49656 100644 --- a/soulseek-rs-lib/src/utils/logger.rs +++ b/soulseek-rs-lib/src/utils/logger.rs @@ -3,7 +3,7 @@ use std::fs::{File, OpenOptions}; use std::io::Write; use std::sync::{ Mutex, Once, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicU8, Ordering}, }; #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] @@ -16,7 +16,23 @@ pub enum LogLevel { } static INIT: Once = Once::new(); -static mut LOG_LEVEL: LogLevel = LogLevel::Warn; +static LOG_LEVEL: AtomicU8 = AtomicU8::new(LogLevel::Warn as u8); + +/// Change the level at runtime, from any thread. +pub fn set_log_level(level: LogLevel) { + LOG_LEVEL.store(level as u8, Ordering::Relaxed); +} + +#[must_use] +pub fn log_level() -> LogLevel { + match LOG_LEVEL.load(Ordering::Relaxed) { + 0 => LogLevel::Error, + 2 => LogLevel::Info, + 3 => LogLevel::Debug, + 4 => LogLevel::Trace, + _ => LogLevel::Warn, + } +} static BUFFER: Mutex> = Mutex::new(Vec::new()); static BUFFERING: AtomicBool = AtomicBool::new(false); @@ -29,15 +45,13 @@ pub fn init() { .unwrap_or_else(|_| "WARN".to_string()) .to_uppercase(); - unsafe { - LOG_LEVEL = match level.as_str() { - "ERROR" => LogLevel::Error, - "INFO" => LogLevel::Info, - "DEBUG" | "VERBOSE" => LogLevel::Debug, // Map VERBOSE to DEBUG - "TRACE" => LogLevel::Trace, - _ => LogLevel::Warn, // "WARN" or default - }; - } + set_log_level(match level.as_str() { + "ERROR" => LogLevel::Error, + "INFO" => LogLevel::Info, + "DEBUG" | "VERBOSE" => LogLevel::Debug, + "TRACE" => LogLevel::Trace, + _ => LogLevel::Warn, + }); // Initialize log file if LOG_FILE env var is set if let Ok(log_file_path) = env::var("LOG_FILE") { @@ -85,8 +99,8 @@ fn has_log_file() -> bool { } pub fn log(level: LogLevel, message: &str) { - unsafe { - if level <= LOG_LEVEL { + if level <= log_level() { + { let (name, colour) = match level { LogLevel::Error => ("ERROR", "\x1b[31m"), // Red LogLevel::Warn => ("WARN", "\x1b[33m"), // Yellow @@ -239,7 +253,16 @@ macro_rules! trace { #[cfg(test)] mod tests { - use super::{LogSink, choose_sink}; + use super::{LogLevel, LogSink, choose_sink, log_level, set_log_level}; + + #[test] + fn the_level_can_be_changed_while_running() { + let before = log_level(); + set_log_level(LogLevel::Trace); + assert_eq!(log_level(), LogLevel::Trace); + set_log_level(before); + assert_eq!(log_level(), before); + } #[test] fn a_configured_file_bypasses_buffering_so_lines_are_not_duplicated() { diff --git a/soulseek-rs-lib/src/utils/mod.rs b/soulseek-rs-lib/src/utils/mod.rs index 5bb6848..e32ce7c 100644 --- a/soulseek-rs-lib/src/utils/mod.rs +++ b/soulseek-rs-lib/src/utils/mod.rs @@ -1,5 +1,6 @@ #[macro_use] pub mod logger; +pub(crate) mod keepalive; pub mod lock; pub mod md5; pub mod path; diff --git a/soulseek-rs-lib/src/utils/zlib.rs b/soulseek-rs-lib/src/utils/zlib.rs index 3a16606..46a6bc8 100644 --- a/soulseek-rs-lib/src/utils/zlib.rs +++ b/soulseek-rs-lib/src/utils/zlib.rs @@ -59,7 +59,8 @@ impl BitReader { use crate::error::{Result, SoulseekRs}; -pub fn deflate(input: &[u8]) -> Result> { +/// Inflate a zlib stream, checking its adler32 trailer. +pub fn inflate(input: &[u8]) -> Result> { let mut r = BitReader::new(input.to_vec()); let cmf = r.read_byte()?; let cm = cmf & 15; // Compression method @@ -83,35 +84,14 @@ pub fn deflate(input: &[u8]) -> Result> { "preset dictionary not supported".to_string(), )); } - let out = inflate(&mut r).map_err(SoulseekRs::CompressionError)?; // decompress DEFLATE data - let _adler32 = r.read_bytes(4)?; // Adler-32 checksum (for this exercise, we ignore it) - Ok(out) -} - -/// Compress `data` into a valid zlib stream using only STORED blocks. -/// -/// No real compressor is needed: standard zlib decoders (and [`deflate`] above) -/// accept a `0x78 0x01` header, one or more uncompressed DEFLATE "stored" blocks -/// of at most `0xFFFF` bytes, then a big-endian adler32 checksum. -#[must_use] -pub fn compress_stored(data: &[u8]) -> Vec { - let mut out = vec![0x78, 0x01]; - if data.is_empty() { - // A single empty, final stored block: BFINAL=1, LEN=0, NLEN=0xFFFF. - out.extend_from_slice(&[0x01, 0x00, 0x00, 0xFF, 0xFF]); - } else { - let mut chunks = data.chunks(0xFFFF).peekable(); - while let Some(chunk) = chunks.next() { - // BFINAL in bit 0, BTYPE=00 (stored) in bits 1-2. - out.push(u8::from(chunks.peek().is_none())); - let len = chunk.len() as u16; - out.extend_from_slice(&len.to_le_bytes()); - out.extend_from_slice(&(!len).to_le_bytes()); - out.extend_from_slice(chunk); - } + let out = inflate_blocks(&mut r).map_err(SoulseekRs::CompressionError)?; + // `read_bytes` assembles little-endian; the zlib trailer is big-endian. + if r.read_bytes(4)?.swap_bytes() != adler32(&out) { + return Err(SoulseekRs::CompressionError( + "adler32 checksum failed".to_string(), + )); } - out.extend_from_slice(&adler32(data).to_be_bytes()); - out + Ok(out) } /// RFC 1950 adler32 checksum. @@ -125,7 +105,7 @@ fn adler32(data: &[u8]) -> u32 { (b << 16) | a } -fn inflate(r: &mut BitReader) -> std::result::Result, String> { +fn inflate_blocks(r: &mut BitReader) -> std::result::Result, String> { let mut bfinal = 0; let mut out = Vec::new(); while bfinal == 0 { @@ -392,11 +372,234 @@ fn inflate_block_fixed( inflate_block_data(r, &literal_length_tree, &distance_tree, o) } +struct BitWriter { + out: Vec, + bit: u32, + held: u32, +} + +impl BitWriter { + fn new() -> Self { + Self { + out: vec![0x78, 0x01], + bit: 0, + held: 0, + } + } + + /// Extra bits, least significant first. + fn bits(&mut self, value: u32, count: u32) { + self.held |= value << self.bit; + self.bit += count; + while self.bit >= 8 { + self.out.push((self.held & 0xFF) as u8); + self.held >>= 8; + self.bit -= 8; + } + } + + /// A Huffman code, most significant bit first. + fn code(&mut self, value: u32, count: u32) { + for i in (0..count).rev() { + self.bits((value >> i) & 1, 1); + } + } + + fn finish(mut self, trailer: u32) -> Vec { + if self.bit > 0 { + self.out.push((self.held & 0xFF) as u8); + } + self.out.extend_from_slice(&trailer.to_be_bytes()); + self.out + } +} + +const MIN_MATCH: usize = 3; +const MAX_MATCH: usize = 258; +const WINDOW: usize = 32_768; +const MATCH_TRIES: usize = 128; +/// A match this long is good enough to stop walking the chain. +const NICE_MATCH: usize = 64; + +/// Hash table size for `len` bytes: a search reply of a few hundred bytes +/// must not zero 256 KiB of table, a listing of megabytes wants all of it. +fn hash_bits(len: usize) -> u32 { + len.max(2).next_power_of_two().trailing_zeros().clamp(8, 15) +} + +/// The fixed Huffman code for a literal or length symbol (RFC 1951 3.2.6). +fn symbol(w: &mut BitWriter, symbol: usize) { + let symbol = symbol as u32; + match symbol { + 0..=143 => w.code(0x30 + symbol, 8), + 144..=255 => w.code(0x190 + symbol - 144, 9), + 256..=279 => w.code(symbol - 256, 7), + _ => w.code(0xC0 + symbol - 280, 8), + } +} + +fn emit_match(w: &mut BitWriter, length: usize, distance: usize) { + let l = LENGTH_BASE + .iter() + .rposition(|&base| base as usize <= length) + .unwrap_or(0); + symbol(w, 257 + l); + w.bits(length as u32 - LENGTH_BASE[l], LENGTH_EXTRA_BITS[l] as u32); + let d = DISTANCE_BASE + .iter() + .rposition(|&base| base as usize <= distance) + .unwrap_or(0); + w.code(d as u32, 5); + w.bits( + distance as u32 - DISTANCE_BASE[d], + DISTANCE_EXTRA_BITS[d] as u32, + ); +} + +fn hash(data: &[u8], at: usize, bits: u32) -> usize { + let three = u32::from(data[at]) << 16 + | u32::from(data[at + 1]) << 8 + | u32::from(data[at + 2]); + (three.wrapping_mul(2_654_435_761) >> (32 - bits)) as usize +} + +/// Chain the three bytes at `at` into the match finder. +fn remember(head: &mut [usize], prev: &mut [usize], data: &[u8], at: usize) { + if at + MIN_MATCH > data.len() { + return; + } + let h = hash(data, at, head.len().trailing_zeros()); + prev[at] = head[h]; + head[h] = at; +} + +/// Deflate `data` into a zlib stream. +/// +/// One fixed-Huffman block over an LZ77 hash-chain match finder, which is +/// what shrinks a listing's repeated path prefixes. ponytail: fixed codes, not +/// dynamic; listings compress 5-10x either way and this is a tenth of the code. +#[must_use] +pub fn deflate(data: &[u8]) -> Vec { + let mut w = BitWriter::new(); + w.bits(1, 1); // final block + w.bits(1, 2); // fixed Huffman codes + let bits = hash_bits(data.len()); + let mut head = vec![usize::MAX; 1 << bits]; + let mut prev = vec![usize::MAX; data.len()]; + let mut i = 0; + while i < data.len() { + let mut best = (0, 0); + if i + MIN_MATCH <= data.len() { + let h = hash(data, i, bits); + let mut candidate = head[h]; + let mut tries = MATCH_TRIES; + let max = MAX_MATCH.min(data.len() - i); + while candidate != usize::MAX + && tries > 0 + && i - candidate <= WINDOW + { + let len = (0..max) + .take_while(|&k| data[candidate + k] == data[i + k]) + .count(); + if len > best.0 { + best = (len, i - candidate); + if len == max || len >= NICE_MATCH { + break; + } + } + candidate = prev[candidate]; + tries -= 1; + } + } + remember(&mut head, &mut prev, data, i); + let (length, distance) = best; + if length < MIN_MATCH { + symbol(&mut w, usize::from(data[i])); + i += 1; + continue; + } + emit_match(&mut w, length, distance); + for at in i + 1..i + length { + remember(&mut head, &mut prev, data, at); + } + i += length; + } + symbol(&mut w, 256); // end of block + w.finish(adler32(data)) +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn a_corrupt_adler32_trailer_is_rejected() { + let mut stream = deflate(b"the bytes a peer sent"); + let last = stream.len() - 1; + stream[last] ^= 0xFF; + assert!(inflate(&stream).is_err(), "a bad checksum must not pass"); + } + + #[test] + fn a_reference_stream_inflates() { + let stream = [ + 120, 156, 203, 72, 205, 201, 201, 87, 8, 207, 47, 202, 73, 1, 0, + 25, 107, 4, 61, + ]; + assert_eq!(inflate(&stream).unwrap(), b"hello World"); + } + + fn listing() -> Vec { + (0..3000) + .flat_map(|n| { + format!("Artist - Album - {n:04} - Track Title.flac") + .into_bytes() + }) + .collect() + } + + #[test] + fn every_shape_round_trips() { + let mut seed = 0x2545_F491_4F6C_DD1Du64; + let mut noise = |n: usize| -> Vec { + (0..n) + .map(|_| { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + (seed >> 24) as u8 + }) + .collect() + }; + let cases = [ + Vec::new(), + vec![0], + vec![0x41; 258], + vec![0x41; 70_000], + vec![0x5A; 100_000], + b"the same words the same words the same words".to_vec(), + listing(), + noise(1), + noise(4096), + noise(70_000), + ]; + for data in cases { + assert_eq!( + inflate(&deflate(&data)).unwrap(), + data, + "{}", + data.len() + ); + } + } + + #[test] + fn a_repetitive_listing_shrinks() { + let data = listing(); + assert!(deflate(&data).len() * 4 < data.len()); + } + #[test] fn decode_symbol_on_degenerate_tree_errors_instead_of_panicking() { // A malformed Huffman table from an untrusted peer can produce a node @@ -409,28 +612,6 @@ mod tests { assert!(decode_symbol(&mut reader, &tree).is_err()); } - #[test] - fn compress_stored_roundtrips_through_the_decoder() { - // Cover the block boundaries: empty, tiny, exactly 0xFFFF, and multi-block. - for size in [0usize, 1, 11, 65_534, 65_535, 65_536, 131_072, 200_000] { - let data: Vec = (0..size).map(|i| (i % 251) as u8).collect(); - let compressed = compress_stored(&data); - assert_eq!( - deflate(&compressed).unwrap(), - data, - "roundtrip failed at size {size}" - ); - } - } - - #[test] - fn compress_stored_header_and_adler32_are_correct() { - assert_eq!(&compress_stored(b"")[0..2], &[120, 1]); - assert_eq!(adler32(b""), 1); - // a = 1+97+98+99 = 295 = 0x127; b = 98+294+589 = 981 = 0x24D. - assert_eq!(adler32(b"abc"), 0x024D_0127); - } - #[test] fn test_bitreader_read_bits() { let data = vec![0b11010010, 0b10110101]; @@ -449,13 +630,6 @@ mod tests { assert_eq!(reader.read_bytes(2).unwrap(), 0x7856); // Little-endian: 0x7856 } - #[test] - fn test_extract_header_success() { - // [120, 156] is a valid zlib header - let result = deflate(&[120, 156, 3, 0, 0, 0, 0, 1]); // Minimal valid zlib stream - assert!(result.is_ok()); - } - #[test] fn test_extract_header_fail_to_short() { let data = vec![120]; // Too short @@ -467,20 +641,7 @@ mod tests { } #[test] - fn test_deflate() { - // Test data from the original test - this should work with our new implementation - let data = vec![ - 120, 156, 203, 72, 205, 201, 201, 87, 8, 207, 47, 202, 73, 1, 0, - 24, 11, 4, 93, - ]; - let result = deflate(&data); - assert!(result.is_ok()); - let decompressed = result.unwrap(); - assert_eq!(decompressed, b"hello World"); - } - - #[test] - fn test_deflate2() { + fn a_dynamic_huffman_stream_from_a_real_peer_inflates() { let data = vec![ 120, 156, 99, 103, 96, 96, 72, 201, 79, 201, 76, 79, 204, 203, 213, 158, 98, 194, 4, 228, 50, 250, 3, 9, 7, 135, 162, 156, 148, 194, @@ -522,7 +683,7 @@ mod tests { 0, 0, 0, 0, 0, ] .to_vec(); - let result = deflate(&data); + let result = inflate(&data); assert!(result.is_ok()); let decompressed = result.unwrap(); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index c5ba362..15f4495 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -13,8 +13,7 @@ //! cargo test -p soulseek-rs-lib --test e2e -- --nocapture //! ``` //! -//! No external crates are used — the library forbids dependencies, so the -//! harness sticks to `std` and the library's own public API. +//! The harness sticks to `std` and the library's own public API. use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; @@ -750,7 +749,7 @@ fn login_raw( /// passes. Returns the matching message. fn read_until_code( stream: &mut TcpStream, - code: u8, + code: u32, timeout: Duration, ) -> Option { let deadline = Instant::now() + timeout; @@ -767,7 +766,7 @@ fn read_until_code( /// Like [`read_until_code`] but turns a miss into an `io::Error` for `?`. fn expect_code( stream: &mut TcpStream, - code: u8, + code: u32, timeout: Duration, ) -> std::io::Result { read_until_code(stream, code, timeout).ok_or_else(|| { @@ -1379,7 +1378,7 @@ fn run_mock_direct_peer(cfg: &MockDirectUpload) -> std::io::Result<()> { p.set_nonblocking(false)?; p.set_read_timeout(Some(Duration::from_secs(10)))?; let mut init = read_framed(&mut p)?; - assert_eq!(init.get_message_code(), 1, "expected inbound PeerInit"); + assert_eq!(init.get_init_code(), 1, "expected inbound PeerInit"); init.set_pointer(5); assert_eq!(init.read_string(), cfg.downloader_username, "PeerInit user"); assert_eq!(init.read_string(), "P", "PeerInit connection type"); @@ -1645,6 +1644,56 @@ fn a_file_downloads_from_a_firewalled_peer_via_server_broker() { // it — the entire search + connect + upload/download stack, no mock peer. // --------------------------------------------------------------------------- +/// A sharer serving `share_dir` and a searcher, both logged in and listening, +/// after a beat for soulfind to register both SetWaitPorts. +fn sharer_and_searcher( + server: &TestServer, + share_dir: &std::path::Path, + sharer: &str, + searcher: &str, +) -> (Client, Client) { + let mut sharing = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.listening_settings( + sharer, + "pw", + free_port().expect("sharer port"), + ) + }); + sharing.connect().expect("sharer connect"); + assert!(sharing.login().expect("sharer login")); + + let mut searching = Client::with_settings(server.listening_settings( + searcher, + "pw", + free_port().expect("searcher port"), + )); + searching.connect().expect("searcher connect"); + assert!(searching.login().expect("searcher login")); + + std::thread::sleep(Duration::from_secs(1)); + (sharing, searching) +} + +/// `sharer`'s answer to `query`, once it reaches `searcher`. +fn reply_from( + searcher: &Client, + query: &str, + sharer: &str, +) -> Option { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let reply = searcher + .get_search_results(query) + .into_iter() + .find(|result| result.username == sharer); + if reply.is_some() || Instant::now() >= deadline { + return reply; + } + std::thread::sleep(Duration::from_millis(200)); + } +} + #[test] fn two_real_clients_search_and_download() { let server = server_or_skip!(); @@ -1655,48 +1704,21 @@ fn two_real_clients_search_and_download() { let filename = "e2e_probe_xyzzy.bin"; std::fs::write(share_dir.join(filename), &content).unwrap(); - let sharer_port = free_port().expect("sharer port"); - let mut sharer = Client::with_settings(ClientSettings { - shared_directories: vec![share_dir.display().to_string()], - ..server.listening_settings("e2e_sharer", "pw", sharer_port) - }); - sharer.connect().expect("sharer connect"); - assert!(sharer.login().expect("sharer login")); - - let leecher_port = free_port().expect("leecher port"); - let mut leecher = Client::with_settings(server.listening_settings( - "e2e_leecher", - "pw", - leecher_port, - )); - leecher.connect().expect("leecher connect"); - assert!(leecher.login().expect("leecher login")); - - // Let soulfind register both SetWaitPorts before the search resolves peers. - std::thread::sleep(Duration::from_secs(1)); + let (sharer, leecher) = + sharer_and_searcher(&server, &share_dir, "e2e_sharer", "e2e_leecher"); let query = "xyzzy"; let _ = leecher.search(query, Duration::from_secs(3)); - // Poll until the sharer's response for our file arrives. - let mut hit: Option<(String, u64)> = None; - let deadline = Instant::now() + Duration::from_secs(20); - while Instant::now() < deadline && hit.is_none() { - for result in leecher.get_search_results(query) { - if result.username == "e2e_sharer" { - for file in &result.files { - if file.name.contains("e2e_probe_xyzzy") { - hit = Some((file.name.clone(), file.size)); - } - } - } - } - if hit.is_none() { - std::thread::sleep(Duration::from_millis(200)); - } - } - let (result_path, size) = - hit.expect("leecher should find the sharer's file"); + let (result_path, size) = reply_from(&leecher, query, "e2e_sharer") + .and_then(|reply| { + reply + .files + .into_iter() + .find(|file| file.name.contains("e2e_probe_xyzzy")) + }) + .map(|file| (file.name, file.size)) + .expect("leecher should find the sharer's file"); assert_eq!(size, content.len() as u64); // Download it from the sharer. @@ -1738,6 +1760,31 @@ fn two_real_clients_search_and_download() { assert_eq!(upload.username, "e2e_leecher"); assert_eq!(upload.bytes_sent, size); + // The next reply advertises what that upload measured. + let _ = leecher.search(query, Duration::from_secs(1)); + let reply = reply_from(&leecher, query, "e2e_sharer") + .expect("the sharer answers again"); + assert!( + reply.speed > 0, + "a finished upload sets the advertised speed" + ); + + // The server hears about it too (SendUploadSpeed), which is where every + // other client's `user` lookup and the server's own ranking read it. + let deadline = Instant::now() + Duration::from_secs(10); + let mut recorded = 0; + while Instant::now() < deadline && recorded == 0 { + leecher + .request_user_info("e2e_sharer") + .expect("ask about the sharer"); + std::thread::sleep(Duration::from_millis(250)); + recorded = leecher + .user_info("e2e_sharer") + .and_then(|info| info.stats) + .map_or(0, |stats| stats.average_speed); + } + assert!(recorded > 0, "the server should record the upload's speed"); + let _ = std::fs::remove_dir_all(share_dir); let _ = std::fs::remove_dir_all(download_dir); } @@ -2056,7 +2103,7 @@ fn a_third_party_client_browses_our_shares_via_the_server_broker() { let mut pierce = read_framed(&mut p).expect("a PierceFirewall frame"); assert_eq!( - pierce.get_message_code(), + pierce.get_init_code(), 0, "the brokered connect-back must start with a PierceFirewall" ); @@ -2317,23 +2364,24 @@ fn place_of(sharer: &Client, username: &str, filename: &str) -> Option { }) } -/// Give `username` privileges in soulfind's own database. +/// Set `column` ("privileges", or "admin" for the user whose private +/// messages to "server" are commands) on `username` in soulfind's own +/// database, good for a day: soulfind stores both as the unix timestamp +/// they expire at. /// -/// `false` means it could not be done — an external server, or no `sqlite3` on -/// this machine — which the caller treats as a skip rather than a failure. -fn grant_privileges(server: &TestServer, username: &str) -> bool { +/// `false` means it could not be done — an external server, or no `sqlite3` +/// on this machine — which the caller treats as a skip rather than a failure. +fn grant(server: &TestServer, username: &str, column: &str) -> bool { let Some(db) = server.db.as_ref() else { return false; }; - // soulfind stores privileges as the unix timestamp they expire at. let expiry = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |since| since.as_secs() + 86_400); Command::new("sqlite3") .arg(db) .arg(format!( - "UPDATE users SET privileges = {expiry} WHERE username = \ - '{username}';" + "UPDATE users SET {column} = {expiry} WHERE username = '{username}';" )) .status() .is_ok_and(|status| status.success()) @@ -2442,7 +2490,7 @@ fn a_privileged_peer_overtakes_one_already_waiting_for_a_slot() { login_raw(&server_addr, "e2e_q_donor", "pw") .expect("the donor registers"), ); - if !grant_privileges(&server, "e2e_q_donor") { + if !grant(&server, "e2e_q_donor", "privileges") { eprintln!( "privilege e2e skipped: cannot write soulfind's database (needs a \ locally spawned server and sqlite3)" @@ -3039,3 +3087,394 @@ fn watching_a_user_the_server_does_not_know_reports_their_absence() { ); assert!(watcher.user_info("e2e_watch_nobody_here").is_none()); } + +// A query that matches most of a big share must not ship the whole share to +// whoever typed it: a reply is a shortlist, and the searcher's other sources +// cover the rest. +#[test] +fn a_search_reply_carries_at_most_the_file_cap() { + let server = server_or_skip!(); + + let cap = soulseek_rs::types::MAX_SEARCH_REPLY_FILES; + let share_dir = unique_download_dir(); + for i in 0..cap + 20 { + std::fs::write(share_dir.join(format!("probe_cap_{i:04}.bin")), b"x") + .unwrap(); + } + let (_sharer, searcher) = sharer_and_searcher( + &server, + &share_dir, + "e2e_cap_sharer", + "e2e_cap_searcher", + ); + + let query = "probe_cap"; + let _ = searcher.search(query, Duration::from_secs(3)); + + let reply = reply_from(&searcher, query, "e2e_cap_sharer") + .expect("the sharer should answer the search"); + assert_eq!(reply.files.len(), cap, "a reply is capped at {cap} files"); + + let _ = std::fs::remove_dir_all(share_dir); +} + +// Tokens were the first five hex digits of the query's MD5, so two queries +// could share one and a peer's answer to either went to whichever search the +// map handed back first. These two queries collide under that scheme. +#[test] +fn two_live_searches_never_share_a_token() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + std::fs::write(share_dir.join("tokenprobe aeabna.bin"), b"x").unwrap(); + let (_sharer, searcher) = sharer_and_searcher( + &server, + &share_dir, + "e2e_token_sharer", + "e2e_token_searcher", + ); + + let matching = "tokenprobe aeabna"; + let colliding = "tokenprobe soyvgt"; + let _ = searcher.search(colliding, Duration::from_millis(100)); + let _ = searcher.search(matching, Duration::from_secs(3)); + + let live = searcher.get_all_searches(); + assert_ne!( + live[matching].token, live[colliding].token, + "two live searches must not share a token" + ); + + assert!( + reply_from(&searcher, matching, "e2e_token_sharer").is_some(), + "the sharer's answer should land in the search it answers" + ); + + let _ = std::fs::remove_dir_all(share_dir); +} + +// Searchers pick sources by the free-slot flag in a reply, and every reply +// said "free" whatever the queue looked like. +#[test] +fn a_search_reply_reports_no_free_slot_while_the_only_slot_is_taken() { + let server = server_or_skip!(); + let (share, folder) = + queue_share("honest", &["blocker.mp3", "probe_honest.bin"]); + let (sharer, searcher) = sharer_and_searcher( + &server, + &share, + "e2e_honest_sharer", + "e2e_honest_searcher", + ); + sharer.set_upload_slots(1); + + let query = "probe_honest"; + let _ = searcher.search(query, Duration::from_secs(1)); + let reply = reply_from(&searcher, query, "e2e_honest_sharer") + .expect("the sharer answers"); + assert_eq!(reply.slots, 1, "nothing is uploading yet"); + + let server_addr = format!("{}:{}", server.host, server.port); + let listen_addr = + format!("127.0.0.1:{}", sharer.listen_port().expect("listening")); + let mut blocker = queue_as( + &server_addr, + &listen_addr, + "e2e_honest_blocker", + &format!("{folder}\\blocker.mp3"), + ) + .expect("blocker queues"); + assert!( + blocker.takes_a_slot(), + "the blocker is offered the only slot" + ); + + let _ = searcher.search(query, Duration::from_secs(1)); + let reply = reply_from(&searcher, query, "e2e_honest_sharer") + .expect("the sharer answers again"); + assert_eq!(reply.slots, 0, "the only slot is taken"); + + let _ = std::fs::remove_dir_all(share); +} + +// "Download folder" in SoulseekQt and Nicotine+ asks the sharer for that one +// folder (peer code 36) instead of the whole share; a sharer that never +// answers leaves the folder download hanging. +#[test] +fn a_third_party_client_fetches_one_folder_of_our_shares() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + std::fs::create_dir_all(share_dir.join("album")).unwrap(); + std::fs::create_dir_all(share_dir.join("other")).unwrap(); + std::fs::write(share_dir.join("album").join("one.flac"), b"xxxx").unwrap(); + std::fs::write(share_dir.join("album").join("two.flac"), b"yy").unwrap(); + std::fs::write(share_dir.join("other").join("three.flac"), b"z").unwrap(); + let root = share_dir.file_name().unwrap().to_str().unwrap().to_string(); + + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.listening_settings("e2e_folder_sharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + let server_addr = format!("{}:{}", server.host, server.port); + let _qt = login_raw(&server_addr, "e2e_folder_browser", "pw") + .expect("third-party client logs in"); + + let folder = format!("{root}\\album"); + let mut p = connect_retry( + &format!("127.0.0.1:{sharer_port}"), + Duration::from_secs(5), + ) + .expect("dial the sharer"); + p.set_read_timeout(Some(Duration::from_secs(15))).unwrap(); + p.write_all(&peer_init_bytes("e2e_folder_browser", "P", 0)) + .unwrap(); + let mut request = Message::new(); + request + .write_int32(36) + .write_int32(77) + .write_string(&folder); + p.write_all(&request.get_buffer()).unwrap(); + p.flush().unwrap(); + + let mut response = expect_code(&mut p, 37, Duration::from_secs(15)) + .expect("the sharer answers with a FolderContentsResponse"); + response.set_pointer(8); + let (token, echoed, directories) = + soulseek_rs::message::peer::parse_folder_contents(&mut response) + .expect("a well-formed folder listing"); + assert_eq!(token, 77); + assert_eq!(echoed, folder); + let mut names: Vec<(String, Vec)> = directories + .into_iter() + .map(|d| (d.name, d.files.into_iter().map(|(name, _)| name).collect())) + .collect(); + for (_, files) in &mut names { + files.sort(); + } + assert_eq!( + names, + [(folder, vec!["one.flac".to_string(), "two.flac".to_string()])], + "only the requested folder, with its files" + ); + + let _ = std::fs::remove_dir_all(share_dir); +} + +// "User info" in SoulseekQt and Nicotine+ asks the peer directly (peer code +// 15); a peer that never answers shows an endless spinner. +#[test] +fn a_third_party_client_reads_our_user_info() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + std::fs::write(share_dir.join("probe.flac"), b"xxxx").unwrap(); + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.listening_settings("e2e_info_sharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + sharer.set_upload_slots(3); + let server_addr = format!("{}:{}", server.host, server.port); + let _qt = login_raw(&server_addr, "e2e_info_asker", "pw") + .expect("third-party client logs in"); + + let mut p = connect_retry( + &format!("127.0.0.1:{sharer_port}"), + Duration::from_secs(5), + ) + .expect("dial the sharer"); + p.set_read_timeout(Some(Duration::from_secs(15))).unwrap(); + p.write_all(&peer_init_bytes("e2e_info_asker", "P", 0)) + .unwrap(); + p.write_all(&Message::new().write_int32(15).get_buffer()) + .unwrap(); + p.flush().unwrap(); + + let mut reply = expect_code(&mut p, 16, Duration::from_secs(15)) + .expect("the sharer answers with a UserInfoResponse"); + reply.set_pointer(8); + let description = reply.read_string(); + let has_picture = reply.read_bool(); + let upload_slots = reply.read_int32(); + let queue_size = reply.read_int32(); + let slots_free = reply.read_bool(); + assert!(description.is_empty(), "no description is configured"); + assert!(!has_picture); + assert_eq!(upload_slots, 3, "the configured slot count"); + assert_eq!(queue_size, 0, "nobody is waiting"); + assert!(slots_free, "nothing is uploading"); + + let _ = std::fs::remove_dir_all(share_dir); +} + +// Progress assumed every read filled the buffer, so a peer that trickles +// small chunks was reported at several times its real rate. +#[test] +fn download_progress_reports_the_rate_actually_received() { + let server = server_or_skip!(); + + let listen_port = free_port().expect("free listen port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_rate_dl", + "pw", + listen_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let filename = "trickle.mp3"; + let content: Vec = (0..400_000u32).map(|i| (i % 251) as u8).collect(); + let download_dir = unique_download_dir(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let cfg = MockUpload { + listen_addr: format!("127.0.0.1:{listen_port}"), + peer_username: "e2e_rate_up".to_string(), + filename: filename.to_string(), + content: content.clone(), + token: 424_260_u32, + ready: ready_tx, + token_delay: Duration::ZERO, + }; + let uploader = + std::thread::spawn(move || run_mock_slow_uploader(&cfg, None)); + 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_rate_up".to_string(), + content.len() as u64, + download_dir.display().to_string(), + ) + .expect("start download"); + + let mut fastest = 0.0f64; + let mut completed = false; + let deadline = Instant::now() + Duration::from_secs(30); + while Instant::now() < deadline && !completed { + match status_rx.recv_timeout(Duration::from_millis(500)) { + Ok(DownloadStatus::InProgress { + speed_bytes_per_sec, + .. + }) => fastest = fastest.max(speed_bytes_per_sec), + Ok(DownloadStatus::Completed) => completed = true, + Ok(DownloadStatus::Failed(_) | DownloadStatus::TimedOut) => break, + _ => {} + } + } + assert!(completed, "the trickled download should complete"); + assert!(fastest > 0.0, "progress should report a rate"); + // The mock sends 4 KiB every 20 ms. + let sent_rate = 4096.0 / 0.020; + assert!( + fastest < 1.5 * sent_rate, + "reported {fastest} B/s for a peer sending {sent_rate} B/s" + ); + + let _ = uploader.join(); + let _ = std::fs::remove_dir_all(&download_dir); +} + +// A listing went out in zlib "stored" blocks, so a big share cost every +// browser its full uncompressed size; names repeat enough to shrink well. +#[test] +fn a_large_listing_travels_compressed() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + let album = share_dir.join("Artist - Album"); + std::fs::create_dir_all(&album).unwrap(); + let mut raw_bytes = 0; + for n in 0..3000 { + let name = format!("Artist - Album - {n:04} - Track Title.flac"); + raw_bytes += name.len() + 21; + std::fs::write(album.join(name), b"x").unwrap(); + } + + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.listening_settings("e2e_zlib_sharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + let server_addr = format!("{}:{}", server.host, server.port); + let _qt = login_raw(&server_addr, "e2e_zlib_browser", "pw") + .expect("third-party client logs in"); + + let mut p = connect_retry( + &format!("127.0.0.1:{sharer_port}"), + Duration::from_secs(5), + ) + .expect("dial the sharer"); + p.set_read_timeout(Some(Duration::from_secs(15))).unwrap(); + p.write_all(&peer_init_bytes("e2e_zlib_browser", "P", 0)) + .unwrap(); + p.write_all(&MessageFactory::build_get_share_file_list().get_buffer()) + .unwrap(); + p.flush().unwrap(); + let mut response = expect_code(&mut p, 5, Duration::from_secs(15)) + .expect("the sharer answers the browse"); + + let frame = response.get_size(); + assert!( + frame < raw_bytes / 4, + "a {raw_bytes}-byte listing travelled as {frame} bytes" + ); + response.set_pointer(8); + let directories = + soulseek_rs::message::peer::parse_shared_file_list(&mut response); + assert_eq!( + directories.iter().map(|d| d.files.len()).sum::(), + 3000 + ); + + let _ = std::fs::remove_dir_all(share_dir); +} + +// The server's announcements (code 66) were dropped on the floor; they are +// how a server tells everyone it is going down. +#[test] +fn a_server_announcement_arrives_as_a_message_from_the_server() { + let server = server_or_skip!(); + + let mut admin = Client::with_settings(server.settings("e2e_admin", "pw")); + let mut bob = + Client::with_settings(server.settings("e2e_announce_bob", "pw")); + admin.connect().expect("admin connect"); + bob.connect().expect("bob connect"); + assert!(admin.login().expect("admin login")); + assert!(bob.login().expect("bob login")); + if !grant(&server, "e2e_admin", "admin") { + println!("e2e skipped: cannot make an admin on this server"); + return; + } + + admin + .send_private_message("server", "announcement going down at nine") + .expect("send the admin command"); + + let deadline = Instant::now() + Duration::from_secs(5); + let mut received = Vec::new(); + while Instant::now() < deadline { + received.extend(bob.take_private_messages()); + if received.iter().any(|m| m.username() == "server") { + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + let announcement = received + .iter() + .find(|m| m.username() == "server") + .expect("bob should hear the announcement"); + assert_eq!(announcement.message(), "going down at nine"); +} diff --git a/website/src/components/layout/footer.tsx b/website/src/components/layout/footer.tsx index d6a856f..a82e19f 100644 --- a/website/src/components/layout/footer.tsx +++ b/website/src/components/layout/footer.tsx @@ -20,6 +20,7 @@ const COLUMNS: readonly FooterColumn[] = [ links: [ { label: "GitHub", href: LINKS.gh }, { label: "Issues", href: LINKS.issues }, + { label: "Contributing", href: LINKS.contributing }, { label: "Releases", href: LINKS.releases }, { label: "Changelog", href: LINKS.changelog }, ], diff --git a/website/src/lib/links.ts b/website/src/lib/links.ts index be69ec3..ca09fe0 100644 --- a/website/src/lib/links.ts +++ b/website/src/lib/links.ts @@ -3,6 +3,7 @@ export const LINKS = { issues: 'https://github.com/michel/soulseek-rs/issues', releases: 'https://github.com/michel/soulseek-rs/releases', changelog: 'https://github.com/michel/soulseek-rs/blob/master/CHANGELOG.md', + contributing: 'https://github.com/michel/soulseek-rs/blob/master/CONTRIBUTING.md', license: 'https://github.com/michel/soulseek-rs/blob/master/LICENSE', cratesClient: 'https://crates.io/crates/soulseek-rs', cratesLib: 'https://crates.io/crates/soulseek-rs-lib', diff --git a/website/src/pages/community.tsx b/website/src/pages/community.tsx index a70d2bc..08964d8 100644 --- a/website/src/pages/community.tsx +++ b/website/src/pages/community.tsx @@ -239,10 +239,15 @@ export const Community = () => { open-source Soulseek server; run it locally to develop against a real protocol implementation without touching the public network.

+

+ Pull requests go to develop; master only moves on a + release. +

+