fix: review findings on the protocol work

Correctness:
- A child that stopped reading could freeze the whole client: the relay
  writes under the context lock and the socket was blocking, so one stalled
  peer took every download, upload and API call with it. Child writes now
  have a deadline, and a child that hits it loses its slot.
- Relaying searches to children was gated by the leaf's answer budget, so a
  parent forwarded a fraction of the stream while advertising itself as
  healthy. The budget bounds our own share scan, which is what it was for;
  the tree's stream goes down whole.
- A dropped client kept its listener port for the life of the process, so a
  retried login advertised an ephemeral port nobody was told about. The
  listener is now stopped and the operations thread let go.
- Excluded phrases were compared against a lowercased path without being
  lowercased themselves, so a capitalised phrase never matched. They are
  normalised where they arrive.
- Losing our own operatorship wiped the room's whole operator roster,
  though we are still a member who can see it.
- A brokered ConnectToPeer of type D was logged and dropped; the peer is
  now told, since we never dial out to adopt a child.
- A peer request made with no server connection returned Ok and parked
  forever; it says NotConnected, as its docs always claimed.
- An empty interest was refused locally but still sent, leaving the
  server's list and ours permanently apart.

Simplification:
- Deleted the excluded-phrase helper nothing called, and the field doc that
  promised behaviour it never had.
- The four roster handlers and four standing handlers were eight
  constructors carrying constants their own message code already implies:
  one newtype each, 100 lines less.
- One place registers a search; the phrase list is borrowed, not cloned per
  relayed search.
