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

cultmesh-rs: the snapshot source says what it would serve for a put

Browse source 
CultMeshRudpSnapshotSource gains served_record: the record the source
would serve for a received put's document. Admission sizes that record,
so a source that serves another shape than it receives (dropping
provenance, or adding its own, as Odin's presence records do) keeps
admission exact by overriding it. The default serves the record as
received, so existing sources, closures included, are unchanged. A
source that fails to answer refuses the put as SnapshotSourceFailed.

Tests: a stripping source admits a put that is over the bound only as
received and serves it at exactly the bound; the existing at-bound
fixture stays as the default-method test; a failing served_record
refuses the put.

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

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

@ -75,6 +75,20 @@ pub trait CultMeshRudpSnapshotSource {
&mut self,
query: &CultMeshRudpSnapshotQuery,
) -> Result<Vec<CultNetRawDocumentRecord>>;
/// The record this source would serve for a received put's `document`.
///
/// The server admits a put only if a snapshot response carrying this record
/// alone fits its limits, so a source that serves a different shape than it
/// receives (dropping provenance, or adding its own) overrides this to keep
/// admission exact. The default serves the record as received. An error
/// refuses the put as `SnapshotSourceFailed`.
fn served_record(
&mut self,
document: &CultNetRawDocumentRecord,
) -> Result<CultNetRawDocumentRecord> {
Ok(document.clone())
}
}
impl<F> CultMeshRudpSnapshotSource for F
@ -192,7 +206,8 @@ 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 received and under the shortest message id CultNet allows, would
/// alone, as the snapshot source would serve it (`served_record`) 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
@ -567,14 +582,21 @@ where
}))
};
// Admit only what some snapshot request can return: the smallest
// 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.
// response that could carry this document is the record the snapshot
// source would serve for it, alone, under the shortest message id
// CultNet encodes, sized by the snapshot path's encoder and fragmented
// as the snapshot path sends it.
let served = match self.snapshot_source.served_record(&document) {
Ok(served) => served,
Err(error) => {
return reject(CultMeshRudpRejectionReason::SnapshotSourceFailed(
format!("{error:#}"),
));
}
};
let alone = CultNetMessage::SnapshotResponseRaw {
message_id: SHORTEST_MESSAGE_ID.into(),
documents: vec![document],
documents: vec![served],
};
let response_bytes = match encode_snapshot_response(&alone) {
Ok(payload) => payload.len(),
@ -594,15 +616,12 @@ 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: documents.remove(0),
document,
};
if let Err(error) = self.sink.accept_raw_document(receipt) {
return reject(CultMeshRudpRejectionReason::SinkRefused(format!(

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

@ -53,6 +53,10 @@ struct SourceState {
documents: Vec<CultNetRawDocumentRecord>,
failures_remaining: usize,
calls: usize,
/// Serves records without source provenance, and says so to admission.
strips_provenance: bool,
/// Refuses to say what it would serve.
served_record_fails: bool,
}
#[derive(Clone, Default)]
@ -70,9 +74,27 @@ impl Source {
Self(Arc::new(Mutex::new(SourceState {
documents,
failures_remaining: 1,
calls: 0,
..Default::default()
})))
}
fn stripping() -> Self {
Self(Arc::new(Mutex::new(SourceState {
strips_provenance: true,
..Default::default()
})))
}
}
/// `document` without the source provenance a peer reports about itself.
fn stripped(document: &CultNetRawDocumentRecord) -> CultNetRawDocumentRecord {
CultNetRawDocumentRecord {
source_runtime_id: None,
source_agent_id: None,
source_role: None,
tags: None,
..document.clone()
}
}
impl CultMeshRudpSnapshotSource for Source {
@ -86,7 +108,27 @@ impl CultMeshRudpSnapshotSource for Source {
state.failures_remaining -= 1;
anyhow::bail!("injected source failure");
}
Ok(state.documents.clone())
let documents = state.documents.clone();
Ok(if state.strips_provenance {
documents.iter().map(stripped).collect()
} else {
documents
})
}
fn served_record(
&mut self,
document: &CultNetRawDocumentRecord,
) -> Result<CultNetRawDocumentRecord> {
let state = self.0.lock().unwrap();
if state.served_record_fails {
anyhow::bail!("injected served-record failure");
}
Ok(if state.strips_provenance {
stripped(document)
} else {
document.clone()
})
}
}
@ -734,7 +776,8 @@ 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
/// source keeps the default `served_record`, so the record is sized as received.
#[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]);
@ -854,6 +897,122 @@ fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served(
Ok(())
}
/// A source that serves records without provenance says so through
/// `served_record`, and admission sizes that: a put over the bound only in its
/// received form is admitted and served at exactly the bound.
#[test]
fn a_put_is_sized_as_its_source_would_serve_it() -> Result<()> {
let provenance = |mut document: CultNetRawDocumentRecord| {
document.source_agent_id = Some("provenance-agent-that-is-not-served".into());
document.source_role = Some("provenance-role".into());
document.tags = Some(vec!["provenance".into(), "tags".into()]);
document
};
let at_bound = provenance(document("fit", vec![7; 100]));
let over_bound = provenance(document("fit", vec![7; 101]));
let limit = served_alone_bytes(&stripped(&at_bound))?;
assert_eq!(
served_alone_bytes(&stripped(&over_bound))?,
limit + 1,
"fixture: the two documents must straddle the bound by one byte as served"
);
assert!(
served_alone_bytes(&at_bound)? > limit + 1,
"fixture: the at-bound put is over the bound as received"
);
let options = CultMeshRudpDocumentServerOptions {
max_snapshot_response_bytes: limit,
..Default::default()
};
let sink = Sink::default();
let source = Source::stripping();
let mut server = server(options, Clock::new(64_000), sink.clone(), source.clone())?;
let target = server.local_addr()?;
let mut refused = client(target, 151)?;
connect(&mut server, &mut [&mut refused])?;
send(
&mut refused,
&CultNetMessage::DocumentPutRaw {
message_id: "put-over".into(),
document: over_bound,
},
)?;
assert_eq!(
poll_until_rejected_or_stored(&mut server, &mut refused, &sink, 1)?,
Some(CultMeshRudpRejectionReason::DocumentUnservable {
response_bytes: limit + 1,
max_snapshot_response_bytes: limit,
fragment_count: 1,
max_fragment_count: 1024,
})
);
let mut writer = client(target, 152)?;
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, &mut writer, &sink, 1)?,
None
);
assert_eq!(
sink.0.lock().unwrap().receipts[0].document,
at_bound,
"the sink receives the record as it was sent"
);
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(), limit);
assert_eq!(
decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?,
CultNetMessage::SnapshotResponseRaw {
message_id: "r".into(),
documents: vec![stripped(&at_bound)],
}
);
Ok(())
}
/// A source that cannot say what it would serve refuses the put.
#[test]
fn a_put_whose_served_record_the_source_cannot_give_is_refused() -> Result<()> {
let sink = Sink::default();
let source = Source::default();
source.0.lock().unwrap().served_record_fails = true;
let mut server = server(Default::default(), Clock::new(65_000), sink.clone(), source)?;
let mut refused = client(server.local_addr()?, 161)?;
connect(&mut server, &mut [&mut refused])?;
send(
&mut refused,
&CultNetMessage::DocumentPutRaw {
message_id: "put".into(),
document: document("unsized", vec![1, 2, 3]),
},
)?;
assert_eq!(
poll_until_rejected_or_stored(&mut server, &mut refused, &sink, 1)?,
Some(CultMeshRudpRejectionReason::SnapshotSourceFailed(
"injected served-record failure".into()
))
);
assert!(sink.0.lock().unwrap().receipts.is_empty());
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