diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index cbac63ff..00dc37db 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -138,6 +138,9 @@ pub struct CultMeshRudpDocumentServerOptions { /// finite without duplicating its fragment/ordering machinery here. 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 + /// refused with `DocumentUnservable`, since no request could return it. pub max_snapshot_response_bytes: usize, pub max_snapshot_documents: usize, pub resend_delay: Duration, @@ -182,6 +185,13 @@ pub enum CultMeshRudpApplicationOperation { 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, with an empty message id, would exceed + /// `max_snapshot_response_bytes`. It was not offered to the sink. + DocumentUnservable { + response_bytes: usize, + max_snapshot_response_bytes: usize, + }, /// The caller's snapshot source failed. Carries its error text. SnapshotSourceFailed(String), SnapshotTooManyDocuments { @@ -204,6 +214,13 @@ 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::DocumentUnservable { + response_bytes, + max_snapshot_response_bytes, + } => write!( + f, + "document can never be served: its snapshot response is {response_bytes} bytes; limit is {max_snapshot_response_bytes}" + ), Self::SnapshotTooManyDocuments { documents, max_snapshot_documents, @@ -536,12 +553,34 @@ where reason, })) }; + // Admit only what some snapshot request can return: the smallest + // response that could carry this document is the document alone + // under an empty message id, sized by the snapshot path's encoder. + let alone = CultNetMessage::SnapshotResponseRaw { + message_id: String::new(), + documents: vec![document], + }; + let response_bytes = match encode_snapshot_response(&alone) { + Ok(payload) => payload.len(), + Err(error) => { + return reject(CultMeshRudpRejectionReason::ResponseEncodingFailed(error)); + } + }; + if response_bytes > self.options.max_snapshot_response_bytes { + return reject(CultMeshRudpRejectionReason::DocumentUnservable { + response_bytes, + max_snapshot_response_bytes: self.options.max_snapshot_response_bytes, + }); + } + 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!( diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 1d18028f..2e51cb2d 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -211,6 +211,19 @@ fn connect( anyhow::bail!("client connection timed out") } +/// The encoded size of the snapshot response that carries `document` alone +/// under an empty message id: the smallest response that could ever serve it. +fn served_alone_bytes(document: &CultNetRawDocumentRecord) -> Result { + Ok(encode_cultnet_message_to_vec( + &CultNetMessage::SnapshotResponseRaw { + message_id: String::new(), + documents: vec![document.clone()], + }, + CultNetWireContract::CultNetSchemaV0, + )? + .len()) +} + fn send(client: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result<()> { client.send( "schema", @@ -655,6 +668,126 @@ fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> { Ok(()) } +/// 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. +#[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 over_bound = document("fit", vec![7; 101]); + let limit = served_alone_bytes(&at_bound)?; + assert_eq!( + served_alone_bytes(&over_bound)?, + limit + 1, + "fixture: the two documents must straddle the bound by one byte" + ); + let options = CultMeshRudpDocumentServerOptions { + max_snapshot_response_bytes: limit, + ..Default::default() + }; + let sink = Sink::default(); + let source = Source::default(); + let mut server = server(options, Clock::new(60_000), sink.clone(), source.clone())?; + let target = server.local_addr()?; + + let mut refused = client(target, 121)?; + let mut writer = client(target, 122)?; + connect(&mut server, &mut [&mut refused, &mut writer])?; + let refused_receipt = send_reliable( + &mut refused, + &CultNetMessage::DocumentPutRaw { + message_id: "put-over".into(), + document: over_bound, + }, + )?; + let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else { + panic!("an unservable put must be refused as an application rejection"); + }; + assert_eq!(rejection.session.connection_id, 121); + assert_eq!( + rejection.operation, + CultMeshRudpApplicationOperation::DocumentPutRaw + ); + assert_eq!(rejection.message_id, "put-over"); + assert_eq!( + rejection.reason, + CultMeshRudpRejectionReason::DocumentUnservable { + response_bytes: limit + 1, + max_snapshot_response_bytes: limit, + } + ); + let sentence = rejection.reason.to_string(); + assert!(sentence.contains(&(limit + 1).to_string()) && sentence.contains(&limit.to_string())); + assert!(sink.0.lock().unwrap().receipts.is_empty()); + assert_eq!(server.session_count(), 1); + refused.receive_once()?; + assert_eq!( + refused.reliable_send_status(&refused_receipt), + CultNetRudpReliableSendStatus::Pending + ); + + send( + &mut writer, + &CultNetMessage::DocumentPutRaw { + message_id: "put-at".into(), + document: at_bound.clone(), + }, + )?; + for _ in 0..500 { + let outcome = server.poll_once()?; + assert!( + !matches!(outcome, CultMeshRudpPollOutcome::ApplicationRejected(_)), + "{outcome:?}" + ); + if !sink.0.lock().unwrap().receipts.is_empty() { + break; + } + thread::sleep(Duration::from_millis(1)); + } + let stored: Vec<_> = sink + .0 + .lock() + .unwrap() + .receipts + .iter() + .map(|receipt| receipt.document.clone()) + .collect(); + assert_eq!(stored, vec![at_bound.clone()]); + source.0.lock().unwrap().documents = stored; + + send( + &mut writer, + &CultNetMessage::SnapshotRequest { + message_id: String::new(), + schema_ids: None, + record_keys: None, + }, + )?; + let mut served = None; + for _ in 0..500 { + let outcome = server.poll_once()?; + assert!( + !matches!(outcome, CultMeshRudpPollOutcome::ApplicationRejected(_)), + "{outcome:?}" + ); + if let Some(frame) = writer.receive_once()? { + served = Some(frame.payload); + break; + } + thread::sleep(Duration::from_millis(1)); + } + let served = served.expect("the admitted document must be served"); + assert_eq!(served.len(), limit); + assert_eq!( + decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?, + CultNetMessage::SnapshotResponseRaw { + message_id: String::new(), + documents: vec![at_bound], + } + ); + Ok(()) +} + #[test] fn session_cap_and_expiry_use_monotonic_time() -> Result<()> { let clock = Clock::new(60_000);