Add Rust RUDP schema message helpers

This commit is contained in:
Meta Vens 2026-06-17 04:29:22 +01:00
commit d131d0f42a
4 changed files with 80 additions and 4 deletions

View file

@ -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

View file

@ -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

View file

@ -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<Option<CultNetMessage>> {
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<u8>) -> Result<()> {
let packet = self.session.create_disconnect(reason);
self.send_packet(&packet)

View file

@ -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<CultNetMessage> {
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(())
}