From d465845516ae8894bc98d414432e7b303d105c03 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 17:42:44 +0200 Subject: [PATCH] cultmesh-rs: a rejected peer is told why, and nothing it sent is acknowledged Every application rejection, DocumentUnservable included, ended with self.sessions.remove(&key): no refusal and no goodbye. The peer could not tell a refusal from packet loss, and a publisher timed out and retried the same put. The session still ends, but the peer is now sent the refusal as a cultnet.error.v0 carrying the reason's text (no code or details, as the TypeScript server's refusals carry none), then the session's goodbye. The refusal is built before the reset and carries the goodbye's emptied acknowledgement fields, so it never acknowledges the refused message: publish_cultnet_message_to_rudp_catalog reads that acknowledgement as admission. A refusal that cannot be encoded or queued is not sent; the goodbye still is. publish_cultnet_message_to_rudp_catalog now returns a catalog's refusal as an error naming it, instead of waiting out its flush deadline. Tests read the refusal from the client side: the refusal text, the put still pending after it, and the goodbye invalidating it; and a publisher refused by a sink hears the sink's reason well inside its deadline. Co-Authored-By: Claude Opus 5.5 --- packages/cultmesh-rs/src/lib.rs | 36 ++++++++-- .../cultmesh-rs/src/rudp_document_server.rs | 48 ++++++++++++- .../cultmesh-rs/tests/rudp_document_server.rs | 72 +++++++++++++++---- 3 files changed, 132 insertions(+), 24 deletions(-) diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index 9c3e4d90..83b713ad 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -866,7 +866,8 @@ fn unix_millis() -> u128 { /// Publish one raw CultNet schema message to a RUDP catalog. /// /// Each call owns a fresh connection epoch and returns only after the catalog -/// acknowledges reliable application admission. +/// acknowledges reliable application admission. A catalog that refuses the +/// message answers with a `cultnet.error.v0`, returned as an error naming it. pub fn publish_cultnet_message_to_rudp_catalog( message: &cultnet_rs::CultNetMessage, options: CultMeshRudpDocumentPublishOptions, @@ -887,7 +888,22 @@ pub fn publish_cultnet_message_to_rudp_catalog( let receipt = client.send_reliable("schema", payload)?; let flush_deadline = Instant::now() + options.flush_timeout; while Instant::now() < flush_deadline { - let _ = client.receive_once()?; + if let Some(frame) = client.receive_once()? { + if frame.channel_id == "schema" { + if let Ok(cultnet_rs::CultNetMessage::Error { error, .. }) = + decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + ) + { + let _ = client.disconnect(b"document-refused".to_vec()); + anyhow::bail!( + "CultMesh RUDP catalog {} refused the message: {error}", + options.target + ); + } + } + } match client.reliable_send_status(&receipt) { cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => { let _ = client.disconnect(b"document-acknowledged".to_vec()); @@ -1280,7 +1296,7 @@ mod tests { } #[test] - fn document_publish_requires_application_acknowledgement() -> Result<()> { + fn a_refused_publish_fails_with_the_catalogs_reason_not_a_timeout() -> Result<()> { let temp = tempfile::tempdir()?; let node = CultMesh::create_node( temp.path().join("cultmesh.cc"), @@ -1308,6 +1324,7 @@ mod tests { Ok(()) }); + let started = Instant::now(); let error = node .publish_document_to_rudp_catalog( "note", @@ -1317,16 +1334,23 @@ mod tests { CultMeshRudpDocumentPublishOptions { target, connect_timeout: Duration::from_millis(100), - flush_timeout: Duration::from_millis(50), + flush_timeout: Duration::from_secs(10), poll_interval: Duration::from_millis(2), resend_delay_ms: 5, ..CultMeshRudpDocumentPublishOptions::default() }, ) - .expect_err("unacknowledged document must fail closed"); + .expect_err("a refused document must fail closed"); + let waited = started.elapsed(); stop.store(true, std::sync::atomic::Ordering::SeqCst); server.join().expect("server thread should not panic")?; - assert!(error.to_string().contains("reliable acknowledgement")); + // The publisher hears the refusal and its reason, long before its flush + // deadline: a refusal is not mistaken for loss. + assert!( + error.to_string().contains("refused the message: injected durable sink failure"), + "{error:#}" + ); + assert!(waited < Duration::from_secs(5), "waited {waited:?}"); Ok(()) } diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 24add376..4d9c8a3f 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -358,8 +358,10 @@ where /// Poll transport maintenance and receive at most one UDP datagram. /// /// Application rejection invalidates only the responsible peer session and - /// is returned as data so a daemon can log it and keep serving. Local - /// socket and server-state failures remain errors. + /// is returned as data so a daemon can log it and keep serving. The peer is + /// sent the refusal as a `cultnet.error.v0` carrying the reason's text, then + /// a goodbye; neither acknowledges the refused message. Local socket and + /// server-state failures remain errors. pub fn poll_once(&mut self) -> Result { self.maintain()?; let mut wire = vec![0_u8; MAX_UDP_DATAGRAM_BYTES]; @@ -458,7 +460,7 @@ where now_unix, now_monotonic, )? { - self.sessions.remove(&key); + self.end_rejected_session(key, rejection.reason.to_string())?; return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)); } } @@ -749,6 +751,46 @@ where self.send_packet(key.remote_addr, &goodbye) } + /// The server refused a peer's message: the session ends, and the peer is + /// sent the refusal before the goodbye, so it can tell a refusal from loss. + /// The refusal carries the goodbye's acknowledgement fields, which the reset + /// has emptied: it must not acknowledge the refused message, because a + /// publisher reads that acknowledgement as admission. A refusal that cannot + /// be encoded or queued is not sent; the goodbye still is. + fn end_rejected_session(&mut self, key: CultMeshRudpSessionKey, reason: String) -> Result<()> { + let Some(mut entry) = self.sessions.remove(&key) else { + return Ok(()); + }; + let refusal = CultNetMessage::Error { + error: reason, + code: None, + details: None, + }; + let packets = encode_cultnet_message_to_vec(&refusal, CultNetWireContract::CultNetSchemaV0) + .and_then(|payload| { + entry.session.send_many( + "schema", + payload, + CultNetRudpSendOptions { + reliable: true, + ordered: true, + sequenced: false, + now_ms: self.clock.now_monotonic_millis(), + reliable_expire_after_ms: None, + }, + Some(self.options.max_fragment_bytes), + ) + }) + .unwrap_or_default(); + let goodbye = entry.session.end_refused_session(); + for mut packet in packets { + packet.ack = goodbye.ack; + packet.ack_mask = goodbye.ack_mask; + self.send_packet(key.remote_addr, &packet)?; + } + 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) diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 1a726570..33a5df78 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -335,6 +335,53 @@ fn poll_until_rejected_or_stored( anyhow::bail!("the put was neither refused nor stored") } +/// What a refused peer is sent: a `cultnet.error.v0`, then the goodbye. The +/// refused message is never acknowledged, so its receipt stays pending until the +/// goodbye invalidates it. Returns the refusal's text. +fn refusal_seen_by( + client: &mut CultNetRudpSocketTransportConnection, + receipt: &CultNetRudpReliableSendReceipt, +) -> Result { + let mut refusal = None; + for _ in 0..2_000 { + if let Some(frame) = client.receive_once()? { + refusal = Some(decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?); + break; + } + thread::sleep(Duration::from_millis(1)); + } + let Some(CultNetMessage::Error { + error, + code, + details, + }) = refusal + else { + panic!("the refused peer was sent no refusal: {refusal:?}"); + }; + assert_eq!((code, details), (None, None)); + assert_eq!( + client.reliable_send_status(receipt), + CultNetRudpReliableSendStatus::Pending, + "a refusal must not acknowledge the refused message" + ); + for _ in 0..2_000 { + client.receive_once()?; + if !client.connected() { + break; + } + thread::sleep(Duration::from_millis(1)); + } + assert!(!client.connected(), "the refusal is followed by a goodbye"); + assert_eq!( + client.reliable_send_status(receipt), + CultNetRudpReliableSendStatus::Invalidated + ); + Ok(error) +} + fn send(client: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result<()> { client.send( "schema", @@ -562,7 +609,7 @@ fn a_restarted_client_starts_with_a_fresh_payload_budget() -> Result<()> { } #[test] -fn application_rejection_is_nonfatal_peer_scoped_and_unacknowledged() -> Result<()> { +fn application_rejection_is_nonfatal_peer_scoped_refused_to_the_peer_and_unacknowledged() -> Result<()> { let clock = Clock::new(48_000); let sink = Sink::fail_once(); let source = Source::fail_once(vec![document("snapshot", vec![4, 5, 6])]); @@ -602,10 +649,9 @@ fn application_rejection_is_nonfatal_peer_scoped_and_unacknowledged() -> Result< CultMeshRudpRejectionReason::SinkRefused("injected sink failure".into()) ); assert_eq!(server.session_count(), 2); - publisher.receive_once()?; assert_eq!( - publisher.reliable_send_status(&publish_receipt), - CultNetRudpReliableSendStatus::Pending + refusal_seen_by(&mut publisher, &publish_receipt)?, + "injected sink failure" ); let snapshot_receipt = send_reliable( @@ -630,10 +676,9 @@ fn application_rejection_is_nonfatal_peer_scoped_and_unacknowledged() -> Result< CultMeshRudpRejectionReason::SnapshotSourceFailed("injected source failure".into()) ); assert_eq!(server.session_count(), 1); - snapshot_client.receive_once()?; assert_eq!( - snapshot_client.reliable_send_status(&snapshot_receipt), - CultNetRudpReliableSendStatus::Pending + refusal_seen_by(&mut snapshot_client, &snapshot_receipt)?, + "injected source failure" ); let survivor_receipt = send_reliable( @@ -777,10 +822,9 @@ fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> { } if response_bytes > 128 )); assert_eq!(snapshot_server.session_count(), 0); - snapshot_client.receive_once()?; assert_eq!( - snapshot_client.reliable_send_status(&receipt), - CultNetRudpReliableSendStatus::Pending + refusal_seen_by(&mut snapshot_client, &receipt)?, + rejection.reason.to_string() ); Ok(()) } @@ -840,11 +884,9 @@ fn a_put_the_server_could_never_serve_is_refused_and_one_at_the_bound_is_served( 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 - ); + // The peer hears the refusal and why, rather than waiting on a put that + // will never be acknowledged. + assert_eq!(refusal_seen_by(&mut refused, &refused_receipt)?, sentence); send( &mut writer,