Prove RUDP response resend across runtimes

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,

packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt

@@ -525,6 +525,7 @@ private fun rudpServeOnce(options: Map<String, String>) {
             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<String, String>) {
     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)) {

packages/cultnet-rs/examples/cultnet_interop_peer.rs

@@ -484,6 +484,7 @@ fn rudp_serve_once(options: &BTreeMap<String, String>) -> 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<String, String>) -> 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<TcpStream>,
     cache: &mut CultCache,

packages/cultnet-ts/test/interop/cultnet-interop.test.ts

@@ -68,6 +68,57 @@ let rustInteropPeerBuild: Promise<void> | undefined;
 let csharpInteropPeerBuild: Promise<void> | undefined;
 let kotlinInteropPeerBuild: Promise<void> | undefined;
 
+interface RudpServerInteropPeerSpec {
+  name: string;
+  connectionId: number;
+  build: () => Promise<void>;
+  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<void> {
   await buildRustInteropPeer();
   await buildCSharpInteropPeer();
@@ -1529,21 +1598,28 @@ async function bindUdpSocket(): Promise<Socket> {
 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<RudpLossyBridge> {
   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();
     },

tests/GameCult.Networking.InteropPeer/Program.cs

@@ -612,6 +612,7 @@ static void RudpServeOnce(Dictionary<string, string> 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<string, string> 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)