diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5ef0829..6241a96 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -127,3 +127,48 @@ jobs: SOULSEEK_E2E_REQUIRED: "1" SOULSEEK_E2E_MODE: daemon run: cargo test -p soulseek-rs --test cli_e2e -- --nocapture --test-threads=1 + + # A mixed load of mock peers around one real client. Every ceiling this + # project has hit only showed under simultaneous searches, downloads, + # uploads and browses, and three of the four lost work silently, so the + # gate is the functional dimensions at 100%, never the throughput number. + stress: + name: Stress (soulfind) + timeout-minutes: 20 + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Setup Rust + uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + - name: Install SQLite runtime/dev libraries + run: sudo apt-get update && sudo apt-get install -y libsqlite3-dev + + - name: Install D toolchain (LDC) + uses: dlang-community/setup-dlang@v2 + with: + compiler: ldc-latest + + - name: Build soulfind server + run: | + git clone --depth 1 https://github.com/soulfind-dev/soulfind.git soulfind + dub build :server --root soulfind + test -x soulfind/bin/soulfind + echo "SOULFIND_BIN=$PWD/soulfind/bin/soulfind" >> "$GITHUB_ENV" + + # An eighth of the local profile: soulfind on a four-core runner brings + # about 150 mock logins online, and the point is lost work, not load. + - name: Run the stress benchmark with the functional gate + env: + STRESS_REQUIRE_FUNCTIONAL: "1" + STRESS_PEERS: "128" + STRESS_SEARCHES: "32" + STRESS_DOWNLOADS: "128" + STRESS_LEECHERS: "64" + STRESS_BROWSES: "32" + run: | + ulimit -n 20000 + cargo run --release -p soulseek-rs-lib --example stress diff --git a/README.md b/README.md index 3628402..0dd16d2 100644 --- a/README.md +++ b/README.md @@ -344,7 +344,7 @@ the message field empty. `serve`'s `status` is `queued`, `uploading`, | Command | Extra fields | | ------------------ | ------------------------------------------------------------------------------------------------------------------------------- | | `search` | `duration`, `slots`, `speed`, `free_slot` | -| `browse` | `directory` | +| `browse` | `directory`, `bitrate`, `duration` | | `download` / `get` | `user`, remote `path`, `size`, beside the local `file` | | `whoami` | `listening`, `listen_port`, `download_dir`, `privilege_seconds` | | `user` | `privileged`, `shared_folders` | diff --git a/benchmarks/README.md b/benchmarks/README.md index f8d7b71..248181d 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -32,6 +32,13 @@ with headroom, so the score still moves when transfer handling regresses. `--save` only overwrites `stress-baseline.json` when the run beats it, so the baseline ratchets upward. Without `--save` a run just prints the delta. +## The gate + +`STRESS_REQUIRE_FUNCTIONAL=1` makes the run exit 1 when any of the four +functional dimensions is below 100%. CI runs it that way on every pull request +with an eighth of the default load; throughput never fails the gate because it +measures the machine, not the client. + ## Tuning the load Every knob is an env var; the defaults are in `Config::from_env`. diff --git a/soulseek-rs-lib/Cargo.toml b/soulseek-rs-lib/Cargo.toml index 34c8298..ebc7d14 100644 --- a/soulseek-rs-lib/Cargo.toml +++ b/soulseek-rs-lib/Cargo.toml @@ -21,3 +21,7 @@ workspace = true [target.'cfg(unix)'.dependencies] libc = "0.2.189" + +[[example]] +name = "stress" +test = true diff --git a/soulseek-rs-lib/README.md b/soulseek-rs-lib/README.md index 0ff5624..33a7756 100644 --- a/soulseek-rs-lib/README.md +++ b/soulseek-rs-lib/README.md @@ -115,6 +115,6 @@ fn main() -> Result<(), Box> { } ``` -`ServerMessage` and `ClientOperation` are `#[non_exhaustive]`: every protocol -message the client learns to handle adds a variant, so match them with a -wildcard arm. +`ClientOperation` is `#[non_exhaustive]`: every protocol message the client +learns to handle adds a variant, so match it with a wildcard arm. `SharedFile` +is too, so build it with the `..` syntax rather than listing every field. diff --git a/soulseek-rs-lib/examples/stress.rs b/soulseek-rs-lib/examples/stress.rs index a14f750..6952ce1 100644 --- a/soulseek-rs-lib/examples/stress.rs +++ b/soulseek-rs-lib/examples/stress.rs @@ -17,6 +17,7 @@ //! STRESS_FILE_KB size of every transferred file //! STRESS_TIMEOUT seconds before the run is called off //! STRESS_SAVE=1 store this run as the new baseline +//! STRESS_REQUIRE_FUNCTIONAL=1 exit 1 if any transfer, search or browse was lost use std::collections::HashMap; use std::io::{Read, Write}; @@ -29,12 +30,24 @@ use std::time::{Duration, Instant}; use soulseek_rs::message::Message; use soulseek_rs::message::peer::{ - FileEntry, SharedDirectory, build_file_search_response, + FileEntry, SharedDirectory, SharedFileEntry, build_file_search_response, build_shared_file_list, }; use soulseek_rs::message::server::MessageFactory; use soulseek_rs::{Client, ClientSettings, ClientVersion, PeerAddress}; +/// A mock's `(name, size)` files in listing form; mocks advertise no attributes. +fn entries(files: &[(String, u64)]) -> Vec { + files + .iter() + .map(|(name, size)| SharedFileEntry { + name: name.clone(), + size: *size, + attributes: Vec::new(), + }) + .collect() +} + /// Files matching this tag are what every mock seeder answers searches for. const TAG: &str = "zqxstress"; @@ -495,18 +508,7 @@ impl SeederState { Err(_) => return, }; match msg.get_message_code() { - // GetShareFileList: reply with everything we hold. - 4 => { - let dir = SharedDirectory { - name: format!("stress\\{}", self.name), - files: self.files.clone(), - }; - let list = - build_shared_file_list(std::slice::from_ref(&dir)); - if !self.reply(&mut p, shared, &list.get_buffer()) { - return; - } - } + 4 if !self.send_listing(&mut p, shared) => return, // QueueUpload: offer the transfer, then stream it once allowed. 43 => { msg.set_pointer(8); @@ -526,7 +528,7 @@ impl SeederState { if !self.reply(&mut p, shared, &tr.get_buffer()) { return; } - if !self.wait_for_allow(&mut p, token) { + if !self.wait_for_allow(&mut p, shared, token) { return; } let content = self.content.clone(); @@ -542,8 +544,25 @@ impl SeederState { } } - /// Wait for the client's `TransferResponse` allowing `token`. - fn wait_for_allow(&self, p: &mut TcpStream, token: u32) -> bool { + /// Everything we hold, the answer to a GetShareFileList. + fn send_listing(&self, p: &mut TcpStream, shared: bool) -> bool { + let dir = SharedDirectory { + name: format!("stress\\{}", self.name), + files: entries(&self.files), + }; + let list = build_shared_file_list(std::slice::from_ref(&dir)); + self.reply(p, shared, &list.get_buffer()) + } + + /// Wait for the client's `TransferResponse` allowing `token`. A browse + /// that arrives meanwhile is answered: a busy client accepts an offer + /// late, and a browse it sent in between is not lost work on its side. + fn wait_for_allow( + &self, + p: &mut TcpStream, + shared: bool, + token: u32, + ) -> bool { let deadline = Instant::now() + Duration::from_secs(30); while Instant::now() < deadline && !self.stop.load(Ordering::Relaxed) { match read_framed(p) { @@ -553,6 +572,12 @@ impl SeederState { return true; } } + Ok(msg) + if msg.get_message_code() == 4 + && !self.send_listing(p, shared) => + { + return false; + } Ok(_) => {} Err(ref e) if is_timeout(e) => {} Err(_) => return false, @@ -766,6 +791,18 @@ struct Score { parts: Vec<(&'static str, f64, f64)>, // name, achieved 0-1, weight } +/// The functional dimensions that fell short of 100%. Throughput is excluded: +/// it measures how fast the machine is, the others measure whether the client +/// silently lost work, which is the only failure a gate should stop on. +fn gate_failures(score: &Score) -> Vec { + score + .parts + .iter() + .filter(|(name, value, _)| *name != "throughput" && *value < 1.0) + .map(|(name, value, _)| format!("{name} {:.1}%", value * 100.0)) + .collect() +} + fn compute_score(m: &Metrics, elapsed: Duration) -> Score { let bytes = Metrics::get(&m.bytes_downloaded) + Metrics::get(&m.bytes_uploaded); @@ -1248,4 +1285,38 @@ fn main() { let _ = std::fs::remove_dir_all(&share_dir); let _ = std::fs::remove_dir_all(&download_dir); + + let failures = gate_failures(&score); + if std::env::var("STRESS_REQUIRE_FUNCTIONAL").is_ok_and(|v| v == "1") + && !failures.is_empty() + { + eprintln!("stress: work was lost: {}", failures.join(", ")); + std::process::exit(1); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn score(parts: &[(&'static str, f64)]) -> Score { + Score { + total: 0.0, + parts: parts.iter().map(|&(n, v)| (n, v, 0.2)).collect(), + } + } + + #[test] + fn lost_work_fails_the_gate_but_slow_transfers_do_not() { + let clean = + score(&[("downloads", 1.0), ("uploads", 1.0), ("throughput", 0.1)]); + assert!(gate_failures(&clean).is_empty()); + + let lossy = score(&[ + ("downloads", 0.98), + ("uploads", 1.0), + ("throughput", 1.0), + ]); + assert_eq!(gate_failures(&lossy), ["downloads 98.0%"]); + } } diff --git a/soulseek-rs-lib/src/actor/mod.rs b/soulseek-rs-lib/src/actor/mod.rs index d9eb252..eed1cfd 100644 --- a/soulseek-rs-lib/src/actor/mod.rs +++ b/soulseek-rs-lib/src/actor/mod.rs @@ -52,7 +52,7 @@ impl ActorHandle { } /// Internal actor message wrapper -pub(crate) enum ActorMessage { +pub enum ActorMessage { UserMessage(M), Stop, } diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index faf21e2..bf63e3a 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -61,6 +61,9 @@ pub enum PeerMessage { size: u64, }, ProcessRead, + /// A newer connection to this peer took over. Deliver what is already + /// on the wire, then close. + Retire, } pub struct PeerActor { @@ -101,6 +104,10 @@ pub struct PeerActor { /// control connection, and each one holds a descriptor and a thread for /// the life of the process. last_activity: Instant, + /// Set once a newer connection to the same peer replaced this one. The + /// stream stays open until then so a reply already in flight, a browse + /// listing say, still arrives; past it the actor closes and stops. + retire_deadline: Option, } /// Cap on queued outbound bytes. A peer that stops reading must not be able to @@ -112,6 +119,10 @@ const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024; /// simply reconnects — every serious client reaps idle peers this way. Five /// minutes comfortably outlives a search's chatter and queue updates. const IDLE_DISCONNECT: Duration = Duration::from_mins(5); +/// How long a replaced connection keeps listening for replies in flight. +const RETIRE_GRACE: Duration = Duration::from_secs(30); +/// Socket buffers read per tick: 64 KiB, against a 100 ms tick. +const READS_PER_TICK: usize = 64; impl PeerActor { #[must_use] @@ -148,6 +159,7 @@ impl PeerActor { serving_tokens: std::collections::HashSet::new(), pending_write: Vec::new(), last_activity: Instant::now(), + retire_deadline: None, } } @@ -218,6 +230,13 @@ impl PeerActor { if matches!(self.connection_state, ConnectionState::Connecting { .. }) { match &msg { PeerMessage::SetUsername(_) | PeerMessage::ProcessRead => {} + // Nothing has gone out on a dial still in progress, so there + // is nothing to wait for. + PeerMessage::Retire => { + self.retire_deadline = Some(Instant::now()); + self.disconnect(); + return; + } _ => { self.queued_messages.push(msg); return; @@ -229,6 +248,9 @@ impl PeerActor { PeerMessage::SendMessage(message) => { self.send_message(message); } + PeerMessage::Retire => { + self.retire_deadline = Some(Instant::now() + RETIRE_GRACE); + } PeerMessage::FileSearchResult(file_search) => { self.handle_file_search_result(file_search); } @@ -484,7 +506,7 @@ impl PeerActor { return; }; - match self.reader.read_from_socket(stream) { + match Self::drain(&mut self.reader, stream) { Ok(()) => { self.last_activity = Instant::now(); } @@ -532,6 +554,25 @@ impl PeerActor { self.extract_and_process_messages(); } + /// Read what the socket holds, up to `READS_PER_TICK` buffers, so a big + /// listing lands in a few ticks rather than a few minutes; the cap keeps + /// a peer that never stops sending from starving the mailbox. Only the + /// first read reports an error: a peer that sends its reply and hangs up + /// has still sent the reply, which must be parsed before the close is + /// acted on, and the socket says so again on the next call. + fn drain( + reader: &mut MessageReader, + stream: &mut TcpStream, + ) -> io::Result<()> { + reader.read_from_socket(stream)?; + for _ in 1..READS_PER_TICK { + if reader.read_from_socket(stream).is_err() { + break; + } + } + Ok(()) + } + fn extract_and_process_messages(&mut self) { let username = self.peer_username(); let mut extracted_count = 0; @@ -664,6 +705,12 @@ impl PeerActor { } fn disconnect_with_error(&mut self, error: Error) { + // A retired connection is expected to go: the peer closing it is not + // evidence the peer is gone, and must not fail its transfers on the + // connection that replaced this one. + if self.retire_deadline.is_some() { + return self.disconnect(); + } let username = self.peer_username(); // Include the cause: a failed outbound dial reports nothing else, so // without it an unreachable peer is indistinguishable from a clean @@ -919,109 +966,27 @@ impl Actor for PeerActor { ); self.disconnect(); } + if self.stream.is_some() + && self + .retire_deadline + .is_some_and(|d| d <= Instant::now()) + { + self.disconnect(); + } + } + } + // Nothing in the registry points at a retired actor any more, so + // it is the one to stop itself. + ConnectionState::Disconnected => { + if self.retire_deadline.is_some() + && let Some(handle) = &self.self_handle + { + let _ = handle.stop(); } } - ConnectionState::Disconnected => {} } } } #[cfg(test)] -mod tests { - use super::*; - use crate::peer::{ConnectionType, Peer}; - use std::net::{TcpListener, TcpStream}; - use std::sync::mpsc::Receiver; - - /// A connected inbound actor over a real loopback socket, plus the far - /// end (kept alive so reads see silence, not EOF). - fn connected_actor() -> (PeerActor, Receiver, TcpStream) { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); - let addr = listener.local_addr().unwrap(); - let stream = TcpStream::connect(addr).unwrap(); - stream.set_nonblocking(true).unwrap(); - let (far_end, _) = listener.accept().unwrap(); - - let (mut actor, rx) = make_actor(Some(stream)); - actor.on_start(); - (actor, rx, far_end) - } - - fn make_actor( - stream: Option, - ) -> (PeerActor, Receiver) { - let (tx, rx) = std::sync::mpsc::channel(); - let peer = Peer::new( - "bob".to_string(), - ConnectionType::P, - "127.0.0.1".to_string(), - 0, - None, - 0, - 0, - 0, - ); - let actor = PeerActor::new(peer, stream, None, tx, "me".to_string(), 7); - (actor, rx) - } - - #[test] - fn a_timed_out_connect_parks_the_actor_in_disconnected() { - let (mut actor, rx) = make_actor(None); - 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" - ); - match rx.try_recv() { - Ok(ClientOperation::PeerConnectFailed(7, username)) => { - assert_eq!(username, "bob"); - } - other => panic!("expected PeerConnectFailed, got {other:?}"), - } - } - - #[test] - fn an_upload_failed_message_reaches_the_client_as_this_peer() { - let (mut actor, rx, _far_end) = connected_actor(); - - actor.handle_message(PeerMessage::UploadFailed( - String::new(), - "song.mp3".to_string(), - )); - - match rx.try_recv() { - Ok(ClientOperation::UploadFailed(username, filename)) => { - assert_eq!(username, "bob"); - assert_eq!(filename, "song.mp3"); - } - other => panic!("expected UploadFailed, got {other:?}"), - } - } - - #[test] - fn tick_reaps_a_peer_idle_past_the_deadline() { - let (mut actor, rx, _far_end) = connected_actor(); - - actor.tick(); - assert!(actor.stream.is_some(), "a fresh connection stays open"); - - actor.last_activity = Instant::now() - .checked_sub(IDLE_DISCONNECT + Duration::from_secs(1)) - .unwrap(); - actor.tick(); - - assert!(actor.stream.is_none(), "an idle stream must be closed"); - match rx.try_recv() { - Ok(ClientOperation::PeerDisconnected(7, username, None)) => { - assert_eq!(username, "bob"); - } - other => panic!("expected a clean PeerDisconnected, got {other:?}"), - } - } -} +mod tests; diff --git a/soulseek-rs-lib/src/actor/peer_actor/tests.rs b/soulseek-rs-lib/src/actor/peer_actor/tests.rs new file mode 100644 index 0000000..de8d7c8 --- /dev/null +++ b/soulseek-rs-lib/src/actor/peer_actor/tests.rs @@ -0,0 +1,235 @@ +use super::*; +use crate::message::peer::build_shared_file_list; +use crate::peer::{ConnectionType, Peer}; +use std::net::{TcpListener, TcpStream}; +use std::sync::mpsc::Receiver; + +/// A connected inbound actor over a real loopback socket, plus the far +/// end (kept alive so reads see silence, not EOF). +fn connected_actor() -> (PeerActor, Receiver, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let stream = TcpStream::connect(addr).unwrap(); + stream.set_nonblocking(true).unwrap(); + let (far_end, _) = listener.accept().unwrap(); + + let (mut actor, rx) = make_actor(Some(stream)); + actor.on_start(); + (actor, rx, far_end) +} + +fn make_actor( + stream: Option, +) -> (PeerActor, Receiver) { + let (tx, rx) = std::sync::mpsc::channel(); + let peer = Peer::new( + "bob".to_string(), + ConnectionType::P, + "127.0.0.1".to_string(), + 0, + None, + 0, + 0, + 0, + ); + let actor = PeerActor::new(peer, stream, None, tx, "me".to_string(), 7); + (actor, rx) +} + +#[test] +fn a_timed_out_connect_parks_the_actor_in_disconnected() { + let (mut actor, rx) = make_actor(None); + 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" + ); + match rx.try_recv() { + Ok(ClientOperation::PeerConnectFailed(7, username)) => { + assert_eq!(username, "bob"); + } + other => panic!("expected PeerConnectFailed, got {other:?}"), + } +} + +#[test] +fn an_upload_failed_message_reaches_the_client_as_this_peer() { + let (mut actor, rx, _far_end) = connected_actor(); + + actor.handle_message(PeerMessage::UploadFailed( + String::new(), + "song.mp3".to_string(), + )); + + match rx.try_recv() { + Ok(ClientOperation::UploadFailed(username, filename)) => { + assert_eq!(username, "bob"); + assert_eq!(filename, "song.mp3"); + } + other => panic!("expected UploadFailed, got {other:?}"), + } +} + +#[test] +fn tick_reaps_a_peer_idle_past_the_deadline() { + let (mut actor, rx, _far_end) = connected_actor(); + + actor.tick(); + assert!(actor.stream.is_some(), "a fresh connection stays open"); + + actor.last_activity = Instant::now() + .checked_sub(IDLE_DISCONNECT + Duration::from_secs(1)) + .unwrap(); + actor.tick(); + + assert!(actor.stream.is_none(), "an idle stream must be closed"); + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, username, None)) => { + assert_eq!(username, "bob"); + } + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } +} + +/// A listing well past one socket buffer, so delivery within a couple of +/// ticks proves the actor drains the socket rather than sipping 1 KiB a tick. +fn big_listing() -> Vec { + let files = (0..400u64) + .map(|i| crate::message::peer::SharedFileEntry { + name: format!("track-{i:03}-{}.flac", "x".repeat(i as usize % 17)), + size: i, + attributes: Vec::new(), + }) + .collect(); + let dir = crate::message::peer::SharedDirectory { + name: "album".to_string(), + files, + }; + let bytes = build_shared_file_list(std::slice::from_ref(&dir)).get_buffer(); + assert!(bytes.len() > 4096, "the listing must span several reads"); + bytes +} + +fn self_handle( + actor: &mut PeerActor, +) -> Receiver> { + let (tx, rx) = std::sync::mpsc::channel(); + actor.set_self_handle(ActorHandle { sender: tx }); + rx +} + +// A connection replaced by a newer one to the same peer may still carry +// a reply already on the wire: it is retired, not cut, and once nothing +// more can come it is the one to stop itself. +#[test] +fn a_retired_actor_delivers_what_is_on_the_wire_then_stops_itself() { + use std::io::Write; + let (mut actor, rx, mut far_end) = connected_actor(); + let stop_rx = self_handle(&mut actor); + + actor.handle_message(PeerMessage::Retire); + far_end.write_all(&big_listing()).unwrap(); + std::thread::sleep(Duration::from_millis(50)); + actor.tick(); + actor.tick(); + + match rx.try_recv() { + Ok(ClientOperation::BrowseResult { username, .. }) => { + assert_eq!(username, "bob"); + } + other => { + panic!("expected the listing to reach the client, got {other:?}") + } + } + assert!(actor.stream.is_some(), "retiring is not yet closing"); + assert!(stop_rx.try_recv().is_err(), "not stopped while listening"); + + actor.retire_deadline = + Some(Instant::now().checked_sub(Duration::from_secs(1)).unwrap()); + actor.tick(); + assert!(actor.stream.is_none(), "past the grace the stream closes"); + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } + + actor.tick(); + assert!( + matches!(stop_rx.try_recv(), Ok(crate::actor::ActorMessage::Stop)), + "a closed retired actor stops itself" + ); +} + +// The peer that replaced the connection considers the old one done too: it +// writes its reply and hangs up in the same breath. The reply still counts. +#[test] +fn a_reply_sent_right_before_the_peer_hangs_up_is_still_delivered() { + use std::io::Write; + let (mut actor, rx, mut far_end) = connected_actor(); + + actor.handle_message(PeerMessage::Retire); + far_end.write_all(&big_listing()).unwrap(); + drop(far_end); + std::thread::sleep(Duration::from_millis(50)); + actor.tick(); + actor.tick(); + + match rx.try_recv() { + Ok(ClientOperation::BrowseResult { username, .. }) => { + assert_eq!(username, "bob"); + } + other => panic!("expected the listing before the close, got {other:?}"), + } + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } +} + +#[test] +fn a_retired_actor_reporting_an_error_reports_a_clean_close() { + let (mut actor, rx, _far_end) = connected_actor(); + + actor.handle_message(PeerMessage::Retire); + actor.disconnect_with_error(Error::from(io::ErrorKind::ConnectionReset)); + + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!( + "a retired connection going away is not an error, got {other:?}" + ), + } +} + +// Retiring an actor whose dial has not completed: nothing went out, so +// there is nothing to wait for, and it must not linger in the queue that +// only a successful connection drains. +#[test] +fn retiring_a_connecting_actor_closes_it_at_once() { + let (mut actor, rx) = make_actor(None); + let stop_rx = self_handle(&mut actor); + actor.connection_state = ConnectionState::Connecting { + since: Instant::now(), + }; + + actor.handle_message(PeerMessage::Retire); + + assert!(matches!( + actor.connection_state, + ConnectionState::Disconnected + )); + match rx.try_recv() { + Ok(ClientOperation::PeerDisconnected(7, _, None)) => {} + other => panic!("expected a clean PeerDisconnected, got {other:?}"), + } + actor.tick(); + assert!(matches!( + stop_rx.try_recv(), + Ok(crate::actor::ActorMessage::Stop) + )); +} diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index 0ff9230..f4b7007 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -126,14 +126,14 @@ impl PeerRegistry { actor.set_self_handle(handle); }) .map_err(|e| format!("failed to spawn peer actor thread: {e}"))?; - // Stop any actor already registered under this username so it does not - // become an orphan pinning a pool worker forever. Eviction on the - // replaced actor's later shutdown is identity-aware (keyed on its id), - // so stopping it here cannot evict this new connection. + // Retire any actor already registered under this username rather than + // stop it: a reply to a request we sent on it may still be on the + // wire. It stops itself once that grace is over, and the eviction its + // shutdown triggers is keyed on its id, so it cannot evict this one. if let Some((_, old_handle)) = peers.insert(username.clone(), (id, handle.clone())) { - let _ = old_handle.stop(); + let _ = old_handle.send(PeerMessage::Retire); debug!( "[peer_registry] Replaced existing peer actor for {}", username diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 1f2620b..ecab497 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -4,6 +4,7 @@ use crate::dispatcher::MessageDispatcher; use crate::message::server::AdminMessageHandler; use crate::message::server::CheckPrivilegesHandler; use crate::message::server::ConnectToPeerHandler; +use crate::message::server::EmbeddedMessageHandler; use crate::message::server::ExcludedSearchPhrasesHandler; use crate::message::server::FileSearchHandler; use crate::message::server::GetPeerAddressHandler; @@ -14,8 +15,10 @@ use crate::message::server::MessageFactory; use crate::message::server::MessageUser; use crate::message::server::ParentMinSpeedHandler; use crate::message::server::ParentSpeedRatioHandler; +use crate::message::server::PossibleParentsHandler; use crate::message::server::PrivilegedUsersHandler; use crate::message::server::ReloggedHandler; +use crate::message::server::ResetDistributedHandler; use crate::message::server::SayChatroomHandler; use crate::message::server::UserJoinedRoomHandler; use crate::message::server::UserLeftRoomHandler; @@ -156,6 +159,11 @@ pub enum ServerMessage { room: String, username: String, }, + /// Peers the server suggests as distributed-network parents: username, + /// host, port. + PossibleParents(Vec<(String, String, u16)>), + /// The server asks us to drop our parent and start over. + ResetDistributed, } pub struct ServerActor { @@ -178,9 +186,9 @@ pub struct ServerActor { } /// The messages a client sends right after a successful login: its shared-file -/// counts, distributed-network opt-out, online status, and (when listening) the -/// port peers should connect to. Kept as a free function so it can be tested -/// without a live connection. +/// counts, online status, and (when listening) +/// the port peers should connect to. Kept as a free function so it can be +/// tested without a live connection. fn post_login_messages( enable_listen: bool, listen_port: u16, @@ -192,7 +200,6 @@ fn post_login_messages( shared_folders, shared_files, ), - MessageFactory::build_no_parent_message(), MessageFactory::build_set_status_message(2), ]; if enable_listen { @@ -291,6 +298,9 @@ impl ServerActor { handlers.register_handler(LoginHandler); handlers.register_handler(ReloggedHandler); handlers.register_handler(AdminMessageHandler); + handlers.register_handler(PossibleParentsHandler); + handlers.register_handler(ResetDistributedHandler); + handlers.register_handler(EmbeddedMessageHandler); handlers.register_handler(RoomListHandler); handlers.register_handler(GetUserStatusHandler); handlers.register_handler(WatchUserHandler); @@ -502,6 +512,14 @@ impl ServerActor { /// length; there is no behaviour here beyond dispatch. fn handle_standing_message(&mut self, message: ServerMessage) { match message { + ServerMessage::PossibleParents(candidates) => { + self.forward_to_client(ClientOperation::PossibleParents( + candidates, + )); + } + ServerMessage::ResetDistributed => { + self.forward_to_client(ClientOperation::ResetDistributed); + } ServerMessage::WishlistSearch { token, query } => { self.queue_message(MessageFactory::build_wishlist_search( token, &query, @@ -554,6 +572,9 @@ impl ServerActor { self.send_message(msg); } self.session.clear(); + // The distributed stance is the leaf's to announce, and a new + // session starts without a parent. + self.forward_to_client(ClientOperation::ResetDistributed); } match self.context.write_safe() { Ok(mut ctx) => ctx.logged_in = Some(message), diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs index bf73bbe..8faba13 100644 --- a/soulseek-rs-lib/src/actor/server_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -36,15 +36,22 @@ fn the_server_socket_keeps_itself_alive() { } fn parked_actor(port: u16) -> ServerActor { - let (tx, _rx) = std::sync::mpsc::channel(); - ServerActor::new( + parked_actor_with_client(port).0 +} + +fn parked_actor_with_client( + port: u16, +) -> (ServerActor, std::sync::mpsc::Receiver) { + let (tx, rx) = std::sync::mpsc::channel(); + let actor = ServerActor::new( PeerAddress::new("127.0.0.1".to_string(), port), tx, 0, false, 0, 0, - ) + ); + (actor, rx) } #[test] @@ -191,12 +198,28 @@ fn a_successful_login_marks_the_session_live_again() { assert_eq!(actor.session.loss(), None); } +// A new session starts without a parent: the leaf drops whatever tree it +// hung from and announces its parentless stance from there. +#[test] +fn a_successful_login_resets_the_distributed_leaf() { + let (mut actor, client) = parked_actor_with_client(1); + + actor.handle_login_status(true); + + assert!(matches!( + client.try_recv(), + Ok(ClientOperation::ResetDistributed) + )); + actor.handle_login_status(false); + assert!(client.try_recv().is_err(), "a failed login resets nothing"); +} + #[test] fn post_login_messages_carry_counts_and_conditional_wait_port() { let messages = post_login_messages(true, 4321, 3, 7); let codes: Vec = messages.iter().map(code_of).collect(); - // SharedFolders, HaveNoParent, SetStatus, SetWaitPort. - assert_eq!(codes, vec![35, 71, 28, 2]); + // SharedFolders; SetStatus; SetWaitPort. + assert_eq!(codes, vec![35, 28, 2]); // The SharedFolders message (code 35) carries the real counts. let shared = messages[0].get_data(); @@ -206,5 +229,5 @@ fn post_login_messages_carry_counts_and_conditional_wait_port() { // Not listening omits SetWaitPort (code 2). let no_listen = post_login_messages(false, 4321, 3, 7); let codes: Vec = no_listen.iter().map(code_of).collect(); - assert_eq!(codes, vec![35, 71, 28]); + assert_eq!(codes, vec![35, 28]); } diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index 8421ce9..e6d0436 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -57,6 +57,7 @@ impl Client { ctx.max_peers.clone(), ); ctx.peer_registry = Some(peer_registry); + ctx.operations = Some(sender.clone()); let listen_sender = sender.clone(); diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs new file mode 100644 index 0000000..013ecf9 --- /dev/null +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -0,0 +1,688 @@ +//! Our place in the distributed search network, as a leaf: which parent we +//! hang from, which candidates we are still sounding out, and what to tell the +//! server when that changes. + +use super::{Client, ClientOperation, Result, ServerMessage}; +use crate::message::MessageReader; +use crate::message::distributed::{self, Distributed}; +use crate::message::server::MessageFactory; +use crate::peer::ConnectionType; +use crate::utils::lock::RwLockExt; +use std::collections::HashMap; +use std::io::{self, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::Sender; +use std::time::{Duration, Instant}; + +/// A candidate that has not answered the dial in this long is not a parent. +const LINK_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// How often an idle link looks at its cancel flag. +const CANCEL_POLL: Duration = Duration::from_secs(1); +/// A parent relays the network's whole search stream; one that says nothing +/// for this long is gone, and holding its slot would block every later +/// candidate. +const LINK_IDLE: Duration = Duration::from_mins(1); +/// Moves of our branch are told to the server at most this often; a parent +/// that flaps its level must not make us flood the server. +const MOVE_INTERVAL: Duration = Duration::from_secs(1); +/// A distributed frame is a search or a branch fact; anything bigger is not +/// one of ours. +const MAX_FRAME: usize = 64 * 1024; +/// The server offers at most ten parents at a time; more candidates than +/// that in flight is a server that keeps offering, and each one is a thread. +const MAX_CANDIDATES: usize = 10; +/// Searches admitted from the parent per second. Every search costs a scan +/// of everything we share, and the tree carries the whole network's stream. +// ponytail: a flat budget; a word index over the shares is the upgrade path. +const SEARCH_BUDGET: u32 = 50; + +/// Where we sit in the tree, as told to the server. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Branch { + pub root: String, + pub level: u32, +} + +impl Branch { + #[must_use] + pub fn parentless(own: &str) -> Self { + Self::under(own, 0) + } + + #[must_use] + pub fn under(root: &str, level: u32) -> Self { + Self { + root: root.to_string(), + level, + } + } +} + +struct Candidate { + link: u64, + level: Option, + root: Option, + cancel: Arc, +} + +/// A candidate to dial: the link number that tags everything it sends, and +/// the flag that tells its thread to stop. +pub struct Dial { + pub link: u64, + pub username: String, + pub host: String, + pub port: u16, + pub cancel: Arc, +} + +/// The leaf's view: at most one parent, adopted from the candidates the +/// server offered, the way Nicotine+ does it. Every method is a pure state +/// change so the rule is testable without a socket. Events carry the link +/// number they came from, so a stale thread cannot speak for a newer link +/// to the same user. +pub struct Leaf { + own: String, + parent: Option, + branch: Branch, + candidates: HashMap, + next_link: u64, + searches: (Instant, u32), + /// When we last told the server we moved, and whether a move since then + /// is still waiting to be told. + moved: Option, + move_pending: bool, +} + +impl Leaf { + #[must_use] + pub fn new(own: &str) -> Self { + Self { + own: own.to_string(), + parent: None, + branch: Branch::parentless(own), + candidates: HashMap::new(), + next_link: 1, + searches: (Instant::now(), 0), + moved: None, + move_pending: false, + } + } + + #[must_use] + pub fn branch(&self) -> Branch { + self.branch.clone() + } + + /// Whether `link` is the live link to our parent. + #[must_use] + pub fn is_parent(&self, username: &str, link: u64) -> bool { + self.parent.as_deref() == Some(username) + && self.has_link(username, link) + } + + /// New candidates to dial: those we are not already sounding out, up to + /// the cap, and none at all while we hang from a parent. + pub fn consider( + &mut self, + candidates: Vec<(String, String, u16)>, + ) -> Vec { + if self.parent.is_some() { + return Vec::new(); + } + let mut dials = Vec::new(); + for (username, host, port) in candidates { + if self.candidates.contains_key(&username) + || self.candidates.len() >= MAX_CANDIDATES + { + continue; + } + let link = self.next_link; + self.next_link += 1; + let cancel = Arc::new(AtomicBool::new(false)); + self.candidates.insert( + username.clone(), + Candidate { + link, + level: None, + root: None, + cancel: cancel.clone(), + }, + ); + dials.push(Dial { + link, + username, + host, + port, + cancel, + }); + } + dials + } + + /// A candidate said how deep it sits. A branch root does not always say + /// so separately: level 0 is its own root. A negative level is nonsense + /// and leaves the candidate unadoptable. When it is the parent talking, + /// our own place moves with it and the new stance comes back. + pub fn branch_level( + &mut self, + username: &str, + link: u64, + level: i32, + ) -> Option { + if !self.has_link(username, link) { + return None; + } + let candidate = self.candidates.get_mut(username)?; + candidate.level = u32::try_from(level).ok(); + if level == 0 { + candidate.root = Some(username.to_string()); + } + self.moved_with_parent(username) + } + + pub fn branch_root( + &mut self, + username: &str, + link: u64, + root: &str, + ) -> Option { + if !self.has_link(username, link) { + return None; + } + self.candidates.get_mut(username)?.root = Some(root.to_string()); + self.moved_with_parent(username) + } + + /// A search came down from `username`. The first one from a candidate + /// that has told us its place adopts it; the stance to announce comes back. + pub fn search_from(&mut self, username: &str, link: u64) -> Option { + if self.parent.is_some() || !self.has_link(username, link) { + return None; + } + let candidate = self.candidates.get(username)?; + let (Some(level), Some(root)) = + (candidate.level, candidate.root.clone()) + else { + return None; + }; + let parent = self.candidates.remove(username)?; + self.cancel_candidates(); + self.candidates.insert(username.to_string(), parent); + self.parent = Some(username.to_string()); + self.branch = Branch::under(&root, level.saturating_add(1)); + Some(self.branch.clone()) + } + + /// Whether to answer one more search from the parent now: at most + /// `SEARCH_BUDGET` a second, the rest dropped. + pub fn admit_search(&mut self, now: Instant) -> bool { + if now.duration_since(self.searches.0) >= Duration::from_secs(1) { + self.searches = (now, 0); + } + self.searches.1 += 1; + self.searches.1 <= SEARCH_BUDGET + } + + /// The link to `username` closed. True when that was the parent, which + /// puts us back to parentless. + pub fn closed(&mut self, username: &str, link: u64) -> bool { + if !self.has_link(username, link) { + return false; + } + self.candidates.remove(username); + if self.parent.as_deref() != Some(username) { + return false; + } + self.parent = None; + self.branch = Branch::parentless(&self.own); + true + } + + pub fn reset(&mut self) { + self.cancel_candidates(); + self.parent = None; + self.branch = Branch::parentless(&self.own); + } + + fn has_link(&self, username: &str, link: u64) -> bool { + self.candidates + .get(username) + .is_some_and(|c| c.link == link) + } + + fn moved_with_parent(&mut self, username: &str) -> Option { + if self.parent.as_deref() != Some(username) { + return None; + } + let candidate = self.candidates.get(username)?; + let branch = Branch::under( + candidate.root.as_deref()?, + candidate.level?.saturating_add(1), + ); + if branch == self.branch { + return None; + } + self.branch = branch; + self.move_pending = true; + self.due_announcement(Instant::now()) + } + + /// The branch to tell the server about, if a move is waiting and the + /// last one was told at least [`MOVE_INTERVAL`] ago. + pub fn due_announcement(&mut self, now: Instant) -> Option { + if !self.move_pending + || self + .moved + .is_some_and(|at| now.duration_since(at) < MOVE_INTERVAL) + { + return None; + } + self.move_pending = false; + self.moved = Some(now); + Some(self.branch.clone()) + } + + fn cancel_candidates(&mut self) { + for candidate in self.candidates.values() { + candidate.cancel.store(true, Ordering::Relaxed); + } + self.candidates.clear(); + } +} + +/// Tell the server where we sit now. +pub fn announce( + server: Option<&Sender>, + branch: &Branch, + has_parent: bool, +) { + let Some(server) = server else { + return; + }; + for message in distributed::stance(&branch.root, branch.level, has_parent) { + let _ = server.send(ServerMessage::SendMessage(message)); + } +} + +/// Dial a parent candidate on its own thread: a `D` PeerInit, then whatever it +/// sends comes back as operations until the link closes or is cancelled. +pub fn spawn_link(dial: Dial, own: String, ops: Sender) { + let closed = ClientOperation::ParentClosed { + parent: dial.username.clone(), + link: dial.link, + }; + let thread_ops = ops.clone(); + let spawned = std::thread::Builder::new() + .name("soulseek-parent".to_string()) + .spawn(move || { + if let Err(e) = link(&dial, &own, &thread_ops) + && !dial.cancel.load(Ordering::Relaxed) + { + crate::debug!( + "[distributed] link to {} ended: {}", + dial.username, + e + ); + let _ = thread_ops.send(ClientOperation::ParentClosed { + parent: dial.username, + link: dial.link, + }); + } + }); + if spawned.is_err() { + let _ = ops.send(closed); + } +} + +fn link( + dial: &Dial, + own: &str, + ops: &Sender, +) -> io::Result<()> { + let addr = (dial.host.as_str(), dial.port) + .to_socket_addrs()? + .next() + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "no address") + })?; + let mut stream = TcpStream::connect_timeout(&addr, LINK_CONNECT_TIMEOUT)?; + if dial.cancel.load(Ordering::Relaxed) { + return Ok(()); + } + stream.set_read_timeout(Some(CANCEL_POLL))?; + stream.write_all( + &MessageFactory::build_peer_init_message(own, ConnectionType::D, 0) + .get_buffer(), + )?; + let mut reader = MessageReader::new(); + let mut heard = Instant::now(); + loop { + if dial.cancel.load(Ordering::Relaxed) { + return Ok(()); + } + match reader.read_from_socket(&mut stream) { + Ok(()) => heard = Instant::now(), + Err(e) + if matches!( + e.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + if heard.elapsed() > LINK_IDLE { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "silent parent", + )); + } + continue; + } + Err(e) => return Err(e), + } + while let Some(mut frame) = reader.extract_message()? { + let Some(operation) = operation_of(dial, &mut frame) else { + continue; + }; + if ops.send(operation).is_err() { + return Ok(()); + } + } + if reader.buffer_len() > MAX_FRAME { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "oversized frame", + )); + } + } +} + +fn operation_of( + dial: &Dial, + frame: &mut crate::message::Message, +) -> Option { + let parent = dial.username.clone(); + let link = dial.link; + Some(match distributed::parse(frame)? { + Distributed::BranchLevel(level) => ClientOperation::ParentBranchLevel { + parent, + link, + level, + }, + Distributed::BranchRoot(root) => { + ClientOperation::ParentBranchRoot { parent, link, root } + } + Distributed::Search { + username, + token, + query, + } => ClientOperation::ParentSearch { + parent, + link, + username, + token, + query, + }, + }) +} + +impl Client { + /// Offer parent candidates as the server does with PossibleParents: + /// `(username, host, port)`. The first candidate to send a search after + /// telling us its branch becomes our parent. + /// + /// # Errors + /// Returns [`crate::SoulseekRs::NotConnected`] before [`Client::connect`]. + pub fn consider_parents( + &self, + candidates: Vec<(String, String, u16)>, + ) -> Result<()> { + let ops = self + .context + .read_safe()? + .operations + .clone() + .ok_or(crate::SoulseekRs::NotConnected)?; + ops.send(ClientOperation::PossibleParents(candidates)) + .map_err(|_| crate::SoulseekRs::NotConnected) + } + + /// Where we sit in the distributed search network: the branch root's name + /// and our level, which is ourselves at 0 while we have no parent. + #[must_use] + pub fn distributed_branch(&self) -> (String, u32) { + self.context.read_safe().map_or_else( + |_| (self.username.clone(), 0), + |ctx| { + let branch = ctx.leaf.branch(); + (branch.root, branch.level) + }, + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn leaf() -> Leaf { + Leaf::new("me") + } + + fn candidate(name: &str) -> (String, String, u16) { + (name.to_string(), "127.0.0.1".to_string(), 2234) + } + + /// Dial the candidates and return their link numbers by name. + fn dial(leaf: &mut Leaf, names: &[&str]) -> HashMap { + leaf.consider(names.iter().map(|n| candidate(n)).collect()) + .into_iter() + .map(|d| (d.username.clone(), d)) + .collect() + } + + #[test] + fn a_leaf_without_a_parent_is_its_own_root_at_level_zero() { + assert_eq!(leaf().branch(), Branch::parentless("me")); + } + + #[test] + fn candidates_are_dialled_only_while_there_is_no_parent() { + let mut leaf = leaf(); + let dials = dial(&mut leaf, &["alice", "bob"]); + assert_eq!(dials.len(), 2); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 1); + leaf.branch_root("alice", alice, "rooty"); + assert!( + leaf.search_from("alice", alice).is_some(), + "alice becomes the parent" + ); + assert!(leaf.consider(vec![candidate("carol")]).is_empty()); + } + + #[test] + fn the_first_search_from_a_candidate_with_branch_info_adopts_it() { + let mut leaf = leaf(); + let dials = dial(&mut leaf, &["alice", "bob"]); + let (alice, bob) = (dials["alice"].link, dials["bob"].link); + assert_eq!(leaf.search_from("alice", alice), None, "no branch yet"); + leaf.branch_level("alice", alice, 3); + assert_eq!(leaf.search_from("alice", alice), None, "level, no root"); + leaf.branch_root("alice", alice, "rooty"); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("rooty", 4)), + "level 3 parent puts us at level 4" + ); + assert_eq!(leaf.branch(), Branch::under("rooty", 4)); + assert!(leaf.is_parent("alice", alice)); + assert_eq!(leaf.search_from("alice", alice), None, "already parent"); + assert_eq!(leaf.search_from("bob", bob), None, "a loser is ignored"); + assert!( + dials["bob"].cancel.load(Ordering::Relaxed), + "the losing link is told to stop" + ); + assert!(!dials["alice"].cancel.load(Ordering::Relaxed)); + } + + #[test] + fn a_level_zero_candidate_is_its_own_root() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_level("alice", alice, 0); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("alice", 1)) + ); + } + + #[test] + fn a_hostile_branch_level_neither_panics_nor_adopts() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_root("alice", alice, "rooty"); + leaf.branch_level("alice", alice, -1); + assert_eq!(leaf.search_from("alice", alice), None, "negative level"); + leaf.branch_level("alice", alice, i32::MAX); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("rooty", 1 << 31)) + ); + } + + #[test] + fn the_parents_later_moves_move_us_too_but_are_told_once_a_second() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); + assert_eq!(leaf.branch_level("alice", alice, 0), None, "unchanged"); + assert_eq!( + leaf.branch_root("alice", alice, "grand"), + Some(Branch::under("grand", 1)) + ); + assert_eq!( + leaf.branch_level("alice", alice, 2), + None, + "a second move within the second waits" + ); + assert_eq!(leaf.branch(), Branch::under("grand", 3), "but it took"); + assert_eq!(leaf.due_announcement(Instant::now()), None); + assert_eq!( + leaf.due_announcement(Instant::now() + MOVE_INTERVAL), + Some(Branch::under("grand", 3)) + ); + assert_eq!( + leaf.due_announcement(Instant::now() + MOVE_INTERVAL), + None, + "told once" + ); + } + + #[test] + fn a_parent_that_becomes_a_root_is_the_root_from_then_on() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_root("alice", alice, "grand"); + leaf.branch_level("alice", alice, 5); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("grand", 6)) + ); + assert_eq!( + leaf.branch_level("alice", alice, 0), + Some(Branch::under("alice", 1)) + ); + } + + #[test] + fn a_stale_link_cannot_speak_for_a_newer_one() { + let mut leaf = leaf(); + let old = dial(&mut leaf, &["alice"])["alice"].link; + assert!(!leaf.closed("alice", old + 7), "unknown link is ignored"); + leaf.closed("alice", old); + let new = dial(&mut leaf, &["alice"])["alice"].link; + assert_ne!(old, new); + leaf.branch_level("alice", old, 0); + assert_eq!(leaf.search_from("alice", old), None, "old link ignored"); + assert!( + !leaf.closed("alice", old), + "a late close leaves the new one" + ); + leaf.branch_level("alice", new, 0); + assert!(leaf.search_from("alice", new).is_some()); + } + + #[test] + fn a_later_batch_keeps_the_candidates_in_flight_and_stays_under_the_cap() { + let mut leaf = leaf(); + let first = dial(&mut leaf, &["alice"])["alice"].link; + let again = dial(&mut leaf, &["alice", "bob"]); + assert!( + !again.contains_key("alice"), + "alice is already being dialled" + ); + assert!(again.contains_key("bob")); + assert!(leaf.has_link("alice", first)); + let names: Vec = (0..20).map(|i| format!("p{i}")).collect(); + let more = leaf + .consider(names.iter().map(|n| candidate(n)).collect::>()); + assert_eq!(more.len(), MAX_CANDIDATES - 2); + } + + #[test] + fn losing_the_parent_or_a_reset_returns_to_parentless() { + let mut leaf = leaf(); + let dials = dial(&mut leaf, &["alice", "bob"]); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); + assert!(!leaf.closed("bob", dials["bob"].link), "not the parent"); + assert!(leaf.closed("alice", alice)); + assert_eq!(leaf.branch(), Branch::parentless("me")); + + let dials = dial(&mut leaf, &["alice", "carol"]); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); + leaf.reset(); + assert_eq!(leaf.branch(), Branch::parentless("me")); + assert!(dials["alice"].cancel.load(Ordering::Relaxed)); + assert!(dials["carol"].cancel.load(Ordering::Relaxed)); + assert!(!leaf.consider(vec![candidate("bob")]).is_empty()); + } + + #[test] + fn searches_beyond_the_budget_are_dropped_until_the_next_second() { + let mut leaf = leaf(); + let now = Instant::now(); + for _ in 0..SEARCH_BUDGET { + assert!(leaf.admit_search(now)); + } + assert!(!leaf.admit_search(now)); + assert!(leaf.admit_search(now + Duration::from_secs(1))); + } + + #[test] + fn announcing_tells_the_server_the_root_the_level_and_that_we_have_a_parent() + { + let (tx, rx) = std::sync::mpsc::channel(); + announce(Some(&tx), &Branch::under("rooty", 4), true); + let mut sent: Vec = Vec::new(); + while let Ok(ServerMessage::SendMessage(m)) = rx.try_recv() { + sent.push(m); + } + let codes: Vec = sent + .iter() + .map(|m| u32::from_le_bytes(m.get_data()[..4].try_into().unwrap())) + .collect(); + assert_eq!(codes, vec![71, 127, 126, 100]); + assert_eq!(sent[0].get_data()[4], 0, "HaveNoParent is false"); + let mut root = sent[1].clone(); + root.set_pointer(4); + assert_eq!(root.read_string(), "rooty"); + assert_eq!(&sent[2].get_data()[4..8], &4u32.to_le_bytes()); + assert_eq!(sent[3].get_data()[4], 0, "AcceptChildren is false"); + } +} diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index d4098d9..5dc3a34 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -60,6 +60,8 @@ pub const DEFAULT_UPLOAD_SLOTS: usize = 10; const BROKER_CONNECT_TIMEOUT: Duration = Duration::from_secs(20); const DEFAULT_MAX_PEERS: usize = 512; +/// How long a message waits for a control connection to its peer. +const PENDING_PEER_TTL: Duration = Duration::from_mins(1); const BROWSE_PROTECT_WINDOW: Duration = Duration::from_mins(5); @@ -107,6 +109,8 @@ struct UploadJob { real_path: std::path::PathBuf, virtual_path: String, size: u64, + /// When the offer went out; `None` once the peer has answered it. + offered: Option, } /// Live bookkeeping for an upload being served (or recently finished). @@ -332,11 +336,45 @@ pub enum ClientOperation { requester_key: String, filename: String, }, + /// Distributed-network parent candidates, from the server or a host. + PossibleParents(Vec<(String, String, u16)>), + /// The server told us to drop our parent. + ResetDistributed, + /// A parent candidate told us how deep it sits. `link` says which dial + /// to that user is talking. + ParentBranchLevel { + parent: String, + link: u64, + level: i32, + }, + /// A parent candidate told us whose branch it is on. + ParentBranchRoot { + parent: String, + link: u64, + root: String, + }, + /// A search came down the tree from `parent`. + ParentSearch { + parent: String, + link: u64, + username: String, + token: u32, + query: String, + }, + /// The link to a parent or candidate is gone. + ParentClosed { + parent: String, + link: u64, + }, } pub struct ClientContext { pub peer_registry: Option, pub downloads: DownloadStore, server_sender: Option>, + /// Where anything outside the operations loop posts operations. + operations: Option>, + /// Our place in the distributed search network. + leaf: distributed::Leaf, searches: HashMap, private_messages: Vec, /// Correlation tokens for server-brokered (firewalled) connections, mapping @@ -349,8 +387,11 @@ pub struct ClientContext { pub shared_directories: Vec, /// Peer listen addresses learned from GetPeerAddress responses. peer_addresses: HashMap, - /// Peer messages waiting for a control connection to that peer. - pending_peer_messages: HashMap>, + /// Peer messages waiting for a control connection to that peer, and + /// when the last was queued: a peer that never answers must not keep + /// them forever. + pending_peer_messages: + HashMap)>, /// Uploads we have offered, keyed by our transfer token. uploads: HashMap, active_uploads: HashMap, @@ -443,6 +484,15 @@ impl ClientContext { } impl ClientContext { + /// A context for the client that logs in as `username`. + #[must_use] + pub fn for_user(username: &str) -> Self { + Self { + leaf: distributed::Leaf::new(username), + ..Self::new() + } + } + #[must_use] pub fn new() -> Self { let actor_system = Arc::new(ActorSystem::new()); @@ -450,6 +500,8 @@ impl ClientContext { Self { peer_registry: None, server_sender: None, + operations: None, + leaf: distributed::Leaf::new(""), searches: HashMap::new(), private_messages: Vec::new(), pending_connect_tokens: HashMap::new(), @@ -690,10 +742,12 @@ impl ClientContext { username: &str, message: crate::message::Message, ) { - self.pending_peer_messages + let entry = self + .pending_peer_messages .entry(username.to_string()) - .or_default() - .push(message); + .or_insert_with(|| (Instant::now(), Vec::new())); + entry.0 = Instant::now(); + entry.1.push(message); } /// Remove and return the messages queued for `username`. @@ -703,9 +757,18 @@ impl ClientContext { ) -> Vec { self.pending_peer_messages .remove(username) + .map(|(_, messages)| messages) .unwrap_or_default() } + /// Drop peer messages nobody could deliver within + /// [`PENDING_PEER_TTL`]: the searcher went offline, or never existed. + pub(crate) fn expire_pending_peer_messages(&mut self, now: Instant) { + self.pending_peer_messages.retain(|_, (since, _)| { + now.duration_since(*since) < PENDING_PEER_TTL + }); + } + /// Store a shared-file listing received from browsing `username`. pub fn store_browse_result( &mut self, @@ -832,11 +895,13 @@ impl Client { listen_port: settings.listen_port, bound_port: None, address: settings.server_address, + context: Arc::new(RwLock::new(ClientContext::for_user( + &settings.username, + ))), username: settings.username, password: settings.password, version: settings.version, shared_directories: settings.shared_directories, - context: Arc::new(RwLock::new(ClientContext::new())), server_handle: None, session: SessionWatch::default(), } @@ -980,6 +1045,7 @@ impl Client { } mod connection; +mod distributed; mod downloads; mod operations; mod rooms; diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 1bc662c..bedf5a3 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -23,6 +23,19 @@ impl Client { if last_sweep.elapsed() >= CONNECT_SWEEP_INTERVAL { last_sweep = Instant::now(); Self::sweep_expired_connects(&client_context); + Self::sweep_stale_offers(&client_context); + if let Ok(mut ctx) = client_context.write_safe() { + ctx.expire_pending_peer_messages(Instant::now()); + if let Some(branch) = + ctx.leaf.due_announcement(Instant::now()) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); + } + } } let operation = match next { Ok(operation) => operation, @@ -695,6 +708,114 @@ impl Client { ); } } + ClientOperation::PossibleParents(candidates) => { + let (dials, ops) = match client_context.write_safe() { + Ok(mut ctx) => ( + ctx.leaf.consider(candidates), + ctx.operations.clone(), + ), + Err(_) => continue, + }; + let Some(ops) = ops else { continue }; + for dial in dials { + super::distributed::spawn_link( + dial, + own_username.clone(), + ops.clone(), + ); + } + } + ClientOperation::ParentBranchLevel { + parent, + link, + level, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = + ctx.leaf.branch_level(&parent, link, level) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); + } + } + ClientOperation::ParentBranchRoot { + parent, + link, + root, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = + ctx.leaf.branch_root(&parent, link, &root) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); + } + } + ClientOperation::ParentSearch { + parent, + link, + username, + token, + query, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = + ctx.leaf.search_from(&parent, link) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); + } + if ctx.leaf.is_parent(&parent, link) + && ctx.leaf.admit_search(Instant::now()) + && let Some(ops) = &ctx.operations + { + let _ = ops.send(ClientOperation::IncomingSearch { + username, + token, + query, + }); + } + } + ClientOperation::ParentClosed { parent, link } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if ctx.leaf.closed(&parent, link) { + super::distributed::announce( + ctx.server_sender.as_ref(), + &ctx.leaf.branch(), + false, + ); + } + } + // Also what a fresh login means: whatever tree we hung + // from belongs to the old session. + ClientOperation::ResetDistributed => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + ctx.leaf.reset(); + super::distributed::announce( + ctx.server_sender.as_ref(), + &ctx.leaf.branch(), + false, + ); + } ClientOperation::PrivilegedUsers(users) => { if let Ok(mut ctx) = client_context.write_safe() { debug!( @@ -718,9 +839,10 @@ impl Client { // address (from the code-9 GetPeerAddress) and // stream the file, or queue until it resolves. let (job_addr, downloader) = match client_context - .read_safe() + .write_safe() { - Ok(ctx) => { + Ok(mut ctx) => { + ctx.mark_offer_answered(token); let Some(job) = ctx.uploads.get(&token) else { continue; }; @@ -761,7 +883,7 @@ impl Client { &requester_key, |ctx| { crate::message::peer::build_shared_file_list( - &shared_directories(ctx), + &ctx.shares.directories(), ) }, ); @@ -790,7 +912,9 @@ impl Client { &client_context, &requester_key, |ctx| { - let dirs: Vec<_> = shared_directories(ctx) + let dirs: Vec<_> = ctx + .shares + .directories() .into_iter() .filter(|dir| dir.name == folder) .collect(); @@ -875,6 +999,15 @@ impl Client { } } + fn sweep_stale_offers(client_context: &Arc>) { + let freed = client_context + .write_safe() + .is_ok_and(|mut ctx| ctx.expire_stale_offers(Instant::now())); + if freed { + Self::pump_upload_queue(client_context); + } + } + fn sweep_expired_connects(client_context: &Arc>) { let expired = match client_context.write_safe() { Ok(mut ctx) => ctx.take_expired_connects(Instant::now()), @@ -885,17 +1018,3 @@ impl Client { } } } - -/// Our shares in the wire form a browse or folder reply carries. -fn shared_directories( - ctx: &ClientContext, -) -> Vec { - ctx.shares - .directories() - .into_iter() - .map(|(name, files)| crate::message::peer::SharedDirectory { - name, - files, - }) - .collect() -} diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 031edb2..1c16b42 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -821,3 +821,18 @@ fn watched_users_are_listed_in_a_stable_order() { context.add_watched_user("bob"); assert_eq!(context.watched_users(), vec!["alice", "bob", "carol"]); } + +// A reply queued for a searcher the server cannot place waits a minute, +// not forever: every such search would otherwise pin a map entry and a +// protected slot in the peer registry. +#[test] +fn a_peer_message_nobody_could_deliver_expires() { + let mut ctx = ClientContext::for_user("me"); + ctx.queue_peer_message("ghost", crate::message::Message::new()); + ctx.expire_pending_peer_messages(Instant::now()); + assert_eq!(ctx.take_peer_messages("ghost").len(), 1, "still fresh"); + + ctx.queue_peer_message("ghost", crate::message::Message::new()); + ctx.expire_pending_peer_messages(Instant::now() + PENDING_PEER_TTL); + assert!(ctx.take_peer_messages("ghost").is_empty(), "expired"); +} diff --git a/soulseek-rs-lib/src/client/upload_queue.rs b/soulseek-rs-lib/src/client/upload_queue.rs index 474c860..4c59f85 100644 --- a/soulseek-rs-lib/src/client/upload_queue.rs +++ b/soulseek-rs-lib/src/client/upload_queue.rs @@ -10,7 +10,7 @@ //! policy (per-user fairness, small-file-first) would be a different promise //! than the one the rules ask for. -use super::{ClientContext, PeerRegistry, UploadJob}; +use super::{ClientContext, Duration, Instant, PeerRegistry, UploadJob}; use crate::types::{UploadInfo, UploadStatus}; use std::path::PathBuf; @@ -18,6 +18,11 @@ use std::path::PathBuf; /// them cannot grow the list without bound. const MAX_UPLOAD_EVENTS: usize = 256; +/// How long an offered slot waits for the peer's TransferResponse. A peer that +/// never answers, having gone away or hung, would otherwise hold the slot for +/// the life of the process; with two slots, two of them shut uploads down. +const OFFER_TIMEOUT: Duration = Duration::from_secs(30); + /// A queued upload that has just been given a slot, ready to be offered to the /// peer that asked for it. #[derive(Debug, Clone)] @@ -142,6 +147,7 @@ impl ClientContext { real_path: job.real_path, virtual_path: job.virtual_path, size: job.size, + offered: Some(Instant::now()), }, ); } @@ -239,6 +245,46 @@ impl ClientContext { self.privileged_users.contains(username) } + /// Drop offers older than [`OFFER_TIMEOUT`], recording each as a failed + /// upload so an operator sees why the peer never got its file. Returns + /// whether anything was dropped, which is the caller's cue to pump again. + pub fn expire_stale_offers(&mut self, now: Instant) -> bool { + let stale: Vec = self + .uploads + .iter() + .filter(|(_, job)| { + job.offered.is_some_and(|sent| { + now.duration_since(sent) >= OFFER_TIMEOUT + }) + }) + .map(|(&token, _)| token) + .collect(); + for token in &stale { + let Some(job) = self.uploads.remove(token) else { + continue; + }; + self.upload_events.push(UploadInfo { + username: job.downloader, + filename: job.virtual_path, + size: job.size, + bytes_sent: 0, + speed_bytes_per_sec: 0.0, + status: UploadStatus::Failed( + "no answer to the transfer offer".to_string(), + ), + }); + } + !stale.is_empty() + } + + /// The peer accepted the offer for `token`; it may still take a while to + /// resolve their address and connect, and that wait must not expire it. + pub fn mark_offer_answered(&mut self, token: u32) { + if let Some(job) = self.uploads.get_mut(&token) { + job.offered = None; + } + } + /// Forget everything `downloader` was waiting for or had been offered, and /// report whether that freed anything. /// @@ -347,6 +393,7 @@ mod tests { /// go untested. #[cfg(test)] mod context_tests { + use super::OFFER_TIMEOUT; use crate::client::{ClientContext, DEFAULT_UPLOAD_SLOTS}; use crate::types::UploadStatus; use std::path::PathBuf; @@ -464,6 +511,38 @@ mod context_tests { assert_eq!(ctx.place_in_queue("second", "f.mp3"), None); } + #[test] + fn an_offer_nobody_answers_frees_its_slot_after_the_timeout() { + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "gone", "f.mp3"); + ask(&mut ctx, "next", "f.mp3"); + assert_eq!(offered(&mut ctx, &mut token), ["gone"]); + + let now = std::time::Instant::now(); + assert!(!ctx.expire_stale_offers(now), "a fresh offer stays"); + assert!( + ctx.expire_stale_offers(now + OFFER_TIMEOUT), + "an unanswered offer expires" + ); + assert_eq!(offered(&mut ctx, &mut token), ["next"]); + assert!(ctx.take_upload_events().iter().any(|event| { + event.username == "gone" + && matches!(event.status, UploadStatus::Failed(_)) + })); + } + + #[test] + fn an_accepted_offer_waiting_on_an_address_does_not_expire() { + let (mut ctx, mut token) = context(1); + ask(&mut ctx, "slow", "f.mp3"); + offered(&mut ctx, &mut token); + + ctx.mark_offer_answered(1); + let later = std::time::Instant::now() + OFFER_TIMEOUT * 2; + assert!(!ctx.expire_stale_offers(later), "an answered offer stays"); + assert!(ctx.uploads.contains_key(&1)); + } + #[test] fn completed_transfers_do_not_keep_occupying_slots() { // active_uploads keeps finished transfers so `uploads()` can report diff --git a/soulseek-rs-lib/src/lib.rs b/soulseek-rs-lib/src/lib.rs index 518c9d7..f9474a5 100644 --- a/soulseek-rs-lib/src/lib.rs +++ b/soulseek-rs-lib/src/lib.rs @@ -2,14 +2,17 @@ // example in it, so a signature change cannot leave the advertised usage stale. #![doc = include_str!("../README.md")] -// Core modules -pub mod actor; +// The public surface is the client, its types, the wire codec and the share +// index. The actors, dispatcher, peer plumbing and download store are how the +// client is built, not what a host programs against, so they stay inside the +// crate and a change to them is not a change to the library's API. +pub(crate) mod actor; pub mod client; -pub mod dispatcher; -pub mod download_store; +pub(crate) mod dispatcher; +pub(crate) mod download_store; pub mod error; pub mod message; -pub mod peer; +pub(crate) mod peer; pub mod shares; pub mod types; #[macro_use] @@ -30,7 +33,8 @@ pub mod prelude { pub use actor::server_actor::{PeerAddress, UserMessage}; pub use client::{Client, ClientSettings}; pub use error::{Result, SoulseekRs}; -pub use message::peer::SharedDirectory; +pub use message::peer::{SharedDirectory, SharedFileEntry}; +pub use peer::{ConnectionType, ParseConnectionTypeError}; pub use types::{ ClientVersion, DownloadStatus, File, RoomEvent, RoomInfo, Search, SearchResult, SessionLoss, Transfer, UploadInfo, UploadStatus, UserInfo, diff --git a/soulseek-rs-lib/src/message/distributed.rs b/soulseek-rs-lib/src/message/distributed.rs new file mode 100644 index 0000000..2ae015c --- /dev/null +++ b/soulseek-rs-lib/src/message/distributed.rs @@ -0,0 +1,244 @@ +//! The distributed search network's wire format. +//! +//! Frames carry a one-byte code, unlike the four-byte server and peer codes, +//! and a leaf only ever reads them: 3 Search, 4 BranchLevel, 5 BranchRoot, +//! 93 EmbeddedMessage. + +use crate::message::Message; +use crate::message::server::MessageFactory; + +/// A search relayed through the tree carries this in its first field; anything +/// else is a frame we do not understand and must not answer. +const SEARCH_IDENTIFIER: u32 = 49; + +/// What a parent sends a leaf. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Distributed { + Search { + username: String, + token: u32, + query: String, + }, + BranchLevel(i32), + BranchRoot(String), +} + +/// Decode one framed distributed message (length prefix included), or `None` +/// for a ping, an unknown code or a truncated frame. +#[must_use] +pub fn parse(message: &mut Message) -> Option { + if message.get_size() < 5 { + return None; + } + message.set_pointer(5); + parse_at(message, message.get_init_code()) +} + +/// Decode the body that follows a one-byte `code`, the pointer already past +/// it. A code-93 envelope is unwrapped exactly once, as Nicotine+ does: an +/// envelope inside an envelope is nobody's search. +#[must_use] +pub fn parse_at(message: &mut Message, code: u8) -> Option { + let code = if code == 93 { + message.read_int8() + } else { + code + }; + match code { + 3 => { + if remaining(message) < 4 + || message.read_int32() != SEARCH_IDENTIFIER + { + return None; + } + let username = message.read_string(); + if remaining(message) < 4 { + return None; + } + let token = message.read_int32(); + let query = message.read_string(); + (!username.is_empty() && !query.is_empty()).then_some( + Distributed::Search { + username, + token, + query, + }, + ) + } + 4 => (remaining(message) >= 4) + .then(|| Distributed::BranchLevel(message.read_int32() as i32)), + 5 => { + let root = message.read_string(); + (!root.is_empty()).then_some(Distributed::BranchRoot(root)) + } + _ => None, + } +} + +const fn remaining(message: &mut Message) -> usize { + message.get_size().saturating_sub(message.get_pointer()) +} + +/// A distributed frame with the one-byte `code`: what a parent (or a test +/// standing in for one) sends down the tree. +fn framed(code: u8, body: impl FnOnce(&mut Message)) -> Message { + let mut message = Message::new(); + message.write_int8(code); + body(&mut message); + message +} + +#[must_use] +pub fn build_search(username: &str, token: u32, query: &str) -> Message { + framed(3, |m| { + m.write_int32(SEARCH_IDENTIFIER) + .write_string(username) + .write_int32(token) + .write_string(query); + }) +} + +#[must_use] +pub fn build_branch_level(level: i32) -> Message { + framed(4, |m| { + m.write_int32(level as u32); + }) +} + +/// What we tell the server about our place in the tree: the four messages +/// Nicotine+ sends together whenever it changes. +#[must_use] +pub fn stance(root: &str, level: u32, has_parent: bool) -> Vec { + vec![ + MessageFactory::build_have_no_parent(!has_parent), + MessageFactory::build_branch_root(root), + MessageFactory::build_branch_level(level), + MessageFactory::build_accept_children(false), + ] +} + +#[must_use] +pub fn build_branch_root(root: &str) -> Message { + framed(5, |m| { + m.write_string(root); + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_distributed_search_carries_who_asked_and_what_for() { + let frame = build_search("seeker", 77, "aphex twin"); + let mut message = Message::new_with_data(frame.get_buffer()); + assert_eq!( + parse(&mut message), + Some(Distributed::Search { + username: "seeker".to_string(), + token: 77, + query: "aphex twin".to_string(), + }) + ); + } + + #[test] + fn a_search_with_the_wrong_identifier_is_dropped() { + let mut body = Message::new(); + body.write_int8(3) + .write_int32(7) + .write_string("seeker") + .write_int32(1) + .write_string("q"); + let mut message = Message::new_with_data(body.get_buffer()); + assert_eq!(parse(&mut message), None); + } + + #[test] + fn branch_level_and_root_frames_parse() { + let mut level = + Message::new_with_data(build_branch_level(2).get_buffer()); + assert_eq!(parse(&mut level), Some(Distributed::BranchLevel(2))); + let mut root = + Message::new_with_data(build_branch_root("rooty").get_buffer()); + assert_eq!( + parse(&mut root), + Some(Distributed::BranchRoot("rooty".to_string())) + ); + } + + #[test] + fn an_embedded_search_unwraps_to_the_search_itself() { + let inner = build_search("seeker", 5, "q").get_data(); + let mut body = Message::new(); + body.write_int8(93).write_raw_bytes(inner); + let mut message = Message::new_with_data(body.get_buffer()); + assert!(matches!( + parse(&mut message), + Some(Distributed::Search { token: 5, .. }) + )); + } + + #[test] + fn unknown_or_truncated_frames_parse_to_nothing() { + let mut ping = + Message::new_with_data(Message::new().write_int8(0).get_buffer()); + assert_eq!(parse(&mut ping), None); + let mut short = Message::new_with_data(vec![1, 0, 0, 0, 3]); + assert_eq!(parse(&mut short), None); + let mut level = Message::new_with_data(vec![3, 0, 0, 0, 4, 1, 0]); + assert_eq!(parse(&mut level), None, "a level needs four bytes"); + let mut root = + Message::new_with_data(build_branch_root("").get_buffer()); + assert_eq!(parse(&mut root), None, "an empty root is no root"); + let mut blank = + Message::new_with_data(build_search("seeker", 1, "").get_buffer()); + assert_eq!(parse(&mut blank), None, "an empty query is no search"); + } + + // An envelope inside an envelope is unwrapped once and found empty; a + // frame that is nothing but envelopes must not recurse its way off the + // stack. + #[test] + fn a_nested_envelope_is_not_unwrapped_twice() { + let mut nested = Message::new(); + nested.write_int8(93).write_raw_bytes( + Message::new() + .write_int8(93) + .write_raw_bytes(build_search("seeker", 5, "q").get_data()) + .get_data(), + ); + let mut message = Message::new_with_data(nested.get_buffer()); + assert_eq!(parse(&mut message), None); + + let mut onions = Message::new_with_data( + Message::new() + .write_raw_bytes(vec![93u8; 64 * 1024]) + .get_buffer(), + ); + assert_eq!(parse(&mut onions), None); + } + + // The frame as the protocol documents it, byte for byte: code 3, the + // constant 49, then user, token and query. + #[test] + fn a_search_frame_matches_the_documented_layout() { + let mut bytes = vec![3u8, 49, 0, 0, 0]; + bytes.extend(2u32.to_le_bytes()); + bytes.extend(b"me"); + bytes.extend(9u32.to_le_bytes()); + bytes.extend(1u32.to_le_bytes()); + bytes.extend(b"q"); + let mut framed = (bytes.len() as u32).to_le_bytes().to_vec(); + framed.extend(&bytes); + assert_eq!(build_search("me", 9, "q").get_buffer(), framed); + assert_eq!( + parse(&mut Message::new_with_data(framed)), + Some(Distributed::Search { + username: "me".to_string(), + token: 9, + query: "q".to_string() + }) + ); + } +} diff --git a/soulseek-rs-lib/src/message/mod.rs b/soulseek-rs-lib/src/message/mod.rs index 28bdf53..03d70d9 100644 --- a/soulseek-rs-lib/src/message/mod.rs +++ b/soulseek-rs-lib/src/message/mod.rs @@ -1,5 +1,6 @@ pub use crate::{debug, error, info, trace, warn}; +pub mod distributed; pub mod handlers; mod message_reader; pub mod peer; diff --git a/soulseek-rs-lib/src/message/peer/folder_contents.rs b/soulseek-rs-lib/src/message/peer/folder_contents.rs index 93f7077..cf0f6f8 100644 --- a/soulseek-rs-lib/src/message/peer/folder_contents.rs +++ b/soulseek-rs-lib/src/message/peer/folder_contents.rs @@ -3,6 +3,8 @@ //! folder" in other clients asks for. use crate::message::peer::SharedDirectory; +#[cfg(test)] +use crate::message::peer::SharedFileEntry; use crate::message::peer::shared_file_list::{ decompress_body, read_directories, write_directories, }; @@ -56,7 +58,18 @@ pub fn parse_folder_contents( 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)], + files: vec![ + SharedFileEntry { + name: "one.flac".to_string(), + size: 40, + attributes: vec![(0, 992)], + }, + SharedFileEntry { + name: "two.flac".to_string(), + size: 50, + attributes: Vec::new(), + }, + ], }]; let built = build_folder_contents(7, "music\\album", &dirs); let mut message = Message::new_with_data(built.get_data()); diff --git a/soulseek-rs-lib/src/message/peer/mod.rs b/soulseek-rs-lib/src/message/peer/mod.rs index 4c37b1a..b3bba95 100644 --- a/soulseek-rs-lib/src/message/peer/mod.rs +++ b/soulseek-rs-lib/src/message/peer/mod.rs @@ -25,8 +25,8 @@ pub use place_in_queue_request::PlaceInQueueRequest; pub use place_in_queue_response::PlaceInQueueResponse; pub use queue_upload::QueueUploadHandler; pub use shared_file_list::{ - SharedDirectory, SharedFileListResponseHandler, build_shared_file_list, - parse_shared_file_list, + SharedDirectory, SharedFileEntry, SharedFileListResponseHandler, + build_shared_file_list, parse_shared_file_list, }; pub use transfer_request::TransferRequest; pub use transfer_response::TransferResponse; diff --git a/soulseek-rs-lib/src/message/peer/shared_file_list.rs b/soulseek-rs-lib/src/message/peer/shared_file_list.rs index 6632797..cc60817 100644 --- a/soulseek-rs-lib/src/message/peer/shared_file_list.rs +++ b/soulseek-rs-lib/src/message/peer/shared_file_list.rs @@ -7,11 +7,32 @@ use crate::peer::PeerMessage; use crate::utils::zlib::{deflate, inflate}; use std::sync::mpsc::Sender; -/// One shared directory and the files directly in it (basename + size). +/// One file in a shared directory: its basename, size and the `(code, value)` +/// audio attributes the owner advertises for it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SharedFileEntry { + pub name: String, + pub size: u64, + pub attributes: Vec<(u32, u32)>, +} + +impl SharedFileEntry { + /// The value of attribute `code` (0 bitrate, 1 duration, 2 VBR, 4 sample + /// rate, 5 bit depth), if the owner advertised it. + #[must_use] + pub fn attribute(&self, code: u32) -> Option { + self.attributes + .iter() + .find(|&&(c, _)| c == code) + .map(|&(_, v)| v) + } +} + +/// One shared directory and the files directly in it. #[derive(Debug, Clone, PartialEq, Eq)] pub struct SharedDirectory { pub name: String, - pub files: Vec<(String, u64)>, + pub files: Vec, } /// Receives a peer's `SharedFileListResponse` (peer code 5) when browsing them. @@ -58,13 +79,16 @@ pub fn write_directories(payload: &mut Message, dirs: &[SharedDirectory]) { payload .write_string(&dir.name) .write_int32(dir.files.len() as u32); - for (name, size) in &dir.files { + for file in &dir.files { payload .write_int8(1) - .write_string(name) - .write_int64(*size) + .write_string(&file.name) + .write_int64(file.size) .write_string("") // extension - .write_int32(0); // attribute count + .write_int32(file.attributes.len() as u32); + for &(code, value) in &file.attributes { + payload.write_int32(code).write_int32(value); + } } } } @@ -87,20 +111,24 @@ pub fn read_directories(body: &mut Message) -> Vec { break; } body.read_int8(); // code - let filename = body.read_string(); - let file_size = body.read_int64(); + let name = body.read_string(); + let size = body.read_int64(); body.read_string(); // extension let attr_count = body.read_int32(); + let mut attributes = Vec::new(); for _ in 0..attr_count { // Each attribute is two int32s (8 bytes); read_int32 does not // advance past the end, so bound the loop explicitly. if body.get_pointer() + 8 > body.get_size() { break; } - body.read_int32(); - body.read_int32(); + attributes.push((body.read_int32(), body.read_int32())); } - files.push((filename, file_size)); + files.push(SharedFileEntry { + name, + size, + attributes, + }); } dirs.push(SharedDirectory { name, files }); } @@ -127,19 +155,56 @@ fn hostile_dir_count_does_not_hang() { assert!(parse_shared_file_list(&mut message).is_empty()); } +#[test] +fn a_listing_carries_each_files_attributes() { + let dirs = vec![SharedDirectory { + name: "music\\album".to_string(), + files: vec![ + SharedFileEntry { + name: "one.mp3".to_string(), + size: 40, + attributes: vec![(0, 320), (1, 187)], + }, + SharedFileEntry { + name: "two.flac".to_string(), + size: 50, + attributes: Vec::new(), + }, + ], + }]; + let mut message = crate::message::framed(|m| { + m.write_raw_bytes( + build_shared_file_list(&dirs).get_data()[4..].to_vec(), + ); + }); + assert_eq!(parse_shared_file_list(&mut message), dirs); +} + #[test] fn shared_file_list_roundtrips() { let dirs = vec![ SharedDirectory { name: "music\\album".to_string(), files: vec![ - ("song one.flac".to_string(), 123), - ("song two.flac".to_string(), 456), + SharedFileEntry { + name: "song one.flac".to_string(), + size: 123, + attributes: vec![(1, 300), (4, 44_100), (5, 16)], + }, + SharedFileEntry { + name: "song two.flac".to_string(), + size: 456, + attributes: Vec::new(), + }, ], }, SharedDirectory { name: "music".to_string(), - files: vec![("top.mp3".to_string(), 789)], + files: vec![SharedFileEntry { + name: "top.mp3".to_string(), + size: 789, + attributes: Vec::new(), + }], }, ]; let message = build_shared_file_list(&dirs); diff --git a/soulseek-rs-lib/src/message/server/distributed.rs b/soulseek-rs-lib/src/message/server/distributed.rs new file mode 100644 index 0000000..8dacc78 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/distributed.rs @@ -0,0 +1,162 @@ +//! What the server says about the distributed network: parent candidates +//! (code 102), a reset (130), and a search embedded for a branch root (93). + +use crate::actor::server_actor::ServerMessage; +use crate::message::distributed::{Distributed, parse_at}; +use crate::message::{Message, MessageHandler}; +use std::net::Ipv4Addr; +use std::sync::mpsc::Sender; + +pub struct PossibleParentsHandler; +impl MessageHandler for PossibleParentsHandler { + fn get_code(&self) -> u32 { + 102 + } + fn handle(&self, message: &mut Message, sender: Sender) { + let count = message.read_int32(); + let candidates = (0..count.min(10)) + .map_while(|_| entry(message)) + .filter_map(|(username, host, port)| { + Some((username, host, u16::try_from(port).ok()?)) + }) + .collect(); + let _ = sender.send(ServerMessage::PossibleParents(candidates)); + } +} + +/// One whole `(username, host, port)` entry, or `None` where the message +/// runs out: a truncated entry is not a parent to dial. +fn entry(message: &mut Message) -> Option<(String, String, u32)> { + let username = message.read_string(); + if username.is_empty() + || message.get_size().saturating_sub(message.get_pointer()) < 8 + { + return None; + } + let host = Ipv4Addr::from(message.read_int32()).to_string(); + Some((username, host, message.read_int32())) +} + +pub struct ResetDistributedHandler; +impl MessageHandler for ResetDistributedHandler { + fn get_code(&self) -> u32 { + 130 + } + fn handle(&self, _message: &mut Message, sender: Sender) { + let _ = sender.send(ServerMessage::ResetDistributed); + } +} + +/// A search the server hands us directly because we are a branch root; it is +/// answered like any other. +pub struct EmbeddedMessageHandler; +impl MessageHandler for EmbeddedMessageHandler { + fn get_code(&self) -> u32 { + 93 + } + fn handle(&self, message: &mut Message, sender: Sender) { + let code = message.read_int8(); + if let Some(Distributed::Search { + username, + token, + query, + }) = parse_at(message, code) + { + let _ = sender.send(ServerMessage::FileSearchRequest { + username, + token, + query, + }); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc::channel; + + #[test] + fn possible_parents_carry_each_candidates_address() { + let (sender, received) = channel(); + let mut message = crate::message::framed(|m| { + m.write_int32(2) + .write_string("alice") + .write_int32(u32::from_le_bytes([1, 0, 0, 127])) + .write_int32(2234) + .write_string("bob") + .write_int32(u32::from_le_bytes([8, 8, 8, 8])) + .write_int32(2235); + }); + PossibleParentsHandler.handle(&mut message, sender); + let ServerMessage::PossibleParents(candidates) = + received.try_recv().unwrap() + else { + panic!("expected PossibleParents"); + }; + assert_eq!( + candidates, + vec![ + ("alice".to_string(), "127.0.0.1".to_string(), 2234), + ("bob".to_string(), "8.8.8.8".to_string(), 2235), + ] + ); + } + + // The count says two, the bytes hold one and a half: the whole one is + // a candidate, the stub is not. A port that is not a port is skipped too. + #[test] + fn a_truncated_or_absurd_entry_is_not_a_candidate() { + let (sender, received) = channel(); + let mut message = crate::message::framed(|m| { + m.write_int32(3) + .write_string("alice") + .write_int32(u32::from_le_bytes([1, 0, 0, 127])) + .write_int32(70_000) + .write_string("bob") + .write_int32(u32::from_le_bytes([8, 8, 8, 8])) + .write_int32(2235) + .write_string("carol"); + }); + PossibleParentsHandler.handle(&mut message, sender); + let ServerMessage::PossibleParents(candidates) = + received.try_recv().unwrap() + else { + panic!("expected PossibleParents"); + }; + assert_eq!( + candidates, + vec![("bob".to_string(), "8.8.8.8".to_string(), 2235)] + ); + } + + #[test] + fn a_reset_and_an_embedded_search_become_client_events() { + let (sender, received) = channel(); + let mut reset = crate::message::framed(|_| {}); + ResetDistributedHandler.handle(&mut reset, sender.clone()); + assert!(matches!( + received.try_recv().unwrap(), + ServerMessage::ResetDistributed + )); + + let inner = crate::message::distributed::build_search("seeker", 9, "q") + .get_data(); + let mut embedded = crate::message::framed(|m| { + m.write_raw_bytes(inner); + }); + EmbeddedMessageHandler.handle(&mut embedded, sender); + let ServerMessage::FileSearchRequest { + username, + token, + query, + } = received.try_recv().unwrap() + else { + panic!("expected FileSearchRequest"); + }; + assert_eq!( + (username.as_str(), token, query.as_str()), + ("seeker", 9, "q") + ); + } +} diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 87fd62c..620508a 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -127,8 +127,32 @@ impl MessageFactory { .clone() } #[must_use] - pub fn build_no_parent_message() -> Message { - Message::new().write_int32(71).write_bool(true).clone() + /// Whether we lack a parent in the distributed network (code 71); the + /// server offers PossibleParents while we say so. + pub fn build_have_no_parent(no_parent: bool) -> Message { + Message::new().write_int32(71).write_bool(no_parent).clone() + } + + /// The root of the branch we hang from (code 127): ourselves while we + /// have no parent. + #[must_use] + pub fn build_branch_root(username: &str) -> Message { + Message::new() + .write_int32(127) + .write_string(username) + .clone() + } + + /// How deep in the tree we sit (code 126): 0 without a parent. + #[must_use] + pub fn build_branch_level(level: u32) -> Message { + Message::new().write_int32(126).write_int32(level).clone() + } + + /// Whether we take children in the distributed network (code 100). + #[must_use] + pub fn build_accept_children(accept: bool) -> Message { + Message::new().write_int32(100).write_bool(accept).clone() } #[must_use] pub fn build_set_wait_port_message(port: u16) -> Message { diff --git a/soulseek-rs-lib/src/message/server/mod.rs b/soulseek-rs-lib/src/message/server/mod.rs index 3d210ab..35fcc3b 100644 --- a/soulseek-rs-lib/src/message/server/mod.rs +++ b/soulseek-rs-lib/src/message/server/mod.rs @@ -1,5 +1,6 @@ mod admin_message; mod connect_to_peer; +mod distributed; mod excluded_search_phrases; mod file_search; mod get_peer_address; @@ -22,6 +23,9 @@ mod wish_list_interval; pub use admin_message::AdminMessageHandler; pub use connect_to_peer::ConnectToPeerHandler; +pub use distributed::{ + EmbeddedMessageHandler, PossibleParentsHandler, ResetDistributedHandler, +}; pub use excluded_search_phrases::ExcludedSearchPhrasesHandler; pub use file_search::FileSearchHandler; pub use get_peer_address::GetPeerAddressHandler; diff --git a/soulseek-rs-lib/src/peer/mod.rs b/soulseek-rs-lib/src/peer/mod.rs index 586320d..301603c 100644 --- a/soulseek-rs-lib/src/peer/mod.rs +++ b/soulseek-rs-lib/src/peer/mod.rs @@ -3,10 +3,9 @@ pub mod listen; pub mod upload_peer; // Export actor types -pub use crate::actor::peer_actor::{PeerActor, PeerMessage}; -pub use crate::actor::peer_registry::PeerRegistry; +pub use crate::actor::peer_actor::PeerMessage; -pub(crate) use download_peer::part_path_of; +pub use download_peer::part_path_of; pub use download_peer::{DownloadError, DownloadPeer}; use crate::message::Message; diff --git a/soulseek-rs-lib/src/shares/audio.rs b/soulseek-rs-lib/src/shares/audio.rs new file mode 100644 index 0000000..6de9282 --- /dev/null +++ b/soulseek-rs-lib/src/shares/audio.rs @@ -0,0 +1,335 @@ +//! The audio attributes a share advertises for a file, read from the file's +//! own headers: Soulseek codes 0 bitrate (kbps), 1 duration (seconds), 2 VBR, +//! 4 sample rate (Hz) and 5 bit depth. Searchers filter and sort on these, so +//! a file without them ranks last in every other client. + +use std::path::Path; + +use std::io::{Read, Seek, SeekFrom}; + +const BITRATE: u32 = 0; +const DURATION: u32 = 1; +const VBR: u32 = 2; +const SAMPLE_RATE: u32 = 4; +const BIT_DEPTH: u32 = 5; + +/// How much of a file's head one read takes. A frame header with its Xing +/// block needs under a hundred bytes; the rest is margin so a typical ID3v2 +/// tag of a few kilobytes fits in the first read too. +const HEAD: usize = 8 * 1024; + +/// The attributes for `path`, or none when the format is not one we read or +/// the file does not parse. Only the file's head is read. +#[must_use] +pub fn probe(path: &Path) -> Vec<(u32, u32)> { + let ext = path + .extension() + .map(|e| e.to_string_lossy().to_ascii_lowercase()); + match ext.as_deref() { + Some("mp3") => mp3(path), + Some("flac") => flac(path), + _ => None, + } + .unwrap_or_default() +} + +fn flac(path: &Path) -> Option> { + let mut head = [0u8; 42]; + std::fs::File::open(path).ok()?.read_exact(&mut head).ok()?; + if &head[0..4] != b"fLaC" || head[4] & 0x7F != 0 { + return None; + } + let packed = u64::from_be_bytes(head[18..26].try_into().ok()?); + let sample_rate = (packed >> 44) as u32; + let bits = ((packed >> 36) & 0x1F) as u32 + 1; + let samples = packed & 0xF_FFFF_FFFF; + if sample_rate == 0 { + return None; + } + let mut attributes = Vec::new(); + if samples > 0 { + let seconds = samples / u64::from(sample_rate); + attributes.push((DURATION, u32::try_from(seconds).unwrap_or(u32::MAX))); + } + attributes.push((SAMPLE_RATE, sample_rate)); + attributes.push((BIT_DEPTH, bits)); + Some(attributes) +} + +/// Layer III bitrates in kbps by the header's four-bit index, for MPEG-1 and +/// for MPEG-2 and 2.5, which halve the table. +const MPEG1_BITRATES: [u32; 15] = [ + 0, 32, 40, 48, 56, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320, +]; +const MPEG2_BITRATES: [u32; 15] = + [0, 8, 16, 24, 32, 40, 48, 56, 64, 80, 96, 112, 128, 144, 160]; +/// Sample rates by the header's two-bit index, one row per version: MPEG-2.5, +/// MPEG-2, MPEG-1. +const SAMPLE_RATES: [[u32; 3]; 3] = [ + [11_025, 12_000, 8_000], + [22_050, 24_000, 16_000], + [44_100, 48_000, 32_000], +]; + +/// One Layer III frame header, decoded. +struct Frame { + bitrate_kbps: u32, + sample_rate: u32, + samples: u32, + /// Offset of the Xing/Info block relative to the header. + xing_at: usize, +} + +fn frame(h: &[u8]) -> Option { + if h.len() < 4 || h[0] != 0xFF || h[1] & 0xE0 != 0xE0 { + return None; + } + let version = (h[1] >> 3) & 3; // 0 = 2.5, 2 = 2, 3 = 1 + let layer = (h[1] >> 1) & 3; // 1 = Layer III + let bitrate_index = usize::from(h[2] >> 4); + let rate_index = usize::from((h[2] >> 2) & 3); + if version == 1 + || layer != 1 + || bitrate_index == 0 + || bitrate_index == 15 + || rate_index == 3 + { + return None; + } + let mpeg1 = version == 3; + let mono = (h[3] >> 6) & 3 == 3; + // Protection bit clear means a two-byte CRC precedes the side info. + let crc = usize::from(h[1] & 1 == 0) * 2; + let rates = SAMPLE_RATES[if version == 0 { + 0 + } else if mpeg1 { + 2 + } else { + 1 + }]; + Some(Frame { + bitrate_kbps: if mpeg1 { + MPEG1_BITRATES + } else { + MPEG2_BITRATES + }[bitrate_index], + sample_rate: rates[rate_index], + samples: if mpeg1 { 1152 } else { 576 }, + // Side info is 32 bytes for MPEG-1 stereo, 17 for MPEG-1 mono and + // MPEG-2 stereo, 9 for MPEG-2 mono. + xing_at: 4 + + crc + + match (mpeg1, mono) { + (true, false) => 32, + (true, true) | (false, false) => 17, + (false, true) => 9, + }, + }) +} + +fn mp3(path: &Path) -> Option> { + let mut file = std::fs::File::open(path).ok()?; + let total = file.metadata().ok()?.len(); + let mut head = [0u8; HEAD]; + let mut len = file.read(&mut head).ok()?; + // An ID3v2 tag in front: its size is four syncsafe bytes, plus a footer. + // `base` is where `head` sits in the file once a big tag forces a second + // read, so the audio length below counts from the first frame. + let mut base = 0usize; + let mut audio_at = 0usize; + if len >= 10 && &head[0..3] == b"ID3" { + let size = head[6..10] + .iter() + .fold(0usize, |acc, &b| (acc << 7) | usize::from(b & 0x7F)); + audio_at = 10 + size + if head[5] & 0x10 != 0 { 10 } else { 0 }; + if audio_at + 4 > len { + file.seek(SeekFrom::Start(audio_at as u64)).ok()?; + len = file.read(&mut head).ok()?; + base = audio_at; + audio_at = 0; + } + } + let head = &head[..len]; + let start = (audio_at..len.saturating_sub(4)) + .find(|&i| frame(&head[i..]).is_some())?; + let frame = frame(&head[start..])?; + let audio_bytes = total.saturating_sub((base + start) as u64); + let at_frame_rate = |bytes: u64| { + (bytes * 8 / (u64::from(frame.bitrate_kbps) * 1000)) as u32 + }; + + let xing = start + frame.xing_at; + let tag = head.get(xing..xing + 4); + let vbr = u32::from(tag == Some(b"Xing")); + if tag != Some(b"Xing") && tag != Some(b"Info") { + return Some(vec![ + (BITRATE, frame.bitrate_kbps), + (DURATION, at_frame_rate(audio_bytes)), + (VBR, 0), + ]); + } + let flags = + u32::from_be_bytes(head.get(xing + 4..xing + 8)?.try_into().ok()?); + let mut at = xing + 8; + let mut frames = None; + if flags & 1 != 0 { + frames = + Some(u32::from_be_bytes(head.get(at..at + 4)?.try_into().ok()?)); + at += 4; + } + let bytes = if flags & 2 != 0 { + u64::from(u32::from_be_bytes(head.get(at..at + 4)?.try_into().ok()?)) + } else { + audio_bytes + }; + // Without a frame count the block adds nothing the first frame did not say. + let Some(frames) = frames else { + return Some(vec![ + (BITRATE, frame.bitrate_kbps), + (DURATION, at_frame_rate(bytes)), + (VBR, vbr), + ]); + }; + let seconds = f64::from(frames) * f64::from(frame.samples) + / f64::from(frame.sample_rate); + let kbps = if seconds > 0.0 { + (bytes as f64 * 8.0 / seconds / 1000.0) as u32 + } else { + frame.bitrate_kbps + }; + Some(vec![ + (BITRATE, kbps), + (DURATION, seconds as u32), + (VBR, vbr), + ]) +} + +#[cfg(test)] +pub(in crate::shares) mod tests { + use super::*; + + /// One MPEG-1 Layer III frame header: 128 kbps, 44.1 kHz, stereo. + const CBR_128: [u8; 4] = [0xFF, 0xFB, 0x90, 0x00]; + const CBR_128_FRAME_LEN: usize = 417; + + pub(in crate::shares) fn cbr_mp3(frames: usize) -> Vec { + let mut out = Vec::new(); + for _ in 0..frames { + out.extend_from_slice(&CBR_128); + out.resize(out.len() + CBR_128_FRAME_LEN - 4, 0); + } + out + } + + fn write(name: &str, bytes: &[u8]) -> std::path::PathBuf { + let dir = std::env::temp_dir() + .join(format!("soulseek-audio-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join(name); + std::fs::write(&path, bytes).unwrap(); + path + } + + #[test] + fn a_constant_bitrate_mp3_reports_bitrate_duration_and_no_vbr() { + // 100 frames of 1152 samples at 44.1 kHz is 2.61 s. + let path = write("cbr.mp3", &cbr_mp3(100)); + assert_eq!(probe(&path), [(0, 128), (1, 2), (2, 0)]); + } + + /// An ID3v2 tag of `size` bytes (syncsafe in the header) with no footer. + fn id3v2(size: usize) -> Vec { + let mut tag = b"ID3\x04\x00\x00".to_vec(); + tag.push(((size >> 21) & 0x7F) as u8); + tag.push(((size >> 14) & 0x7F) as u8); + tag.push(((size >> 7) & 0x7F) as u8); + tag.push((size & 0x7F) as u8); + tag.resize(10 + size, 0); + tag + } + + #[test] + fn an_id3v2_tag_in_front_of_the_frames_is_skipped() { + // 50 frames is 1.31 s; counting the 8000-byte tag as audio, or the + // tag twice, would put the duration at 0. + let mut bytes = id3v2(8000); + bytes.extend_from_slice(&cbr_mp3(50)); + let path = write("tagged.mp3", &bytes); + assert_eq!(probe(&path), [(0, 128), (1, 1), (2, 0)]); + } + + #[test] + fn a_tag_larger_than_the_first_read_is_skipped_too() { + // 20 KiB of tag pushes the frames past the first read; 100 frames is + // 2.61 s, and counting the tag as audio would say 3. + let mut bytes = id3v2(20_480); + bytes.extend_from_slice(&cbr_mp3(100)); + let path = write("big-tag.mp3", &bytes); + assert_eq!(probe(&path), [(0, 128), (1, 2), (2, 0)]); + } + + #[test] + fn a_crc_protected_frame_still_finds_its_xing_header() { + // Protection bit clear: a two-byte CRC sits before the side info. + let mut bytes = cbr_mp3(1); + bytes[1] = 0xFA; + let xing_at = 4 + 2 + 32; + bytes[xing_at..xing_at + 4].copy_from_slice(b"Xing"); + bytes[xing_at + 4..xing_at + 8].copy_from_slice(&3u32.to_be_bytes()); + bytes[xing_at + 8..xing_at + 12].copy_from_slice(&200u32.to_be_bytes()); + bytes[xing_at + 12..xing_at + 16] + .copy_from_slice(&104_250u32.to_be_bytes()); + let path = write("crc.mp3", &bytes); + assert_eq!(probe(&path), [(0, 159), (1, 5), (2, 1)]); + } + + #[test] + fn a_xing_header_without_a_frame_count_falls_back_to_the_frame_header() { + let mut bytes = cbr_mp3(100); + let xing_at = 4 + 32; + bytes[xing_at..xing_at + 4].copy_from_slice(b"Xing"); + bytes[xing_at + 4..xing_at + 8].copy_from_slice(&2u32.to_be_bytes()); + bytes[xing_at + 8..xing_at + 12] + .copy_from_slice(&41_700u32.to_be_bytes()); + let path = write("no-frames.mp3", &bytes); + assert_eq!(probe(&path), [(0, 128), (1, 2), (2, 1)]); + } + + #[test] + fn a_variable_bitrate_mp3_uses_its_xing_header() { + // A Xing header in the first frame's data: 200 frames, 104_250 bytes + // in the stream, so 5.22 s at an average of 160 kbps. + let mut bytes = cbr_mp3(1); + let xing_at = 4 + 32; // MPEG-1 stereo side info + bytes[xing_at..xing_at + 4].copy_from_slice(b"Xing"); + bytes[xing_at + 4..xing_at + 8].copy_from_slice(&3u32.to_be_bytes()); + bytes[xing_at + 8..xing_at + 12].copy_from_slice(&200u32.to_be_bytes()); + bytes[xing_at + 12..xing_at + 16] + .copy_from_slice(&104_250u32.to_be_bytes()); + let path = write("vbr.mp3", &bytes); + assert_eq!(probe(&path), [(0, 159), (1, 5), (2, 1)]); + } + + #[test] + fn a_flac_reports_duration_sample_rate_and_bit_depth() { + // STREAMINFO: 44.1 kHz, 2 channels, 16 bits, 441_000 samples (10 s). + let mut bytes = b"fLaC".to_vec(); + bytes.extend_from_slice(&[0x80, 0x00, 0x00, 0x22]); + bytes.extend_from_slice(&[0x10, 0x00, 0x10, 0x00]); // block sizes + bytes.extend_from_slice(&[0, 0, 0, 0, 0, 0]); // frame sizes + let packed: u64 = + (44_100u64 << 44) | (1u64 << 41) | (15u64 << 36) | 0x0006_BAA8; + bytes.extend_from_slice(&packed.to_be_bytes()); + bytes.extend_from_slice(&[0u8; 16]); // md5 + let path = write("lossless.flac", &bytes); + assert_eq!(probe(&path), [(1, 10), (4, 44_100), (5, 16)]); + } + + #[test] + fn anything_else_or_a_truncated_file_has_no_attributes() { + assert!(probe(&write("notes.txt", b"hello")).is_empty()); + assert!(probe(&write("short.mp3", b"ID3")).is_empty()); + assert!(probe(&write("short.flac", b"fLaC\x80\x00")).is_empty()); + assert!(probe(Path::new("/nonexistent/x.mp3")).is_empty()); + } +} diff --git a/soulseek-rs-lib/src/shares/mod.rs b/soulseek-rs-lib/src/shares/mod.rs index 341ec89..fb85657 100644 --- a/soulseek-rs-lib/src/shares/mod.rs +++ b/soulseek-rs-lib/src/shares/mod.rs @@ -4,12 +4,19 @@ //! peer-facing *virtual path* (the shared directory's own name followed by the //! backslash-separated relative path, matching the Soulseek wire convention). +mod audio; + +use crate::message::peer::{SharedDirectory, SharedFileEntry}; use std::collections::{HashMap, HashSet}; use std::io; use std::path::{Path, PathBuf}; /// One shared file: its peer-facing virtual path and where it lives on disk. +/// +/// Built by the scan, never by a host, so fields can be added without a +/// breaking release. #[derive(Debug, Clone)] +#[non_exhaustive] pub struct SharedFile { /// Backslash-separated path exposed to peers, e.g. `music\album\song.mp3`. pub virtual_path: String, @@ -19,7 +26,8 @@ pub struct SharedFile { /// The real filesystem path used to serve the bytes. pub real_path: PathBuf, pub size: u64, - /// Optional `(code, value)` audio attributes (empty for now). + /// `(code, value)` audio attributes read from the file's headers: bitrate, + /// duration, VBR, sample rate, bit depth, as far as the format gives them. pub attributes: Vec<(u32, u32)>, } @@ -126,12 +134,13 @@ impl Shares { } /// Files grouped by their virtual directory (everything before the final - /// backslash), as `(directory, [(basename, size)])`. Used to build a - /// SharedFileListResponse. + /// backslash), in the form a browse or folder listing carries. #[must_use] - pub fn directories(&self) -> Vec<(String, Vec<(String, u64)>)> { - let mut by_dir: std::collections::BTreeMap> = - std::collections::BTreeMap::new(); + pub fn directories(&self) -> Vec { + let mut by_dir: std::collections::BTreeMap< + String, + Vec, + > = std::collections::BTreeMap::new(); for file in &self.files { let (dir, base) = file .virtual_path @@ -140,9 +149,16 @@ impl Shares { by_dir .entry(dir.to_string()) .or_default() - .push((base.to_string(), file.size)); + .push(SharedFileEntry { + name: base.to_string(), + size: file.size, + attributes: file.attributes.clone(), + }); } - by_dir.into_iter().collect() + by_dir + .into_iter() + .map(|(name, files)| SharedDirectory { name, files }) + .collect() } #[must_use] @@ -202,9 +218,9 @@ fn scan_root( files.push(SharedFile { search_key: virtual_path.to_lowercase(), virtual_path, + attributes: audio::probe(&path), real_path: path, size: meta.len(), - attributes: Vec::new(), }); folders_with_files.insert(dir.clone()); } @@ -244,6 +260,39 @@ mod tests { root } + #[test] + fn a_scanned_mp3_carries_its_bitrate() { + let root = std::env::temp_dir() + .join(format!("soulseek-shares-audio-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).unwrap(); + // 100 frames at 128 kbps is 2.6 seconds. + std::fs::write( + root.join("track.mp3"), + super::audio::tests::cbr_mp3(100), + ) + .unwrap(); + std::fs::write(root.join("cover.jpg"), b"jpeg").unwrap(); + + let shares = Shares::scan(&root).unwrap(); + let track = shares + .get( + "soulseek-shares-audio-{pid}\\track.mp3" + .replace("{pid}", &std::process::id().to_string()) + .as_str(), + ) + .unwrap(); + assert!(track.attributes.contains(&(0, 128))); + assert!(track.attributes.contains(&(1, 2))); + let cover = shares + .files() + .iter() + .find(|f| f.virtual_path.ends_with("cover.jpg")) + .unwrap(); + assert!(cover.attributes.is_empty()); + let _ = std::fs::remove_dir_all(root); + } + #[test] fn scan_counts_files_and_folders() { let root = temp_tree(); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 15f4495..f4e7725 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -23,11 +23,11 @@ use std::sync::mpsc::{Receiver, Sender}; use std::time::{Duration, Instant}; use soulseek_rs::message::Message; +use soulseek_rs::message::distributed; use soulseek_rs::message::server::MessageFactory; -use soulseek_rs::peer::ConnectionType; use soulseek_rs::{ - Client, ClientSettings, ClientVersion, DownloadStatus, PeerAddress, - SessionLoss, UploadStatus, + Client, ClientSettings, ClientVersion, ConnectionType, DownloadStatus, + PeerAddress, SessionLoss, UploadStatus, }; /// Only one test at a time may drive a server. @@ -1838,7 +1838,7 @@ fn a_runtime_share_update_is_visible_to_browsers() { assert!( directories .iter() - .any(|d| d.files.iter().any(|(name, _)| name == "late.mp3")), + .any(|d| d.files.iter().any(|f| f.name == "late.mp3")), "the listing should include the file shared at runtime" ); @@ -1888,7 +1888,7 @@ fn browse_a_peers_shared_files() { assert!( directories .iter() - .any(|d| { d.files.iter().any(|(name, _)| name == "track.flac") }), + .any(|d| { d.files.iter().any(|f| f.name == "track.flac") }), "the listing should include the shared file" ); @@ -1949,7 +1949,7 @@ fn browse_a_firewalled_peer_via_broker() { assert!( directories .iter() - .any(|d| d.files.iter().any(|(name, _)| name == "hidden.flac")), + .any(|d| d.files.iter().any(|f| f.name == "hidden.flac")), "the brokered listing should include the shared file" ); @@ -2019,7 +2019,7 @@ fn a_third_party_client_browses_our_shares_directly() { assert!( directories .iter() - .any(|d| d.files.iter().any(|(name, _)| name == "track.flac")), + .any(|d| d.files.iter().any(|f| f.name == "track.flac")), "the listing should include the shared file, got {directories:?}" ); @@ -2127,7 +2127,7 @@ fn a_third_party_client_browses_our_shares_via_the_server_broker() { assert!( directories .iter() - .any(|d| d.files.iter().any(|(name, _)| name == "brokered.flac")), + .any(|d| d.files.iter().any(|f| f.name == "brokered.flac")), "the brokered listing should include the shared file, got {directories:?}" ); @@ -2180,7 +2180,7 @@ fn a_stalled_peer_connection_does_not_wedge_the_listener() { assert!( directories .iter() - .any(|d| d.files.iter().any(|(name, _)| name == "still.flac")), + .any(|d| d.files.iter().any(|f| f.name == "still.flac")), "the listing should include the shared file, got {directories:?}" ); @@ -2312,7 +2312,12 @@ impl QueueingPeer { /// place in the queue" is also true of a peer that never asked, so polling /// for that would pass against a completely broken pump. fn takes_a_slot(&mut self) -> bool { - read_until_code(&mut self.peer, 40, Duration::from_secs(10)).is_some() + self.takes_a_slot_within(Duration::from_secs(10)) + } + + fn takes_a_slot_within(&mut self, timeout: Duration) -> bool { + let _ = self.peer.set_read_timeout(Some(timeout)); + read_until_code(&mut self.peer, 40, timeout).is_some() } } @@ -3250,7 +3255,7 @@ fn a_third_party_client_fetches_one_folder_of_our_shares() { assert_eq!(echoed, folder); let mut names: Vec<(String, Vec)> = directories .into_iter() - .map(|d| (d.name, d.files.into_iter().map(|(name, _)| name).collect())) + .map(|d| (d.name, d.files.into_iter().map(|f| f.name).collect())) .collect(); for (_, files) in &mut names { files.sort(); @@ -3478,3 +3483,288 @@ fn a_server_announcement_arrives_as_a_message_from_the_server() { .expect("bob should hear the announcement"); assert_eq!(announcement.message(), "going down at nine"); } + +// A peer that is offered a slot and never answers must not keep it: with +// two slots by default, two such peers shut uploads down for the session. +#[test] +fn an_unanswered_upload_offer_frees_its_slot() { + let server = server_or_skip!(); + let (share, folder) = queue_share("expiry", &["silent.mp3", "patient.mp3"]); + let server_addr = format!("{}:{}", server.host, server.port); + + let sharer_port = free_port().expect("sharer port"); + let mut sharer = Client::with_settings(ClientSettings { + shared_directories: vec![share.display().to_string()], + ..server.listening_settings("e2e_expiry_sharer", "pw", sharer_port) + }); + sharer.connect().expect("sharer connect"); + assert!(sharer.login().expect("sharer login")); + sharer.set_upload_slots(1); + let listen_addr = format!("127.0.0.1:{sharer_port}"); + + let mut silent = queue_as( + &server_addr, + &listen_addr, + "e2e_expiry_silent", + &format!("{folder}\\silent.mp3"), + ) + .expect("silent peer queues"); + assert!(silent.takes_a_slot(), "the silent peer is offered the slot"); + + let mut patient = queue_as( + &server_addr, + &listen_addr, + "e2e_expiry_patient", + &format!("{folder}\\patient.mp3"), + ) + .expect("patient peer queues"); + assert!( + patient.takes_a_slot_within(Duration::from_mins(1)), + "the slot must come free once the silent peer's offer expires" + ); + assert!( + sharer.take_upload_events().iter().any(|event| { + event.username == "e2e_expiry_silent" + && matches!(event.status, UploadStatus::Failed(_)) + }), + "the operator sees the unanswered offer fail" + ); + + let _ = std::fs::remove_dir_all(share); +} + +/// A constant-bitrate MP3: `frames` MPEG-1 Layer III frames at 128 kbps, +/// 44.1 kHz, so every searcher sees a bitrate to filter and sort on. +fn cbr_mp3(frames: usize) -> Vec { + let mut out = Vec::new(); + for _ in 0..frames { + out.extend_from_slice(&[0xFF, 0xFB, 0x90, 0x00]); + out.resize(out.len() + 413, 0); + } + out +} + +// A shared file with no bitrate fails every `--min-bitrate` filter and sorts +// last in SoulseekQt; the attributes come from the file's own headers. +#[test] +fn a_shared_mp3_advertises_its_bitrate_and_duration() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + // 1000 frames is 26 seconds. + std::fs::write(share_dir.join("attrprobe song.mp3"), cbr_mp3(1000)) + .unwrap(); + let (_sharer, searcher) = sharer_and_searcher( + &server, + &share_dir, + "e2e_attr_sharer", + "e2e_attr_searcher", + ); + + let query = "attrprobe"; + let _ = searcher.search(query, Duration::from_secs(3)); + let reply = reply_from(&searcher, query, "e2e_attr_sharer") + .expect("the sharer answers"); + let file = &reply.files[0]; + assert_eq!(file.attribs.get(&0), Some(&128), "bitrate in kbps"); + assert_eq!(file.attribs.get(&1), Some(&26), "duration in seconds"); + assert_eq!(file.attribs.get(&2), Some(&0), "constant bitrate"); + + let _ = std::fs::remove_dir_all(share_dir); +} + +// A browse shows what a search shows: the bitrate and duration a searcher +// filters on are in the listing too, as Nicotine+ sends them. +#[test] +fn a_browse_listing_carries_the_files_attributes() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + std::fs::write(share_dir.join("listed.mp3"), cbr_mp3(1000)).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_lattr_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_lattr_browser", "pw") + .expect("third-party client logs in"); + + let directories = third_party_browse( + &format!("127.0.0.1:{sharer_port}"), + "e2e_lattr_browser", + Duration::ZERO, + Duration::from_secs(15), + ) + .expect("a listing"); + let listed = directories + .iter() + .flat_map(|d| d.files.iter()) + .find(|f| f.name == "listed.mp3") + .expect("the mp3 is listed"); + assert!( + listed.attributes.contains(&(0, 128)), + "{:?}", + listed.attributes + ); + assert!( + listed.attributes.contains(&(1, 26)), + "{:?}", + listed.attributes + ); + + let _ = std::fs::remove_dir_all(share_dir); +} + +/// The next inbound connection, or a timeout error: a mock that never gets +/// dialled must fail its test rather than hang it. +fn accept_within( + listener: &std::net::TcpListener, + timeout: Duration, +) -> std::io::Result { + listener.set_nonblocking(true)?; + let deadline = Instant::now() + timeout; + let stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + if Instant::now() >= deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "no inbound connection", + )); + } + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => return Err(e), + } + }; + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(Duration::from_secs(15)))?; + Ok(stream) +} + +/// The PeerInit that opens an inbound connection: who dialled, and as what. +fn peer_init_of(stream: &mut TcpStream) -> std::io::Result<(String, String)> { + let mut init = read_framed(stream)?; + assert_eq!(init.get_init_code(), 1, "expected a PeerInit"); + init.set_pointer(5); + Ok((init.read_string(), init.read_string())) +} + +// Searches also travel peer to peer, down a tree of parents and children the +// server assembles from PossibleParents. A leaf dials the candidates with a D +// connection, adopts the first that passes it a search after stating its +// branch, and answers those searches exactly like ones from the server. +// soulfind never hands out parents, so the parent here is a mock: a logged-in +// user the leaf is pointed at directly. +#[test] +fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { + let server = server_or_skip!(); + + let share_dir = unique_download_dir(); + std::fs::create_dir_all(share_dir.join("album")).unwrap(); + std::fs::write(share_dir.join("album").join("treesearch.mp3"), b"xxxx") + .unwrap(); + + // The parent and the searcher are two logged-in users with a listener + // each. Bind all interfaces: soulfind reports the LAN address. + let server_addr = format!("{}:{}", server.host, server.port); + let raw_user = |name: &str| { + let port = free_port().expect("a port for the raw user"); + let listener = std::net::TcpListener::bind(("0.0.0.0", port)).unwrap(); + let mut srv = login_raw(&server_addr, name, "pw").expect("raw login"); + srv.write_all( + &MessageFactory::build_set_wait_port_message(port).get_buffer(), + ) + .unwrap(); + (listener, port, srv) + }; + let (listener, parent_port, _parent_srv) = raw_user("e2e_parent"); + let (searcher, _, _searcher_srv) = raw_user("e2e_searcher"); + + let mut leaf = Client::with_settings(ClientSettings { + shared_directories: vec![share_dir.display().to_string()], + ..server.listening_settings( + "e2e_leaf", + "pw", + free_port().expect("leaf port"), + ) + }); + leaf.connect().expect("leaf connect"); + assert!(leaf.login().expect("leaf login")); + assert_eq!( + leaf.distributed_branch(), + ("e2e_leaf".to_string(), 0), + "a leaf without a parent is its own branch root" + ); + + leaf.consider_parents(vec![( + "e2e_parent".to_string(), + "127.0.0.1".to_string(), + parent_port, + )]) + .expect("consider parents"); + + let mut link = accept_within(&listener, Duration::from_secs(15)) + .expect("the leaf dials the candidate parent"); + assert_eq!( + peer_init_of(&mut link).unwrap(), + ("e2e_leaf".to_string(), "D".to_string()), + "a parent link opens with a distributed PeerInit" + ); + + // We are a branch root at level 0; the leaf hangs one level below us. + link.write_all(&distributed::build_branch_level(0).get_buffer()) + .unwrap(); + link.write_all(&distributed::build_branch_root("e2e_parent").get_buffer()) + .unwrap(); + link.write_all( + &distributed::build_search("e2e_searcher", 4242, "treesearch") + .get_buffer(), + ) + .unwrap(); + + let deadline = Instant::now() + Duration::from_secs(10); + while leaf.distributed_branch().1 == 0 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + assert_eq!( + leaf.distributed_branch(), + ("e2e_parent".to_string(), 1), + "the first search from a candidate with a branch adopts it" + ); + + // The answer goes to whoever asked, not to the parent that relayed it: + // a P connection to the searcher carrying a FileSearchResponse for the + // token. + let mut p = accept_within(&searcher, Duration::from_secs(20)) + .expect("the leaf dials the searcher to answer"); + assert_eq!( + peer_init_of(&mut p).unwrap(), + ("e2e_leaf".to_string(), "P".to_string()) + ); + let response = expect_code(&mut p, 9, Duration::from_secs(15)) + .expect("a FileSearchResponse for the search from the tree"); + let body = soulseek_rs::utils::zlib::inflate(&response.get_data()[8..]) + .expect("the response body inflates"); + let name_len = u32::from_le_bytes(body[..4].try_into().unwrap()) as usize; + let token_at = 4 + name_len; + assert_eq!(&body[4..token_at], b"e2e_leaf"); + assert_eq!( + u32::from_le_bytes(body[token_at..token_at + 4].try_into().unwrap()), + 4242 + ); + + // Losing the parent puts the leaf back on its own. + drop(link); + let deadline = Instant::now() + Duration::from_secs(10); + while leaf.distributed_branch().1 != 0 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + assert_eq!(leaf.distributed_branch(), ("e2e_leaf".to_string(), 0)); + + let _ = std::fs::remove_dir_all(share_dir); +} diff --git a/soulseek-rs/docs/openrpc.json b/soulseek-rs/docs/openrpc.json index 365ebe3..d9ab8e0 100644 --- a/soulseek-rs/docs/openrpc.json +++ b/soulseek-rs/docs/openrpc.json @@ -948,21 +948,8 @@ "SharedDirectoryDto": { "properties": { "files": { - "description": "Basename and size; join with the directory to get a downloadable path.", "items": { - "maxItems": 2, - "minItems": 2, - "prefixItems": [ - { - "type": "string" - }, - { - "format": "uint64", - "minimum": 0, - "type": "integer" - } - ], - "type": "array" + "$ref": "#/components/schemas/SharedFileEntryDto" }, "type": "array" }, @@ -976,6 +963,46 @@ ], "type": "object" }, + "SharedFileEntryDto": { + "properties": { + "attributes": { + "description": "`(code, value)` audio attributes the owner advertises.", + "items": { + "maxItems": 2, + "minItems": 2, + "prefixItems": [ + { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + { + "format": "uint32", + "minimum": 0, + "type": "integer" + } + ], + "type": "array" + }, + "type": "array" + }, + "name": { + "description": "Basename; join with the directory to get a downloadable path.", + "type": "string" + }, + "size": { + "format": "uint64", + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "name", + "size", + "attributes" + ], + "type": "object" + }, "SharesStatus": { "description": "What the share index holds, from `shares.status` / `shares.reindex`.", "properties": { @@ -1313,7 +1340,7 @@ "name": "MIT" }, "title": "soulseek-rs daemon", - "version": "1" + "version": "2" }, "methods": [ { diff --git a/soulseek-rs/src/SKILL.md b/soulseek-rs/src/SKILL.md index 220d3e0..b416efd 100644 --- a/soulseek-rs/src/SKILL.md +++ b/soulseek-rs/src/SKILL.md @@ -131,7 +131,8 @@ something that is not on the network right now: add it as a wish, then run the wishlist later rather than repeating the same search. **`browse `** — one object per file the peer shares: `user`, `directory`, -`path`, `size`. +`path`, `size`, `bitrate`, `duration`. The last two are `null` when the peer +advertises no audio attributes for the file. **`user `** — one object: `user`, `status` (`online`/`away`/`offline`), `privileged`, `average_speed`, `shared_files`, `shared_folders`. Any field may diff --git a/soulseek-rs/src/cli.rs b/soulseek-rs/src/cli.rs index 7b96954..ad2e2cf 100644 --- a/soulseek-rs/src/cli.rs +++ b/soulseek-rs/src/cli.rs @@ -172,7 +172,11 @@ pub struct Cli { pub log_file: Option, } +/// Non-exhaustive: this enum is public through the crate's library target so +/// the integration tests can reach it, and a new subcommand must not read as a +/// breaking change of that library. #[derive(Subcommand, Debug)] +#[non_exhaustive] pub enum Commands { /// Search the network and print matching files Search(SearchArgs), @@ -254,6 +258,7 @@ pub struct DaemonArgs { } #[derive(Subcommand, Debug)] +#[non_exhaustive] pub enum DaemonCommand { /// Print the token remote clients authenticate with Token, @@ -266,6 +271,7 @@ pub enum DaemonCommand { } #[derive(Subcommand, Debug)] +#[non_exhaustive] pub enum CompletionsCommand { /// Write the completion script where the shell looks for it Install(CompletionsArgs), diff --git a/soulseek-rs/src/commands/social.rs b/soulseek-rs/src/commands/social.rs index 4bedea4..f9b897c 100644 --- a/soulseek-rs/src/commands/social.rs +++ b/soulseek-rs/src/commands/social.rs @@ -61,13 +61,15 @@ pub fn browse(ctx: &Ctx, args: &BrowseArgs) -> CliResult { let mut files = 0usize; for directory in &directories { - for (basename, size) in &directory.files { + for file in &directory.files { files += 1; ctx.out.emit(&BrowseRecord { user: args.user.clone(), directory: directory.name.clone(), - path: full_path(&directory.name, basename), - size: *size, + path: full_path(&directory.name, &file.name), + size: file.size, + bitrate: file.attribute(super::transfer::ATTR_BITRATE), + duration: file.attribute(super::transfer::ATTR_DURATION), }); } } diff --git a/soulseek-rs/src/commands/transfer.rs b/soulseek-rs/src/commands/transfer.rs index bd37b5b..b97b054 100644 --- a/soulseek-rs/src/commands/transfer.rs +++ b/soulseek-rs/src/commands/transfer.rs @@ -19,8 +19,8 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; /// Soulseek file attribute codes the network agrees on. -const ATTR_BITRATE: u32 = 0; -const ATTR_DURATION: u32 = 1; +pub const ATTR_BITRATE: u32 = 0; +pub const ATTR_DURATION: u32 = 1; /// What to keep out of the raw result set. Every field is a further /// restriction: a candidate has to satisfy all of them. @@ -388,10 +388,10 @@ fn resolve_sizes( listing .iter() .flat_map(|directory| { - directory.files.iter().map(|(basename, size)| { + directory.files.iter().map(|file| { ( - social::full_path(&directory.name, basename), - *size, + social::full_path(&directory.name, &file.name), + file.size, ) }) }) diff --git a/soulseek-rs/src/daemon/proto/dto.rs b/soulseek-rs/src/daemon/proto/dto.rs index 304ecfb..82ea25a 100644 --- a/soulseek-rs/src/daemon/proto/dto.rs +++ b/soulseek-rs/src/daemon/proto/dto.rs @@ -12,7 +12,8 @@ use soulseek_rs::types::{ }; use soulseek_rs::{ DownloadStatus, File, RoomEvent, RoomInfo, SearchResult, SessionLoss, - SharedDirectory, UploadInfo, UploadStatus, UserInfo, UserMessage, + SharedDirectory, SharedFileEntry, UploadInfo, UploadStatus, UserInfo, + UserMessage, }; use std::collections::HashMap; @@ -553,15 +554,32 @@ impl From for UserInfo { #[cfg_attr(test, derive(schemars::JsonSchema))] pub struct SharedDirectoryDto { pub name: String, - /// Basename and size; join with the directory to get a downloadable path. - pub files: Vec<(String, u64)>, + pub files: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] +pub struct SharedFileEntryDto { + /// Basename; join with the directory to get a downloadable path. + pub name: String, + pub size: u64, + /// `(code, value)` audio attributes the owner advertises. + pub attributes: Vec<(u32, u32)>, } impl From<&SharedDirectory> for SharedDirectoryDto { fn from(directory: &SharedDirectory) -> Self { Self { name: directory.name.clone(), - files: directory.files.clone(), + files: directory + .files + .iter() + .map(|file| SharedFileEntryDto { + name: file.name.clone(), + size: file.size, + attributes: file.attributes.clone(), + }) + .collect(), } } } @@ -570,7 +588,15 @@ impl From for SharedDirectory { fn from(dto: SharedDirectoryDto) -> Self { Self { name: dto.name, - files: dto.files, + files: dto + .files + .into_iter() + .map(|file| SharedFileEntry { + name: file.name, + size: file.size, + attributes: file.attributes, + }) + .collect(), } } } diff --git a/soulseek-rs/src/daemon/proto/mod.rs b/soulseek-rs/src/daemon/proto/mod.rs index 7a3fea7..e0e243a 100644 --- a/soulseek-rs/src/daemon/proto/mod.rs +++ b/soulseek-rs/src/daemon/proto/mod.rs @@ -13,7 +13,7 @@ /// Bumped when a change would break a client written against the old shape. /// Sent in the `auth` reply so a mismatch is caught at connect time rather /// than as a confusing failure three calls later. -pub const PROTOCOL_VERSION: u32 = 1; +pub const PROTOCOL_VERSION: u32 = 2; /// The OpenRPC document, served by `rpc.discover` so a generator can point at /// a running daemon instead of needing this repository. @@ -322,7 +322,11 @@ mod tests { fn a_shared_listing_survives_the_round_trip() { let directory = SharedDirectory { name: "@@x\\Music".into(), - files: vec![("a.mp3".into(), 4096)], + files: vec![soulseek_rs::SharedFileEntry { + name: "a.mp3".into(), + size: 4096, + attributes: vec![(0, 320)], + }], }; let json = serde_json::to_string(&SharedDirectoryDto::from(&directory)) .unwrap(); diff --git a/soulseek-rs/src/models/browse.rs b/soulseek-rs/src/models/browse.rs index c20bf16..305558f 100644 --- a/soulseek-rs/src/models/browse.rs +++ b/soulseek-rs/src/models/browse.rs @@ -123,13 +123,13 @@ pub fn build_browse_tree(directories: &[SharedDirectory]) -> BuiltTree { // File paths come from the original name so downloads match exactly. let dir_path = dir.name.trim_end_matches('\\'); - for (basename, size) in &dir.files { + for file in &dir.files { let file_path = if dir_path.is_empty() { - basename.clone() + file.name.clone() } else { - format!("{dir_path}\\{basename}") + format!("{dir_path}\\{}", file.name) }; - node.files.push((basename.clone(), *size, file_path)); + node.files.push((file.name.clone(), file.size, file_path)); } } @@ -416,7 +416,14 @@ mod tests { fn dir(name: &str, files: &[(&str, u64)]) -> SharedDirectory { SharedDirectory { name: name.to_string(), - files: files.iter().map(|(n, s)| ((*n).to_string(), *s)).collect(), + files: files + .iter() + .map(|(n, s)| soulseek_rs::SharedFileEntry { + name: (*n).to_string(), + size: *s, + attributes: Vec::new(), + }) + .collect(), } } diff --git a/soulseek-rs/src/output.rs b/soulseek-rs/src/output.rs index 64d2c09..abe9d61 100644 --- a/soulseek-rs/src/output.rs +++ b/soulseek-rs/src/output.rs @@ -207,6 +207,8 @@ pub struct BrowseRecord { pub directory: String, pub path: String, pub size: u64, + pub bitrate: Option, + pub duration: Option, } impl Record for BrowseRecord { @@ -667,6 +669,8 @@ mod tests { directory: "@@music".into(), path: "@@music\\a.flac".into(), size: 12, + bitrate: Some(320), + duration: None, }; assert_eq!(record.text(), "bob\t12\t@@music\\a.flac"); } diff --git a/soulseek-rs/src/remote/mirror.rs b/soulseek-rs/src/remote/mirror.rs index 10472f4..d7439f0 100644 --- a/soulseek-rs/src/remote/mirror.rs +++ b/soulseek-rs/src/remote/mirror.rs @@ -227,14 +227,18 @@ mod tests { Event::Browse, json!({ "username": "bob", - "directories": [{ "name": "@@x", "files": [["a.mp3", 42]] }], + "directories": [{ + "name": "@@x", + "files": [{ "name": "a.mp3", "size": 42, "attributes": [[0, 320]] }], + }], }), ); assert!(mirror.take_browse("sue").is_none()); let listing = mirror.take_browse("bob").expect("bob answered"); assert_eq!(listing[0].name, "@@x"); - assert_eq!(listing[0].files, [("a.mp3".to_string(), 42)]); + assert_eq!(listing[0].files[0].name, "a.mp3"); + assert_eq!(listing[0].files[0].attribute(0), Some(320)); assert!(mirror.take_browse("bob").is_none(), "taken means taken"); } diff --git a/soulseek-rs/src/ui/main_tui/mod.rs b/soulseek-rs/src/ui/main_tui/mod.rs index bfffd32..75ee091 100644 --- a/soulseek-rs/src/ui/main_tui/mod.rs +++ b/soulseek-rs/src/ui/main_tui/mod.rs @@ -1628,7 +1628,13 @@ mod tests { tui.state.browse.open("bob"); let listing = vec![soulseek_rs::SharedDirectory { name: "Music".to_string(), - files: (0..files).map(|i| (format!("{i:02}.mp3"), 1)).collect(), + files: (0..files) + .map(|i| soulseek_rs::SharedFileEntry { + name: format!("{i:02}.mp3"), + size: 1, + attributes: Vec::new(), + }) + .collect(), }]; tui.state .browse diff --git a/soulseek-rs/tests/cli_e2e.rs b/soulseek-rs/tests/cli_e2e.rs index 3086a6c..b6b0cb9 100644 --- a/soulseek-rs/tests/cli_e2e.rs +++ b/soulseek-rs/tests/cli_e2e.rs @@ -1084,8 +1084,11 @@ impl FilterFixture { fn new(server: &TestServer, sharer: &str) -> (Self, Client) { let share = Scratch::new("filter-share"); + // The mp3 is a real 128 kbps stream so the bitrate filter has + // something to read; the others are padding of a known size. + std::fs::write(share.path().join("flt_alpha_keepme.mp3"), cbr_mp3(10)) + .expect("share file"); for (name, size) in [ - ("flt_alpha_keepme.mp3", Self::MEDIUM), ("flt_beta_skipme.flac", Self::SMALL), ("flt_gamma_keepme.ogg", Self::LARGE), ] { @@ -1098,6 +1101,18 @@ impl FilterFixture { } } +/// A constant-bitrate MP3: `frames` MPEG-1 Layer III frames at 128 kbps and +/// 44.1 kHz, 417 bytes each; ten of them land between the fixture's SMALL +/// and LARGE sizes. +fn cbr_mp3(frames: usize) -> Vec { + let mut out = Vec::new(); + for _ in 0..frames { + out.extend_from_slice(&[0xFF, 0xFB, 0x90, 0x00]); + out.resize(out.len() + 413, 0); + } + out +} + /// Search for the fixture as a fresh user and return the matching file names, /// sorted so the assertion does not depend on result order. /// @@ -1155,6 +1170,30 @@ fn extension_keeps_one_type_and_repeats_into_a_union() { assert_eq!(several, ["flt_beta_skipme.flac", "flt_gamma_keepme.ogg"]); } +#[test] +fn a_bitrate_floor_keeps_only_files_that_advertise_at_least_that_much() { + let server = server_or_skip!(); + let (_fixture, _sharer) = FilterFixture::new(&server, "cli_e2e_flt_kbps"); + + // Only the mp3 carries a bitrate; the padding files have none and an + // unknown bitrate cannot satisfy a floor. + let (exit, enough) = filtered_names( + &server, + "cli_e2e_seek_kbps_128", + &["--min-bitrate", "128"], + ); + assert_eq!(exit, EXIT_OK); + assert_eq!(enough, ["flt_alpha_keepme.mp3"]); + + let (exit, none) = filtered_names( + &server, + "cli_e2e_seek_kbps_320", + &["--min-bitrate", "320"], + ); + assert_eq!(exit, EXIT_NO_RESULTS); + assert!(none.is_empty()); +} + #[test] fn size_bounds_narrow_the_result_set_from_either_end() { let server = server_or_skip!();