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: an application rejection carries a typed reason

Browse source 
CultMeshRudpApplicationRejection.reason was a String, so a consumer could
only log a refusal, never act on its cause. It is now
CultMeshRudpRejectionReason, one variant per refusal the document server
makes, with sizes and limits as fields. Display keeps the existing
sentences, so `{}` logging reads the same.

The snapshot path encodes its response through one helper, which put
admission will share.

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

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

@ -174,12 +174,67 @@ pub enum CultMeshRudpApplicationOperation {
SnapshotRequest,
}
/// Why the server refused an application message.
///
/// `Display` gives the operator-facing sentence; match on the variant to act on
/// the cause.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CultMeshRudpRejectionReason {
/// The caller's sink refused the document. Carries the sink's error text.
SinkRefused(String),
/// The caller's snapshot source failed. Carries its error text.
SnapshotSourceFailed(String),
SnapshotTooManyDocuments {
documents: usize,
max_snapshot_documents: usize,
},
SnapshotResponseTooLarge {
response_bytes: usize,
max_snapshot_response_bytes: usize,
},
/// A snapshot response could not be encoded. Carries the encoder's error text.
ResponseEncodingFailed(String),
/// The server's retained payload budget cannot hold the response.
PayloadBudgetFull,
/// The session could not queue the response. Carries the session's error text.
ResponseQueueFailed(String),
}
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::SnapshotTooManyDocuments {
documents,
max_snapshot_documents,
} => write!(
f,
"snapshot returned {documents} documents; limit is {max_snapshot_documents}"
),
Self::SnapshotResponseTooLarge {
response_bytes,
max_snapshot_response_bytes,
} => write!(
f,
"snapshot response is {response_bytes} bytes; limit is {max_snapshot_response_bytes}"
),
Self::ResponseEncodingFailed(error) => {
write!(f, "snapshot response could not be encoded: {error}")
}
Self::PayloadBudgetFull => f.write_str("retained payload budget is full"),
Self::ResponseQueueFailed(error) => {
write!(f, "snapshot response could not be queued: {error}")
}
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CultMeshRudpApplicationRejection {
pub session: CultMeshRudpSessionKey,
pub operation: CultMeshRudpApplicationOperation,
pub message_id: String,
pub reason: String,
pub reason: CultMeshRudpRejectionReason,
}
#[derive(Clone, Debug, PartialEq, Eq)]
@ -473,6 +528,14 @@ where
message_id,
document,
} => {
let reject = |reason| {
Ok(Some(CultMeshRudpApplicationRejection {
session: key,
operation: CultMeshRudpApplicationOperation::DocumentPutRaw,
message_id: message_id.clone(),
reason,
}))
};
let receipt = CultMeshRudpRawDocumentReceipt {
session: key,
message_id: message_id.clone(),
@ -481,12 +544,9 @@ where
document,
};
if let Err(error) = self.sink.accept_raw_document(receipt) {
return Ok(Some(CultMeshRudpApplicationRejection {
session: key,
operation: CultMeshRudpApplicationOperation::DocumentPutRaw,
message_id,
reason: format!("{error:#}"),
}));
return reject(CultMeshRudpRejectionReason::SinkRefused(format!(
"{error:#}"
)));
}
}
CultNetMessage::SnapshotRequest {
@ -508,7 +568,9 @@ where
session: key,
operation: CultMeshRudpApplicationOperation::SnapshotRequest,
message_id,
reason: format!("{error:#}"),
reason: CultMeshRudpRejectionReason::SnapshotSourceFailed(format!(
"{error:#}"
)),
}));
}
};
@ -517,28 +579,24 @@ where
session: key,
operation: CultMeshRudpApplicationOperation::SnapshotRequest,
message_id,
reason: format!(
"snapshot returned {} documents; limit is {}",
documents.len(),
self.options.max_snapshot_documents
),
reason: CultMeshRudpRejectionReason::SnapshotTooManyDocuments {
documents: documents.len(),
max_snapshot_documents: self.options.max_snapshot_documents,
},
}));
}
let response = CultNetMessage::SnapshotResponseRaw {
message_id: message_id.clone(),
documents,
};
let payload = match encode_cultnet_message_to_vec(
&response,
CultNetWireContract::CultNetSchemaV0,
) {
let payload = match encode_snapshot_response(&response) {
Ok(payload) => payload,
Err(error) => {
return Ok(Some(CultMeshRudpApplicationRejection {
session: key,
operation: CultMeshRudpApplicationOperation::SnapshotRequest,
message_id,
reason: format!("snapshot response could not be encoded: {error:#}"),
reason: CultMeshRudpRejectionReason::ResponseEncodingFailed(error),
}));
}
};
@ -547,11 +605,10 @@ where
session: key,
operation: CultMeshRudpApplicationOperation::SnapshotRequest,
message_id,
reason: format!(
"snapshot response is {} bytes; limit is {}",
payload.len(),
self.options.max_snapshot_response_bytes
),
reason: CultMeshRudpRejectionReason::SnapshotResponseTooLarge {
response_bytes: payload.len(),
max_snapshot_response_bytes: self.options.max_snapshot_response_bytes,
},
}));
}
if !self.payload_budget_allows(key, payload.len()) {
@ -559,7 +616,7 @@ where
session: key,
operation: CultMeshRudpApplicationOperation::SnapshotRequest,
message_id,
reason: "retained payload budget is full".into(),
reason: CultMeshRudpRejectionReason::PayloadBudgetFull,
}));
}
let payload_bytes = payload.len();
@ -585,7 +642,9 @@ where
session: key,
operation: CultMeshRudpApplicationOperation::SnapshotRequest,
message_id,
reason: format!("snapshot response could not be queued: {error:#}"),
reason: CultMeshRudpRejectionReason::ResponseQueueFailed(format!(
"{error:#}"
)),
}));
}
};
@ -686,6 +745,13 @@ fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> {
Ok(())
}
/// 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> {
encode_cultnet_message_to_vec(response, CultNetWireContract::CultNetSchemaV0)
.map_err(|error| format!("{error:#}"))
}
fn duration_millis(duration: Duration) -> u64 {
duration.as_millis().try_into().unwrap_or(u64::MAX)
}

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

