Shut kicked and removed connections when their close can't be queued

Fixes #1070

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

crates/collab-net/src/barred_tests.rs

@@ -51,3 +51,47 @@ fn a_closed_file_forgets_whom_it_barred() {
     assert!(!hub.barred.contains_key("f"), "and its barred addresses went with it");
     assert!(hub.barred.contains_key("other"), "another file's stay");
 }
+
+#[test]
+fn a_kicked_connection_with_a_full_queue_is_shut() {
+    use std::io::Read;
+    use std::net::TcpListener;
+    use std::sync::Arc;
+    use std::sync::atomic::AtomicUsize;
+    use std::sync::mpsc::sync_channel;
+    use std::time::Duration;
+
+    use super::{Conn, Kind, Line, Out};
+
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let mut client = std::net::TcpStream::connect(listener.local_addr().unwrap()).unwrap();
+    let (server_side, _) = listener.accept().unwrap();
+    client.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+    // Stands in for the writer thread's handle: it keeps the socket open after the Conn is gone.
+    let _writer = server_side.try_clone().unwrap();
+    let (tx, _rx) = sync_channel(2);
+    while tx.try_send(Out::Bytes(Default::default(), None)).is_ok() {}
+    let out = Line {
+        tx,
+        queued: Arc::default(),
+        block: Arc::new(AtomicUsize::new(0)),
+        all: Arc::new(AtomicUsize::new(0)),
+        max: 1 << 20,
+        max_block: usize::MAX,
+        max_all: usize::MAX,
+    };
+    let ip = IpAddr::V4(Ipv4Addr::LOCALHOST);
+    let mut hub = Hub {
+        in_flight: InFlight(Default::default()),
+        backend: Closes,
+        rooms: HashMap::new(),
+        conns: HashMap::from([(1, Conn { out, sock: server_side, kind: Kind::Managing, ip })]),
+        local: None,
+        outbox: Vec::new(),
+        barred: HashMap::new(),
+    };
+    hub.kick(&[1], "kicked");
+    assert!(hub.conns.is_empty());
+    let mut buf = [0u8; 1];
+    assert_eq!(client.read(&mut buf).ok(), Some(0), "the kicked socket must be shut (client sees EOF)");
+}

crates/collab-net/src/hub.rs

@@ -811,10 +811,15 @@ fn encode(msg: &ServerMsg) -> Option<Bytes> {
 
 /// Queue `Refused {reason}` and a close for one connection (small: not counted against its caps).
 fn refuse(out: &Line, why: &str) {
+    refused(out, why);
+}
+
+/// [`refuse`], and whether the close was queued (it isn't when the queue is full).
+fn refused(out: &Line, why: &str) -> bool {
     if let Some(b) = encode(&ServerMsg::Refused { reason: why.to_string() }) {
         let _ = out.tx.try_send(Out::Bytes(b, None));
     }
-    let _ = out.tx.try_send(Out::Close(CloseCode::Policy, reason(why)));
+    out.tx.try_send(Out::Close(CloseCode::Policy, reason(why))).is_ok()
 }
 
 /// What the server tells a client when it is too busy to take more from it.
@@ -1429,7 +1434,9 @@ impl<B: Backend> Hub<B> {
             barred.insert(c.ip);
         }
         log::info!("event=removed file={file} client={id} ip={}", c.ip);
-        let _ = c.out.tx.try_send(Out::Close(CloseCode::Away, REMOVED.into()));
+        if c.out.tx.try_send(Out::Close(CloseCode::Away, REMOVED.into())).is_err() {
+            shutdown(&c.sock);
+        }
         self.gone(conn);
         Ok(())
     }
@@ -1543,8 +1550,10 @@ impl<B: Backend> Hub<B> {
             // People in the file are told the session is over for them, as when a host stops (they
             // keep their copy; nothing retries); a library request is refused.
             for conn in &people {
-                if let Some(c) = self.conns.get(conn) {
+                if let Some(c) = self.conns.get(conn)
-                    let _ = c.out.tx.try_send(Out::Close(CloseCode::Away, STOPPED.into()));
+                    && c.out.tx.try_send(Out::Close(CloseCode::Away, STOPPED.into())).is_err()
+                {
+                    shutdown(&c.sock);
                 }
                 self.gone(*conn);
             }
@@ -1593,8 +1602,10 @@ impl<B: Backend> Hub<B> {
     /// Refuse connections with `why` and drop them at once (nothing more they send counts).
     fn kick(&mut self, conns: &[u64], why: &str) {
         for conn in conns {
-            if let Some(c) = self.conns.get(conn) {
+            if let Some(c) = self.conns.get(conn)
-                refuse(&c.out, why);
+                && !refused(&c.out, why)
+            {
+                shutdown(&c.sock);
             }
             self.gone(*conn);
         }