diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 2e50471e..9001b70c 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -182,7 +182,7 @@ Current progress: exchange reliable ordered `schema` frames through the shared handshake and packet language. - The same harness now proves schema-v0 MessagePack messages over RUDP in both - directions for C#, Kotlin, and Python peers against TypeScript. Those tests + directions for C#, Kotlin, Python, and Rust peers against TypeScript. Those tests send `cultnet.schema_catalog_request.v0` payloads through reliable ordered `schema` frames and validate `cultnet.hello.v0` responses with advertised RUDP transport profiles where the responding runtime owns them. @@ -239,8 +239,8 @@ Current progress: creating a second authority path. - The remaining parity work is to deepen Python's server-side use of its port and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those - ports, then broaden RUDP interop into reconnect policy, Rust schema-v0 - MessagePack-over-RUDP coverage, and the wider schema-message matrix. + ports, then broaden RUDP interop into reconnect policy and the wider + schema-message matrix. ## RUDP Packet Contract V0 diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 3089e457..27fd3051 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, shared transport profiles, shared RUDP socket transport, CultMesh node/session/catalog/stream/simulation surfaces | 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, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, 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, transport profiles, TCP-framed transport port, shared RUDP socket transport, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | -| Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport | Full CultMesh ergonomic facade or production game-session server | +| Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, schema-v0 MessagePack over RUDP evidence | Full CultMesh ergonomic facade or production game-session server | | Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs, authority leases, stream catalog, branded `CultMesh` facade, RUDP client/server sugar, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | ## Evidence Gates @@ -42,7 +42,8 @@ CultMesh. wire harness. It covers schema-v0 interop and RUDP exchange across C#, TypeScript, Rust, Python, and Kotlin, including loss, reordering, fragmentation, disconnect reason propagation, ping/pong, and schema-v0 - MessagePack over RUDP for C#, Python, and Kotlin peers against TypeScript. + MessagePack over RUDP for C#, Rust, Python, and Kotlin peers against + TypeScript. - `packages/cultcache-py/PARITY.md` is the Python-specific audit ledger. Its "Still Not Claimed" section is part of the contract, not an apology. - `docs/cultnet-transport-parity.md` is the transport ownership map. It tracks @@ -56,8 +57,8 @@ CultMesh. - Broaden service adoption so local servers and daemon paths use the shared transport port consistently instead of keeping raw TCP/LiteNetLib/WebSocket ownership in individual bodies. -- Extend RUDP interop evidence through reconnect policy, Rust schema-v0 - MessagePack-over-RUDP coverage, and the wider schema-message matrix. +- Extend RUDP interop evidence through reconnect policy and the wider + schema-message matrix. - Keep Kotlin ergonomic parity moving with Android/JVM-shaped APIs instead of backfilling C# server responsibilities it should not own. - Keep performance claims evidence-bound: C# owns SoA; Python has benchmark diff --git a/packages/cultnet-rs/examples/cultnet_interop_peer.rs b/packages/cultnet-rs/examples/cultnet_interop_peer.rs index 2e86eff9..1e360541 100644 --- a/packages/cultnet-rs/examples/cultnet_interop_peer.rs +++ b/packages/cultnet-rs/examples/cultnet_interop_peer.rs @@ -18,9 +18,11 @@ use cultnet_rs::CultNetSchemaRegistration; use cultnet_rs::CultNetSchemaRegistry; use cultnet_rs::CultNetTransportProfile; use cultnet_rs::CultNetWireContract; +use cultnet_rs::RudpTransportProfileOptions; use cultnet_rs::TcpFramedTransportConnection; use cultnet_rs::TcpFramedTransportProfileOptions; 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::encode_cultnet_message_to_vec; @@ -206,7 +208,7 @@ struct DialConfig { fn main() -> Result<()> { let mut args = std::env::args().skip(1); let mode = args.next().ok_or_else(|| { - anyhow!("expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once") + anyhow!("expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once | rudp-serve-message-once | rudp-dial-message-once") })?; let options = parse_args(args.collect()); @@ -216,6 +218,8 @@ fn main() -> Result<()> { "dial" => dial(parse_dial_config(&options)?)?, "rudp-serve-once" => rudp_serve_once(&options)?, "rudp-dial-once" => rudp_dial_once(&options)?, + "rudp-serve-message-once" => rudp_serve_message_once(&options)?, + "rudp-dial-message-once" => rudp_dial_message_once(&options)?, _ => return Err(anyhow!("unknown mode {mode}")), } @@ -566,6 +570,149 @@ fn rudp_dial_once(options: &BTreeMap) -> Result<()> { Err(anyhow!("timed out waiting for TypeScript RUDP response")) } +fn rudp_serve_message_once(options: &BTreeMap) -> Result<()> { + let bind_host = options + .get("bind-host") + .cloned() + .unwrap_or_else(|| "127.0.0.1".to_string()); + let bind_port = options + .get("bind-port") + .map(|value| value.parse::()) + .transpose() + .with_context(|| "argument --bind-port must be a u16")? + .unwrap_or(0); + let socket = UdpSocket::bind((bind_host.as_str(), bind_port)).with_context(|| { + format!("failed to bind RUDP message server on {bind_host}:{bind_port}") + })?; + socket.set_read_timeout(Some(Duration::from_millis(20)))?; + let local_port = socket.local_addr()?.port(); + let mut transport_options = + CultNetRudpSocketTransportOptions::server("rust-rudp-message-interop", socket, 0x22446689); + transport_options.initial_sequence = 100; + transport_options.resend_delay_ms = 25; + let mut transport = CultNetRudpSocketTransportConnection::new(transport_options)?; + + print_json(&serde_json::json!({ + "status": "ready", + "port": local_port, + }))?; + + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + if let Some(frame) = transport.receive_once()? { + if frame.channel_id != "schema" { + return Err(anyhow!("unexpected RUDP channel: {}", frame.channel_id)); + } + let message = decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?; + let CultNetMessage::SchemaCatalogRequest { .. } = message else { + return Err(anyhow!("unexpected schema message: {message:?}")); + }; + let response = CultNetMessage::Hello { + runtime_id: "rust-rudp-message-interop".to_string(), + runtime_kind: "rust".to_string(), + agent_id: None, + role: None, + display_name: Some("Rust RUDP Message Interop".to_string()), + supported_document_types: Some(vec![]), + supported_mutation_contracts: Some(vec![]), + supported_message_versions: Some(vec![ + "cultnet.hello.v0".to_string(), + "cultnet.schema_catalog_request.v0".to_string(), + ]), + transport_profiles: Some(vec![create_rudp_transport_profile( + "rust-rudp-message-interop", + RudpTransportProfileOptions { + host: Some("127.0.0.1".to_string()), + port: Some(local_port), + ..RudpTransportProfileOptions::default() + }, + )]), + supports_schema_catalog: Some(true), + }; + transport.send( + "schema", + encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0)?, + )?; + poll_rudp_after_send(&mut transport, Duration::from_millis(250))?; + print_json(&serde_json::json!({ "status": "ok" }))?; + return Ok(()); + } + transport.poll_resends()?; + thread::sleep(Duration::from_millis(5)); + } + + Err(anyhow!( + "timed out waiting for TypeScript RUDP schema-v0 message" + )) +} + +fn rudp_dial_message_once(options: &BTreeMap) -> Result<()> { + let target_host = require_arg(options, "target-host")?; + let target_port = parse_u16_arg(options, "target-port")?; + let remote_addr: SocketAddr = format!("{target_host}:{target_port}") + .parse() + .with_context(|| { + format!("failed to parse RUDP remote endpoint {target_host}:{target_port}") + })?; + let socket = UdpSocket::bind(("127.0.0.1", 0))?; + socket.set_read_timeout(Some(Duration::from_millis(20)))?; + let mut transport_options = CultNetRudpSocketTransportOptions::client( + "rust-rudp-message-client-interop", + socket, + remote_addr, + 0x88664423, + ); + transport_options.resend_delay_ms = 25; + let mut transport = CultNetRudpSocketTransportConnection::new(transport_options)?; + transport.connect(b"rust-message-join".to_vec())?; + + let request = CultNetMessage::SchemaCatalogRequest { + message_id: "rust-ts-schema-message".to_string(), + include_schema_json: Some(false), + schema_ids: Some(vec![]), + kinds: Some(vec![CultNetSchemaKind::WireMessage]), + }; + + let mut sent = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + if let Some(frame) = transport.receive_once()? { + if frame.channel_id != "schema" { + return Err(anyhow!("unexpected RUDP channel: {}", frame.channel_id)); + } + let message = decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?; + match message { + CultNetMessage::Hello { runtime_id, .. } + if runtime_id == "ts-rust-rudp-message-server" => + { + print_json(&serde_json::json!({ "status": "ok" }))?; + return Ok(()); + } + other => return Err(anyhow!("unexpected schema message: {other:?}")), + } + } + transport.poll_resends()?; + if transport.connected() && !sent { + transport.send( + "schema", + encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0)?, + )?; + sent = true; + } + thread::sleep(Duration::from_millis(5)); + } + + Err(anyhow!( + "timed out waiting for TypeScript RUDP schema-v0 response" + )) +} + fn poll_rudp_after_send( transport: &mut CultNetRudpSocketTransportConnection, duration: Duration, diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 1442f323..8a71976c 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -947,12 +947,12 @@ test("CultNet Python and TypeScript exchange schema frames when Python dials a T try { const receivedFrame = once(server, "frame"); - const [frame] = await withTimeout(receivedFrame, 2_000, "Python RUDP client frame"); + const [frame] = await withTimeout(receivedFrame, 5_000, "Python RUDP client frame"); assert.equal(frame.channelId, "schema"); assert.deepEqual(Buffer.from(frame.payload), Buffer.from("python-client-state")); server.send("schema", Buffer.from("ts-server-state")); - const [exitCode] = await withTimeout(once(pythonClient.child, "exit"), 2_000, "Python RUDP client exit"); + const [exitCode] = await withTimeout(once(pythonClient.child, "exit"), 5_000, "Python RUDP client exit"); assert.equal(exitCode, 0, pythonClient.stderr.join("")); } finally { server.close(); @@ -1037,7 +1037,7 @@ test("CultNet Python and TypeScript exchange schema-v0 MessagePack messages over try { const receivedFrame = once(server, "frame"); - const [frame] = await withTimeout(receivedFrame, 2_000, "Python RUDP schema-v0 client frame"); + const [frame] = await withTimeout(receivedFrame, 5_000, "Python RUDP schema-v0 client frame"); assert.equal(frame.channelId, "schema"); const message = decode(frame.payload) as { schemaVersion?: string; @@ -1058,7 +1058,7 @@ test("CultNet Python and TypeScript exchange schema-v0 MessagePack messages over supportsSchemaCatalog: true, }))); - const [exitCode] = await withTimeout(once(pythonClient.child, "exit"), 2_000, "Python RUDP message client exit"); + const [exitCode] = await withTimeout(once(pythonClient.child, "exit"), 5_000, "Python RUDP message client exit"); assert.equal(exitCode, 0, pythonClient.stderr.join("")); } finally { server.close(); @@ -1145,6 +1145,114 @@ test("CultNet Rust and TypeScript exchange schema frames when Rust dials a TypeS } }); +test("CultNet TypeScript and Rust exchange schema-v0 MessagePack messages over RUDP", async () => { + await buildRustInteropPeer(); + const rustPeer = spawnRustRudpMessageServer(); + const clientSocket = await bindUdpSocket(); + let client: CultNetRudpSocketTransportConnection | undefined; + + try { + const rustPort = await rustPeer.ready; + client = new CultNetRudpSocketTransportConnection({ + runtimeId: "ts-rust-rudp-message-interop", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: rustPort, + connectionId: 0x22446689, + initialSequence: 1, + resendDelayMs: 25, + resendPollMs: 5, + }); + + const receivedFrame = once(client, "frame"); + client.connect(Buffer.from("ts-rust-message-join")); + await waitFor(() => client?.connected === true, "TypeScript RUDP client to complete Rust message handshake"); + client.send("schema", Buffer.from(encode({ + schemaVersion: "cultnet.schema_catalog_request.v0", + messageId: "ts-rust-schema-message", + includeSchemaJson: false, + schemaIds: [], + kinds: ["wire_message"], + }))); + + const [frame] = await withTimeout(receivedFrame, 2_000, "Rust RUDP schema-v0 response frame"); + assert.equal(frame.channelId, "schema"); + const message = decode(frame.payload) as { + schemaVersion?: string; + runtimeId?: string; + supportsSchemaCatalog?: boolean; + transportProfiles?: Array<{ transports?: Array<{ protocol?: string; channels?: Array<{ channelId?: string }> }> }>; + }; + assert.equal(message.schemaVersion, "cultnet.hello.v0"); + assert.equal(message.runtimeId, "rust-rudp-message-interop"); + assert.equal(message.supportsSchemaCatalog, true); + assert.equal(message.transportProfiles?.[0]?.transports?.[0]?.protocol, "rudp"); + assert.equal(message.transportProfiles?.[0]?.transports?.[0]?.channels?.[0]?.channelId, "schema"); + + const [exitCode] = await withTimeout(once(rustPeer.child, "exit"), 2_000, "Rust RUDP message peer exit"); + assert.equal(exitCode, 0, rustPeer.stderr.join("")); + } finally { + if (client) { + client.close(); + } else { + clientSocket.close(); + } + if (rustPeer.child.exitCode === null && !rustPeer.child.killed) { + rustPeer.child.kill("SIGTERM"); + await once(rustPeer.child, "exit").catch(() => undefined); + } + } +}); + +test("CultNet Rust and TypeScript exchange schema-v0 MessagePack messages over RUDP when Rust dials", async () => { + await buildRustInteropPeer(); + const serverSocket = await bindUdpSocket(); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "ts-rust-rudp-message-server", + socket: serverSocket, + mode: "server", + connectionId: 0x88664423, + initialSequence: 100, + resendDelayMs: 25, + resendPollMs: 5, + }); + const rustClient = spawnRustRudpMessageClient(udpSocketPort(serverSocket)); + + try { + const receivedFrame = once(server, "frame"); + const [frame] = await withTimeout(receivedFrame, 2_000, "Rust RUDP schema-v0 client frame"); + assert.equal(frame.channelId, "schema"); + const message = decode(frame.payload) as { + schemaVersion?: string; + messageId?: string; + kinds?: string[]; + }; + assert.equal(message.schemaVersion, "cultnet.schema_catalog_request.v0"); + assert.equal(message.messageId, "rust-ts-schema-message"); + assert.deepEqual(message.kinds, ["wire_message"]); + server.send("schema", Buffer.from(encode({ + schemaVersion: "cultnet.hello.v0", + runtimeId: "ts-rust-rudp-message-server", + runtimeKind: "typescript", + supportedDocumentTypes: [], + supportedMutationContracts: [], + supportedMessageVersions: ["cultnet.hello.v0", "cultnet.schema_catalog_request.v0"], + transportProfiles: [], + supportsSchemaCatalog: true, + }))); + + const [exitCode] = await withTimeout(once(rustClient.child, "exit"), 2_000, "Rust RUDP message client exit"); + assert.equal(exitCode, 0, rustClient.stderr.join("")); + } finally { + server.close(); + if (rustClient.child.exitCode === null && !rustClient.child.killed) { + rustClient.child.kill("SIGTERM"); + await once(rustClient.child, "exit").catch(() => undefined); + } + } +}); + test("CultNet TypeScript and C# exchange schema frames over the shared RUDP socket transport", async () => { await buildCSharpInteropPeer(); const csharpPeer = spawnCSharpRudpServer(); @@ -2214,6 +2322,59 @@ function spawnRustRudpServer(options: RudpServerSpawnOptions = {}): RunningPytho return { child, ready, stderr }; } +function spawnRustRudpMessageServer(): RunningPythonRudpPeer { + const child = spawn(rustBinaryPath, [ + "rudp-serve-message-once", + "--bind-host", "127.0.0.1", + ], { + cwd: cultnetRsRoot, + env: process.env, + stdio: ["ignore", "pipe", "pipe"], + }); + const stderr: string[] = []; + let stdoutBuffer = ""; + + const ready = new Promise((resolveReady, rejectReady) => { + child.stdout.setEncoding("utf8"); + child.stdout.on("data", (chunk: string) => { + stdoutBuffer += chunk; + while (true) { + const newline = stdoutBuffer.indexOf("\n"); + if (newline === -1) { + break; + } + + const line = stdoutBuffer.slice(0, newline).trim(); + stdoutBuffer = stdoutBuffer.slice(newline + 1); + if (!line) { + continue; + } + + try { + const parsed = JSON.parse(line) as { status?: string; port?: number }; + if (parsed.status === "ready" && typeof parsed.port === "number") { + resolveReady(parsed.port); + } else if (parsed.status !== "ok") { + rejectReady(new Error(`Rust RUDP message peer emitted unexpected stdout: ${line}`)); + } + } catch (error) { + rejectReady(new Error(`Rust RUDP message peer emitted non-JSON stdout: ${line}`)); + } + } + }); + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk: string) => { + stderr.push(chunk); + }); + child.once("exit", (code, signal) => { + rejectReady(new Error(`Rust RUDP message peer exited before publishing a port (code=${code}, signal=${signal}).\n${stderr.join("")}`)); + }); + child.once("error", rejectReady); + }); + + return { child, ready, stderr }; +} + function spawnCSharpRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer { const child = spawn(dotnetCommand, [ csharpDllPath, @@ -2573,6 +2734,25 @@ function spawnRustRudpClient(remotePort: number): Omit { + const child = spawn(rustBinaryPath, [ + "rudp-dial-message-once", + "--target-host", "127.0.0.1", + "--target-port", String(remotePort), + ], { + cwd: cultnetRsRoot, + env: process.env, + stdio: ["ignore", "pipe", "pipe"], + }); + const stderr: string[] = []; + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk: string) => { + stderr.push(chunk); + }); + + return { child, stderr }; +} + async function bindUdpSocket(): Promise { const socket = dgram.createSocket("udp4"); await new Promise((resolveBind, rejectBind) => {