cultmesh-rs: a sink's refusal of a put is a typed write_not_authorized

The RUDP document server sent a host sink's refusal as an untyped Error (code None).
It now sends write_not_authorized with details naming the record, as C#, TS and
Python do; the text stays the fixed variant text, never the sink's. The rejection
carries the put's record_key. Other rejections stay untyped. Unit test pins the
bytes to the error-vectors.json write_not_authorized vector; integration tests
check code/details on the wire and that a refused put reaches no sink.

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

packages/cultmesh-rs/src/lib.rs

@@ -865,6 +865,7 @@ pub fn request_raw_snapshot_from_rudp_catalog(
 /// fixed refusals a CultMesh catalog sends; any other text could carry
 /// anything, so it is left out.
 fn recognised_refusal(text: &str) -> String {
+    let text = text.strip_prefix("write_not_authorized: ").unwrap_or(text);
     if REFUSAL_TEXTS.contains(&text) {
         format!(": {text}")
     } else {

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -1,6 +1,6 @@
 use anyhow::{Result, anyhow};
 use cultnet_rs::{
-    CultNetMessage, CultNetRawDocumentRecord, CultNetRudpPacket, CultNetRudpPacketType,
+    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, send_error_code,
@@ -316,14 +316,50 @@ impl CultMeshRudpRejectionReason {
     }
 }
 
+/// The typed `write_not_authorized` refusal for a put whose record is `record_key`, as C#
+/// `CultNetErrorMessage.ForRemoteWriteRefused` builds it: code, and `details` naming the record.
+pub(crate) fn write_refused_message(record_key: &str, reason: &str) -> CultNetMessage {
+    CultNetMessage::Error {
+        error: format!("write_not_authorized: {reason}"),
+        code: Some(CultNetErrorCode::WriteNotAuthorized),
+        details: Some(CultNetErrorDetails {
+            field: Some("recordKey".into()),
+            value: Some(record_key.into()),
+            as_of: None,
+            current: None,
+        }),
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CultMeshRudpApplicationRejection {
     pub session: CultMeshRudpSessionKey,
     pub operation: CultMeshRudpApplicationOperation,
     pub message_id: String,
+    /// The record a rejected put concerned; `None` for a snapshot request. A sink refusal names it to the peer.
+    pub record_key: Option<String>,
     pub reason: CultMeshRudpRejectionReason,
 }
 
+impl CultMeshRudpApplicationRejection {
+    /// What the refused peer is sent: a sink's refusal of a put is a typed `write_not_authorized` naming
+    /// the record; every other rejection is the variant's fixed text, untyped. Neither carries the
+    /// sink's or source's own text.
+    fn refusal_message(&self) -> CultNetMessage {
+        let text = self.reason.refusal_text();
+        match (&self.reason, &self.record_key) {
+            (CultMeshRudpRejectionReason::SinkRefused(_), Some(record_key)) => {
+                write_refused_message(record_key, text)
+            }
+            _ => CultNetMessage::Error {
+                error: text.into(),
+                code: None,
+                details: None,
+            },
+        }
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum CultMeshRudpPollOutcome {
     Idle,
@@ -335,8 +371,8 @@ pub enum CultMeshRudpPollOutcome {
 enum SessionEnd {
     /// The session refused a packet.
     Refused,
-    /// The server rejected a peer's message; the peer is sent this fixed text.
-    Rejected(&'static str),
+    /// The server rejected a peer's message; the peer is sent this refusal.
+    Rejected(CultNetMessage),
     /// The session owes its peer a packet that can never be sent as built.
     Unsendable(std::io::Error),
 }
@@ -555,7 +591,7 @@ where
                     rejection.reason,
                     CultMeshRudpRejectionReason::ResponseSendFailed(_)
                 ) {
-                    self.end_session(key, SessionEnd::Rejected(rejection.reason.refusal_text()))?;
+                    self.end_session(key, SessionEnd::Rejected(rejection.refusal_message()))?;
                 }
                 return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection));
             }
@@ -683,11 +719,13 @@ where
                 message_id,
                 document,
             } => {
+                let record_key = document.record_key.clone();
                 let reject = |reason| {
                     Ok(Some(CultMeshRudpApplicationRejection {
                         session: key,
                         operation: CultMeshRudpApplicationOperation::DocumentPutRaw,
                         message_id: message_id.clone(),
+                        record_key: Some(record_key.clone()),
                         reason,
                     }))
                 };
@@ -758,6 +796,7 @@ where
                             session: key,
                             operation: CultMeshRudpApplicationOperation::SnapshotRequest,
                             message_id,
+                            record_key: None,
                             reason: CultMeshRudpRejectionReason::SnapshotSourceFailed(format!(
                                 "{error:#}"
                             )),
@@ -769,6 +808,7 @@ where
                         session: key,
                         operation: CultMeshRudpApplicationOperation::SnapshotRequest,
                         message_id,
+                        record_key: None,
                         reason: CultMeshRudpRejectionReason::SnapshotTooManyDocuments {
                             documents: documents.len(),
                             max_snapshot_documents: self.options.max_snapshot_documents,
@@ -786,6 +826,7 @@ where
                             session: key,
                             operation: CultMeshRudpApplicationOperation::SnapshotRequest,
                             message_id,
+                            record_key: None,
                             reason: CultMeshRudpRejectionReason::ResponseEncodingFailed(error),
                         }));
                     }
@@ -795,6 +836,7 @@ where
                         session: key,
                         operation: CultMeshRudpApplicationOperation::SnapshotRequest,
                         message_id,
+                        record_key: None,
                         reason: CultMeshRudpRejectionReason::SnapshotResponseTooLarge {
                             response_bytes: payload.len(),
                             max_snapshot_response_bytes: self.options.max_snapshot_response_bytes,
@@ -806,6 +848,7 @@ where
                         session: key,
                         operation: CultMeshRudpApplicationOperation::SnapshotRequest,
                         message_id,
+                        record_key: None,
                         reason: CultMeshRudpRejectionReason::PayloadBudgetFull,
                     }));
                 }
