From ecb80b4fd8e838725efc10b991c3d68d45fd7bfb Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 17:44:37 +0200 Subject: [PATCH] cultnet-ts: RUDP socket transport drops and counts strays instead of emitting error A foreign connection id, a malformed frame or a refused packet was emitted as an error event; with no listener that throws and ends the process. They are now counted in stats.packetsDropped. The id is checked before the peer endpoint or session state is touched, and an unadmitted first packet in server mode no longer claims the endpoint. Listener faults still surface as error events. Co-Authored-By: Claude Sonnet 5 --- packages/cultnet-ts/src/rudp.ts | 43 +++++++++++++--- packages/cultnet-ts/src/transport.ts | 3 ++ packages/cultnet-ts/test/cultnet.test.ts | 63 ++++++++++++++++++++++++ 3 files changed, 102 insertions(+), 7 deletions(-) diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 56dc67fa..89e478c2 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -807,6 +807,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement bytesSent: 0, framesReceived: 0, framesSent: 0, + packetsDropped: 0, }; constructor(options: CultNetRudpSocketTransportOptions) { @@ -922,15 +923,29 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement #receiveDatagram(wire: Buffer, remote: RemoteInfo): void { this.#stats.bytesReceived += wire.length; + // What a datagram carries is the sender's business, not a fault of this + // process: a malformed frame, another session's connection id and a packet + // the session refuses are dropped and counted, never emitted as "error" + // (an "error" with no listener throws and ends the process). The id is + // checked before the peer endpoint or session state is touched, so a stray + // Connect can neither move the endpoint nor reset the session. let packet: CultNetRudpPacket; try { packet = decodeRudpPacket(wire); - } catch (error) { - this.emit("error", error instanceof Error ? error : new Error(String(error))); + } catch { + this.#stats.packetsDropped += 1; + return; + } + if (packet.connectionId !== this.#session.connectionId) { + this.#stats.packetsDropped += 1; return; } if (!this.#remoteHost || this.#remotePort === undefined) { + if (this.#mode === "server" && packet.packetType !== "connect") { + this.#stats.packetsDropped += 1; + return; + } this.#remoteHost = remote.address; this.#remotePort = remote.port; } else if (remote.address !== this.#remoteHost || remote.port !== this.#remotePort) { @@ -938,18 +953,32 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement this.#remoteHost = remote.address; this.#remotePort = remote.port; } else { + this.#stats.packetsDropped += 1; return; } } - try { - if (this.#mode === "server" && packet.packetType === "connect") { - this.#session.resetPeerState(); - this.#sendPacket(this.#session.acceptConnect(packet, Date.now())); + if (this.#mode === "server" && packet.packetType === "connect") { + this.#session.resetPeerState(); + let accept: CultNetRudpPacket; + try { + accept = this.#session.acceptConnect(packet, Date.now()); + } catch { + this.#stats.packetsDropped += 1; return; } + this.#sendPacket(accept); + return; + } - const result = this.#session.receive(packet, Date.now()); + let result: ReturnType; + try { + result = this.#session.receive(packet, Date.now()); + } catch { + this.#stats.packetsDropped += 1; + return; + } + try { if (result.reply) { this.#sendPacket(result.reply); } diff --git a/packages/cultnet-ts/src/transport.ts b/packages/cultnet-ts/src/transport.ts index a1b9b3ce..f900aa23 100644 --- a/packages/cultnet-ts/src/transport.ts +++ b/packages/cultnet-ts/src/transport.ts @@ -14,6 +14,8 @@ export interface CultNetTransportStats { bytesSent: number; framesReceived: number; framesSent: number; + /** Datagrams read and discarded because they belong to no session on this transport. */ + packetsDropped: number; } export interface CultNetReconnectPolicy { @@ -192,6 +194,7 @@ export class TcpFramedTransportConnection extends EventEmitter implements CultNe bytesSent: 0, framesReceived: 0, framesSent: 0, + packetsDropped: 0, }; constructor(stream: Duplex, profile: CultNetTransportProfile) { diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 2f540a86..735418e0 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1867,3 +1867,66 @@ test("rudp receiver evicts stranded fragment sets instead of refusing new ones", assert.equal(second.delivered.length, 1); assert.equal(Buffer.from(second.delivered[0]!.payload).toString(), "hello"); }); + +test("rudp socket transport drops strays without an error event and keeps serving its peer", async () => { + const serverSocket = await bindUdpSocket(); + const clientSocket = await bindUdpSocket(); + const strayHost = await bindUdpSocket(); + const connectionId = 0x10203050; + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendPollMs: 5, + }); + const client = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-client", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: udpPort(serverSocket), + connectionId, + resendPollMs: 5, + }); + + try { + // No "error" listener on either side: an unhandled "error" event throws + // and would end the process, which is what a stray must never do. + const frames: string[] = []; + server.on("frame", (frame) => frames.push(Buffer.from(frame.payload).toString("utf8"))); + client.connect(); + await waitFor(() => client.connected && server.connected, "RUDP socket handshake"); + + const foreign = new CultNetRudpSession({ connectionId: 0x0badf10e }); + const strays = [ + Buffer.from("not a rudp packet"), + Buffer.alloc(0), + Buffer.from(encodeRudpPacket(foreign.createConnect(0, Buffer.from("foreign")))), + ]; + for (const stray of strays) { + // From an unrelated address, and from the peer's own address. + strayHost.send(stray, udpPort(serverSocket), "127.0.0.1"); + clientSocket.send(stray, udpPort(serverSocket), "127.0.0.1"); + } + await waitFor(() => server.stats.packetsDropped >= strays.length * 2, "strays counted"); + + client.send("schema", Buffer.from("still here", "utf8")); + await waitFor(() => frames.length === 1, "the real frame"); + assert.deepEqual(frames, ["still here"]); + + // The client side too: a moved flow answers from the server address. + for (const stray of strays) { + serverSocket.send(stray, udpPort(clientSocket), "127.0.0.1"); + } + await waitFor(() => client.stats.packetsDropped >= strays.length, "client strays counted"); + const clientFrames: string[] = []; + client.on("frame", (frame) => clientFrames.push(Buffer.from(frame.payload).toString("utf8"))); + server.send("schema", Buffer.from("server still here", "utf8")); + await waitFor(() => clientFrames.length === 1, "the real server frame"); + } finally { + strayHost.close(); + client.close(); + server.close(); + } +});