cultnet-rs: repeated Connect has one primitive; the goodbye must not ack the refused frame

The Disconnect a refusing session sent carried the ack of the very frame it
refused, so the sender saw the frame acknowledged. The session forgets its
peer's sequences before building the goodbye, and a server-mode transport
forgets its endpoint so only a Connect can claim it again.

CultNetRudpSession::answer_repeated_connect is the one answer to a repeated
Connect for the hub and the document server: it records the sequence (the
document server's old accept_connect did, and a client that reconnects in one
session depends on it), applies acks, queues nothing, and returns the pending
Accept or an Ack.

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

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -411,17 +411,17 @@ where
         now: u64,
     ) -> Result<bool> {
         let reply = match self.sessions.get_mut(&key) {
-            // The same key is a retransmitted Connect for the existing epoch:
-            // repeat the Accept still awaiting acknowledgement, or acknowledge
-            // if none is. It queues nothing, so a Connect storm cannot grow the
-            // session's reliable queue. A genuinely fresh client incarnation
-            // must choose a fresh id.
+            // The same key is a retransmitted Connect for the existing epoch.
+            // A genuinely fresh client incarnation must choose a fresh id.
             Some(entry) => {
                 entry.last_activity_monotonic_millis = now;
-                entry
-                    .session
-                    .pending_accept_for_resend(now)
-                    .unwrap_or_else(|| entry.session.create_ack())
+                match entry.session.answer_repeated_connect(packet, now) {
+                    Ok(reply) => reply,
+                    Err(_) => {
+                        self.sessions.remove(&key);
+                        return Ok(false);
+                    }
+                }
             }
             None => {
                 if self.sessions.len() >= self.options.max_sessions {

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -581,7 +581,7 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> {
             max_pending_reliable_packets: None,
         })
     };
-    let mut send_raw = |server: &mut Server, packet: &cultnet_rs::CultNetRudpPacket| -> Result<()> {
+    let send_raw = |server: &mut Server, packet: &cultnet_rs::CultNetRudpPacket| -> Result<()> {
         raw.send_to(&encode_rudp_packet(packet)?, target)?;
         server.poll_once()?;
         Ok(())

packages/cultnet-rs/src/rudp.rs

@@ -271,9 +271,7 @@ impl CultNetRudpSession {
         self.pending_reliable.len() + self.queued_reliable.len()
     }
 
-    /// The unacknowledged Accept, stamped as just sent, for a peer that
-    /// repeats its Connect. `None` once the peer has acknowledged it.
-    pub fn pending_accept_for_resend(&mut self, now_ms: u64) -> Option<CultNetRudpPacket> {
+    fn pending_accept_for_resend(&mut self, now_ms: u64) -> Option<CultNetRudpPacket> {
         let pending = self
             .pending_reliable
             .values_mut()
@@ -397,6 +395,30 @@ impl CultNetRudpSession {
         Ok(response)
     }
 
+    /// Answers a Connect from a peer this session has already accepted. The
+    /// packet's sequence is remembered and nothing is queued, so a Connect storm
+    /// cannot grow the reliable queue. The reply is the Accept still awaiting
+    /// acknowledgement, or an Ack once it was acknowledged.
+    pub fn answer_repeated_connect(
+        &mut self,
+        packet: &CultNetRudpPacket,
+        now_ms: u64,
+    ) -> Result<CultNetRudpPacket> {
+        self.require_connection(packet)?;
+        if packet.packet_type != CultNetRudpPacketType::Connect {
+            return Err(anyhow!(
+                "Expected RUDP connect packet, got {:?}",
+                packet.packet_type
+            ));
+        }
+        self.apply_acknowledgements(packet);
+        self.remember_received(packet.sequence);
+        self.last_received_at_ms = Some(now_ms);
+        Ok(self
+            .pending_accept_for_resend(now_ms)
+            .unwrap_or_else(|| self.create_ack()))
+    }
+
     pub fn send(
         &mut self,
         channel_id: &str,
@@ -1594,14 +1616,10 @@ impl CultNetRudpServerHub {
             if let Some(peer) = self.peers.get_mut(&remote_addr)
                 && peer.context.connect_payload == packet.payload
             {
-                if peer.session.receive(&packet, now_ms()).is_err() {
+                let Ok(reply) = peer.session.answer_repeated_connect(&packet, now_ms()) else {
                     self.end_refusing_peer(remote_addr)?;
                     return Ok(true);
-                }
-                let reply = peer
-                    .session
-                    .pending_accept_for_resend(now_ms())
-                    .unwrap_or_else(|| peer.session.create_ack());
+                };
                 self.send_packet(remote_addr, &reply)?;
                 return Ok(true);
             }
@@ -1712,6 +1730,9 @@ impl CultNetRudpServerHub {
             return Ok(());
         };
         let reason = RUDP_REFUSED_PACKET_REASON.to_vec();
+        // Forget the peer's sequences first: the goodbye's ack field would
+        // otherwise acknowledge the very frame the session refused.
+        peer.session.reset_peer_state();
         let goodbye = peer.session.create_disconnect(reason.clone());
         self.pending_events
             .push_back(CultNetRudpServerEvent::Disconnected {
@@ -2068,10 +2089,18 @@ impl CultNetRudpSocketTransportConnection {
             // the session cannot be kept: a retransmit would be acknowledged
             // and the frame silently lost. End it and tell the peer.
             self.stats.packets_dropped += 1;
+            // The goodbye is built after the reset, or its ack field would
+            // acknowledge the very frame the session refused. A server-mode
+            // transport also forgets its peer: only a Connect can claim it again.
             let reason = RUDP_REFUSED_PACKET_REASON.to_vec();
+            self.session.reset_peer_state();
             let goodbye = self.session.create_disconnect(reason.clone());
             self.disconnect_reason = Some(reason);
-            self.send_packet(&goodbye)?;
+            let sent = self.send_packet(&goodbye);
+            if self.mode == CultNetRudpSocketMode::Server {
+                self.remote_addr = None;
+            }
+            sent?;
             return Ok(false);
         };
         if let Some(reply) = result.reply {

packages/cultnet-rs/tests/rudp_stray_packets.rs

@@ -388,9 +388,12 @@ fn client_ends_its_session_when_it_refuses_a_packet() -> Result<()> {
     assert!(client.disconnect_reason().is_some());
 
     // The server is told, not left holding a live session.
-    let (received, _) = server.recv_from(&mut buffer)?;
-    let told = decode_rudp_packet(&buffer[..received])?;
-    assert_eq!(told.packet_type, CultNetRudpPacketType::Disconnect);
+    let mut told = false;
+    while let Ok((received, _)) = server.recv_from(&mut buffer) {
+        told |= decode_rudp_packet(&buffer[..received])?.packet_type
+            == CultNetRudpPacketType::Disconnect;
+    }
+    assert!(told, "the server must be told the session ended");
     Ok(())
 }