cultnet-ts: an operation handler's result belongs to the generation its request arrived on

When that generation ends while the handler runs (a goodbye, a refusal, a
timeout, a new Connect from the same endpoint), the response is not sent
and a failure sends no second goodbye: a client that connected again owns
the endpoint.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:54:33 +02:00
commit 25ea551bd5
2 changed files with 90 additions and 0 deletions

View file

@ -226,6 +226,11 @@ async function handleServerDatagram(
sessions.delete(key);
return true;
}
// A request belongs to the session generation it arrived on. If that generation ends while a
// handler runs (a goodbye, a refusal, a timeout, a new Connect from the same endpoint), what
// the handler returns or throws belongs to no live session, and nothing of it may reach the
// endpoint: a client that connected again there owns it now.
const generation = peer.session.generation;
for (const frame of result.delivered) {
if (frame.channelId !== "schema") continue;
try {
@ -239,6 +244,7 @@ async function handleServerDatagram(
peer.handling -= 1;
peer.lastHandledAtMs = Date.now();
}
if (peer.session.generation !== generation) return true;
const payload = encode(encodeCultNetMessageForWire(response, "cultnet.schema.v0"));
for (const responsePacket of peer.session.sendMany("schema", payload, {
reliable: true,
@ -247,6 +253,7 @@ async function handleServerDatagram(
maxFragmentBytes: options.maxFragmentBytes ?? 2048,
})) sendPacket(key, peer, responsePacket);
} catch {
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);

View file

@ -2825,6 +2825,89 @@ test("server-mode transport admits a restarted client and keeps its session when
}
});
// A request belongs to the session generation it arrived on. A client that connects again
// from the same endpoint while a handler runs owns that endpoint now: nothing the handler
// returns or throws for the old generation may reach it.
async function restartDuringHandler(settle: (release: { resolve: () => void; reject: (error: Error) => void }) => void) {
let release: { resolve: () => void; reject: (error: Error) => void } | undefined;
const server = await startCultNetOperationServer({
runtimeId: "sai-sidecar",
handler: async (request) => {
if (request.messageId === "held") {
await new Promise<void>((resolve, reject) => { release = { resolve, reject: (error) => reject(error) }; });
}
return {
schemaVersion: "cultnet.operation_response.v0",
messageId: request.messageId,
serviceId: request.serviceId,
operation: request.operation,
status: "ok",
payloadSchema: "gamecult.eve.plugin_abi.response.v1",
payloadEncoding: "messagepack-base64",
payload: request.payload,
diagnostics: [],
sourceRuntimeId: "sai-sidecar",
};
},
});
const connectionId = 0x43554c54;
const port = Number(new URL(server.endpoint).port);
const socket = await bindUdpSocket();
let session = new CultNetRudpSession({ connectionId, initialSequence: 50_000 });
const answered: string[] = [];
const goodbyes: string[] = [];
socket.on("message", (wire) => {
const packet = decodeRudpPacket(wire);
if (packet.packetType === "disconnect") goodbyes.push(Buffer.from(packet.payload ?? []).toString("utf8"));
const result = session.receive(packet, Date.now());
if (packet.packetType === "accept" || packet.reliable) {
socket.send(encodeRudpPacket(session.createAckForReceived(packet.sequence)), port, "127.0.0.1");
}
for (const frame of result.delivered) {
answered.push((decodeMsgpack(frame.payload) as { messageId: string }).messageId);
}
});
const toServer = (packet: CultNetRudpPacket) => socket.send(encodeRudpPacket(packet), port, "127.0.0.1");
const request = (messageId: string) => Buffer.from(encode(encodeCultNetMessageForWire({
schemaVersion: "cultnet.operation_request.v0",
messageId,
serviceId: "sai.vn",
operation: "describe",
payloadSchema: "gamecult.eve.plugin_abi.request.v1",
payloadEncoding: "messagepack-base64",
payload: "gaZzY2hlbWE=",
} satisfies CultNetOperationRequestMessage)));
try {
toServer(session.createConnect(Date.now()));
await waitFor(() => session.connected, "the first Accept");
toServer(session.send("schema", request("held"), { reliable: true, ordered: true }));
await waitFor(() => release !== undefined, "the held handler");
session = new CultNetRudpSession({ connectionId, initialSequence: 7 });
toServer(session.createConnect(Date.now()));
await waitFor(() => session.connected, "the restarted client's Accept");
settle(release!);
await new Promise((resolve) => setTimeout(resolve, 100));
toServer(session.send("schema", request("after-restart"), { reliable: true, ordered: true }));
await waitFor(() => answered.includes("after-restart"), "the restarted client's answer");
assert.deepEqual(answered, ["after-restart"], "the old generation's response reached the new session");
assert.deepEqual(goodbyes, [], "a goodbye for the old generation reached the new session");
assert.equal(session.connected, true);
} finally {
socket.close();
await server.close();
}
}
test("an operation response for a generation a new Connect replaced is not sent into the new session", async () => {
await restartDuringHandler((release) => release.resolve());
});
test("an operation handler that fails after a new Connect replaced its generation says no goodbye to the new session", async () => {
await restartDuringHandler((release) => release.reject(new Error("handler failed")));
});
test("RUDP operation service admits a restarted client on the same address", async () => {
const server = await startCultNetOperationServer({
runtimeId: "sai-sidecar",