From 85f26134eda8434be7647de3d521d98c7c1c8a63 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 01:25:02 +0200 Subject: [PATCH] fix(lib): free a silent parent's slot and pace the moves we report A candidate that accepted the dial and then said nothing kept its slot for the session, and ten of those blocked every later batch; a link now closes after a minute of silence. A parent that flaps its level made us tell the server four messages per flap; moves are now told at most once a second, the latest one winning. A parent that reports level zero is its own root from then on, whatever it said before. The wait for a searcher's control connection runs from the last message queued for it, not the first. A successful login is pinned to reset the leaf. --- .../src/actor/server_actor/tests.rs | 29 ++++++- soulseek-rs-lib/src/client/distributed.rs | 75 +++++++++++++++++-- soulseek-rs-lib/src/client/mod.rs | 12 +-- soulseek-rs-lib/src/client/operations.rs | 9 +++ 4 files changed, 111 insertions(+), 14 deletions(-) diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs index aa99422..8faba13 100644 --- a/soulseek-rs-lib/src/actor/server_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -36,15 +36,22 @@ fn the_server_socket_keeps_itself_alive() { } fn parked_actor(port: u16) -> ServerActor { - let (tx, _rx) = std::sync::mpsc::channel(); - ServerActor::new( + parked_actor_with_client(port).0 +} + +fn parked_actor_with_client( + port: u16, +) -> (ServerActor, std::sync::mpsc::Receiver) { + let (tx, rx) = std::sync::mpsc::channel(); + let actor = ServerActor::new( PeerAddress::new("127.0.0.1".to_string(), port), tx, 0, false, 0, 0, - ) + ); + (actor, rx) } #[test] @@ -191,6 +198,22 @@ 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); diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs index b936d40..013ecf9 100644 --- a/soulseek-rs-lib/src/client/distributed.rs +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -20,6 +20,13 @@ use std::time::{Duration, Instant}; 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; @@ -82,6 +89,10 @@ pub struct Leaf { candidates: HashMap, next_link: u64, searches: (Instant, u32), + /// When we last told the server we moved, and whether a move since then + /// is still waiting to be told. + moved: Option, + move_pending: bool, } impl Leaf { @@ -94,6 +105,8 @@ impl Leaf { candidates: HashMap::new(), next_link: 1, searches: (Instant::now(), 0), + moved: None, + move_pending: false, } } @@ -163,7 +176,7 @@ impl Leaf { } let candidate = self.candidates.get_mut(username)?; candidate.level = u32::try_from(level).ok(); - if level == 0 && candidate.root.is_none() { + if level == 0 { candidate.root = Some(username.to_string()); } self.moved_with_parent(username) @@ -251,8 +264,24 @@ impl Leaf { if branch == self.branch { return None; } - self.branch = branch.clone(); - Some(branch) + self.branch = branch; + self.move_pending = true; + self.due_announcement(Instant::now()) + } + + /// The branch to tell the server about, if a move is waiting and the + /// last one was told at least [`MOVE_INTERVAL`] ago. + pub fn due_announcement(&mut self, now: Instant) -> Option { + if !self.move_pending + || self + .moved + .is_some_and(|at| now.duration_since(at) < MOVE_INTERVAL) + { + return None; + } + self.move_pending = false; + self.moved = Some(now); + Some(self.branch.clone()) } fn cancel_candidates(&mut self) { @@ -328,18 +357,25 @@ fn link( .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(()) => {} + 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), @@ -516,7 +552,7 @@ mod tests { } #[test] - fn the_parents_later_moves_move_us_too() { + 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); @@ -528,9 +564,36 @@ mod tests { ); 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.branch(), 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] diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 160e7a6..5dc3a34 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -388,7 +388,8 @@ pub struct ClientContext { /// Peer listen addresses learned from GetPeerAddress responses. peer_addresses: HashMap, /// Peer messages waiting for a control connection to that peer, and - /// since when: a peer that never answers must not keep them forever. + /// when the last was queued: a peer that never answers must not keep + /// them forever. pending_peer_messages: HashMap)>, /// Uploads we have offered, keyed by our transfer token. @@ -741,11 +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_insert_with(|| (Instant::now(), Vec::new())) - .1 - .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`. diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 6d47d09..bedf5a3 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -26,6 +26,15 @@ impl Client { 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 {