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,