Merge branch 'develop'

.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

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`                                                                                                  |

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`.

soulseek-rs-lib/Cargo.toml

@@ -21,3 +21,7 @@ workspace = true
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2.189"
+
+[[example]]
+name = "stress"
+test = true

soulseek-rs-lib/README.md

@@ -115,6 +115,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 }
 ```
 
-`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.

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<SharedFileEntry> {
+    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<String> {
+    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%"]);
+    }
 }

soulseek-rs-lib/src/actor/mod.rs

@@ -52,7 +52,7 @@ impl<M: Send> ActorHandle<M> {
 }
 
 /// Internal actor message wrapper
-pub(crate) enum ActorMessage<M> {
+pub enum ActorMessage<M> {
     UserMessage(M),
     Stop,
 }

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<Instant>,
 }
 
 /// 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<ClientOperation>, 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<TcpStream>,
-    ) -> (PeerActor, Receiver<ClientOperation>) {
-        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;

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<ClientOperation>, 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<TcpStream>,
+) -> (PeerActor, Receiver<ClientOperation>) {
+    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<u8> {
+    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<crate::actor::ActorMessage<PeerMessage>> {
+    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)
+    ));
+}

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

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),

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<ClientOperation>) {
+    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<u32> = 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<u32> = no_listen.iter().map(code_of).collect();
-    assert_eq!(codes, vec![35, 71, 28]);
+    assert_eq!(codes, vec![35, 28]);
 }

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();
 

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<u32>,
+    root: Option<String>,
+    cancel: Arc<AtomicBool>,
+}
+
+/// 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<AtomicBool>,
+}
+
+/// 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<String>,
+    branch: Branch,
+    candidates: HashMap<String, Candidate>,
+    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<Instant>,
+    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<Dial> {
+        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<Branch> {
+        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<Branch> {
+        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<Branch> {
+        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<Branch> {
+        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<Branch> {
+        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<ServerMessage>>,
+    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<ClientOperation>) {
+    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<ClientOperation>,
+) -> 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<ClientOperation> {
+    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<String, Dial> {
+        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<String> = (0..20).map(|i| format!("p{i}")).collect();
+        let more = leaf
+            .consider(names.iter().map(|n| candidate(n)).collect::<Vec<_>>());
+        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<crate::message::Message> = Vec::new();
+        while let Ok(ServerMessage::SendMessage(m)) = rx.try_recv() {
+            sent.push(m);
+        }
+        let codes: Vec<u32> = 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");
+    }
+}

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<Instant>,
 }
 
 /// 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<PeerRegistry>,
     pub downloads: DownloadStore,
     server_sender: Option<Sender<ServerMessage>>,
+    /// Where anything outside the operations loop posts operations.
+    operations: Option<Sender<ClientOperation>>,
+    /// Our place in the distributed search network.
+    leaf: distributed::Leaf,
     searches: HashMap<String, Search>,
     private_messages: Vec<UserMessage>,
     /// Correlation tokens for server-brokered (firewalled) connections, mapping
@@ -349,8 +387,11 @@ pub struct ClientContext {
     pub shared_directories: Vec<String>,
     /// Peer listen addresses learned from GetPeerAddress responses.
     peer_addresses: HashMap<String, (String, u32)>,
-    /// Peer messages waiting for a control connection to that peer.
-    pending_peer_messages: HashMap<String, Vec<crate::message::Message>>,
+    /// 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<String, (Instant, Vec<crate::message::Message>)>,
     /// Uploads we have offered, keyed by our transfer token.
     uploads: HashMap<u32, UploadJob>,
     active_uploads: HashMap<u32, ActiveUpload>,
@@ -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<crate::message::Message> {
         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;

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<RwLock<ClientContext>>) {
+        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<RwLock<ClientContext>>) {
         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<crate::message::peer::SharedDirectory> {
-    ctx.shares
-        .directories()
-        .into_iter()
-        .map(|(name, files)| crate::message::peer::SharedDirectory {
-            name,
-            files,
-        })
-        .collect()
-}

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");
+}

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<u32> = 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

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,

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<Distributed> {
+    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<Distributed> {
+    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<Message> {
+    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()
+            })
+        );
+    }
+}

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;

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());

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;

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<u32> {
+        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<SharedFileEntry>,
 }
 
 /// 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<SharedDirectory> {
                 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);

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<ServerMessage> for PossibleParentsHandler {
+    fn get_code(&self) -> u32 {
+        102
+    }
+    fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
+        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<ServerMessage> for ResetDistributedHandler {
+    fn get_code(&self) -> u32 {
+        130
+    }
+    fn handle(&self, _message: &mut Message, sender: Sender<ServerMessage>) {
+        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<ServerMessage> for EmbeddedMessageHandler {
+    fn get_code(&self) -> u32 {
+        93
+    }
+    fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
+        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")
+        );
+    }
+}

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 {

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;

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;

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<Vec<(u32, u32)>> {
+    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<Frame> {
+    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<Vec<(u32, u32)>> {
+    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<u8> {
+        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<u8> {
+        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());
+    }
+}

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<String, Vec<(String, u64)>> =
-            std::collections::BTreeMap::new();
+    pub fn directories(&self) -> Vec<SharedDirectory> {
+        let mut by_dir: std::collections::BTreeMap<
+            String,
+            Vec<SharedFileEntry>,
+        > = 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();

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<String>)> = 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<u8> {
+    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<TcpStream> {
+    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);
+}

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": [
     {

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 <user>`** — 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 <name>`** — one object: `user`, `status` (`online`/`away`/`offline`),
 `privileged`, `average_speed`, `shared_files`, `shared_folders`. Any field may

soulseek-rs/src/cli.rs

@@ -172,7 +172,11 @@ pub struct Cli {
     pub log_file: Option<PathBuf>,
 }
 
+/// 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),

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),
             });
         }
     }

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,
                             )
                         })
                     })

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<UserInfoDto> 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<SharedFileEntryDto>,
+}
+
+#[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<SharedDirectoryDto> 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(),
         }
     }
 }

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();

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(),
         }
     }
 

soulseek-rs/src/output.rs

@@ -207,6 +207,8 @@ pub struct BrowseRecord {
     pub directory: String,
     pub path: String,
     pub size: u64,
+    pub bitrate: Option<u32>,
+    pub duration: Option<u32>,
 }
 
 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");
     }

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");
     }
 

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

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<u8> {
+    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!();