From 03141bc93bad3e2e1269edb8092224f383773169 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:04:22 +0000 Subject: [PATCH 1/2] cultmesh-rs: send the capped frame's answer reliably and end the session once it is acknowledged The capped (8th) refusal went out unreliable and unordered with the goodbye at once, so a lost datagram left the peer without it. It is now queued like any response and resent until the peer acknowledges it; the session ends then (or after a 2 s grace) and serves nothing after the capped frame meanwhile. The answer, its resends and the poll that carried the capped frame acknowledge nothing, as every refusal. A reliable queue that cannot take the answer ends the session at once with the unreliable refusal, as before. Co-Authored-By: Claude Sonnet 5.5 --- .../cultmesh-rs/src/rudp_document_server.rs | 101 ++++++++++++++- .../cultmesh-rs/tests/rudp_document_server.rs | 120 +++++++++++++++++- 2 files changed, 210 insertions(+), 11 deletions(-) diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 1553168b..30f76273 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -25,6 +25,10 @@ const SHORTEST_MESSAGE_ID: &str = "0"; /// is answered too, then the session ends. The same number is the cap in every runtime. pub const MAX_CONSECUTIVE_REFUSED_FRAMES: u32 = 8; +/// How long a session whose capped frame was answered waits for the peer to acknowledge the answer +/// before it ends anyway. +const CAPPED_ANSWER_GRACE_MILLIS: u64 = 2_000; + /// The transport identity of one remote CultNet RUDP session. /// /// Connection identifiers are scoped to a remote socket address. Two clients @@ -424,6 +428,9 @@ struct SessionEntry { admitted_payload_bytes: usize, undecodable_frames_in_a_row: u32, refused_deletes_in_a_row: u32, + /// Set once the peer's capped frame has been answered: the session ends when the answer is acknowledged, + /// or at this time. + ending_deadline_monotonic_millis: Option, } /// A synchronous, multi-session CultNet RUDP raw-document server. @@ -612,6 +619,14 @@ where if frame.channel_id != "schema" { continue; } + // The capped frame was answered: nothing after it is served while the answer is delivered. + if self + .sessions + .get(&key) + .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some()) + { + break; + } let decoded = decode_cultnet_message_from_slice( &frame.payload, CultNetWireContract::CultNetSchemaV0, @@ -664,7 +679,7 @@ where let refusal = rejection.refusal_message(); let packets = match self.sessions.remove(&key) { Some(mut entry) => { - let packets = self.refusal_packets(&mut entry.session, &refusal); + let packets = self.refusal_packets(&mut entry.session, &refusal, false); self.sessions.insert(key, entry); packets } @@ -679,6 +694,45 @@ where kept_rejection = Some(rejection); continue; } + // The capped frame is answered reliably, so a lost datagram is resent; the session ends once + // the peer has acknowledged the answer or the grace runs out (`end_answered_sessions`). When + // the answer cannot be queued, the session ends at once with the refusal as before. + if (is_unreadable || is_delete) + && run.is_some() + && !matches!( + rejection.reason, + CultMeshRudpRejectionReason::ResponseSendFailed(_) + ) + { + let refusal = rejection.refusal_message(); + let packets = match self.sessions.remove(&key) { + Some(mut entry) => { + let packets = self.refusal_packets(&mut entry.session, &refusal, true); + if !packets.is_empty() { + entry.ending_deadline_monotonic_millis = + Some(now_monotonic + CAPPED_ANSWER_GRACE_MILLIS); + } + self.sessions.insert(key, entry); + packets + } + None => Vec::new(), + }; + if packets.is_empty() { + self.end_session(key, SessionEnd::Rejected(refusal))?; + return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)); + } + for mut packet in packets { + // Acknowledges nothing, as every refusal: the refused frame was not admitted. + packet.ack = 0; + packet.ack_mask = 0; + if let Some(error) = self.send_packet(key.remote_addr, &packet)? { + unsendable = Some(error); + break; + } + } + kept_rejection = Some(rejection); + break; + } // The only reply `receive` returns is a Pong, which delivers no // frame, so no earlier send of this poll can precede a rejection. // A response that could not be sent ended its session where the @@ -693,7 +747,11 @@ where } } - if packet.reliable && unsendable.is_none() { + let answering_capped_frame = self + .sessions + .get(&key) + .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some()); + if packet.reliable && unsendable.is_none() && !answering_capped_frame { let ack = self .sessions .get_mut(&key) @@ -705,6 +763,7 @@ where if let Some(error) = unsendable { self.end_session(key, SessionEnd::Unsendable(error))?; } + self.end_answered_sessions()?; Ok(match kept_rejection { Some(rejection) => CultMeshRudpPollOutcome::ApplicationRejected(rejection), None => CultMeshRudpPollOutcome::Handled, @@ -722,10 +781,16 @@ where && now.saturating_sub(entry.created_at_monotonic_millis) <= lifetime_ms }); let expired = before - self.sessions.len(); + self.end_answered_sessions()?; let mut resends = Vec::new(); for (key, entry) in &mut self.sessions { - for packet in entry.session.due_resends(now) { + for mut packet in entry.session.due_resends(now) { + if entry.ending_deadline_monotonic_millis.is_some() { + // The resent answer to a capped frame acknowledges nothing either. + packet.ack = 0; + packet.ack_mask = 0; + } resends.push((*key, packet)); } } @@ -744,6 +809,26 @@ where }) } + /// Ends each session whose capped frame was answered once the peer has acknowledged every answer (a lost + /// answer is resent first), or when the grace has run out. + fn end_answered_sessions(&mut self) -> Result<()> { + let now = self.clock.now_monotonic_millis(); + let due: Vec = self + .sessions + .iter() + .filter(|(_, entry)| { + entry.ending_deadline_monotonic_millis.is_some_and(|deadline| { + entry.session.outstanding_reliable_packet_count() == 0 || now >= deadline + }) + }) + .map(|(key, _)| *key) + .collect(); + for key in due { + self.end_session(key, SessionEnd::Refused)?; + } + Ok(()) + } + /// `Ok(false)` when the Connect was refused because the session cap is full; /// only a socket failure is an `Err`. fn accept_connection( @@ -795,6 +880,7 @@ where admitted_payload_bytes: 0, undecodable_frames_in_a_row: 0, refused_deletes_in_a_row: 0, + ending_deadline_monotonic_millis: None, }, ); accept @@ -1034,7 +1120,7 @@ where }; let (refusal, mut unsendable) = match end { SessionEnd::Refused => (Vec::new(), None), - SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal), None), + SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal, false), None), SessionEnd::Unsendable(error) => (Vec::new(), Some(error)), }; entry.session.reset_peer_state(); @@ -1057,11 +1143,12 @@ where Ok(()) } - /// The refusal a rejected peer is sent, as unreliable, unordered packets. + /// The refusal a rejected peer is sent: unreliable and unordered, or (`reliable`) queued like any response. fn refusal_packets( &self, session: &mut CultNetRudpSession, refusal: &CultNetMessage, + reliable: bool, ) -> Vec { encode_cultnet_message_to_vec(refusal, CultNetWireContract::CultNetSchemaV0) .and_then(|payload| { @@ -1069,8 +1156,8 @@ where "schema", payload, CultNetRudpSendOptions { - reliable: false, - ordered: false, + reliable, + ordered: reliable, sequenced: false, now_ms: self.clock.now_monotonic_millis(), reliable_expire_after_ms: None, diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 466037c7..a1954955 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -343,14 +343,17 @@ fn refusal_seen_by( client: &mut CultNetRudpSocketTransportConnection, receipt: &CultNetRudpReliableSendReceipt, ) -> Result { - let (error, code, details) = typed_refusal_seen_by(client, receipt)?; + let (error, code, details) = typed_refusal_seen_by(client, receipt, None)?; assert_eq!((code, details), (None, None)); Ok(error) } +/// `server` is polled while the goodbye is awaited when the refusal ends its session only once the peer has +/// acknowledged it (the capped frame's answer). fn typed_refusal_seen_by( client: &mut CultNetRudpSocketTransportConnection, receipt: &CultNetRudpReliableSendReceipt, + mut server: Option<&mut Server>, ) -> Result<(String, Option, Option)> { let mut refusal = None; for _ in 0..2_000 { @@ -381,6 +384,9 @@ fn typed_refusal_seen_by( "a refusal must not acknowledge the refused message" ); for _ in 0..2_000 { + if let Some(server) = server.as_deref_mut() { + server.poll_once()?; + } client.receive_once()?; if !client.connected() { break; @@ -681,7 +687,7 @@ fn application_rejection_is_nonfatal_peer_scoped_refused_to_the_peer_and_unackno // The caller keeps the sink's text; the peer is sent fixed text, never it. // A refused put is typed write_not_authorized and names the record; nothing of it reached the sink. assert_eq!( - typed_refusal_seen_by(&mut publisher, &publish_receipt)?, + typed_refusal_seen_by(&mut publisher, &publish_receipt, None)?, ( "write_not_authorized: the catalog refused the document".to_string(), Some(CultNetErrorCode::WriteNotAuthorized), @@ -863,7 +869,7 @@ fn an_unreadable_frame_is_refused_typed_and_the_session_kept_until_eight_in_a_ro panic!("the eighth consecutive unreadable frame must be refused"); }; assert_eq!(rejection.operation, CultMeshRudpApplicationOperation::Unreadable); - let (error, code, _) = typed_refusal_seen_by(&mut garbler, &receipt)?; + let (error, code, _) = typed_refusal_seen_by(&mut garbler, &receipt, Some(&mut server))?; assert_eq!((error.as_str(), code), (message_invalid.2.as_str(), message_invalid.0)); drain_until_idle(&mut server)?; assert_eq!(server.session_count(), 1, "only the capped peer's session ended"); @@ -905,7 +911,7 @@ fn a_refused_delete_keeps_the_session_until_eight_in_a_row() -> Result<()> { }; assert_eq!(rejection.operation, CultMeshRudpApplicationOperation::DocumentDelete); assert_eq!(rejection.reason, CultMeshRudpRejectionReason::DeleteNotSupported); - let (error, code, details) = typed_refusal_seen_by(&mut deleter, &receipt)?; + let (error, code, details) = typed_refusal_seen_by(&mut deleter, &receipt, Some(&mut server))?; assert_eq!( (error, code, details), { @@ -1623,6 +1629,11 @@ impl RawPeer { /// Sends `message` reliably and in order; returns its sequences. fn send(&mut self, message: &CultNetMessage) -> Result> { let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; + self.send_payload(payload) + } + + /// Sends `payload` as one reliable, ordered schema frame; returns its sequences. + fn send_payload(&mut self, payload: Vec) -> Result> { let options = cultnet_rs::CultNetRudpSendOptions { reliable: true, ordered: true, @@ -1910,3 +1921,104 @@ fn a_response_that_is_too_large_to_send_is_named_emsgsize() -> Result<()> { ); Ok(()) } + +/// A peer on `server` that has sent seven unreadable frames, each answered and processed, and then the +/// eighth, whose answer is waiting at its socket. Returns the peer and the eighth frame's sequence. +fn peer_at_the_cap(server: &mut Server, connection_id: u32) -> Result<(RawPeer, u32)> { + let mut peer = RawPeer::connect(server, connection_id)?; + for _ in 0..7 { + peer.send_payload(garbage())?; + assert!(matches!( + server.poll_once()?, + CultMeshRudpPollOutcome::ApplicationRejected(_) + )); + for packet in peer.drain() { + peer.session.receive(&packet, 3)?; + } + } + let capped = peer.send_payload(garbage())?[0]; + assert!(matches!( + server.poll_once()?, + CultMeshRudpPollOutcome::ApplicationRejected(_) + )); + Ok((peer, capped)) +} + +fn is_goodbye(packet: &cultnet_rs::CultNetRudpPacket) -> bool { + packet.packet_type == cultnet_rs::CultNetRudpPacketType::Disconnect +} + +fn is_answer(packet: &cultnet_rs::CultNetRudpPacket) -> bool { + packet.packet_type == cultnet_rs::CultNetRudpPacketType::Data && packet.reliable +} + +/// The capped frame's answer is reliable: its datagram is lost, the server resends it, and only once the +/// peer has acknowledged it does the goodbye follow. Neither transmission acknowledges the capped frame. +#[test] +fn a_lost_capped_answer_is_resent_and_the_goodbye_waits_for_its_acknowledgement() -> Result<()> { + let clock = Clock::new(81_000); + let mut server = server(Default::default(), clock.clone(), Sink::default(), Source::default())?; + let (mut peer, capped) = peer_at_the_cap(&mut server, 81)?; + + let first = peer.drain(); + assert!(first.iter().any(is_answer), "fixture: the answer was sent, and is lost here"); + assert!(!first.iter().any(is_goodbye), "no goodbye before the answer is acknowledged"); + assert!( + !first.iter().any(|packet| acknowledges(packet, capped)), + "nothing sent for the capped frame acknowledges it" + ); + assert_eq!(server.session_count(), 1); + + clock.set_monotonic(81_000 + 60); + server.maintain()?; + let resent = peer.drain(); + let answer = resent.iter().find(|packet| is_answer(packet)).expect("the lost answer is resent"); + assert!(!acknowledges(answer, capped), "the resent answer acknowledges nothing the peer sent"); + assert!(!resent.iter().any(is_goodbye), "still no goodbye"); + let delivered = peer.session.receive(answer, 4)?.delivered; + let Ok(CultNetMessage::Error { code, .. }) = + decode_cultnet_message_from_slice(&delivered[0].payload, CultNetWireContract::CultNetSchemaV0) + else { + panic!("the resent answer is a typed refusal"); + }; + assert_eq!(code, Some(CultNetErrorCode::MessageInvalid)); + + peer.send_packet(&peer.session.create_ack_for(answer.sequence))?; + assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); + assert!(peer.drain().iter().any(is_goodbye), "acknowledged: the goodbye follows"); + assert_eq!(server.session_count(), 0); + Ok(()) +} + +/// A peer that never acknowledges the answer is not kept for ever: the session ends after the grace. +#[test] +fn a_capped_session_whose_answer_is_never_acknowledged_ends_after_the_grace() -> Result<()> { + let clock = Clock::new(82_000); + let mut server = server(Default::default(), clock.clone(), Sink::default(), Source::default())?; + let (peer, _) = peer_at_the_cap(&mut server, 82)?; + peer.drain(); + + clock.set_monotonic(82_000 + 1_999); + server.maintain()?; + assert!(!peer.drain().iter().any(is_goodbye), "inside the grace"); + assert_eq!(server.session_count(), 1); + clock.set_monotonic(82_000 + 2_000); + server.maintain()?; + assert!(peer.drain().iter().any(is_goodbye), "the grace ran out"); + assert_eq!(server.session_count(), 0); + Ok(()) +} + +/// Nothing behind the capped frame is served while its answer is delivered. +#[test] +fn nothing_behind_the_capped_frame_is_served() -> Result<()> { + let sink = Sink::default(); + let mut server = server(Default::default(), Clock::new(83_000), sink.clone(), Source::default())?; + let (mut peer, _) = peer_at_the_cap(&mut server, 83)?; + peer.send_payload(valid_put("behind-the-cap"))?; + for _ in 0..3 { + server.poll_once()?; + } + assert!(sink.0.lock().unwrap().receipts.is_empty(), "the put behind the cap was not stored"); + Ok(()) +} -- 2.47.3 From b2077fa2bfa4734f3151eeafede5f7a9c2c27848 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:21:08 +0000 Subject: [PATCH 2/2] cultmesh-rs: a Pong in the capped grace acknowledges nothing; the grace deadline saturates A Ping in the 2 s grace after the capped frame was answered with a Pong built from the session's ack state, which includes the capped frame. Every reply while the session is ending now zeroes its ack fields. The deadline no longer wraps at the end of the clock. Co-Authored-By: Claude Sonnet 5.5 --- .../cultmesh-rs/src/rudp_document_server.rs | 15 ++++++-- .../cultmesh-rs/tests/rudp_document_server.rs | 34 +++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 30f76273..647589e0 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -605,7 +605,18 @@ where }; let mut unsendable = match result.reply { - Some(reply) => self.send_packet(key.remote_addr, &reply)?, + Some(mut reply) => { + // A reply (a Pong) in the grace after the capped frame acknowledges nothing either. + if self + .sessions + .get(&key) + .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some()) + { + reply.ack = 0; + reply.ack_mask = 0; + } + self.send_packet(key.remote_addr, &reply)? + } None => None, }; @@ -710,7 +721,7 @@ where let packets = self.refusal_packets(&mut entry.session, &refusal, true); if !packets.is_empty() { entry.ending_deadline_monotonic_millis = - Some(now_monotonic + CAPPED_ANSWER_GRACE_MILLIS); + Some(now_monotonic.saturating_add(CAPPED_ANSWER_GRACE_MILLIS)); } self.sessions.insert(key, entry); packets diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index a1954955..0e3cba66 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -2022,3 +2022,37 @@ fn nothing_behind_the_capped_frame_is_served() -> Result<()> { assert!(sink.0.lock().unwrap().receipts.is_empty(), "the put behind the cap was not stored"); Ok(()) } + +/// A Ping inside the grace is answered with a Pong that acknowledges nothing: its ack fields are built from +/// what the session received, which includes the capped frame. +#[test] +fn a_pong_in_the_grace_after_the_cap_acknowledges_nothing() -> Result<()> { + let mut server = server(Default::default(), Clock::new(84_000), Sink::default(), Source::default())?; + let (mut peer, capped) = peer_at_the_cap(&mut server, 84)?; + peer.drain(); + + let ping = peer.session.create_ping(b"alive".to_vec()); + peer.send_packet(&ping)?; + assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); + let pong = peer + .drain() + .into_iter() + .find(|packet| packet.packet_type == cultnet_rs::CultNetRudpPacketType::Pong) + .expect("the ping is answered"); + assert_eq!(pong.payload, b"alive"); + assert!(!acknowledges(&pong, capped), "the Pong acknowledges the capped frame"); + assert_eq!((pong.ack, pong.ack_mask), (0, 0)); + Ok(()) +} + +/// The grace deadline saturates: at the end of the clock the answer is not given up on at once. +#[test] +fn the_grace_deadline_does_not_wrap_at_the_end_of_the_clock() -> Result<()> { + let mut server = server(Default::default(), Clock::new(u64::MAX - 10), Sink::default(), Source::default())?; + let (peer, _) = peer_at_the_cap(&mut server, 85)?; + peer.drain(); + server.maintain()?; + assert!(!peer.drain().iter().any(is_goodbye), "the deadline wrapped and the session ended before its grace"); + assert_eq!(server.session_count(), 1); + Ok(()) +} -- 2.47.3