diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 8950f638..6b37087d 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -106,6 +106,8 @@ export { createRudpTransportProfile, decodeRudpPacket, encodeRudpPacket, + isPermanentSendError, + sendRudpDatagram, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS, CultNetRudpReconnectLoop, CultNetRudpSession, diff --git a/packages/cultnet-ts/src/operation-service.ts b/packages/cultnet-ts/src/operation-service.ts index 606cd418..ceebaedc 100644 --- a/packages/cultnet-ts/src/operation-service.ts +++ b/packages/cultnet-ts/src/operation-service.ts @@ -12,6 +12,8 @@ import { CultNetRudpSocketTransportConnection, decodeRudpPacket, encodeRudpPacket, + isPermanentSendError, + sendRudpDatagram, type CultNetRudpPacket, } from "./rudp"; import { CultNetPeer } from "./peer"; @@ -38,6 +40,12 @@ export interface CultNetOperationServer { readonly endpoint: string; /** Datagrams read and discarded: malformed, unadmitted, refused by their session, or failed in handling. */ readonly packetsDropped: number; + /** + * Datagrams that could not be sent to a peer and may yet be (no route, full buffers): + * each is that peer's lost datagram, resent if it was reliable. One that can never be + * sent as built ends that peer's session instead and is not counted here. + */ + readonly sendFailures: number; close(): Promise; } @@ -55,8 +63,12 @@ interface RemoteSession { handling: number; /** When the last handler finished: the response still needs its acknowledgement. */ lastHandledAtMs: number; + /** A packet it owed could never be sent, so the session ended. */ + unsendable: boolean; } +type SendPacket = (key: string, peer: RemoteSession, packet: CultNetRudpPacket) => void; + export async function startCultNetOperationServer( options: CultNetOperationServerOptions, ): Promise { @@ -65,11 +77,23 @@ export async function startCultNetOperationServer( const socket = createSocket("udp4"); const connectionId = options.connectionId ?? DEFAULT_CONNECTION_ID; const sessions = new Map(); - const sendPacket = (remote: RemoteInfo, packet: CultNetRudpPacket): void => { - const wire = encodeRudpPacket(packet); - socket.send(wire, remote.port, remote.address); - }; let packetsDropped = 0; + let sendFailures = 0; + const sendPacket: SendPacket = (key, peer, packet) => { + sendRudpDatagram(socket, encodeRudpPacket(packet), peer.remote.port, peer.remote.address, (error) => { + if (!isPermanentSendError(error)) { + sendFailures += 1; + return; + } + if (peer.unsendable) return; + // The datagram can never be sent as built: the session that owes it ends, never + // the server, and the peer is told. + peer.unsendable = true; + if (sessions.get(key) === peer) sessions.delete(key); + const goodbye = encodeRudpPacket(peer.session.endUnsendableSession(error)); + sendRudpDatagram(socket, goodbye, peer.remote.port, peer.remote.address, () => {}); + }); + }; socket.on("message", (wire, remote) => { // What a datagram carries is the sender's business: a rejection here must // drop and count the packet, never become an unhandled rejection that ends @@ -89,8 +113,8 @@ export async function startCultNetOperationServer( if (peer.handling > 0 || now - peer.lastHandledAtMs < idleTimeoutMs) continue; if (peer.session.checkTimeout(now, idleTimeoutMs)) sessions.delete(key); } - for (const peer of sessions.values()) { - for (const packet of peer.session.dueResends(Date.now())) sendPacket(peer.remote, packet); + for (const [key, peer] of sessions) { + for (const packet of peer.session.dueResends(Date.now())) sendPacket(key, peer, packet); } }, 25); resendTimer.unref?.(); @@ -99,6 +123,7 @@ export async function startCultNetOperationServer( return { endpoint, get packetsDropped() { return packetsDropped; }, + get sendFailures() { return sendFailures; }, close: async () => { clearInterval(resendTimer); await closeSocket(socket); @@ -153,7 +178,7 @@ async function handleServerDatagram( options: CultNetOperationServerOptions, wire: Buffer, remote: RemoteInfo, - sendPacket: (remote: RemoteInfo, packet: CultNetRudpPacket) => void, + sendPacket: SendPacket, ): Promise { let packet: CultNetRudpPacket; try { @@ -165,14 +190,24 @@ async function handleServerDatagram( const key = `${remote.address}:${remote.port}`; let peer = sessions.get(key); if (packet.packetType === "connect") { - if (!peer) { - if (sessions.size >= MAX_OPERATION_SESSIONS) return false; - peer = { session: new CultNetRudpSession({ connectionId, resendDelayMs: 25 }), remote, handling: 0, lastHandledAtMs: 0 }; - sessions.set(key, peer); + if (peer) { + // A Connect from an admitted peer repeats: the session answers with the + // Accept already owed and queues nothing. + sendPacket(key, peer, peer.session.acceptConnect(packet, Date.now(), encode("cultnet-operation-service"))); + return true; } - // A Connect from an admitted peer repeats: the session answers with the - // Accept already owed and queues nothing. - sendPacket(remote, peer.session.acceptConnect(packet, Date.now(), encode("cultnet-operation-service"))); + if (sessions.size >= MAX_OPERATION_SESSIONS) return false; + const candidate: RemoteSession = { + session: new CultNetRudpSession({ connectionId, resendDelayMs: 25 }), + remote, + handling: 0, + lastHandledAtMs: 0, + unsendable: false, + }; + sendPacket(key, candidate, candidate.session.acceptConnect(packet, Date.now(), encode("cultnet-operation-service"))); + // A peer whose Accept can never be sent cannot be answered, so no session starts. + if (candidate.unsendable) return false; + sessions.set(key, candidate); return true; } if (!peer) return false; @@ -183,8 +218,8 @@ async function handleServerDatagram( endSession(sessions, key, peer, sendPacket); return false; } - if (result.reply) sendPacket(remote, result.reply); - for (const ready of result.readyToSend ?? []) sendPacket(remote, ready); + if (result.reply) sendPacket(key, peer, result.reply); + for (const ready of result.readyToSend ?? []) sendPacket(key, peer, ready); if (result.disconnected) { sessions.delete(key); return true; @@ -208,7 +243,7 @@ async function handleServerDatagram( ordered: true, nowMs: Date.now(), maxFragmentBytes: options.maxFragmentBytes ?? 2048, - })) sendPacket(remote, responsePacket); + })) sendPacket(key, peer, responsePacket); } catch { // The session recorded this request's sequence, so keeping it would // acknowledge the retransmit of a request that was never handled. @@ -217,7 +252,7 @@ async function handleServerDatagram( } } if (packet.packetType === "data" || result.delivered.length > 0) { - sendPacket(remote, peer.session.createAckForReceived(packet.sequence)); + sendPacket(key, peer, peer.session.createAckForReceived(packet.sequence)); } return true; } @@ -231,11 +266,11 @@ function endSession( sessions: Map, key: string, peer: RemoteSession, - sendPacket: (remote: RemoteInfo, packet: CultNetRudpPacket) => void, + sendPacket: SendPacket, ): void { if (sessions.get(key) === peer) sessions.delete(key); peer.session.resetPeerState(); - sendPacket(peer.remote, peer.session.createDisconnect(Buffer.from("session refused a packet", "utf8"))); + sendPacket(key, peer, peer.session.createDisconnect(Buffer.from("session refused a packet", "utf8"))); } function parseRudpEndpoint(endpoint: string): { host: string; port: number } { diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index e32bb2d7..4b166d95 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -645,6 +645,18 @@ export class CultNetRudpSession { }); } + /// Ends a session that owes its peer a packet that can never be sent as + /// built (see `isPermanentSendError`). Resending it would fail forever, and + /// dropping it would leave a reliable sequence the peer waits on for good, so + /// the session cannot be kept. Returns the goodbye for the peer; its reason + /// names the error's code, never its message, which can quote the address. + endUnsendableSession(error: Error): CultNetRudpPacket { + this.resetPeerState(); + return this.createDisconnect( + Buffer.from(`${RUDP_UNSENDABLE_PACKET_REASON}: ${sendErrorCode(error)}`, "utf8"), + ); + } + checkTimeout(nowMs: number, timeoutMs: number): boolean { if (!this.#connected || this.#lastReceivedAtMs === undefined) { return false; @@ -1100,11 +1112,22 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement // The goodbye is built after the reset, or its ack field would acknowledge // the very frame the session refused. this.#session.resetPeerState(); + this.#endWithGoodbye(this.#session.createDisconnect(reason)); + } + + /** The session owes its peer a packet that can never be sent as built. */ + #endUnsendableSession(error: Error): void { + this.#endWithGoodbye(this.#session.endUnsendableSession(error)); + } + + #endWithGoodbye(goodbye: CultNetRudpPacket): void { + const reason = goodbye.payload; this.#endedReason = reason; - try { - this.#sendPacket(this.#session.createDisconnect(reason)); - } catch { + if (this.#remoteHost && this.#remotePort !== undefined) { // Best-effort: the session ends whether or not the peer hears it. + const wire = encodeRudpPacket(goodbye); + this.#stats.bytesSent += wire.length; + sendRudpDatagram(this.#socket, wire, this.#remotePort, this.#remoteHost, () => {}); } if (this.#mode === "server") { // Only a Connect can claim the endpoint again. @@ -1175,7 +1198,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement this.#stats.packetsDropped += 1; return; } - this.#sendPacket(accept); + this.#sendInPoll(accept); return; } @@ -1192,10 +1215,10 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement } try { if (result.reply) { - this.#sendPacket(result.reply); + this.#sendInPoll(result.reply); } for (const ready of result.readyToSend ?? []) { - this.#sendPacket(ready); + this.#sendInPoll(ready); } if (result.pong) { this.emit("pong", { payload: result.pongPayload ?? new Uint8Array() }); @@ -1214,7 +1237,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement return; } if (packet.packetType === "accept" || packet.packetType === "data" || result.delivered.length > 0) { - this.#sendPacket(this.#session.createAckForReceived(packet.sequence)); + this.#sendInPoll(this.#session.createAckForReceived(packet.sequence)); } } catch (error) { this.emit("error", error instanceof Error ? error : new Error(String(error))); @@ -1222,18 +1245,62 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement } #sendDueResends(): void { + const generation = this.#session.generation; for (const packet of this.#session.dueResends(Date.now())) { - this.#sendPacket(packet); + this.#sendInPoll(packet); + // A resend that can never be sent ended the session and what it owed. + if (this.#session.generation !== generation) return; } } + /** A caller-directed send: a failure the send throws is the caller's error. */ #sendPacket(packet: CultNetRudpPacket): void { + const thrown = this.#transmit(packet); + if (thrown) throw thrown; + } + + /** + * A send with no caller to hand a failure to (a receive or a resend): the failure + * is handled as every later one is, and never escapes the timer or the socket's + * "message" handler. + */ + #sendInPoll(packet: CultNetRudpPacket): void { + const thrown = this.#transmit(packet); + if (thrown) this.#sendFailed(thrown); + } + + /** Sends one packet and returns the failure the send threw, if any. */ + #transmit(packet: CultNetRudpPacket): Error | undefined { if (!this.#remoteHost || this.#remotePort === undefined) { throw new Error("RUDP socket transport does not have a remote endpoint."); } const wire = encodeRudpPacket(packet); this.#stats.bytesSent += wire.length; - this.#socket.send(wire, this.#remotePort, this.#remoteHost); + let thrown: Error | undefined; + let sending = true; + const generation = this.#session.generation; + sendRudpDatagram(this.#socket, wire, this.#remotePort, this.#remoteHost, (error) => { + if (sending) { + thrown = error; + } else if (this.#session.generation === generation) { + // A later failure belongs to the session generation that sent it. + this.#sendFailed(error); + } + }); + sending = false; + return thrown; + } + + /** + * A datagram that can never be sent ends the session; any other failure is a lost + * datagram, reported on "error" as the socket reports it. + */ + #sendFailed(error: Error): void { + if (isPermanentSendError(error)) { + this.#endUnsendableSession(error); + return; + } + this.emit("error", error); } } @@ -1327,6 +1394,61 @@ export class CultNetRudpReconnectLoop extends EventEmitter { } } +/** Prefix of the goodbye reason for a session that owed a packet which can never be sent. */ +const RUDP_UNSENDABLE_PACKET_REASON = "packet could not be sent"; + +/** + * Failures that can never pass for the datagram as built: too large for a datagram + * (`EMSGSIZE`), an address malformed for the socket (`EINVAL`, and Node's own + * `ERR_SOCKET_BAD_PORT`, which it throws for port 0 before the system would say + * `EINVAL`), or an address of another family (`EAFNOSUPPORT`). libuv names these the + * same on every platform. + */ +const PERMANENT_SEND_ERROR_CODES: ReadonlySet = new Set([ + "EMSGSIZE", + "EINVAL", + "EAFNOSUPPORT", + "ERR_SOCKET_BAD_PORT", +]); + +function sendErrorCode(error: Error): string { + const code = (error as NodeJS.ErrnoException).code; + return typeof code === "string" && code.length > 0 ? code : "UNKNOWN"; +} + +/** + * Whether a failed UDP send can never succeed for the datagram as built, so resending it is + * pointless and the session that owes it cannot be kept. Everything else (no route, full + * buffers, a closed socket, an interrupted call) is a property of the path or the moment: a + * lost datagram, not a reason to end a session. + */ +export function isPermanentSendError(error: Error): boolean { + return PERMANENT_SEND_ERROR_CODES.has(sendErrorCode(error)); +} + +/** + * Sends one datagram. Node reports a failed send two ways: it throws for an argument it + * refuses (port 0 among them) and passes a system failure to the callback later. Both + * reach `onFailure` and nothing else, so a datagram that cannot be sent to one peer never + * escapes a timer or a receive handler, and never reaches the socket's "error" event. A + * failure the send throws reaches `onFailure` before this returns. + */ +export function sendRudpDatagram( + socket: Socket, + wire: Uint8Array, + port: number, + address: string, + onFailure: (error: Error) => void, +): void { + try { + socket.send(wire, port, address, (error) => { + if (error) onFailure(error); + }); + } catch (error) { + onFailure(error instanceof Error ? error : new Error(String(error))); + } +} + export function createRudpTransportProfile( runtimeId: string, options: RudpTransportProfileOptions = {}, diff --git a/packages/cultnet-ts/test/rudp-send-failure.test.ts b/packages/cultnet-ts/test/rudp-send-failure.test.ts new file mode 100644 index 00000000..9abfe3d8 --- /dev/null +++ b/packages/cultnet-ts/test/rudp-send-failure.test.ts @@ -0,0 +1,231 @@ +// A datagram that cannot be sent to one peer is that peer's problem alone. Node throws +// for a send it refuses outright (a peer at UDP port 0, which any spoofer can claim) +// and reports a system failure later through the callback; neither may escape a +// timer or a receive handler, and only a datagram that can never be sent as built +// ends the session that owes it. +import assert from "node:assert/strict"; +import dgram, { type RemoteInfo, type Socket } from "node:dgram"; +import test from "node:test"; +import { + CultNetRudpSession, + CultNetRudpSocketTransportConnection, + encodeRudpPacket, + isPermanentSendError, + startCultNetOperationServer, + type CultNetOperationResponseMessage, +} from "../src"; + +async function bindUdpSocket(): Promise { + const socket = dgram.createSocket("udp4"); + await new Promise((resolve) => socket.bind(0, "127.0.0.1", resolve)); + return socket; +} + +function udpPort(socket: Socket): number { + const address = socket.address(); + assert.notEqual(typeof address, "string"); + return address.port; +} + +async function waitFor(predicate: () => boolean, description: string): Promise { + const startedAt = Date.now(); + while (!predicate()) { + if (Date.now() - startedAt > 2_000) { + throw new Error(`Timed out waiting for ${description}.`); + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** A Connect as it arrives from UDP source port 0: Node refuses to send anything back. */ +function connectFromPortZero(socket: Socket, connectionId: number): void { + const wire = Buffer.from(encodeRudpPacket(new CultNetRudpSession({ connectionId }).createConnect(Date.now()))); + const remote: RemoteInfo = { address: "127.0.0.1", family: "IPv4", port: 0, size: wire.length }; + socket.emit("message", wire, remote); +} + +/** Records every exception that would otherwise end the process. */ +function watchUncaught(): { errors: unknown[]; stop: () => void } { + const errors: unknown[] = []; + const listener = (error: unknown) => errors.push(error); + process.on("uncaughtException", listener); + return { errors, stop: () => process.off("uncaughtException", listener) }; +} + +test("only a datagram that can never be sent as built is a permanent send failure", () => { + const failure = (code: string) => Object.assign(new Error("send failed"), { code }); + for (const code of ["EMSGSIZE", "EINVAL", "EAFNOSUPPORT", "ERR_SOCKET_BAD_PORT"]) { + assert.equal(isPermanentSendError(failure(code)), true, code); + } + for (const code of ["EHOSTUNREACH", "ENETUNREACH", "ECONNREFUSED", "ENOBUFS", "EAGAIN", "ERR_SOCKET_DGRAM_NOT_RUNNING"]) { + assert.equal(isPermanentSendError(failure(code)), false, code); + } + assert.equal(isPermanentSendError(new Error("no code")), false); +}); + +test("a server transport answering a Connect from port 0 ends that session and keeps serving", async () => { + const serverSocket = await bindUdpSocket(); + const clientSocket = await bindUdpSocket(); + const connectionId = 0x10203090; + const uncaught = watchUncaught(); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendDelayMs: 10, + resendPollMs: 5, + }); + const errors: Error[] = []; + const reasons: string[] = []; + server.on("error", (error: Error) => errors.push(error)); + server.on("disconnect", ({ reason }: { reason: Uint8Array }) => reasons.push(Buffer.from(reason).toString("utf8"))); + const client = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-client", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: udpPort(serverSocket), + connectionId, + resendDelayMs: 10, + resendPollMs: 5, + }); + try { + connectFromPortZero(serverSocket, connectionId); + // Many resend ticks: nothing the port-0 session owed is ever retried. + await sleep(100); + // The goodbye names the error by its code, never by a message quoting the address. + assert.deepEqual(reasons, ["packet could not be sent: ERR_SOCKET_BAD_PORT"]); + + const frames: string[] = []; + server.on("frame", (frame: { payload: Uint8Array }) => frames.push(Buffer.from(frame.payload).toString("utf8"))); + client.connect(); + await waitFor(() => client.connected && server.connected, "a real client's handshake"); + client.send("schema", Buffer.from("after", "utf8")); + await waitFor(() => frames.length === 1, "the real client's frame"); + assert.deepEqual(frames, ["after"]); + assert.deepEqual(errors, []); + assert.deepEqual(uncaught.errors, []); + } finally { + uncaught.stop(); + client.close(); + server.close(); + } +}); + +test("an operation server answering a Connect from port 0 starts no session and keeps serving", async () => { + const connectionId = 0x43554c54; + const uncaught = watchUncaught(); + // The server owns its socket; this captures it so a datagram can arrive from port 0. + const createSocket = dgram.createSocket; + let serverSocket: Socket | undefined; + Object.assign(dgram, { + createSocket: (...args: Parameters) => { + serverSocket = createSocket(...args); + return serverSocket; + }, + }); + const server = await startCultNetOperationServer({ + runtimeId: "operations", + handler: (request): CultNetOperationResponseMessage => ({ + schemaVersion: "cultnet.operation_response.v0", + messageId: request.messageId, + serviceId: request.serviceId, + operation: request.operation, + status: "ok", + payloadSchema: request.payloadSchema, + payloadEncoding: request.payloadEncoding, + payload: request.payload, + diagnostics: [], + sourceRuntimeId: "operations", + }), + }).finally(() => Object.assign(dgram, { createSocket })); + const raw = await bindUdpSocket(); + const port = Number(new URL(server.endpoint).port); + const accepts: number[] = []; + raw.on("message", () => accepts.push(Date.now())); + try { + connectFromPortZero(serverSocket!, connectionId); + // The resend timer runs every 25 ms; a session that had started would owe its Accept. + await sleep(150); + assert.equal(server.packetsDropped, 1, "the Connect started no session"); + assert.equal(server.sendFailures, 0); + + // A real peer is still answered. + raw.send(encodeRudpPacket(new CultNetRudpSession({ connectionId }).createConnect(0)), port, "127.0.0.1"); + await waitFor(() => accepts.length > 0, "the real peer's Accept"); + assert.deepEqual(uncaught.errors, []); + } finally { + uncaught.stop(); + raw.close(); + await server.close(); + } +}); + +test("a server transport whose resend can never be sent ends the session and sends nothing more of it", async () => { + const serverSocket = await bindUdpSocket(); + const clientSocket = await bindUdpSocket(); + const connectionId = 0x10203094; + const uncaught = watchUncaught(); + const clientPort = udpPort(clientSocket); + // Sends to the client fail as the test says: later through the callback, or at once. + let mode: "healthy" | "lost" | "never" = "healthy"; + const send = serverSocket.send.bind(serverSocket) as (...args: unknown[]) => void; + Object.assign(serverSocket, { + send: (message: Uint8Array, port: number, address: string, callback?: (error: Error | null) => void) => { + if (port !== clientPort || mode === "healthy") { + send(message, port, address, callback); + } else if (mode === "lost") { + process.nextTick(() => callback?.(Object.assign(new Error("lost"), { code: "EHOSTUNREACH" }))); + } else { + throw Object.assign(new Error("never"), { code: "EMSGSIZE" }); + } + }, + }); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendDelayMs: 10, + resendPollMs: 5, + }); + const errors: string[] = []; + const reasons: string[] = []; + server.on("error", (error: NodeJS.ErrnoException) => errors.push(error.code ?? "")); + server.on("disconnect", ({ reason }: { reason: Uint8Array }) => reasons.push(Buffer.from(reason).toString("utf8"))); + const client = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-client", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: udpPort(serverSocket), + connectionId, + resendDelayMs: 10, + resendPollMs: 5, + }); + try { + client.connect(); + await waitFor(() => client.connected && server.connected, "the handshake"); + mode = "lost"; + server.send("schema", Buffer.from("one", "utf8")); + server.send("schema", Buffer.from("two", "utf8")); + await waitFor(() => errors.length >= 2, "both writes lost"); + assert.deepEqual([...new Set(errors)], ["EHOSTUNREACH"], "a loss that may pass is reported and ends nothing"); + assert.deepEqual(reasons, []); + + // Both writes are due again together; the first can never be sent. + mode = "never"; + await waitFor(() => reasons.length > 0, "the session to end"); + await sleep(50); + assert.deepEqual(reasons, ["packet could not be sent: EMSGSIZE"]); + assert.equal(server.connected, false); + assert.deepEqual(uncaught.errors, []); + } finally { + uncaught.stop(); + client.close(); + server.close(); + } +});