From 4209bbb9b6bcb3ebdb198ddc412c923c8352dc6a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 15 Jun 2026 01:40:43 +0100 Subject: [PATCH] Prove RUDP response resend across runtimes --- docs/cultnet-transport-parity.md | 7 +- .../kotlin/org/gamecult/cultmesh/CultMesh.kt | 10 ++ .../examples/cultnet_interop_peer.rs | 14 ++ .../test/interop/cultnet-interop.test.ts | 165 +++++++++++++----- .../Program.cs | 12 ++ 5 files changed, 161 insertions(+), 47 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index b65a9553..6b18e5b1 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -177,10 +177,9 @@ Current progress: exchange reliable ordered `schema` frames through the shared handshake and packet language. - The same harness now proves reliable ordered `schema` delivery survives one - dropped client data packet when TypeScript sends through RUDP to C#, Kotlin, - Python, and Rust. The drop happens below the transport connection through a - UDP bridge, so the recovery is RUDP resend/ack behavior, not application-level - retry. + dropped data packet in both directions between TypeScript and C#/Kotlin/Python/Rust. + The drops happen below the transport connection through a UDP bridge, so the + recovery is RUDP resend/ack behavior, not application-level retry. - 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 the full loss, reordering, diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index ceda8564..e482afdd 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt @@ -525,6 +525,7 @@ private fun rudpServeOnce(options: Map) { if (frame != null) { requireRudpFrame(frame, "schema", "ts-kotlin-client-state") transport.send("schema", "kotlin-server-state".toByteArray(StandardCharsets.UTF_8)) + pollRudpAfterSend(transport, 250) println("""{"status":"ok"}""") return } @@ -570,6 +571,15 @@ private fun rudpDialOnce(options: Map) { error("Timed out waiting for TypeScript RUDP response") } +private fun pollRudpAfterSend(transport: CultNetRudpSocketTransportConnection, durationMs: Long) { + val deadline = System.nanoTime() + durationMs * 1_000_000L + while (System.nanoTime() < deadline) { + transport.receiveOnce() + transport.pollResends() + Thread.sleep(5) + } +} + private fun requireRudpFrame(frame: CultNetTransportFrame, expectedChannelId: String, expectedPayload: String) { val expectedBytes = expectedPayload.toByteArray(StandardCharsets.UTF_8) if (frame.channelId != expectedChannelId || !frame.payload.contentEquals(expectedBytes)) { diff --git a/packages/cultnet-rs/examples/cultnet_interop_peer.rs b/packages/cultnet-rs/examples/cultnet_interop_peer.rs index 499ce0dd..56609de9 100644 --- a/packages/cultnet-rs/examples/cultnet_interop_peer.rs +++ b/packages/cultnet-rs/examples/cultnet_interop_peer.rs @@ -484,6 +484,7 @@ fn rudp_serve_once(options: &BTreeMap) -> Result<()> { )); } transport.send("schema", b"rust-server-state".to_vec())?; + poll_rudp_after_send(&mut transport, Duration::from_millis(250))?; print_json(&serde_json::json!({ "status": "ok" }))?; return Ok(()); } @@ -539,6 +540,19 @@ fn rudp_dial_once(options: &BTreeMap) -> Result<()> { Err(anyhow!("timed out waiting for TypeScript RUDP response")) } +fn poll_rudp_after_send( + transport: &mut CultNetRudpSocketTransportConnection, + duration: Duration, +) -> Result<()> { + let deadline = Instant::now() + duration; + while Instant::now() < deadline { + let _ = transport.receive_once()?; + transport.poll_resends()?; + thread::sleep(Duration::from_millis(5)); + } + Ok(()) +} + fn mutate_remote_note( transport: &mut TcpFramedTransportConnection, cache: &mut CultCache, diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 53c4fed2..bbade2d5 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -68,6 +68,57 @@ let rustInteropPeerBuild: Promise | undefined; let csharpInteropPeerBuild: Promise | undefined; let kotlinInteropPeerBuild: Promise | undefined; +interface RudpServerInteropPeerSpec { + name: string; + connectionId: number; + build: () => Promise; + spawn: () => RunningPythonRudpPeer; + joinPayload: string; + clientPayload: string; + expectedPayload: string; +} + +function rudpServerInteropPeers(): RudpServerInteropPeerSpec[] { + return [ + { + name: "Python", + connectionId: 0x55667788, + build: async () => undefined, + spawn: spawnPythonRudpPeer, + joinPayload: "ts-join", + clientPayload: "client-state", + expectedPayload: "python-state", + }, + { + name: "Rust", + connectionId: 0x22446688, + build: buildRustInteropPeer, + spawn: spawnRustRudpServer, + joinPayload: "ts-rust-join", + clientPayload: "ts-rust-client-state", + expectedPayload: "rust-server-state", + }, + { + name: "C#", + connectionId: 0x33557799, + build: buildCSharpInteropPeer, + spawn: spawnCSharpRudpServer, + joinPayload: "ts-csharp-join", + clientPayload: "ts-csharp-client-state", + expectedPayload: "csharp-server-state", + }, + { + name: "Kotlin", + connectionId: 0x446688aa, + build: buildKotlinInteropPeer, + spawn: spawnKotlinRudpServer, + joinPayload: "ts-kotlin-join", + clientPayload: "ts-kotlin-client-state", + expectedPayload: "kotlin-server-state", + }, + ]; +} + const pythonRudpPeerScript = String.raw` import socket import time @@ -989,46 +1040,7 @@ test("CultNet Kotlin and TypeScript exchange schema frames when Kotlin dials a T }); test("CultNet RUDP reliable schema delivery survives one dropped client data packet across runtimes", async () => { - const peers = [ - { - name: "Python", - connectionId: 0x55667788, - build: async () => undefined, - spawn: spawnPythonRudpPeer, - joinPayload: "ts-join", - clientPayload: "client-state", - expectedPayload: "python-state", - }, - { - name: "Rust", - connectionId: 0x22446688, - build: buildRustInteropPeer, - spawn: spawnRustRudpServer, - joinPayload: "ts-rust-join", - clientPayload: "ts-rust-client-state", - expectedPayload: "rust-server-state", - }, - { - name: "C#", - connectionId: 0x33557799, - build: buildCSharpInteropPeer, - spawn: spawnCSharpRudpServer, - joinPayload: "ts-csharp-join", - clientPayload: "ts-csharp-client-state", - expectedPayload: "csharp-server-state", - }, - { - name: "Kotlin", - connectionId: 0x446688aa, - build: buildKotlinInteropPeer, - spawn: spawnKotlinRudpServer, - joinPayload: "ts-kotlin-join", - clientPayload: "ts-kotlin-client-state", - expectedPayload: "kotlin-server-state", - }, - ]; - - for (const peer of peers) { + for (const peer of rudpServerInteropPeers()) { await peer.build(); const serverPeer = peer.spawn(); const clientSocket = await bindUdpSocket(); @@ -1046,7 +1058,7 @@ test("CultNet RUDP reliable schema delivery survives one dropped client data pac try { client = new CultNetRudpSocketTransportConnection({ - runtimeId: `ts-lossy-${peer.name.toLowerCase()}-rudp-interop`, + runtimeId: `ts-lossy-${peer.name.replace(/\W/g, "").toLowerCase()}-rudp-interop`, socket: clientSocket, mode: "client", remoteHost: "127.0.0.1", @@ -1084,6 +1096,63 @@ test("CultNet RUDP reliable schema delivery survives one dropped client data pac } }); +test("CultNet RUDP reliable schema response survives one dropped server data packet across runtimes", async () => { + for (const peer of rudpServerInteropPeers()) { + await peer.build(); + const serverPeer = peer.spawn(); + const clientSocket = await bindUdpSocket(); + const lossyBridge = await createRudpLossyBridge({ + remoteHost: "127.0.0.1", + remotePort: await serverPeer.ready, + shouldDropServerPacket: (wire) => { + const packet = decodeRudpPacket(wire); + return packet.packetType === "data" + && packet.channelId === "schema" + && Buffer.from(packet.payload ?? []).equals(Buffer.from(peer.expectedPayload)); + }, + }); + let client: CultNetRudpSocketTransportConnection | undefined; + + try { + client = new CultNetRudpSocketTransportConnection({ + runtimeId: `ts-server-loss-${peer.name.replace(/\W/g, "").toLowerCase()}-rudp-interop`, + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: lossyBridge.port, + connectionId: peer.connectionId, + initialSequence: 1, + resendDelayMs: 25, + resendPollMs: 5, + }); + + const receivedFrame = once(client, "frame"); + client.connect(Buffer.from(peer.joinPayload)); + await waitFor(() => client?.connected === true, `TypeScript RUDP client to complete ${peer.name} handshake through server-loss bridge`); + client.send("schema", Buffer.from(peer.clientPayload)); + + const [frame] = await withTimeout(receivedFrame, 2_000, `${peer.name} RUDP response frame after dropped server data packet`); + assert.equal(frame.channelId, "schema"); + assert.deepEqual(Buffer.from(frame.payload), Buffer.from(peer.expectedPayload)); + assert.equal(lossyBridge.droppedServerPackets, 1, `${peer.name} bridge should drop exactly one server data packet`); + + const [exitCode] = await withTimeout(once(serverPeer.child, "exit"), 2_000, `${peer.name} RUDP peer exit after server resend`); + assert.equal(exitCode, 0, serverPeer.stderr.join("")); + } finally { + if (client) { + client.close(); + } else { + clientSocket.close(); + } + lossyBridge.close(); + if (serverPeer.child.exitCode === null && !serverPeer.child.killed) { + serverPeer.child.kill("SIGTERM"); + await once(serverPeer.child, "exit").catch(() => undefined); + } + } + } +}); + async function buildInteropPeers(): Promise { await buildRustInteropPeer(); await buildCSharpInteropPeer(); @@ -1529,21 +1598,28 @@ async function bindUdpSocket(): Promise { interface RudpLossyBridge { port: number; readonly droppedClientPackets: number; + readonly droppedServerPackets: number; close(): void; } async function createRudpLossyBridge(options: { remoteHost: string; remotePort: number; - shouldDropClientPacket: (wire: Buffer) => boolean; + shouldDropClientPacket?: (wire: Buffer) => boolean; + shouldDropServerPacket?: (wire: Buffer) => boolean; }): Promise { const socket = dgram.createSocket("udp4"); const remote = { address: options.remoteHost, port: options.remotePort }; let client: { address: string; port: number } | undefined; let droppedClientPackets = 0; + let droppedServerPackets = 0; socket.on("message", (wire, sender) => { if (sender.address === remote.address && sender.port === remote.port) { + if (droppedServerPackets === 0 && options.shouldDropServerPacket?.(wire)) { + droppedServerPackets += 1; + return; + } if (client) { socket.send(wire, client.port, client.address); } @@ -1551,7 +1627,7 @@ async function createRudpLossyBridge(options: { } client ??= { address: sender.address, port: sender.port }; - if (droppedClientPackets === 0 && options.shouldDropClientPacket(wire)) { + if (droppedClientPackets === 0 && options.shouldDropClientPacket?.(wire)) { droppedClientPackets += 1; return; } @@ -1572,6 +1648,9 @@ async function createRudpLossyBridge(options: { get droppedClientPackets() { return droppedClientPackets; }, + get droppedServerPackets() { + return droppedServerPackets; + }, close() { socket.close(); }, diff --git a/tests/GameCult.Networking.InteropPeer/Program.cs b/tests/GameCult.Networking.InteropPeer/Program.cs index be918e2e..1f6eb374 100644 --- a/tests/GameCult.Networking.InteropPeer/Program.cs +++ b/tests/GameCult.Networking.InteropPeer/Program.cs @@ -612,6 +612,7 @@ static void RudpServeOnce(Dictionary options) { RequireRudpFrame(frame, "schema", "ts-csharp-client-state"); transport.Send("schema", Ascii("csharp-server-state")); + PollRudpAfterSend(transport, TimeSpan.FromMilliseconds(250)); WriteJsonLine(new { status = "ok" }); return; } @@ -667,6 +668,17 @@ static void RudpDialOnce(Dictionary options) throw new TimeoutException("Timed out waiting for TypeScript RUDP response."); } +static void PollRudpAfterSend(CultNetRudpSocketTransportConnection transport, TimeSpan duration) +{ + var deadline = DateTimeOffset.UtcNow.Add(duration); + while (DateTimeOffset.UtcNow < deadline) + { + _ = transport.ReceiveOnce(); + transport.PollResends(); + Thread.Sleep(5); + } +} + static void RequireRudpFrame(CultNetTransportFrame frame, string expectedChannelId, string expectedPayload) { if (!string.Equals(frame.ChannelId, expectedChannelId, StringComparison.Ordinal)