GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultnet-ts: an operation server's send failure belongs to the generation that sent it

Browse source 
A permanent failure reported after that generation ended (the client
connected again from the same endpoint) ended the generation that replaced
it and said goodbye to the new client. The callback now acts only while the
generation that sent the datagram is live, which is the one in `sessions`.

That makes two guards unreachable, so they are deleted: `peer.unsendable`
before the goodbye (the first permanent failure ends the generation, so a
second callback from it is already ignored) and the conditional delete (a
live generation is the entry at its key). A refusal's goodbye is sent
best-effort, as the transport and the document server send theirs, instead
of through the failure handling meant for datagrams a session owes.

Tests: late failure after a reconnect, for the transport's generation check
and for the operation server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 16:02:27 +02:00
commit de8e2e1b88
2 changed files with 158 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

25
packages/cultnet-ts/src/operation-service.ts
View file

@ -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<boolean> {
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<string, RemoteSession>,
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 } {

142
packages/cultnet-ts/test/rudp-send-failure.test.ts
View file

@ -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<typeof decodeRudpPacket>[] = [];
socket.on("message", (wire) => received.push(decodeRudpPacket(wire)));
const toServer = (packet: Parameters<typeof encodeRudpPacket>[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<typeof dgram.createSocket>) => {
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();
}
});