From a71f62e99739151d06b774c9432e51b188bd9e46 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 23:01:33 +0200 Subject: [PATCH] cultmesh-rs: pin the refusal at a queue full to the brim, and a real EMSGSIZE The queue-full test used an empty queue, so a reliable refusal fitted and the reliable-refusal mutant lived. The new test fills a queue of 8 exactly with a reply the peer never acknowledges; the next reply cannot be queued, and its refusal must still arrive, acknowledging nothing the peer sent. A RawPeer helper drives the peer packet by packet; the gap test uses it too. A response too large for any datagram fails with the kernel's EMSGSIZE: the rejection is ResponseSendFailed("EMSGSIZE"), the goodbye is exactly "packet could not be sent: EMSGSIZE", and no refusal is sent. Co-Authored-By: Claude Opus 5.5 --- .../cultmesh-rs/tests/rudp_document_server.rs | 298 ++++++++++++++---- 1 file changed, 236 insertions(+), 62 deletions(-) diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 7cb02938..30fe4678 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -1391,90 +1391,128 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> { Ok(()) } +/// A peer driven packet by packet, so a test decides what it receives and what +/// it acknowledges. +struct RawPeer { + socket: UdpSocket, + target: SocketAddr, + session: cultnet_rs::CultNetRudpSession, +} + +impl RawPeer { + fn connect( + server: &mut CultMeshRudpDocumentServer, + connection_id: u32, + ) -> Result { + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_read_timeout(Some(Duration::from_millis(50)))?; + let mut peer = Self { + socket, + target: server.local_addr()?, + session: cultnet_rs::CultNetRudpSession::new(cultnet_rs::CultNetRudpSessionOptions { + connection_id, + initial_sequence: Some(100), + resend_delay_ms: 10_000, + max_pending_reliable_packets: Some(64), + }), + }; + let connect = peer.session.create_connect(0, Vec::new())?; + peer.send_packet(&connect)?; + server.poll_once()?; + let accept = peer.receive().expect("the server accepts the connect"); + assert_eq!( + accept.packet_type, + cultnet_rs::CultNetRudpPacketType::Accept + ); + peer.session.receive(&accept, 1)?; + Ok(peer) + } + + fn send_packet(&self, packet: &cultnet_rs::CultNetRudpPacket) -> Result<()> { + self.socket + .send_to(&cultnet_rs::encode_rudp_packet(packet)?, self.target)?; + Ok(()) + } + + /// 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)?; + let options = cultnet_rs::CultNetRudpSendOptions { + 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))?; + for packet in &packets { + self.send_packet(packet)?; + } + Ok(packets.iter().map(|packet| packet.sequence).collect()) + } + + fn receive(&self) -> Option { + let mut wire = vec![0_u8; 65_535]; + let (bytes, _) = self.socket.recv_from(&mut wire).ok()?; + cultnet_rs::decode_rudp_packet(&wire[..bytes]).ok() + } + + /// Every packet waiting, none of them processed or acknowledged. + fn drain(&self) -> Vec { + std::iter::from_fn(|| self.receive()).collect() + } +} + +/// Whether `packet`'s acknowledgement fields cover `sequence`. +fn acknowledges(packet: &cultnet_rs::CultNetRudpPacket, sequence: u32) -> bool { + packet.ack == sequence + || (0..32).any(|bit| { + packet.ack_mask & (1 << bit) != 0 + && packet.ack > bit + && packet.ack - bit - 1 == sequence + }) +} + /// The refusal is unreliable and unordered: a peer missing an earlier reliable /// packet from the server still receives it. An ordered refusal would wait /// behind the gap for a resend that never comes, because the session ends with it. #[test] fn a_refusal_reaches_a_peer_missing_an_earlier_packet() -> Result<()> { - use cultnet_rs::{ - CultNetRudpPacket, CultNetRudpPacketType, CultNetRudpSendOptions, CultNetRudpSession, - CultNetRudpSessionOptions, decode_rudp_packet, encode_rudp_packet, - }; let mut server = server( Default::default(), Clock::new(71_000), Sink::fail_once(), Source::documents(vec![document("held", vec![1, 2, 3])]), )?; - let target = server.local_addr()?; - let peer = UdpSocket::bind("127.0.0.1:0")?; - peer.set_read_timeout(Some(Duration::from_millis(50)))?; - let received = || -> Option { - let mut wire = vec![0_u8; 65_535]; - let (bytes, _) = peer.recv_from(&mut wire).ok()?; - decode_rudp_packet(&wire[..bytes]).ok() - }; - let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { - connection_id: 78, - initial_sequence: Some(100), - resend_delay_ms: 10_000, - max_pending_reliable_packets: Some(64), - }); - peer.send_to( - &encode_rudp_packet(&session.create_connect(0, Vec::new())?)?, - target, - )?; - server.poll_once()?; - let accept = received().expect("the server accepts the connect"); - assert_eq!(accept.packet_type, CultNetRudpPacketType::Accept); - session.receive(&accept, 1)?; - let reliable = CultNetRudpSendOptions { - reliable: true, - ordered: true, - sequenced: false, - now_ms: 2, - reliable_expire_after_ms: None, - }; - let send = |session: &mut CultNetRudpSession, message: &CultNetMessage| -> Result<()> { - let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; - for packet in session.send_many("schema", payload, reliable.clone(), Some(1200))? { - peer.send_to(&encode_rudp_packet(&packet)?, target)?; - } - Ok(()) - }; + let mut peer = RawPeer::connect(&mut server, 78)?; // The snapshot reply is lost: the peer never sees it, so its ordered // stream from the server has a gap. - send( - &mut session, - &CultNetMessage::SnapshotRequest { - message_id: "lost".into(), - schema_ids: None, - record_keys: None, - }, - )?; + peer.send(&CultNetMessage::SnapshotRequest { + message_id: "lost".into(), + schema_ids: None, + record_keys: None, + })?; assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); - let mut lost = 0; - while received().is_some() { - lost += 1; - } - assert!(lost > 0, "fixture: the server sent the reply that is lost"); + assert!( + !peer.drain().is_empty(), + "fixture: the server sent the reply that is lost" + ); - send( - &mut session, - &CultNetMessage::DocumentPutRaw { - message_id: "refused".into(), - document: document("refused", vec![4, 5, 6]), - }, - )?; + peer.send(&CultNetMessage::DocumentPutRaw { + message_id: "refused".into(), + document: document("refused", vec![4, 5, 6]), + })?; assert!(matches!( server.poll_once()?, CultMeshRudpPollOutcome::ApplicationRejected(_) )); let mut delivered = Vec::new(); let mut disconnected = false; - while let Some(packet) = received() { - let result = session.receive(&packet, 3)?; + for packet in peer.drain() { + let result = peer.session.receive(&packet, 3)?; delivered.extend(result.delivered); disconnected |= result.disconnected; } @@ -1496,6 +1534,92 @@ fn a_refusal_reaches_a_peer_missing_an_earlier_packet() -> Result<()> { Ok(()) } +/// The peer's reliable queue is full to the brim: an 8-fragment reply the peer +/// never acknowledges fills a queue of 8 exactly. The next reply cannot be +/// queued, and its refusal still reaches the peer, because it takes no room in +/// that queue; it acknowledges nothing the peer sent. +#[test] +fn a_refusal_reaches_a_peer_whose_queue_is_full_to_the_brim() -> Result<()> { + let options = CultMeshRudpDocumentServerOptions { + max_fragment_bytes: 100, + max_pending_reliable_packets_per_session: 8, + max_snapshot_response_bytes: 8192, + ..Default::default() + }; + let source = Source::documents(vec![document_served_at("q1", 800)?]); + let mut server = server(options, Clock::new(74_000), Sink::default(), source)?; + let mut peer = RawPeer::connect(&mut server, 79)?; + peer.send(&CultNetMessage::SnapshotRequest { + message_id: "a".into(), + schema_ids: None, + record_keys: None, + })?; + assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); + let unacknowledged = peer + .drain() + .into_iter() + .filter(|packet| packet.packet_type == cultnet_rs::CultNetRudpPacketType::Data) + .count(); + assert_eq!( + unacknowledged, 8, + "fixture: the first reply fills the queue exactly" + ); + + let refused = peer.send(&CultNetMessage::SnapshotRequest { + message_id: "b".into(), + schema_ids: None, + record_keys: None, + })?; + let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else { + panic!("the second reply cannot be queued"); + }; + assert!( + matches!( + rejection.reason, + CultMeshRudpRejectionReason::ResponseQueueFailed(_) + ), + "{:?}", + rejection.reason + ); + let arrived = peer.drain(); + let mut parts: Vec<_> = arrived + .iter() + .filter(|packet| { + packet.packet_type == cultnet_rs::CultNetRudpPacketType::Data + && packet.channel_id == "schema" + && !packet.reliable + }) + .collect(); + assert!(!parts.is_empty(), "the refusal reaches the peer"); + for part in &parts { + assert!( + refused + .iter() + .all(|sequence| !acknowledges(part, *sequence)), + "the refusal acknowledges the refused request: ack {} mask {:#x}", + part.ack, + part.ack_mask + ); + } + parts.sort_by_key(|part| part.fragment_index); + let payload: Vec = parts.iter().flat_map(|part| part.payload.clone()).collect(); + assert_eq!( + decode_cultnet_message_from_slice(&payload, CultNetWireContract::CultNetSchemaV0)?, + CultNetMessage::Error { + error: "the snapshot response could not be queued".into(), + code: None, + details: None, + } + ); + assert!( + arrived + .iter() + .any(|packet| packet.packet_type == cultnet_rs::CultNetRudpPacketType::Disconnect), + "the refusal is followed by a goodbye" + ); + Ok(()) +} + /// A snapshot the reliable queue cannot take is answered with a refusal, not /// left to time out: the refusal needs no room in that queue. #[test] @@ -1545,3 +1669,53 @@ fn a_snapshot_the_queue_cannot_take_is_answered() -> Result<()> { ); Ok(()) } + +/// A response too large for any datagram fails with the kernel's EMSGSIZE. The +/// caller's rejection and the peer's goodbye both carry its fixed name, never +/// the platform's text, and the peer is sent no refusal. +#[test] +fn a_response_that_is_too_large_to_send_is_named_emsgsize() -> Result<()> { + let options = CultMeshRudpDocumentServerOptions { + max_fragment_bytes: 70_000, + ..Default::default() + }; + let source = Source::documents(vec![document("big", vec![7; 66_000])]); + let mut server = server(options, Clock::new(73_000), Sink::default(), source)?; + let mut client = client(server.local_addr()?, 152)?; + connect(&mut server, &mut [&mut client])?; + send_reliable( + &mut client, + &CultNetMessage::SnapshotRequest { + message_id: "big".into(), + schema_ids: None, + record_keys: None, + }, + )?; + let mut rejection = None; + for _ in 0..100 { + if let CultMeshRudpPollOutcome::ApplicationRejected(found) = server.poll_once()? { + rejection = Some(found); + break; + } + } + assert_eq!( + rejection.expect("the response can never be sent").reason, + CultMeshRudpRejectionReason::ResponseSendFailed("EMSGSIZE") + ); + let mut frames = 0; + for _ in 0..500 { + if client.receive_once()?.is_some() { + frames += 1; + } + if !client.connected() { + break; + } + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(frames, 0, "no refusal is sent to an unreachable peer"); + assert_eq!( + client.disconnect_reason(), + Some(&b"packet could not be sent: EMSGSIZE"[..]) + ); + Ok(()) +}