@@ -832,6 +875,7 @@ where
                             session: key,
                             operation: CultMeshRudpApplicationOperation::SnapshotRequest,
                             message_id,
+                            record_key: None,
                             reason: CultMeshRudpRejectionReason::ResponseQueueFailed(format!(
                                 "{error:#}"
                             )),
@@ -848,6 +892,7 @@ where
                             session: key,
                             operation: CultMeshRudpApplicationOperation::SnapshotRequest,
                             message_id,
+                            record_key: None,
                             reason: CultMeshRudpRejectionReason::ResponseSendFailed(code),
                         }));
                     }
@@ -875,7 +920,7 @@ where
         };
         let (refusal, mut unsendable) = match end {
             SessionEnd::Refused => (Vec::new(), None),
-            SessionEnd::Rejected(text) => (self.refusal_packets(&mut entry.session, text), None),
+            SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal), None),
             SessionEnd::Unsendable(error) => (Vec::new(), Some(error)),
         };
         entry.session.reset_peer_state();
@@ -902,14 +947,9 @@ where
     fn refusal_packets(
         &self,
         session: &mut CultNetRudpSession,
-        text: &str,
+        refusal: &CultNetMessage,
     ) -> Vec<CultNetRudpPacket> {
-        let refusal = CultNetMessage::Error {
-            error: text.into(),
-            code: None,
-            details: None,
-        };
-        encode_cultnet_message_to_vec(&refusal, CultNetWireContract::CultNetSchemaV0)
+        encode_cultnet_message_to_vec(refusal, CultNetWireContract::CultNetSchemaV0)
             .and_then(|payload| {
                 session.send_many(
                     "schema",
@@ -1080,3 +1120,26 @@ fn encode_snapshot_response(response: &CultNetMessage) -> std::result::Result<Ve
 fn duration_millis(duration: Duration) -> u64 {
     duration.as_millis().try_into().unwrap_or(u64::MAX)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // contracts/cultnet/interop/error-vectors.json, `write_not_authorized`: bytes captured from C# MessagePack
+    // (`ForRemoteWriteRefused(request for "some:key", "not granted")`) and pinned by C#, TS and Python too.
+    const WRITE_NOT_AUTHORIZED_HEX: &str = "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D92177726974655F6E6F745F617574686F72697A65643A206E6F74206772616E746564AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A8736F6D653A6B6579A461734F66C0A763757272656E74C0";
+
+    fn hex(bytes: &[u8]) -> String {
+        bytes.iter().map(|byte| format!("{byte:02X}")).collect()
+    }
+
+    #[test]
+    fn the_sink_refusal_is_the_pinned_write_not_authorized_bytes() {
+        let bytes = encode_cultnet_message_to_vec(
+            &write_refused_message("some:key", "not granted"),
+            CultNetWireContract::CultNetSchemaV0,
+        )
+        .unwrap();
+        assert_eq!(hex(&bytes), WRITE_NOT_AUTHORIZED_HEX);
+    }
+}

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -6,7 +6,7 @@ use cultmesh_rs::{
     CultMeshRudpSnapshotQuery, CultMeshRudpSnapshotSource,
 };
 use cultnet_rs::{
-    CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding,
+    CultNetErrorCode, CultNetErrorDetails, CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding,
     CultNetRudpReliableSendReceipt, CultNetRudpReliableSendStatus, CultNetRudpSocketMode,
     CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract,
     decode_cultnet_message_from_slice, encode_cultnet_message_to_vec,
@@ -343,6 +343,15 @@ fn refusal_seen_by(
     client: &mut CultNetRudpSocketTransportConnection,
     receipt: &CultNetRudpReliableSendReceipt,
 ) -> Result<String> {
+    let (error, code, details) = typed_refusal_seen_by(client, receipt)?;
+    assert_eq!((code, details), (None, None));
+    Ok(error)
+}
+
+fn typed_refusal_seen_by(
+    client: &mut CultNetRudpSocketTransportConnection,
+    receipt: &CultNetRudpReliableSendReceipt,
+) -> Result<(String, Option<CultNetErrorCode>, Option<CultNetErrorDetails>)> {
     let mut refusal = None;
     for _ in 0..2_000 {
         if let Some(frame) = client.receive_once()? {
@@ -362,7 +371,6 @@ fn refusal_seen_by(
     else {
         panic!("the refused peer was sent no refusal: {refusal:?}");
     };
-    assert_eq!((code, details), (None, None));
     assert!(
         !error.contains("CANARY"),
         "a sink's or source's text reached the peer: {error}"
@@ -384,7 +392,7 @@ fn refusal_seen_by(
         client.reliable_send_status(receipt),
         CultNetRudpReliableSendStatus::Invalidated
     );
-    Ok(error)
+    Ok((error, code, details))
 }
 
 /// Reads every datagram waiting at the server, such as a refused peer's
@@ -671,10 +679,21 @@ fn application_rejection_is_nonfatal_peer_scoped_refused_to_the_peer_and_unackno
     );
     assert_eq!(server.session_count(), 2);
     // The caller keeps the sink's text; the peer is sent fixed text, never it.
+    // A refused put is typed write_not_authorized and names the record; nothing of it reached the sink.
     assert_eq!(
-        refusal_seen_by(&mut publisher, &publish_receipt)?,
-        "the catalog refused the document"
+        typed_refusal_seen_by(&mut publisher, &publish_receipt)?,
+        (
+            "write_not_authorized: the catalog refused the document".to_string(),
+            Some(CultNetErrorCode::WriteNotAuthorized),
+            Some(CultNetErrorDetails {
+                field: Some("recordKey".into()),
+                value: Some(document("rejected", vec![0, 255, 1]).record_key),
+                as_of: None,
+                current: None,
+            })
+        )
     );
+    assert_eq!(sink.0.lock().unwrap().receipts.len(), 0, "a refused put changes nothing");
     drain_until_idle(&mut server)?;
 
     let snapshot_receipt = send_reliable(
@@ -1525,9 +1544,14 @@ fn a_refusal_reaches_a_peer_missing_an_earlier_packet() -> Result<()> {
     assert_eq!(
         refusals,
         vec![CultNetMessage::Error {
-            error: "the catalog refused the document".into(),
-            code: None,
-            details: None,
+            error: "write_not_authorized: the catalog refused the document".into(),
+            code: Some(CultNetErrorCode::WriteNotAuthorized),
+            details: Some(CultNetErrorDetails {
+                field: Some("recordKey".into()),
+                value: Some(document("refused", vec![4, 5, 6]).record_key),
+                as_of: None,
+                current: None,
+            }),
         }]
     );
     assert!(disconnected, "the refusal is followed by a goodbye");