dumbgoy/soulseek-rs-modded
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge branch 'develop'

Browse Source
This commit is contained in:
Michel de Graaf
2026-09-07 18:02:27 +02:00
parent 6d47b3e594 cbf7813cf6
commit adf5bc906e
66 changed files with 1817 additions and 629 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CONTRIBUTING.md
+43
View File
@@ -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.
README.md
+6
View File
@@ -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).
soulseek-rs-lib/examples/stress.rs
+9 -5
View File
@@ -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);
soulseek-rs-lib/src/actor/peer_actor.rs
+37 -26
View File
@@ -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<SharedDirectory>),
/// 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<SharedDirectory>) {
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())
);
soulseek-rs-lib/src/actor/server_actor.rs
+41 -322
View File
@@ -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<bool>,
}
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<u32> = 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<u32> = no_listen.iter().map(code_of).collect();
assert_eq!(codes, vec![35, 71, 28]);
}
}
mod tests;
soulseek-rs-lib/src/actor/server_actor/tests.rs
+210
View File
@@ -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<u32> = 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<u32> = no_listen.iter().map(code_of).collect();
assert_eq!(codes, vec![35, 71, 28]);
}
soulseek-rs-lib/src/actor/server_actor/types.rs
+99
View File
@@ -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<bool>,
}
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
}
}
soulseek-rs-lib/src/client/mod.rs
+33 -3
View File
@@ -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<crate::message::Message> {
let matches = shares.search(query);
if matches.is_empty() {
@@ -146,6 +160,7 @@ fn build_search_response(
}
let entries: Vec<FileEntry> = 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,
soulseek-rs-lib/src/client/operations.rs
+75 -28
View File
@@ -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::<Vec<_>>();
(
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<RwLock<ClientContext>>,
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<RwLock<ClientContext>>) {
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<crate::message::peer::SharedDirectory> {
ctx.shares
.directories()
.into_iter()
.map(|(name, files)| crate::message::peer::SharedDirectory {
name,
files,
})
.collect()
}
soulseek-rs-lib/src/client/search.rs
+2 -3
View File
@@ -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(),
soulseek-rs-lib/src/client/tests.rs
+7 -3
View File
@@ -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);
}
soulseek-rs-lib/src/client/upload_queue.rs
+5 -1
View File
@@ -123,7 +123,7 @@ impl ClientContext {
mut next_token: impl FnMut() -> u32,
) -> (Option<PeerRegistry>, Vec<UploadOffer>) {
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<UploadInfo> {
soulseek-rs-lib/src/client/uploads.rs
+21 -2
View File
@@ -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.
soulseek-rs-lib/src/dispatcher.rs
+37
View File
@@ -37,3 +37,40 @@ impl<Op> MessageDispatcher<Op> {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::message::MessageHandler;
use std::sync::mpsc::channel;
struct Echo(u32);
impl MessageHandler<u32> for Echo {
fn get_code(&self) -> u32 {
self.0
}
fn handle(&self, _: &mut Message, sender: Sender<u32>) {
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::<Vec<_>>(), [1001]);
}
}
soulseek-rs-lib/src/message/handlers.rs
+3 -3
View File
@@ -4,11 +4,11 @@ use crate::message::Message;
use std::sync::mpsc::Sender;
pub trait MessageHandler<Op>: Send {
fn get_code(&self) -> u8;
fn get_code(&self) -> u32;
fn handle(&self, message: &mut Message, sender: Sender<Op>);
}
pub struct Handlers<Op> {
handlers: HashMap<u8, Box<dyn MessageHandler<Op> + Send>>,
handlers: HashMap<u32, Box<dyn MessageHandler<Op> + Send>>,
}
impl<Op> Default for Handlers<Op> {
@@ -35,7 +35,7 @@ impl<Op> Handlers<Op> {
#[must_use]
pub fn get_handler(
&self,
code: u8,
code: u32,
) -> Option<&(dyn MessageHandler<Op> + Send)> {
self.handlers.get(&code).map(|v| &**v)
}
soulseek-rs-lib/src/message/mod.rs
+17 -9
View File
@@ -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]
soulseek-rs-lib/src/message/peer/file_search_response.rs
+24 -5
View File
@@ -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<PeerMessage> for FileSearchResponse {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
9
}
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
@@ -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)
);
}
soulseek-rs-lib/src/message/peer/folder_contents.rs
+68
View File
@@ -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<PeerMessage> for FolderContentsRequest {
fn get_code(&self) -> u32 {
36
}
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
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<SharedDirectory>)> {
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))
);
}
soulseek-rs-lib/src/message/peer/get_share_file_list.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use std::sync::mpsc::Sender;
/// owns the shares) builds the real SharedFileListResponse in reply.
pub struct GetShareFileList;
impl MessageHandler<PeerMessage> for GetShareFileList {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
4
}
fn handle(&self, _message: &mut Message, sender: Sender<PeerMessage>) {
soulseek-rs-lib/src/message/peer/mod.rs
+6
View File
@@ -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};
soulseek-rs-lib/src/message/peer/peer_init.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use crate::{
pub struct PeerInit;
impl MessageHandler<PeerMessage> for PeerInit {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
1
}
soulseek-rs-lib/src/message/peer/place_in_queue_request.rs
+1 -1
View File
@@ -12,7 +12,7 @@ use std::sync::mpsc::Sender;
pub struct PlaceInQueueRequest;
impl MessageHandler<PeerMessage> for PlaceInQueueRequest {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
51
}
soulseek-rs-lib/src/message/peer/place_in_queue_response.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct PlaceInQueueResponse;
impl MessageHandler<PeerMessage> for PlaceInQueueResponse {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
44
}
soulseek-rs-lib/src/message/peer/queue_upload.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use std::sync::mpsc::Sender;
pub struct QueueUploadHandler;
impl MessageHandler<PeerMessage> for QueueUploadHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
43
}
soulseek-rs-lib/src/message/peer/shared_file_list.rs
+39 -28
View File
@@ -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<PeerMessage> for SharedFileListResponseHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
5
}
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
@@ -30,6 +30,29 @@ impl MessageHandler<PeerMessage> 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<SharedDirectory> {
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<SharedDirectory> {
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<SharedDirectory> {
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<SharedDirectory> {
dirs
}
/// Inflate a compressed peer payload positioned at the blob.
pub fn decompress_body(message: &mut Message) -> Option<Message> {
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);
});
soulseek-rs-lib/src/message/peer/transfer_request.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct TransferRequest;
impl MessageHandler<PeerMessage> for TransferRequest {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
40
}
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
soulseek-rs-lib/src/message/peer/transfer_response.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct TransferResponse;
impl MessageHandler<PeerMessage> for TransferResponse {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
41
}
soulseek-rs-lib/src/message/peer/upload_denied.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct UploadDeniedHandler;
impl MessageHandler<PeerMessage> for UploadDeniedHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
50
}
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
soulseek-rs-lib/src/message/peer/upload_failed.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct UploadFailedHandler;
impl MessageHandler<PeerMessage> for UploadFailedHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
46
}
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
soulseek-rs-lib/src/message/peer/user_info.rs
+33
View File
@@ -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<PeerMessage> for UserInfoRequest {
fn get_code(&self) -> u32 {
15
}
fn handle(&self, _message: &mut Message, sender: Sender<PeerMessage>) {
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()
}
soulseek-rs-lib/src/message/server/admin_message.rs
+30
View File
@@ -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<ServerMessage> for AdminMessageHandler {
fn get_code(&self) -> u32 {
66
}
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
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));
}
}
soulseek-rs-lib/src/message/server/connect_to_peer.rs
+1 -1
View File
@@ -5,7 +5,7 @@ use std::sync::mpsc::Sender;
pub struct ConnectToPeerHandler;
impl MessageHandler<ServerMessage> for ConnectToPeerHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
18
}
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
soulseek-rs-lib/src/message/server/excluded_search_phrases.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use std::sync::mpsc::Sender;
pub struct ExcludedSearchPhrasesHandler;
impl MessageHandler<ServerMessage> for ExcludedSearchPhrasesHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
160
}
soulseek-rs-lib/src/message/server/file_search.rs
+1 -1
View File
@@ -9,7 +9,7 @@ use crate::{
pub struct FileSearchHandler;
impl MessageHandler<ServerMessage> for FileSearchHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
26
}
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
soulseek-rs-lib/src/message/server/get_peer_address.rs
+1 -1
View File
@@ -5,7 +5,7 @@ use std::sync::mpsc::Sender;
pub struct GetPeerAddressHandler;
impl MessageHandler<ServerMessage> for GetPeerAddressHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
3
}
soulseek-rs-lib/src/message/server/join_room.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use std::sync::mpsc::Sender;
pub struct JoinRoomHandler;
impl MessageHandler<ServerMessage> for JoinRoomHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
14
}
soulseek-rs-lib/src/message/server/leave_room.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct LeaveRoomHandler;
impl MessageHandler<ServerMessage> for LeaveRoomHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
15
}
soulseek-rs-lib/src/message/server/login.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use crate::message::MessageHandler;
pub struct LoginHandler;
impl MessageHandler<ServerMessage> for LoginHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
1
}
soulseek-rs-lib/src/message/server/message_factory.rs
+20 -1
View File
@@ -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<u8> = [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);
}
soulseek-rs-lib/src/message/server/message_user.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use std::sync::mpsc::Sender;
pub struct MessageUser;
impl MessageHandler<ServerMessage> for MessageUser {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
22
}
soulseek-rs-lib/src/message/server/mod.rs
+2
View File
@@ -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;
soulseek-rs-lib/src/message/server/parent_min_speed.rs
+1 -1
View File
@@ -9,7 +9,7 @@ use crate::{
pub struct ParentMinSpeedHandler;
impl MessageHandler<ServerMessage> for ParentMinSpeedHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
83
}
soulseek-rs-lib/src/message/server/parent_speed_ratio.rs
+1 -1
View File
@@ -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<ServerMessage> for ParentSpeedRatioHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
84
}
soulseek-rs-lib/src/message/server/privileged_users.rs
+2 -2
View File
@@ -14,7 +14,7 @@ use crate::{
pub struct PrivilegedUsersHandler;
impl MessageHandler<ServerMessage> for PrivilegedUsersHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
69
}
@@ -37,7 +37,7 @@ impl MessageHandler<ServerMessage> for PrivilegedUsersHandler {
pub struct CheckPrivilegesHandler;
impl MessageHandler<ServerMessage> for CheckPrivilegesHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
92
}
soulseek-rs-lib/src/message/server/relogged.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct ReloggedHandler;
impl MessageHandler<ServerMessage> for ReloggedHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
41
}
soulseek-rs-lib/src/message/server/room_list.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use std::sync::mpsc::Sender;
pub struct RoomListHandler;
impl MessageHandler<ServerMessage> for RoomListHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
64
}
soulseek-rs-lib/src/message/server/say_chatroom.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct SayChatroomHandler;
impl MessageHandler<ServerMessage> for SayChatroomHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
13
}
soulseek-rs-lib/src/message/server/user_info.rs
+2 -2
View File
@@ -12,7 +12,7 @@ use std::sync::mpsc::Sender;
pub struct GetUserStatusHandler;
impl MessageHandler<ServerMessage> for GetUserStatusHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
7
}
@@ -33,7 +33,7 @@ impl MessageHandler<ServerMessage> for GetUserStatusHandler {
pub struct GetUserStatsHandler;
impl MessageHandler<ServerMessage> for GetUserStatsHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
36
}
soulseek-rs-lib/src/message/server/user_joined_room.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct UserJoinedRoomHandler;
impl MessageHandler<ServerMessage> for UserJoinedRoomHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
16
}
soulseek-rs-lib/src/message/server/user_left_room.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use std::sync::mpsc::Sender;
pub struct UserLeftRoomHandler;
impl MessageHandler<ServerMessage> for UserLeftRoomHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
17
}
soulseek-rs-lib/src/message/server/watch_user.rs
+1 -1
View File
@@ -21,7 +21,7 @@ use std::sync::mpsc::Sender;
pub struct WatchUserHandler;
impl MessageHandler<ServerMessage> for WatchUserHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
5
}
soulseek-rs-lib/src/message/server/wish_list_interval.rs
+1 -1
View File
@@ -13,7 +13,7 @@ use crate::{
pub struct WishListIntervalHandler;
impl MessageHandler<ServerMessage> for WishListIntervalHandler {
fn get_code(&self) -> u8 {
fn get_code(&self) -> u32 {
104
}
soulseek-rs-lib/src/peer/download_peer.rs
+12 -8
View File
@@ -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<RwLock<ClientContext>>,
download: Option<Download>,
) -> 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();
}
soulseek-rs-lib/src/peer/listen.rs
+1 -1
View File
@@ -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,
);
soulseek-rs-lib/src/peer/mod.rs
+1 -1
View File
@@ -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();
soulseek-rs-lib/src/peer/upload_peer.rs
+6 -3
View File
@@ -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<u64> {
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)]
soulseek-rs-lib/src/shares/mod.rs
+19 -4
View File
@@ -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();
soulseek-rs-lib/src/types.rs
+11 -3
View File
@@ -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<u8> = 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");
soulseek-rs-lib/src/utils/keepalive.rs
+99
View File
@@ -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::<libc::c_int>() 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::<libc::c_int>() 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));
}
soulseek-rs-lib/src/utils/logger.rs
+37 -14
View File
@@ -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<Vec<String>> = 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() {
soulseek-rs-lib/src/utils/mod.rs
+1
View File
@@ -1,5 +1,6 @@
#[macro_use]
pub mod logger;
pub(crate) mod keepalive;
pub mod lock;
pub mod md5;
pub mod path;
soulseek-rs-lib/src/utils/zlib.rs
+235 -74
View File
@@ -59,7 +59,8 @@ impl BitReader {
use crate::error::{Result, SoulseekRs};
pub fn deflate(input: &[u8]) -> Result<Vec<u8>> {
/// Inflate a zlib stream, checking its adler32 trailer.
pub fn inflate(input: &[u8]) -> Result<Vec<u8>> {
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<Vec<u8>> {
"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<u8> {
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<Vec<u8>, String> {
fn inflate_blocks(r: &mut BitReader) -> std::result::Result<Vec<u8>, 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<u8>,
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<u8> {
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<u8> {
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<u8> {
(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<u8> {
(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<u8> = (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();
soulseek-rs-lib/tests/e2e.rs
+491 -52
View File
@@ -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<Message> {
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<Message> {
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<soulseek_rs::SearchResult> {
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<u32> {
})
}
/// 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<String>)> = 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<u8> = (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::<usize>(),
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");
}
website/src/components/layout/footer.tsx
+1
View File
@@ -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 },
],
website/src/lib/links.ts
+1
View File
@@ -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',
website/src/pages/community.tsx
+5
View File
@@ -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.
</p>
<p>
Pull requests go to <Code>develop</Code>; <Code>master</Code> only moves on a
release.
</p>
<div className="mt-1 flex flex-wrap gap-2 sm:gap-3">
<Button href={LINKS.issues} variant="accent">
Open an issue
</Button>
<Button href={LINKS.contributing}>Contributing</Button>
<Button href={LINKS.gh}>GitHub</Button>
</div>
</Prose>