GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultmesh-rs: a repeated Connect re-sends the pending Accept and queues nothing

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 19:33:30 +02:00
commit f9034eedb0
2 changed files with 53 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

77
packages/cultmesh-rs/src/rudp_document_server.rs
View file

@ -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<bool> {
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)
}

20
packages/cultmesh-rs/tests/rudp_document_server.rs
View file

@ -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(