packages/cultnet-rs/src/contracts.rs

@@ -249,6 +249,9 @@ pub enum CultNetErrorCode {
     WriteNotAuthorized,
     UnsupportedSchemaVersion,
     MessageInvalid,
+    /// `cultnet.error.v1` only: the session's remembered-write limit is full. `cultnet.error.v0` neither
+    /// encodes nor decodes it, so its enum stays the schema's.
+    LimitReached,
 }
 
 impl CultNetErrorCode {
@@ -263,6 +266,7 @@ impl CultNetErrorCode {
             Self::WriteNotAuthorized => "write_not_authorized",
             Self::UnsupportedSchemaVersion => "unsupported_schema_version",
             Self::MessageInvalid => "message_invalid",
+            Self::LimitReached => "limit_reached",
         }
     }
 
@@ -277,6 +281,7 @@ impl CultNetErrorCode {
             "write_not_authorized" => Ok(Self::WriteNotAuthorized),
             "unsupported_schema_version" => Ok(Self::UnsupportedSchemaVersion),
             "message_invalid" => Ok(Self::MessageInvalid),
+            "limit_reached" => Ok(Self::LimitReached),
             _ => Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code")),
         }
     }
@@ -359,6 +364,26 @@ pub enum CultNetMessage {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         details: Option<CultNetErrorDetails>,
     },
+    /// `cultnet.error.v1` (contracts/cultnet/interop/write-receipt.md): `cultnet.error.v0` plus the id of the
+    /// request it answers. Sent only to a connection that declared the write-receipt capability.
+    #[serde(skip_deserializing)]
+    #[serde(rename = "cultnet.error.v1", rename_all = "camelCase")]
+    ErrorV1 {
+        error: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        code: Option<CultNetErrorCode>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        details: Option<CultNetErrorDetails>,
+        message_id: Option<String>,
+    },
+    /// `cultnet.write_admitted.v0`: the receipt for one admitted put or delete, sent after its commit point.
+    #[serde(rename = "cultnet.write_admitted.v0", rename_all = "camelCase")]
+    WriteAdmitted {
+        message_id: String,
+        operation: String,
+        schema_id: String,
+        record_key: String,
+    },
     #[serde(rename = "cultnet.sample.change_name.v0", rename_all = "camelCase")]
     SampleChangeName { name: String },
     #[serde(rename = "cultnet.sample.chat.v0", rename_all = "camelCase")]
@@ -543,7 +568,9 @@ pub fn encode_cultnet_message_for_wire(
             | CultNetMessage::SnapshotResponseRawV1 { .. }
             | CultNetMessage::ContentChunkRequest { .. }
             | CultNetMessage::ContentChunkResponse { .. }
-            | CultNetMessage::Error { .. } => encode_raw_cultnet_schema_message(message),
+            | CultNetMessage::Error { .. }
+            | CultNetMessage::ErrorV1 { .. }
+            | CultNetMessage::WriteAdmitted { .. } => encode_raw_cultnet_schema_message(message),
             _ => Ok(cultcache_rs::decode_payload(&rmp_serde::to_vec(
                 &serde_json::to_value(message)?,
             )?)?),
@@ -573,7 +600,8 @@ pub fn decode_cultnet_message_from_slice(
                     | "cultnet.snapshot_response_raw.v0"
                     | "cultmesh.content_chunk_request.v1"
                     | "cultmesh.content_chunk_response.v1"
-                    | "cultnet.error.v0",
+                    | "cultnet.error.v0"
+                    | "cultnet.error.v1",
                 ) => parse_raw_cultnet_schema_message(&wire_value)
                     .with_context(|| received_version_label(&wire_value))?,
                 _ => cultcache_rs::decode_payload::<CultNetMessage>(bytes)
@@ -645,6 +673,20 @@ fn validate_message(message: &CultNetMessage) -> Result<()> {
             require_non_empty(session, "session")?;
         }
         CultNetMessage::Error { error, code: _, details: _ } => require_non_empty(error, "error")?,
+        CultNetMessage::ErrorV1 { error, message_id, .. } => {
+            require_non_empty(error, "error")?;
+            if let Some(message_id) = message_id {
+                require_non_empty(message_id, "messageId")?;
+            }
+        }
+        CultNetMessage::WriteAdmitted { message_id, operation, schema_id, record_key } => {
+            require_non_empty(message_id, "messageId")?;
+            if operation != "put" && operation != "delete" {
+                return Err(anyhow!("operation must be put or delete"));
+            }
+            require_non_empty(schema_id, "schemaId")?;
+            require_non_empty(record_key, "recordKey")?;
+        }
         CultNetMessage::SampleChangeName { name } => require_non_empty(name, "name")?,
         CultNetMessage::SampleChat { text } => require_non_empty(text, "text")?,
         CultNetMessage::DocumentDelete {
@@ -1021,7 +1063,7 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result<CultNetMessag
                 .to_vec(),
             error: require_legacy_string(get("error"), "error")?,
         }),
