cultmesh-rs: refuse a put the document server could never serve

The server admitted a put up to the 4 MiB per-session budget but serves at
most max_snapshot_response_bytes (1 MiB by default), so it could accept a
document no snapshot request would ever return.

DocumentPutRaw now sizes the snapshot response that carries the document
alone under an empty message id, the smallest response that could serve
it, with the snapshot path's own encoder. Over max_snapshot_response_bytes
it is refused as DocumentUnservable, naming both sizes, before the sink
sees it. At the bound it is admitted and served at exactly that size.

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

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -138,6 +138,9 @@ pub struct CultMeshRudpDocumentServerOptions {
     /// finite without duplicating its fragment/ordering machinery here.
     pub max_admitted_payload_bytes: usize,
     pub max_admitted_payload_bytes_per_session: usize,
+    /// The largest encoded snapshot response the server sends. It also bounds
+    /// puts: a document whose snapshot response alone would exceed it is
+    /// refused with `DocumentUnservable`, since no request could return it.
     pub max_snapshot_response_bytes: usize,
     pub max_snapshot_documents: usize,
     pub resend_delay: Duration,
@@ -182,6 +185,13 @@ pub enum CultMeshRudpApplicationOperation {
 pub enum CultMeshRudpRejectionReason {
     /// The caller's sink refused the document. Carries the sink's error text.
     SinkRefused(String),
+    /// The document could never be served: a snapshot response carrying it
+    /// alone, with an empty message id, would exceed
+    /// `max_snapshot_response_bytes`. It was not offered to the sink.
+    DocumentUnservable {
+        response_bytes: usize,
+        max_snapshot_response_bytes: usize,
+    },
     /// The caller's snapshot source failed. Carries its error text.
     SnapshotSourceFailed(String),
     SnapshotTooManyDocuments {
@@ -204,6 +214,13 @@ impl std::fmt::Display for CultMeshRudpRejectionReason {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::SinkRefused(error) | Self::SnapshotSourceFailed(error) => f.write_str(error),
+            Self::DocumentUnservable {
+                response_bytes,
+                max_snapshot_response_bytes,
+            } => write!(
+                f,
+                "document can never be served: its snapshot response is {response_bytes} bytes; limit is {max_snapshot_response_bytes}"
+            ),
             Self::SnapshotTooManyDocuments {
                 documents,
                 max_snapshot_documents,
@@ -536,12 +553,34 @@ where
                         reason,
                     }))
                 };
+                // Admit only what some snapshot request can return: the smallest
+                // response that could carry this document is the document alone
+                // under an empty message id, sized by the snapshot path's encoder.
+                let alone = CultNetMessage::SnapshotResponseRaw {
+                    message_id: String::new(),
+                    documents: vec![document],
+                };
+                let response_bytes = match encode_snapshot_response(&alone) {
+                    Ok(payload) => payload.len(),
+                    Err(error) => {
+                        return reject(CultMeshRudpRejectionReason::ResponseEncodingFailed(error));
+                    }
+                };
+                if response_bytes > self.options.max_snapshot_response_bytes {
+                    return reject(CultMeshRudpRejectionReason::DocumentUnservable {
+                        response_bytes,
+                        max_snapshot_response_bytes: self.options.max_snapshot_response_bytes,
+                    });
+                }
+                let CultNetMessage::SnapshotResponseRaw { mut documents, .. } = alone else {
+                    unreachable!("constructed as a snapshot response above");
+                };
                 let receipt = CultMeshRudpRawDocumentReceipt {
                     session: key,
                     message_id: message_id.clone(),
                     transport_sequence,
                     received_at_unix_millis: now_unix,
-                    document,
+                    document: documents.remove(0),
                 };
                 if let Err(error) = self.sink.accept_raw_document(receipt) {
                     return reject(CultMeshRudpRejectionReason::SinkRefused(format!(

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -211,6 +211,19 @@ fn connect(
     anyhow::bail!("client connection timed out")
 }
 
+/// The encoded size of the snapshot response that carries `document` alone
+/// under an empty message id: the smallest response that could ever serve it.
+fn served_alone_bytes(document: &CultNetRawDocumentRecord) -> Result<usize> {
+    Ok(encode_cultnet_message_to_vec(
+        &CultNetMessage::SnapshotResponseRaw {
+            message_id: String::new(),
+            documents: vec![document.clone()],
+        },
+        CultNetWireContract::CultNetSchemaV0,
+    )?
+    .len())
+}
+
 fn send(client: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result<()> {
     client.send(
         "schema",
@@ -655,6 +668,126 @@ fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> {
     Ok(())
 }
 
+/// A put is admitted only if some snapshot request could return it. One byte
+/// over the served bound is refused before the sink sees it; exactly at the
+/// bound is admitted, and a snapshot then serves it at exactly that size.
+#[test]
+fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served() -> Result<()> {
+    let at_bound = document("fit", vec![7; 100]);
+    let over_bound = document("fit", vec![7; 101]);
+    let limit = served_alone_bytes(&at_bound)?;
+    assert_eq!(
+        served_alone_bytes(&over_bound)?,
+        limit + 1,
+        "fixture: the two documents must straddle the bound by one byte"
+    );
+    let options = CultMeshRudpDocumentServerOptions {
+        max_snapshot_response_bytes: limit,
+        ..Default::default()
+    };
+    let sink = Sink::default();
+    let source = Source::default();
+    let mut server = server(options, Clock::new(60_000), sink.clone(), source.clone())?;
+    let target = server.local_addr()?;
+
+    let mut refused = client(target, 121)?;
+    let mut writer = client(target, 122)?;
+    connect(&mut server, &mut [&mut refused, &mut writer])?;
+    let refused_receipt = send_reliable(
+        &mut refused,
+        &CultNetMessage::DocumentPutRaw {
+            message_id: "put-over".into(),
+            document: over_bound,
+        },
+    )?;
+    let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else {
+        panic!("an unservable put must be refused as an application rejection");
+    };
+    assert_eq!(rejection.session.connection_id, 121);
+    assert_eq!(
+        rejection.operation,
+        CultMeshRudpApplicationOperation::DocumentPutRaw
+    );
+    assert_eq!(rejection.message_id, "put-over");
+    assert_eq!(
+        rejection.reason,
+        CultMeshRudpRejectionReason::DocumentUnservable {
+            response_bytes: limit + 1,
+            max_snapshot_response_bytes: limit,
+        }
+    );
+    let sentence = rejection.reason.to_string();
+    assert!(sentence.contains(&(limit + 1).to_string()) && sentence.contains(&limit.to_string()));
+    assert!(sink.0.lock().unwrap().receipts.is_empty());
+    assert_eq!(server.session_count(), 1);
+    refused.receive_once()?;
+    assert_eq!(
+        refused.reliable_send_status(&refused_receipt),
+        CultNetRudpReliableSendStatus::Pending
+    );
+
+    send(
+        &mut writer,
+        &CultNetMessage::DocumentPutRaw {
+            message_id: "put-at".into(),
+            document: at_bound.clone(),
+        },
+    )?;
+    for _ in 0..500 {
+        let outcome = server.poll_once()?;
+        assert!(
+            !matches!(outcome, CultMeshRudpPollOutcome::ApplicationRejected(_)),
+            "{outcome:?}"
+        );
+        if !sink.0.lock().unwrap().receipts.is_empty() {
+            break;
+        }
+        thread::sleep(Duration::from_millis(1));
+    }
+    let stored: Vec<_> = sink
+        .0
+        .lock()
+        .unwrap()
+        .receipts
+        .iter()
+        .map(|receipt| receipt.document.clone())
+        .collect();
+    assert_eq!(stored, vec![at_bound.clone()]);
+    source.0.lock().unwrap().documents = stored;
+
+    send(
+        &mut writer,
+        &CultNetMessage::SnapshotRequest {
+            message_id: String::new(),
+            schema_ids: None,
+            record_keys: None,
+        },
+    )?;
+    let mut served = None;
+    for _ in 0..500 {
+        let outcome = server.poll_once()?;
+        assert!(
+            !matches!(outcome, CultMeshRudpPollOutcome::ApplicationRejected(_)),
+            "{outcome:?}"
+        );
+        if let Some(frame) = writer.receive_once()? {
+            served = Some(frame.payload);
+            break;
+        }
+        thread::sleep(Duration::from_millis(1));
+    }
+    let served = served.expect("the admitted document must be served");
+    assert_eq!(served.len(), limit);
+    assert_eq!(
+        decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?,
+        CultNetMessage::SnapshotResponseRaw {
+            message_id: String::new(),
+            documents: vec![at_bound],
+        }
+    );
+    Ok(())
+}
+
 #[test]
 fn session_cap_and_expiry_use_monotonic_time() -> Result<()> {
     let clock = Clock::new(60_000);