diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 1553168b..647589e0 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. @@ -598,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, }; @@ -612,6 +630,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 +690,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 +705,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.saturating_add(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 +758,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 +774,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 +792,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 +820,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 +891,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 +1131,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 +1154,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 +1167,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..0e3cba66 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,138 @@ 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(()) +} + +/// 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(()) +}