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: the snapshot source keeps owning the served record's provenance

Browse source 
The previous commit cleared source provenance from every served record.
Odin's snapshot source serves runtime-presence records with provenance it
authors itself (source_runtime_id, source_agent_id, source_role and an
odin-observed tag), and a crate test pins that a raw response is served as
the source returned it. The server does not own that shape, so it no
longer rewrites it, and admission again sizes the record as received.
Which rule a Rust server should size by is reported as a fork.

The fragment-count admission, the construction cross-check and the
exact payload-budget fixture stand. The budget fixture's put carries a
longer message id so its frame, the exact budget, is at least its
served-alone response. Polling for a refusal sleeps when the server is
idle, so a put of many fragments has time to arrive.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:37:34 +02:00
commit b1d7d31f59
2 changed files with 29 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -68,11 +68,8 @@ where
/// Caller-owned catalog port for serving raw snapshot requests.
///
/// The caller decides which records the requester may see. CultMesh serves
/// The caller decides which records the requester may see. CultMesh preserves
/// 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,
@ -195,7 +192,7 @@ 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, as served and under the shortest message id CultNet allows, would
/// alone, as received 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
@ -570,12 +567,14 @@ where
}))
};
// Admit only what some snapshot request can return: the smallest
// response that could carry this document is the record as served,
// alone, under the shortest message id CultNet encodes, sized by the
// response that could carry this document is the document alone, as
// received, under the shortest message id CultNet encodes, sized by the
// snapshot path's encoder and fragmented as the snapshot path sends it.
// The snapshot source owns the served record's shape, so the record
// as received is the server's measure of it.
let alone = CultNetMessage::SnapshotResponseRaw {
message_id: SHORTEST_MESSAGE_ID.into(),
documents: vec![served_record(document.clone())],
documents: vec![document],
};
let response_bytes = match encode_snapshot_response(&alone) {
Ok(payload) => payload.len(),
@ -595,12 +594,15 @@ where
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,
document: documents.remove(0),
};
if let Err(error) = self.sink.accept_raw_document(receipt) {
return reject(CultMeshRudpRejectionReason::SinkRefused(format!(
@ -646,7 +648,7 @@ where
}
let response = CultNetMessage::SnapshotResponseRaw {
message_id: message_id.clone(),
documents: documents.into_iter().map(served_record).collect(),
documents,
};
let payload = match encode_snapshot_response(&response) {
Ok(payload) => payload,
@ -835,16 +837,6 @@ fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> {
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.

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

@ -211,37 +211,14 @@ fn connect(
anyhow::bail!("client connection timed out")
}
/// 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
/// The encoded size of the snapshot response that carries `document` alone
/// 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![as_served(document)],
documents: vec![document.clone()],
},
CultNetWireContract::CultNetSchemaV0,
)?
@ -287,8 +264,12 @@ fn poll_until_rejected_or_stored(
receipts: usize,
) -> Result<Option<CultMeshRudpRejectionReason>> {
for _ in 0..5_000 {
if let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? {
return Ok(Some(rejection.reason));
match server.poll_once()? {
CultMeshRudpPollOutcome::ApplicationRejected(rejection) => {
return Ok(Some(rejection.reason));
}
CultMeshRudpPollOutcome::Idle => thread::sleep(Duration::from_millis(1)),
_ => {}
}
if sink.0.lock().unwrap().receipts.len() >= receipts {
return Ok(None);
@ -675,8 +656,10 @@ fn snapshot_response_resends_until_acknowledged() -> Result<()> {
fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> {
let clock = Clock::new(55_000);
let sink = Sink::default();
// The message id is long enough that the put's frame is at least its
// served-alone response, so the snapshot limit below admits it.
let message = CultNetMessage::DocumentPutRaw {
message_id: "budget".into(),
message_id: "budget-exact".into(),
document: document("budget", vec![7; 64]),
};
let encoded = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?;
@ -747,12 +730,10 @@ 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. The
/// bound is the record as served: provenance that makes the received record
/// larger than the bound neither refuses it nor reaches the snapshot.
/// 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 = with_provenance(document("fit", vec![7; 100]));
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!(
@ -760,18 +741,6 @@ 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()
@ -845,11 +814,7 @@ 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()],
"the sink receives the record as it was sent"
);
assert_eq!(stored, vec![at_bound.clone()]);
source.0.lock().unwrap().documents = stored;
send(
@ -879,7 +844,7 @@ 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![as_served(&at_bound)],
documents: vec![at_bound],
}
);
Ok(())
@ -957,7 +922,7 @@ fn a_put_whose_response_would_overflow_the_reliable_queue_is_refused() -> Result
decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?,
CultNetMessage::SnapshotResponseRaw {
message_id: "r".into(),
documents: vec![as_served(&at_bound)],
documents: vec![at_bound],
}
);
Ok(())