cultnet-rs, cultmesh-rs: null or missing document/schemaId refused naming the field; budget tests use literal bytes; unreliable-refusal local failure test

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -3,7 +3,7 @@ use cultnet_rs::{
     CultNetErrorCode, CultNetErrorDetails, CultNetMessage, CultNetRawDocumentRecord, CultNetRudpPacket, CultNetRudpPacketType,
     CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, CultNetWireContract,
     decode_cultnet_message_from_slice, decode_rudp_packet, encode_cultnet_message_to_vec,
-    encode_rudp_packet, is_permanent_send_error, readable_message_id, message_id_digest, send_error_code, write_content_hash,
+    encode_rudp_packet, is_permanent_send_error, readable_message_id, message_id_digest, missing_required_write_field, send_error_code, write_content_hash,
 };
 use std::collections::{BTreeMap, HashMap};
 #[cfg(test)]
@@ -289,6 +289,10 @@ pub enum CultMeshRudpRejectionReason {
     MessageIdTooLong,
     /// A receipt-capable session sent a write on an unreliable channel, where no answer could be owed once.
     ChannelNotReliable,
+    /// A receipt-capable session's put or delete has a null or missing `document`.
+    DocumentMissing,
+    /// A receipt-capable session's put or delete has a null or missing `schemaId`.
+    SchemaIdMissing,
 }
 
 impl std::fmt::Display for CultMeshRudpRejectionReason {
@@ -301,6 +305,8 @@ impl std::fmt::Display for CultMeshRudpRejectionReason {
             Self::WriteLimitReached => f.write_str("the session remembers as many writes as it will"),
             Self::MessageIdTooLong => f.write_str("the message id is longer than 256 bytes"),
             Self::ChannelNotReliable => f.write_str("a receipt session's write arrived on an unreliable channel"),
+            Self::DocumentMissing => f.write_str("the write has no document"),
+            Self::SchemaIdMissing => f.write_str("the write has no schemaId"),
             Self::DocumentUnservable {
                 response_bytes,
                 max_snapshot_response_bytes,
@@ -341,7 +347,7 @@ impl std::fmt::Display for CultMeshRudpRejectionReason {
 /// The texts a refused peer is sent, one per rejection variant. They are fixed:
 /// a sink's or source's error text can quote what a peer sent, so it stays with
 /// the caller that receives the rejection and never reaches the wire.
-pub(crate) const REFUSAL_TEXTS: [&str; 15] = [
+pub(crate) const REFUSAL_TEXTS: [&str; 17] = [
     "the catalog refused the document",
     "the document can never be served",
     "the snapshot source failed",
@@ -357,6 +363,8 @@ pub(crate) const REFUSAL_TEXTS: [&str; 15] = [
     "the session's remembered-write limit is full",
     "messageId exceeds 256 bytes",
     "receipt sessions require the reliable channel",
+    "document is required",
+    "schemaId is required",
 ];
 
 impl CultMeshRudpRejectionReason {
@@ -383,7 +391,9 @@ impl CultMeshRudpRejectionReason {
             11 => Self::MessageIdReused,
             12 => Self::WriteLimitReached,
             13 => Self::MessageIdTooLong,
-            _ => Self::ChannelNotReliable,
+            14 => Self::ChannelNotReliable,
+            15 => Self::DocumentMissing,
+            _ => Self::SchemaIdMissing,
         }
     }
 
@@ -404,6 +414,8 @@ impl CultMeshRudpRejectionReason {
             Self::WriteLimitReached => 12,
             Self::MessageIdTooLong => 13,
             Self::ChannelNotReliable => 14,
+            Self::DocumentMissing => 15,
+            Self::SchemaIdMissing => 16,
         }
     }
 }
@@ -484,6 +496,8 @@ impl CultMeshRudpApplicationRejection {
                 message_invalid_message("messageId", text)
             }
             (CultMeshRudpRejectionReason::MessageIdTooLong, _) => message_invalid_message("messageId", text),
+            (CultMeshRudpRejectionReason::DocumentMissing, _) => message_invalid_message("document", text),
+            (CultMeshRudpRejectionReason::SchemaIdMissing, _) => message_invalid_message("schemaId", text),
             (CultMeshRudpRejectionReason::ChannelNotReliable, _) => message_invalid_message("channel", text),
             (CultMeshRudpRejectionReason::WriteLimitReached, _) => CultNetMessage::Error {
                 error: format!("limit_reached: {text}"),
@@ -844,13 +858,26 @@ where
                     now_unix,
                     now_monotonic,
                 )?,
-                Err(error) => Some(CultMeshRudpApplicationRejection {
-                    session: key,
-                    operation: CultMeshRudpApplicationOperation::Unreadable,
-                    message_id: readable_message_id(&frame.payload).unwrap_or_default(),
-                    record_key: None,
-                    reason: CultMeshRudpRejectionReason::MessageUnreadable(format!("{error:#}")),
-                }),
+                Err(error) => {
+                    // A receipt session's put or delete missing a required field is a refused write that
+                    // names the field, not an unreadable frame.
+                    let missing = if receipt_session { missing_required_write_field(&frame.payload) } else { None };
+                    Some(CultMeshRudpApplicationRejection {
+                        session: key,
+                        operation: match missing {
+                            Some(("put", _)) => CultMeshRudpApplicationOperation::DocumentPutRaw,
+                            Some(_) => CultMeshRudpApplicationOperation::DocumentDelete,
+                            None => CultMeshRudpApplicationOperation::Unreadable,
+                        },
+                        message_id: readable_message_id(&frame.payload).unwrap_or_default(),
+                        record_key: None,
+                        reason: match missing {
+                            Some((_, "document")) => CultMeshRudpRejectionReason::DocumentMissing,
+                            Some(_) => CultMeshRudpRejectionReason::SchemaIdMissing,
+                            None => CultMeshRudpRejectionReason::MessageUnreadable(format!("{error:#}")),
+                        },
+                    })
+                }
             };
             if let Some(rejection) = rejection {
                 // A refused delete or unreadable frame below the cap is answered on the live session,

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -2646,7 +2646,7 @@ fn error_text_and_field(message: &CultNetMessage) -> (String, Option<String>) {
 #[test]
 fn a_server_wide_memory_budget_refuses_the_write_over_it_and_keeps_the_sessions() -> Result<()> {
     let options = CultMeshRudpDocumentServerOptions {
-        max_remembered_write_bytes: 2 * CultMeshRudpDocumentServerOptions::REMEMBERED_ENTRY_BYTES,
+        max_remembered_write_bytes: 2 * 65,
         ..Default::default()
     };
     let sink = Sink::default();
@@ -2685,7 +2685,7 @@ fn a_server_wide_memory_budget_refuses_the_write_over_it_and_keeps_the_sessions(
 #[test]
 fn a_message_id_over_256_bytes_is_refused_before_anything_is_remembered() -> Result<()> {
     let options = CultMeshRudpDocumentServerOptions {
-        max_remembered_write_bytes: CultMeshRudpDocumentServerOptions::REMEMBERED_ENTRY_BYTES,
+        max_remembered_write_bytes: 65,
         ..Default::default()
     };
     let sink = Sink::default();
@@ -2753,7 +2753,7 @@ fn an_unreliable_write_on_a_receipt_session_is_refused_message_invalid_naming_th
 #[test]
 fn limit_refusals_are_remembered_in_a_bounded_set() -> Result<()> {
     let options = CultMeshRudpDocumentServerOptions {
-        max_remembered_write_bytes: CultMeshRudpDocumentServerOptions::REMEMBERED_ENTRY_BYTES,
+        max_remembered_write_bytes: 65,
         ..Default::default()
     };
     let mut server = server(options, Clock::new(70_000), Sink::default(), Source::default())?;
@@ -2806,3 +2806,65 @@ fn an_unreliable_write_on_a_legacy_session_is_dropped_silently() -> Result<()> {
     assert_eq!(sink.0.lock().unwrap().receipts.len(), 0);
     Ok(())
 }
+
+/// A put or delete whose document or schemaId is null or missing is refused `message_invalid` naming that
+/// field (document first), with the spec's fixed text and the write's id; the session goes on.
+#[test]
+fn a_write_missing_its_document_or_schema_id_is_refused_naming_the_field() -> Result<()> {
+    let sink = Sink::default();
+    let mut server = server(Default::default(), Clock::new(70_000), sink.clone(), Source::default())?;
+    let mut peer = client(server.local_addr()?, 236)?;
+    connect(&mut server, &mut [&mut peer])?;
+    send_reliable(&mut peer, &receipt_hello())?;
+    let frames = [
+        ("p-missing", serde_json::json!({"schemaVersion": "cultnet.document_put_raw.v0", "messageId": "p-missing"}), "document"),
+        ("p-null", serde_json::json!({"schemaVersion": "cultnet.document_put_raw.v0", "messageId": "p-null", "document": null}), "document"),
+        ("p-noschema", serde_json::json!({"schemaVersion": "cultnet.document_put_raw.v0", "messageId": "p-noschema",
+            "document": {"recordKey": "k", "storedAt": "t", "payloadEncoding": "messagepack", "payload": [1]}}), "schemaId"),
+        ("d-null", serde_json::json!({"schemaVersion": "cultnet.document_delete.v0", "messageId": "d-null", "schemaId": null, "recordKey": "k"}), "schemaId"),
+        ("d-missing", serde_json::json!({"schemaVersion": "cultnet.document_delete.v0", "messageId": "d-missing", "recordKey": "k"}), "schemaId"),
+    ];
+    for (_, frame, _) in &frames {
+        peer.send_reliable("schema", rmp_serde::to_vec_named(frame)?)?;
+    }
+    let answers = answers_after(&mut server, &mut peer, frames.len())?;
+    for ((id, _, field), answer) in frames.iter().zip(&answers) {
+        assert_eq!(refusal_of(answer), (Some(CultNetErrorCode::MessageInvalid), Some((*id).into())));
+        assert_eq!(
+            error_text_and_field(answer),
+            (format!("message_invalid: {field} is required"), Some((*field).into()))
+        );
+    }
+    assert_eq!(sink.0.lock().unwrap().receipts.len(), 0);
+    assert_eq!(server.session_count(), 1);
+    Ok(())
+}
+
+/// An unreliable write whose refusal the full reliable queue cannot take ends the session as a local failure
+/// (the peer's write is then unconfirmed, never silent and never "refused a packet").
+#[test]
+fn an_unreliable_refusal_that_cannot_be_queued_ends_the_session_as_a_local_failure() -> Result<()> {
+    let options = CultMeshRudpDocumentServerOptions { max_pending_reliable_packets_per_session: 3, ..Default::default() };
+    let mut server = server(options, Clock::new(70_000), Sink::default(), Source::default())?;
+    let mut peer = RawPeer::connect(&mut server, 237)?;
+    peer.send(&receipt_hello())?;
+    server.poll_once()?;
+    // The peer acknowledges nothing, so each refusal stays in the server's reliable queue.
+    for index in 0..8 {
+        let payload = encode_cultnet_message_to_vec(&put(&format!("u{index}"), "note:1", vec![1]), CultNetWireContract::CultNetSchemaV0)?;
+        let packets = peer.session.send_many(
+            "schema",
+            payload,
+            cultnet_rs::CultNetRudpSendOptions { reliable: false, ordered: false, sequenced: false, now_ms: 3, reliable_expire_after_ms: None },
+            Some(1200),
+        )?;
+        for packet in &packets {
+            peer.send_packet(packet)?;
+        }
+        server.poll_once()?;
+    }
+    assert_eq!(server.session_count(), 0, "the refusal the queue could not take ended the session");
+    let goodbye = peer.drain().into_iter().find(is_goodbye).expect("the peer is told the session ended");
+    assert_eq!(goodbye.payload, b"server could not answer a write".to_vec());
+    Ok(())
+}

packages/cultnet-rs/src/contracts.rs

@@ -2020,3 +2020,29 @@ pub fn readable_message_id(bytes: &[u8]) -> Option<String> {
         _ => None,
     })
 }
+
+/// The first required field (`document`, then `schemaId`) that a put or delete frame lacks or holds as nil
+/// (write-receipt.md, "Required fields"), with the write's operation (`"put"` or `"delete"`). `None` when the
+/// bytes are not such a frame, or both fields are present. A put's `schemaId` is its document's.
+pub fn missing_required_write_field(bytes: &[u8]) -> Option<(&'static str, &'static str)> {
+    let rmpv::Value::Map(entries) = rmpv::decode::read_value(&mut &*bytes).ok()? else {
+        return None;
+    };
+    let field = |entries: &[(rmpv::Value, rmpv::Value)], name: &str| {
+        entries
+            .iter()
+            .find(|(key, _)| key.as_str() == Some(name))
+            .map(|(_, value)| value.clone())
+            .filter(|value| !value.is_nil())
+    };
+    match field(&entries, "schemaVersion")?.as_str()? {
+        "cultnet.document_put_raw.v0" => match field(&entries, "document") {
+            Some(rmpv::Value::Map(document)) => {
+                field(&document, "schemaId").is_none().then_some(("put", "schemaId"))
+            }
+            _ => Some(("put", "document")),
+        },
+        "cultnet.document_delete.v0" => field(&entries, "schemaId").is_none().then_some(("delete", "schemaId")),
+        _ => None,
+    }
+}