From 37a2b002dbc358d6c26019c6461ff733df99b476 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 01:30:29 +0100 Subject: [PATCH] cultnet-ts: CultNetRudpSocketTransportServer serves many peers on one socket, one reliable session per remote endpoint (matches Rust server hub and C# socket server; wire unchanged) Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-ts/src/index.ts | 6 + packages/cultnet-ts/src/rudp-server.ts | 370 +++++++++++++++++++ packages/cultnet-ts/src/rudp.ts | 8 +- packages/cultnet-ts/test/rudp-server.test.ts | 172 +++++++++ 4 files changed, 552 insertions(+), 4 deletions(-) create mode 100644 packages/cultnet-ts/src/rudp-server.ts create mode 100644 packages/cultnet-ts/test/rudp-server.test.ts diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 1e856822..c321bd07 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -123,6 +123,12 @@ export { type CultNetRudpSessionOptions, type RudpTransportProfileOptions, } from "./rudp"; +export { + CultNetRudpSocketTransportServer, + type CultNetRudpSocketServerFrame, + type CultNetRudpSocketServerOptions, + type CultNetRudpSocketServerPeer, +} from "./rudp-server"; export { invokeCultNetOperation, startCultNetOperationServer, diff --git a/packages/cultnet-ts/src/rudp-server.ts b/packages/cultnet-ts/src/rudp-server.ts new file mode 100644 index 00000000..e6e68464 --- /dev/null +++ b/packages/cultnet-ts/src/rudp-server.ts @@ -0,0 +1,370 @@ +import { EventEmitter } from "node:events"; +import { type RemoteInfo, type Socket } from "node:dgram"; + +import type { CultNetTransportProfile } from "./contracts"; +import { + CultNetRudpSession, + MAX_UDP4_DATAGRAM_BYTES, + MAX_UDP6_DATAGRAM_BYTES, + RUDP_FIXED_HEADER_BYTES, + channelOptions, + createRudpTransportProfile, + decodeRudpPacket, + encodeRudpPacket, + isPermanentSendError, + sendRudpDatagram, + type CultNetRudpPacket, +} from "./rudp"; +import type { CultNetReconnectPolicy, CultNetTransportFrame, CultNetTransportStats } from "./transport"; + +export interface CultNetRudpSocketServerOptions { + runtimeId: string; + /** A bound UDP socket. The server owns it from here: `close()` closes it. */ + socket: Socket; + connectionId: number; + /** The first local sequence of every peer session. Unset, each session draws its own. */ + initialSequence?: number; + resendDelayMs?: number; + resendPollMs?: number; + transportId?: string; + maxPayloadBytes?: number; + maxFragmentBytes?: number; + maxPendingReliablePackets?: number; + reconnectPolicy?: CultNetReconnectPolicy; + /** 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. + */ + peerTimeoutMs?: number; +} + +/** One remote endpoint's session. A peer that reconnects from the same endpoint is a new object. */ +export interface CultNetRudpSocketServerPeer { + readonly remoteHost: string; + readonly remotePort: number; + /** The exact bytes of the Connect that opened this peer's session. */ + readonly connectPayload: Uint8Array; + readonly connected: boolean; + readonly outstandingReliablePacketCount: number; +} + +export interface CultNetRudpSocketServerFrame extends CultNetTransportFrame { + peer: CultNetRudpSocketServerPeer; +} + +type PeerEntry = { + peer: CultNetRudpSocketServerPeer; + session: CultNetRudpSession; + key: string; +}; + +/** The goodbye for a session that refused a packet; built after the reset, or its ack would acknowledge the refused frame. */ +function refusedGoodbye(session: CultNetRudpSession): CultNetRudpPacket { + session.resetPeerState(); + return session.createDisconnect(utf8("session refused a packet")); +} + +const utf8 = (text: string) => new TextEncoder().encode(text); + +/** + * One UDP socket serving many peers. Every remote address and port gets its own reliable + * session, opened by its Connect: a second client never takes the first one's session, and + * a client that reconnects from the same endpoint replaces only its own. The wire format is + * the single-peer transport's; the shape matches the Rust `CultNetRudpServerHub` and the C# + * `CultNetRudpSocketTransportServer`. + * + * Reply to a peer with `send(peer, ...)` on the peer a frame arrived from. + * + * Events: `peerConnected` (peer), `frame` ({ peer, channelId, payload }), `pong` ({ peer, payload }), + * `peerDisconnected` ({ peer, reason }), `close`, `error`. + */ +export class CultNetRudpSocketTransportServer extends EventEmitter { + readonly profile: CultNetTransportProfile; + readonly #socket: Socket; + readonly #options: CultNetRudpSocketServerOptions; + readonly #entries = new Map(); + readonly #timer: NodeJS.Timeout; + readonly #maxDatagramBytes: number; + readonly #maxPeers: number; + readonly #peerTimeoutMs: number; + #closed = false; + readonly #stats: CultNetTransportStats = { + bytesReceived: 0, + bytesSent: 0, + framesReceived: 0, + framesSent: 0, + packetsDropped: 0, + sendFailures: 0, + }; + + constructor(options: CultNetRudpSocketServerOptions) { + super(); + this.#options = options; + this.#socket = options.socket; + this.#maxPeers = options.maxPeers ?? 256; + this.#peerTimeoutMs = options.peerTimeoutMs ?? 60_000; + if (!Number.isInteger(this.#maxPeers) || this.#maxPeers < 1) { + throw new Error("RUDP server maxPeers must be a positive integer."); + } + const address = this.#socket.address(); + this.#maxDatagramBytes = + typeof address !== "string" && String(address.family).endsWith("6") ? MAX_UDP6_DATAGRAM_BYTES : MAX_UDP4_DATAGRAM_BYTES; + this.profile = createRudpTransportProfile(options.runtimeId, { + transportId: options.transportId, + host: typeof address === "string" ? undefined : address.address, + port: typeof address === "string" ? undefined : address.port, + maxPayloadBytes: options.maxPayloadBytes, + maxFragmentBytes: options.maxFragmentBytes, + maxPendingReliablePackets: options.maxPendingReliablePackets, + reconnectPolicy: options.reconnectPolicy, + }); + + this.#socket.on("message", (wire, remote) => this.#receiveDatagram(wire, remote)); + this.#socket.on("close", () => { + if (this.#closed) return; + this.#closed = true; + clearInterval(this.#timer); + this.#entries.clear(); + this.emit("close"); + }); + this.#socket.on("error", (error) => this.emit("error", error instanceof Error ? error : new Error(String(error)))); + this.#timer = setInterval(() => this.#tick(), options.resendPollMs ?? 25); + this.#timer.unref?.(); + } + + /** The peers with a live session. */ + get peers(): CultNetRudpSocketServerPeer[] { + return [...this.#entries.values()].map((entry) => entry.peer); + } + + get stats(): CultNetTransportStats { + return { ...this.#stats }; + } + + /** Sends to one peer. Throws when that peer's session has ended or was replaced. */ + send(peer: CultNetRudpSocketServerPeer, channelId: string, payload: Uint8Array): void { + const entry = this.#liveEntry(peer); + const largestPayload = + this.#options.maxFragmentBytes === undefined ? payload.byteLength : Math.min(payload.byteLength, this.#options.maxFragmentBytes); + if (RUDP_FIXED_HEADER_BYTES + Buffer.byteLength(channelId, "utf8") + largestPayload > this.#maxDatagramBytes) { + throw Object.assign(new Error("RUDP datagram is larger than UDP can carry."), { code: "EMSGSIZE" }); + } + const packets = entry.session.sendMany(channelId, payload, { + ...channelOptions(channelId), + nowMs: Date.now(), + maxFragmentBytes: this.#options.maxFragmentBytes, + }); + for (const packet of packets) { + const thrown = this.#transmit(entry, packet); + if (thrown) throw thrown; + } + this.#stats.framesSent += 1; + } + + /** Ends one peer's session, telling it why. */ + disconnect(peer: CultNetRudpSocketServerPeer, reason: Uint8Array = new Uint8Array()): void { + const entry = this.#liveEntry(peer); + this.#endPeer(entry, entry.session.createDisconnect(Uint8Array.from(reason))); + } + + /** Ends every peer and releases the socket. */ + close(): void { + if (this.#closed) return; + clearInterval(this.#timer); + for (const entry of [...this.#entries.values()]) { + this.#endPeer(entry, entry.session.createDisconnect(utf8("server closed"))); + } + this.#closed = true; + this.#socket.close(); + } + + #liveEntry(peer: CultNetRudpSocketServerPeer): PeerEntry { + const entry = this.#entries.get(`${peer.remoteHost}:${peer.remotePort}`); + if (!entry || entry.peer !== peer) { + throw new Error("RUDP peer is not connected."); + } + return entry; + } + + /** 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; + 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, () => {}); + } + + #receiveDatagram(wire: Buffer, remote: RemoteInfo): void { + 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. + let packet: CultNetRudpPacket; + try { + packet = decodeRudpPacket(wire); + } catch { + this.#stats.packetsDropped += 1; + return; + } + if (packet.connectionId !== this.#options.connectionId) { + this.#stats.packetsDropped += 1; + return; + } + const key = `${remote.address}:${remote.port}`; + const existing = this.#entries.get(key); + + if (packet.packetType === "connect") { + this.#receiveConnect(packet, remote, key, existing); + return; + } + if (!existing) { + this.#stats.packetsDropped += 1; + return; + } + this.#receiveFromPeer(existing, packet); + } + + #receiveConnect(packet: CultNetRudpPacket, remote: RemoteInfo, key: string, existing: PeerEntry | undefined): void { + if (existing?.session.connectRepeats(packet)) { + // A repeat of the Connect it accepted: the Accept owed, nothing restarted. + try { + this.#sendInPoll(existing, existing.session.acceptConnect(packet, Date.now())); + } catch { + this.#stats.packetsDropped += 1; + this.#endPeer(existing, refusedGoodbye(existing.session)); + } + return; + } + if (!existing && this.#entries.size >= this.#maxPeers) { + this.#stats.packetsDropped += 1; + return; + } + const session = new CultNetRudpSession({ + connectionId: this.#options.connectionId, + initialSequence: this.#options.initialSequence, + resendDelayMs: this.#options.resendDelayMs, + maxPendingReliablePackets: this.#options.maxPendingReliablePackets, + }); + let accept: CultNetRudpPacket; + try { + accept = session.acceptConnect(packet, Date.now()); + } catch { + this.#stats.packetsDropped += 1; + return; + } + const entry: PeerEntry = { + key, + session, + peer: { + remoteHost: remote.address, + remotePort: remote.port, + connectPayload: Uint8Array.from(packet.payload ?? []), + get connected() { return session.connected; }, + get outstandingReliablePacketCount() { return session.outstandingReliablePacketCount; }, + }, + }; + if (this.#transmit(entry, accept)) { + // The peer cannot be answered, so no session starts. + this.#stats.packetsDropped += 1; + return; + } + if (existing) { + // Same endpoint, new Connect generation: only this peer's old session ends. + this.#entries.delete(key); + this.emit("peerDisconnected", { peer: existing.peer, reason: utf8("replaced by a new Connect generation") }); + } + this.#entries.set(key, entry); + this.emit("peerConnected", entry.peer); + } + + #receiveFromPeer(entry: PeerEntry, packet: CultNetRudpPacket): void { + const session = entry.session; + let result: ReturnType; + try { + result = session.receive(packet, Date.now()); + } catch { + // receive() has already recorded the packet's reliable sequence, so the session + // cannot be kept: a retransmit would be acknowledged and the frame silently lost. + this.#stats.packetsDropped += 1; + this.#endPeer(entry, refusedGoodbye(session)); + return; + } + try { + if (result.reply) this.#sendInPoll(entry, result.reply); + for (const ready of result.readyToSend ?? []) this.#sendInPoll(entry, ready); + if (result.pong) { + this.emit("pong", { peer: entry.peer, payload: result.pongPayload ?? new Uint8Array() }); + } + for (const frame of result.delivered) { + this.#stats.framesReceived += 1; + this.emit("frame", { peer: entry.peer, channelId: frame.channelId, payload: frame.payload }); + } + if (result.disconnected) { + if (this.#entries.get(entry.key) === entry) { + this.#entries.delete(entry.key); + this.emit("peerDisconnected", { peer: entry.peer, reason: result.disconnectReason ?? new Uint8Array(0) }); + } + return; + } + if (this.#entries.get(entry.key) === entry && (packet.packetType === "data" || result.delivered.length > 0)) { + this.#sendInPoll(entry, session.createAckForReceived(packet.sequence)); + } + } catch (error) { + this.emit("error", error instanceof Error ? error : new Error(String(error))); + } + } + + #tick(): void { + const now = Date.now(); + for (const entry of [...this.#entries.values()]) { + if (this.#entries.get(entry.key) !== entry) continue; + if (entry.session.checkTimeout(now, this.#peerTimeoutMs)) { + this.#endPeer(entry, entry.session.createDisconnect(utf8("session timed out"))); + continue; + } + const generation = entry.session.generation; + for (const packet of entry.session.dueResends(now)) { + this.#sendInPoll(entry, packet); + // A resend that can never be sent ended the session and what it owed. + if (entry.session.generation !== generation || this.#entries.get(entry.key) !== entry) break; + } + } + } + + /** A send with no caller to hand a failure to: it ends that peer, never the poll. */ + #sendInPoll(entry: PeerEntry, packet: CultNetRudpPacket): void { + const thrown = this.#transmit(entry, packet); + if (thrown) this.#sendFailed(entry, thrown); + } + + /** Sends one packet to the peer and returns the failure the send threw, if any. */ + #transmit(entry: PeerEntry, packet: CultNetRudpPacket): Error | undefined { + const wire = encodeRudpPacket(packet); + this.#stats.bytesSent += wire.length; + let thrown: Error | undefined; + let sending = true; + const generation = entry.session.generation; + sendRudpDatagram(this.#socket, wire, entry.peer.remotePort, entry.peer.remoteHost, (error) => { + if (sending) { + thrown = error; + } else if (entry.session.generation === generation) { + this.#sendFailed(entry, error); + } + }); + sending = false; + return thrown; + } + + /** A datagram that can never be sent ends that peer; any other failure is a lost datagram. */ + #sendFailed(entry: PeerEntry, error: Error): void { + if (isPermanentSendError(error)) { + this.#endPeer(entry, entry.session.endUnsendableSession(error)); + return; + } + this.#stats.sendFailures += 1; + } +} diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index e68293d3..89939ef3 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -29,11 +29,11 @@ import { const RUDP_MAGIC = [0x43, 0x4e, 0x52, 0x30] as const; // CNR0 const RUDP_VERSION = 0; -const RUDP_FIXED_HEADER_BYTES = 36; +export const RUDP_FIXED_HEADER_BYTES = 36; /// The largest UDP payload a datagram can carry: 65,535 less the UDP header and, /// over IPv4, the minimal IP header. IPv6's header is outside its payload length. -const MAX_UDP4_DATAGRAM_BYTES = 65_507; -const MAX_UDP6_DATAGRAM_BYTES = 65_527; +export const MAX_UDP4_DATAGRAM_BYTES = 65_507; +export const MAX_UDP6_DATAGRAM_BYTES = 65_527; const MAX_CHANNEL_ID_BYTES = 255; export const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS = 32; const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096; @@ -1588,7 +1588,7 @@ export function createRudpTransportProfile( }; } -function channelOptions(channelId: string): { reliable: boolean; ordered: boolean; sequenced: boolean } { +export function channelOptions(channelId: string): { reliable: boolean; ordered: boolean; sequenced: boolean } { if (channelId === "schema") { return { reliable: true, ordered: true, sequenced: false }; } diff --git a/packages/cultnet-ts/test/rudp-server.test.ts b/packages/cultnet-ts/test/rudp-server.test.ts new file mode 100644 index 00000000..d35ca9f7 --- /dev/null +++ b/packages/cultnet-ts/test/rudp-server.test.ts @@ -0,0 +1,172 @@ +// One server socket serves many peers, each with its own reliable session. +import assert from "node:assert/strict"; +import test from "node:test"; +import dgram, { type Socket } from "node:dgram"; +import { + CultNetRudpSession, + CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportServer, + decodeRudpPacket, + encodeRudpPacket, + type CultNetRudpSocketServerPeer, +} from "../src"; + +const connectionId = 0x10203091; +const enc = (text: string) => new TextEncoder().encode(text); +const dec = (bytes: Uint8Array) => new TextDecoder().decode(bytes); +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function bind(): Promise { + const socket = dgram.createSocket("udp4"); + await new Promise((resolve) => socket.bind(0, "127.0.0.1", resolve)); + return socket; +} + +async function waitFor(predicate: () => boolean, description: string): Promise { + const startedAt = Date.now(); + while (!predicate()) { + if (Date.now() - startedAt > 3_000) throw new Error(`Timed out waiting for ${description}.`); + await sleep(5); + } +} + +type Client = { connection: CultNetRudpSocketTransportConnection; socket: Socket; received: string[]; closed: () => boolean }; + +async function startServer(extra: Record = {}) { + const socket = await bind(); + const server = new CultNetRudpSocketTransportServer({ + runtimeId: "many-peer-server", socket, connectionId, resendDelayMs: 10, resendPollMs: 5, ...extra, + }); + const requests: { peer: CultNetRudpSocketServerPeer; text: string }[] = []; + const connected: CultNetRudpSocketServerPeer[] = []; + const left: { peer: CultNetRudpSocketServerPeer; reason: string }[] = []; + server.on("peerConnected", (peer: CultNetRudpSocketServerPeer) => connected.push(peer)); + server.on("peerDisconnected", ({ peer, reason }: { peer: CultNetRudpSocketServerPeer; reason: Uint8Array }) => left.push({ peer, reason: dec(reason) })); + // Every request is answered on the peer it arrived from. + server.on("frame", ({ peer, payload }: { peer: CultNetRudpSocketServerPeer; payload: Uint8Array }) => { + requests.push({ peer, text: dec(payload) }); + server.send(peer, "schema", enc(`re:${dec(payload)}`)); + }); + const port = (socket.address() as { port: number }).port; + return { server, port, requests, connected, left }; +} + +async function startClient(port: number, connectPayload = new Uint8Array()): Promise { + const socket = await bind(); + const connection = new CultNetRudpSocketTransportConnection({ + runtimeId: "client", socket, mode: "client", remoteHost: "127.0.0.1", remotePort: port, + connectionId, resendDelayMs: 10, resendPollMs: 5, + }); + const received: string[] = []; + let closed = false; + connection.on("frame", ({ payload }: { payload: Uint8Array }) => received.push(dec(payload))); + connection.on("close", () => { closed = true; }); + connection.connect(connectPayload); + await waitFor(() => connection.connected, "client handshake"); + return { connection, socket, received, closed: () => closed }; +} + +/** A bare session on a socket, so a goodbye can be sent and the socket released only after it left. */ +async function rawClient(port: number) { + const socket = await bind(); + const session = new CultNetRudpSession({ connectionId }); + const transmit = (packet: ReturnType) => + new Promise((resolve) => socket.send(encodeRudpPacket(packet), port, "127.0.0.1", () => resolve())); + socket.on("message", (wire) => { session.receive(decodeRudpPacket(wire), Date.now()); }); + await transmit(session.createConnect(Date.now())); + await waitFor(() => session.connected, "raw client handshake"); + return { async leave() { await transmit(session.createDisconnect()); socket.close(); } }; +} + +test("two clients on different ports talk through one server socket and neither resets the other", async () => { + const { server, port, requests } = await startServer(); + const a = await startClient(port, enc("alpha")); + const b = await startClient(port, enc("beta")); + try { + assert.equal(server.peers.length, 2); + assert.deepEqual(server.peers.map((peer) => dec(peer.connectPayload)).sort(), ["alpha", "beta"]); + for (let round = 0; round < 5; round += 1) { + a.connection.send("schema", enc(`a${round}`)); + b.connection.send("schema", enc(`b${round}`)); + } + await waitFor(() => a.received.length === 5 && b.received.length === 5, "both clients' replies"); + assert.deepEqual(a.received, [0, 1, 2, 3, 4].map((n) => `re:a${n}`)); + assert.deepEqual(b.received, [0, 1, 2, 3, 4].map((n) => `re:b${n}`)); + assert.equal(requests.length, 10); + assert.equal(a.connection.connected && b.connection.connected, true); + assert.equal(a.closed() || b.closed(), false); + assert.equal(server.peers.length, 2); + } finally { + a.connection.close(); b.connection.close(); server.close(); + } +}); + +test("a client reconnecting from the same endpoint replaces only its own session", async () => { + const { server, port, connected, left } = await startServer(); + const a = await startClient(port); + const b = await startClient(port); + try { + const oldPeer = connected[0]!; + a.connection.connect(); + await waitFor(() => connected.length === 3, "the reconnect to be accepted"); + assert.equal(left.length, 1); + assert.equal(left[0]!.peer, oldPeer); + assert.equal(left[0]!.reason, "replaced by a new Connect generation"); + assert.throws(() => server.send(oldPeer, "schema", enc("stale")), /not connected/); + assert.equal(server.peers.length, 2); + + a.connection.send("schema", enc("after")); + b.connection.send("schema", enc("still")); + 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(); + } +}); + +test("a peer that leaves, or goes silent, is dropped and the others keep their sessions", async () => { + const { server, port, left } = await startServer({ peerTimeoutMs: 150 }); + const leaver = await rawClient(port); + const silent = await startClient(port); + const steady = await startClient(port); + try { + await leaver.leave(); + await waitFor(() => server.peers.length === 2, "the leaver to be dropped"); + assert.deepEqual(left.map((entry) => entry.reason), [""]); + + // Only the steady peer speaks; the silent one is reaped and told so. + const keepAlive = setInterval(() => steady.connection.ping(), 20); + try { + await waitFor(() => server.peers.length === 1, "the silent peer to be reaped"); + } finally { + clearInterval(keepAlive); + } + assert.ok(left.some((entry) => entry.reason === "session timed out")); + await waitFor(() => silent.closed(), "the reaped client to hear the goodbye"); + assert.equal(steady.closed(), false); + 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(); + } +}); + +test("Connects beyond maxPeers and datagrams from unadmitted endpoints are dropped", async () => { + const { server, port } = await startServer({ maxPeers: 1 }); + const first = await startClient(port); + const stranger = await bind(); + try { + const wire = Buffer.from(encodeRudpPacket(new CultNetRudpSession({ connectionId }).createConnect(Date.now()))); + stranger.send(wire, port, "127.0.0.1"); + const data = new CultNetRudpSession({ connectionId, initialSequence: 9 }); + data.createConnect(0); + stranger.send(Buffer.from(encodeRudpPacket(data.createPing())), port, "127.0.0.1"); + await waitFor(() => server.stats.packetsDropped >= 1, "the extra Connect to be dropped"); + await sleep(30); + assert.equal(server.peers.length, 1); + 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(); + } +});