GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Add Rust interop RUDP serve dial path

Browse source
This commit is contained in:
Meta Vens 2026-06-17 10:07:37 +01:00
commit c80703172a
5 changed files with 578 additions and 142 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/cultnet-interop.yml vendored
View file

@ -93,7 +93,7 @@ jobs:
- name: Interop RUDP Rust lane
timeout-minutes: 8
run: node --test --test-name-pattern "TypeScript and Rust|Rust and TypeScript" packages/cultnet-ts/dist-test/test/interop/cultnet-interop.test.js
run: node --test --test-concurrency=1 --test-name-pattern "TypeScript and Rust|Rust and TypeScript" packages/cultnet-ts/dist-test/test/interop/cultnet-interop.test.js
- name: Interop RUDP C# lane
timeout-minutes: 8

7
docs/cultnet-transport-parity.md
View file

@ -30,7 +30,7 @@ CultNet owns cross-runtime transport semantics:
| C# `GameCult.Networking` | LiteNetLib UDP `NetManager` / `NetPeer`; sends legacy union messages and schema-v0 messages with `DeliveryMethod.ReliableOrdered`; single-peer RUDP socket transport exists in the library | LiteNetLib connection requests and app-level peer/catalog surfaces | Production LiteNetLib adapter profile plus UDP socket binding for the shared RUDP reliability owner |
| C# interop peer | TCP stream with 4-byte length-prefixed MessagePack frames | UDP multicast probe/announce | Test harness only |
| TypeScript `cultnet-ts` | `CultNetPeer` over any Node `Duplex`, TCP-framed transport, or single-peer RUDP socket transport; interop uses TCP | UDP multicast probe/announce in the interop peer | First UDP socket binding for the shared RUDP reliability owner |
| Rust `cultnet-rs` | Interop example uses TCP length-prefixed MessagePack frames; single-peer RUDP socket transport exists in the library | UDP multicast probe/announce | UDP socket binding for the shared RUDP reliability owner |
| Rust `cultnet-rs` | Interop example serves and dials schema-v0 MessagePack over TCP-framed or shared RUDP transport; single-peer RUDP socket transport exists in the library | UDP multicast probe/announce with TCP and RUDP transport profiles | UDP socket binding for the shared RUDP reliability owner |
| Python `cultcache-py` | TCP sockets with 4-byte length-prefixed MessagePack frames for local CultMesh/CultNet server and client; single-peer RUDP socket transport exists in the library | Endpoint lists and CultMesh peer/Verse catalogs | UDP socket binding for the shared RUDP reliability owner |
| Kotlin `cultmesh-kotlin` | Channel-aware WebSocket transport connection for reliable ordered `schema` frames; single-peer RUDP socket transport exists in the library | CultMesh Verse and peer catalogs, plus endpoint lists carried in schema-v0 catalog messages | WebSocket adapter plus UDP socket binding for the shared RUDP reliability owner, with build-script self-test |
@ -180,8 +180,9 @@ Current progress:
side-channel for Python's schema lane.
- Rust now has the same `tcp_framed` transport profile helper plus a
`TcpFramedTransportConnection` with schema-channel `send`, `receive`, and
transfer stats. The Rust interop peer advertises and uses that shared port
instead of owning raw TCP frame I/O directly.
transfer stats. The Rust interop peer advertises TCP and RUDP profiles and
serves/dials schema-v0 through the shared RUDP path without giving RUDP its
own document semantics.
- Kotlin now has a `websocket` transport profile helper plus
`CultNetWebSocketTransportConnection`, which wraps the older binary
WebSocket client behind the same `CultNetTransportFrame` and stats shape for

2
docs/runtime-parity-scope.md
View file

@ -33,7 +33,7 @@ CultMesh.
| C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, schema-client transport port for service request/response clients, CultMesh node/session/catalog/stream/simulation/RUDP surfaces, authorized peer lookup | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption |
| Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup, shard catalogs/logs, schema-transport factory for raw client request/subscription flows, local TCP and RUDP CultMesh server listeners, Python interop RUDP serve/dial path, session/prediction/simulation fact helpers, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe |
| TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, advertised schema/shard catalog wire contracts plus request/sync helpers, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP interop serve/dial path, RUDP reconnect policy/controller, RUDP reconnect-loop helper, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup | Full C# game session/server stack |
| Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, schema and shard catalog helpers, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for schema registry, shard catalog, peer cards, authority leases, authorized peer lookup, RUDP endpoint parsing, and peer-card/authorized-peer RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server |
| Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, schema and shard catalog helpers, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, Rust interop RUDP serve/dial path, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for schema registry, shard catalog, peer cards, authority leases, authorized peer lookup, RUDP endpoint parsing, and peer-card/authorized-peer RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server |
| Kotlin | JVM/Android client substrate | Typed MessagePack cache, built-in schema catalog advertisement, schema/shard catalogs with shared schema-message transport fetch/sync helpers, authority leases, authorized peer lookup, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout |
## Evidence Gates

