From 74705c7cd1e1ecb4fed629a189d513a670f78dc4 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 22:13:24 +0000 Subject: [PATCH] Rust RUDP server: answer a superseded sequenced write; order frame and packet by serial arithmetic Superseded packet: a reliable-sequenced packet that arrives after a newer one delivers no frame, yet is acknowledged and never resent, so its write went unanswered. The test (red first: only "new" was refused, "old" got nothing) showed the drop happens in the session below frame decoding, but an unfragmented packet carries its whole frame, so the server can read the id from the packet payload. Chosen: refuse it from the packet payload. A superseded fragment of a larger write has no readable id in the packet, so write-receipt.md now says that write may go unanswered and the client's unconfirmed outcome covers it. Ordering: frame.sequence <= packet.sequence is replaced by sequence_at_or_before, serial arithmetic modulo 2^32. Deviation from the brief: cultnet-rs' at_or_before carries a 4096-sequence receive window made for duplicate detection; a frame's first fragment may lie further back than that, so reusing it would leave such a sequenced write unrefused and applied. The new test pins that, and a hand mutant with the window fails it. A wrapped sequence cannot reach this compare through the session today (it is judged a duplicate), so the unit test pins the helper, not an end-to-end wrap. Co-Authored-By: Claude Sonnet 5.5 --- contracts/cultnet/interop/write-receipt.md | 2 +- .../cultmesh-rs/src/rudp_document_server.rs | 33 +++++- .../cultmesh-rs/tests/rudp_document_server.rs | 102 ++++++++++++++++++ 3 files changed, 135 insertions(+), 2 deletions(-) diff --git a/contracts/cultnet/interop/write-receipt.md b/contracts/cultnet/interop/write-receipt.md index 8581cc53..04d4b2fd 100644 --- a/contracts/cultnet/interop/write-receipt.md +++ b/contracts/cultnet/interop/write-receipt.md @@ -38,7 +38,7 @@ A connection is receipt-capable when its `cultnet.hello.v0` (sent before its fir - **Limit refusals are remembered.** A `limit_reached` refusal is remembered like any other: a resend of the same id and content hash replays `limit_reached`. No slot is freed mid-session, so an id refused at a bound never becomes admitted later in that session (a server at the cap keeps the refused digest pair in a bounded refusal set of at most 1024 per session, outside the entry budget; past it, a further new write is refused `limit_reached` without being remembered, and the session refuses every id it has not seen for the rest of its life, even if a server-wide budget has since freed, so a refusal that was not remembered is never followed by an admission of the same id). The client mints a new id on a new session. - **Required fields.** A put or delete whose document or `schemaId` is null or missing is refused `message_invalid` with `details.field` `"document"` or `"schemaId"` (the first missing in that order) and fixed text; the write is not applied. A `schemaId` or `recordKey` that is empty or only whitespace (present but names nothing) is not "missing": the write is refused `message_invalid` exactly as a frame that does not read, with `details.field` `"message"` and the write's id (vector `error_v1_empty_schema_id_is_unreadable_and_names_the_write`), and is never applied; a server must not hand it to its store. - **Fixed texts.** The `error` string of these refusals is fixed, so vectors can compare it: `message_invalid: messageId exceeds 256 bytes`, `message_invalid: document is required`, `message_invalid: schemaId is required`, `message_invalid: receipt sessions require the reliable channel`, `limit_reached: the session's remembered-write limit is full`. - - **Reliable channel only.** A write is accepted on a receipt session only on a channel that is reliable and not sequenced (reliable ordered and reliable unordered both qualify; a sequenced channel drops superseded packets, so a write on it can vanish without trace). A write arriving on any other channel is refused `message_invalid` with `details.field: "channel"` (the transport profile's word for it) and fixed text. Receipts and refusals travel the reliable channel; a write that could be lost without trace cannot be answered once. + - **Reliable channel only.** A write is accepted on a receipt session only on a channel that is reliable and not sequenced (reliable ordered and reliable unordered both qualify; a sequenced channel drops superseded packets, so a write on it can vanish without trace). A write arriving on any other channel is refused `message_invalid` with `details.field: "channel"` (the transport profile's word for it) and fixed text. Receipts and refusals travel the reliable channel; a write that could be lost without trace cannot be answered once. The server refuses a sequenced write whenever it can read the write's id: from the delivered frame, and from a superseded packet that carries its whole frame. A write sent as several fragments whose frame is superseded is dropped whole and may go unanswered; the client's unconfirmed outcome covers it. - Past the session (or a server's longer retention, if it keeps one) a resend is a new write. 6. **Forwarded writes.** A server that forwards a non-primary write answers the original peer only with what the primary answered: the primary's receipt becomes `write_admitted` with the original id, its refusal becomes `error.v1` with the original id. If the forward gets no answer the server relays nothing: no receipt, no error, no synthesized refusal. The client's own rule below then reports the write as unconfirmed. A forwarder never turns an absent or ambiguous primary answer into an admission. 7. **Looks alike.** An admitted and a refused write differ in `schemaVersion`; a receipt has no `error`, an `error.v1` has no admission. A transport ack is neither. diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 12fdf3e6..a0ab51ac 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -821,13 +821,26 @@ where } let mut kept_rejection = None; + // A superseded packet delivers no frame, yet it was acknowledged and will not be resent, so its write would + // 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 + && packet.fragment_count == 0 + && !result.delivered.iter().any(|frame| frame.channel_id == "schema" && frame.sequence == packet.sequence) + { + if let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = + self.refuse_unreliable_write(key, &packet.payload)? + { + kept_rejection = Some(rejection); + } + } for frame in result.delivered { if frame.channel_id != "schema" { continue; } // `delivered` also holds the ordered frames this packet's arrival released from the hold; those were // not sent on the sequenced channel. They sit above the packet's own sequence, its own frame at or below. - if refuse_sequenced && frame.sequence <= packet.sequence { + if refuse_sequenced && sequence_at_or_before(frame.sequence, packet.sequence) { if let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = self.refuse_unreliable_write(key, &frame.payload)? { @@ -1816,6 +1829,13 @@ where #[cfg(test)] mod send_failure_tests; +/// Whether `sequence` is at or before `mark` in serial order (RFC 1982): the 32-bit space is compared modulo 2^32, +/// so a frame that began before the wrap and a packet after it still order correctly. The receive window of +/// `cultnet_rs`' duplicate test does not belong here: a frame's first fragment may lie any distance back. +fn sequence_at_or_before(sequence: u32, mark: u32) -> bool { + mark.wrapping_sub(sequence) < 1 << 31 +} + fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> { if options.max_sessions == 0 { return Err(anyhow!("max_sessions must be greater than zero")); @@ -1931,4 +1951,15 @@ mod tests { .unwrap(); assert_eq!(hex(&bytes), MESSAGE_INVALID_HEX); } + + #[test] + fn sequence_order_holds_across_the_wrap_and_beyond_the_receive_window() { + assert!(sequence_at_or_before(7, 7)); + assert!(sequence_at_or_before(100, 5_200), "further back than the receive window"); + assert!(!sequence_at_or_before(5_200, 100)); + assert!(sequence_at_or_before(u32::MAX - 1, 1), "begun before the wrap, ended after it"); + assert!(!sequence_at_or_before(1, u32::MAX - 1), "after the wrap is later than before it"); + assert!(sequence_at_or_before(u32::MAX, 0), "the wrap itself"); + assert!(!sequence_at_or_before(0, u32::MAX)); + } } diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 50e63425..69813874 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -3097,6 +3097,108 @@ fn a_sequenced_packet_releasing_a_held_ordered_write_refuses_only_its_own() -> R Ok(()) } +/// The refusal of a write on a sequenced channel does not depend on the packet being the newest. A reliable-sequenced +/// packet that arrives after a newer one on its channel is superseded: the session delivers no frame for it, yet +/// the packet was acknowledged and will not be resent, so the server answers it from the packet's own payload. +#[test] +fn a_superseded_sequenced_write_is_still_refused() -> 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, 253)?; + 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 older = packet_for(&mut peer, "old")?; + let newer = packet_for(&mut peer, "new")?; + let mut refused_ids = Vec::new(); + for packets in [&newer, &older] { + for packet in packets { + peer.send_packet(packet)?; + } + server.poll_once()?; + 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 { .. }) { + refused_ids.push(refusal_of(&answer).1.expect("a refusal names its write")); + } + } + } + } + assert_eq!(refused_ids, vec!["new".to_string(), "old".to_string()], "the superseded write is answered too"); + assert_eq!(sink.0.lock().unwrap().receipts.len(), 0, "and nothing was applied"); + Ok(()) +} + +/// The refusal reaches a sequenced write however far apart its fragments' sequences sit: other traffic can take the +/// sequences between the first fragment and the last, further back than the duplicate test's receive window. +#[test] +fn a_sequenced_write_whose_fragments_span_more_than_the_receive_window_is_refused() -> 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, 254)?; + peer.send(&receipt_hello())?; + server.poll_once()?; + let payload = encode_cultnet_message_to_vec(&put("far", "note:far", vec![1; 64]), CultNetWireContract::CultNetSchemaV0)?; + let mut packets = peer.session.send_many( + "schema", + payload, + cultnet_rs::CultNetRudpSendOptions { reliable: true, ordered: false, sequenced: true, now_ms: 3, reliable_expire_after_ms: None }, + Some(32), + )?; + let last = packets.pop().expect("the write is fragmented"); + assert!(!packets.is_empty()); + let mut next = last.sequence; + let mut refused_ids = Vec::new(); + for packet in &packets { + peer.send_packet(packet)?; + server.poll_once()?; + } + // 5,000 reliable packets on another channel, in order, take the sequences between the fragments. + for count in 0..5_000 { + let mut filler = packets[0].clone(); + filler.channel_id = "filler".into(); + filler.sequenced = false; + filler.fragment_id = 0; + filler.fragment_index = 0; + filler.fragment_count = 0; + filler.payload = vec![0]; + filler.sequence = next; + next += 1; + peer.send_packet(&filler)?; + server.poll_once()?; + if count % 100 == 99 { + // Read what the server sent, so the socket buffer never holds back the refusal. + for packet in peer.drain() { + peer.session.receive(&packet, 4)?; + } + } + } + let mut last = last; + last.sequence = next; + peer.send_packet(&last)?; + server.poll_once()?; + 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 { .. }) { + refused_ids.push(refusal_of(&answer).1.expect("a refusal names its write")); + } + } + } + assert_eq!(refused_ids, vec!["far".to_string()]); + assert_eq!(sink.0.lock().unwrap().receipts.len(), 0, "and nothing was applied"); + Ok(()) +} + /// A refused write is traffic: the peer that sends it is not idle, so its session outlives the idle timeout measured /// from the refusal and not from the packet before it. #[test]