diff --git a/packages/cultnet-ts/CHANGELOG.md b/packages/cultnet-ts/CHANGELOG.md index b87eebd3..eae823c4 100644 --- a/packages/cultnet-ts/CHANGELOG.md +++ b/packages/cultnet-ts/CHANGELOG.md @@ -7,7 +7,7 @@ All notable changes to `cultnet-ts` are recorded here, following ### Added -- `CultNetRudpSocketTransportServer`: one UDP socket serving many peers, each remote address and port with its own reliable session (`peerConnected`, `frame`, `peerDisconnected` events; `send(peer, ...)` answers on the peer a frame came from). A second client no longer replaces the first; idle peers are reaped after `peerTimeoutMs` (default 60 s) and Connects beyond `maxPeers` (default 256) are dropped. Matches the Rust `CultNetRudpServerHub` and C# `CultNetRudpSocketTransportServer`; the wire format is unchanged. `CultNetRudpSocketTransportConnection` in `server` mode still serves exactly one peer. +- `CultNetRudpSocketTransportServer`: one UDP socket serving many peers, each remote address and port with its own reliable session (`peerConnected`, `frame`, `peerDisconnected` events; `send(peer, ...)` answers on the peer a frame came from). A second client no longer replaces the first; Connects beyond `maxPeers` (default 256) are dropped, and `close()` resolves once every peer's goodbye has left the socket. The shape follows the Rust `CultNetRudpServerHub` and C# `CultNetRudpSocketTransportServer`, and the wire format is unchanged. The optional `peerTimeoutMs` (idle peers ended and told so; default off) exists only in TypeScript. `CultNetRudpSocketTransportConnection` in `server` mode still serves exactly one peer. ### Breaking diff --git a/packages/cultnet-ts/src/rudp-server.ts b/packages/cultnet-ts/src/rudp-server.ts index e6e68464..37e8695a 100644 --- a/packages/cultnet-ts/src/rudp-server.ts +++ b/packages/cultnet-ts/src/rudp-server.ts @@ -34,10 +34,12 @@ export interface CultNetRudpSocketServerOptions { /** A Connect from an endpoint beyond this many peers is dropped. Default 256. */ maxPeers?: number; /** - * A peer that has sent nothing for this long is ended and told so. Default 60 000; - * `Infinity` keeps idle peers until they leave. + * A peer that has sent nothing for this long is ended and told so. TypeScript only: the + * Rust and C# servers never reap. Default `Infinity`, which keeps idle peers until they leave. */ peerTimeoutMs?: number; + /** How long `close()` waits for the goodbyes to leave the socket. Default 1000. */ + closeGraceMs?: number; } /** One remote endpoint's session. A peer that reconnects from the same endpoint is a new object. */ @@ -90,6 +92,7 @@ export class CultNetRudpSocketTransportServer extends EventEmitter { readonly #maxPeers: number; readonly #peerTimeoutMs: number; #closed = false; + #closing = false; readonly #stats: CultNetTransportStats = { bytesReceived: 0, bytesSent: 0, @@ -104,7 +107,7 @@ export class CultNetRudpSocketTransportServer extends EventEmitter { this.#options = options; this.#socket = options.socket; this.#maxPeers = options.maxPeers ?? 256; - this.#peerTimeoutMs = options.peerTimeoutMs ?? 60_000; + this.#peerTimeoutMs = options.peerTimeoutMs ?? Infinity; if (!Number.isInteger(this.#maxPeers) || this.#maxPeers < 1) { throw new Error("RUDP server maxPeers must be a positive integer."); } @@ -169,13 +172,23 @@ export class CultNetRudpSocketTransportServer extends EventEmitter { this.#endPeer(entry, entry.session.createDisconnect(Uint8Array.from(reason))); } - /** Ends every peer and releases the socket. */ - close(): void { + /** + * Ends every peer and releases the socket once the goodbyes have left it (bounded by + * `closeGraceMs`, default 1000): closing the socket at once would drop them. + */ + async close(): Promise { if (this.#closed) return; + this.#closing = true; clearInterval(this.#timer); - for (const entry of [...this.#entries.values()]) { - this.#endPeer(entry, entry.session.createDisconnect(utf8("server closed"))); - } + const sent = [...this.#entries.values()].map((entry) => + this.#endPeer(entry, entry.session.createDisconnect(utf8("server closed")))); + let bound: NodeJS.Timeout | undefined; + await Promise.race([ + Promise.all(sent), + new Promise((resolve) => { bound = setTimeout(resolve, this.#options.closeGraceMs ?? 1_000); }), + ]); + clearTimeout(bound); + if (this.#closed) return; this.#closed = true; this.#socket.close(); } @@ -189,17 +202,24 @@ export class CultNetRudpSocketTransportServer extends EventEmitter { } /** Removes the peer, tells the caller, and says goodbye on the wire. Idempotent per entry. */ - #endPeer(entry: PeerEntry, goodbye: CultNetRudpPacket): void { - if (this.#entries.get(entry.key) !== entry) return; + #endPeer(entry: PeerEntry, goodbye: CultNetRudpPacket): Promise { + if (this.#entries.get(entry.key) !== entry) return Promise.resolve(); this.#entries.delete(entry.key); this.emit("peerDisconnected", { peer: entry.peer, reason: goodbye.payload }); // Best-effort: the session is over whether or not the peer hears it. const wire = encodeRudpPacket(goodbye); this.#stats.bytesSent += wire.length; - sendRudpDatagram(this.#socket, wire, entry.peer.remotePort, entry.peer.remoteHost, () => {}); + return new Promise((resolve) => { + try { + this.#socket.send(wire, entry.peer.remotePort, entry.peer.remoteHost, () => resolve()); + } catch { + resolve(); + } + }); } #receiveDatagram(wire: Buffer, remote: RemoteInfo): void { + if (this.#closing) return; this.#stats.bytesReceived += wire.length; // What a datagram carries is the sender's business: a malformed frame, another // session's connection id and a sender no Connect admitted are dropped and counted. diff --git a/packages/cultnet-ts/test/rudp-server.test.ts b/packages/cultnet-ts/test/rudp-server.test.ts index d35ca9f7..6ba976d3 100644 --- a/packages/cultnet-ts/test/rudp-server.test.ts +++ b/packages/cultnet-ts/test/rudp-server.test.ts @@ -97,7 +97,7 @@ test("two clients on different ports talk through one server socket and neither assert.equal(a.closed() || b.closed(), false); assert.equal(server.peers.length, 2); } finally { - a.connection.close(); b.connection.close(); server.close(); + a.connection.close(); b.connection.close(); await server.close(); } }); @@ -120,7 +120,7 @@ test("a client reconnecting from the same endpoint replaces only its own session await waitFor(() => a.received.includes("re:after") && b.received.includes("re:still"), "replies after the reconnect"); assert.equal(b.closed(), false); } finally { - a.connection.close(); b.connection.close(); server.close(); + a.connection.close(); b.connection.close(); await server.close(); } }); @@ -147,7 +147,7 @@ test("a peer that leaves, or goes silent, is dropped and the others keep their s steady.connection.send("schema", enc("alive")); await waitFor(() => steady.received.includes("re:alive"), "the surviving peer's reply"); } finally { - silent.connection.close(); steady.connection.close(); server.close(); + silent.connection.close(); steady.connection.close(); await server.close(); } }); @@ -167,6 +167,30 @@ test("Connects beyond maxPeers and datagrams from unadmitted endpoints are dropp first.connection.send("schema", enc("ok")); await waitFor(() => first.received.includes("re:ok"), "the admitted peer's reply"); } finally { - stranger.close(); first.connection.close(); server.close(); + stranger.close(); first.connection.close(); await server.close(); + } +}); + +test("close() delivers a Disconnect to every peer before the socket is released", async () => { + const { server, port } = await startServer(); + const peers = await Promise.all([0, 1, 2].map(async () => { + const socket = await bind(); + const session = new CultNetRudpSession({ connectionId }); + const types: string[] = []; + socket.on("message", (wire) => { + const packet = decodeRudpPacket(wire); + types.push(packet.packetType); + session.receive(packet, Date.now()); + }); + socket.send(encodeRudpPacket(session.createConnect(Date.now())), port, "127.0.0.1"); + await waitFor(() => session.connected, "raw handshake"); + return { socket, types }; + })); + try { + assert.equal(server.peers.length, 3); + await server.close(); + await waitFor(() => peers.every((peer) => peer.types.includes("disconnect")), "every peer to observe the Disconnect"); + } finally { + peers.forEach((peer) => peer.socket.close()); } });