GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultmesh-rs: admission sizes the record as served, in the fragments it is sent in

Browse source 
A served record carries no source provenance: the server now clears
source_runtime_id, source_agent_id, source_role and tags from every record
it sends in a snapshot response, as the TypeScript and Python servers
already serve only what their stores keep. Put admission sizes that served
record, so provenance that makes a put larger only in its received form no
longer refuses it; the sink still receives the record as it was sent.

Admission also counts the response's fragments: a put whose served-alone
response would take more fragments of max_fragment_bytes than
max_pending_reliable_packets_per_session or 65535 is refused with
DocumentUnservable, which now names the fragment count and its limit.
Construction refuses options that could not carry an empty snapshot
response, in bytes or in fragments.

The payload-budget fixture sets both budgets to exactly the put's frame, so
a charge equal to a limit must fit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:28:55 +02:00
commit 9118be879d
2 changed files with 339 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

87
packages/cultmesh-rs/src/rudp_document_server.rs
View file

@ -11,6 +11,8 @@ use std::net::{SocketAddr, UdpSocket};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
const MAX_UDP_DATAGRAM_BYTES: usize = 65_535;
/// The most fragments one RUDP message can carry: its fragment count is a u16.
const MAX_FRAGMENTS_PER_MESSAGE: usize = u16::MAX as usize;
/// The shortest message id CultNet encodes: one character. A snapshot request
/// can carry no shorter id, so no response to it can be smaller.
const SHORTEST_MESSAGE_ID: &str = "0";
@ -66,8 +68,11 @@ where
/// Caller-owned catalog port for serving raw snapshot requests.
///
/// The caller decides which records the requester may see. CultMesh preserves
/// The caller decides which records the requester may see. CultMesh serves
/// those records as raw CultNet documents and does not interpret their schemas.
/// A served record carries no source provenance: the server clears
/// `source_runtime_id`, `source_agent_id`, `source_role` and `tags` before it
/// sizes or sends a response, as every CultMesh runtime's server does.
pub trait CultMeshRudpSnapshotSource {
fn raw_snapshot(
&mut self,
@ -142,7 +147,8 @@ pub struct CultMeshRudpDocumentServerOptions {
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
/// puts: a document whose snapshot response alone would exceed it, or would
/// take more fragments than one message or the reliable queue can hold, is
/// refused with `DocumentUnservable`, since no request could return it.
pub max_snapshot_response_bytes: usize,
pub max_snapshot_documents: usize,
@ -189,11 +195,16 @@ 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, under the shortest message id CultNet allows, would exceed
/// `max_snapshot_response_bytes`. It was not offered to the sink.
/// alone, as served and under the shortest message id CultNet allows, would
/// exceed `max_snapshot_response_bytes`, or would take more than
/// `max_fragment_count` fragments of `max_fragment_bytes`: the lesser of
/// 65535 and `max_pending_reliable_packets_per_session`. It was not offered
/// to the sink.
DocumentUnservable {
response_bytes: usize,
max_snapshot_response_bytes: usize,
fragment_count: usize,
max_fragment_count: usize,
},
/// The caller's snapshot source failed. Carries its error text.
SnapshotSourceFailed(String),
@ -220,9 +231,11 @@ impl std::fmt::Display for CultMeshRudpRejectionReason {
Self::DocumentUnservable {
response_bytes,
max_snapshot_response_bytes,
fragment_count,
max_fragment_count,
} => write!(
f,
"document can never be served: its snapshot response is {response_bytes} bytes; limit is {max_snapshot_response_bytes}"
"document can never be served: its snapshot response is {response_bytes} bytes in {fragment_count} fragments; limits are {max_snapshot_response_bytes} bytes and {max_fragment_count} fragments"
),
Self::SnapshotTooManyDocuments {
documents,
@ -557,11 +570,12 @@ where
}))
};
// Admit only what some snapshot request can return: the smallest
// response that could carry this document is the document alone
// under the shortest message id CultNet encodes, sized by the snapshot path's encoder.
// response that could carry this document is the record as served,
// alone, under the shortest message id CultNet encodes, sized by the
// snapshot path's encoder and fragmented as the snapshot path sends it.
let alone = CultNetMessage::SnapshotResponseRaw {
message_id: SHORTEST_MESSAGE_ID.into(),
documents: vec![document],
documents: vec![served_record(document.clone())],
};
let response_bytes = match encode_snapshot_response(&alone) {
Ok(payload) => payload.len(),
@ -569,21 +583,24 @@ where
return reject(CultMeshRudpRejectionReason::ResponseEncodingFailed(error));
}
};
if response_bytes > self.options.max_snapshot_response_bytes {
let fragment_count = self.fragments_for(response_bytes);
let max_fragment_count = self.max_fragments_per_response();
if response_bytes > self.options.max_snapshot_response_bytes
|| fragment_count > max_fragment_count
{
return reject(CultMeshRudpRejectionReason::DocumentUnservable {
response_bytes,
max_snapshot_response_bytes: self.options.max_snapshot_response_bytes,
fragment_count,
max_fragment_count,
});
}
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: documents.remove(0),
document,
};
if let Err(error) = self.sink.accept_raw_document(receipt) {
return reject(CultMeshRudpRejectionReason::SinkRefused(format!(
@ -629,7 +646,7 @@ where
}
let response = CultNetMessage::SnapshotResponseRaw {
message_id: message_id.clone(),
documents,
documents: documents.into_iter().map(served_record).collect(),
};
let payload = match encode_snapshot_response(&response) {
Ok(payload) => payload,
@ -711,6 +728,17 @@ where
self.send_packet(key.remote_addr, &goodbye)
}
/// The fragments `send_many` splits a response of `bytes` into.
fn fragments_for(&self, bytes: usize) -> usize {
bytes.div_ceil(self.options.max_fragment_bytes).max(1)
}
/// The most fragments one response can take: one message's fragment count,
/// and the reliable queue it must fit into whole.
fn max_fragments_per_response(&self) -> usize {
MAX_FRAGMENTS_PER_MESSAGE.min(self.options.max_pending_reliable_packets_per_session)
}
fn payload_budget_allows(&self, key: CultMeshRudpSessionKey, bytes: usize) -> bool {
let Some(session_bytes) = self
.sessions
@ -784,9 +812,40 @@ fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> {
if options.max_fragment_bytes == 0 {
return Err(anyhow!("max_fragment_bytes must be greater than zero"));
}
// The limits must leave room for the smallest response the server sends, an
// empty snapshot: a server that could answer nothing is misconfigured.
let empty = encode_snapshot_response(&CultNetMessage::SnapshotResponseRaw {
message_id: SHORTEST_MESSAGE_ID.into(),
documents: Vec::new(),
})
.map_err(|error| anyhow!(error))?
.len();
if empty > options.max_snapshot_response_bytes {
return Err(anyhow!(
"max_snapshot_response_bytes cannot hold an empty snapshot response"
));
}
let max_fragments =
MAX_FRAGMENTS_PER_MESSAGE.min(options.max_pending_reliable_packets_per_session);
if empty.div_ceil(options.max_fragment_bytes) > max_fragments {
return Err(anyhow!(
"max_fragment_bytes and max_pending_reliable_packets_per_session cannot carry an empty snapshot response"
));
}
Ok(())
}
/// The record as the server serves it: without source provenance. What a peer
/// says about its own runtime, agent, role and tags is receipt data for the
/// sink, not part of the served document.
fn served_record(mut document: CultNetRawDocumentRecord) -> CultNetRawDocumentRecord {
document.source_runtime_id = None;
document.source_agent_id = None;
document.source_role = None;
document.tags = None;
document
}
/// The one encoder for a snapshot response, shared by the snapshot path and by
/// put admission so the size a put is judged by is the size it would be served at.
fn encode_snapshot_response(response: &CultNetMessage) -> std::result::Result<Vec<u8>, String> {

276
packages/cultmesh-rs/tests/rudp_document_server.rs
View file

@ -211,20 +211,92 @@ fn connect(
anyhow::bail!("client connection timed out")
}
/// The encoded size of the snapshot response that carries `document` alone
/// under a one-character message id, the shortest CultNet encodes: the
/// The record as a server serves it: the record the peer sent, without the
/// source provenance a peer reports about itself. Spelled out here rather than
/// asked of the server, so a server that served another shape is caught.
fn as_served(document: &CultNetRawDocumentRecord) -> CultNetRawDocumentRecord {
CultNetRawDocumentRecord {
source_runtime_id: None,
source_agent_id: None,
source_role: None,
tags: None,
..document.clone()
}
}
/// `document` with every source provenance field filled, so its received form
/// is larger than its served form.
fn with_provenance(mut document: CultNetRawDocumentRecord) -> CultNetRawDocumentRecord {
document.source_runtime_id = Some("provenance-runtime-that-is-not-served".into());
document.source_agent_id = Some("provenance-agent".into());
document.source_role = Some("provenance-role".into());
document.tags = Some(vec!["provenance".into(), "tags".into()]);
document
}
/// The encoded size of the snapshot response that carries `document` alone, as
/// served, under a one-character message id, the shortest CultNet encodes: 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: "r".into(),
documents: vec![document.clone()],
documents: vec![as_served(document)],
},
CultNetWireContract::CultNetSchemaV0,
)?
.len())
}
/// A document keyed `key` whose served-alone response is exactly `bytes`.
fn document_served_at(key: &str, bytes: usize) -> Result<CultNetRawDocumentRecord> {
let probe = bytes / 2;
let base = served_alone_bytes(&document(key, vec![7; probe]))?;
let fitted = document(key, vec![7; probe + bytes - base]);
assert_eq!(
served_alone_bytes(&fitted)?,
bytes,
"fixture: the served size is linear in the payload"
);
Ok(fitted)
}
/// Polls the server and client until the client receives a frame.
fn serve_until_frame(
server: &mut Server,
client: &mut CultNetRudpSocketTransportConnection,
) -> Result<Vec<u8>> {
for _ in 0..2_000 {
let outcome = server.poll_once()?;
assert!(
!matches!(outcome, CultMeshRudpPollOutcome::ApplicationRejected(_)),
"{outcome:?}"
);
if let Some(frame) = client.receive_once()? {
return Ok(frame.payload);
}
thread::sleep(Duration::from_millis(1));
}
anyhow::bail!("the client received no frame")
}
/// Polls the server until it returns a rejection or the sink holds `receipts`.
fn poll_until_rejected_or_stored(
server: &mut Server,
sink: &Sink,
receipts: usize,
) -> Result<Option<CultMeshRudpRejectionReason>> {
for _ in 0..5_000 {
if let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? {
return Ok(Some(rejection.reason));
}
if sink.0.lock().unwrap().receipts.len() >= receipts {
return Ok(None);
}
}
anyhow::bail!("the put was neither refused nor stored")
}
fn send(client: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result<()> {
client.send(
"schema",
@ -608,9 +680,13 @@ fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> {
document: document("budget", vec![7; 64]),
};
let encoded = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?;
let budget = encoded
.len()
.max(served_alone_bytes(&document("budget", vec![7; 64]))?);
// The put's frame is exactly both budgets, so the first put is admitted only
// if a charge equal to a limit fits it.
let budget = encoded.len();
assert!(
served_alone_bytes(&document("budget", vec![7; 64]))? <= budget,
"fixture: the snapshot limit must admit the put"
);
let options = CultMeshRudpDocumentServerOptions {
max_admitted_payload_bytes: budget,
max_admitted_payload_bytes_per_session: budget,
@ -671,10 +747,12 @@ fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> {
/// 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.
/// bound is admitted, and a snapshot then serves it at exactly that size. The
/// bound is the record as served: provenance that makes the received record
/// larger than the bound neither refuses it nor reaches the snapshot.
#[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 at_bound = with_provenance(document("fit", vec![7; 100]));
let over_bound = document("fit", vec![7; 101]);
let limit = served_alone_bytes(&at_bound)?;
assert_eq!(
@ -682,6 +760,18 @@ fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served(
limit + 1,
"fixture: the two documents must straddle the bound by one byte"
);
let received_alone = encode_cultnet_message_to_vec(
&CultNetMessage::SnapshotResponseRaw {
message_id: "r".into(),
documents: vec![at_bound.clone()],
},
CultNetWireContract::CultNetSchemaV0,
)?
.len();
assert!(
received_alone > limit + 1,
"fixture: the at-bound put is over the bound in its received form"
);
let options = CultMeshRudpDocumentServerOptions {
max_snapshot_response_bytes: limit,
..Default::default()
@ -715,6 +805,8 @@ fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served(
CultMeshRudpRejectionReason::DocumentUnservable {
response_bytes: limit + 1,
max_snapshot_response_bytes: limit,
fragment_count: 1,
max_fragment_count: 1024,
}
);
let sentence = rejection.reason.to_string();
@ -753,7 +845,11 @@ fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served(
.iter()
.map(|receipt| receipt.document.clone())
.collect();
assert_eq!(stored, vec![at_bound.clone()]);
assert_eq!(
stored,
vec![at_bound.clone()],
"the sink receives the record as it was sent"
);
source.0.lock().unwrap().documents = stored;
send(
@ -783,12 +879,172 @@ fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served(
decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?,
CultNetMessage::SnapshotResponseRaw {
message_id: "r".into(),
documents: vec![at_bound],
documents: vec![as_served(&at_bound)],
}
);
Ok(())
}
/// A put is admitted only if its served response fits the fragments one
/// response may take: here eight of 100 bytes, the reliable queue, although the
/// byte limit is far larger. Eight fragments are admitted and served; nine are
/// refused, and so is Soul's 2000-byte probe, each naming its sizes.
#[test]
fn a_put_whose_response_would_overflow_the_reliable_queue_is_refused() -> Result<()> {
let options = CultMeshRudpDocumentServerOptions {
max_fragment_bytes: 100,
max_pending_reliable_packets_per_session: 8,
max_snapshot_response_bytes: 8192,
..Default::default()
};
let sink = Sink::default();
let source = Source::default();
let mut server = server(options, Clock::new(61_000), sink.clone(), source.clone())?;
let target = server.local_addr()?;
let unservable = |response_bytes, fragment_count| {
Some(CultMeshRudpRejectionReason::DocumentUnservable {
response_bytes,
max_snapshot_response_bytes: 8192,
fragment_count,
max_fragment_count: 8,
})
};
for (id, bytes, fragments) in [(131, 2000, 20), (132, 801, 9)] {
let mut refused = client(target, id)?;
connect(&mut server, &mut [&mut refused])?;
send(
&mut refused,
&CultNetMessage::DocumentPutRaw {
message_id: "put-over".into(),
document: document_served_at("queue", bytes)?,
},
)?;
let reason = poll_until_rejected_or_stored(&mut server, &sink, 1)?;
assert_eq!(reason, unservable(bytes, fragments));
let sentence = reason.unwrap().to_string();
assert!(
sentence.contains(&format!("{bytes} bytes in {fragments} fragments"))
&& sentence.contains("8 fragments"),
"{sentence}"
);
}
assert!(sink.0.lock().unwrap().receipts.is_empty());
let at_bound = document_served_at("queue", 800)?;
let mut writer = client(target, 133)?;
connect(&mut server, &mut [&mut writer])?;
send(
&mut writer,
&CultNetMessage::DocumentPutRaw {
message_id: "put-at".into(),
document: at_bound.clone(),
},
)?;
assert_eq!(poll_until_rejected_or_stored(&mut server, &sink, 1)?, None);
source.0.lock().unwrap().documents = vec![at_bound.clone()];
send(
&mut writer,
&CultNetMessage::SnapshotRequest {
message_id: "r".into(),
schema_ids: None,
record_keys: None,
},
)?;
let served = serve_until_frame(&mut server, &mut writer)?;
assert_eq!(served.len(), 800);
assert_eq!(
decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?,
CultNetMessage::SnapshotResponseRaw {
message_id: "r".into(),
documents: vec![as_served(&at_bound)],
}
);
Ok(())
}
/// One message carries at most 65535 fragments, however deep the reliable
/// queue: a response of 65536 one-byte fragments is refused, one of 65535 is not.
#[test]
fn a_put_whose_response_needs_more_than_65535_fragments_is_refused() -> Result<()> {
let options = CultMeshRudpDocumentServerOptions {
max_fragment_bytes: 1,
max_pending_reliable_packets_per_session: 100_000,
max_snapshot_response_bytes: 70_000,
..Default::default()
};
let sink = Sink::default();
let mut server = server(options, Clock::new(62_000), sink.clone(), Source::default())?;
let target = server.local_addr()?;
let mut refused = client(target, 141)?;
connect(&mut server, &mut [&mut refused])?;
send(
&mut refused,
&CultNetMessage::DocumentPutRaw {
message_id: "put-over".into(),
document: document_served_at("wide", 65_536)?,
},
)?;
assert_eq!(
poll_until_rejected_or_stored(&mut server, &sink, 1)?,
Some(CultMeshRudpRejectionReason::DocumentUnservable {
response_bytes: 65_536,
max_snapshot_response_bytes: 70_000,
fragment_count: 65_536,
max_fragment_count: 65_535,
})
);
let mut writer = client(target, 142)?;
connect(&mut server, &mut [&mut writer])?;
send(
&mut writer,
&CultNetMessage::DocumentPutRaw {
message_id: "put-at".into(),
document: document_served_at("wide", 65_535)?,
},
)?;
assert_eq!(poll_until_rejected_or_stored(&mut server, &sink, 1)?, None);
Ok(())
}
/// The limits must leave room for an empty snapshot response, the smallest the
/// server sends; exactly enough room is accepted.
#[test]
fn options_that_cannot_carry_an_empty_snapshot_response_are_refused() -> Result<()> {
let empty = encode_cultnet_message_to_vec(
&CultNetMessage::SnapshotResponseRaw {
message_id: "r".into(),
documents: Vec::new(),
},
CultNetWireContract::CultNetSchemaV0,
)?
.len();
let build = |options: CultMeshRudpDocumentServerOptions| {
server(options, Clock::new(63_000), Sink::default(), Source::default()).map(|_| ())
};
let bytes = |max_snapshot_response_bytes| CultMeshRudpDocumentServerOptions {
max_snapshot_response_bytes,
..Default::default()
};
let fragments = |max_pending_reliable_packets_per_session| CultMeshRudpDocumentServerOptions {
max_fragment_bytes: 1,
max_pending_reliable_packets_per_session,
..Default::default()
};
build(bytes(empty))?;
let error = build(bytes(empty - 1)).unwrap_err().to_string();
assert!(error.contains("max_snapshot_response_bytes"), "{error}");
build(fragments(empty))?;
let error = build(fragments(empty - 1)).unwrap_err().to_string();
assert!(
error.contains("max_pending_reliable_packets_per_session"),
"{error}"
);
Ok(())
}
#[test]
fn session_cap_and_expiry_use_monotonic_time() -> Result<()> {
let clock = Clock::new(60_000);