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 <noreply@anthropic.com>

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(())
     }
 

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<CultMeshRudpPollOutcome> {
         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)

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -335,6 +335,53 @@ fn poll_until_rejected_or_stored<Q: CultMeshRudpSnapshotSource>(
     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<String> {
+    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,