diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 47348bfe..cbac63ff 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -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, 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) } diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index e43e4e59..1d18028f 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -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.