cultnet-rs, cultmesh-rs: an ended session fails its flush; issued sequences survive a reset; the doc server says goodbye

A session that refused a packet, or was disconnected, forgot its
unacknowledged writes, so flush_reliable returned success for writes the peer
never acknowledged. The transport now fails the flush with the reason and
reads such writes as Invalidated. Reset no longer reissues sequence numbers or
fragment ids: a late duplicate Connect in server mode reused sequences the
peer had seen. end_refused_session owns the reset-then-goodbye ordering for
the hub, the transport and the document server, which used to end a refusing
session silently.

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

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -325,7 +325,7 @@ where
         let result = match result {
             Ok(result) => result,
             Err(_) => {
-                self.sessions.remove(&key);
+                self.end_refused_session(key)?;
                 self.packets_dropped += 1;
                 return Ok(CultMeshRudpPollOutcome::Handled);
             }
@@ -418,7 +418,7 @@ where
                 match entry.session.answer_repeated_connect(packet, now) {
                     Ok(reply) => reply,
                     Err(_) => {
-                        self.sessions.remove(&key);
+                        self.end_refused_session(key)?;
                         return Ok(false);
                     }
                 }
@@ -592,6 +592,16 @@ where
         Ok(None)
     }
 
+    /// The session refused a packet: it ends, and the client is told so it does
+    /// not wait on a session the server no longer holds.
+    fn end_refused_session(&mut self, key: CultMeshRudpSessionKey) -> Result<()> {
+        let Some(mut entry) = self.sessions.remove(&key) else {
+            return Ok(());
+        };
+        let goodbye = entry.session.end_refused_session();
+        self.send_packet(key.remote_addr, &goodbye)
+    }
+
     fn payload_budget_allows(&self, key: CultMeshRudpSessionKey, bytes: usize) -> bool {
         let Some(session_bytes) = self
             .sessions

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -627,6 +627,15 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> {
     send_raw(&mut server, &poison)?;
     assert_eq!(server.packets_dropped(), 3);
     assert_eq!(server.session_count(), 1);
+    // ...and the refused client is told, after the Accepts its storm drew.
+    raw.set_read_timeout(Some(Duration::from_millis(200)))?;
+    let mut buffer = vec![0_u8; 65_535];
+    let mut told = false;
+    while let Ok(received) = raw.recv(&mut buffer) {
+        told |= cultnet_rs::decode_rudp_packet(&buffer[..received])?.packet_type
+            == cultnet_rs::CultNetRudpPacketType::Disconnect;
+    }
+    assert!(told, "the refused client must be sent a goodbye");
 
     send(
         &mut real,

packages/cultnet-rs/src/rudp.rs

@@ -334,11 +334,12 @@ impl CultNetRudpSession {
         Ok(())
     }
 
+    /// Forgets everything learned from the peer. Sequence numbers and fragment
+    /// ids already issued stay issued: a peer that still remembers them must
+    /// never see one reused.
     pub fn reset_peer_state(&mut self) {
         self.session_scope = Uuid::new_v4();
-        self.next_sequence = self.initial_sequence;
         self.next_sequenced_by_channel.clear();
-        self.next_fragment_id = 1;
         self.connected = false;
         self.last_received_at_ms = None;
         self.highest_received_sequence = None;
@@ -351,6 +352,16 @@ impl CultNetRudpSession {
         self.fragment_buffers.clear();
     }
 
+    /// Ends a session that refused a packet. `receive` has already recorded the
+    /// refused packet's reliable sequence, so the session cannot be kept: a
+    /// retransmit would be acknowledged and the frame silently lost. Returns the
+    /// goodbye for the peer, built after the reset or its ack field would
+    /// acknowledge the very frame that was refused.
+    pub fn end_refused_session(&mut self) -> CultNetRudpPacket {
+        self.reset_peer_state();
+        self.create_disconnect(RUDP_REFUSED_PACKET_REASON.to_vec())
+    }
+
     pub fn create_connect(&mut self, now_ms: u64, payload: Vec<u8>) -> Result<CultNetRudpPacket> {
         self.ensure_reliable_capacity(1)?;
         let packet = self.create_packet(
@@ -1729,15 +1740,11 @@ impl CultNetRudpServerHub {
         let Some(mut peer) = self.peers.remove(&remote_addr) else {
             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());
+        let goodbye = peer.session.end_refused_session();
         self.pending_events
             .push_back(CultNetRudpServerEvent::Disconnected {
                 session: peer.context,
-                reason,
+                reason: goodbye.payload.clone(),
             });
         self.send_packet(remote_addr, &goodbye)
     }
@@ -1861,6 +1868,7 @@ impl CultNetRudpSocketTransportConnection {
             ));
         }
         let packet = self.session.create_connect(now_ms(), payload)?;
+        self.disconnect_reason = None;
         self.send_packet(&packet)
     }
 
@@ -1914,7 +1922,12 @@ impl CultNetRudpSocketTransportConnection {
         &self,
         receipt: &CultNetRudpReliableSendReceipt,
     ) -> CultNetRudpReliableSendStatus {
-        self.session.reliable_send_status(receipt)
+        let status = self.session.reliable_send_status(receipt);
+        // A write still unacknowledged when the session ended never will be.
+        if status == CultNetRudpReliableSendStatus::Pending && self.disconnect_reason.is_some() {
+            return CultNetRudpReliableSendStatus::Invalidated;
+        }
+        status
     }
 
     fn send_data_packets(&mut self, packets: &[CultNetRudpPacket]) -> Result<()> {
@@ -1981,7 +1994,18 @@ impl CultNetRudpSocketTransportConnection {
         let deadline = Instant::now() + timeout;
         let mut preserved_frames = VecDeque::new();
         let result = (|| {
-            while self.outstanding_reliable_packet_count() > 0 {
+            loop {
+                // An ended session forgot its unacknowledged writes; reporting
+                // them flushed would be a lie.
+                if let Some(reason) = &self.disconnect_reason {
+                    return Err(anyhow!(
+                        "RUDP session ended before its reliable writes were acknowledged: {}",
+                        String::from_utf8_lossy(reason)
+                    ));
+                }
+                if self.outstanding_reliable_packet_count() == 0 {
+                    return Ok(());
+                }
                 if Instant::now() >= deadline {
                     return Err(anyhow!(
                         "RUDP reliable flush timed out with {} packets outstanding",
@@ -1993,7 +2017,6 @@ impl CultNetRudpSocketTransportConnection {
                 }
                 self.poll_resends()?;
             }
-            Ok(())
         })();
 
         preserved_frames.append(&mut self.delivered_frames);
@@ -2076,6 +2099,7 @@ impl CultNetRudpSocketTransportConnection {
             && packet.packet_type == CultNetRudpPacketType::Connect
         {
             self.session.reset_peer_state();
+            self.disconnect_reason = None;
             let Ok(accept) = self.session.accept_connect(&packet, now_ms(), Vec::new()) else {
                 self.stats.packets_dropped += 1;
                 return Ok(true);
@@ -2085,17 +2109,11 @@ impl CultNetRudpSocketTransportConnection {
         }
 
         let Ok(result) = self.session.receive(&packet, now_ms()) else {
-            // `receive` has already recorded the packet's reliable sequence, so
-            // the session cannot be kept: a retransmit would be acknowledged
-            // and the frame silently lost. End it and tell the peer.
+            // End the session and tell the peer. A server-mode transport also
+            // forgets its peer: only a Connect can claim it again.
             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);
+            let goodbye = self.session.end_refused_session();
+            self.disconnect_reason = Some(goodbye.payload.clone());
             let sent = self.send_packet(&goodbye);
             if self.mode == CultNetRudpSocketMode::Server {
                 self.remote_addr = None;

packages/cultnet-rs/tests/rudp_stray_packets.rs

@@ -467,3 +467,118 @@ fn constructors_reject_limits_that_make_a_connect_unadmittable() -> Result<()> {
     }
     Ok(())
 }
+
+/// The distinct reliable Accept sequences a peer has been sent so far.
+fn accept_sequences(peer: &UdpSocket) -> Result<std::collections::BTreeSet<u32>> {
+    let mut buffer = vec![0_u8; 65_535];
+    let mut sequences = std::collections::BTreeSet::new();
+    while let Ok((received, _)) = peer.recv_from(&mut buffer) {
+        let packet = decode_rudp_packet(&buffer[..received])?;
+        if packet.packet_type == CultNetRudpPacketType::Accept && packet.reliable {
+            sequences.insert(packet.sequence);
+        }
+    }
+    Ok(sequences)
+}
+
+/// A Connect that repeats (a moved flow, a retransmit, a storm) is answered
+/// with the Accept already owed, never a fresh reliable Accept per Connect.
+#[test]
+fn hub_owes_one_reliable_accept_however_many_connects_repeat() -> Result<()> {
+    let mut hub =
+        CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID))?;
+    let hub_addr = hub.local_addr()?;
+    let peer = socket()?;
+    let mut peer_session = raw_session(CONNECTION_ID);
+    let connect = peer_session.create_connect(0, b"peer".to_vec())?;
+    for _ in 0..20 {
+        send_to(&peer, hub_addr, &connect)?;
+        while hub.receive_event_once()?.is_some() {}
+    }
+    assert_eq!(accept_sequences(&peer)?.len(), 1);
+    Ok(())
+}
+
+#[test]
+fn server_mode_owes_one_reliable_accept_however_many_connects_repeat() -> Result<()> {
+    let server_socket = socket()?;
+    let server_addr = server_socket.local_addr()?;
+    let mut server = CultNetRudpSocketTransportConnection::new(
+        CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID),
+    )?;
+    let peer = socket()?;
+    let mut peer_session = raw_session(CONNECTION_ID);
+    let connect = peer_session.create_connect(0, b"peer".to_vec())?;
+    for _ in 0..20 {
+        send_to(&peer, server_addr, &connect)?;
+        let _ = server.receive_once()?;
+    }
+    assert!(server.outstanding_reliable_packet_count() <= 1);
+    Ok(())
+}
+
+/// A late duplicate Connect resets what the server learned of the peer, never
+/// what it already issued: a reused sequence is dropped by the peer as a
+/// duplicate.
+#[test]
+fn server_mode_never_reuses_a_sequence_after_a_late_duplicate_connect() -> Result<()> {
+    let server_socket = socket()?;
+    let server_addr = server_socket.local_addr()?;
+    let mut server = CultNetRudpSocketTransportConnection::new(
+        CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID),
+    )?;
+    let peer = socket()?;
+    let mut peer_session = raw_session(CONNECTION_ID);
+    let connect = peer_session.create_connect(0, b"peer".to_vec())?;
+    send_to(&peer, server_addr, &connect)?;
+    let _ = server.receive_once()?;
+    server.send_reliable("schema", b"before".to_vec())?;
+    send_to(&peer, server_addr, &connect)?;
+    let _ = server.receive_once()?;
+    server.send_reliable("schema", b"after".to_vec())?;
+
+    let mut buffer = vec![0_u8; 65_535];
+    let mut delivered = Vec::new();
+    while let Ok((received, _)) = peer.recv_from(&mut buffer) {
+        let result = peer_session.receive(&decode_rudp_packet(&buffer[..received])?, 0)?;
+        delivered.extend(result.delivered.into_iter().map(|frame| frame.payload));
+    }
+    assert!(delivered.contains(&b"before".to_vec()));
+    assert!(
+        delivered.contains(&b"after".to_vec()),
+        "the frame sent after the duplicate Connect was dropped as a reused sequence"
+    );
+    Ok(())
+}
+
+/// A session that ended forgot its unacknowledged writes; a flush that then
+/// reports success would claim writes the peer never acknowledged.
+#[test]
+fn a_write_unacknowledged_when_the_session_is_refused_fails_its_flush() -> Result<()> {
+    let server = socket()?;
+    let server_addr = server.local_addr()?;
+    let mut client = CultNetRudpSocketTransportConnection::new(
+        CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID),
+    )?;
+    client.connect(b"hello".to_vec())?;
+    let mut buffer = vec![0_u8; 65_535];
+    let (received, client_addr) = server.recv_from(&mut buffer)?;
+    let mut server_session = raw_session(CONNECTION_ID);
+    let accept =
+        server_session.accept_connect(&decode_rudp_packet(&buffer[..received])?, 0, Vec::new())?;
+    send_to(&server, client_addr, &accept)?;
+    let _ = client.receive_once()?;
+
+    let receipt = client.send_reliable("schema", b"never acknowledged".to_vec())?;
+    send_to(&server, client_addr, &poisoned(&mut server_session)?)?;
+    let _ = client.receive_once()?;
+    assert!(client.disconnect_reason().is_some());
+
+    let flushed = client.flush_reliable(Duration::from_millis(200));
+    assert!(flushed.is_err(), "the write was never acknowledged");
+    assert_eq!(
+        client.reliable_send_status(&receipt),
+        CultNetRudpReliableSendStatus::Invalidated
+    );
+    Ok(())
+}