diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 72917040..c8cb034c 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -398,7 +398,10 @@ Current progress: first peer authorized by the authority lease catalog for a Verse role. The facade delegates to the existing Rust schema registry, shard catalog, and RUDP socket transport owners; discovery contact remains separate from trust, which - is still gated by the authority lease catalog. + is still gated by the authority lease catalog. Rust RUDP transports now also + send and receive typed `cultnet.schema.v0` messages on the reliable ordered + `schema` channel, so callers do not have to hand-encode MessagePack payloads + before using the shared RUDP pipe. - Rust now also has substrate-shaped shard catalog helpers: `CultNetShardDescriptor`, `CultNetShardCatalog`, request filtering by schema id and record key, response creation, response application, and built-in diff --git a/packages/cultnet-rs/README.md b/packages/cultnet-rs/README.md index a21e31ba..b46fc472 100644 --- a/packages/cultnet-rs/README.md +++ b/packages/cultnet-rs/README.md @@ -124,7 +124,8 @@ contact and authority ergonomics: ```rust use cultnet_rs::{ - CultMesh, CultMeshAuthorityLease, CultMeshPeerCard, CultMeshRudpSocketOptions, CultNetSchemaKind, + CultMesh, CultMeshAuthorityLease, CultMeshPeerCard, CultMeshRudpSocketOptions, CultNetMessage, + CultNetSchemaKind, }; use chrono::{Duration, Utc}; @@ -177,6 +178,12 @@ let mut client = CultMesh::create_rudp_client_for_authorized_peer( CultMeshRudpSocketOptions::default(), )?; client.connect(b"join".to_vec())?; +client.send_schema_message(&CultNetMessage::SchemaCatalogRequest { + message_id: "rust-schema-catalog".to_string(), + include_schema_json: Some(false), + schema_ids: None, + kinds: Some(vec![CultNetSchemaKind::DocumentPayload]), +})?; ``` For caller-owned reconnect loops, `CultNetRudpReconnectLoop` keeps transport diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 1fee0167..94f77e20 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -9,6 +9,7 @@ use std::net::UdpSocket; use std::time::SystemTime; use std::time::UNIX_EPOCH; +use crate::CultNetMessage; use crate::CultNetReconnectController; use crate::CultNetReconnectDecision; use crate::CultNetReconnectPolicy; @@ -20,7 +21,10 @@ use crate::CultNetTransportOrdering; use crate::CultNetTransportProfile; use crate::CultNetTransportProtocol; use crate::CultNetTransportStats; +use crate::CultNetWireContract; use crate::create_reconnect_policy; +use crate::decode_cultnet_message_from_slice; +use crate::encode_cultnet_message_to_vec; const RUDP_MAGIC: [u8; 4] = [0x43, 0x4e, 0x52, 0x30]; const RUDP_VERSION: u8 = 0; @@ -935,6 +939,27 @@ impl CultNetRudpSocketTransportConnection { Ok(()) } + pub fn send_schema_message(&mut self, message: &CultNetMessage) -> Result<()> { + let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; + self.send("schema", payload) + } + + pub fn receive_schema_message_once(&mut self) -> Result> { + let Some(frame) = self.receive_once()? else { + return Ok(None); + }; + if frame.channel_id != "schema" { + return Err(anyhow!( + "Expected RUDP schema frame, received channel {}", + frame.channel_id + )); + } + Ok(Some(decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?)) + } + pub fn disconnect(&mut self, reason: Vec) -> Result<()> { let packet = self.session.create_disconnect(reason); self.send_packet(&packet) diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 360c5f13..bb814e0f 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -100,6 +100,18 @@ fn receive_rudp_frame( anyhow::bail!("RUDP socket frame was not delivered") } +fn receive_rudp_schema_message( + transport: &mut CultNetRudpSocketTransportConnection, +) -> Result { + for _ in 0..20 { + if let Some(message) = transport.receive_schema_message_once()? { + return Ok(message); + } + thread::sleep(StdDuration::from_millis(5)); + } + anyhow::bail!("RUDP schema message was not delivered") +} + #[derive(Clone, Debug, PartialEq, DatabaseEntry)] #[cultcache( type = "ghostlight.agent-state", @@ -1077,12 +1089,41 @@ fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() let client_frame = receive_rudp_frame(&mut client)?; assert_eq!(client_frame.channel_id, "schema"); assert_eq!(client_frame.payload, b"server-state"); + + client.send_schema_message(&CultNetMessage::Hello { + runtime_id: "rust-rudp-client".to_string(), + runtime_kind: "rust-worker".to_string(), + agent_id: None, + role: Some("schema".to_string()), + display_name: None, + supported_document_types: None, + supported_mutation_contracts: None, + supported_message_versions: None, + transport_profiles: Some(vec![client.profile.clone()]), + supports_schema_catalog: Some(true), + })?; + let message = receive_rudp_schema_message(&mut server)?; + let CultNetMessage::Hello { + runtime_id, + transport_profiles, + supports_schema_catalog, + .. + } = message + else { + anyhow::bail!("RUDP schema message did not decode as hello"); + }; + assert_eq!(runtime_id, "rust-rudp-client"); + assert_eq!(supports_schema_catalog, Some(true)); + assert_eq!( + transport_profiles.expect("hello advertises transport")[0].transports[0].protocol, + CultNetTransportProtocol::Rudp + ); assert_eq!( server.profile.transports[0].protocol, CultNetTransportProtocol::Rudp ); - assert_eq!(client.stats().frames_sent, 1); - assert_eq!(server.stats().frames_received, 1); + assert_eq!(client.stats().frames_sent, 2); + assert_eq!(server.stats().frames_received, 2); Ok(()) }