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 <noreply@anthropic.com>

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<CultNetRudpSession["receive"]>;
+    try {
+      result = this.#session.receive(packet, Date.now());
+    } catch {
+      this.#stats.packetsDropped += 1;
+      return;
+    }
+    try {
       if (result.reply) {
         this.#sendPacket(result.reply);
       }

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) {

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();
+  }
+});