cultnet-rs, cultmesh-rs: remembered-write cap refusal is error.v1 limit_reached; pin authorize-once with a counting sink

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

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -437,6 +437,11 @@ impl CultMeshRudpApplicationRejection {
             (CultMeshRudpRejectionReason::MessageIdReused, _) => {
                 message_invalid_message("messageId", text)
             }
+            (CultMeshRudpRejectionReason::WriteLimitReached, _) => CultNetMessage::Error {
+                error: format!("limit_reached: {text}"),
+                code: Some(CultNetErrorCode::LimitReached),
+                details: Some(CultNetErrorDetails::default()),
+            },
             _ => CultNetMessage::Error {
                 error: text.into(),
                 code: None,

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -24,6 +24,8 @@ struct SinkState {
     failures_remaining: usize,
     /// Record keys the sink refuses to authorize, before and at the write.
     denied_keys: Vec<String>,
+    /// Every `authorize_raw_document` call, by message id.
+    authorized: Vec<String>,
 }
 
 #[derive(Clone, Default)]
@@ -47,6 +49,7 @@ impl Sink {
 
 impl CultMeshRudpRawDocumentSink for Sink {
     fn authorize_raw_document(&mut self, receipt: &CultMeshRudpRawDocumentReceipt) -> Result<()> {
+        self.0.lock().unwrap().authorized.push(receipt.message_id.clone());
         if self.0.lock().unwrap().denied_keys.contains(&receipt.document.record_key) {
             anyhow::bail!("not authorized");
         }
@@ -2328,9 +2331,9 @@ fn a_full_dedupe_table_refuses_new_writes_and_keeps_the_session_and_its_receipts
     send_reliable(&mut peer, &put("over", "note:over", vec![1]))?;
     send_reliable(&mut peer, &put("w0", "note:0", vec![1]))?;
     let answers = frames_after(&mut server, &mut peer, 2)?;
-    assert_eq!(refusal_of(&answers[0]), (None, Some("over".into())));
+    assert_eq!(refusal_of(&answers[0]), (Some(CultNetErrorCode::LimitReached), Some("over".into())));
     let CultNetMessage::ErrorV1 { error, .. } = &answers[0] else { unreachable!() };
-    assert_eq!(error, "the session's remembered-write limit is full", "the cap has its own fixed text");
+    assert_eq!(error, "limit_reached: the session's remembered-write limit is full", "the cap has its own fixed text");
     assert_eq!(answers[1], admitted("w0", "note:0"), "a remembered id is still answered idempotently");
     assert_eq!(server.session_count(), 1);
     Ok(())
@@ -2477,6 +2480,35 @@ fn a_refused_write_is_remembered_and_its_resend_replays_the_refusal() -> Result<
     Ok(())
 }
 
+/// The sink is asked once, on an id's first sighting: not again for a resend (admitted or refused), and
+/// not at all on a session that never declared the receipt capability. The sink here only counts, so a
+/// flag that stopped or inverted the ask cannot hide behind the sink refusing at the write.
+#[test]
+fn authorize_is_called_once_per_first_sighting_and_only_for_receipt_sessions() -> 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()?, 219)?;
+    let mut legacy = client(server.local_addr()?, 220)?;
+    connect(&mut server, &mut [&mut peer, &mut legacy])?;
+    send_reliable(&mut peer, &receipt_hello())?;
+    send_reliable(&mut peer, &put("m1", "note:1", vec![1]))?;
+    assert_eq!(answers_after(&mut server, &mut peer, 1)?, vec![admitted("m1", "note:1")]);
+    assert_eq!(sink.0.lock().unwrap().authorized, vec!["m1".to_string()], "first sighting asks once");
+    send_reliable(&mut peer, &put("m1", "note:1", vec![1]))?;
+    assert_eq!(frames_after(&mut server, &mut peer, 1)?, vec![admitted("m1", "note:1")]);
+    assert_eq!(sink.0.lock().unwrap().authorized, vec!["m1".to_string()], "the resend does not ask again");
+    send_reliable(&mut legacy, &put("l1", "note:9", vec![1]))?;
+    for _ in 0..40 {
+        server.poll_once()?;
+        legacy.poll_resends()?;
+        while legacy.receive_once()?.is_some() {}
+    }
+    let state = sink.0.lock().unwrap();
+    assert_eq!(state.receipts.len(), 2, "the legacy put was committed");
+    assert_eq!(state.authorized, vec!["m1".to_string()], "a legacy session is never asked");
+    Ok(())
+}
+
 #[test]
 fn a_delete_that_reuses_a_puts_id_is_message_invalid() -> Result<()> {
     let sink = Sink::default();

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")),
         }
     }
@@ -1067,6 +1072,9 @@ 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")?;
             let error = require_legacy_string(get("error"), "ErrorMessage.Error")?;
             if schema_version == "cultnet.error.v1" {
@@ -1134,6 +1142,9 @@ 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 {
         CultNetMessage::DocumentPutRaw {
             message_id,

packages/cultnet-rs/tests/write_receipt.rs

@@ -1,7 +1,7 @@
 //! Wire parity with contracts/cultnet/interop/write-receipt-vectors.json: the content hash, and the two new frames.
 
 use cultnet_rs::{
-    CultNetErrorCode, CultNetMessage, CultNetWireContract, decode_cultnet_message_from_slice,
+    CultNetErrorCode, CultNetErrorDetails, CultNetMessage, CultNetWireContract, decode_cultnet_message_from_slice,
     encode_cultnet_message_to_vec, write_content_hash,
 };
 
@@ -89,3 +89,48 @@ fn the_spec_wire_vectors_decode_to_their_expected_fields() {
     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!(hex(&bytes), 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()));
+}
+
+/// `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());
+}