-        "cultnet.error.v0" => {
+        "cultnet.error.v0" | "cultnet.error.v1" => {
             let code = get("code")
                 .filter(|value| !value.is_nil())
                 .map(|value| {
@@ -1031,12 +1073,23 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result<CultNetMessag
                         .and_then(CultNetErrorCode::from_wire_str)
                 })
                 .transpose()?;
+            if schema_version == "cultnet.error.v0" && code == Some(CultNetErrorCode::LimitReached) {
+                return Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code"));
+            }
             let details = require_error_details(get("details"), "ErrorMessage.Details")?;
-            Ok(CultNetMessage::Error {
-                error: require_legacy_string(get("error"), "ErrorMessage.Error")?,
-                code,
-                details,
-            })
+            let error = require_legacy_string(get("error"), "ErrorMessage.Error")?;
+            if schema_version == "cultnet.error.v1" {
+                return Ok(CultNetMessage::ErrorV1 {
+                    error,
+                    code,
+                    details,
+                    message_id: match get("messageId") {
+                        None | Some(rmpv::Value::Nil) => None,
+                        other => Some(require_legacy_string(other, "messageId")?),
+                    },
+                });
+            }
+            Ok(CultNetMessage::Error { error, code, details })
         }
         _ => Err(anyhow!(
             "Unsupported raw cultnet.schema.v0 message"
@@ -1090,7 +1143,18 @@ fn require_legacy_optional_u64(
 }
 
 fn encode_raw_cultnet_schema_message(message: &CultNetMessage) -> Result<rmpv::Value> {
+    if let CultNetMessage::Error { code: Some(CultNetErrorCode::LimitReached), .. } = message {
+        return Err(anyhow!("limit_reached is a cultnet.error.v1 code; cultnet.error.v0 cannot carry it"));
+    }
     Ok(rmpv::Value::Map(match message {
+        // Spec field order (write-receipt.md), not serde_json's sorted map.
+        CultNetMessage::WriteAdmitted { message_id, operation, schema_id, record_key } => vec![
+            (rmpv::Value::from("schemaVersion"), rmpv::Value::from("cultnet.write_admitted.v0")),
+            (rmpv::Value::from("messageId"), rmpv::Value::from(message_id.as_str())),
+            (rmpv::Value::from("operation"), rmpv::Value::from(operation.as_str())),
+            (rmpv::Value::from("schemaId"), rmpv::Value::from(schema_id.as_str())),
+            (rmpv::Value::from("recordKey"), rmpv::Value::from(record_key.as_str())),
+        ],
         CultNetMessage::DocumentPutRaw {
             message_id,
             document,
@@ -1255,6 +1319,36 @@ fn encode_raw_cultnet_schema_message(message: &CultNetMessage) -> Result<rmpv::V
                     .unwrap_or(rmpv::Value::Nil),
             ),
         ],
+        CultNetMessage::ErrorV1 {
+            error,
+            code,
+            details,
+            message_id,
+        } => vec![
+            (rmpv::Value::from("schemaVersion"), rmpv::Value::from("cultnet.error.v1")),
+            (rmpv::Value::from("error"), rmpv::Value::from(error.as_str())),
+            (rmpv::Value::from("routingHint"), rmpv::Value::Nil),
+            (
+                rmpv::Value::from("code"),
+                code.as_ref()
+                    .map(|code| rmpv::Value::from(code.as_wire_str()))
+                    .unwrap_or(rmpv::Value::Nil),
+            ),
+            (
+                rmpv::Value::from("details"),
+                details
+                    .as_ref()
+                    .map(encode_error_details)
+                    .unwrap_or(rmpv::Value::Nil),
+            ),
+            (
+                rmpv::Value::from("messageId"),
+                message_id
+                    .as_deref()
+                    .map(rmpv::Value::from)
+                    .unwrap_or(rmpv::Value::Nil),
+            ),
+        ],
         _ => {
             return Err(anyhow!(
                 "Message is not a raw cultnet.schema.v0 binary replication message"
@@ -1884,3 +1978,71 @@ fn legacy_schema_descriptor(schema: &CultNetSchemaDescriptor) -> Result<rmpv::Va
         ),
     ]))
 }
+
+/// The content hash a write-receipt server binds to a `(actor, messageId)` dedupe key
+/// (contracts/cultnet/interop/write-receipt.md, rule 5): SHA-256 of the MessagePack array
+/// `[operation, schemaId, recordKey, payload]`, `payload` a `bin` for a put and nil for a delete.
+pub fn write_content_hash(
+    operation: &str,
+    schema_id: &str,
+    record_key: &str,
+    payload: Option<&[u8]>,
+) -> Result<[u8; 32]> {
+    use sha2::{Digest, Sha256};
+    let value = rmpv::Value::Array(vec![
+        rmpv::Value::from(operation),
+        rmpv::Value::from(schema_id),
+        rmpv::Value::from(record_key),
+        payload.map_or(rmpv::Value::Nil, |payload| rmpv::Value::Binary(payload.to_vec())),
+    ]);
+    let mut bytes = Vec::new();
+    rmpv::encode::write_value(&mut bytes, &value)?;
+    Ok(Sha256::digest(&bytes).into())
+}
+
+/// SHA-256 of a write's `messageId`: what a receipt server remembers in place of the id itself, so an
+/// entry's memory does not depend on the id's length (write-receipt.md, "What is remembered").
+pub fn message_id_digest(message_id: &str) -> [u8; 32] {
+    use sha2::{Digest, Sha256};
+    Sha256::digest(message_id.as_bytes()).into()
+}
+
+/// The `messageId` of a schema frame that does not decode as a message (write-receipt.md, rule 3: a refusal
+/// carries the write's id whenever the frame is readable enough to find it). `None` when the bytes are not
+/// a MessagePack map or hold no non-empty string `messageId`.
+pub fn readable_message_id(bytes: &[u8]) -> Option<String> {
+    let value = rmpv::decode::read_value(&mut &*bytes).ok()?;
+    let rmpv::Value::Map(entries) = value else {
+        return None;
+    };
+    entries.into_iter().find_map(|(key, value)| match (key.as_str(), value) {
+        (Some("messageId"), rmpv::Value::String(id)) => id.into_str().filter(|id| !id.is_empty()),
+        _ => 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,
+    }
+}

packages/cultnet-rs/src/rudp.rs

@@ -468,6 +468,14 @@ impl CultNetRudpSession {
         self.create_disconnect(RUDP_REFUSED_PACKET_REASON.to_vec())
     }
 
+    /// Ends a session because the server itself could not answer a write it owes an answer to (its answer
+    /// could not be encoded, queued or sent). The peer did nothing wrong: this goodbye is not the
+    /// "session refused a packet" of a protocol refusal.
+    pub fn end_local_failure_session(&mut self) -> CultNetRudpPacket {
+        self.reset_peer_state();
+        self.create_disconnect(RUDP_LOCAL_FAILURE_REASON.to_vec())
+    }
+
     /// Ends a session that owes its peer a packet that can never be sent as
     /// built (see `is_permanent_send_error`). Resending it would fail forever,
     /// and dropping it would leave a reliable sequence the peer waits on for
@@ -2609,6 +2617,8 @@ where
 /// Sent to a peer whose session refused one of its packets, and reported to
 /// the local caller as the disconnect reason.
 const RUDP_REFUSED_PACKET_REASON: &[u8] = b"session refused a packet";
+/// Goodbye reason of a session whose server could not answer a write (`end_local_failure_session`).
+const RUDP_LOCAL_FAILURE_REASON: &[u8] = b"server could not answer a write";
 const SESSION_TIMED_OUT_REASON: &[u8] = b"session timed out";
 /// Prefix of the goodbye reason for a session that owed a packet which can
 /// never be sent; the failure's fixed name (`send_error_code`) follows it.

packages/cultnet-rs/tests/write_receipt.rs

@@ -0,0 +1,155 @@
+//! Wire parity with contracts/cultnet/interop/write-receipt-vectors.json: the content hash, and the two new frames.
+
+use cultnet_rs::{
+    CultNetErrorCode, CultNetErrorDetails, CultNetMessage, CultNetWireContract, decode_cultnet_message_from_slice,
+    encode_cultnet_message_to_vec, write_content_hash,
+};
+
+fn hex(bytes: &[u8]) -> String {
+    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+#[test]
+fn the_content_hash_matches_the_spec_vectors() {
+    let a = [0x81, 0xA2, 0x69, 0x64, 0xA1, 0x31];
+    let b = [0x81, 0xA2, 0x69, 0x64, 0xA1, 0x32];
+    let cases = [
+        (write_content_hash("put", "note.v0", "note:1", Some(&a)), "7051c10071123b6ade50fbd5075e3859cae4ea70ce358a24a0b0c6980090e95d"),
+        (write_content_hash("put", "note.v0", "note:1", Some(&b)), "58d3fc6b858d2605a68550c9d4811ccdca6b9db7a0c3bfebf304d279257f6f9f"),
+        (write_content_hash("put", "note.v0", "note:2", Some(&a)), "a248cd920e58d41bc6ccca9e108030b1d491239000047fef088122364a32b6f8"),
+        (write_content_hash("delete", "note.v0", "note:1", None), "232b0f96651b28aa8ab7673a6106944a9f6a8d66852b6e21291a23ed084b8b37"),
+    ];
+    for (hash, expected) in cases {
+        assert_eq!(hex(&hash.unwrap()), expected);
+    }
+}
+
+#[test]
+fn the_new_frames_decode_back_equal() {
+    let admitted = CultNetMessage::WriteAdmitted {
+        message_id: "put-1".into(),
+        operation: "put".into(),
+        schema_id: "note.v0".into(),
+        record_key: "note:1".into(),
+    };
+    let bytes = encode_cultnet_message_to_vec(&admitted, CultNetWireContract::CultNetSchemaV0).unwrap();
+    assert_eq!(
+        decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0).unwrap(),
+        admitted
+    );
+    // error.v1 with a nil id round-trips as nil, and the code enum is v0's.
+    let error = CultNetMessage::ErrorV1 {
+        error: "message_invalid: unreadable frame".into(),
+        code: Some(CultNetErrorCode::MessageInvalid),
+        details: None,
+        message_id: None,
+    };
+    let bytes = encode_cultnet_message_to_vec(&error, CultNetWireContract::CultNetSchemaV0).unwrap();
+    assert_eq!(
+        decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0).unwrap(),
+        error
+    );
+}
+
+fn unhex(text: &str) -> Vec<u8> {
+    (0..text.len()).step_by(2).map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap()).collect()
+}
+
+/// The spec's wire vectors (write-receipt-vectors.json `wire`): the bytes decode to the vector's fields, not
+/// merely to something that encodes back to itself.
+#[test]
+fn the_spec_wire_vectors_decode_to_their_expected_fields() {
+    let admitted = decode_cultnet_message_from_slice(
+        &unhex("85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A57075742D31A96F7065726174696F6EA3707574A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"),
+        CultNetWireContract::CultNetSchemaV0,
+    )
+    .unwrap();
+    assert_eq!(
+        admitted,
+        CultNetMessage::WriteAdmitted {
+            message_id: "put-1".into(),
+            operation: "put".into(),
+            schema_id: "note.v0".into(),
+            record_key: "note:1".into(),
+        }
+    );
+    let refused = decode_cultnet_message_from_slice(
+        &unhex("86AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7631A56572726F72D93677726974655F6E6F745F617574686F72697A65643A2074686520636174616C6F6720726566757365642074686520646F63756D656E74AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A66E6F74653A32A461734F66C0A763757272656E74C0A96D6573736167654964A57075742D32"),
+        CultNetWireContract::CultNetSchemaV0,
+    )
+    .unwrap();
+    let CultNetMessage::ErrorV1 { error, code, details, message_id } = refused else {
+        panic!("expected error.v1");
+    };
+    assert_eq!(error, "write_not_authorized: the catalog refused the document");
+    assert_eq!(code, Some(CultNetErrorCode::WriteNotAuthorized));
+    assert_eq!(message_id.as_deref(), Some("put-2"));
+    let details = details.unwrap();
+    assert_eq!(details.field.as_deref(), Some("recordKey"));
+    assert_eq!(details.value.as_deref(), Some("note:2"));
+    assert_eq!((details.as_of, details.current), (None, None));
+}
+
+/// Spec vector `error_v1_limit_reached`: the bytes the server must send, and what they decode to.
+#[test]
+fn error_v1_limit_reached_is_the_spec_vector() {
+    let vector = "86AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7631A56572726F72D93B6C696D69745F726561636865643A207468652073657373696F6E27732072656D656D62657265642D7772697465206C696D69742069732066756C6CAB726F7574696E6748696E74C0A4636F6465AD6C696D69745F72656163686564A764657461696C7384A56669656C64C0A576616C7565C0A461734F66C0A763757272656E74C0A96D6573736167654964A87075742D31303235";
+    let message = CultNetMessage::ErrorV1 {
+        error: "limit_reached: the session's remembered-write limit is full".into(),
+        code: Some(CultNetErrorCode::LimitReached),
+        details: Some(CultNetErrorDetails::default()),
+        message_id: Some("put-1025".into()),
+    };
+    let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap();
+    assert_eq!(bytes, unhex(vector), "encoded bytes must equal the spec vector");
+    let CultNetMessage::ErrorV1 { error, code, details, message_id } =
+        decode_cultnet_message_from_slice(&unhex(vector), CultNetWireContract::CultNetSchemaV0).unwrap()
+    else {
+        panic!("expected error.v1");
+    };
+    assert_eq!(error, "limit_reached: the session's remembered-write limit is full");
+    assert_eq!(code, Some(CultNetErrorCode::LimitReached));
+    assert_eq!(message_id.as_deref(), Some("put-1025"));
+    assert_eq!(details, Some(CultNetErrorDetails::default()));
+}
+
+/// Spec vectors `write_admitted_put` and `write_admitted_delete`: the encoder emits the spec's field order.
+#[test]
+fn write_admitted_encodes_to_the_spec_vector_bytes() {
+    let cases = [
+        ("put-1", "put", "85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A57075742D31A96F7065726174696F6EA3707574A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"),
+        ("del-1", "delete", "85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A564656C2D31A96F7065726174696F6EA664656C657465A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"),
+    ];
+    for (id, operation, vector) in cases {
+        let message = CultNetMessage::WriteAdmitted {
+            message_id: id.into(),
+            operation: operation.into(),
+            schema_id: "note.v0".into(),
+            record_key: "note:1".into(),
+        };
+        let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap();
+        assert_eq!(bytes, unhex(vector));
+    }
+}
+
+/// `limit_reached` is in the error.v1 enum only: a v0 error neither encodes nor decodes it.
+#[test]
+fn error_v0_refuses_the_v1_only_limit_reached_code() {
+    let v0 = CultNetMessage::Error {
+        error: "x".into(),
+        code: Some(CultNetErrorCode::LimitReached),
+        details: None,
+    };
+    assert!(encode_cultnet_message_to_vec(&v0, CultNetWireContract::CultNetSchemaV0).is_err());
+    let v1 = CultNetMessage::ErrorV1 {
+        error: "x".into(),
+        code: Some(CultNetErrorCode::LimitReached),
+        details: None,
+        message_id: None,
+    };
+    let v1_bytes = encode_cultnet_message_to_vec(&v1, CultNetWireContract::CultNetSchemaV0).unwrap();
+    let at = v1_bytes.windows(16).position(|w| w == b"cultnet.error.v1").unwrap();
+    let mut as_v0 = v1_bytes.clone();
+    as_v0[at + 15] = b'0';
+    assert!(decode_cultnet_message_from_slice(&as_v0, CultNetWireContract::CultNetSchemaV0).is_err());
+}