From f9034eedb0e49005d69977a6aaeeafb05076286d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 19:33:30 +0200 Subject: [PATCH] cultmesh-rs: a repeated Connect re-sends the pending Accept and queues nothing The document server called accept_connect again for a Connect on an existing key, queuing another reliable Accept each time: a resend storm, and the full queue Err that the previous commit turned into a drop. The repeat now re-sends the Accept still pending, or an Ack once acknowledged, as the hub does. A fresh session's accept_connect cannot fail under validated options, so its error is propagated again rather than counted as a drop. Co-Authored-By: Claude Opus 5.5 --- .../cultmesh-rs/src/rudp_document_server.rs | 77 ++++++++++--------- .../cultmesh-rs/tests/rudp_document_server.rs | 20 +++-- 2 files changed, 53 insertions(+), 44 deletions(-) diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index d5e0b473..524d17c7 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -240,8 +240,8 @@ where } /// Datagrams read and discarded because they belong to no admissible - /// session: malformed frames, unknown sessions, a Connect the session or - /// the session cap refuses, a packet that fails its session. Never an + /// session: malformed frames, unknown sessions, a Connect the session cap + /// refuses, a packet that fails its session. Never an /// error: a moved flow or a scanner must not end the daemon loop. pub fn packets_dropped(&self) -> u64 { self.packets_dropped @@ -402,50 +402,53 @@ where }) } - /// `Ok(false)` when the Connect was refused (session cap, or the session - /// cannot queue another Accept); only a socket failure is an `Err`. + /// `Ok(false)` when the Connect was refused because the session cap is full; + /// only a socket failure is an `Err`. fn accept_connection( &mut self, key: CultMeshRudpSessionKey, packet: &CultNetRudpPacket, now: u64, ) -> Result { - if !self.sessions.contains_key(&key) { - if self.sessions.len() >= self.options.max_sessions { - return Ok(false); + 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. + Some(entry) => { + entry.last_activity_monotonic_millis = now; + entry + .session + .pending_accept_for_resend(now) + .unwrap_or_else(|| entry.session.create_ack()) } - self.sessions.insert( - key, - SessionEntry { - session: CultNetRudpSession::new(CultNetRudpSessionOptions { - connection_id: key.connection_id, - initial_sequence: 1, - resend_delay_ms: duration_millis(self.options.resend_delay), - max_pending_reliable_packets: Some( - self.options.max_pending_reliable_packets_per_session, - ), - }), - created_at_monotonic_millis: now, - last_activity_monotonic_millis: now, - admitted_payload_bytes: 0, - }, - ); - } - - let accept = { - let entry = self - .sessions - .get_mut(&key) - .expect("accepted session must be present"); - // The same key is a retransmitted Connect for the existing epoch. - // A genuinely fresh client incarnation must choose a fresh id. - entry.last_activity_monotonic_millis = now; - match entry.session.accept_connect(packet, now, Vec::new()) { - Ok(accept) => accept, - Err(_) => return Ok(false), + None => { + if self.sessions.len() >= self.options.max_sessions { + return Ok(false); + } + let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: key.connection_id, + initial_sequence: 1, + resend_delay_ms: duration_millis(self.options.resend_delay), + max_pending_reliable_packets: Some( + self.options.max_pending_reliable_packets_per_session, + ), + }); + let accept = session.accept_connect(packet, now, Vec::new())?; + self.sessions.insert( + key, + SessionEntry { + session, + created_at_monotonic_millis: now, + last_activity_monotonic_millis: now, + admitted_payload_bytes: 0, + }, + ); + accept } }; - self.send_packet(key.remote_addr, &accept)?; + self.send_packet(key.remote_addr, &reply)?; Ok(true) } diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index e1e912e0..980bf04b 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -566,7 +566,7 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> { max_pending_reliable_packets_per_session: 4, ..Default::default() }; - let mut server = server(options, clock, sink.clone(), Source::default())?; + let mut server = server(options, clock.clone(), sink.clone(), Source::default())?; let target = server.local_addr()?; let mut real = client(target, 101)?; connect(&mut server, &mut [&mut real])?; @@ -602,16 +602,22 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> { send_raw(&mut server, &unadmitted.send("schema", vec![1], reliable.clone())?)?; assert_eq!(server.packets_dropped(), 2); - // A moved flow re-sends Connect from a socket that never hears the Accept: - // each one asks the session to queue one more unacknowledged Accept, and - // the fifth finds the queue full. + // A moved flow re-sends Connect from a socket that never hears the Accept. + // Each repeat re-sends the Accept still awaiting acknowledgement and queues + // nothing, so the storm neither errors nor grows the session's queue. let mut stormer = session(102); let connect_packet = stormer.create_connect(0, Vec::new())?; - for _ in 0..12 { + for _ in 0..200 { send_raw(&mut server, &connect_packet)?; } - assert_eq!(server.packets_dropped(), 2 + 8); + assert_eq!(server.packets_dropped(), 2); assert_eq!(server.session_count(), 2); + clock.set(60_000 + 1_000); + assert_eq!( + server.maintain()?.packets_resent, + 1, + "one Accept awaits acknowledgement, however many Connects repeated" + ); // A packet its own admitted session refuses ends that session, not the loop. stormer.assume_connected(0); @@ -619,7 +625,7 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> { poison.fragment_count = 2; poison.fragment_id = 0; send_raw(&mut server, &poison)?; - assert_eq!(server.packets_dropped(), 2 + 8 + 1); + assert_eq!(server.packets_dropped(), 3); assert_eq!(server.session_count(), 1); send(