packages/cultmesh-rs/src/lib.rs

@@ -1451,6 +1451,126 @@ mod tests {
         Ok((target, stop, catalog))
     }
 
+    /// A catalog that acknowledges every reliable message it receives and refuses it with a typed
+    /// write_not_authorized, the acknowledgement either before the refusal or after it (the second packet
+    /// follows the first by 100 ms). The refusal carries no acknowledgement of its own.
+    fn acknowledging_refusing_catalog(
+        refusal_first: bool,
+    ) -> Result<(
+        SocketAddr,
+        std::sync::Arc<std::sync::atomic::AtomicBool>,
+        thread::JoinHandle<Result<()>>,
+    )> {
+        use cultnet_rs::{
+            CultNetRudpPacketType, CultNetRudpSendOptions, CultNetRudpSession,
+            CultNetRudpSessionOptions, decode_rudp_packet, encode_rudp_packet,
+        };
+        let socket = UdpSocket::bind("127.0.0.1:0")?;
+        socket.set_read_timeout(Some(Duration::from_millis(5)))?;
+        let target = socket.local_addr()?;
+        let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let catalog_stop = stop.clone();
+        let catalog = thread::spawn(move || -> Result<()> {
+            let mut session: Option<CultNetRudpSession> = None;
+            let mut wire = vec![0_u8; 65_535];
+            while !catalog_stop.load(std::sync::atomic::Ordering::SeqCst) {
+                let Ok((bytes, peer)) = socket.recv_from(&mut wire) else {
+                    continue;
+                };
+                let Ok(packet) = decode_rudp_packet(&wire[..bytes]) else {
+                    continue;
+                };
+                if packet.packet_type == CultNetRudpPacketType::Connect {
+                    let mut accepted = CultNetRudpSession::new(CultNetRudpSessionOptions {
+                        connection_id: packet.connection_id,
+                        initial_sequence: None,
+                        resend_delay_ms: 10_000,
+                        max_pending_reliable_packets: Some(64),
+                    });
+                    let accept = accepted.accept_connect(&packet, 0, Vec::new())?;
+                    socket.send_to(&encode_rudp_packet(&accept)?, peer)?;
+                    session = Some(accepted);
+                    continue;
+                }
+                let Some(session) = session.as_mut() else {
+                    continue;
+                };
+                let Ok(received) = session.receive(&packet, 1) else {
+                    continue;
+                };
+                if !packet.reliable || received.delivered.is_empty() {
+                    continue;
+                }
+                let ack = session.create_ack_for_received(packet.sequence);
+                let refusal = encode_cultnet_message_to_vec(
+                    &cultnet_rs::CultNetMessage::Error {
+                        error: "write_not_authorized: the catalog refused the document".into(),
+                        code: Some(cultnet_rs::CultNetErrorCode::WriteNotAuthorized),
+                        details: None,
+                    },
+                    CultNetWireContract::CultNetSchemaV0,
+                )?;
+                let options = CultNetRudpSendOptions {
+                    reliable: false,
+                    ordered: false,
+                    sequenced: false,
+                    now_ms: 2,
+                    reliable_expire_after_ms: None,
+                };
+                let mut refusals = session.send_many("schema", refusal, options, Some(1200))?;
+                for refusal in &mut refusals {
+                    refusal.ack = 0;
+                    refusal.ack_mask = 0;
+                }
+                let send_refusals = |socket: &UdpSocket| -> Result<()> {
+                    for refusal in &refusals {
+                        socket.send_to(&encode_rudp_packet(refusal)?, peer)?;
+                    }
+                    Ok(())
+                };
+                if refusal_first {
+                    send_refusals(&socket)?;
+                    thread::sleep(Duration::from_millis(100));
+                    socket.send_to(&encode_rudp_packet(&ack)?, peer)?;
+                } else {
+                    socket.send_to(&encode_rudp_packet(&ack)?, peer)?;
+                    thread::sleep(Duration::from_millis(100));
+                    send_refusals(&socket)?;
+                }
+            }
+            Ok(())
+        });
+        Ok((target, stop, catalog))
+    }
+
+    fn publish_options(target: SocketAddr) -> CultMeshRudpDocumentPublishOptions {
+        CultMeshRudpDocumentPublishOptions {
+            target,
+            connect_timeout: Duration::from_secs(2),
+            flush_timeout: Duration::from_secs(3),
+            poll_interval: Duration::from_millis(2),
+            resend_delay_ms: 5,
+            ..CultMeshRudpDocumentPublishOptions::default()
+        }
+    }
+
+    fn refused_note() -> cultnet_rs::CultNetMessage {
+        cultnet_rs::CultNetMessage::DocumentPutRaw {
+            message_id: "completion".into(),
+            document: cultnet_rs::CultNetRawDocumentRecord {
+                schema_id: "cultmesh.test.raw.v0".into(),
+                record_key: "tests:completion:a".into(),
+                stored_at: "2026-11-08T00:00:00Z".into(),
+                payload_encoding: cultnet_rs::CultNetRawPayloadEncoding::Messagepack,
+                payload: vec![0xc0],
+                source_runtime_id: None,
+                source_agent_id: None,
+                source_role: None,
+                tags: None,
+            },
+        }
+    }
+
     /// A catalog's refusal text is quoted only when it is one of the fixed
     /// refusals a CultMesh catalog sends. Any other text, which could carry
     /// anything, stays out of the publisher's error.
@@ -1901,4 +2021,18 @@ mod tests {
         assert_eq!(lease.bytes(), &[1, 2, 3]);
         Ok(())
     }
+
+    // Heimdall's invariant: no client treats a reliable-transport ack as write admission. The catalog here
+    // acknowledges the put and refuses it, the refusal arriving before or after the ack.
+    #[test]
+    fn a_publish_whose_refusal_arrives_before_the_ack_is_refused() -> Result<()> {
+        let (target, stop, catalog) = acknowledging_refusing_catalog(true)?;
+        let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target));
+        stop.store(true, std::sync::atomic::Ordering::SeqCst);
+        catalog.join().expect("catalog thread should not panic")?;
+        let error = result.expect_err("a refused put is never reported as admitted");
+        assert!(error.to_string().contains("refused the put"), "{error:#}");
+        Ok(())
+    }
+
 }