diff --git a/packages/cultnet-ts/src/operation-service.ts b/packages/cultnet-ts/src/operation-service.ts index de41e405..90760591 100644 --- a/packages/cultnet-ts/src/operation-service.ts +++ b/packages/cultnet-ts/src/operation-service.ts @@ -81,16 +81,20 @@ export async function startCultNetOperationServer( let packetsDropped = 0; let sendFailures = 0; const sendPacket: SendPacket = (key, peer, packet) => { + const generation = peer.session.generation; sendRudpDatagram(socket, encodeRudpPacket(packet), peer.remote.port, peer.remote.address, (error) => { if (!isPermanentSendError(error)) { sendFailures += 1; return; } - if (peer.unsendable) return; + // A failure reported after the generation that sent the datagram ended is owed by + // nothing: that generation's end already told the peer, and a client that connected + // again from the endpoint owns it now. A live generation is the one in `sessions`. + if (peer.session.generation !== generation) 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); + sessions.delete(key); const goodbye = encodeRudpPacket(peer.session.endUnsendableSession(error)); sendRudpDatagram(socket, goodbye, peer.remote.port, peer.remote.address, () => {}); }); @@ -99,7 +103,7 @@ export async function startCultNetOperationServer( // 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 // the process. - handleServerDatagram(sessions, connectionId, options, wire, remote, sendPacket).then( + handleServerDatagram(sessions, connectionId, options, wire, remote, socket, sendPacket).then( admitted => { if (!admitted) packetsDropped += 1; }, () => { packetsDropped += 1; }, ); @@ -180,6 +184,7 @@ async function handleServerDatagram( options: CultNetOperationServerOptions, wire: Buffer, remote: RemoteInfo, + socket: Socket, sendPacket: SendPacket, ): Promise { let packet: CultNetRudpPacket; @@ -217,7 +222,7 @@ async function handleServerDatagram( try { result = peer.session.receive(packet, Date.now()); } catch { - endSession(sessions, key, peer, sendPacket); + endSession(sessions, key, peer, socket); return false; } if (result.reply) sendPacket(key, peer, result.reply); @@ -256,7 +261,7 @@ async function handleServerDatagram( if (peer.session.generation !== generation) return true; // The session recorded this request's sequence, so keeping it would // acknowledge the retransmit of a request that was never handled. - endSession(sessions, key, peer, sendPacket); + endSession(sessions, key, peer, socket); return false; } } @@ -269,17 +274,19 @@ async function handleServerDatagram( /** * Ends a session that took a packet it could not serve, and tells the peer. The * reset comes first, or the goodbye's ack field would acknowledge the very - * packet that was refused. A session already replaced under the same key is left. + * packet that was refused. The goodbye is best-effort: the session ends whether + * or not the peer hears it. */ function endSession( sessions: Map, key: string, peer: RemoteSession, - sendPacket: SendPacket, + socket: Socket, ): void { - if (sessions.get(key) === peer) sessions.delete(key); + sessions.delete(key); peer.session.resetPeerState(); - sendPacket(key, peer, peer.session.createDisconnect(Buffer.from("session refused a packet", "utf8"))); + const goodbye = encodeRudpPacket(peer.session.createDisconnect(Buffer.from("session refused a packet", "utf8"))); + sendRudpDatagram(socket, goodbye, peer.remote.port, peer.remote.address, () => {}); } function parseRudpEndpoint(endpoint: string): { host: string; port: number } { diff --git a/packages/cultnet-ts/test/rudp-send-failure.test.ts b/packages/cultnet-ts/test/rudp-send-failure.test.ts index edcf0d08..cb62d199 100644 --- a/packages/cultnet-ts/test/rudp-send-failure.test.ts +++ b/packages/cultnet-ts/test/rudp-send-failure.test.ts @@ -6,13 +6,17 @@ import assert from "node:assert/strict"; import dgram, { type RemoteInfo, type Socket } from "node:dgram"; import test from "node:test"; +import { encode } from "@msgpack/msgpack"; import { CultNetRudpSession, CultNetRudpSocketTransportConnection, + decodeRudpPacket, + encodeCultNetMessageForWire, encodeRudpPacket, invokeCultNetOperation, isPermanentSendError, startCultNetOperationServer, + type CultNetOperationRequestMessage, type CultNetOperationResponseMessage, } from "../src"; @@ -291,3 +295,141 @@ test("an operation call the network refuses rejects with a timeout and does not uncaught.stop(); } }); + +// A send's failure can be reported after the generation that sent it ended: the client +// connected again from the same endpoint in the meantime. That failure is owed by nothing; +// it may not end the generation that replaced it or say goodbye to the endpoint. + +/** Makes the next send to `port` fail as can never pass as built, reported `delayMs` later. */ +function failNextSendLate(socket: Socket, port: number, delayMs: number): { arm: () => void } { + let armed = false; + const send = socket.send.bind(socket) as (...args: unknown[]) => void; + Object.assign(socket, { + send: (message: Uint8Array, toPort: number, address: string, callback?: (error: Error | null) => void) => { + if (toPort !== port || !armed) { + send(message, toPort, address, callback); + return; + } + armed = false; + setTimeout(() => callback?.(Object.assign(new Error("late"), { code: "EMSGSIZE" })), delayMs); + }, + }); + return { arm: () => { armed = true; } }; +} + +/** A client driven packet by packet from one socket, so it can connect again from the same endpoint. */ +async function rawClient(serverPort: number, connectionId: number) { + const socket = await bindUdpSocket(); + const received: ReturnType[] = []; + socket.on("message", (wire) => received.push(decodeRudpPacket(wire))); + const toServer = (packet: Parameters[0]) => socket.send(encodeRudpPacket(packet), serverPort, "127.0.0.1"); + const connect = async (initialSequence: number) => { + const session = new CultNetRudpSession({ connectionId, initialSequence }); + received.length = 0; + toServer(session.createConnect(Date.now())); + await waitFor(() => received.some((packet) => packet.packetType === "accept"), "the Accept"); + const accept = received.find((packet) => packet.packetType === "accept")!; + session.receive(accept, Date.now()); + toServer(session.createAckForReceived(accept.sequence)); + return session; + }; + return { socket, received, toServer, connect }; +} + +test("a transport ignores a send failure reported after the generation that sent it ended", async () => { + const serverSocket = await bindUdpSocket(); + const connectionId = 0x10203096; + const uncaught = watchUncaught(); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendPollMs: 1_000, + }); + const reasons: string[] = []; + server.on("disconnect", ({ reason }: { reason: Uint8Array }) => reasons.push(Buffer.from(reason).toString("utf8"))); + const client = await rawClient(udpPort(serverSocket), connectionId); + const late = failNextSendLate(serverSocket, udpPort(client.socket), 150); + try { + await client.connect(10); + late.arm(); + server.send("schema", Buffer.from("old", "utf8")); + await client.connect(5_000); + await sleep(250); + + assert.deepEqual(reasons, []); + assert.equal(server.connected, true); + assert.deepEqual(client.received.filter((packet) => packet.packetType === "disconnect"), []); + assert.deepEqual(uncaught.errors, []); + } finally { + uncaught.stop(); + client.socket.close(); + server.close(); + } +}); + +test("an operation server ignores a send failure reported after the generation that sent it ended", async () => { + const connectionId = 0x43554c54; + 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 client = await rawClient(Number(new URL(server.endpoint).port), connectionId); + const late = failNextSendLate(serverSocket!, udpPort(client.socket), 150); + const request = (session: CultNetRudpSession, messageId: string) => session.sendMany("schema", encode(encodeCultNetMessageForWire({ + schemaVersion: "cultnet.operation_request.v0", + messageId, + serviceId: "service", + operation: "operation", + payloadSchema: "payload", + payloadEncoding: "messagepack", + payload: new Uint8Array(), + diagnostics: [], + sourceRuntimeId: "caller", + } satisfies CultNetOperationRequestMessage, "cultnet.schema.v0")), { reliable: true, ordered: true, nowMs: Date.now() }); + try { + const old = await client.connect(10); + // The response to this request is the next send, and it fails after the client reconnected. + late.arm(); + for (const packet of request(old, "old")) client.toServer(packet); + const renewed = await client.connect(5_000); + await sleep(250); + assert.deepEqual(client.received.filter((packet) => packet.packetType === "disconnect"), []); + assert.equal(server.sendFailures, 0); + + // The session that replaced it is still served. + let read = client.received.length; + let answered = 0; + for (const packet of request(renewed, "after")) client.toServer(packet); + await waitFor(() => { + for (; read < client.received.length; read += 1) { + const packet = client.received[read]!; + if (packet.packetType === "data") answered += renewed.receive(packet, Date.now()).delivered.length; + } + return answered > 0; + }, "the renewed session's response"); + } finally { + client.socket.close(); + await server.close(); + } +});