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

Fence provider RUDP connection generations

Browse source
This commit is contained in:
Meta Vens 2026-07-19 00:49:41 +01:00
commit 5c075bc212
8 changed files with 325 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

5
packages/cultmesh-ts/README.md
View file

@ -280,6 +280,7 @@ const odinTransport = new CultMeshProviderRudpTransport({
endpoint: "rudp://127.0.0.1:17871",
runtimeId: "voidbot-worker-7",
connectionId: 0x43554c54,
sessionToken: process.env.ODIN_PROVIDER_SESSION_TOKEN,
});
const session = new CultMeshProviderSession({
@ -321,6 +322,10 @@ desired documents and durable receipt outbox.
Do not advertise it as a public provider boundary until the surrounding
CultNet session authenticates a claim authorizing the four-part provider
identity. Source address and successful contact are not authority.
The adapter carries `sessionToken` in a typed RUDP Connect-evidence map beside
a fresh `clientSessionId`. Packet retransmits reuse that id; an actual reconnect
mints another. The receiver must decode and validate the token as authorization
evidence for the requested provider identity.
## RUDP Helpers

31
packages/cultmesh-ts/src/provider-session-broker.ts
View file

@ -74,6 +74,8 @@ export class CultMeshProviderSessionBroker {
readonly #commands = new Map<string, CultMeshProviderCommandWire>();
readonly #receipts = new Map<string, CultMeshProviderCommandReceiptWire>();
readonly #publicationOwners = new Map<string, { identityKey: string; publicationId: string }>();
readonly #closedSessions = new Set<string>();
readonly #inFlightBySession = new Map<string, number>();
readonly #expiryTimer: NodeJS.Timeout;
#work: Promise<void> = Promise.resolve();
#leaseSequence = 0;
@ -101,15 +103,34 @@ export class CultMeshProviderSessionBroker {
request: CultNetOperationRequestMessage,
session: CultMeshRudpServerSession,
): Promise<CultNetOperationResponseMessage> {
this.#inFlightBySession.set(
session.sessionId,
(this.#inFlightBySession.get(session.sessionId) ?? 0) + 1,
);
const work = this.#work.then(() => this.#handle(request, session));
this.#work = work.then(() => undefined, () => undefined);
return await work;
try {
return await work;
} finally {
const remaining = (this.#inFlightBySession.get(session.sessionId) ?? 1) - 1;
if (remaining > 0) {
this.#inFlightBySession.set(session.sessionId, remaining);
} else {
this.#inFlightBySession.delete(session.sessionId);
this.#closedSessions.delete(session.sessionId);
}
}
}
async #handle(
request: CultNetOperationRequestMessage,
session: CultMeshRudpServerSession,
): Promise<CultNetOperationResponseMessage> {
if (this.#closedSessions.has(session.sessionId)) {
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
"The physical provider session is closed or replaced.",
]);
}
if (request.serviceId !== CULTMESH_PROVIDER_SESSION_SERVICE_ID) {
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
`Unsupported service ${request.serviceId}.`,
@ -148,6 +169,9 @@ export class CultMeshProviderSessionBroker {
}
public sessionClosed(session: CultMeshRudpServerSession): void {
if ((this.#inFlightBySession.get(session.sessionId) ?? 0) > 0) {
this.#closedSessions.add(session.sessionId);
}
for (const leaseId of this.#sessionLeases.get(session.sessionId) ?? []) {
const record = this.#leases.get(leaseId);
if (record?.session?.sessionId === session.sessionId) record.session = undefined;
@ -207,6 +231,11 @@ export class CultMeshProviderSessionBroker {
"The CultNet session does not authorize this provider identity.",
]);
}
if (this.#closedSessions.has(session.sessionId)) {
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
"The physical provider session closed while registration authority was being checked.",
]);
}
const identityKey = identityTuple(identity);
const previousLeaseId = this.#identityLeases.get(identityKey);
let publications = new Map<string, CultNetRawDocumentRecord>();

23
packages/cultmesh-ts/src/provider-session-wire.ts
View file

@ -31,6 +31,13 @@ export type CultMeshProviderOperationStatus =
| "denied"
| "invalid";
/**
* Transport evidence carried by a provider's RUDP Connect packet.
*
* `clientSessionId` distinguishes an actual reconnect from retransmission of
* the same Connect packet. The token remains authorization evidence only; it
* is not a transport-session identifier.
*/
export interface CultMeshProviderConnectEvidenceWire {
clientSessionId: string;
sessionToken: string | null;
@ -123,21 +130,29 @@ export function decodeProviderSessionPayload<T>(payload: string): T {
return decode(Buffer.from(payload, "base64")) as T;
}
export function encodeProviderConnectEvidence(evidence: CultMeshProviderConnectEvidenceWire): Uint8Array {
export function encodeProviderConnectEvidence(
evidence: CultMeshProviderConnectEvidenceWire,
): Uint8Array {
assertProviderConnectEvidence(evidence);
return encode(evidence);
}
export function decodeProviderConnectEvidence(payload: Uint8Array): CultMeshProviderConnectEvidenceWire {
export function decodeProviderConnectEvidence(
payload: Uint8Array,
): CultMeshProviderConnectEvidenceWire {
const evidence = decode(payload) as unknown;
assertProviderConnectEvidence(evidence);
return evidence;
}
export function assertProviderConnectEvidence(value: unknown): asserts value is CultMeshProviderConnectEvidenceWire {
export function assertProviderConnectEvidence(
value: unknown,
): asserts value is CultMeshProviderConnectEvidenceWire {
const evidence = requireObject(value, "provider Connect evidence");
requireText(evidence.clientSessionId, "clientSessionId");
if (evidence.sessionToken !== null) requireText(evidence.sessionToken, "sessionToken");
if (evidence.sessionToken !== null) {
requireText(evidence.sessionToken, "sessionToken");
}
}
export function decodeProviderCommandDocument(document: CultNetRawDocumentRecord): CultMeshProviderCommandWire {

177
packages/cultmesh-ts/test/provider-rudp-transport.test.ts
View file

@ -1,12 +1,25 @@
import test from "node:test";
import assert from "node:assert/strict";
import { createSocket } from "node:dgram";
import { decode } from "@msgpack/msgpack";
import { CultNetDocumentRegistry } from "cultnet-ts";
import {
CultNetDocumentRegistry,
CultNetRudpSession,
decodeRudpPacket,
encodeRudpPacket,
type CultNetOperationRequestMessage,
} from "cultnet-ts";
import {
CultMesh,
CultMeshProviderRudpTransport,
CultMeshProviderSessionBroker,
CULTMESH_PROVIDER_SESSION_SERVICE_ID,
cultMeshProviderSessionOperations,
cultMeshProviderSessionSchemas,
decodeProviderConnectEvidence,
encodeProviderConnectEvidence,
encodeProviderSessionPayload,
type CultMeshProviderCommand,
type CultMeshProviderIdentity,
type CultMeshProviderPublication,
@ -28,7 +41,9 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle"
const broker = new CultMeshProviderSessionBroker({
runtimeId: "odin-test",
createLeaseId: () => `lease-${++leaseSequence}`,
authorizeRegistration: () => true,
authorizeRegistration: (_identity, session) =>
session.connectPayload !== undefined
&& decodeProviderConnectEvidence(session.connectPayload).sessionToken === "signed-provider-token",
acceptPublication: (_identity, publicationId, document) => {
accepted.set(publicationId, decode(document.payload));
acceptedProvenance.set(publicationId, {
@ -54,10 +69,44 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle"
});
await server.start();
const unauthorizedConnection = await new CultMeshProviderRudpTransport({
endpoint: `rudp://127.0.0.1:${server.bind.port}`,
runtimeId: "unauthorized-provider",
connectionId: 0x43554c54,
sessionToken: "not-authority",
}).connect(identity, new AbortController().signal);
try {
await assert.rejects(
unauthorizedConnection.register(
{ identity, requestedLeaseDurationMs: 10_000 },
new AbortController().signal,
),
/denied|not authorized/i,
);
assert.equal(
broker.activeLeaseCount,
0,
"a successful RUDP handshake does not grant provider authority",
);
} finally {
unauthorizedConnection.close();
}
const providerSocket = createSocket("udp4");
const rawSend = providerSocket.send.bind(providerSocket);
let originalConnectWire: Buffer | undefined;
providerSocket.send = ((message: Uint8Array, ...args: unknown[]) => {
if (decodeRudpPacket(message).packetType === "connect" && !originalConnectWire) {
originalConnectWire = Buffer.from(message);
}
return (rawSend as (...parameters: unknown[]) => boolean)(message, ...args);
}) as typeof providerSocket.send;
const transport = new CultMeshProviderRudpTransport({
endpoint: `rudp://127.0.0.1:${server.bind.port}`,
runtimeId: identity.serviceInstanceId,
connectionId: 0x43554c54,
sessionToken: "signed-provider-token",
socketFactory: () => providerSocket,
});
const connection = await transport.connect(identity, new AbortController().signal);
const secondIdentity: CultMeshProviderIdentity = {
@ -70,6 +119,7 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle"
endpoint: `rudp://127.0.0.1:${server.bind.port}`,
runtimeId: secondIdentity.serviceInstanceId,
connectionId: 0x43554c54,
sessionToken: "signed-provider-token",
}).connect(secondIdentity, new AbortController().signal);
try {
broker.enqueueCommand({
@ -81,6 +131,12 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle"
});
const lease = await connection.register({ identity, requestedLeaseDurationMs: 10_000 }, new AbortController().signal);
assert.equal(lease.leaseId, "lease-1");
assert.ok(originalConnectWire);
await new Promise<void>((resolve, reject) => {
rawSend(originalConnectWire!, server.bind.port, "127.0.0.1", error => error ? reject(error) : resolve());
});
await new Promise(resolve => setTimeout(resolve, 25));
assert.equal(broker.activeLeaseCount, 1, "a duplicate Connect cannot revoke the live provider lease");
const secondLease = await secondConnection.register({
identity: secondIdentity,
requestedLeaseDurationMs: 10_000,
@ -155,7 +211,124 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle"
}
});
test("closed physical sessions cannot finish delayed registration", async () => {
const authorization = deferred<boolean>();
const broker = new CultMeshProviderSessionBroker({
runtimeId: "odin-test",
authorizeRegistration: () => authorization.promise,
acceptPublication: () => undefined,
deletePublications: () => undefined,
acceptReceipt: () => undefined,
});
const session = {
sessionId: "closed-generation",
remote: { address: "127.0.0.1", family: "IPv4", port: 40000 },
connectPayload: new Uint8Array(),
send: () => undefined,
};
const request: CultNetOperationRequestMessage = {
schemaVersion: "cultnet.operation_request.v0",
messageId: "late-register",
serviceId: CULTMESH_PROVIDER_SESSION_SERVICE_ID,
operation: cultMeshProviderSessionOperations.register,
payloadSchema: cultMeshProviderSessionSchemas.registration,
payloadEncoding: "messagepack-base64",
payload: encodeProviderSessionPayload({ ...identity, requestedLeaseDurationMs: 10_000 }),
};
try {
const pending = broker.handle(request, session);
await new Promise(resolve => setTimeout(resolve, 0));
broker.sessionClosed(session);
authorization.resolve(true);
const response = await pending;
assert.equal(response.status, "denied");
assert.equal(broker.activeLeaseCount, 0);
} finally {
broker.close();
}
});
test("a replaced RUDP generation drops operations queued behind an active handler", async () => {
const connectionId = 0x43554c54;
const handlerGate = deferred<void>();
let handlerCalls = 0;
let closedSessions = 0;
const server = CultMesh.createRudpDocumentServer("odin-queue-test", connectionId, {
bindPort: 0,
documents: new CultNetDocumentRegistry(),
onOperationRequest: async request => {
handlerCalls += 1;
if (handlerCalls === 1) await handlerGate.promise;
return {
schemaVersion: "cultnet.operation_response.v0",
messageId: request.messageId,
serviceId: request.serviceId,
operation: request.operation,
status: "denied",
payloadSchema: cultMeshProviderSessionSchemas.mutationAcceptance,
payloadEncoding: "messagepack-base64",
payload: encodeProviderSessionPayload({}),
};
},
onSessionClosed: () => { closedSessions += 1; },
onError: () => undefined,
});
await server.start();
const socket = createSocket("udp4");
const rawSend = socket.send.bind(socket);
const connection = await new CultMeshProviderRudpTransport({
endpoint: `rudp://127.0.0.1:${server.bind.port}`,
runtimeId: identity.serviceInstanceId,
connectionId,
sessionToken: "signed-provider-token",
operationTimeoutMs: 100,
socketFactory: () => socket,
}).connect(identity, new AbortController().signal);
try {
const first = connection.register(
{ identity, requestedLeaseDurationMs: 10_000 },
new AbortController().signal,
);
const second = connection.register(
{ identity, requestedLeaseDurationMs: 10_000 },
new AbortController().signal,
);
await waitFor(() => handlerCalls === 1, "first queued provider operation");
const replacement = new CultNetRudpSession({ connectionId });
const replacementWire = encodeRudpPacket(replacement.createConnect(
Date.now(),
Buffer.from(encodeProviderConnectEvidence({
clientSessionId: "replacement-generation",
sessionToken: "signed-provider-token",
})),
));
await new Promise<void>((resolve, reject) => {
rawSend(replacementWire, server.bind.port, "127.0.0.1", error => error ? reject(error) : resolve());
});
await waitFor(() => closedSessions === 1, "old RUDP generation closure");
handlerGate.resolve();
await new Promise(resolve => setTimeout(resolve, 20));
assert.equal(handlerCalls, 1, "transport-queued work from the closed generation must be dropped");
connection.close();
await Promise.allSettled([first, second]);
} finally {
connection.close();
server.close();
}
});
function deferred<T>(): { promise: Promise<T>; resolve(value: T): void } {
let resolve!: (value: T) => void;
return { promise: new Promise<T>(done => resolve = done), resolve };
}
async function waitFor(predicate: () => boolean, description: string): Promise<void> {
const startedAt = Date.now();
while (!predicate()) {
if (Date.now() - startedAt > 1_000) throw new Error(`Timed out waiting for ${description}.`);
await new Promise(resolve => setTimeout(resolve, 5));
}
}

28
packages/cultmesh-ts/test/provider-session-wire.test.ts
View file

@ -1,9 +1,15 @@
import test from "node:test";
import assert from "node:assert/strict";
import { encodeProviderSessionPayload } from "../src/provider-session-wire";
import {
decodeProviderConnectEvidence,
encodeProviderConnectEvidence,
encodeProviderSessionPayload,
} from "../src/provider-session-wire";
const REGISTRATION_GOLDEN = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03";
const CONNECT_EVIDENCE_GOLDEN = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbrJvZGluLXNlc3Npb24tdG9rZW4=";
const TOKENLESS_CONNECT_EVIDENCE_GOLDEN = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbsA=";
test("provider registration matches the shared C# MessagePack golden payload", () => {
assert.equal(encodeProviderSessionPayload({
@ -15,3 +21,23 @@ test("provider registration matches the shared C# MessagePack golden payload", (
authorityLeaseId: "authority-7",
}), REGISTRATION_GOLDEN);
});
test("provider Connect evidence keeps transport generation separate from authority", () => {
const evidence = {
clientSessionId: "aetheria-client-42",
sessionToken: "odin-session-token",
};
const encoded = encodeProviderConnectEvidence(evidence);
assert.equal(Buffer.from(encoded).toString("base64"), CONNECT_EVIDENCE_GOLDEN);
assert.deepEqual(
decodeProviderConnectEvidence(encoded),
evidence,
);
assert.equal(
Buffer.from(encodeProviderConnectEvidence({
clientSessionId: "aetheria-client-42",
sessionToken: null,
})).toString("base64"),
TOKENLESS_CONNECT_EVIDENCE_GOLDEN,
);
});

8
packages/cultnet-ts/test/cultnet.test.ts
View file

@ -415,6 +415,14 @@ test("rudp session handshake acks reliable connect and accept packets", () => {
assert.equal(server.connected, true);
assert.deepEqual(server.pendingReliableSequences, [100]);
const retransmittedAccept = server.acceptConnect(connect, 15, Buffer.from("ok", "utf8"));
assert.deepEqual(
encodeRudpPacket(retransmittedAccept),
encodeRudpPacket(accept),
"a retransmitted Connect reuses the pending Accept packet",
);
assert.deepEqual(server.pendingReliableSequences, [100]);
client.receive(accept, 20);
assert.equal(client.connected, true);
assert.deepEqual(client.pendingReliableSequences, []);

35
src/GameCult.Mesh/CultMeshProviderSessionWire.cs
View file

@ -60,6 +60,19 @@ namespace GameCult.Mesh
public const string PayloadEncoding = "messagepack-base64";
}
/// <summary>
/// Carries provider authorization evidence and a transport-owned connection
/// generation in an RUDP Connect packet. The token is not a session id.
/// </summary>
[MessagePackObject]
public sealed class CultMeshProviderConnectEvidenceWire
{
/// <summary>Gets or sets the fresh id minted for this physical connection.</summary>
[Key("clientSessionId")] public string ClientSessionId { get; set; } = string.Empty;
/// <summary>Gets or sets the optional signed provider authorization token.</summary>
[Key("sessionToken")] public string? SessionToken { get; set; }
}
/// <summary>Requests a fenced provider lease from the provider-session broker.</summary>
[MessagePackObject]
public sealed class CultMeshProviderRegistrationWire
@ -245,6 +258,24 @@ namespace GameCult.Mesh
return value;
}
/// <summary>Encodes provider RUDP Connect evidence as a named MessagePack map.</summary>
public static byte[] EncodeConnectEvidence(CultMeshProviderConnectEvidenceWire evidence)
{
if (evidence == null) throw new ArgumentNullException(nameof(evidence));
Validate(evidence);
return MessagePackSerializer.Serialize(evidence, Options);
}
/// <summary>Decodes and validates provider RUDP Connect evidence.</summary>
public static CultMeshProviderConnectEvidenceWire DecodeConnectEvidence(byte[] payload)
{
if (payload == null) throw new ArgumentNullException(nameof(payload));
var evidence = MessagePackSerializer.Deserialize<CultMeshProviderConnectEvidenceWire>(payload, Options);
if (evidence == null) throw new MessagePackSerializationException("Provider Connect evidence decoded to null.");
Validate(evidence);
return evidence;
}
/// <summary>Decodes a retained provider command carried as a raw CultNet document.</summary>
public static CultMeshProviderCommandWire DecodeCommandDocument(CultNetRawDocumentRecord document)
{
@ -336,6 +367,10 @@ namespace GameCult.Mesh
{
switch (value)
{
case CultMeshProviderConnectEvidenceWire evidence:
RequireText(evidence.ClientSessionId, nameof(evidence.ClientSessionId));
RequireOptionalText(evidence.SessionToken, nameof(evidence.SessionToken));
break;
case CultMeshProviderRegistrationWire registration:
RequireIdentity(registration.ProviderId, registration.ServiceInstanceId, registration.EndpointId, registration.VerseId);
RequirePositive(registration.RequestedLeaseDurationMs, nameof(registration.RequestedLeaseDurationMs));

26
tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs
View file

@ -10,6 +10,32 @@ namespace GameCult.Mesh.Tests;
public sealed class CultMeshProviderSessionWireTests
{
private const string RegistrationGolden = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03";
private const string ConnectEvidenceGolden = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbrJvZGluLXNlc3Npb24tdG9rZW4=";
private const string TokenlessConnectEvidenceGolden = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbsA=";
[Test]
public void ConnectEvidence_SeparatesTransportGenerationFromAuthorityToken()
{
var encoded = CultMeshProviderSessionWire.EncodeConnectEvidence(
new CultMeshProviderConnectEvidenceWire
{
ClientSessionId = "aetheria-client-42",
SessionToken = "odin-session-token"
});
var decoded = CultMeshProviderSessionWire.DecodeConnectEvidence(encoded);
Assert.Multiple(() =>
{
Assert.That(Convert.ToBase64String(encoded), Is.EqualTo(ConnectEvidenceGolden),
"C#, TypeScript, and Rust must emit the same Connect evidence map");
Assert.That(decoded.ClientSessionId, Is.EqualTo("aetheria-client-42"));
Assert.That(decoded.SessionToken, Is.EqualTo("odin-session-token"));
Assert.That(
Convert.ToBase64String(CultMeshProviderSessionWire.EncodeConnectEvidence(
new CultMeshProviderConnectEvidenceWire { ClientSessionId = "aetheria-client-42" })),
Is.EqualTo(TokenlessConnectEvidenceGolden));
});
}
[Test]
public void Registration_RoundTrips_InsideCanonicalCultNetOperationEnvelope()