fix(lib): harden the distributed leaf against hostile and slow parents

Review of the leaf found the seams a stranger on the other end of a D
connection could pull at, and a few that a slow one would.

- A code-93 envelope is unwrapped once, as Nicotine+ does; a frame made
  of nothing but envelopes used to recurse once per byte.
- A branch level is validated before it is used: negative levels leave
  the candidate unadoptable and the largest one no longer overflows.
- The link reads through the crate's MessageReader, so a frame split
  across a read timeout is no longer torn or mistaken for a closed link.
- Every event a link raises carries its link number, so a stale thread
  cannot adopt, close or move a newer link to the same user.
- Later branch facts from the parent move our own place and are told to
  the server; a fresh login resets the leaf, whose stance is now the one
  place the four announcement messages come from.
- Candidates from a later batch join the ones in flight instead of
  cancelling them, capped at ten; searches from the parent are budgeted
  at fifty a second; a reply queued for a searcher the server cannot
  place expires after a minute.
- Truncated frames and half-read PossibleParents entries are dropped
  instead of yielding an empty root, a level of zero or a phantom
  candidate.

The e2e now has a searcher of its own, so the answer must go to whoever
asked and not to the parent that relayed it.

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

@@ -183,12 +183,10 @@ pub struct ServerActor {
     shared_folder_count: u32,
     shared_file_count: u32,
     session: SessionWatch,
-    /// The name we logged in as, once a login has been asked for.
-    username: String,
 }
 
 /// The messages a client sends right after a successful login: its shared-file
-/// counts, its distributed-network stance, online status, and (when listening)
+/// 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(
@@ -196,7 +194,6 @@ fn post_login_messages(
     listen_port: u16,
     shared_folders: u32,
     shared_files: u32,
-    own_username: &str,
 ) -> Vec<Message> {
     let mut messages = vec![
         MessageFactory::build_shared_folders_message(
@@ -205,19 +202,12 @@ fn post_login_messages(
         ),
         MessageFactory::build_set_status_message(2),
     ];
-    messages.splice(1..1, parentless_stance(own_username));
     if enable_listen {
         messages.push(MessageFactory::build_set_wait_port_message(listen_port));
     }
     messages
 }
 
-/// What a leaf tells the server while it has no parent: it is its own branch
-/// root at level 0.
-fn parentless_stance(own_username: &str) -> Vec<Message> {
-    crate::message::distributed::stance(own_username, 0, false)
-}
-
 impl ServerActor {
     #[must_use]
     pub fn new(
@@ -245,7 +235,6 @@ impl ServerActor {
             shared_folder_count,
             shared_file_count,
             session: SessionWatch::default(),
-            username: String::new(),
         }
     }
 
@@ -579,11 +568,13 @@ impl ServerActor {
                 self.listen_port,
                 self.shared_folder_count,
                 self.shared_file_count,
-                &self.username,
             ) {
                 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),
@@ -651,7 +642,6 @@ impl ServerActor {
         version: ClientVersion,
         response: std::sync::mpsc::Sender<Result<bool, SoulseekRs>>,
     ) {
-        self.username.clone_from(&username);
         if self.stream.is_none() && !self.initiate_connection() {
             let _ = response.send(Err(SoulseekRs::NotConnected));
             return;

soulseek-rs-lib/src/actor/server_actor/tests.rs

@@ -193,18 +193,10 @@ fn a_successful_login_marks_the_session_live_again() {
 
 #[test]
 fn post_login_messages_carry_counts_and_conditional_wait_port() {
-    let messages = post_login_messages(true, 4321, 3, 7, "me");
+    let messages = post_login_messages(true, 4321, 3, 7);
     let codes: Vec<u32> = messages.iter().map(code_of).collect();
-    // SharedFolders; the distributed stance HaveNoParent, BranchRoot,
+    // SharedFolders; SetStatus; SetWaitPort.
-    // BranchLevel, AcceptChildren; SetStatus; SetWaitPort.
+    assert_eq!(codes, vec![35, 28, 2]);
-    assert_eq!(codes, vec![35, 71, 127, 126, 100, 28, 2]);
-    // A leaf with no parent is its own branch root at level 0, and it
-    // declines children.
-    let mut root = messages[2].clone();
-    root.set_pointer(4);
-    assert_eq!(root.read_string(), "me");
-    assert_eq!(&messages[3].get_data()[4..8], &0u32.to_le_bytes());
-    assert_eq!(messages[4].get_data()[4], 0);
 
     // The SharedFolders message (code 35) carries the real counts.
     let shared = messages[0].get_data();
@@ -212,7 +204,7 @@ fn post_login_messages_carry_counts_and_conditional_wait_port() {
     assert_eq!(u32::from_le_bytes(shared[8..12].try_into().unwrap()), 7);
 
     // Not listening omits SetWaitPort (code 2).
-    let no_listen = post_login_messages(false, 4321, 3, 7, "me");
+    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, 127, 126, 100, 28]);
+    assert_eq!(codes, vec![35, 28]);
 }

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

@@ -3,18 +3,18 @@
 //! server when that changes.
 
 use super::{Client, ClientOperation, Result, ServerMessage};
-use crate::message::Message;
+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, Read, Write};
+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;
+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);
@@ -23,6 +23,13 @@ const CANCEL_POLL: 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)]
@@ -46,15 +53,17 @@ impl Branch {
     }
 }
 
