From d708247e9b8bddfbfd7e689cf1d37e80e3b01707 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 1 Oct 2026 01:03:41 +0200 Subject: [PATCH] cultmesh-rs: pin the in-poll release ack, per-session answers and expiry bounds The in-poll release test from the last commit sent its later put on an unreliable packet. The server never reads unreliable schema frames, so that put was never offered and the test failed. The release it meant to reach happens when an answer arrives from another thread while a poll runs. The test clock now accepts a held reply at its next wall-clock read, which models that. A Ping (unreliable) then gets exactly one release ack and no Pong. Reliable data on a channel the server does not read gets exactly its own end-of-poll ack. Two answers queued in one poll, an acceptance for one session and a refusal for another, each reach only their own session. This kills take_answers' session guard replaced by false. A session is kept at exactly its idle and lifetime limits and expires one millisecond past them. The expiry comparisons moved into this cut's diff when every removal went through remove_session. cargo-mutants --in-diff over the cut: 57 mutants, 44 caught, 8 unviable, 5 timeouts (take_answers scan mutants that never terminate), 0 missed. Co-Authored-By: Claude Opus 5.5 --- .../cultmesh-rs/tests/rudp_document_server.rs | 179 +++++++++++++----- 1 file changed, 131 insertions(+), 48 deletions(-) diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 340d0abb..9f793e75 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -158,6 +158,9 @@ fn stripping_server( struct Clock { unix: Arc, monotonic: Arc, + /// A reply accepted at the next wall-clock read: an answer that arrives + /// from another thread while a poll runs. + accept_on_read: Arc>>, } impl Clock { @@ -165,6 +168,7 @@ impl Clock { Self { unix: Arc::new(AtomicU64::new(now)), monotonic: Arc::new(AtomicU64::new(now)), + accept_on_read: Arc::default(), } } @@ -184,6 +188,9 @@ impl Clock { impl CultMeshRudpServerClock for Clock { fn now_unix_millis(&self) -> u64 { + if let Some(reply) = self.accept_on_read.lock().unwrap().take() { + reply.accept(); + } self.unix.load(Ordering::SeqCst) } @@ -1487,21 +1494,22 @@ impl RawPeer { /// Sends `message` reliably and in order; returns its sequences. fn send(&mut self, message: &CultNetMessage) -> Result> { - self.send_with(message, true) + let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; + self.send_on("schema", payload) } - fn send_with(&mut self, message: &CultNetMessage, reliable: bool) -> Result> { - let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; + /// Sends `payload` reliably and in order on `channel`; returns its sequences. + fn send_on(&mut self, channel: &str, payload: Vec) -> Result> { let options = cultnet_rs::CultNetRudpSendOptions { - reliable, - ordered: reliable, + reliable: true, + ordered: true, sequenced: false, now_ms: 2, reliable_expire_after_ms: None, }; let packets = self .session - .send_many("schema", payload, options, Some(1200))?; + .send_many(channel, payload, options, Some(1200))?; for packet in &packets { self.send_packet(packet)?; } @@ -2377,58 +2385,133 @@ fn an_unservable_put_is_never_offered_to_the_sink() -> Result<()> { Ok(()) } -/// A sink that holds each put's reply until it is offered the put `last`, -/// then accepts every held reply and that one, inside the same poll. -#[derive(Default)] -struct AnswersOnLast(Vec); - -impl CultMeshRudpRawDocumentSink for AnswersOnLast { - fn accept_raw_document( - &mut self, - receipt: CultMeshRudpRawDocumentReceipt, - reply: CultMeshRudpPutReply, - ) { - self.0.push(reply); - if receipt.message_id == "last" { - self.0.drain(..).for_each(CultMeshRudpPutReply::accept); - } - } -} - -/// A poll whose packet releases a withheld session sends exactly one -/// acknowledgement. A reliable packet's own acknowledgement is that one; an -/// unreliable packet has none, so the release sends it. +/// A poll whose packet finds its session released, by an answer that arrived +/// while the poll ran, sends exactly one acknowledgement. A reliable packet's +/// own acknowledgement is that one. An unreliable packet (a Ping) has none, +/// so the release sends it. #[test] fn a_release_inside_a_poll_sends_exactly_one_acknowledgement() -> Result<()> { for reliable in [false, true] { - let mut server = CultMeshRudpDocumentServer::new( - UdpSocket::bind("127.0.0.1:0")?, - AnswersOnLast::default(), - Source::default(), - Clock::new(89_000), - Default::default(), - )?; + let clock = Clock::new(89_000); + let sink = HeldSink::default(); + let mut server = held_server(&sink, &clock)?; let mut peer = RawPeer::connect(&mut server, 102)?; - let first = peer.send(&put("first"))?; + let held = peer.send(&put("held"))?; poll_until_idle(&mut server)?; - assert_eq!(peer.drain(), Vec::new(), "fixture: the first put is held"); + assert_eq!(peer.drain(), Vec::new(), "fixture: the put is held"); - let last = peer.send_with(&put("last"), reliable)?; + *clock.accept_on_read.lock().unwrap() = Some(sink.reply("held")); + let own = if reliable { + // Reliable data on a channel the server does not read. + peer.send_on("other", vec![1])? + } else { + let ping = peer.session.create_ping(Vec::new()); + peer.send_packet(&ping)?; + Vec::new() + }; assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); let sent = peer.drain(); - assert_eq!(sent.len(), 1, "reliable {reliable}: {sent:?}"); - assert_eq!(sent[0].packet_type, cultnet_rs::CultNetRudpPacketType::Ack); - assert!( - first - .iter() - .all(|sequence| acknowledges(&sent[0], *sequence)) + let kinds: Vec<_> = sent.iter().map(|packet| packet.packet_type).collect(); + // No Pong: the session was withheld when the Ping arrived. + assert_eq!( + kinds, + vec![cultnet_rs::CultNetRudpPacketType::Ack], + "reliable {reliable}" ); - if reliable { - assert!( - last.iter() - .all(|sequence| acknowledges(&sent[0], *sequence)) - ); + for sequence in held.iter().chain(&own) { + assert!(acknowledges(&sent[0], *sequence)); } } Ok(()) } + +/// Answers are handled per session: one session's answer never releases or +/// ends another. One session's put is accepted and another's refused in the +/// same poll; each session gets only its own answer. +#[test] +fn one_sessions_answer_never_releases_or_ends_another() -> Result<()> { + let clock = Clock::new(90_000); + let sink = HeldSink::default(); + let mut server = held_server(&sink, &clock)?; + let mut accepted = RawPeer::connect(&mut server, 103)?; + let mut refused = RawPeer::connect(&mut server, 104)?; + let accepted_put = accepted.send(&put("accepted"))?; + let refused_put = refused.send(&put("refused"))?; + poll_until_idle(&mut server)?; + + sink.reply("accepted").accept(); + sink.reply("refused").refuse("no CANARY-7f3a"); + let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else { + panic!("the refusal is the poll's outcome"); + }; + assert_eq!( + ( + rejection.session.connection_id, + rejection.message_id.as_str() + ), + (104, "refused") + ); + let to_accepted = accepted.drain(); + assert_eq!( + to_accepted + .iter() + .map(|packet| packet.packet_type) + .collect::>(), + vec![cultnet_rs::CultNetRudpPacketType::Ack] + ); + assert!( + accepted_put + .iter() + .all(|sequence| acknowledges(&to_accepted[0], *sequence)) + ); + let to_refused = refused.drain(); + assert_acknowledges_none(&to_refused, &refused_put); + assert_eq!( + to_refused + .iter() + .map(|packet| packet.packet_type) + .collect::>(), + vec![ + cultnet_rs::CultNetRudpPacketType::Data, + cultnet_rs::CultNetRudpPacketType::Disconnect + ] + ); + assert_eq!(server.session_count(), 1); + Ok(()) +} + +/// A session is kept while its idle time and its age are at their limits, and +/// expires once either is past it. +#[test] +fn a_session_is_kept_at_its_limits_and_expires_past_them() -> Result<()> { + for (idle, lifetime) in [(1_000, 60_000), (60_000, 1_000)] { + let clock = Clock::new(91_000); + let mut server = CultMeshRudpDocumentServer::new( + UdpSocket::bind("127.0.0.1:0")?, + HeldSink::default(), + Source::default(), + clock.clone(), + CultMeshRudpDocumentServerOptions { + session_idle_timeout: Duration::from_millis(idle), + session_max_lifetime: Duration::from_millis(lifetime), + ..Default::default() + }, + )?; + RawPeer::connect(&mut server, 105)?; + clock.set(92_000); + assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Idle); + assert_eq!( + server.session_count(), + 1, + "kept at the limit ({idle}, {lifetime})" + ); + clock.set(92_001); + assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Idle); + assert_eq!( + server.session_count(), + 0, + "expired past it ({idle}, {lifetime})" + ); + } + Ok(()) +}