diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 34794fc5..4a3ec060 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -11,6 +11,8 @@ use std::net::{SocketAddr, UdpSocket}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; const MAX_UDP_DATAGRAM_BYTES: usize = 65_535; +/// The most fragments one RUDP message can carry: its fragment count is a u16. +const MAX_FRAGMENTS_PER_MESSAGE: usize = u16::MAX as usize; /// The shortest message id CultNet encodes: one character. A snapshot request /// can carry no shorter id, so no response to it can be smaller. const SHORTEST_MESSAGE_ID: &str = "0"; @@ -66,8 +68,11 @@ where /// Caller-owned catalog port for serving raw snapshot requests. /// -/// The caller decides which records the requester may see. CultMesh preserves +/// The caller decides which records the requester may see. CultMesh serves /// 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, @@ -142,7 +147,8 @@ pub struct CultMeshRudpDocumentServerOptions { 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 + /// puts: a document whose snapshot response alone would exceed it, or would + /// take more fragments than one message or the reliable queue can hold, is /// refused with `DocumentUnservable`, since no request could return it. pub max_snapshot_response_bytes: usize, pub max_snapshot_documents: usize, @@ -189,11 +195,16 @@ 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, under the shortest message id CultNet allows, would exceed - /// `max_snapshot_response_bytes`. It was not offered to the sink. + /// alone, as served 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 + /// to the sink. DocumentUnservable { response_bytes: usize, max_snapshot_response_bytes: usize, + fragment_count: usize, + max_fragment_count: usize, }, /// The caller's snapshot source failed. Carries its error text. SnapshotSourceFailed(String), @@ -220,9 +231,11 @@ impl std::fmt::Display for CultMeshRudpRejectionReason { Self::DocumentUnservable { response_bytes, max_snapshot_response_bytes, + fragment_count, + max_fragment_count, } => write!( f, - "document can never be served: its snapshot response is {response_bytes} bytes; limit is {max_snapshot_response_bytes}" + "document can never be served: its snapshot response is {response_bytes} bytes in {fragment_count} fragments; limits are {max_snapshot_response_bytes} bytes and {max_fragment_count} fragments" ), Self::SnapshotTooManyDocuments { documents, @@ -557,11 +570,12 @@ where })) }; // Admit only what some snapshot request can return: the smallest - // response that could carry this document is the document alone - // under the shortest message id CultNet encodes, sized by the snapshot path's encoder. + // response that could carry this document is the record as served, + // alone, under the shortest message id CultNet encodes, sized by the + // snapshot path's encoder and fragmented as the snapshot path sends it. let alone = CultNetMessage::SnapshotResponseRaw { message_id: SHORTEST_MESSAGE_ID.into(), - documents: vec![document], + documents: vec![served_record(document.clone())], }; let response_bytes = match encode_snapshot_response(&alone) { Ok(payload) => payload.len(), @@ -569,21 +583,24 @@ where return reject(CultMeshRudpRejectionReason::ResponseEncodingFailed(error)); } }; - if response_bytes > self.options.max_snapshot_response_bytes { + let fragment_count = self.fragments_for(response_bytes); + let max_fragment_count = self.max_fragments_per_response(); + if response_bytes > self.options.max_snapshot_response_bytes + || fragment_count > max_fragment_count + { return reject(CultMeshRudpRejectionReason::DocumentUnservable { response_bytes, max_snapshot_response_bytes: self.options.max_snapshot_response_bytes, + fragment_count, + 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!( @@ -629,7 +646,7 @@ where } let response = CultNetMessage::SnapshotResponseRaw { message_id: message_id.clone(), - documents, + documents: documents.into_iter().map(served_record).collect(), }; let payload = match encode_snapshot_response(&response) { Ok(payload) => payload, @@ -711,6 +728,17 @@ where self.send_packet(key.remote_addr, &goodbye) } + /// The fragments `send_many` splits a response of `bytes` into. + fn fragments_for(&self, bytes: usize) -> usize { + bytes.div_ceil(self.options.max_fragment_bytes).max(1) + } + + /// The most fragments one response can take: one message's fragment count, + /// and the reliable queue it must fit into whole. + fn max_fragments_per_response(&self) -> usize { + MAX_FRAGMENTS_PER_MESSAGE.min(self.options.max_pending_reliable_packets_per_session) + } + fn payload_budget_allows(&self, key: CultMeshRudpSessionKey, bytes: usize) -> bool { let Some(session_bytes) = self .sessions @@ -784,9 +812,40 @@ fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> { if options.max_fragment_bytes == 0 { return Err(anyhow!("max_fragment_bytes must be greater than zero")); } + // The limits must leave room for the smallest response the server sends, an + // empty snapshot: a server that could answer nothing is misconfigured. + let empty = encode_snapshot_response(&CultNetMessage::SnapshotResponseRaw { + message_id: SHORTEST_MESSAGE_ID.into(), + documents: Vec::new(), + }) + .map_err(|error| anyhow!(error))? + .len(); + if empty > options.max_snapshot_response_bytes { + return Err(anyhow!( + "max_snapshot_response_bytes cannot hold an empty snapshot response" + )); + } + let max_fragments = + MAX_FRAGMENTS_PER_MESSAGE.min(options.max_pending_reliable_packets_per_session); + if empty.div_ceil(options.max_fragment_bytes) > max_fragments { + return Err(anyhow!( + "max_fragment_bytes and max_pending_reliable_packets_per_session cannot carry an empty snapshot response" + )); + } 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. fn encode_snapshot_response(response: &CultNetMessage) -> std::result::Result, String> { diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 1d338058..df4164fe 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -211,20 +211,92 @@ fn connect( anyhow::bail!("client connection timed out") } -/// The encoded size of the snapshot response that carries `document` alone -/// under a one-character message id, the shortest CultNet encodes: the +/// 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 /// smallest response that could ever serve it. fn served_alone_bytes(document: &CultNetRawDocumentRecord) -> Result { Ok(encode_cultnet_message_to_vec( &CultNetMessage::SnapshotResponseRaw { message_id: "r".into(), - documents: vec![document.clone()], + documents: vec![as_served(document)], }, CultNetWireContract::CultNetSchemaV0, )? .len()) } +/// A document keyed `key` whose served-alone response is exactly `bytes`. +fn document_served_at(key: &str, bytes: usize) -> Result { + let probe = bytes / 2; + let base = served_alone_bytes(&document(key, vec![7; probe]))?; + let fitted = document(key, vec![7; probe + bytes - base]); + assert_eq!( + served_alone_bytes(&fitted)?, + bytes, + "fixture: the served size is linear in the payload" + ); + Ok(fitted) +} + +/// Polls the server and client until the client receives a frame. +fn serve_until_frame( + server: &mut Server, + client: &mut CultNetRudpSocketTransportConnection, +) -> Result> { + for _ in 0..2_000 { + let outcome = server.poll_once()?; + assert!( + !matches!(outcome, CultMeshRudpPollOutcome::ApplicationRejected(_)), + "{outcome:?}" + ); + if let Some(frame) = client.receive_once()? { + return Ok(frame.payload); + } + thread::sleep(Duration::from_millis(1)); + } + anyhow::bail!("the client received no frame") +} + +/// Polls the server until it returns a rejection or the sink holds `receipts`. +fn poll_until_rejected_or_stored( + server: &mut Server, + sink: &Sink, + receipts: usize, +) -> Result> { + for _ in 0..5_000 { + if let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? { + return Ok(Some(rejection.reason)); + } + if sink.0.lock().unwrap().receipts.len() >= receipts { + return Ok(None); + } + } + anyhow::bail!("the put was neither refused nor stored") +} + fn send(client: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result<()> { client.send( "schema", @@ -608,9 +680,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]))?); + // The put's frame is exactly both budgets, so the first put is admitted only + // if a charge equal to a limit fits it. + let budget = encoded.len(); + assert!( + served_alone_bytes(&document("budget", vec![7; 64]))? <= budget, + "fixture: the snapshot limit must admit the put" + ); let options = CultMeshRudpDocumentServerOptions { max_admitted_payload_bytes: budget, max_admitted_payload_bytes_per_session: budget, @@ -671,10 +747,12 @@ 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 +/// bound is the record as served: provenance that makes the received record +/// larger than the bound neither refuses it nor reaches the snapshot. #[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 at_bound = with_provenance(document("fit", vec![7; 100])); let over_bound = document("fit", vec![7; 101]); let limit = served_alone_bytes(&at_bound)?; assert_eq!( @@ -682,6 +760,18 @@ 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() @@ -715,6 +805,8 @@ fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served( CultMeshRudpRejectionReason::DocumentUnservable { response_bytes: limit + 1, max_snapshot_response_bytes: limit, + fragment_count: 1, + max_fragment_count: 1024, } ); let sentence = rejection.reason.to_string(); @@ -753,7 +845,11 @@ 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()]); + assert_eq!( + stored, + vec![at_bound.clone()], + "the sink receives the record as it was sent" + ); source.0.lock().unwrap().documents = stored; send( @@ -783,12 +879,172 @@ 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![at_bound], + documents: vec![as_served(&at_bound)], } ); 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 +/// refused, and so is Soul's 2000-byte probe, each naming its sizes. +#[test] +fn a_put_whose_response_would_overflow_the_reliable_queue_is_refused() -> Result<()> { + let options = CultMeshRudpDocumentServerOptions { + max_fragment_bytes: 100, + max_pending_reliable_packets_per_session: 8, + max_snapshot_response_bytes: 8192, + ..Default::default() + }; + let sink = Sink::default(); + let source = Source::default(); + let mut server = server(options, Clock::new(61_000), sink.clone(), source.clone())?; + let target = server.local_addr()?; + let unservable = |response_bytes, fragment_count| { + Some(CultMeshRudpRejectionReason::DocumentUnservable { + response_bytes, + max_snapshot_response_bytes: 8192, + fragment_count, + max_fragment_count: 8, + }) + }; + + for (id, bytes, fragments) in [(131, 2000, 20), (132, 801, 9)] { + let mut refused = client(target, id)?; + connect(&mut server, &mut [&mut refused])?; + send( + &mut refused, + &CultNetMessage::DocumentPutRaw { + message_id: "put-over".into(), + document: document_served_at("queue", bytes)?, + }, + )?; + let reason = poll_until_rejected_or_stored(&mut server, &sink, 1)?; + assert_eq!(reason, unservable(bytes, fragments)); + let sentence = reason.unwrap().to_string(); + assert!( + sentence.contains(&format!("{bytes} bytes in {fragments} fragments")) + && sentence.contains("8 fragments"), + "{sentence}" + ); + } + assert!(sink.0.lock().unwrap().receipts.is_empty()); + + let at_bound = document_served_at("queue", 800)?; + let mut writer = client(target, 133)?; + 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, &sink, 1)?, None); + 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(), 800); + assert_eq!( + decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?, + CultNetMessage::SnapshotResponseRaw { + message_id: "r".into(), + documents: vec![as_served(&at_bound)], + } + ); + Ok(()) +} + +/// One message carries at most 65535 fragments, however deep the reliable +/// queue: a response of 65536 one-byte fragments is refused, one of 65535 is not. +#[test] +fn a_put_whose_response_needs_more_than_65535_fragments_is_refused() -> Result<()> { + let options = CultMeshRudpDocumentServerOptions { + max_fragment_bytes: 1, + max_pending_reliable_packets_per_session: 100_000, + max_snapshot_response_bytes: 70_000, + ..Default::default() + }; + let sink = Sink::default(); + let mut server = server(options, Clock::new(62_000), sink.clone(), Source::default())?; + let target = server.local_addr()?; + + let mut refused = client(target, 141)?; + connect(&mut server, &mut [&mut refused])?; + send( + &mut refused, + &CultNetMessage::DocumentPutRaw { + message_id: "put-over".into(), + document: document_served_at("wide", 65_536)?, + }, + )?; + assert_eq!( + poll_until_rejected_or_stored(&mut server, &sink, 1)?, + Some(CultMeshRudpRejectionReason::DocumentUnservable { + response_bytes: 65_536, + max_snapshot_response_bytes: 70_000, + fragment_count: 65_536, + max_fragment_count: 65_535, + }) + ); + + let mut writer = client(target, 142)?; + connect(&mut server, &mut [&mut writer])?; + send( + &mut writer, + &CultNetMessage::DocumentPutRaw { + message_id: "put-at".into(), + document: document_served_at("wide", 65_535)?, + }, + )?; + assert_eq!(poll_until_rejected_or_stored(&mut server, &sink, 1)?, None); + Ok(()) +} + +/// The limits must leave room for an empty snapshot response, the smallest the +/// server sends; exactly enough room is accepted. +#[test] +fn options_that_cannot_carry_an_empty_snapshot_response_are_refused() -> Result<()> { + let empty = encode_cultnet_message_to_vec( + &CultNetMessage::SnapshotResponseRaw { + message_id: "r".into(), + documents: Vec::new(), + }, + CultNetWireContract::CultNetSchemaV0, + )? + .len(); + let build = |options: CultMeshRudpDocumentServerOptions| { + server(options, Clock::new(63_000), Sink::default(), Source::default()).map(|_| ()) + }; + let bytes = |max_snapshot_response_bytes| CultMeshRudpDocumentServerOptions { + max_snapshot_response_bytes, + ..Default::default() + }; + let fragments = |max_pending_reliable_packets_per_session| CultMeshRudpDocumentServerOptions { + max_fragment_bytes: 1, + max_pending_reliable_packets_per_session, + ..Default::default() + }; + build(bytes(empty))?; + let error = build(bytes(empty - 1)).unwrap_err().to_string(); + assert!(error.contains("max_snapshot_response_bytes"), "{error}"); + build(fragments(empty))?; + let error = build(fragments(empty - 1)).unwrap_err().to_string(); + assert!( + error.contains("max_pending_reliable_packets_per_session"), + "{error}" + ); + Ok(()) +} + #[test] fn session_cap_and_expiry_use_monotonic_time() -> Result<()> { let clock = Clock::new(60_000);