feat(lib): decode distributed frames and announce a parentless stance

The distributed search network carries searches from parent to child
over connections whose frames have a one-byte code; the library had no
reader for them. This adds the codec for Search, BranchLevel,
BranchRoot and the embedded envelope, with the identifier check that
keeps malformed frames from being answered, and completes the post-login
stance Nicotine+ sends: HaveNoParent, BranchRoot (ourselves),
BranchLevel 0 and AcceptChildren false, in place of HaveNoParent alone.

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

@@ -175,32 +175,46 @@ 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, 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, its distributed-network stance, 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,
     shared_folders: u32,
     shared_files: u32,
+    own_username: &str,
 ) -> Vec<Message> {
     let mut messages = vec![
         MessageFactory::build_shared_folders_message(
             shared_folders,
             shared_files,
         ),
-        MessageFactory::build_no_parent_message(),
         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 and takes no children. Nicotine+ sends the same four.
+fn parentless_stance(own_username: &str) -> Vec<Message> {
+    vec![
+        MessageFactory::build_have_no_parent(true),
+        MessageFactory::build_branch_root(own_username),
+        MessageFactory::build_branch_level(0),
+        MessageFactory::build_accept_children(false),
+    ]
+}
+
 impl ServerActor {
     #[must_use]
     pub fn new(
@@ -228,6 +242,7 @@ impl ServerActor {
             shared_folder_count,
             shared_file_count,
             session: SessionWatch::default(),
+            username: String::new(),
         }
     }
 
@@ -550,6 +565,7 @@ impl ServerActor {
                 self.listen_port,
                 self.shared_folder_count,
                 self.shared_file_count,
+                &self.username,
             ) {
                 self.send_message(msg);
             }
@@ -621,6 +637,7 @@ 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,10 +193,18 @@ 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);
+    let messages = post_login_messages(true, 4321, 3, 7, "me");
     let codes: Vec<u32> = messages.iter().map(code_of).collect();
-    // SharedFolders, HaveNoParent, SetStatus, SetWaitPort.
-    assert_eq!(codes, vec![35, 71, 28, 2]);
+    // SharedFolders; the distributed stance HaveNoParent, BranchRoot,
+    // BranchLevel, AcceptChildren; SetStatus; SetWaitPort.
+    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();
@@ -204,7 +212,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);
+    let no_listen = post_login_messages(false, 4321, 3, 7, "me");
     let codes: Vec<u32> = no_listen.iter().map(code_of).collect();
-    assert_eq!(codes, vec![35, 71, 28]);
+    assert_eq!(codes, vec![35, 71, 127, 126, 100, 28]);
 }

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

@@ -0,0 +1,160 @@
+//! 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;
+
+/// 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_body(message, message.get_init_code())
+}
+
+fn parse_body(message: &mut Message, code: u8) -> Option<Distributed> {
+    match code {
+        3 => {
+            if message.read_int32() != SEARCH_IDENTIFIER {
+                return None;
+            }
+            let username = message.read_string();
+            let token = message.read_int32();
+            let query = message.read_string();
+            (!username.is_empty() && !query.is_empty()).then_some(
+                Distributed::Search {
+                    username,
+                    token,
+                    query,
+                },
+            )
+        }
+        4 => Some(Distributed::BranchLevel(message.read_int32() as i32)),
+        5 => Some(Distributed::BranchRoot(message.read_string())),
+        // Older branch roots wrap a search in a code-93 envelope.
+        93 => {
+            let inner = message.read_int8();
+            parse_body(message, inner)
+        }
+        _ => None,
+    }
+}
+
+/// 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);
+    })
+}
+
+#[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);
+    }
+}

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/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 {