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]