-#[derive(Default)]
 struct Candidate {
-    level: Option<i32>,
+    link: u64,
+    level: Option<u32>,
     root: Option<String>,
     cancel: Arc<AtomicBool>,
 }
 
-/// A candidate to dial, and the flag that tells its link to stop.
+/// 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,
@@ -63,12 +72,16 @@ pub struct Dial {
 
 /// 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.
+/// 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),
 }
 
 impl Leaf {
@@ -79,6 +92,8 @@ impl Leaf {
             parent: None,
             branch: Branch::parentless(own),
             candidates: HashMap::new(),
+            next_link: 1,
+            searches: (Instant::now(), 0),
         }
     }
 
@@ -87,13 +102,15 @@ impl Leaf {
         self.branch.clone()
     }
 
+    /// Whether `link` is the live link to our parent.
     #[must_use]
-    pub fn is_parent(&self, username: &str) -> bool {
+    pub fn is_parent(&self, username: &str, link: u64) -> bool {
         self.parent.as_deref() == Some(username)
+            && self.has_link(username, link)
     }
 
-    /// New candidates: dial them all, unless we already hang from a parent.
+    /// New candidates to dial: those we are not already sounding out, up to
-    /// Candidates still being sounded out from an earlier batch are dropped.
+    /// the cap, and none at all while we hang from a parent.
     pub fn consider(
         &mut self,
         candidates: Vec<(String, String, u16)>,
@@ -101,50 +118,74 @@ impl Leaf {
         if self.parent.is_some() {
             return Vec::new();
         }
-        self.cancel_candidates();
+        let mut dials = Vec::new();
-        candidates
+        for (username, host, port) in candidates {
-            .into_iter()
+            if self.candidates.contains_key(&username)
-            .map(|(username, host, port)| {
+                || 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(),
-                        ..Candidate::default()
                 },
             );
-                Dial {
+            dials.push(Dial {
+                link,
                 username,
                 host,
                 port,
                 cancel,
+            });
         }
-            })
+        dials
-            .collect()
     }
 
-    /// A branch root does not always say so separately: level 0 is its own
+    /// A candidate said how deep it sits. A branch root does not always say
-    /// root.
+    /// so separately: level 0 is its own root. A negative level is nonsense
-    pub fn branch_level(&mut self, username: &str, level: i32) {
+    /// and leaves the candidate unadoptable. When it is the parent talking,
-        let Some(candidate) = self.candidates.get_mut(username) else {
+    /// our own place moves with it and the new stance comes back.
-            return;
+    pub fn branch_level(
-        };
+        &mut self,
-        candidate.level = Some(level);
+        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.is_none() {
             candidate.root = Some(username.to_string());
         }
+        self.moved_with_parent(username)
     }
 
-    pub fn branch_root(&mut self, username: &str, root: &str) {
+    pub fn branch_root(
-        if let Some(candidate) = self.candidates.get_mut(username) {
+        &mut self,
-            candidate.root = Some(root.to_string());
+        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
+    /// A search came down from `username`. The first one from a candidate
-    /// has told us its place adopts it; the stance to announce comes back.
+    /// that has told us its place adopts it; the stance to announce comes back.
-    pub fn search_from(&mut self, username: &str) -> Option<Branch> {
+    pub fn search_from(&mut self, username: &str, link: u64) -> Option<Branch> {
-        if self.parent.is_some() {
+        if self.parent.is_some() || !self.has_link(username, link) {
             return None;
         }
         let candidate = self.candidates.get(username)?;
@@ -157,16 +198,28 @@ impl Leaf {
         self.cancel_candidates();
         self.candidates.insert(username.to_string(), parent);
         self.parent = Some(username.to_string());
-        self.branch =
+        self.branch = Branch::under(&root, level.saturating_add(1));
-            Branch::under(&root, u32::try_from(level + 1).unwrap_or(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) -> bool {
+    pub fn closed(&mut self, username: &str, link: u64) -> bool {
+        if !self.has_link(username, link) {
+            return false;
+        }
         self.candidates.remove(username);
-        if !self.is_parent(username) {
+        if self.parent.as_deref() != Some(username) {
             return false;
         }
         self.parent = None;
@@ -180,6 +233,28 @@ impl Leaf {
         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.clone();
+        Some(branch)
+    }
+
     fn cancel_candidates(&mut self) {
         for candidate in self.candidates.values() {
             candidate.cancel.store(true, Ordering::Relaxed);
@@ -205,17 +280,31 @@ pub fn announce(
 /// 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 _ = std::thread::Builder::new()
+    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 || {
-            let parent = dial.username.clone();
+            if let Err(e) = link(&dial, &own, &thread_ops)
-            if let Err(e) = link(&dial, &own, &ops)
                 && !dial.cancel.load(Ordering::Relaxed)
             {
-                crate::debug!("[distributed] link to {} ended: {}", parent, e);
+                crate::debug!(
-                let _ = ops.send(ClientOperation::ParentClosed { parent });
+                    "[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(
@@ -230,48 +319,20 @@ fn link(
             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();
     loop {
         if dial.cancel.load(Ordering::Relaxed) {
             return Ok(());
         }
-        let Some(mut frame) = read_frame(&mut stream)? else {
+        match reader.read_from_socket(&mut stream) {
-            continue;
-        };
-        let parent = dial.username.clone();
-        let operation = match distributed::parse(&mut frame) {
-            Some(Distributed::BranchLevel(level)) => {
-                ClientOperation::ParentBranchLevel { parent, level }
-            }
-            Some(Distributed::BranchRoot(root)) => {
-                ClientOperation::ParentBranchRoot { parent, root }
-            }
-            Some(Distributed::Search {
-                username,
-                token,
-                query,
-            }) => ClientOperation::ParentSearch {
-                parent,
-                username,
-                token,
-                query,
-            },
-            None => continue,
-        };
-        if ops.send(operation).is_err() {
-            return Ok(());
-        }
-    }
-}
-
-/// One framed message, or `None` when the read timed out with nothing sent.
-fn read_frame(stream: &mut TcpStream) -> io::Result<Option<Message>> {
-    let mut len = [0u8; 4];
-    match stream.read_exact(&mut len) {
             Ok(()) => {}
             Err(e)
                 if matches!(
@@ -279,21 +340,54 @@ fn read_frame(stream: &mut TcpStream) -> io::Result<Option<Message>> {
                     io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
                 ) =>
             {
-            return Ok(None);
+                continue;
             }
             Err(e) => return Err(e),
         }
-    let size = u32::from_le_bytes(len) as usize;
+        while let Some(mut frame) = reader.extract_message()? {
-    if size > MAX_FRAME {
+            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",
             ));
         }
-    let mut data = len.to_vec();
+    }
-    data.resize(4 + size, 0);
+}
-    stream.read_exact(&mut data[4..])?;
+
-    Ok(Some(Message::new_with_data(data)))
+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 {
@@ -339,6 +433,18 @@ mod tests {
         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"));
@@ -347,12 +453,13 @@ mod tests {
     #[test]
     fn candidates_are_dialled_only_while_there_is_no_parent() {
         let mut leaf = leaf();
-        let dial = leaf.consider(vec![candidate("alice"), candidate("bob")]);
+        let dials = dial(&mut leaf, &["alice", "bob"]);
-        assert_eq!(dial.len(), 2);
+        assert_eq!(dials.len(), 2);
-        leaf.branch_level("alice", 1);
+        let alice = dials["alice"].link;
-        leaf.branch_root("alice", "rooty");
+        leaf.branch_level("alice", alice, 1);
+        leaf.branch_root("alice", alice, "rooty");
         assert!(
-            leaf.search_from("alice").is_some(),
+            leaf.search_from("alice", alice).is_some(),
             "alice becomes the parent"
         );
         assert!(leaf.consider(vec![candidate("carol")]).is_empty());
@@ -361,48 +468,158 @@ mod tests {
     #[test]
     fn the_first_search_from_a_candidate_with_branch_info_adopts_it() {
         let mut leaf = leaf();
-        leaf.consider(vec![candidate("alice"), candidate("bob")]);
+        let dials = dial(&mut leaf, &["alice", "bob"]);
-        assert_eq!(leaf.search_from("alice"), None, "no branch info yet");
+        let (alice, bob) = (dials["alice"].link, dials["bob"].link);
-        leaf.branch_level("alice", 3);
+        assert_eq!(leaf.search_from("alice", alice), None, "no branch yet");
-        assert_eq!(leaf.search_from("alice"), None, "level without root");
+        leaf.branch_level("alice", alice, 3);
-        leaf.branch_root("alice", "rooty");
+        assert_eq!(leaf.search_from("alice", alice), None, "level, no root");
+        leaf.branch_root("alice", alice, "rooty");
         assert_eq!(
-            leaf.search_from("alice"),
+            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_eq!(leaf.search_from("alice"), None, "already the parent");
+        assert!(leaf.is_parent("alice", alice));
-        assert_eq!(leaf.search_from("bob"), None, "a loser is ignored");
+        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();
-        leaf.consider(vec![candidate("alice")]);
+        let alice = dial(&mut leaf, &["alice"])["alice"].link;
-        leaf.branch_level("alice", 0);
+        leaf.branch_level("alice", alice, 0);
-        assert_eq!(leaf.search_from("alice"), Some(Branch::under("alice", 1)));
+        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() {
+        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),
+            Some(Branch::under("grand", 3))
+        );
+        assert_eq!(leaf.branch(), Branch::under("grand", 3));
+    }
+
+    #[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();
-        leaf.consider(vec![candidate("alice")]);
+        let dials = dial(&mut leaf, &["alice", "bob"]);
-        leaf.branch_level("alice", 0);
+        let alice = dials["alice"].link;
-        leaf.search_from("alice");
+        leaf.branch_level("alice", alice, 0);
-        assert!(!leaf.closed("bob"), "a stranger closing changes nothing");
+        leaf.search_from("alice", alice);
-        assert!(leaf.closed("alice"));
+        assert!(!leaf.closed("bob", dials["bob"].link), "not the parent");
+        assert!(leaf.closed("alice", alice));
         assert_eq!(leaf.branch(), Branch::parentless("me"));
 
-        leaf.consider(vec![candidate("alice")]);
+        let dials = dial(&mut leaf, &["alice", "carol"]);
-        leaf.branch_level("alice", 0);
+        let alice = dials["alice"].link;
-        leaf.search_from("alice");
+        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());
     }
 
-    fn candidate(name: &str) -> (String, String, u16) {
+    #[test]
-        (name.to_string(), "127.0.0.1".to_string(), 2234)
+    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);
 
@@ -338,19 +340,23 @@ pub enum ClientOperation {
     PossibleParents(Vec<(String, String, u16)>),
     /// The server told us to drop our parent.
     ResetDistributed,
-    /// A parent candidate told us how deep it sits.
+    /// 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,
@@ -358,6 +364,7 @@ pub enum ClientOperation {
     /// The link to a parent or candidate is gone.
     ParentClosed {
         parent: String,
+        link: u64,
     },
 }
 pub struct ClientContext {
@@ -380,8 +387,10 @@ 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.
+    /// Peer messages waiting for a control connection to that peer, and
-    pending_peer_messages: HashMap<String, Vec<crate::message::Message>>,
+    /// since when: 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>,
@@ -474,6 +483,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());
@@ -725,7 +743,8 @@ impl ClientContext {
     ) {
         self.pending_peer_messages
             .entry(username.to_string())
-            .or_default()
+            .or_insert_with(|| (Instant::now(), Vec::new()))
+            .1
             .push(message);
     }
 
@@ -736,9 +755,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,
@@ -865,11 +893,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(),
         }

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

@@ -17,9 +17,6 @@ impl Client {
         own_username: String,
     ) {
         thread::spawn(move || {
-            if let Ok(mut ctx) = client_context.write_safe() {
-                ctx.leaf = super::distributed::Leaf::new(&own_username);
-            }
             let mut last_sweep = Instant::now();
             loop {
                 let next = reader.recv_timeout(CONNECT_SWEEP_INTERVAL);
@@ -27,6 +24,9 @@ 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());
+                    }
                 }
                 let operation = match next {
                     Ok(operation) => operation,
@@ -716,18 +716,45 @@ impl Client {
                             );
                         }
                     }
-                    ClientOperation::ParentBranchLevel { parent, level } => {
+                    ClientOperation::ParentBranchLevel {
-                        if let Ok(mut ctx) = client_context.write_safe() {
+                        parent,
-                            ctx.leaf.branch_level(&parent, level);
+                        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, root } => {
+                    ClientOperation::ParentBranchRoot {
-                        if let Ok(mut ctx) = client_context.write_safe() {
+                        parent,
-                            ctx.leaf.branch_root(&parent, &root);
+                        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,
@@ -735,14 +762,17 @@ impl Client {
                         let Ok(mut ctx) = client_context.write_safe() else {
                             continue;
                         };
-                        if let Some(branch) = ctx.leaf.search_from(&parent) {
+                        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)
+                        if ctx.leaf.is_parent(&parent, link)
+                            && ctx.leaf.admit_search(Instant::now())
                             && let Some(ops) = &ctx.operations
                         {
                             let _ = ops.send(ClientOperation::IncomingSearch {
@@ -752,11 +782,11 @@ impl Client {
                             });
                         }
                     }
-                    ClientOperation::ParentClosed { parent } => {
+                    ClientOperation::ParentClosed { parent, link } => {
                         let Ok(mut ctx) = client_context.write_safe() else {
                             continue;
                         };
-                        if ctx.leaf.closed(&parent) {
+                        if ctx.leaf.closed(&parent, link) {
                             super::distributed::announce(
                                 ctx.server_sender.as_ref(),
                                 &ctx.leaf.branch(),
@@ -764,6 +794,8 @@ impl Client {
                             );
                         }
                     }
+                    // 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;

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

@@ -34,15 +34,27 @@ pub fn parse(message: &mut Message) -> Option<Distributed> {
     parse_at(message, message.get_init_code())
 }
 
-/// Decode the body that follows a one-byte `code`, the pointer already past it.
+/// 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 message.read_int32() != SEARCH_IDENTIFIER {
+            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(
@@ -53,17 +65,20 @@ pub fn parse_at(message: &mut Message, code: u8) -> Option<Distributed> {
                 },
             )
         }
-        4 => Some(Distributed::BranchLevel(message.read_int32() as i32)),
+        4 => (remaining(message) >= 4)
-        5 => Some(Distributed::BranchRoot(message.read_string())),
+            .then(|| Distributed::BranchLevel(message.read_int32() as i32)),
-        // Older branch roots wrap a search in a code-93 envelope.
+        5 => {
-        93 => {
+            let root = message.read_string();
-            let inner = message.read_int8();
+            (!root.is_empty()).then_some(Distributed::BranchRoot(root))
-            parse_at(message, inner)
         }
         _ => 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 {
@@ -171,5 +186,59 @@ mod tests {
         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/server/distributed.rs

@@ -4,6 +4,7 @@
 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;
@@ -13,22 +14,29 @@ impl MessageHandler<ServerMessage> for PossibleParentsHandler {
     }
     fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
         let count = message.read_int32();
-        let mut candidates = Vec::new();
+        let candidates = (0..count.min(10))
-        for _ in 0..count.min(10) {
+            .map_while(|_| entry(message))
-            if message.get_pointer() >= message.get_size() {
+            .filter_map(|(username, host, port)| {
-                break;
+                Some((username, host, u16::try_from(port).ok()?))
-            }
+            })
-            let username = message.read_string();
+            .collect();
-            // The address travels as four bytes in network order.
-            let ip: Vec<u8> = (0..4).map(|_| message.read_int8()).collect();
-            let host = format!("{}.{}.{}.{}", ip[3], ip[2], ip[1], ip[0]);
-            let port = message.read_int32() as u16;
-            candidates.push((username, host, port));
-        }
         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 {
@@ -95,6 +103,33 @@ mod tests {
         );
     }
 
+    // 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();

soulseek-rs-lib/tests/e2e.rs

@@ -3669,20 +3669,21 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() {
     std::fs::write(share_dir.join("album").join("treesearch.mp3"), b"xxxx")
         .unwrap();
 
-    // The parent doubles as the searcher, so the leaf's answer comes back to
+    // The parent and the searcher are two logged-in users with a listener
-    // the same listener. Bind all interfaces: soulfind reports the LAN address.
+    // each. Bind all interfaces: soulfind reports the LAN address.
-    let parent_port = free_port().expect("parent port");
-    let listener =
-        std::net::TcpListener::bind(("0.0.0.0", parent_port)).unwrap();
     let server_addr = format!("{}:{}", server.host, server.port);
-    let mut parent_srv =
+    let raw_user = |name: &str| {
-        login_raw(&server_addr, "e2e_parent", "pw").expect("parent logs in");
+        let port = free_port().expect("a port for the raw user");
-    parent_srv
+        let listener = std::net::TcpListener::bind(("0.0.0.0", port)).unwrap();
-        .write_all(
+        let mut srv = login_raw(&server_addr, name, "pw").expect("raw login");
-            &MessageFactory::build_set_wait_port_message(parent_port)
+        srv.write_all(
-                .get_buffer(),
+            &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()],
@@ -3721,7 +3722,7 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() {
     link.write_all(&distributed::build_branch_root("e2e_parent").get_buffer())
         .unwrap();
     link.write_all(
-        &distributed::build_search("e2e_parent", 4242, "treesearch")
+        &distributed::build_search("e2e_searcher", 4242, "treesearch")
             .get_buffer(),
     )
     .unwrap();
@@ -3736,9 +3737,10 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() {
         "the first search from a candidate with a branch adopts it"
     );
 
-    // The answer arrives the ordinary way: a P connection to the searcher
+    // The answer goes to whoever asked, not to the parent that relayed it:
-    // carrying a FileSearchResponse for our token.
+    // a P connection to the searcher carrying a FileSearchResponse for the
-    let mut p = accept_within(&listener, Duration::from_secs(20))
+    // 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(),