From c80703172a38504d73ad7d22f22b3df8ff4ec0dd Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 17 Jun 2026 10:07:37 +0100 Subject: [PATCH] Add Rust interop RUDP serve dial path --- .github/workflows/cultnet-interop.yml | 2 +- docs/cultnet-transport-parity.md | 7 +- docs/runtime-parity-scope.md | 2 +- .../examples/cultnet_interop_peer.rs | 615 ++++++++++++++---- .../test/interop/cultnet-interop.test.ts | 100 +++ 5 files changed, 581 insertions(+), 145 deletions(-) diff --git a/.github/workflows/cultnet-interop.yml b/.github/workflows/cultnet-interop.yml index 212284db..1bde67ef 100644 --- a/.github/workflows/cultnet-interop.yml +++ b/.github/workflows/cultnet-interop.yml @@ -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 diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 848cee3a..d93b160b 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -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 diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index bd8548ce..3d3c446b 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -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 diff --git a/packages/cultnet-rs/examples/cultnet_interop_peer.rs b/packages/cultnet-rs/examples/cultnet_interop_peer.rs index 1e360541..ab00c6c8 100644 --- a/packages/cultnet-rs/examples/cultnet_interop_peer.rs +++ b/packages/cultnet-rs/examples/cultnet_interop_peer.rs @@ -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, + target_rudp_port: Option, schema_path: String, } +trait SchemaMessageSender { + fn send_schema_message(&mut self, message: &CultNetMessage) -> Result<()>; +} + +trait SchemaMessageTransport: SchemaMessageSender { + fn read_schema_message(&mut self) -> Result; +} + 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::(&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, ¬e)?, - "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, ¬e)?, + "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, + 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::(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::(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, + 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::(cache, &receipt_message)?; Ok(serde_json::json!({ @@ -827,10 +853,11 @@ fn start_udp_discovery_server(config: Arc) -> 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, + cache: Arc>, + document_registry: Arc, + schema_registry: Arc, +) -> 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, + cache: Arc>, + document_registry: Arc, + schema_registry: Arc, +) -> Result<()> { + let mut peers = BTreeMap::::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, @@ -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>, + 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, + transport: &mut dyn SchemaMessageSender, config: &PeerConfig, cache: &Arc>, document_registry: &CultNetDocumentRegistry, @@ -1008,8 +1161,8 @@ fn handle_raw_put( ¬e, options, )?; - send_message(transport, &receipt_message)?; - send_message(transport, ¬e_message)?; + transport.send_schema_message(&receipt_message)?; + transport.send_schema_message(¬e_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 { - 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 { + vec![ + tcp_transport_profile(runtime_id, host, tcp_port), + rudp_transport_profile(runtime_id, host, rudp_port), + ] +} + +fn dial_transport_profiles(config: &DialConfig) -> Vec { + 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> { + 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, - message: &CultNetMessage, -) -> Result<()> { - let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; - transport.send("schema", &payload) -} - -fn read_message(transport: &mut TcpFramedTransportConnection) -> Result { - let frame = transport.receive()?; - decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) -} - fn is_eof_like(error: &anyhow::Error) -> bool { error .downcast_ref::() .is_some_and(|io| io.kind() == std::io::ErrorKind::UnexpectedEof) } +impl SchemaMessageSender for TcpFramedTransportConnection { + 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 { + fn read_schema_message(&mut self) -> Result { + 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 { + 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 { 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 { } fn parse_peer_config(options: &BTreeMap) -> Result { + 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) -> Result { .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::()) + .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) -> Result { 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::()) + .transpose() + .with_context(|| "argument --target-port must be a u16")?, + target_rudp_port: options + .get("target-rudp-port") + .map(|value| value.parse::()) + .transpose() + .with_context(|| "argument --target-rudp-port must be a u16")?, schema_path: require_arg(options, "schema-path")?.to_string(), }) } diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 3f982239..487a1fb4 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -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();