diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 4a3ec060..fd671e95 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -68,11 +68,8 @@ where /// Caller-owned catalog port for serving raw snapshot requests. /// -/// The caller decides which records the requester may see. CultMesh serves +/// The caller decides which records the requester may see. CultMesh preserves /// 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, @@ -195,7 +192,7 @@ 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 served and under the shortest message id CultNet allows, would + /// alone, as received 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 @@ -570,12 +567,14 @@ where })) }; // Admit only what some snapshot request can return: the smallest - // response that could carry this document is the record as served, - // alone, under the shortest message id CultNet encodes, sized by the + // 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. let alone = CultNetMessage::SnapshotResponseRaw { message_id: SHORTEST_MESSAGE_ID.into(), - documents: vec![served_record(document.clone())], + documents: vec![document], }; let response_bytes = match encode_snapshot_response(&alone) { Ok(payload) => payload.len(), @@ -595,12 +594,15 @@ 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, + document: documents.remove(0), }; if let Err(error) = self.sink.accept_raw_document(receipt) { return reject(CultMeshRudpRejectionReason::SinkRefused(format!( @@ -646,7 +648,7 @@ where } let response = CultNetMessage::SnapshotResponseRaw { message_id: message_id.clone(), - documents: documents.into_iter().map(served_record).collect(), + documents, }; let payload = match encode_snapshot_response(&response) { Ok(payload) => payload, @@ -835,16 +837,6 @@ fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> { 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. diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index df4164fe..1aa24211 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -211,37 +211,14 @@ fn connect( anyhow::bail!("client connection timed out") } -/// 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 +/// The encoded size of the snapshot response that carries `document` alone +/// 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![as_served(document)], + documents: vec![document.clone()], }, CultNetWireContract::CultNetSchemaV0, )? @@ -287,8 +264,12 @@ fn poll_until_rejected_or_stored( receipts: usize, ) -> Result> { for _ in 0..5_000 { - if let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? { - return Ok(Some(rejection.reason)); + match server.poll_once()? { + CultMeshRudpPollOutcome::ApplicationRejected(rejection) => { + return Ok(Some(rejection.reason)); + } + CultMeshRudpPollOutcome::Idle => thread::sleep(Duration::from_millis(1)), + _ => {} } if sink.0.lock().unwrap().receipts.len() >= receipts { return Ok(None); @@ -675,8 +656,10 @@ fn snapshot_response_resends_until_acknowledged() -> Result<()> { fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> { let clock = Clock::new(55_000); let sink = Sink::default(); + // The message id is long enough that the put's frame is at least its + // served-alone response, so the snapshot limit below admits it. let message = CultNetMessage::DocumentPutRaw { - message_id: "budget".into(), + message_id: "budget-exact".into(), document: document("budget", vec![7; 64]), }; let encoded = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?; @@ -747,12 +730,10 @@ 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. The -/// bound is the record as served: provenance that makes the received record -/// larger than the bound neither refuses it nor reaches the snapshot. +/// 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 = with_provenance(document("fit", vec![7; 100])); + 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!( @@ -760,18 +741,6 @@ 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() @@ -845,11 +814,7 @@ 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()], - "the sink receives the record as it was sent" - ); + assert_eq!(stored, vec![at_bound.clone()]); source.0.lock().unwrap().documents = stored; send( @@ -879,7 +844,7 @@ 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![as_served(&at_bound)], + documents: vec![at_bound], } ); Ok(()) @@ -957,7 +922,7 @@ fn a_put_whose_response_would_overflow_the_reliable_queue_is_refused() -> Result decode_cultnet_message_from_slice(&served, CultNetWireContract::CultNetSchemaV0)?, CultNetMessage::SnapshotResponseRaw { message_id: "r".into(), - documents: vec![as_served(&at_bound)], + documents: vec![at_bound], } ); Ok(())