This commit is contained in:
Michel de Graaf
2026-09-09 18:38:34 +00:00
parent 435ff382d8
commit 6f32a4e090
11 changed files with 294 additions and 194 deletions
+9 -10
View File
@@ -427,16 +427,14 @@ impl ServerActor {
handlers.register_handler(CantConnectToPeerHandler); handlers.register_handler(CantConnectToPeerHandler);
handlers.register_handler(RoomMembersHandler); handlers.register_handler(RoomMembersHandler);
handlers.register_handler(RoomOperatorsHandler); handlers.register_handler(RoomOperatorsHandler);
handlers.register_handler(RoomRosterChangeHandler::member_added()); // Members added/removed (134/135), operators added/removed (143/144).
handlers.register_handler(RoomRosterChangeHandler::member_removed()); for code in [134, 135, 143, 144] {
handlers.register_handler(RoomRosterChangeHandler::operator_added()); handlers.register_handler(RoomRosterChangeHandler(code));
handlers.register_handler(RoomRosterChangeHandler::operator_removed()); }
handlers.register_handler(OwnRoomStandingHandler::membership_granted()); // Our own membership (139/140) and operatorship (145/146).
handlers.register_handler(OwnRoomStandingHandler::membership_revoked()); for code in [139, 140, 145, 146] {
handlers handlers.register_handler(OwnRoomStandingHandler(code));
.register_handler(OwnRoomStandingHandler::operatorship_granted()); }
handlers
.register_handler(OwnRoomStandingHandler::operatorship_revoked());
handlers.register_handler(CantCreateRoomHandler); handlers.register_handler(CantCreateRoomHandler);
handlers.register_handler(RoomTickersHandler); handlers.register_handler(RoomTickersHandler);
handlers.register_handler(RoomTickerAddedHandler); handlers.register_handler(RoomTickerAddedHandler);
@@ -954,6 +952,7 @@ impl ServerActor {
) { ) {
if let Err(e) = if let Err(e) =
self.client_channel.send(ClientOperation::IncomingSearch { self.client_channel.send(ClientOperation::IncomingSearch {
from_parent: false,
username, username,
token, token,
query, query,
+42 -11
View File
@@ -1,9 +1,10 @@
use super::{ use super::{
Arc, Client, ClientContext, ClientOperation, ConnectionType, DownloadPeer, Arc, AtomicBool, Client, ClientContext, ClientOperation, ConnectionType,
Listen, Peer, PeerRegistry, Receiver, Result, RwLock, RwLockExt, Sender, DownloadPeer, Listen, Peer, PeerRegistry, Receiver, Result, RwLock,
ServerActor, ServerMessage, Shares, SoulseekRs, TcpStream, error, info, RwLockExt, Sender, ServerActor, ServerMessage, Shares, SoulseekRs,
mpsc, thread, trace, warn, TcpStream, error, info, mpsc, thread, trace, warn,
}; };
use std::sync::atomic::Ordering;
/// Ceiling on the wait for a login verdict. Generous enough for a slow server /// Ceiling on the wait for a login verdict. Generous enough for a slow server
/// and a retrying connect, short enough that an unattended caller ends. /// and a retrying connect, short enough that an unattended caller ends.
@@ -96,6 +97,7 @@ impl Client {
} else { } else {
None None
}; };
self.listener_stopped = Arc::new(AtomicBool::new(false));
let mut server_actor = ServerActor::new( let mut server_actor = ServerActor::new(
self.address.clone(), self.address.clone(),
@@ -118,9 +120,16 @@ impl Client {
let client_sender = listen_sender; let client_sender = listen_sender;
let context = self.context.clone(); let context = self.context.clone();
let own_username = self.username.clone(); let own_username = self.username.clone();
let stopped = self.listener_stopped.clone();
thread::spawn(move || { thread::spawn(move || {
Listen::serve(&listener, client_sender, context, own_username); Listen::serve(
&listener,
client_sender,
context,
own_username,
stopped,
);
}); });
} }
@@ -147,14 +156,26 @@ impl Client {
if let Some(handle) = self.server_handle.take() { if let Some(handle) = self.server_handle.take() {
let _ = handle.stop(); let _ = handle.stop();
} }
let registry = self let registry = match self.context.write_safe() {
.context Ok(mut ctx) => {
.read_safe() // Dropping our end of the operations channel is what lets its
.ok() // thread finish once the peers and the listener have gone.
.and_then(|ctx| ctx.peer_registry.clone()); ctx.operations = None;
ctx.peer_registry.clone()
}
Err(_) => None,
};
if let Some(registry) = registry { if let Some(registry) = registry {
registry.stop_all(); registry.stop_all();
} }
// The listener thread is parked in accept(), so it only notices the
// flag when a connection arrives: dial it once ourselves. Without
// this the port stays bound for the life of the process, and the next
// client falls back to an ephemeral port nobody was told about.
self.listener_stopped.store(true, Ordering::Relaxed);
if let Some(port) = self.bound_port.take() {
let _ = TcpStream::connect(("127.0.0.1", port));
}
} }
/// Log in and wait for the server's verdict. /// Log in and wait for the server's verdict.
@@ -355,7 +376,17 @@ impl Client {
} }
} }
ConnectionType::D => { ConnectionType::D => {
error!("ConnectionType::D not implemented"); // ponytail: we never dial out to adopt a child — children
// reach us through the listener. Tell the peer rather than
// leave it waiting; wire the outbound half if firewalled
// children turn out to matter.
if let (true, Some(token)) = (peer.brokered, peer.token) {
Self::report_cant_connect(
&client_context,
token,
&peer.username,
);
}
} }
} }
} }
+41 -32
View File
@@ -315,6 +315,10 @@ pub enum ClientOperation {
username: String, username: String,
token: u32, token: u32,
query: String, query: String,
/// True when it came down the tree from our parent, which has
/// already passed it to our children and is subject to the answer
/// budget. A search the server relayed is neither.
from_parent: bool,
}, },
/// A peer queued one of our shared files; `requester_key` is the registry /// A peer queued one of our shared files; `requester_key` is the registry
/// key of the peer actor — the peer's username. /// key of the peer actor — the peer's username.
@@ -555,8 +559,8 @@ pub struct ClientContext {
privileged_users: HashSet<String>, privileged_users: HashSet<String>,
/// Seconds of our own privileges left (code 92), once we have asked. /// Seconds of our own privileges left (code 92), once we have asked.
own_privileges: Option<u32>, own_privileges: Option<u32>,
/// Phrases the server will not search for (code 160). A query carrying one /// Phrases the server excludes from the search network (code 160). Files
/// is refused locally rather than spent on the server. /// whose path carries one are left out of the replies we send.
excluded_search_phrases: Vec<String>, excluded_search_phrases: Vec<String>,
/// Peers waiting for one of our upload slots. /// Peers waiting for one of our upload slots.
upload_queue: Vec<QueuedUpload>, upload_queue: Vec<QueuedUpload>,
@@ -779,15 +783,12 @@ impl ClientContext {
members, members,
granted, granted,
} => { } => {
// Our own standing is rendered from the same rosters, so a // Losing membership takes both rosters with it: a room we
// revocation must drop the room rather than leave a roster // cannot enter has none worth showing. Losing operatorship
// we are no longer part of looking current. // leaves us a member who still sees them, and the server
if !granted { // narrates that roster change separately (code 144).
// Losing membership takes the operator roster with it: if !granted && *members {
// a room we cannot enter has no roster worth showing.
if *members {
self.private_room_members.remove(room); self.private_room_members.remove(room);
}
self.private_room_operators.remove(room); self.private_room_operators.remove(room);
} }
} }
@@ -802,29 +803,45 @@ impl ClientContext {
/// Interests are matched by the server case-insensitively, and Nicotine+ /// Interests are matched by the server case-insensitively, and Nicotine+
/// lowercases them before sending; do the same so two spellings of one /// lowercases them before sending; do the same so two spellings of one
/// interest cannot both be held. /// interest cannot both be held.
pub fn add_own_interest(&mut self, item: &str, liked: bool) -> String { /// Returns the stored spelling, or `None` for an empty item — which is
/// no interest, and must not be sent to the server either.
pub fn add_own_interest(
&mut self,
item: &str,
liked: bool,
) -> Option<String> {
let item = item.trim().to_lowercase(); let item = item.trim().to_lowercase();
if item.is_empty() {
return None;
}
let list = if liked { let list = if liked {
&mut self.own_interests.likes &mut self.own_interests.likes
} else { } else {
&mut self.own_interests.hates &mut self.own_interests.hates
}; };
if !item.is_empty() && !list.contains(&item) { if !list.contains(&item) {
list.push(item.clone()); list.push(item.clone());
} }
item Some(item)
} }
/// Forget one of our own interests. /// Forget one of our own interests. `None` for an empty item, as above.
pub fn remove_own_interest(&mut self, item: &str, liked: bool) -> String { pub fn remove_own_interest(
&mut self,
item: &str,
liked: bool,
) -> Option<String> {
let item = item.trim().to_lowercase(); let item = item.trim().to_lowercase();
if item.is_empty() {
return None;
}
let list = if liked { let list = if liked {
&mut self.own_interests.likes &mut self.own_interests.likes
} else { } else {
&mut self.own_interests.hates &mut self.own_interests.hates
}; };
list.retain(|held| held != &item); list.retain(|held| held != &item);
item Some(item)
} }
/// What we like and hate, as last set. /// What we like and hate, as last set.
@@ -1042,9 +1059,11 @@ impl ClientContext {
self.user_interests.get(username).cloned() self.user_interests.get(username).cloned()
} }
/// Record the phrases the server refuses to search for (code 160). /// Record the phrases the server excludes (code 160), lowercased here so
/// the per-file check is a plain `contains` on a lowercased path.
pub fn set_excluded_search_phrases(&mut self, phrases: Vec<String>) { pub fn set_excluded_search_phrases(&mut self, phrases: Vec<String>) {
self.excluded_search_phrases = phrases; self.excluded_search_phrases =
phrases.iter().map(|p| p.to_lowercase()).collect();
} }
/// The phrases the server refuses to search for. /// The phrases the server refuses to search for.
@@ -1053,20 +1072,6 @@ impl ClientContext {
self.excluded_search_phrases.clone() self.excluded_search_phrases.clone()
} }
/// The first excluded phrase `query` contains, if any. Matching is
/// case-insensitive and on substrings, which is how the server applies its
/// own filter.
#[must_use]
pub fn excluded_phrase_in(&self, query: &str) -> Option<String> {
let haystack = query.to_lowercase();
self.excluded_search_phrases
.iter()
.find(|phrase| {
!phrase.is_empty() && haystack.contains(&phrase.to_lowercase())
})
.cloned()
}
/// Forget any interests cached for `username`, so a poll after a fresh /// Forget any interests cached for `username`, so a poll after a fresh
/// request cannot mistake the previous answer for this one. /// request cannot mistake the previous answer for this one.
pub fn invalidate_user_interests(&mut self, username: &str) { pub fn invalidate_user_interests(&mut self, username: &str) {
@@ -1383,6 +1388,9 @@ pub struct Client {
server_handle: Option<ActorHandle<ServerMessage>>, server_handle: Option<ActorHandle<ServerMessage>>,
context: Arc<RwLock<ClientContext>>, context: Arc<RwLock<ClientContext>>,
session: SessionWatch, session: SessionWatch,
/// Tells the peer listener to stop, so a disconnected client releases the
/// port it bound.
listener_stopped: Arc<AtomicBool>,
} }
impl Drop for Client { impl Drop for Client {
@@ -1416,6 +1424,7 @@ impl Client {
shared_directories: settings.shared_directories, shared_directories: settings.shared_directories,
server_handle: None, server_handle: None,
session: SessionWatch::default(), session: SessionWatch::default(),
listener_stopped: Arc::new(AtomicBool::new(false)),
} }
} }
+49 -12
View File
@@ -16,6 +16,11 @@ const CONNECT_SWEEP_INTERVAL: Duration = Duration::from_secs(1);
/// long queue is not pestered. /// long queue is not pestered.
const QUEUE_POSITION_INTERVAL: Duration = Duration::from_secs(300); const QUEUE_POSITION_INTERVAL: Duration = Duration::from_secs(300);
/// How long a write to a child may block. The relay runs under the client
/// context lock, so an unbounded write would freeze every other operation
/// behind one child that stopped reading.
const CHILD_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
impl Client { impl Client {
pub(crate) fn listen_to_client_operations( pub(crate) fn listen_to_client_operations(
reader: Receiver<ClientOperation>, reader: Receiver<ClientOperation>,
@@ -582,30 +587,48 @@ impl Client {
username, username,
token, token,
query, query,
from_parent,
} => { } => {
// Every search we receive comes from the tree: from // Every search we receive comes from the tree: from
// our parent, or from the server acting as one. That // our parent, or from the server acting as one. That
// is what qualifies us to carry children at all — // is what qualifies us to carry children at all —
// Nicotine+ refuses them until something feeds it — // Nicotine+ refuses them until something feeds it —
// and passing the search down is the whole duty. // and passing the search down is the whole duty.
let capacity_changed = match client_context.write_safe() let (capacity_changed, answer) = match client_context
.write_safe()
{ {
Ok(mut ctx) => { Ok(mut ctx) => {
let was_fed = ctx.children.is_fed(); let was_fed = ctx.children.is_fed();
let before = ctx.children.len(); let before = ctx.children.len();
ctx.set_fed_by(true); ctx.set_fed_by(true);
ctx.children if !from_parent {
.broadcast_search(&username, token, &query); // A parent's search was passed down where
// it arrived; this is the server's.
ctx.children.broadcast_search(
&username, token, &query,
);
}
// Answering costs a scan of everything we
// share, so the budget applies to searches
// from the tree. The server's own are few.
let answer = !from_parent
|| ctx.leaf.admit_search(Instant::now());
// A child whose socket has gone is dropped by // A child whose socket has gone is dropped by
// that write, which frees a slot the server // that write, which frees a slot the server
// should hear about. // should hear about.
!was_fed || ctx.children.len() != before (
!was_fed || ctx.children.len() != before,
answer,
)
} }
Err(_) => false, Err(_) => (false, false),
}; };
if capacity_changed { if capacity_changed {
Self::announce_child_capacity(&client_context); Self::announce_child_capacity(&client_context);
} }
if !answer {
continue;
}
// Don't answer our own distributed search. // Don't answer our own distributed search.
if username == own_username { if username == own_username {
@@ -620,7 +643,7 @@ impl Client {
ctx.has_free_upload_slot(), ctx.has_free_upload_slot(),
ctx.last_upload_speed, ctx.last_upload_speed,
ctx.upload_queue.len() as u32, ctx.upload_queue.len() as u32,
&ctx.excluded_search_phrases(), &ctx.excluded_search_phrases,
), ),
Err(e) => { Err(e) => {
error!("[client] IncomingSearch read: {}", e); error!("[client] IncomingSearch read: {}", e);
@@ -786,17 +809,24 @@ impl Client {
&mut ctx, &branch, true, &mut ctx, &branch, true,
); );
} }
if ctx.leaf.is_parent(&parent, link) if ctx.leaf.is_parent(&parent, link) {
&& ctx.leaf.admit_search(Instant::now()) // The tree's stream is not ours to throttle: every
&& let Some(ops) = &ctx.operations // search goes down to our children. The budget
{ // below bounds only the share scan we do for our
let _ = ops.send(ClientOperation::IncomingSearch { // own reply.
ctx.children
.broadcast_search(&username, token, &query);
if let Some(ops) = &ctx.operations {
let _ =
ops.send(ClientOperation::IncomingSearch {
username, username,
token, token,
query, query,
from_parent: true,
}); });
} }
} }
}
ClientOperation::ParentClosed { parent, link } => { ClientOperation::ParentClosed { parent, link } => {
let Ok(mut ctx) = client_context.write_safe() else { let Ok(mut ctx) = client_context.write_safe() else {
continue; continue;
@@ -893,7 +923,14 @@ impl Client {
if !ctx.children.has_room() { if !ctx.children.has_room() {
continue; continue;
} }
if stream.set_nodelay(true).is_err() { // A child that stops reading must not wedge us: the
// relay writes under the context lock, so the write
// is bounded and a child that hits it loses its slot.
if stream.set_nodelay(true).is_err()
|| stream
.set_write_timeout(Some(CHILD_WRITE_TIMEOUT))
.is_err()
{
continue; continue;
} }
if username == own_username { if username == own_username {
+6 -2
View File
@@ -263,14 +263,18 @@ impl Client {
.send_to_peer(username, PeerMessage::SendMessage(request)); .send_to_peer(username, PeerMessage::SendMessage(request));
} }
} else { } else {
// Without a server there is nobody to resolve the peer's address,
// so the request would sit in the queue forever: say so instead.
let handle = self
.server_handle
.as_ref()
.ok_or(SoulseekRs::NotConnected)?;
self.context self.context
.write_safe()? .write_safe()?
.queue_peer_message(username, request); .queue_peer_message(username, request);
if let Some(handle) = &self.server_handle {
let _ = handle let _ = handle
.send(ServerMessage::GetPeerAddress(username.to_string())); .send(ServerMessage::GetPeerAddress(username.to_string()));
} }
}
Ok(()) Ok(())
} }
+1 -10
View File
@@ -59,19 +59,10 @@ impl Client {
fn send_search(&self, query: &str, wishlist: bool) -> Result<()> { fn send_search(&self, query: &str, wishlist: bool) -> Result<()> {
info!("Searching for {}", query); info!("Searching for {}", query);
let token = self.register_search(query)?;
let Some(handle) = &self.server_handle else { let Some(handle) = &self.server_handle else {
return Err(SoulseekRs::NotConnected); return Err(SoulseekRs::NotConnected);
}; };
let token = next_search_token();
self.context.write_safe()?.searches.insert(
query.to_string(),
Search {
token,
results: Vec::new(),
},
);
let query = query.to_string(); let query = query.to_string();
let _ = handle.send(if wishlist { let _ = handle.send(if wishlist {
ServerMessage::WishlistSearch { token, query } ServerMessage::WishlistSearch { token, query }
+20 -4
View File
@@ -301,7 +301,11 @@ impl Client {
/// # Errors /// # Errors
/// [`crate::SoulseekRs::NotConnected`] when there is no server connection. /// [`crate::SoulseekRs::NotConnected`] when there is no server connection.
pub fn add_interest(&self, item: &str) -> Result<()> { pub fn add_interest(&self, item: &str) -> Result<()> {
let item = self.context.write_safe()?.add_own_interest(item, true); let Some(item) =
self.context.write_safe()?.add_own_interest(item, true)
else {
return Ok(()); // nothing to like or unlike
};
self.send_server_message(MessageFactory::build_add_thing_i_like(&item)) self.send_server_message(MessageFactory::build_add_thing_i_like(&item))
} }
@@ -310,7 +314,11 @@ impl Client {
/// # Errors /// # Errors
/// [`crate::SoulseekRs::NotConnected`] when there is no server connection. /// [`crate::SoulseekRs::NotConnected`] when there is no server connection.
pub fn remove_interest(&self, item: &str) -> Result<()> { pub fn remove_interest(&self, item: &str) -> Result<()> {
let item = self.context.write_safe()?.remove_own_interest(item, true); let Some(item) =
self.context.write_safe()?.remove_own_interest(item, true)
else {
return Ok(()); // nothing to like or unlike
};
self.send_server_message(MessageFactory::build_remove_thing_i_like( self.send_server_message(MessageFactory::build_remove_thing_i_like(
&item, &item,
)) ))
@@ -321,7 +329,11 @@ impl Client {
/// # Errors /// # Errors
/// [`crate::SoulseekRs::NotConnected`] when there is no server connection. /// [`crate::SoulseekRs::NotConnected`] when there is no server connection.
pub fn add_dislike(&self, item: &str) -> Result<()> { pub fn add_dislike(&self, item: &str) -> Result<()> {
let item = self.context.write_safe()?.add_own_interest(item, false); let Some(item) =
self.context.write_safe()?.add_own_interest(item, false)
else {
return Ok(()); // nothing to like or unlike
};
self.send_server_message(MessageFactory::build_add_thing_i_hate(&item)) self.send_server_message(MessageFactory::build_add_thing_i_hate(&item))
} }
@@ -330,7 +342,11 @@ impl Client {
/// # Errors /// # Errors
/// [`crate::SoulseekRs::NotConnected`] when there is no server connection. /// [`crate::SoulseekRs::NotConnected`] when there is no server connection.
pub fn remove_dislike(&self, item: &str) -> Result<()> { pub fn remove_dislike(&self, item: &str) -> Result<()> {
let item = self.context.write_safe()?.remove_own_interest(item, false); let Some(item) =
self.context.write_safe()?.remove_own_interest(item, false)
else {
return Ok(()); // nothing to like or unlike
};
self.send_server_message(MessageFactory::build_remove_thing_i_hate( self.send_server_message(MessageFactory::build_remove_thing_i_hate(
&item, &item,
)) ))
+50 -4
View File
@@ -1015,7 +1015,10 @@ fn revoked_membership_drops_the_room_we_can_no_longer_see() {
} }
#[test] #[test]
fn losing_operatorship_keeps_the_membership_it_does_not_touch() { fn losing_operatorship_keeps_the_roster_we_can_still_see() {
// Demotion does not blind us: we are still a member, and the room's other
// operators are still ours to show. The server narrates our own removal
// from that roster separately (code 144).
let mut ctx = ClientContext::new(); let mut ctx = ClientContext::new();
ctx.apply_room_event(RoomEvent::PrivateMembers { ctx.apply_room_event(RoomEvent::PrivateMembers {
room: "club".into(), room: "club".into(),
@@ -1023,7 +1026,7 @@ fn losing_operatorship_keeps_the_membership_it_does_not_touch() {
}); });
ctx.apply_room_event(RoomEvent::PrivateOperators { ctx.apply_room_event(RoomEvent::PrivateOperators {
room: "club".into(), room: "club".into(),
users: vec!["alice".into()], users: vec!["alice".into(), "bob".into()],
}); });
ctx.apply_room_event(RoomEvent::OwnStandingChanged { ctx.apply_room_event(RoomEvent::OwnStandingChanged {
@@ -1032,7 +1035,23 @@ fn losing_operatorship_keeps_the_membership_it_does_not_touch() {
granted: false, granted: false,
}); });
assert_eq!(ctx.private_room_members("club"), vec!["alice".to_string()]); assert_eq!(ctx.private_room_members("club"), vec!["alice".to_string()]);
assert!(ctx.private_room_operators("club").is_empty()); assert_eq!(
ctx.private_room_operators("club"),
vec!["alice".to_string(), "bob".to_string()],
"the other operators are still there to show"
);
ctx.apply_room_event(RoomEvent::PrivateRosterChanged {
room: "club".into(),
username: "alice".into(),
members: false,
added: false,
});
assert_eq!(
ctx.private_room_operators("club"),
vec!["bob".to_string()],
"only the demotion the server narrates removes us"
);
} }
#[test] #[test]
@@ -1082,6 +1101,25 @@ fn a_file_the_server_excludes_is_left_out_of_a_reply() {
"an exclusion the path does not carry changes nothing" "an exclusion the path does not carry changes nothing"
); );
// The phrases are lowercased where they arrive, so a server that sends
// one capitalised still matches a path.
let mut ctx = ClientContext::new();
ctx.set_excluded_search_phrases(vec!["SPAM".to_string()]);
assert!(
build_search_response(
&shares,
"me",
1,
"xyzzy",
true,
0,
0,
&ctx.excluded_search_phrases(),
)
.is_none(),
"a capitalised phrase from the server still withholds the file"
);
let _ = std::fs::remove_dir_all(&dir); let _ = std::fs::remove_dir_all(&dir);
} }
@@ -1091,7 +1129,15 @@ fn our_own_interests_are_kept_lowercased_and_deduplicated() {
// the session ends, so they are held here in one spelling to be sent // the session ends, so they are held here in one spelling to be sent
// again next login. // again next login.
let mut ctx = ClientContext::new(); let mut ctx = ClientContext::new();
assert_eq!(ctx.add_own_interest(" Krautrock ", true), "krautrock"); assert_eq!(
ctx.add_own_interest(" Krautrock ", true).as_deref(),
Some("krautrock")
);
assert_eq!(
ctx.add_own_interest(" ", true),
None,
"an empty interest is not stored, and is not sent either"
);
ctx.add_own_interest("KRAUTROCK", true); ctx.add_own_interest("KRAUTROCK", true);
ctx.add_own_interest("Muzak", false); ctx.add_own_interest("Muzak", false);
@@ -57,55 +57,14 @@ impl MessageHandler<ServerMessage> for RoomOperatorsHandler {
} }
} }
/// One user added to (`added`) or removed from a private room's roster: /// One user added to or removed from a private room's roster, by the code it
/// members (134/135) or operators (143/144). /// arrived under: members added (134) or removed (135), operators added (143)
pub struct RoomRosterChangeHandler { /// or removed (144).
code: u32, pub struct RoomRosterChangeHandler(pub u32);
members: bool,
added: bool,
}
impl RoomRosterChangeHandler {
#[must_use]
pub const fn member_added() -> Self {
Self {
code: 134,
members: true,
added: true,
}
}
#[must_use]
pub const fn member_removed() -> Self {
Self {
code: 135,
members: true,
added: false,
}
}
#[must_use]
pub const fn operator_added() -> Self {
Self {
code: 143,
members: false,
added: true,
}
}
#[must_use]
pub const fn operator_removed() -> Self {
Self {
code: 144,
members: false,
added: false,
}
}
}
impl MessageHandler<ServerMessage> for RoomRosterChangeHandler { impl MessageHandler<ServerMessage> for RoomRosterChangeHandler {
fn get_code(&self) -> u32 { fn get_code(&self) -> u32 {
self.code self.0
} }
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) { fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
@@ -114,69 +73,27 @@ impl MessageHandler<ServerMessage> for RoomRosterChangeHandler {
let _ = sender.send(ServerMessage::PrivateRoomRosterChanged { let _ = sender.send(ServerMessage::PrivateRoomRosterChanged {
room, room,
username, username,
members: self.members, members: matches!(self.0, 134 | 135),
added: self.added, added: matches!(self.0, 134 | 143),
}); });
} }
} }
/// Our own standing in a private room changed: membership granted (139) or /// Our own standing in a private room changed: membership granted (139) or
/// revoked (140), operatorship granted (145) or revoked (146). /// revoked (140), operatorship granted (145) or revoked (146).
pub struct OwnRoomStandingHandler { pub struct OwnRoomStandingHandler(pub u32);
code: u32,
members: bool,
granted: bool,
}
impl OwnRoomStandingHandler {
#[must_use]
pub const fn membership_granted() -> Self {
Self {
code: 139,
members: true,
granted: true,
}
}
#[must_use]
pub const fn membership_revoked() -> Self {
Self {
code: 140,
members: true,
granted: false,
}
}
#[must_use]
pub const fn operatorship_granted() -> Self {
Self {
code: 145,
members: false,
granted: true,
}
}
#[must_use]
pub const fn operatorship_revoked() -> Self {
Self {
code: 146,
members: false,
granted: false,
}
}
}
impl MessageHandler<ServerMessage> for OwnRoomStandingHandler { impl MessageHandler<ServerMessage> for OwnRoomStandingHandler {
fn get_code(&self) -> u32 { fn get_code(&self) -> u32 {
self.code self.0
} }
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) { fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
let room = message.read_string(); let room = message.read_string();
let _ = sender.send(ServerMessage::OwnRoomStandingChanged { let _ = sender.send(ServerMessage::OwnRoomStandingChanged {
room, room,
members: self.members, members: matches!(self.0, 139 | 140),
granted: self.granted, granted: matches!(self.0, 139 | 145),
}); });
} }
} }
@@ -243,10 +160,10 @@ mod tests {
#[test] #[test]
fn each_roster_change_reports_which_roster_and_which_way() { fn each_roster_change_reports_which_roster_and_which_way() {
for (handler, members, added) in [ for (handler, members, added) in [
(RoomRosterChangeHandler::member_added(), true, true), (RoomRosterChangeHandler(134), true, true),
(RoomRosterChangeHandler::member_removed(), true, false), (RoomRosterChangeHandler(135), true, false),
(RoomRosterChangeHandler::operator_added(), false, true), (RoomRosterChangeHandler(143), false, true),
(RoomRosterChangeHandler::operator_removed(), false, false), (RoomRosterChangeHandler(144), false, false),
] { ] {
let (tx, rx) = std::sync::mpsc::channel(); let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| { let mut message = framed(|m| {
@@ -274,10 +191,10 @@ mod tests {
#[test] #[test]
fn our_own_standing_reports_the_room_and_the_change() { fn our_own_standing_reports_the_room_and_the_change() {
for (handler, members, granted) in [ for (handler, members, granted) in [
(OwnRoomStandingHandler::membership_granted(), true, true), (OwnRoomStandingHandler(139), true, true),
(OwnRoomStandingHandler::membership_revoked(), true, false), (OwnRoomStandingHandler(140), true, false),
(OwnRoomStandingHandler::operatorship_granted(), false, true), (OwnRoomStandingHandler(145), false, true),
(OwnRoomStandingHandler::operatorship_revoked(), false, false), (OwnRoomStandingHandler(146), false, false),
] { ] {
let (tx, rx) = std::sync::mpsc::channel(); let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| { let mut message = framed(|m| {
+6
View File
@@ -1,5 +1,6 @@
use std::io; use std::io;
use std::net::{TcpListener, TcpStream}; use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::Sender; use std::sync::mpsc::Sender;
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::thread; use std::thread;
@@ -422,6 +423,7 @@ impl Listen {
client_sender: Sender<ClientOperation>, client_sender: Sender<ClientOperation>,
client_context: Arc<RwLock<ClientContext>>, client_context: Arc<RwLock<ClientContext>>,
own_username: String, own_username: String,
stopped: Arc<AtomicBool>,
) { ) {
info!("[listener] listening on {:?}", listener.local_addr()); info!("[listener] listening on {:?}", listener.local_addr());
@@ -433,6 +435,10 @@ impl Listen {
let handshakes = Arc::new(Semaphore::new(MAX_HANDSHAKES)); let handshakes = Arc::new(Semaphore::new(MAX_HANDSHAKES));
for stream in listener.incoming() { for stream in listener.incoming() {
if stopped.load(Ordering::Relaxed) {
debug!("[listener] stopping, releasing the port");
return;
}
let Ok(stream) = stream else { let Ok(stream) = stream else {
error!( error!(
"[listener] Failed to accept connection: {}", "[listener] Failed to accept connection: {}",
+44
View File
@@ -5452,3 +5452,47 @@ fn privileges_can_be_handed_to_another_user() {
"the gifted privileges should show up as time of our own" "the gifted privileges should show up as time of our own"
); );
} }
#[test]
fn a_disconnected_client_gives_its_listener_port_back() {
// A client that goes out of scope must release the port it bound: the
// next one has to be able to advertise the port it was configured with,
// not fall back to an ephemeral one nobody was told about. No server is
// needed — binding the listener happens in connect(), before the dial.
let port = free_port().expect("a port to hold");
let settings = |port: u16| ClientSettings {
username: "e2e_port_release".to_string(),
password: "pw".to_string(),
server_address: PeerAddress::new("127.0.0.1".to_string(), 1),
enable_listen: true,
listen_port: port,
shared_directories: Vec::new(),
accept_children: false,
version: ClientVersion::default(),
};
let mut first = Client::with_settings(settings(port));
first.connect().expect("bind the listener");
assert_eq!(first.listen_port(), Some(port));
drop(first);
// The listener thread notices the stop the moment it is woken, which the
// disconnect does by dialling it; give it a beat to unwind.
let deadline = Instant::now() + Duration::from_secs(5);
let mut second_port = None;
while Instant::now() < deadline && second_port != Some(port) {
let mut second = Client::with_settings(settings(port));
second.connect().expect("bind again");
second_port = second.listen_port();
if second_port == Some(port) {
break;
}
drop(second);
std::thread::sleep(Duration::from_millis(200));
}
assert_eq!(
second_port,
Some(port),
"the port should be free again once the first client is dropped"
);
}