diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index b89cd823..212e3836 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -21,7 +21,9 @@ import { defineCultNetDocumentBinding, encodeCultNetMessageForWire, encodeRudpPacket, + isPermanentSendError, parseCultNetMessage, + sendRudpDatagram, type CultNetErrorMessage, type CultNetDocumentBinding, type CultNetMessage, @@ -29,6 +31,7 @@ import { type CultNetOperationResponseMessage, type CultNetRawDocumentRecord, type CultNetReconnectPolicy, + type CultNetRudpPacket, type CultNetSchemaCatalogOptions, type CultNetSnapshotResponseRawMessage, type CultNetWireContract, @@ -5761,14 +5764,17 @@ export class CultMesh { const port = options.bindPort ?? 0; const bind = { host, port }; const socket = options.socket ?? createSocket(host.includes(":") ? "udp6" : "udp4"); - const sessions = new Map; - }>(); + }; + const sessions = new Map(); let sessionSequence = 0; const resendPollMs = Math.max(10, options.resendPollMs ?? 25); const sessionTimeoutMs = Math.max(1_000, options.sessionTimeoutMs ?? 30_000); @@ -5794,19 +5800,15 @@ export class CultMesh { if (packet.packetType === "connect") { const connectPayload = Uint8Array.from(packet.payload ?? []); if (record?.session.connectRepeats(packet)) { - socket.send( - encodeRudpPacket(record.session.acceptConnect(packet, nowMs)), - remote.port, - remote.address, - ); + sendPacket(record, record.session.acceptConnect(packet, nowMs)); return; } - if (record) closeSession(record); - record = { + const next: SessionRecord = { sessionId: `${runtimeId}:${++sessionSequence}`, remote: { address: remote.address, family: remote.family, port: remote.port }, connectPayload, closed: false, + unsendable: false, session: new CultNetRudpSession({ connectionId, initialSequence: options.initialSequence, @@ -5815,8 +5817,12 @@ export class CultMesh { }), work: Promise.resolve(), }; - sessions.set(key, record); - socket.send(encodeRudpPacket(record.session.acceptConnect(packet, nowMs)), remote.port, remote.address); + sendPacket(next, next.session.acceptConnect(packet, nowMs)); + // A peer whose Accept can never be sent cannot be answered: no session + // starts, and one already there is left as it was. + if (next.unsendable) return; + if (record) closeSession(record); + sessions.set(key, next); return; } @@ -5824,12 +5830,22 @@ export class CultMesh { return; } - const result = record.session.receive(packet, nowMs); + let result: ReturnType; + try { + result = record.session.receive(packet, nowMs); + } 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. It ends, never the server, and the peer is told. What a + // datagram carries is the sender's business, so nothing is reported. + endRefusedSession(key, record); + return; + } if (result.reply) { - socket.send(encodeRudpPacket(result.reply), record.remote.port, record.remote.address); + sendPacket(record, result.reply); } for (const ready of result.readyToSend ?? []) { - socket.send(encodeRudpPacket(ready), record.remote.port, record.remote.address); + sendPacket(record, ready); } for (const frame of result.delivered) { if (frame.channelId !== "schema") { @@ -5845,11 +5861,7 @@ export class CultMesh { return; } if (packet.reliable) { - socket.send( - encodeRudpPacket(record.session.createAckForReceived(packet.sequence)), - record.remote.port, - record.remote.address, - ); + sendPacket(record, record.session.createAckForReceived(packet.sequence)); } } catch (error) { reportError(error); @@ -5858,7 +5870,7 @@ export class CultMesh { socket.on("error", reportError); async function handleDocumentServerFrame( - record: { sessionId: string; remote: { address: string; family: string; port: number }; session: CultNetRudpSession; connectPayload: Uint8Array; closed: boolean }, + record: SessionRecord, payload: Uint8Array, ): Promise { if (record.closed) return; @@ -5920,13 +5932,7 @@ export class CultMesh { } } - function publicSession(record: { - sessionId: string; - remote: { address: string; family: string; port: number }; - session: CultNetRudpSession; - connectPayload: Uint8Array; - closed: boolean; - }): CultMeshRudpServerSession { + function publicSession(record: SessionRecord): CultMeshRudpServerSession { return { sessionId: record.sessionId, remote: { ...record.remote }, @@ -5938,31 +5944,58 @@ export class CultMesh { }; } - function notifySessionClosed(record: { - sessionId: string; - remote: { address: string; family: string; port: number }; - session: CultNetRudpSession; - connectPayload: Uint8Array; - closed: boolean; - }): void { + function notifySessionClosed(record: SessionRecord): void { if (!options.onSessionClosed) return; void Promise.resolve(options.onSessionClosed(publicSession(record))).catch(reportError); } - function closeSession(record: { - sessionId: string; - remote: { address: string; family: string; port: number }; - session: CultNetRudpSession; - connectPayload: Uint8Array; - closed: boolean; - }): void { + function closeSession(record: SessionRecord): void { if (record.closed) return; record.closed = true; notifySessionClosed(record); } + /** + * Sends one packet to a session's peer. A failure is that peer's alone: one that may + * pass (no route, full buffers) is a lost datagram, reported, and resent if it was + * reliable; one that can never pass as built ends that peer's session. Neither ends + * the server, a timer or another peer's session. + */ + function sendPacket(record: SessionRecord, packet: CultNetRudpPacket): void { + sendRudpDatagram(socket, encodeRudpPacket(packet), record.remote.port, record.remote.address, (error) => { + if (isPermanentSendError(error)) { + endUnsendableSession(record, error); + } else { + reportError(error); + } + }); + } + + function endRefusedSession(key: string, record: SessionRecord): void { + sessions.delete(key); + closeSession(record); + // The goodbye is built after the reset, or its ack field would acknowledge the + // very frame the session refused. + record.session.resetPeerState(); + const goodbye = encodeRudpPacket(record.session.createDisconnect(Buffer.from("session refused a packet", "utf8"))); + sendRudpDatagram(socket, goodbye, record.remote.port, record.remote.address, () => {}); + } + + /** The session owes its peer a packet that can never be sent as built. */ + function endUnsendableSession(record: SessionRecord, error: Error): void { + if (record.unsendable) return; + record.unsendable = true; + const key = `${record.remote.address}:${record.remote.port}`; + // A session that never started, or was already replaced, has nothing to end. + if (sessions.get(key) !== record) return; + sessions.delete(key); + closeSession(record); + const goodbye = encodeRudpPacket(record.session.endUnsendableSession(error)); + sendRudpDatagram(socket, goodbye, record.remote.port, record.remote.address, () => {}); + } + function sendSchemaMessage( - record: { sessionId: string; remote: { address: string; port: number }; session: CultNetRudpSession; closed: boolean }, + record: SessionRecord, message: CultNetMessage, ): void { if (record.closed) throw new Error(`CultMesh RUDP session ${record.sessionId} is closed.`); @@ -5972,7 +6005,7 @@ export class CultMesh { ordered: true, nowMs: Date.now(), })) { - socket.send(encodeRudpPacket(packet), record.remote.port, record.remote.address); + sendPacket(record, packet); } } @@ -5999,7 +6032,7 @@ export class CultMesh { continue; } for (const packet of record.session.dueResends(nowMs)) { - socket.send(encodeRudpPacket(packet), record.remote.port, record.remote.address); + sendPacket(record, packet); } } }, resendPollMs); diff --git a/packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts b/packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts new file mode 100644 index 00000000..8fa0f6a4 --- /dev/null +++ b/packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts @@ -0,0 +1,209 @@ +// A datagram the document server cannot send to one peer is that peer's problem +// alone. It never escapes a timer or the receive handler, never stops the server +// serving the others, and only a datagram that can never be sent as built ends +// the session that owes it. +import assert from "node:assert/strict"; +import { createSocket, type RemoteInfo, type Socket } from "node:dgram"; +import test from "node:test"; +import { encode } from "@msgpack/msgpack"; +import { + CultNetDocumentRegistry, + CultNetRudpSession, + decodeRudpPacket, + encodeRudpPacket, + type CultNetPeer, +} from "cultnet-ts"; +import { CultMesh } from "../src/index"; + +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 new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +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) }; +} + +function put(recordKey: string) { + return { + schemaVersion: "cultnet.document_put_raw.v0" as const, + messageId: `put-${recordKey}`, + document: { + schemaId: "cultmesh.note.v0", + recordKey, + storedAt: new Date().toISOString(), + payloadEncoding: "messagepack" as const, + payload: encode(recordKey), + }, + }; +} + +interface Server { + socket: Socket; + server: ReturnType; + /** The record key of every put received, with the port it came from. */ + puts: Map; + closed: number[]; + errors: Error[]; + peer(name: string): Promise; +} + +async function startServer(connectionId: number, socket: Socket): Promise { + const puts = new Map(); + const closed: number[] = []; + const errors: Error[] = []; + const server = CultMesh.createRudpDocumentServer("send-failure-server", connectionId, { + documents: new CultNetDocumentRegistry(), + socket, + bindHost: "127.0.0.1", + bindPort: 0, + resendDelayMs: 10, + resendPollMs: 10, + sessionTimeoutMs: 1_000, + onDocumentPutRaw: (document) => { + puts.set(document.recordKey, document.remote.port); + }, + onSessionClosed: (session) => { + closed.push(session.remote.port); + }, + onError: (error) => errors.push(error), + }); + await server.start(); + const peer = (name: string) => + CultMesh.createRudpPeer(name, connectionId, `rudp://127.0.0.1:${server.bind.port}`, { + resendDelayMs: 10, + resendPollMs: 5, + connectTimeoutMs: 2_000, + }); + return { socket, server, puts, closed, errors, peer }; +} + +test("a Connect whose Accept can never be sent starts no session, and the server keeps serving", async () => { + const connectionId = 0x10203091; + const uncaught = watchUncaught(); + const socket = createSocket("udp4"); + const served = await startServer(connectionId, socket); + let peer: CultNetPeer | undefined; + try { + // A Connect as it arrives from UDP source port 0, which Node refuses to answer. + 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); + // Past the idle timeout: a session that had started would have been closed by now. + await sleep(1_300); + + peer = await served.peer("after-port-zero"); + peer.send(put("after")); + await waitFor(() => served.puts.has("after"), "the real peer's put"); + assert.deepEqual(served.closed, [], "no session started for port 0"); + assert.deepEqual(served.errors.map((error) => error.message), []); + assert.deepEqual(uncaught.errors, []); + } finally { + uncaught.stop(); + peer?.close(); + served.server.close(); + } +}); + +test("a datagram that may yet pass is reported; one that never can ends only that peer's session", async () => { + const connectionId = 0x10203092; + const socket = createSocket("udp4"); + // Sends to a failing port report the failure the way Node reports a system one: + // later, through the send's callback. + const failing = new Map(); + const send = socket.send.bind(socket) as (...args: unknown[]) => void; + Object.assign(socket, { + send: (message: Uint8Array, port: number, address: string, callback?: (error: Error | null) => void) => { + const code = failing.get(port); + if (code === undefined) { + send(message, port, address, callback); + return; + } + process.nextTick(() => callback?.(Object.assign(new Error(`send failed: ${code}`), { code }))); + }, + }); + const served = await startServer(connectionId, socket); + let x: CultNetPeer | undefined; + let y: CultNetPeer | undefined; + try { + x = await served.peer("x"); + y = await served.peer("y"); + x.send(put("x-1")); + await waitFor(() => served.puts.has("x-1"), "X's first put"); + const xPort = served.puts.get("x-1")!; + + // No route: X's acknowledgements are lost datagrams, reported, and end nothing. + failing.set(xPort, "EHOSTUNREACH"); + x.send(put("x-2")); + await waitFor(() => served.errors.some((error) => (error as NodeJS.ErrnoException).code === "EHOSTUNREACH"), "the reported loss"); + y.send(put("y-1")); + await waitFor(() => served.puts.has("y-1"), "Y's put"); + assert.deepEqual(served.closed, [], "a loss that may pass ends no session"); + + // X's retransmitted put is acknowledged again, and that datagram can never be sent. + failing.set(xPort, "EMSGSIZE"); + await waitFor(() => served.closed.length > 0, "X's session to end"); + assert.deepEqual(served.closed, [xPort]); + assert.equal( + served.errors.some((error) => (error as NodeJS.ErrnoException).code === "EMSGSIZE"), + false, + "a datagram that can never be sent ends the session instead of being reported", + ); + + y.send(put("y-2")); + await waitFor(() => served.puts.has("y-2"), "Y's put after X ended"); + assert.deepEqual(served.closed, [xPort]); + } finally { + x?.close(); + y?.close(); + served.server.close(); + } +}); + +test("a packet the session refuses ends only that peer's session, and the peer is told", async () => { + const connectionId = 0x10203093; + const socket = createSocket("udp4"); + const served = await startServer(connectionId, socket); + const raw = createSocket("udp4"); + await new Promise((resolve) => raw.bind(0, "127.0.0.1", resolve)); + const arrived: ReturnType[] = []; + raw.on("message", (wire) => arrived.push(decodeRudpPacket(wire))); + let y: CultNetPeer | undefined; + try { + const session = new CultNetRudpSession({ connectionId }); + raw.send(encodeRudpPacket(session.createConnect(Date.now())), served.server.bind.port, "127.0.0.1"); + await waitFor(() => arrived.some((packet) => packet.packetType === "accept"), "the Accept"); + session.receive(arrived.find((packet) => packet.packetType === "accept")!, Date.now()); + y = await served.peer("y"); + + // A fragment whose metadata the session refuses: fragment id 0 names no set. + const [fragment] = session.sendMany("schema", new Uint8Array([1]), { reliable: false }); + raw.send( + encodeRudpPacket({ ...fragment!, fragmentId: 0, fragmentIndex: 0, fragmentCount: 2 }), + served.server.bind.port, + "127.0.0.1", + ); + await waitFor(() => arrived.some((packet) => packet.packetType === "disconnect"), "the goodbye"); + const goodbye = arrived.find((packet) => packet.packetType === "disconnect")!; + assert.equal(Buffer.from(goodbye.payload ?? []).toString("utf8"), "session refused a packet"); + assert.deepEqual(served.closed, [(raw.address() as { port: number }).port]); + + y.send(put("y-after")); + await waitFor(() => served.puts.has("y-after"), "Y's put"); + assert.deepEqual(served.errors.map((error) => error.message), []); + } finally { + y?.close(); + raw.close(); + served.server.close(); + } +});