Merge pull request #84 from michel/lib/distributed-leaf

feat(lib): join the distributed search network as a leaf

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);
 
@@ -334,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
@@ -351,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>,
@@ -445,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());
@@ -452,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(),
@@ -692,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`.
@@ -705,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,
@@ -834,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(),
         }
@@ -982,6 +1045,7 @@ impl Client {
 }
 
 mod connection;
+mod distributed;
 mod downloads;
 mod operations;
 mod rooms;

soulseek-rs-lib/src/client/operations.rs

@@ -24,6 +24,18 @@ impl Client {
                     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,
@@ -696,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!(

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/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/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/tests/e2e.rs

@@ -23,6 +23,7 @@ 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::{
     Client, ClientSettings, ClientVersion, ConnectionType, DownloadStatus,
@@ -3616,3 +3617,154 @@ fn a_browse_listing_carries_the_files_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);
+}