diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 1e729407..9aae5be9 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -780,6 +780,13 @@ where return Ok(CultMeshRudpPollOutcome::Handled); } + // Whether this packet is the first the session takes for its sequence: a repeat of an answered write, or a + // packet too far ahead to be remembered, is not, and the write is not answered for it again. + let first_receipt = packet.reliable + && self + .sessions + .get(&key) + .is_some_and(|entry| !entry.session.has_received(packet.sequence)); let result = { let entry = self .sessions @@ -825,6 +832,8 @@ where // go unanswered. A packet that carries its whole frame is answered from its own payload. A superseded // fragment cannot be: the write's id may lie in a fragment this packet does not hold. if refuse_sequenced + && first_receipt + && self.sessions.get(&key).is_some_and(|entry| entry.session.has_received(packet.sequence)) && packet.fragment_count == 0 && !result.delivered.iter().any(|frame| frame.channel_id == "schema" && frame.sequence == packet.sequence) { diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 6ca0007a..60596ee5 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -3138,6 +3138,64 @@ fn a_superseded_sequenced_write_is_still_refused() -> Result<()> { Ok(()) } +/// One answer per write (write-receipt.md, rule 1): a sequenced packet the server already answered is answered +/// once, however often the peer sends it again, and a packet too far ahead to be received draws nothing until it +/// is received (the peer resends it, and the answer comes then). +#[test] +fn a_repeated_or_far_ahead_sequenced_write_draws_one_answer_not_one_per_send() -> Result<()> { + let sink = Sink::default(); + let mut server = server(Default::default(), Clock::new(70_000), sink.clone(), Source::default())?; + let mut peer = RawPeer::connect(&mut server, 256)?; + peer.send(&receipt_hello())?; + server.poll_once()?; + let packet_for = |peer: &mut RawPeer, id: &str| -> Result> { + let payload = encode_cultnet_message_to_vec(&put(id, &format!("note:{id}"), vec![1]), CultNetWireContract::CultNetSchemaV0)?; + peer.session.send_many( + "schema", + payload, + cultnet_rs::CultNetRudpSendOptions { reliable: true, ordered: false, sequenced: true, now_ms: 3, reliable_expire_after_ms: None }, + Some(1200), + ) + }; + let refusals_of = |peer: &mut RawPeer| -> Result> { + let mut ids = Vec::new(); + for packet in peer.drain() { + for frame in peer.session.receive(&packet, 4)?.delivered { + let answer = decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0)?; + if matches!(answer, CultNetMessage::ErrorV1 { .. }) { + ids.push(refusal_of(&answer).1.expect("a refusal names its write")); + } + } + } + Ok(ids) + }; + // The same packet, sent three times (a resend whose acknowledgement was lost), is refused once. + let once = packet_for(&mut peer, "dup")?; + let mut refused = Vec::new(); + for _ in 0..3 { + for packet in &once { + peer.send_packet(packet)?; + } + server.poll_once()?; + refused.extend(refusals_of(&mut peer)?); + } + assert_eq!(refused, vec!["dup".to_string()], "a repeat of an answered write is not answered again"); + + // A packet far past the receive window is not received: no answer yet, and none per resend. + let ahead = packet_for(&mut peer, "ahead")?; + let mut far = ahead[0].clone(); + far.sequence = once[0].sequence + 5_000; + let mut refused = Vec::new(); + for _ in 0..3 { + peer.send_packet(&far)?; + server.poll_once()?; + refused.extend(refusals_of(&mut peer)?); + } + assert_eq!(refused, Vec::::new(), "an unreceived packet draws no answer"); + assert_eq!(sink.0.lock().unwrap().receipts.len(), 0); + Ok(()) +} + /// A superseded sequenced packet is answered even when its arrival fills a gap and releases another write: the /// frames that arrive with it do not stand in for its own. #[test] diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index eae7a2ee..427a38c1 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1277,6 +1277,12 @@ impl CultNetRudpSession { } } + /// Whether `receive` has taken this reliable sequence: delivered, held, superseded or a duplicate. A packet + /// `receive` refused unremembered (too far ahead, or before the handshake) is not, and its sender resends it. + pub fn has_received(&self, sequence: u32) -> bool { + self.was_received(sequence) + } + /// The one duplicate test: true for a sequence in the window that was /// received, for one below the window, and for one at or before the /// watermark. The watermark starts at the handshake's seed, so a frame the