615
packages/cultnet-rs/examples/cultnet_interop_peer.rs
View file

@ -11,6 +11,10 @@ use cultnet_rs::CultNetDocumentPutOptions;
use cultnet_rs::CultNetDocumentRegistry;
use cultnet_rs::CultNetMessage;
use cultnet_rs::CultNetMutationAuthority;
use cultnet_rs::CultNetRudpPacketType;
use cultnet_rs::CultNetRudpSendOptions;
use cultnet_rs::CultNetRudpSession;
use cultnet_rs::CultNetRudpSessionOptions;
use cultnet_rs::CultNetRudpSocketTransportConnection;
use cultnet_rs::CultNetRudpSocketTransportOptions;
use cultnet_rs::CultNetSchemaKind;
@ -25,7 +29,9 @@ use cultnet_rs::builtin_schema_registry;
use cultnet_rs::create_rudp_transport_profile;
use cultnet_rs::create_tcp_framed_transport_profile;
use cultnet_rs::decode_cultnet_message_from_slice;
use cultnet_rs::decode_rudp_packet;
use cultnet_rs::encode_cultnet_message_to_vec;
use cultnet_rs::encode_rudp_packet;
use serde::Deserialize;
use serde::Serialize;
use socket2::Domain;
@ -61,6 +67,10 @@ const FIRE_RECEIPT_TYPE: &str = "cultnet.interop-fire-weapon-receipt";
const FIRE_RECEIPT_SCHEMA_ID: &str = "https://github.com/GameCult/cultnet-ts/integration/contracts/cultnet.interop-fire-weapon-receipt.schema.json";
const FIRE_RECEIPT_SCHEMA_VERSION: &str = "cultnet.interop_fire_weapon_receipt.v0";
const DISCOVERY_ANNOUNCE_SCHEMA_VERSION: &str = "cultnet.discovery_announce.v0";
const RUDP_INTEROP_CONNECTION_ID: u32 = 0x43554C54;
const RUDP_INTEROP_MAX_FRAGMENT_BYTES: usize = 1024;
const RUDP_INTEROP_RESEND_DELAY_MS: u64 = 25;
const RUDP_INTEROP_READ_TIMEOUT: Duration = Duration::from_secs(15);
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "cultnet.interop-note", schema = "CultNetInteropNote")]
@ -189,6 +199,7 @@ struct PeerConfig {
bind_host: String,
advertise_host: String,
tcp_port: u16,
rudp_port: u16,
discovery_port: u16,
discovery_group: Ipv4Addr,
schema_path: String,
@ -201,10 +212,19 @@ struct DialConfig {
display_name: String,
agent_id: String,
target_host: String,
target_port: u16,
target_port: Option<u16>,
target_rudp_port: Option<u16>,
schema_path: String,
}
trait SchemaMessageSender {
fn send_schema_message(&mut self, message: &CultNetMessage) -> Result<()>;
}
trait SchemaMessageTransport: SchemaMessageSender {
fn read_schema_message(&mut self) -> Result<CultNetMessage>;
}
fn main() -> Result<()> {
let mut args = std::env::args().skip(1);
let mode = args.next().ok_or_else(|| {
@ -250,7 +270,13 @@ fn serve(config: PeerConfig) -> Result<()> {
let config = Arc::new(config);
start_udp_discovery_server(config.clone())?;
start_tcp_server(config.clone(), cache, document_registry, schema_registry)?;
start_tcp_server(
config.clone(),
cache.clone(),
document_registry.clone(),
schema_registry.clone(),
)?;
start_rudp_server(config.clone(), cache, document_registry, schema_registry)?;
print_json(&serde_json::json!({
"status": "ready",
@ -258,6 +284,7 @@ fn serve(config: PeerConfig) -> Result<()> {
"runtimeId": config.runtime_id,
"runtimeKind": config.runtime_kind,
"tcpPort": config.tcp_port,
"rudpPort": config.rudp_port,
"discoveryPort": config.discovery_port,
"discoveryGroup": config.discovery_group.to_string(),
}))?;
@ -359,50 +386,43 @@ fn dial(config: DialConfig) -> Result<()> {
let mut document_registry = CultNetDocumentRegistry::new();
register_capability_bindings(&mut document_registry, &schema_registration.schema_id);
let stream = TcpStream::connect((config.target_host.as_str(), config.target_port))
.with_context(|| {
format!(
"failed to connect to {}:{}",
config.target_host, config.target_port
)
})?;
let mut transport = TcpFramedTransportConnection::new(
stream,
tcp_transport_profile(&config.runtime_id, &config.target_host, config.target_port),
);
let transport_name = if config.target_rudp_port.is_some() {
"rudp"
} else {
"tcp"
};
let mut transport = open_dial_transport(&config)?;
send_message(
&mut transport,
&CultNetMessage::Hello {
runtime_id: config.runtime_id.clone(),
runtime_kind: config.runtime_kind.clone(),
agent_id: Some(config.agent_id.clone()),
role: None,
display_name: Some(config.display_name.clone()),
supported_document_types: Some(vec![INTEROP_DOCUMENT_TYPE.to_string()]),
supported_mutation_contracts: Some(interaction_contracts()),
supported_message_versions: Some(vec![INTEROP_SCHEMA_VERSION.to_string()]),
transport_profiles: None,
supports_schema_catalog: Some(true),
},
)?;
transport.send_schema_message(&CultNetMessage::Hello {
runtime_id: config.runtime_id.clone(),
runtime_kind: config.runtime_kind.clone(),
agent_id: Some(config.agent_id.clone()),
role: None,
display_name: Some(config.display_name.clone()),
supported_document_types: Some(vec![INTEROP_DOCUMENT_TYPE.to_string()]),
supported_mutation_contracts: Some(interaction_contracts()),
supported_message_versions: Some(vec![INTEROP_SCHEMA_VERSION.to_string()]),
transport_profiles: Some(dial_transport_profiles(&config)),
supports_schema_catalog: Some(true),
})?;
let remote_hello = read_message(&mut transport)?;
let remote_hello = transport
.read_schema_message()
.context("waiting for remote hello")?;
let remote_runtime_id = match &remote_hello {
CultNetMessage::Hello { runtime_id, .. } => runtime_id.clone(),
other => return Err(anyhow!("expected hello response, got {other:?}")),
};
send_message(
&mut transport,
&CultNetMessage::SchemaCatalogRequest {
message_id: format!("{}-catalog", config.runtime_id),
include_schema_json: Some(true),
schema_ids: None,
kinds: None,
},
)?;
let catalog_response = read_message(&mut transport)?;
transport.send_schema_message(&CultNetMessage::SchemaCatalogRequest {
message_id: format!("{}-catalog", config.runtime_id),
include_schema_json: Some(true),
schema_ids: None,
kinds: None,
})?;
let catalog_response = transport
.read_schema_message()
.context("waiting for schema catalog response")?;
let has_interop_schema = match &catalog_response {
CultNetMessage::SchemaCatalogResponse { schemas, .. } => schemas.iter().any(|schema| {
schema.schema_id == schema_registration.schema_id
@ -411,15 +431,14 @@ fn dial(config: DialConfig) -> Result<()> {
other => return Err(anyhow!("expected catalog response, got {other:?}")),
};
send_message(
&mut transport,
&CultNetMessage::SnapshotRequest {
message_id: format!("{}-snapshot", config.runtime_id),
schema_ids: Some(vec![schema_registration.schema_id.clone()]),
record_keys: None,
},
)?;
let snapshot_response = read_message(&mut transport)?;
transport.send_schema_message(&CultNetMessage::SnapshotRequest {
message_id: format!("{}-snapshot", config.runtime_id),
schema_ids: Some(vec![schema_registration.schema_id.clone()]),
record_keys: None,
})?;
let snapshot_response = transport
.read_schema_message()
.context("waiting for snapshot response")?;
let applied = document_registry
.apply_raw_snapshot_response::<CultNetInteropNote>(&mut cache, &snapshot_response)?;
let note = applied
@ -429,9 +448,10 @@ fn dial(config: DialConfig) -> Result<()> {
print_json(&serde_json::json!({
"mode": "dial",
"transport": transport_name,
"runtimeId": config.runtime_id,
"targetHost": config.target_host,
"targetPort": config.target_port,
"targetPort": config.target_rudp_port.or(config.target_port),
"remoteHello": {
"schemaVersion": "cultnet.hello.v0",
"runtimeId": remote_runtime_id,
@ -445,8 +465,8 @@ fn dial(config: DialConfig) -> Result<()> {
"body": note.body,
"tags": note.tags,
},
"mutatedNote": mutate_remote_note(&mut transport, &mut cache, &document_registry, &schema_registration.schema_id, &config.runtime_id, &note)?,
"fireReceipt": fire_remote_weapon(&mut transport, &mut cache, &document_registry, &config.runtime_id, &remote_runtime_id)?,
"mutatedNote": mutate_remote_note(&mut *transport, &mut cache, &document_registry, &schema_registration.schema_id, &config.runtime_id, &note)?,
"fireReceipt": fire_remote_weapon(&mut *transport, &mut cache, &document_registry, &config.runtime_id, &remote_runtime_id)?,
}))?;
Ok(())
}
@ -727,7 +747,7 @@ fn poll_rudp_after_send(
}
fn mutate_remote_note(
transport: &mut TcpFramedTransportConnection<TcpStream>,
transport: &mut dyn SchemaMessageTransport,
cache: &mut CultCache,
document_registry: &CultNetDocumentRegistry,
note_schema_id: &str,
@ -747,12 +767,16 @@ fn mutate_remote_note(
&intent,
CultNetDocumentPutOptions::default(),
)?;
send_message(transport, &message)?;
transport.send_schema_message(&message)?;
let receipt_message = read_message(transport)?;
let receipt_message = transport
.read_schema_message()
.context("waiting for mutation receipt")?;
let _receipt = document_registry
.apply_raw_document_put_message::<CultNetInteropMutationReceipt>(cache, &receipt_message)?;
let mutated_message = read_message(transport)?;
let mutated_message = transport
.read_schema_message()
.context("waiting for mutated note")?;
let mutated = document_registry
.apply_raw_document_put_message::<CultNetInteropNote>(cache, &mutated_message)?;
if mutated_message_schema_id(&mutated_message) != Some(note_schema_id) {
@ -769,7 +793,7 @@ fn mutate_remote_note(
}
fn fire_remote_weapon(
transport: &mut TcpFramedTransportConnection<TcpStream>,
transport: &mut dyn SchemaMessageTransport,
cache: &mut CultCache,
document_registry: &CultNetDocumentRegistry,
runtime_id: &str,
@ -787,8 +811,10 @@ fn fire_remote_weapon(
&command,
CultNetDocumentPutOptions::default(),
)?;
send_message(transport, &message)?;
let receipt_message = read_message(transport)?;
transport.send_schema_message(&message)?;
let receipt_message = transport
.read_schema_message()
.context("waiting for fire receipt")?;
let receipt = document_registry
.apply_raw_document_put_message::<CultNetInteropFireReceipt>(cache, &receipt_message)?;
Ok(serde_json::json!({
@ -827,10 +853,11 @@ fn start_udp_discovery_server(config: Arc<PeerConfig>) -> Result<()> {
tcp_port: config.tcp_port,
wire_contract: "cultnet.schema.v0".to_string(),
supported_document_types: vec![INTEROP_DOCUMENT_TYPE.to_string()],
transport_profiles: Some(tcp_transport_profiles(
transport_profiles: Some(interop_transport_profiles(
&config.runtime_id,
&config.advertise_host,
config.tcp_port,
config.rudp_port,
)),
supports_schema_catalog: true,
};
@ -887,6 +914,122 @@ fn start_tcp_server(
Ok(())
}
fn start_rudp_server(
config: Arc<PeerConfig>,
cache: Arc<Mutex<CultCache>>,
document_registry: Arc<CultNetDocumentRegistry>,
schema_registry: Arc<CultNetSchemaRegistry>,
) -> Result<()> {
let socket =
UdpSocket::bind((config.bind_host.as_str(), config.rudp_port)).with_context(|| {
format!(
"failed to bind RUDP listener on {}:{}",
config.bind_host, config.rudp_port
)
})?;
socket.set_read_timeout(Some(Duration::from_millis(20)))?;
thread::spawn(move || {
if let Err(error) =
run_rudp_server(socket, config, cache, document_registry, schema_registry)
{
eprintln!("{error:#}");
}
});
Ok(())
}
struct RudpPeerSession {
session: CultNetRudpSession,
}
fn run_rudp_server(
socket: UdpSocket,
config: Arc<PeerConfig>,
cache: Arc<Mutex<CultCache>>,
document_registry: Arc<CultNetDocumentRegistry>,
schema_registry: Arc<CultNetSchemaRegistry>,
) -> Result<()> {
let mut peers = BTreeMap::<SocketAddr, RudpPeerSession>::new();
let mut buffer = vec![0_u8; 65_535];
loop {
match socket.recv_from(&mut buffer) {
Ok((len, remote_addr)) => {
let packet = match decode_rudp_packet(&buffer[..len]) {
Ok(packet) if packet.connection_id == RUDP_INTEROP_CONNECTION_ID => packet,
Ok(_) => continue,
Err(error) => {
eprintln!("{error:#}");
continue;
}
};
let peer = peers.entry(remote_addr).or_insert_with(|| RudpPeerSession {
session: CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: RUDP_INTEROP_CONNECTION_ID,
initial_sequence: 100,
resend_delay_ms: RUDP_INTEROP_RESEND_DELAY_MS,
max_pending_reliable_packets: None,
}),
});
if packet.packet_type == CultNetRudpPacketType::Connect {
let accept = peer.session.accept_connect(
&packet,
rudp_now_ms(),
b"cultnet-interop-rudp".to_vec(),
)?;
send_rudp_packet(&socket, remote_addr, &accept)?;
continue;
}
let result = peer.session.receive(&packet, rudp_now_ms())?;
if let Some(reply) = result.reply {
send_rudp_packet(&socket, remote_addr, &reply)?;
}
if packet.packet_type == CultNetRudpPacketType::Data {
let ack = peer.session.create_ack();
send_rudp_packet(&socket, remote_addr, &ack)?;
}
for frame in result.delivered {
if frame.channel_id != "schema" {
continue;
}
let message = decode_cultnet_message_from_slice(
&frame.payload,
CultNetWireContract::CultNetSchemaV0,
)?;
let mut transport = RudpSessionSender {
socket: &socket,
remote_addr,
session: &mut peer.session,
};
handle_server_message(
&mut transport,
message,
&config,
&cache,
&document_registry,
&schema_registry,
)?;
}
}
Err(error)
if error.kind() == std::io::ErrorKind::WouldBlock
|| error.kind() == std::io::ErrorKind::TimedOut
|| error.kind() == std::io::ErrorKind::ConnectionReset =>
{
let now = rudp_now_ms();
for (remote_addr, peer) in peers.iter_mut() {
for packet in peer.session.due_resends(now) {
send_rudp_packet(&socket, *remote_addr, &packet)?;
}
}
}
Err(error) => return Err(error.into()),
}
}
}
fn handle_connection(
stream: TcpStream,
config: Arc<PeerConfig>,
@ -899,78 +1042,88 @@ fn handle_connection(
tcp_transport_profile(&config.runtime_id, &config.advertise_host, config.tcp_port),
);
loop {
let message = match read_message(&mut transport) {
let message = match transport.read_schema_message() {
Ok(message) => message,
Err(error) if is_eof_like(&error) => break,
Err(error) => return Err(error),
};
match message {
CultNetMessage::Hello { .. } => {
send_message(
&mut transport,
&CultNetMessage::Hello {
runtime_id: config.runtime_id.clone(),
runtime_kind: config.runtime_kind.clone(),
agent_id: Some(config.agent_id.clone()),
role: None,
display_name: Some(config.display_name.clone()),
supported_document_types: Some(vec![INTEROP_DOCUMENT_TYPE.to_string()]),
supported_mutation_contracts: Some(interaction_contracts()),
supported_message_versions: Some(vec![INTEROP_SCHEMA_VERSION.to_string()]),
transport_profiles: Some(tcp_transport_profiles(
&config.runtime_id,
&config.advertise_host,
config.tcp_port,
)),
supports_schema_catalog: Some(true),
},
)?;
}
request @ CultNetMessage::SchemaCatalogRequest { .. } => {
let response = schema_registry.create_catalog_response(&request)?;
send_message(&mut transport, &response)?;
}
CultNetMessage::SnapshotRequest {
message_id,
schema_ids,
record_keys,
} => {
let mut response = document_registry.create_raw_snapshot_response(
&cache.lock().expect("cache poisoned"),
message_id,
schema_ids.as_deref(),
record_keys.as_deref(),
)?;
if let CultNetMessage::SnapshotResponseRaw { documents, .. } = &mut response {
for document in documents.iter_mut() {
document.source_runtime_id = Some(config.runtime_id.clone());
document.source_agent_id = Some(config.agent_id.clone());
document.source_role = Some("peer".to_string());
document.tags =
Some(vec!["interop".to_string(), config.runtime_id.clone()]);
}
}
send_message(&mut transport, &response)?;
}
message @ CultNetMessage::DocumentPutRaw { .. } => {
handle_raw_put(
&mut transport,
&config,
&cache,
&document_registry,
&message,
)?;
}
_ => {}
}
handle_server_message(
&mut transport,
message,
&config,
&cache,
&document_registry,
&schema_registry,
)?;
}
Ok(())
}
fn handle_server_message(
transport: &mut dyn SchemaMessageSender,
message: CultNetMessage,
config: &PeerConfig,
cache: &Arc<Mutex<CultCache>>,
document_registry: &CultNetDocumentRegistry,
schema_registry: &CultNetSchemaRegistry,
) -> Result<()> {
match message {
CultNetMessage::Hello { .. } => {
transport.send_schema_message(&CultNetMessage::Hello {
runtime_id: config.runtime_id.clone(),
runtime_kind: config.runtime_kind.clone(),
agent_id: Some(config.agent_id.clone()),
role: None,
display_name: Some(config.display_name.clone()),
supported_document_types: Some(vec![INTEROP_DOCUMENT_TYPE.to_string()]),
supported_mutation_contracts: Some(interaction_contracts()),
supported_message_versions: Some(vec![INTEROP_SCHEMA_VERSION.to_string()]),
transport_profiles: Some(interop_transport_profiles(
&config.runtime_id,
&config.advertise_host,
config.tcp_port,
config.rudp_port,
)),
supports_schema_catalog: Some(true),
})?;
}
request @ CultNetMessage::SchemaCatalogRequest { .. } => {
let response = schema_registry.create_catalog_response(&request)?;
transport.send_schema_message(&response)?;
}
CultNetMessage::SnapshotRequest {
message_id,
schema_ids,
record_keys,
} => {
let mut response = document_registry.create_raw_snapshot_response(
&cache.lock().expect("cache poisoned"),
message_id,
schema_ids.as_deref(),
record_keys.as_deref(),
)?;
if let CultNetMessage::SnapshotResponseRaw { documents, .. } = &mut response {
for document in documents.iter_mut() {
document.source_runtime_id = Some(config.runtime_id.clone());
document.source_agent_id = Some(config.agent_id.clone());
document.source_role = Some("peer".to_string());
document.tags = Some(vec!["interop".to_string(), config.runtime_id.clone()]);
}
}
transport.send_schema_message(&response)?;
}
message @ CultNetMessage::DocumentPutRaw { .. } => {
handle_raw_put(transport, config, cache, document_registry, &message)?;
}
_ => {}
}
Ok(())
}
fn handle_raw_put(
transport: &mut TcpFramedTransportConnection<TcpStream>,
transport: &mut dyn SchemaMessageSender,
config: &PeerConfig,
cache: &Arc<Mutex<CultCache>>,
document_registry: &CultNetDocumentRegistry,
@ -1008,8 +1161,8 @@ fn handle_raw_put(
&note,
options,
)?;
send_message(transport, &receipt_message)?;
send_message(transport, &note_message)?;
transport.send_schema_message(&receipt_message)?;
transport.send_schema_message(&note_message)?;
} else if document.schema_id == FIRE_COMMAND_SCHEMA_ID {
let mut cache = cache.lock().expect("cache poisoned");
let command = document_registry
@ -1029,7 +1182,7 @@ fn handle_raw_put(
&receipt,
response_options(config, "side-effect"),
)?;
send_message(transport, &receipt_message)?;
transport.send_schema_message(&receipt_message)?;
}
Ok(())
}
@ -1140,8 +1293,91 @@ fn tcp_transport_profile(runtime_id: &str, host: &str, port: u16) -> CultNetTran
)
}
fn tcp_transport_profiles(runtime_id: &str, host: &str, port: u16) -> Vec<CultNetTransportProfile> {
vec![tcp_transport_profile(runtime_id, host, port)]
fn rudp_transport_profile(runtime_id: &str, host: &str, port: u16) -> CultNetTransportProfile {
create_rudp_transport_profile(
runtime_id,
RudpTransportProfileOptions {
transport_id: Some("interop-rudp".to_string()),
host: Some(host.to_string()),
port: Some(port),
max_fragment_bytes: Some(RUDP_INTEROP_MAX_FRAGMENT_BYTES as u32),
..RudpTransportProfileOptions::default()
},
)
}
fn interop_transport_profiles(
runtime_id: &str,
host: &str,
tcp_port: u16,
rudp_port: u16,
) -> Vec<CultNetTransportProfile> {
vec![
tcp_transport_profile(runtime_id, host, tcp_port),
rudp_transport_profile(runtime_id, host, rudp_port),
]
}
fn dial_transport_profiles(config: &DialConfig) -> Vec<CultNetTransportProfile> {
let mut profiles = Vec::new();
if let Some(target_port) = config.target_port {
profiles.push(tcp_transport_profile(
&config.runtime_id,
&config.target_host,
target_port,
));
}
if let Some(target_rudp_port) = config.target_rudp_port {
profiles.push(rudp_transport_profile(
&config.runtime_id,
&config.target_host,
target_rudp_port,
));
}
profiles
}
fn open_dial_transport(config: &DialConfig) -> Result<Box<dyn SchemaMessageTransport>> {
if let Some(target_rudp_port) = config.target_rudp_port {
let remote_addr: SocketAddr = format!("{}:{}", config.target_host, target_rudp_port)
.parse()
.with_context(|| {
format!(
"failed to parse RUDP remote endpoint {}:{}",
config.target_host, target_rudp_port
)
})?;
let socket = UdpSocket::bind(("127.0.0.1", 0))?;
socket.set_read_timeout(Some(Duration::from_millis(20)))?;
let mut options = CultNetRudpSocketTransportOptions::client(
format!("{}-interop-rudp-dial", config.runtime_id),
socket,
remote_addr,
RUDP_INTEROP_CONNECTION_ID,
);
options.transport_id = Some("interop-rudp".to_string());
options.resend_delay_ms = RUDP_INTEROP_RESEND_DELAY_MS;
options.max_fragment_bytes = Some(RUDP_INTEROP_MAX_FRAGMENT_BYTES as u32);
let mut transport = CultNetRudpSocketTransportConnection::new(options)?;
transport.connect(b"cultnet-interop-rudp".to_vec())?;
wait_for_rudp_connected(&mut transport, Duration::from_secs(5))?;
return Ok(Box::new(transport));
}
let target_port = config
.target_port
.ok_or_else(|| anyhow!("dial requires --target-port or --target-rudp-port"))?;
let stream =
TcpStream::connect((config.target_host.as_str(), target_port)).with_context(|| {
format!(
"failed to connect to {}:{}",
config.target_host, target_port
)
})?;
Ok(Box::new(TcpFramedTransportConnection::new(
stream,
tcp_transport_profile(&config.runtime_id, &config.target_host, target_port),
)))
}
fn response_options(config: &PeerConfig, tag: &str) -> CultNetDocumentPutOptions {
@ -1178,25 +1414,108 @@ fn build_note(runtime_id: &str, display_name: &str) -> CultNetInteropNote {
}
}
fn send_message(
transport: &mut TcpFramedTransportConnection<TcpStream>,
message: &CultNetMessage,
) -> Result<()> {
let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
transport.send("schema", &payload)
}
fn read_message(transport: &mut TcpFramedTransportConnection<TcpStream>) -> Result<CultNetMessage> {
let frame = transport.receive()?;
decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0)
}
fn is_eof_like(error: &anyhow::Error) -> bool {
error
.downcast_ref::<std::io::Error>()
.is_some_and(|io| io.kind() == std::io::ErrorKind::UnexpectedEof)
}
impl SchemaMessageSender for TcpFramedTransportConnection<TcpStream> {
fn send_schema_message(&mut self, message: &CultNetMessage) -> Result<()> {
let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
self.send("schema", &payload)
}
}
impl SchemaMessageTransport for TcpFramedTransportConnection<TcpStream> {
fn read_schema_message(&mut self) -> Result<CultNetMessage> {
let frame = self.receive()?;
decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0)
}
}
impl SchemaMessageSender for CultNetRudpSocketTransportConnection {
fn send_schema_message(&mut self, message: &CultNetMessage) -> Result<()> {
CultNetRudpSocketTransportConnection::send_schema_message(self, message)
}
}
impl SchemaMessageTransport for CultNetRudpSocketTransportConnection {
fn read_schema_message(&mut self) -> Result<CultNetMessage> {
let deadline = Instant::now() + RUDP_INTEROP_READ_TIMEOUT;
while Instant::now() < deadline {
if let Some(message) = self.receive_schema_message_once()? {
return Ok(message);
}
self.poll_resends()?;
thread::sleep(Duration::from_millis(5));
}
Err(anyhow!("timed out waiting for RUDP schema message"))
}
}
struct RudpSessionSender<'a> {
socket: &'a UdpSocket,
remote_addr: SocketAddr,
session: &'a mut CultNetRudpSession,
}
impl SchemaMessageSender for RudpSessionSender<'_> {
fn send_schema_message(&mut self, message: &CultNetMessage) -> Result<()> {
let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
let packets = self.session.send_many(
"schema",
payload,
CultNetRudpSendOptions {
reliable: true,
ordered: true,
sequenced: false,
now_ms: rudp_now_ms(),
},
Some(RUDP_INTEROP_MAX_FRAGMENT_BYTES),
)?;
for packet in packets {
send_rudp_packet(self.socket, self.remote_addr, &packet)?;
}
Ok(())
}
}
fn send_rudp_packet(
socket: &UdpSocket,
remote_addr: SocketAddr,
packet: &cultnet_rs::CultNetRudpPacket,
) -> Result<()> {
socket.send_to(&encode_rudp_packet(packet)?, remote_addr)?;
Ok(())
}
fn wait_for_rudp_connected(
transport: &mut CultNetRudpSocketTransportConnection,
timeout: Duration,
) -> Result<()> {
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
if transport.connected() {
return Ok(());
}
let _ = transport.receive_once()?;
transport.poll_resends()?;
thread::sleep(Duration::from_millis(5));
}
Err(anyhow!("timed out waiting for RUDP connect"))
}
fn rudp_now_ms() -> u64 {
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
fn create_discovery_socket(port: u16, _join_group: bool) -> Result<UdpSocket> {
let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?;
socket.set_reuse_address(true)?;
@ -1209,6 +1528,7 @@ fn create_discovery_socket(port: u16, _join_group: bool) -> Result<UdpSocket> {
}
fn parse_peer_config(options: &BTreeMap<String, String>) -> Result<PeerConfig> {
let tcp_port = parse_u16_arg(options, "tcp-port")?;
Ok(PeerConfig {
runtime_id: require_arg(options, "runtime-id")?.to_string(),
runtime_kind: require_arg(options, "runtime-kind")?.to_string(),
@ -1219,7 +1539,13 @@ fn parse_peer_config(options: &BTreeMap<String, String>) -> Result<PeerConfig> {
.cloned()
.unwrap_or_else(|| "127.0.0.1".to_string()),
advertise_host: require_arg(options, "advertise-host")?.to_string(),
tcp_port: parse_u16_arg(options, "tcp-port")?,
tcp_port,
rudp_port: options
.get("rudp-port")
.map(|value| value.parse::<u16>())
.transpose()
.with_context(|| "argument --rudp-port must be a u16")?
.unwrap_or(tcp_port),
discovery_port: parse_u16_arg(options, "discovery-port")?,
discovery_group: parse_ipv4_arg(options, "discovery-group")?,
schema_path: require_arg(options, "schema-path")?.to_string(),
@ -1233,7 +1559,16 @@ fn parse_dial_config(options: &BTreeMap<String, String>) -> Result<DialConfig> {
display_name: require_arg(options, "display-name")?.to_string(),
agent_id: require_arg(options, "agent-id")?.to_string(),
target_host: require_arg(options, "target-host")?.to_string(),
target_port: parse_u16_arg(options, "target-port")?,
target_port: options
.get("target-port")
.map(|value| value.parse::<u16>())
.transpose()
.with_context(|| "argument --target-port must be a u16")?,
target_rudp_port: options
.get("target-rudp-port")
.map(|value| value.parse::<u16>())
.transpose()
.with_context(|| "argument --target-rudp-port must be a u16")?,
schema_path: require_arg(options, "schema-path")?.to_string(),
})
}

100
packages/cultnet-ts/test/interop/cultnet-interop.test.ts
View file

@ -972,6 +972,106 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state
], cultcachePyRoot, { PYTHONPATH: cultcachePySrc }, expectedPeers);
});
test("CultNet TypeScript and Rust full interop peers exchange schema-v0 over RUDP", async (t) => {
logInteropPhase("ts-rust-full-rudp", "build peers");
await buildRustInteropPeer();
cleanInteropStores([
"rust-rudp-peer",
"rust-rudp-client-dial",
"ts-rust-rudp-peer",
"ts-rust-rudp-client-dial",
]);
const rustTcpPort = await getFreePort();
const rustRudpPort = await getFreePort();
const tsTcpPort = await getFreePort();
const tsRudpPort = await getFreePort();
const discoveryPort = await getFreePort();
const advertiseHost = findAdvertiseHost();
const servers: RunningServeProcess[] = [];
servers.push(await spawnServeProcess("rust-rudp", {
command: rustBinaryPath,
args: [
"serve",
"--runtime-id", "rust-rudp-peer",
"--runtime-kind", "rust",
"--display-name", "Rust RUDP Peer",
"--agent-id", "rust-rudp-agent",
"--bind-host", "127.0.0.1",
"--advertise-host", advertiseHost,
"--tcp-port", String(rustTcpPort),
"--rudp-port", String(rustRudpPort),
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--schema-path", interopSchemaPath,
],
cwd: cultnetRsRoot,
}));
await servers[servers.length - 1].ready;
servers.push(await spawnServeProcess("ts-rust-rudp", {
command: process.execPath,
args: [
tsPeerScript,
"serve",
"--runtime-id", "ts-rust-rudp-peer",
"--runtime-kind", "node",
"--display-name", "TypeScript Rust RUDP Peer",
"--agent-id", "ts-rust-rudp-agent",
"--bind-host", "127.0.0.1",
"--advertise-host", advertiseHost,
"--tcp-port", String(tsTcpPort),
"--rudp-port", String(tsRudpPort),
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--schema-path", interopSchemaPath,
],
cwd: cultNetTsRoot,
}));
await servers[servers.length - 1].ready;
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
logInteropPhase("ts-rust-full-rudp", "typescript dials rust over rudp");
const tsRudpDial = await runJsonCommand("ts-rust-rudp-dial", process.execPath, [
tsPeerScript,
"dial",
"--runtime-id", "ts-rust-rudp-client",
"--runtime-kind", "node",
"--display-name", "TS Rust RUDP Dialer",
"--agent-id", "ts-rust-rudp-client-agent",
"--target-host", "127.0.0.1",
"--target-rudp-port", String(rustRudpPort),
"--schema-path", interopSchemaPath,
], cultNetTsRoot);
assert.equal(tsRudpDial.transport, "rudp");
assert.equal(tsRudpDial.remoteHello.runtimeId, "rust-rudp-peer");
assert.equal(tsRudpDial.hasInteropSchema, true);
assert.equal(tsRudpDial.retrievedNote.authorRuntimeId, "rust-rudp-peer");
assert.ok(tsRudpDial.mutatedNote.tags.includes("decorated:ts-rust-rudp-client"));
assert.equal(tsRudpDial.fireReceipt.accepted, true);
logInteropPhase("ts-rust-full-rudp", "rust dials typescript over rudp");
const rustRudpDial = await runJsonCommand("rust-rudp-dial", rustBinaryPath, [
"dial",
"--runtime-id", "rust-rudp-client",
"--runtime-kind", "rust",
"--display-name", "Rust RUDP Dialer",
"--agent-id", "rust-rudp-client-agent",
"--target-host", "127.0.0.1",
"--target-rudp-port", String(tsRudpPort),
"--schema-path", interopSchemaPath,
], cultnetRsRoot);
assert.equal(rustRudpDial.transport, "rudp");
assert.equal(rustRudpDial.remoteHello.runtimeId, "ts-rust-rudp-peer");
assert.equal(rustRudpDial.hasInteropSchema, true);
assert.equal(rustRudpDial.retrievedNote.authorRuntimeId, "ts-rust-rudp-peer");
assert.ok(rustRudpDial.mutatedNote.tags.includes("decorated:rust-rudp-client"));
assert.equal(rustRudpDial.fireReceipt.accepted, true);
});
test("CultNet TypeScript and Python exchange schema frames over the shared RUDP socket transport", async () => {
const pythonPeer = spawnPythonRudpPeer();
const clientSocket = await bindUdpSocket();