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

Ack Cut D D2 (cultmesh-ts): the document server's sends are guarded, classified, and a refused packet ends only its peer

Browse source 
Every socket send in the RUDP document server goes through
sendRudpDatagram. A Connect from UDP port 0 used to store a session whose
pending Accept threw from the resend interval on every tick, an
uncaughtException that ends the process. Now:

- a Connect whose Accept can never be sent starts no session and leaves a
  session already at that endpoint as it was;
- a datagram that can never be sent as built ends only that peer's session
  (goodbye, onSessionClosed); any other failure is a lost datagram,
  reported through onError, and ends nothing;
- a packet the session refuses (bad fragment metadata) ends that peer's
  session with "session refused a packet" instead of leaving a session
  that has recorded a sequence it never delivered.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:03:05 +02:00
commit cc750d4e4c
2 changed files with 286 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

121
packages/cultmesh-ts/src/index.ts
View file

@ -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<string, {
type SessionRecord = {
sessionId: string;
remote: { address: string; family: string; port: number };
session: CultNetRudpSession;
connectPayload: Uint8Array;
closed: boolean;
/** A packet it owed could never be sent, so the session ended. */
unsendable: boolean;
work: Promise<void>;
}>();
};
const sessions = new Map<string, SessionRecord>();
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<CultNetRudpSession["receive"]>;
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<void> {
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);

209
packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts Normal file
View file

@ -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<void> {
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<typeof CultMesh.createRudpDocumentServer>;
/** The record key of every put received, with the port it came from. */
puts: Map<string, number>;
closed: number[];
errors: Error[];
peer(name: string): Promise<CultNetPeer>;
}
async function startServer(connectionId: number, socket: Socket): Promise<Server> {
const puts = new Map<string, number>();
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<number, string>();
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<void>((resolve) => raw.bind(0, "127.0.0.1", resolve));
const arrived: ReturnType<typeof decodeRudpPacket>[] = [];
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();
}
});