@ -2,8 +2,8 @@ use anyhow::Result;
use cultmesh_rs::{
CultMeshRudpApplicationOperation, CultMeshRudpDocumentServer,
CultMeshRudpDocumentServerOptions, CultMeshRudpPollOutcome, CultMeshRudpRawDocumentReceipt,
CultMeshRudpRawDocumentSink, CultMeshRudpServerClock, CultMeshRudpSnapshotQuery,
CultMeshRudpSnapshotSource,
CultMeshRudpRawDocumentSink, CultMeshRudpRejectionReason, CultMeshRudpServerClock,
CultMeshRudpSnapshotQuery, CultMeshRudpSnapshotSource,
};
use cultnet_rs::{
CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding,
@ -299,7 +299,8 @@ fn same_connection_id_is_peer_scoped_and_raw_bytes_are_untouched() -> Result<()>
}
#[test]
fn a_second_connect_from_a_client_replaces_its_session_and_a_fresh_epoch_is_separate() -> Result<()> {
fn a_second_connect_from_a_client_replaces_its_session_and_a_fresh_epoch_is_separate() -> Result<()>
{
let clock = Clock::new(45_000);
let sink = Sink::default();
let mut server = server(Default::default(), clock, sink.clone(), Source::default())?;
@ -339,7 +340,12 @@ fn a_second_connect_from_a_client_replaces_its_session_and_a_fresh_epoch_is_sepa
#[test]
fn a_restarted_client_on_the_same_address_and_connection_id_is_admitted() -> Result<()> {
let sink = Sink::default();
let mut server = server(Default::default(), Clock::new(46_000), sink.clone(), Source::default())?;
let mut server = server(
Default::default(),
Clock::new(46_000),
sink.clone(),
Source::default(),
)?;
let target = server.local_addr()?;
let shared = UdpSocket::bind("127.0.0.1:0")?;
let twin = shared.try_clone()?;
@ -376,7 +382,10 @@ fn a_restarted_client_on_the_same_address_and_connection_id_is_admitted() -> Res
}
let receipts = sink.0.lock().unwrap();
assert!(
receipts.receipts.iter().any(|r| r.message_id == "after-restart"),
receipts
.receipts
.iter()
.any(|r| r.message_id == "after-restart"),
"the restarted client's frame was not delivered"
);
Ok(())
@ -390,11 +399,15 @@ fn a_restarted_client_starts_with_a_fresh_payload_budget() -> Result<()> {
message_id: id.into(),
document: document("budget", vec![7; 64]),
};
let encoded = encode_cultnet_message_to_vec(&put("put-a"), CultNetWireContract::CultNetSchemaV0)?;
let encoded =
encode_cultnet_message_to_vec(&put("put-a"), CultNetWireContract::CultNetSchemaV0)?;
let budget = encoded
.len()
.max(served_alone_bytes(&document("budget", vec![7; 64]))?);
let options = CultMeshRudpDocumentServerOptions {
max_admitted_payload_bytes: encoded.len() * 2,
max_admitted_payload_bytes_per_session: encoded.len(),
max_snapshot_response_bytes: encoded.len(),
max_admitted_payload_bytes: budget * 2,
max_admitted_payload_bytes_per_session: budget,
max_snapshot_response_bytes: budget,
..Default::default()
};
let sink = Sink::default();
@ -460,7 +473,10 @@ fn application_rejection_is_nonfatal_peer_scoped_and_unacknowledged() -> Result<
CultMeshRudpApplicationOperation::DocumentPutRaw
);
assert_eq!(rejection.message_id, "rejected-put");
assert!(rejection.reason.contains("injected sink failure"));
assert_eq!(
rejection.reason,
CultMeshRudpRejectionReason::SinkRefused("injected sink failure".into())
);
assert_eq!(server.session_count(), 2);
publisher.receive_once()?;
assert_eq!(
@ -485,7 +501,10 @@ fn application_rejection_is_nonfatal_peer_scoped_and_unacknowledged() -> Result<
CultMeshRudpApplicationOperation::SnapshotRequest
);
assert_eq!(rejection.message_id, "rejected-snapshot");
assert!(rejection.reason.contains("injected source failure"));
assert_eq!(
rejection.reason,
CultMeshRudpRejectionReason::SnapshotSourceFailed("injected source failure".into())
);
assert_eq!(server.session_count(), 1);
snapshot_client.receive_once()?;
assert_eq!(
@ -575,10 +594,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]))?);
let options = CultMeshRudpDocumentServerOptions {
max_admitted_payload_bytes: encoded.len(),
max_admitted_payload_bytes_per_session: encoded.len(),
max_snapshot_response_bytes: encoded.len(),
max_admitted_payload_bytes: budget,
max_admitted_payload_bytes_per_session: budget,
max_snapshot_response_bytes: budget,
..Default::default()
};
let mut budget_server = server(options, clock, sink.clone(), Source::default())?;
@ -617,7 +639,13 @@ fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> {
CultMeshRudpApplicationOperation::SnapshotRequest
);
assert_eq!(rejection.message_id, "too-large");
assert!(rejection.reason.contains("snapshot response is"));
assert!(matches!(
rejection.reason,
CultMeshRudpRejectionReason::SnapshotResponseTooLarge {
response_bytes,
max_snapshot_response_bytes: 128,
} if response_bytes > 128
));
assert_eq!(snapshot_server.session_count(), 0);
snapshot_client.receive_once()?;
assert_eq!(
@ -656,8 +684,7 @@ fn session_cap_and_expiry_use_monotonic_time() -> Result<()> {
#[test]
fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> {
use cultnet_rs::{
CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions,
encode_rudp_packet,
CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, encode_rudp_packet,
};
let clock = Clock::new(60_000);
let sink = Sink::default();
@ -698,7 +725,10 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> {
reliable: true,
..Default::default()
};
send_raw(&mut server, &unadmitted.send("schema", vec![1], reliable.clone())?)?;
send_raw(
&mut server,
&unadmitted.send("schema", vec![1], reliable.clone())?,
)?;
assert_eq!(server.packets_dropped(), 2);
// A moved flow re-sends Connect from a socket that never hears the Accept.