Add provider session wire transport
This commit is contained in:
Meta Vens
parent
18cebf740f
commit
262b5cc53d
10 changed files with 2092 additions and 5 deletions
61
docs/cultmesh-provider-session.md
Normal file
61
docs/cultmesh-provider-session.md
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
# CultMesh Provider Session
|
||||
|
||||
## Authority map
|
||||
|
||||
- Owner: the provider-session broker owns registration, active lease fencing,
|
||||
accepted publication membership, command retention, receipt acceptance, and
|
||||
withdrawal.
|
||||
- Inputs: a physical CultNet session, an explicit registration-authorization
|
||||
decision, correlated typed operations, the broker clock, provider commands,
|
||||
and persistence callbacks. The current RUDP adapter does not itself provide
|
||||
an authenticated claim.
|
||||
- Outputs: leases, application acceptances, accepted raw documents, pushed
|
||||
typed commands, persisted receipts, and deterministic removals.
|
||||
- Derived state: socket sessions, retry schedules, connection health, and
|
||||
provider projections are observations. They do not authorize publications.
|
||||
- Forbidden writers: RUDP packet acknowledgements, source addresses, direct raw
|
||||
document callbacks, renderer-owned catalogs, and arrival-time freshness maps
|
||||
cannot create accepted provider truth.
|
||||
- Shared paths: initial connect and reconnect use `provider.register`; initial,
|
||||
replayed, and live documents use `provider.publication.put`; explicit
|
||||
withdrawal and lease expiry remove the broker-owned publication set; command
|
||||
retry ends only when `provider.receipt.put` is accepted.
|
||||
- Cut line: a receiver must stop accepting unleased provider raw puts before it
|
||||
claims this protocol as its ingress authority.
|
||||
|
||||
## Wire contract
|
||||
|
||||
Provider lifecycle operations use the reliable ordered CultNet schema channel
|
||||
and the existing `cultnet.operation_request.v0` / `cultnet.operation_response.v0`
|
||||
envelopes with service id `gamecult.mesh.provider_session`. Inner payloads are
|
||||
camel-case MessagePack maps encoded as `messagepack-base64`.
|
||||
|
||||
The v1 operations are:
|
||||
|
||||
| Operation | Request payload | Successful payload |
|
||||
| --- | --- | --- |
|
||||
| `provider.register` | `gamecult.mesh.provider_registration.v1` | `gamecult.mesh.provider_lease.v1` |
|
||||
| `provider.renew` | `gamecult.mesh.provider_lease_renewal.v1` | `gamecult.mesh.provider_lease.v1` |
|
||||
| `provider.publication.put` | `gamecult.mesh.provider_publication_put.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` |
|
||||
| `provider.publication.delete` | `gamecult.mesh.provider_publication_delete.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` |
|
||||
| `provider.receipt.put` | `gamecult.mesh.provider_receipt_put.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` |
|
||||
| `provider.withdraw` | `gamecult.mesh.provider_withdrawal.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` |
|
||||
|
||||
Broker-to-provider commands are raw typed documents using
|
||||
`gamecult.mesh.provider_command.v1`. Asset bodies do not travel inside these
|
||||
operation payloads.
|
||||
|
||||
Responses use `ok`, `conflict`, `expired`, `denied`, or `invalid`. A transport
|
||||
ACK is never an application acceptance. Renewal issues a new lease id and
|
||||
immediately fences the previous id. The broker derives the set removed by full
|
||||
withdrawal; a provider cannot use a supplied list to hide owned publications.
|
||||
|
||||
Public deployment additionally requires a CultNet authentication claim that
|
||||
authorizes the provider identity. Until that claim is wired and checked, the
|
||||
TypeScript RUDP adapter is a private-development transport.
|
||||
|
||||
`CultMeshProviderSessionBroker` is an in-memory reference broker and socket
|
||||
proof. Its command and receipt retention does not survive process restart.
|
||||
Production receivers must persist that control-plane state atomically with
|
||||
accepted publication ownership; Odin must not substitute these volatile maps
|
||||
for its durable broker authority.
|
||||
|
|
@ -271,7 +271,16 @@ publication, and withdrawal. Provider code owns domain state and idempotent
|
|||
command transactions; the injected transport owns CultNet wire details.
|
||||
|
||||
```ts
|
||||
import { CultMeshProviderSession } from "cultmesh-ts";
|
||||
import {
|
||||
CultMeshProviderRudpTransport,
|
||||
CultMeshProviderSession,
|
||||
} from "cultmesh-ts";
|
||||
|
||||
const odinTransport = new CultMeshProviderRudpTransport({
|
||||
endpoint: "rudp://127.0.0.1:17871",
|
||||
runtimeId: "voidbot-worker-7",
|
||||
connectionId: 0x43554c54,
|
||||
});
|
||||
|
||||
const session = new CultMeshProviderSession({
|
||||
identity: {
|
||||
|
|
@ -300,6 +309,19 @@ is persisted before transport publication, remains pending across reconnects,
|
|||
and is marked published only after the current connection accepts it. Receipt
|
||||
store failure degrades command intake without inventing a network outage.
|
||||
|
||||
The RUDP provider transport carries lifecycle payloads inside the existing
|
||||
`cultnet.operation_request.v0` and `cultnet.operation_response.v0` envelopes.
|
||||
RUDP acknowledgements prove byte delivery only. Registration, renewal,
|
||||
publication changes, receipts, and withdrawal complete only after the
|
||||
provider-session broker returns a correlated application response. The broker
|
||||
owns lease fencing and accepted publication membership; the provider owns its
|
||||
desired documents and durable receipt outbox.
|
||||
|
||||
`CultMeshProviderRudpTransport` is currently a private-development transport.
|
||||
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.
|
||||
|
||||
## RUDP Helpers
|
||||
|
||||
Node runtimes can build the shared CultNet reliable-UDP transport from the
|
||||
|
|
|
|||
|
|
@ -25,6 +25,8 @@ import {
|
|||
type CultNetErrorMessage,
|
||||
type CultNetDocumentBinding,
|
||||
type CultNetMessage,
|
||||
type CultNetOperationRequestMessage,
|
||||
type CultNetOperationResponseMessage,
|
||||
type CultNetRawDocumentRecord,
|
||||
type CultNetReconnectPolicy,
|
||||
type CultNetSchemaCatalogOptions,
|
||||
|
|
@ -33,6 +35,9 @@ import {
|
|||
} from "cultnet-ts";
|
||||
|
||||
export * from "./provider-session";
|
||||
export * from "./provider-session-wire";
|
||||
export * from "./provider-rudp-transport";
|
||||
export * from "./provider-session-broker";
|
||||
|
||||
export interface CultMeshVec2 {
|
||||
readonly x: number;
|
||||
|
|
@ -4095,10 +4100,21 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt
|
|||
getCache?: () => Promise<CultCache> | CultCache;
|
||||
onError?: (error: Error) => void;
|
||||
onDocumentPutRaw?: (document: CultMeshRudpDocumentPut) => void | Promise<void>;
|
||||
onOperationRequest?: (
|
||||
request: CultNetOperationRequestMessage,
|
||||
session: CultMeshRudpServerSession,
|
||||
) => CultNetOperationResponseMessage | Promise<CultNetOperationResponseMessage>;
|
||||
onSessionClosed?: (session: CultMeshRudpServerSession) => void | Promise<void>;
|
||||
wireContract?: CultNetWireContract;
|
||||
sessionTimeoutMs?: number;
|
||||
}
|
||||
|
||||
export interface CultMeshRudpServerSession {
|
||||
readonly sessionId: string;
|
||||
readonly remote: { address: string; family: string; port: number };
|
||||
send(message: CultNetMessage): void;
|
||||
}
|
||||
|
||||
export interface CultMeshRudpDocumentServer {
|
||||
readonly bind: { host: string; port: number };
|
||||
start(): Promise<void>;
|
||||
|
|
@ -5542,9 +5558,12 @@ export class CultMesh {
|
|||
const bind = { host, port };
|
||||
const socket = options.socket ?? createSocket(host.includes(":") ? "udp6" : "udp4");
|
||||
const sessions = new Map<string, {
|
||||
sessionId: string;
|
||||
remote: { address: string; family: string; port: number };
|
||||
session: CultNetRudpSession;
|
||||
work: Promise<void>;
|
||||
}>();
|
||||
let sessionSequence = 0;
|
||||
const resendPollMs = Math.max(10, options.resendPollMs ?? 25);
|
||||
const sessionTimeoutMs = Math.max(1_000, options.sessionTimeoutMs ?? 30_000);
|
||||
const wireContract = options.wireContract ?? "cultnet.schema.v0";
|
||||
|
|
@ -5567,7 +5586,9 @@ export class CultMesh {
|
|||
let record = sessions.get(key);
|
||||
|
||||
if (packet.packetType === "connect") {
|
||||
if (record) notifySessionClosed(record);
|
||||
record = {
|
||||
sessionId: `${runtimeId}:${++sessionSequence}`,
|
||||
remote: { address: remote.address, family: remote.family, port: remote.port },
|
||||
session: new CultNetRudpSession({
|
||||
connectionId,
|
||||
|
|
@ -5575,6 +5596,7 @@ export class CultMesh {
|
|||
resendDelayMs: options.resendDelayMs,
|
||||
maxPendingReliablePackets: options.maxPendingReliablePackets,
|
||||
}),
|
||||
work: Promise.resolve(),
|
||||
};
|
||||
sessions.set(key, record);
|
||||
socket.send(encodeRudpPacket(record.session.acceptConnect(packet, nowMs)), remote.port, remote.address);
|
||||
|
|
@ -5593,12 +5615,13 @@ export class CultMesh {
|
|||
if (frame.channelId !== "schema") {
|
||||
continue;
|
||||
}
|
||||
handleDocumentServerFrame(record, frame.payload).catch((error) => {
|
||||
reportError(error);
|
||||
});
|
||||
record.work = record.work
|
||||
.then(() => handleDocumentServerFrame(record, frame.payload))
|
||||
.catch(reportError);
|
||||
}
|
||||
if (result.disconnected) {
|
||||
sessions.delete(key);
|
||||
notifySessionClosed(record);
|
||||
return;
|
||||
}
|
||||
if (packet.packetType === "accept" || result.delivered.length > 0) {
|
||||
|
|
@ -5611,7 +5634,7 @@ export class CultMesh {
|
|||
socket.on("error", reportError);
|
||||
|
||||
async function handleDocumentServerFrame(
|
||||
record: { remote: { address: string; family: string; port: number }; session: CultNetRudpSession },
|
||||
record: { sessionId: string; remote: { address: string; family: string; port: number }; session: CultNetRudpSession },
|
||||
payload: Uint8Array,
|
||||
): Promise<void> {
|
||||
const message = parseCultNetMessage(decode(payload), wireContract);
|
||||
|
|
@ -5643,6 +5666,16 @@ export class CultMesh {
|
|||
await options.onDocumentPutRaw(normalizeRudpDocumentPut(message, record.remote));
|
||||
}
|
||||
return;
|
||||
case "cultnet.operation_request.v0":
|
||||
if (!options.onOperationRequest) {
|
||||
sendSchemaMessage(record, {
|
||||
schemaVersion: "cultnet.error.v0",
|
||||
error: "CultMesh RUDP document server has no operation handler.",
|
||||
});
|
||||
return;
|
||||
}
|
||||
sendSchemaMessage(record, await options.onOperationRequest(message, publicSession(record)));
|
||||
return;
|
||||
default:
|
||||
sendSchemaMessage(record, {
|
||||
schemaVersion: "cultnet.error.v0",
|
||||
|
|
@ -5651,6 +5684,27 @@ export class CultMesh {
|
|||
}
|
||||
}
|
||||
|
||||
function publicSession(record: {
|
||||
sessionId: string;
|
||||
remote: { address: string; family: string; port: number };
|
||||
session: CultNetRudpSession;
|
||||
}): CultMeshRudpServerSession {
|
||||
return {
|
||||
sessionId: record.sessionId,
|
||||
remote: { ...record.remote },
|
||||
send: message => sendSchemaMessage(record, message),
|
||||
};
|
||||
}
|
||||
|
||||
function notifySessionClosed(record: {
|
||||
sessionId: string;
|
||||
remote: { address: string; family: string; port: number };
|
||||
session: CultNetRudpSession;
|
||||
}): void {
|
||||
if (!options.onSessionClosed) return;
|
||||
void Promise.resolve(options.onSessionClosed(publicSession(record))).catch(reportError);
|
||||
}
|
||||
|
||||
function sendSchemaMessage(
|
||||
record: { remote: { address: string; port: number }; session: CultNetRudpSession },
|
||||
message: CultNetMessage,
|
||||
|
|
@ -5684,6 +5738,7 @@ export class CultMesh {
|
|||
for (const [key, record] of sessions) {
|
||||
if (record.session.checkTimeout(nowMs, sessionTimeoutMs)) {
|
||||
sessions.delete(key);
|
||||
notifySessionClosed(record);
|
||||
continue;
|
||||
}
|
||||
for (const packet of record.session.dueResends(nowMs)) {
|
||||
|
|
@ -5698,6 +5753,7 @@ export class CultMesh {
|
|||
clearInterval(resendTimer);
|
||||
resendTimer = undefined;
|
||||
}
|
||||
for (const record of sessions.values()) notifySessionClosed(record);
|
||||
sessions.clear();
|
||||
socket.close();
|
||||
},
|
||||
|
|
|
|||
390
packages/cultmesh-ts/src/provider-rudp-transport.ts
Normal file
390
packages/cultmesh-ts/src/provider-rudp-transport.ts
Normal file
|
|
@ -0,0 +1,390 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { createSocket, type Socket } from "node:dgram";
|
||||
import { encode } from "@msgpack/msgpack";
|
||||
import {
|
||||
CultNetPeer,
|
||||
CultNetRudpSocketTransportConnection,
|
||||
type CultNetMessage,
|
||||
type CultNetOperationRequestMessage,
|
||||
type CultNetOperationResponseMessage,
|
||||
type CultNetRawDocumentRecord,
|
||||
} from "cultnet-ts";
|
||||
|
||||
import type {
|
||||
CultMeshProviderCommand,
|
||||
CultMeshProviderCommandListener,
|
||||
CultMeshProviderCommandReceipt,
|
||||
CultMeshProviderConnection,
|
||||
CultMeshProviderIdentity,
|
||||
CultMeshProviderLease,
|
||||
CultMeshProviderPublication,
|
||||
CultMeshProviderRegistration,
|
||||
CultMeshProviderTransport,
|
||||
CultMeshProviderUnsubscribe,
|
||||
CultMeshProviderWithdrawal,
|
||||
} from "./provider-session";
|
||||
import {
|
||||
CULTMESH_PROVIDER_SESSION_SERVICE_ID,
|
||||
assertMutationAcceptance,
|
||||
assertProviderLease,
|
||||
cultMeshProviderSessionOperations,
|
||||
cultMeshProviderSessionSchemas,
|
||||
decodeProviderCommandDocument,
|
||||
decodeProviderSessionPayload,
|
||||
encodeProviderSessionPayload,
|
||||
type CultMeshProviderCommandReceiptWire,
|
||||
type CultMeshProviderLeaseWire,
|
||||
type CultMeshProviderMutationAcceptanceWire,
|
||||
} from "./provider-session-wire";
|
||||
|
||||
export interface CultMeshProviderRudpTransportOptions {
|
||||
readonly endpoint: string;
|
||||
readonly runtimeId: string;
|
||||
readonly connectionId: number;
|
||||
readonly bindHost?: string;
|
||||
readonly connectTimeoutMs?: number;
|
||||
readonly operationTimeoutMs?: number;
|
||||
readonly maxFragmentBytes?: number;
|
||||
readonly socketFactory?: () => Socket;
|
||||
}
|
||||
|
||||
/**
|
||||
* Private-development provider transport. Public deployment additionally needs
|
||||
* an authenticated CultNet session whose claims authorize the provider identity.
|
||||
*/
|
||||
export class CultMeshProviderRudpTransport implements CultMeshProviderTransport {
|
||||
readonly #options: CultMeshProviderRudpTransportOptions;
|
||||
|
||||
public constructor(options: CultMeshProviderRudpTransportOptions) {
|
||||
if (!options.endpoint) throw new Error("CultMesh provider RUDP transport requires endpoint.");
|
||||
if (!options.runtimeId) throw new Error("CultMesh provider RUDP transport requires runtimeId.");
|
||||
if (!Number.isInteger(options.connectionId)) throw new Error("CultMesh provider RUDP transport requires integer connectionId.");
|
||||
this.#options = options;
|
||||
}
|
||||
|
||||
public async connect(
|
||||
identity: CultMeshProviderIdentity,
|
||||
signal: AbortSignal,
|
||||
): Promise<CultMeshProviderConnection> {
|
||||
if (signal.aborted) throw abortError();
|
||||
const endpoint = parseEndpoint(this.#options.endpoint);
|
||||
const socket = this.#options.socketFactory?.() ?? createSocket(endpoint.host.includes(":") ? "udp6" : "udp4");
|
||||
await bindSocket(socket, this.#options.bindHost ?? (endpoint.host.includes(":") ? "::" : "0.0.0.0"), signal);
|
||||
const transport = new CultNetRudpSocketTransportConnection({
|
||||
mode: "client",
|
||||
runtimeId: this.#options.runtimeId,
|
||||
transportId: `${this.#options.runtimeId}.cultmesh-provider-session`,
|
||||
socket,
|
||||
remoteHost: endpoint.host,
|
||||
remotePort: endpoint.port,
|
||||
connectionId: this.#options.connectionId,
|
||||
maxFragmentBytes: this.#options.maxFragmentBytes ?? 2_048,
|
||||
});
|
||||
const peer = new CultNetPeer(transport, { wireContract: "cultnet.schema.v0" });
|
||||
transport.connect();
|
||||
try {
|
||||
await waitUntilConnected(transport, signal, this.#options.connectTimeoutMs ?? 2_000);
|
||||
return new RudpProviderConnection(identity, peer, transport, this.#options.operationTimeoutMs ?? 5_000);
|
||||
} catch (error) {
|
||||
transport.close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class RudpProviderConnection implements CultMeshProviderConnection {
|
||||
readonly #identity: CultMeshProviderIdentity;
|
||||
readonly #peer: CultNetPeer;
|
||||
readonly #transport: CultNetRudpSocketTransportConnection;
|
||||
readonly #timeoutMs: number;
|
||||
readonly #listeners = new Set<CultMeshProviderCommandListener>();
|
||||
readonly #pending = new Map<string, {
|
||||
operation: string;
|
||||
payloadSchema: string;
|
||||
resolve: (response: CultNetOperationResponseMessage) => void;
|
||||
reject: (error: Error) => void;
|
||||
timer: NodeJS.Timeout;
|
||||
}>();
|
||||
readonly #publications = new Map<string, CultMeshProviderPublication>();
|
||||
readonly #bufferedCommands = new Map<string, CultMeshProviderCommand>();
|
||||
#closed = false;
|
||||
#requestedLeaseDurationMs = 30_000;
|
||||
|
||||
public constructor(
|
||||
identity: CultMeshProviderIdentity,
|
||||
peer: CultNetPeer,
|
||||
transport: CultNetRudpSocketTransportConnection,
|
||||
timeoutMs: number,
|
||||
) {
|
||||
this.#identity = identity;
|
||||
this.#peer = peer;
|
||||
this.#transport = transport;
|
||||
this.#timeoutMs = timeoutMs;
|
||||
peer.on("message", message => this.#onMessage(message));
|
||||
peer.on("invalidMessage", error => this.#failPending(error));
|
||||
peer.on("close", () => this.#failPending(new Error("CultMesh provider RUDP connection closed.")));
|
||||
peer.on("error", error => this.#failPending(error));
|
||||
}
|
||||
|
||||
public async register(
|
||||
registration: CultMeshProviderRegistration,
|
||||
signal: AbortSignal,
|
||||
): Promise<CultMeshProviderLease> {
|
||||
this.#requestedLeaseDurationMs = registration.requestedLeaseDurationMs;
|
||||
const response = await this.#operation(
|
||||
cultMeshProviderSessionOperations.register,
|
||||
cultMeshProviderSessionSchemas.registration,
|
||||
cultMeshProviderSessionSchemas.lease,
|
||||
{ ...registration.identity, requestedLeaseDurationMs: registration.requestedLeaseDurationMs },
|
||||
signal,
|
||||
);
|
||||
const wire = decodeProviderSessionPayload<CultMeshProviderLeaseWire>(response.payload);
|
||||
assertProviderLease(wire);
|
||||
assertIdentity(this.#identity, wire);
|
||||
return { leaseId: wire.leaseId, expiresAt: new Date(wire.expiresAtUtc) };
|
||||
}
|
||||
|
||||
public async renew(lease: CultMeshProviderLease): Promise<CultMeshProviderLease> {
|
||||
const response = await this.#operation(
|
||||
cultMeshProviderSessionOperations.renew,
|
||||
cultMeshProviderSessionSchemas.leaseRenewal,
|
||||
cultMeshProviderSessionSchemas.lease,
|
||||
{ leaseId: lease.leaseId, requestedLeaseDurationMs: this.#requestedLeaseDurationMs },
|
||||
);
|
||||
const wire = decodeProviderSessionPayload<CultMeshProviderLeaseWire>(response.payload);
|
||||
assertProviderLease(wire);
|
||||
assertIdentity(this.#identity, wire);
|
||||
return { leaseId: wire.leaseId, expiresAt: new Date(wire.expiresAtUtc) };
|
||||
}
|
||||
|
||||
public async publish(
|
||||
publication: CultMeshProviderPublication,
|
||||
lease: CultMeshProviderLease,
|
||||
): Promise<void> {
|
||||
const document: CultNetRawDocumentRecord = {
|
||||
schemaId: publication.schemaId,
|
||||
recordKey: publication.recordKey,
|
||||
storedAt: new Date().toISOString(),
|
||||
payloadEncoding: "messagepack",
|
||||
payload: encode(publication.value),
|
||||
sourceRuntimeId: this.#identity.serviceInstanceId,
|
||||
sourceAgentId: this.#identity.providerId,
|
||||
sourceRole: "provider",
|
||||
};
|
||||
await this.#accepted(cultMeshProviderSessionOperations.publicationPut, cultMeshProviderSessionSchemas.publicationPut, {
|
||||
leaseId: lease.leaseId,
|
||||
publicationId: publication.publicationId,
|
||||
document,
|
||||
});
|
||||
this.#publications.set(publication.publicationId, publication);
|
||||
}
|
||||
|
||||
public async withdrawPublication(publicationId: string, lease: CultMeshProviderLease): Promise<void> {
|
||||
const publication = this.#publications.get(publicationId);
|
||||
if (!publication) throw new Error(`CultMesh provider publication ${publicationId} is not owned by this connection.`);
|
||||
await this.#accepted(cultMeshProviderSessionOperations.publicationDelete, cultMeshProviderSessionSchemas.publicationDelete, {
|
||||
leaseId: lease.leaseId,
|
||||
publicationId,
|
||||
schemaId: publication.schemaId,
|
||||
recordKey: publication.recordKey,
|
||||
});
|
||||
this.#publications.delete(publicationId);
|
||||
}
|
||||
|
||||
public watchCommands(listener: CultMeshProviderCommandListener): CultMeshProviderUnsubscribe {
|
||||
this.#listeners.add(listener);
|
||||
for (const [commandId, command] of this.#bufferedCommands) {
|
||||
this.#bufferedCommands.delete(commandId);
|
||||
void Promise.resolve(listener(command)).catch(() => undefined);
|
||||
}
|
||||
return () => {
|
||||
this.#listeners.delete(listener);
|
||||
};
|
||||
}
|
||||
|
||||
public async publishReceipt(
|
||||
receipt: CultMeshProviderCommandReceipt,
|
||||
lease: CultMeshProviderLease,
|
||||
): Promise<void> {
|
||||
const wire: CultMeshProviderCommandReceiptWire = {
|
||||
receiptId: receipt.receiptId,
|
||||
commandId: receipt.commandId,
|
||||
commandKind: receipt.commandKind,
|
||||
providerId: receipt.providerId,
|
||||
serviceInstanceId: receipt.serviceInstanceId,
|
||||
state: receipt.state,
|
||||
completedAtUtc: receipt.completedAt.toISOString(),
|
||||
result: receipt.result,
|
||||
error: receipt.error,
|
||||
};
|
||||
await this.#accepted(cultMeshProviderSessionOperations.receiptPut, cultMeshProviderSessionSchemas.receiptPut, {
|
||||
leaseId: lease.leaseId,
|
||||
receipt: wire,
|
||||
});
|
||||
}
|
||||
|
||||
public async withdraw(withdrawal: CultMeshProviderWithdrawal): Promise<void> {
|
||||
await this.#accepted(cultMeshProviderSessionOperations.withdraw, cultMeshProviderSessionSchemas.withdrawal, {
|
||||
leaseId: withdrawal.leaseId,
|
||||
});
|
||||
this.#publications.clear();
|
||||
}
|
||||
|
||||
public close(): void {
|
||||
if (this.#closed) return;
|
||||
this.#closed = true;
|
||||
this.#failPending(new Error("CultMesh provider RUDP connection closed."));
|
||||
this.#listeners.clear();
|
||||
this.#transport.close();
|
||||
}
|
||||
|
||||
async #accepted(operation: string, payloadSchema: string, payload: unknown): Promise<CultMeshProviderMutationAcceptanceWire> {
|
||||
const response = await this.#operation(operation, payloadSchema, cultMeshProviderSessionSchemas.mutationAcceptance, payload);
|
||||
const acceptance = decodeProviderSessionPayload<CultMeshProviderMutationAcceptanceWire>(response.payload);
|
||||
assertMutationAcceptance(acceptance);
|
||||
return acceptance;
|
||||
}
|
||||
|
||||
async #operation(
|
||||
operation: string,
|
||||
payloadSchema: string,
|
||||
responsePayloadSchema: string,
|
||||
payload: unknown,
|
||||
signal?: AbortSignal,
|
||||
): Promise<CultNetOperationResponseMessage> {
|
||||
if (this.#closed) throw new Error("CultMesh provider RUDP connection is closed.");
|
||||
if (signal?.aborted) throw abortError();
|
||||
const messageId = randomUUID();
|
||||
const request: CultNetOperationRequestMessage = {
|
||||
schemaVersion: "cultnet.operation_request.v0",
|
||||
messageId,
|
||||
serviceId: CULTMESH_PROVIDER_SESSION_SERVICE_ID,
|
||||
operation,
|
||||
payloadSchema,
|
||||
payloadEncoding: "messagepack-base64",
|
||||
payload: encodeProviderSessionPayload(payload),
|
||||
sourceRuntimeId: this.#identity.serviceInstanceId,
|
||||
};
|
||||
return await new Promise<CultNetOperationResponseMessage>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
this.#pending.delete(messageId);
|
||||
reject(new Error(`CultMesh provider operation ${operation} timed out.`));
|
||||
}, this.#timeoutMs);
|
||||
const abort = (): void => {
|
||||
clearTimeout(timer);
|
||||
this.#pending.delete(messageId);
|
||||
reject(abortError());
|
||||
};
|
||||
signal?.addEventListener("abort", abort, { once: true });
|
||||
this.#pending.set(messageId, {
|
||||
operation,
|
||||
payloadSchema: responsePayloadSchema,
|
||||
timer,
|
||||
resolve: response => {
|
||||
signal?.removeEventListener("abort", abort);
|
||||
if (response.serviceId !== CULTMESH_PROVIDER_SESSION_SERVICE_ID || response.operation !== operation) {
|
||||
reject(new Error(`CultMesh provider response correlation does not match ${operation}.`));
|
||||
return;
|
||||
}
|
||||
if (response.payloadEncoding !== "messagepack-base64") {
|
||||
reject(new Error(`CultMesh provider response for ${operation} has incompatible payload encoding.`));
|
||||
return;
|
||||
}
|
||||
if (response.status !== "ok") {
|
||||
reject(new Error(`CultMesh provider operation ${operation} was ${response.status}: ${(response.diagnostics ?? []).join("; ")}`));
|
||||
return;
|
||||
}
|
||||
if (response.payloadSchema !== responsePayloadSchema) {
|
||||
reject(new Error(`CultMesh provider response for ${operation} has incompatible payload schema.`));
|
||||
return;
|
||||
}
|
||||
resolve(response);
|
||||
},
|
||||
reject: error => {
|
||||
signal?.removeEventListener("abort", abort);
|
||||
reject(error);
|
||||
},
|
||||
});
|
||||
this.#peer.send(request);
|
||||
});
|
||||
}
|
||||
|
||||
#onMessage(message: CultNetMessage): void {
|
||||
if (message.schemaVersion === "cultnet.operation_response.v0") {
|
||||
const pending = this.#pending.get(message.messageId);
|
||||
if (!pending) return;
|
||||
this.#pending.delete(message.messageId);
|
||||
clearTimeout(pending.timer);
|
||||
pending.resolve(message);
|
||||
return;
|
||||
}
|
||||
if (message.schemaVersion !== "cultnet.document_put_raw.v0") return;
|
||||
if (message.document.schemaId !== cultMeshProviderSessionSchemas.command) return;
|
||||
const wire = decodeProviderCommandDocument(message.document);
|
||||
if (wire.providerId !== this.#identity.providerId || wire.serviceInstanceId !== this.#identity.serviceInstanceId) return;
|
||||
const command: CultMeshProviderCommand = { ...wire };
|
||||
if (this.#listeners.size === 0) {
|
||||
this.#bufferedCommands.set(command.commandId, command);
|
||||
return;
|
||||
}
|
||||
for (const listener of [...this.#listeners]) void Promise.resolve(listener(command)).catch(() => undefined);
|
||||
}
|
||||
|
||||
#failPending(error: Error): void {
|
||||
for (const [messageId, pending] of this.#pending) {
|
||||
this.#pending.delete(messageId);
|
||||
clearTimeout(pending.timer);
|
||||
pending.reject(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function assertIdentity(identity: CultMeshProviderIdentity, lease: CultMeshProviderLeaseWire): void {
|
||||
for (const key of ["providerId", "serviceInstanceId", "endpointId", "verseId"] as const) {
|
||||
if (identity[key] !== lease[key]) throw new Error(`CultMesh provider lease ${key} does not match the registered identity.`);
|
||||
}
|
||||
}
|
||||
|
||||
function parseEndpoint(endpoint: string): { host: string; port: number } {
|
||||
const url = new URL(endpoint);
|
||||
const port = Number(url.port);
|
||||
if (url.protocol !== "rudp:" || !url.hostname || !Number.isInteger(port) || port <= 0 || port > 65_535) {
|
||||
throw new Error(`Invalid CultMesh provider RUDP endpoint ${endpoint}.`);
|
||||
}
|
||||
return { host: url.hostname, port };
|
||||
}
|
||||
|
||||
function bindSocket(socket: Socket, host: string, signal: AbortSignal): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const abort = (): void => {
|
||||
socket.close();
|
||||
reject(abortError());
|
||||
};
|
||||
signal.addEventListener("abort", abort, { once: true });
|
||||
socket.once("error", reject);
|
||||
socket.bind(0, host, () => {
|
||||
signal.removeEventListener("abort", abort);
|
||||
socket.removeListener("error", reject);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function waitUntilConnected(
|
||||
transport: CultNetRudpSocketTransportConnection,
|
||||
signal: AbortSignal,
|
||||
timeoutMs: number,
|
||||
): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (!transport.connected) {
|
||||
if (signal.aborted) throw abortError();
|
||||
if (Date.now() >= deadline) throw new Error("CultMesh provider RUDP connection timed out.");
|
||||
await new Promise(resolve => setTimeout(resolve, 5));
|
||||
}
|
||||
}
|
||||
|
||||
function abortError(): Error {
|
||||
const error = new Error("CultMesh provider RUDP operation aborted.");
|
||||
error.name = "AbortError";
|
||||
return error;
|
||||
}
|
||||
499
packages/cultmesh-ts/src/provider-session-broker.ts
Normal file
499
packages/cultmesh-ts/src/provider-session-broker.ts
Normal file
|
|
@ -0,0 +1,499 @@
|
|||
import { encode } from "@msgpack/msgpack";
|
||||
import type {
|
||||
CultNetDocumentPutRawMessage,
|
||||
CultNetOperationRequestMessage,
|
||||
CultNetOperationResponseMessage,
|
||||
CultNetRawDocumentRecord,
|
||||
} from "cultnet-ts";
|
||||
|
||||
import type { CultMeshRudpServerSession } from "./index";
|
||||
import {
|
||||
CULTMESH_PROVIDER_SESSION_SERVICE_ID,
|
||||
cultMeshProviderSessionOperations,
|
||||
cultMeshProviderSessionSchemas,
|
||||
decodeProviderSessionPayload,
|
||||
encodeProviderSessionPayload,
|
||||
type CultMeshProviderCommandReceiptWire,
|
||||
type CultMeshProviderCommandWire,
|
||||
type CultMeshProviderLeaseRenewalWire,
|
||||
type CultMeshProviderLeaseWire,
|
||||
type CultMeshProviderOperationStatus,
|
||||
type CultMeshProviderPublicationDeleteWire,
|
||||
type CultMeshProviderPublicationPutWire,
|
||||
type CultMeshProviderReceiptPutWire,
|
||||
type CultMeshProviderRegistrationWire,
|
||||
type CultMeshProviderWithdrawalWire,
|
||||
} from "./provider-session-wire";
|
||||
|
||||
export interface CultMeshProviderBrokerClock {
|
||||
now(): Date;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderSessionBrokerOptions {
|
||||
readonly runtimeId: string;
|
||||
readonly clock?: CultMeshProviderBrokerClock;
|
||||
readonly createLeaseId?: () => string;
|
||||
readonly maximumLeaseDurationMs?: number;
|
||||
readonly expiryPollMs?: number;
|
||||
readonly onError?: (error: Error) => void;
|
||||
readonly authorizeRegistration: (
|
||||
identity: CultMeshProviderRegistrationWire,
|
||||
session: CultMeshRudpServerSession,
|
||||
) => boolean | Promise<boolean>;
|
||||
readonly acceptPublication: (
|
||||
identity: CultMeshProviderRegistrationWire,
|
||||
publicationId: string,
|
||||
document: CultNetRawDocumentRecord,
|
||||
) => void | Promise<void>;
|
||||
readonly deletePublications: (
|
||||
identity: CultMeshProviderRegistrationWire,
|
||||
publications: readonly {
|
||||
publicationId: string;
|
||||
document: Pick<CultNetRawDocumentRecord, "schemaId" | "recordKey">;
|
||||
}[],
|
||||
) => void | Promise<void>;
|
||||
readonly acceptReceipt: (
|
||||
identity: CultMeshProviderRegistrationWire,
|
||||
receipt: CultMeshProviderCommandReceiptWire,
|
||||
) => void | Promise<void>;
|
||||
}
|
||||
|
||||
interface LeaseRecord {
|
||||
identity: CultMeshProviderRegistrationWire;
|
||||
lease: CultMeshProviderLeaseWire;
|
||||
session?: CultMeshRudpServerSession;
|
||||
publications: Map<string, CultNetRawDocumentRecord>;
|
||||
}
|
||||
|
||||
export class CultMeshProviderSessionBroker {
|
||||
readonly #options: CultMeshProviderSessionBrokerOptions;
|
||||
readonly #clock: CultMeshProviderBrokerClock;
|
||||
readonly #leases = new Map<string, LeaseRecord>();
|
||||
readonly #identityLeases = new Map<string, string>();
|
||||
readonly #sessionLeases = new Map<string, Set<string>>();
|
||||
readonly #commands = new Map<string, CultMeshProviderCommandWire>();
|
||||
readonly #receipts = new Map<string, CultMeshProviderCommandReceiptWire>();
|
||||
readonly #publicationOwners = new Map<string, { identityKey: string; publicationId: string }>();
|
||||
readonly #expiryTimer: NodeJS.Timeout;
|
||||
#work: Promise<void> = Promise.resolve();
|
||||
#leaseSequence = 0;
|
||||
|
||||
public constructor(options: CultMeshProviderSessionBrokerOptions) {
|
||||
if (!options.runtimeId) throw new Error("CultMesh provider-session broker requires runtimeId.");
|
||||
if (!options.authorizeRegistration) throw new Error("CultMesh provider-session broker requires an explicit registration authority.");
|
||||
this.#options = options;
|
||||
this.#clock = options.clock ?? { now: () => new Date() };
|
||||
this.#expiryTimer = setInterval(() => {
|
||||
void this.expireLeases().catch(error => {
|
||||
const normalized = error instanceof Error ? error : new Error(String(error));
|
||||
if (this.#options.onError) this.#options.onError(normalized);
|
||||
else console.error(`CultMesh provider-session expiry failed: ${normalized.message}`);
|
||||
});
|
||||
}, Math.max(10, options.expiryPollMs ?? 1_000));
|
||||
this.#expiryTimer.unref?.();
|
||||
}
|
||||
|
||||
public close(): void {
|
||||
clearInterval(this.#expiryTimer);
|
||||
}
|
||||
|
||||
public async handle(
|
||||
request: CultNetOperationRequestMessage,
|
||||
session: CultMeshRudpServerSession,
|
||||
): Promise<CultNetOperationResponseMessage> {
|
||||
const work = this.#work.then(() => this.#handle(request, session));
|
||||
this.#work = work.then(() => undefined, () => undefined);
|
||||
return await work;
|
||||
}
|
||||
|
||||
async #handle(
|
||||
request: CultNetOperationRequestMessage,
|
||||
session: CultMeshRudpServerSession,
|
||||
): Promise<CultNetOperationResponseMessage> {
|
||||
if (request.serviceId !== CULTMESH_PROVIDER_SESSION_SERVICE_ID) {
|
||||
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
`Unsupported service ${request.serviceId}.`,
|
||||
]);
|
||||
}
|
||||
if (request.payloadEncoding !== "messagepack-base64") {
|
||||
return this.#response(request, "invalid", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
"Provider-session operation payloads must use messagepack-base64.",
|
||||
]);
|
||||
}
|
||||
try {
|
||||
switch (request.operation) {
|
||||
case cultMeshProviderSessionOperations.register:
|
||||
return await this.#register(request, session);
|
||||
case cultMeshProviderSessionOperations.renew:
|
||||
return await this.#renew(request, session);
|
||||
case cultMeshProviderSessionOperations.publicationPut:
|
||||
return await this.#publicationPut(request, session);
|
||||
case cultMeshProviderSessionOperations.publicationDelete:
|
||||
return await this.#publicationDelete(request, session);
|
||||
case cultMeshProviderSessionOperations.receiptPut:
|
||||
return await this.#receiptPut(request, session);
|
||||
case cultMeshProviderSessionOperations.withdraw:
|
||||
return await this.#withdraw(request, session);
|
||||
default:
|
||||
return this.#response(request, "invalid", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
`Unsupported provider-session operation ${request.operation}.`,
|
||||
]);
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof BrokerInfrastructureError) throw error.cause;
|
||||
return this.#response(request, error instanceof BrokerStatusError ? error.status : "invalid", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
error instanceof Error ? error.message : String(error),
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
public sessionClosed(session: CultMeshRudpServerSession): void {
|
||||
for (const leaseId of this.#sessionLeases.get(session.sessionId) ?? []) {
|
||||
const record = this.#leases.get(leaseId);
|
||||
if (record?.session?.sessionId === session.sessionId) record.session = undefined;
|
||||
}
|
||||
this.#sessionLeases.delete(session.sessionId);
|
||||
}
|
||||
|
||||
public enqueueCommand(command: CultMeshProviderCommandWire): void {
|
||||
requireText(command.commandId, "commandId");
|
||||
requireText(command.providerId, "providerId");
|
||||
requireText(command.serviceInstanceId, "serviceInstanceId");
|
||||
const completed = this.#receipts.get(command.commandId);
|
||||
if (completed) {
|
||||
if (completed.commandKind !== command.commandKind || completed.providerId !== command.providerId || completed.serviceInstanceId !== command.serviceInstanceId) {
|
||||
throw new Error(`Command id ${command.commandId} is already completed for another command transaction.`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const existing = this.#commands.get(command.commandId);
|
||||
if (existing && !wireEquivalent(existing, command)) {
|
||||
throw new Error(`Command id ${command.commandId} is already retained for another command transaction.`);
|
||||
}
|
||||
this.#commands.set(command.commandId, existing ?? command);
|
||||
for (const record of this.#leases.values()) {
|
||||
if (!this.#isActive(record)) continue;
|
||||
if (record.identity.providerId !== command.providerId || record.identity.serviceInstanceId !== command.serviceInstanceId) continue;
|
||||
this.#sendCommand(record, command);
|
||||
}
|
||||
}
|
||||
|
||||
public get activeLeaseCount(): number {
|
||||
return [...this.#leases.values()].filter(record => this.#isActive(record)).length;
|
||||
}
|
||||
|
||||
public async expireLeases(): Promise<void> {
|
||||
const work = this.#work.then(() => this.#expireLeases());
|
||||
this.#work = work.then(() => undefined, () => undefined);
|
||||
await work;
|
||||
}
|
||||
|
||||
async #expireLeases(): Promise<void> {
|
||||
const now = this.#clock.now().getTime();
|
||||
for (const [leaseId, record] of [...this.#leases]) {
|
||||
if (Date.parse(record.lease.expiresAtUtc) > now) continue;
|
||||
await this.#options.deletePublications(record.identity, publicationEntries(record));
|
||||
this.#releasePublicationOwners(record);
|
||||
this.#removeLease(leaseId, record);
|
||||
}
|
||||
}
|
||||
|
||||
async #register(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise<CultNetOperationResponseMessage> {
|
||||
expectSchema(request, cultMeshProviderSessionSchemas.registration);
|
||||
const identity = decodeProviderSessionPayload<CultMeshProviderRegistrationWire>(request.payload);
|
||||
validateRegistration(identity);
|
||||
if (!await infrastructure(() => this.#options.authorizeRegistration(identity, session))) {
|
||||
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
"The CultNet session does not authorize this provider identity.",
|
||||
]);
|
||||
}
|
||||
const identityKey = identityTuple(identity);
|
||||
const previousLeaseId = this.#identityLeases.get(identityKey);
|
||||
let publications = new Map<string, CultNetRawDocumentRecord>();
|
||||
if (previousLeaseId) {
|
||||
const previous = this.#leases.get(previousLeaseId);
|
||||
if (previous) {
|
||||
publications = previous.publications;
|
||||
this.#removeLease(previousLeaseId, previous);
|
||||
}
|
||||
}
|
||||
const lease = this.#createLease(identity, identity.requestedLeaseDurationMs);
|
||||
const record: LeaseRecord = { identity, lease, session, publications };
|
||||
this.#leases.set(lease.leaseId, record);
|
||||
this.#identityLeases.set(identityKey, lease.leaseId);
|
||||
this.#attachSession(session, lease.leaseId);
|
||||
this.#replayCommands(record);
|
||||
return this.#response(request, "ok", cultMeshProviderSessionSchemas.lease, lease);
|
||||
}
|
||||
|
||||
async #renew(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise<CultNetOperationResponseMessage> {
|
||||
expectSchema(request, cultMeshProviderSessionSchemas.leaseRenewal);
|
||||
const renewal = decodeProviderSessionPayload<CultMeshProviderLeaseRenewalWire>(request.payload);
|
||||
const current = this.#requireLease(renewal.leaseId, session);
|
||||
const lease = this.#createLease(current.identity, renewal.requestedLeaseDurationMs);
|
||||
const replacement: LeaseRecord = { ...current, lease, session };
|
||||
this.#removeLease(renewal.leaseId, current);
|
||||
this.#leases.set(lease.leaseId, replacement);
|
||||
this.#identityLeases.set(identityTuple(current.identity), lease.leaseId);
|
||||
this.#attachSession(session, lease.leaseId);
|
||||
return this.#response(request, "ok", cultMeshProviderSessionSchemas.lease, lease);
|
||||
}
|
||||
|
||||
async #publicationPut(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise<CultNetOperationResponseMessage> {
|
||||
expectSchema(request, cultMeshProviderSessionSchemas.publicationPut);
|
||||
const mutation = decodeProviderSessionPayload<CultMeshProviderPublicationPutWire>(request.payload);
|
||||
const record = this.#requireLease(mutation.leaseId, session);
|
||||
requireText(mutation.publicationId, "publicationId");
|
||||
validateDocument(mutation.document);
|
||||
const previous = record.publications.get(mutation.publicationId);
|
||||
if (previous && (previous.schemaId !== mutation.document.schemaId || previous.recordKey !== mutation.document.recordKey)) {
|
||||
return this.#response(request, "conflict", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
`Publication ${mutation.publicationId} already owns ${previous.schemaId}:${previous.recordKey}. Delete it before changing its tuple.`,
|
||||
]);
|
||||
}
|
||||
const tuple = documentTuple(mutation.document);
|
||||
const owner = this.#publicationOwners.get(tuple);
|
||||
const ownerIdentity = identityTuple(record.identity);
|
||||
if (owner && (owner.identityKey !== ownerIdentity || owner.publicationId !== mutation.publicationId)) {
|
||||
return this.#response(request, "conflict", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
`Publication tuple ${mutation.document.schemaId}:${mutation.document.recordKey} is already owned.`,
|
||||
]);
|
||||
}
|
||||
const normalizedDocument: CultNetRawDocumentRecord = {
|
||||
...mutation.document,
|
||||
sourceRuntimeId: record.identity.serviceInstanceId,
|
||||
sourceAgentId: record.identity.providerId,
|
||||
sourceRole: "provider",
|
||||
};
|
||||
await infrastructure(() => this.#options.acceptPublication(record.identity, mutation.publicationId, normalizedDocument));
|
||||
record.publications.set(mutation.publicationId, normalizedDocument);
|
||||
this.#publicationOwners.set(tuple, { identityKey: ownerIdentity, publicationId: mutation.publicationId });
|
||||
return this.#accepted(request, record.lease.leaseId, { publicationId: mutation.publicationId });
|
||||
}
|
||||
|
||||
async #publicationDelete(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise<CultNetOperationResponseMessage> {
|
||||
expectSchema(request, cultMeshProviderSessionSchemas.publicationDelete);
|
||||
const mutation = decodeProviderSessionPayload<CultMeshProviderPublicationDeleteWire>(request.payload);
|
||||
const record = this.#requireLease(mutation.leaseId, session);
|
||||
const owned = record.publications.get(mutation.publicationId);
|
||||
if (!owned || owned.schemaId !== mutation.schemaId || owned.recordKey !== mutation.recordKey) {
|
||||
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
`Publication ${mutation.publicationId} is not owned by this lease with the supplied schema and key.`,
|
||||
]);
|
||||
}
|
||||
await infrastructure(() => this.#options.deletePublications(record.identity, [{ publicationId: mutation.publicationId, document: owned }]));
|
||||
record.publications.delete(mutation.publicationId);
|
||||
this.#publicationOwners.delete(documentTuple(owned));
|
||||
return this.#accepted(request, record.lease.leaseId, { publicationId: mutation.publicationId });
|
||||
}
|
||||
|
||||
async #receiptPut(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise<CultNetOperationResponseMessage> {
|
||||
expectSchema(request, cultMeshProviderSessionSchemas.receiptPut);
|
||||
const mutation = decodeProviderSessionPayload<CultMeshProviderReceiptPutWire>(request.payload);
|
||||
const record = this.#requireLease(mutation.leaseId, session);
|
||||
const receipt = mutation.receipt;
|
||||
validateReceipt(receipt);
|
||||
if (receipt.providerId !== record.identity.providerId || receipt.serviceInstanceId !== record.identity.serviceInstanceId) {
|
||||
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, ["Receipt identity does not match its lease."]);
|
||||
}
|
||||
const previousReceipt = this.#receipts.get(receipt.commandId);
|
||||
if (previousReceipt) {
|
||||
if (!wireEquivalent(previousReceipt, receipt)) {
|
||||
return this.#response(request, "conflict", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
`Command ${receipt.commandId} already has a different accepted receipt.`,
|
||||
]);
|
||||
}
|
||||
return this.#accepted(request, record.lease.leaseId, { commandId: receipt.commandId, receiptId: receipt.receiptId });
|
||||
}
|
||||
const command = this.#commands.get(receipt.commandId);
|
||||
if (!command || command.commandKind !== receipt.commandKind || command.providerId !== receipt.providerId || command.serviceInstanceId !== receipt.serviceInstanceId) {
|
||||
return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [
|
||||
`Receipt ${receipt.receiptId} does not match a retained command for this provider generation.`,
|
||||
]);
|
||||
}
|
||||
await infrastructure(() => this.#options.acceptReceipt(record.identity, receipt));
|
||||
this.#receipts.set(receipt.commandId, receipt);
|
||||
this.#commands.delete(receipt.commandId);
|
||||
return this.#accepted(request, record.lease.leaseId, { commandId: receipt.commandId, receiptId: receipt.receiptId });
|
||||
}
|
||||
|
||||
async #withdraw(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise<CultNetOperationResponseMessage> {
|
||||
expectSchema(request, cultMeshProviderSessionSchemas.withdrawal);
|
||||
const mutation = decodeProviderSessionPayload<CultMeshProviderWithdrawalWire>(request.payload);
|
||||
const record = this.#requireLease(mutation.leaseId, session);
|
||||
await infrastructure(() => this.#options.deletePublications(record.identity, publicationEntries(record)));
|
||||
this.#releasePublicationOwners(record);
|
||||
this.#removeLease(mutation.leaseId, record);
|
||||
return this.#accepted(request, mutation.leaseId);
|
||||
}
|
||||
|
||||
#requireLease(leaseId: string, session: CultMeshRudpServerSession): LeaseRecord {
|
||||
requireText(leaseId, "leaseId");
|
||||
const record = this.#leases.get(leaseId);
|
||||
if (!record || !this.#isActive(record)) throw new BrokerStatusError("expired", `Provider lease ${leaseId} is expired or fenced.`);
|
||||
if (record.session?.sessionId !== session.sessionId) throw new BrokerStatusError("denied", `Provider lease ${leaseId} belongs to another physical session.`);
|
||||
return record;
|
||||
}
|
||||
|
||||
#isActive(record: LeaseRecord): boolean {
|
||||
return Date.parse(record.lease.expiresAtUtc) > this.#clock.now().getTime();
|
||||
}
|
||||
|
||||
#createLease(identity: CultMeshProviderRegistrationWire, requestedMs: number): CultMeshProviderLeaseWire {
|
||||
if (!Number.isFinite(requestedMs) || requestedMs <= 0) throw new Error("requestedLeaseDurationMs must be positive.");
|
||||
const durationMs = Math.min(requestedMs, this.#options.maximumLeaseDurationMs ?? 120_000);
|
||||
const now = this.#clock.now();
|
||||
return {
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
endpointId: identity.endpointId,
|
||||
verseId: identity.verseId,
|
||||
leaseId: this.#options.createLeaseId?.() ?? `${this.#options.runtimeId}:lease:${++this.#leaseSequence}`,
|
||||
validFromUtc: now.toISOString(),
|
||||
expiresAtUtc: new Date(now.getTime() + durationMs).toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
#attachSession(session: CultMeshRudpServerSession, leaseId: string): void {
|
||||
let leases = this.#sessionLeases.get(session.sessionId);
|
||||
if (!leases) this.#sessionLeases.set(session.sessionId, leases = new Set());
|
||||
leases.add(leaseId);
|
||||
}
|
||||
|
||||
#removeLease(leaseId: string, record: LeaseRecord): void {
|
||||
this.#leases.delete(leaseId);
|
||||
if (this.#identityLeases.get(identityTuple(record.identity)) === leaseId) this.#identityLeases.delete(identityTuple(record.identity));
|
||||
this.#sessionLeases.get(record.session?.sessionId ?? "")?.delete(leaseId);
|
||||
}
|
||||
|
||||
#releasePublicationOwners(record: LeaseRecord): void {
|
||||
const ownerIdentity = identityTuple(record.identity);
|
||||
for (const [publicationId, document] of record.publications) {
|
||||
const tuple = documentTuple(document);
|
||||
const owner = this.#publicationOwners.get(tuple);
|
||||
if (owner?.identityKey === ownerIdentity && owner.publicationId === publicationId) this.#publicationOwners.delete(tuple);
|
||||
}
|
||||
}
|
||||
|
||||
#replayCommands(record: LeaseRecord): void {
|
||||
for (const command of this.#commands.values()) {
|
||||
if (command.providerId === record.identity.providerId && command.serviceInstanceId === record.identity.serviceInstanceId) this.#sendCommand(record, command);
|
||||
}
|
||||
}
|
||||
|
||||
#sendCommand(record: LeaseRecord, command: CultMeshProviderCommandWire): void {
|
||||
if (!record.session) return;
|
||||
const message: CultNetDocumentPutRawMessage = {
|
||||
schemaVersion: "cultnet.document_put_raw.v0",
|
||||
messageId: `${this.#options.runtimeId}:command:${command.commandId}`,
|
||||
document: {
|
||||
schemaId: cultMeshProviderSessionSchemas.command,
|
||||
recordKey: `${command.providerId}/${command.serviceInstanceId}/${command.commandId}`,
|
||||
storedAt: this.#clock.now().toISOString(),
|
||||
payloadEncoding: "messagepack",
|
||||
payload: encode(command),
|
||||
sourceRuntimeId: this.#options.runtimeId,
|
||||
sourceRole: "provider-session-broker",
|
||||
},
|
||||
};
|
||||
record.session.send(message);
|
||||
}
|
||||
|
||||
#accepted(request: CultNetOperationRequestMessage, leaseId: string, extra: Record<string, string> = {}): CultNetOperationResponseMessage {
|
||||
return this.#response(request, "ok", cultMeshProviderSessionSchemas.mutationAcceptance, {
|
||||
acceptedAtUtc: this.#clock.now().toISOString(),
|
||||
leaseId,
|
||||
...extra,
|
||||
});
|
||||
}
|
||||
|
||||
#response(
|
||||
request: CultNetOperationRequestMessage,
|
||||
status: CultMeshProviderOperationStatus,
|
||||
payloadSchema: string,
|
||||
payload: unknown,
|
||||
diagnostics?: string[],
|
||||
): CultNetOperationResponseMessage {
|
||||
const response: CultNetOperationResponseMessage = {
|
||||
schemaVersion: "cultnet.operation_response.v0",
|
||||
messageId: request.messageId,
|
||||
serviceId: request.serviceId,
|
||||
operation: request.operation,
|
||||
status,
|
||||
payloadSchema,
|
||||
payloadEncoding: "messagepack-base64",
|
||||
payload: encodeProviderSessionPayload(payload),
|
||||
sourceRuntimeId: this.#options.runtimeId,
|
||||
};
|
||||
if (diagnostics) response.diagnostics = diagnostics;
|
||||
return response;
|
||||
}
|
||||
}
|
||||
|
||||
class BrokerStatusError extends Error {
|
||||
public constructor(public readonly status: CultMeshProviderOperationStatus, message: string) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
|
||||
class BrokerInfrastructureError extends Error {
|
||||
public constructor(public readonly cause: unknown) {
|
||||
super(cause instanceof Error ? cause.message : String(cause));
|
||||
}
|
||||
}
|
||||
|
||||
async function infrastructure<T>(action: () => T | Promise<T>): Promise<T> {
|
||||
try {
|
||||
return await action();
|
||||
} catch (error) {
|
||||
throw new BrokerInfrastructureError(error);
|
||||
}
|
||||
}
|
||||
|
||||
function expectSchema(request: CultNetOperationRequestMessage, schema: string): void {
|
||||
if (request.payloadSchema !== schema) throw new Error(`Operation ${request.operation} requires payload schema ${schema}.`);
|
||||
}
|
||||
|
||||
function validateRegistration(value: CultMeshProviderRegistrationWire): void {
|
||||
for (const field of ["providerId", "serviceInstanceId", "endpointId", "verseId"] as const) requireText(value[field], field);
|
||||
if (!Number.isFinite(value.requestedLeaseDurationMs) || value.requestedLeaseDurationMs <= 0) throw new Error("requestedLeaseDurationMs must be positive.");
|
||||
}
|
||||
|
||||
function validateDocument(document: CultNetRawDocumentRecord): void {
|
||||
requireText(document?.schemaId, "document.schemaId");
|
||||
requireText(document?.recordKey, "document.recordKey");
|
||||
if (document.payloadEncoding !== "messagepack" || !(document.payload instanceof Uint8Array)) throw new Error("Provider publications must carry a raw MessagePack document.");
|
||||
}
|
||||
|
||||
function validateReceipt(receipt: CultMeshProviderCommandReceiptWire): void {
|
||||
requireText(receipt?.receiptId, "receipt.receiptId");
|
||||
requireText(receipt?.commandId, "receipt.commandId");
|
||||
requireText(receipt?.commandKind, "receipt.commandKind");
|
||||
requireText(receipt?.providerId, "receipt.providerId");
|
||||
requireText(receipt?.serviceInstanceId, "receipt.serviceInstanceId");
|
||||
if (receipt.state !== "applied" && receipt.state !== "rejected" && receipt.state !== "failed") throw new Error("CultMesh provider-session receipt.state is invalid.");
|
||||
if (!Number.isFinite(Date.parse(receipt.completedAtUtc))) throw new Error("CultMesh provider-session receipt.completedAtUtc must be RFC3339.");
|
||||
}
|
||||
|
||||
function identityTuple(identity: CultMeshProviderRegistrationWire): string {
|
||||
return `${identity.verseId}\u0000${identity.providerId}\u0000${identity.serviceInstanceId}\u0000${identity.endpointId}`;
|
||||
}
|
||||
|
||||
function documentTuple(document: Pick<CultNetRawDocumentRecord, "schemaId" | "recordKey">): string {
|
||||
return `${document.schemaId}\u0000${document.recordKey}`;
|
||||
}
|
||||
|
||||
function wireEquivalent(left: unknown, right: unknown): boolean {
|
||||
return Buffer.from(encode(left)).equals(Buffer.from(encode(right)));
|
||||
}
|
||||
|
||||
function publicationEntries(record: LeaseRecord): readonly {
|
||||
publicationId: string;
|
||||
document: Pick<CultNetRawDocumentRecord, "schemaId" | "recordKey">;
|
||||
}[] {
|
||||
return [...record.publications].map(([publicationId, document]) => ({ publicationId, document }));
|
||||
}
|
||||
|
||||
function requireText(value: unknown, name: string): asserts value is string {
|
||||
if (typeof value !== "string" || value.trim().length === 0) throw new Error(`CultMesh provider-session ${name} must be non-empty.`);
|
||||
}
|
||||
170
packages/cultmesh-ts/src/provider-session-wire.ts
Normal file
170
packages/cultmesh-ts/src/provider-session-wire.ts
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
import { decode, encode } from "@msgpack/msgpack";
|
||||
import type { CultNetRawDocumentRecord } from "cultnet-ts";
|
||||
|
||||
export const CULTMESH_PROVIDER_SESSION_SERVICE_ID = "gamecult.mesh.provider_session";
|
||||
|
||||
export const cultMeshProviderSessionOperations = {
|
||||
register: "provider.register",
|
||||
renew: "provider.renew",
|
||||
publicationPut: "provider.publication.put",
|
||||
publicationDelete: "provider.publication.delete",
|
||||
receiptPut: "provider.receipt.put",
|
||||
withdraw: "provider.withdraw",
|
||||
} as const;
|
||||
|
||||
export const cultMeshProviderSessionSchemas = {
|
||||
registration: "gamecult.mesh.provider_registration.v1",
|
||||
lease: "gamecult.mesh.provider_lease.v1",
|
||||
leaseRenewal: "gamecult.mesh.provider_lease_renewal.v1",
|
||||
publicationPut: "gamecult.mesh.provider_publication_put.v1",
|
||||
publicationDelete: "gamecult.mesh.provider_publication_delete.v1",
|
||||
command: "gamecult.mesh.provider_command.v1",
|
||||
receiptPut: "gamecult.mesh.provider_receipt_put.v1",
|
||||
withdrawal: "gamecult.mesh.provider_withdrawal.v1",
|
||||
mutationAcceptance: "gamecult.mesh.provider_mutation_acceptance.v1",
|
||||
} as const;
|
||||
|
||||
export type CultMeshProviderOperationStatus =
|
||||
| "ok"
|
||||
| "conflict"
|
||||
| "expired"
|
||||
| "denied"
|
||||
| "invalid";
|
||||
|
||||
export interface CultMeshProviderRegistrationWire {
|
||||
providerId: string;
|
||||
serviceInstanceId: string;
|
||||
endpointId: string;
|
||||
verseId: string;
|
||||
requestedLeaseDurationMs: number;
|
||||
authorityLeaseId?: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderLeaseWire {
|
||||
providerId: string;
|
||||
serviceInstanceId: string;
|
||||
endpointId: string;
|
||||
verseId: string;
|
||||
leaseId: string;
|
||||
validFromUtc: string;
|
||||
expiresAtUtc: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderLeaseRenewalWire {
|
||||
leaseId: string;
|
||||
requestedLeaseDurationMs: number;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderPublicationPutWire {
|
||||
leaseId: string;
|
||||
publicationId: string;
|
||||
document: CultNetRawDocumentRecord;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderPublicationDeleteWire {
|
||||
leaseId: string;
|
||||
publicationId: string;
|
||||
schemaId: string;
|
||||
recordKey: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderCommandWire {
|
||||
commandId: string;
|
||||
commandKind: string;
|
||||
providerId: string;
|
||||
serviceInstanceId: string;
|
||||
payload: unknown;
|
||||
}
|
||||
|
||||
export type CultMeshProviderReceiptStateWire = "applied" | "rejected" | "failed";
|
||||
|
||||
export interface CultMeshProviderCommandReceiptWire {
|
||||
receiptId: string;
|
||||
commandId: string;
|
||||
commandKind: string;
|
||||
providerId: string;
|
||||
serviceInstanceId: string;
|
||||
state: CultMeshProviderReceiptStateWire;
|
||||
completedAtUtc: string;
|
||||
result?: unknown;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderReceiptPutWire {
|
||||
leaseId: string;
|
||||
receipt: CultMeshProviderCommandReceiptWire;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderWithdrawalWire {
|
||||
leaseId: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderMutationAcceptanceWire {
|
||||
acceptedAtUtc: string;
|
||||
leaseId?: string;
|
||||
publicationId?: string;
|
||||
commandId?: string;
|
||||
receiptId?: string;
|
||||
}
|
||||
|
||||
export function encodeProviderSessionPayload(value: unknown): string {
|
||||
return Buffer.from(encode(value)).toString("base64");
|
||||
}
|
||||
|
||||
export function decodeProviderSessionPayload<T>(payload: string): T {
|
||||
if (typeof payload !== "string" || payload.length === 0) {
|
||||
throw new Error("CultMesh provider-session payload must be non-empty MessagePack base64.");
|
||||
}
|
||||
return decode(Buffer.from(payload, "base64")) as T;
|
||||
}
|
||||
|
||||
export function decodeProviderCommandDocument(document: CultNetRawDocumentRecord): CultMeshProviderCommandWire {
|
||||
if (document.schemaId !== cultMeshProviderSessionSchemas.command) {
|
||||
throw new Error(`Expected ${cultMeshProviderSessionSchemas.command}, received ${document.schemaId}.`);
|
||||
}
|
||||
if (document.payloadEncoding !== "messagepack") {
|
||||
throw new Error(`Provider command ${document.recordKey} must use MessagePack.`);
|
||||
}
|
||||
const command = decode(document.payload) as Partial<CultMeshProviderCommandWire>;
|
||||
requireText(command.commandId, "commandId");
|
||||
requireText(command.commandKind, "commandKind");
|
||||
requireText(command.providerId, "providerId");
|
||||
requireText(command.serviceInstanceId, "serviceInstanceId");
|
||||
return command as CultMeshProviderCommandWire;
|
||||
}
|
||||
|
||||
export function assertProviderLease(value: unknown): asserts value is CultMeshProviderLeaseWire {
|
||||
const lease = requireObject(value, "provider lease");
|
||||
requireText(lease.providerId, "providerId");
|
||||
requireText(lease.serviceInstanceId, "serviceInstanceId");
|
||||
requireText(lease.endpointId, "endpointId");
|
||||
requireText(lease.verseId, "verseId");
|
||||
requireText(lease.leaseId, "leaseId");
|
||||
requireTimestamp(lease.validFromUtc, "validFromUtc");
|
||||
requireTimestamp(lease.expiresAtUtc, "expiresAtUtc");
|
||||
}
|
||||
|
||||
export function assertMutationAcceptance(value: unknown): asserts value is CultMeshProviderMutationAcceptanceWire {
|
||||
const acceptance = requireObject(value, "provider mutation acceptance");
|
||||
requireTimestamp(acceptance.acceptedAtUtc, "acceptedAtUtc");
|
||||
}
|
||||
|
||||
function requireObject(value: unknown, name: string): Record<string, unknown> {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) {
|
||||
throw new Error(`CultMesh ${name} must be a MessagePack map.`);
|
||||
}
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function requireText(value: unknown, name: string): asserts value is string {
|
||||
if (typeof value !== "string" || value.trim().length === 0) {
|
||||
throw new Error(`CultMesh provider-session ${name} must be non-empty.`);
|
||||
}
|
||||
}
|
||||
|
||||
function requireTimestamp(value: unknown, name: string): asserts value is string {
|
||||
requireText(value, name);
|
||||
if (!Number.isFinite(Date.parse(value))) {
|
||||
throw new Error(`CultMesh provider-session ${name} must be an RFC3339 timestamp.`);
|
||||
}
|
||||
}
|
||||
161
packages/cultmesh-ts/test/provider-rudp-transport.test.ts
Normal file
161
packages/cultmesh-ts/test/provider-rudp-transport.test.ts
Normal file
|
|
@ -0,0 +1,161 @@
|
|||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { decode } from "@msgpack/msgpack";
|
||||
import { CultNetDocumentRegistry } from "cultnet-ts";
|
||||
|
||||
import {
|
||||
CultMesh,
|
||||
CultMeshProviderRudpTransport,
|
||||
CultMeshProviderSessionBroker,
|
||||
type CultMeshProviderCommand,
|
||||
type CultMeshProviderIdentity,
|
||||
type CultMeshProviderPublication,
|
||||
} from "../src";
|
||||
|
||||
const identity: CultMeshProviderIdentity = {
|
||||
providerId: "voidbot.swarm",
|
||||
serviceInstanceId: "voidbot-worker-7",
|
||||
endpointId: "voidbot-worker-7.rudp",
|
||||
verseId: "voidbot.local",
|
||||
};
|
||||
|
||||
test("RUDP provider transport requires broker acceptance for the full lifecycle", async () => {
|
||||
const accepted = new Map<string, unknown>();
|
||||
const acceptedProvenance = new Map<string, { runtime?: string; agent?: string }>();
|
||||
const deleted: string[] = [];
|
||||
const receipts: string[] = [];
|
||||
let leaseSequence = 0;
|
||||
const broker = new CultMeshProviderSessionBroker({
|
||||
runtimeId: "odin-test",
|
||||
createLeaseId: () => `lease-${++leaseSequence}`,
|
||||
authorizeRegistration: () => true,
|
||||
acceptPublication: (_identity, publicationId, document) => {
|
||||
accepted.set(publicationId, decode(document.payload));
|
||||
acceptedProvenance.set(publicationId, {
|
||||
runtime: document.sourceRuntimeId,
|
||||
agent: document.sourceAgentId,
|
||||
});
|
||||
},
|
||||
deletePublications: (_identity, publications) => {
|
||||
for (const { publicationId } of publications) {
|
||||
deleted.push(publicationId);
|
||||
accepted.delete(publicationId);
|
||||
}
|
||||
},
|
||||
acceptReceipt: (_identity, receipt) => {
|
||||
receipts.push(receipt.receiptId);
|
||||
},
|
||||
});
|
||||
const server = CultMesh.createRudpDocumentServer("odin-test", 0x43554c54, {
|
||||
bindPort: 0,
|
||||
documents: new CultNetDocumentRegistry(),
|
||||
onOperationRequest: (request, session) => broker.handle(request, session),
|
||||
onSessionClosed: session => broker.sessionClosed(session),
|
||||
});
|
||||
await server.start();
|
||||
|
||||
const transport = new CultMeshProviderRudpTransport({
|
||||
endpoint: `rudp://127.0.0.1:${server.bind.port}`,
|
||||
runtimeId: identity.serviceInstanceId,
|
||||
connectionId: 0x43554c54,
|
||||
});
|
||||
const connection = await transport.connect(identity, new AbortController().signal);
|
||||
const secondIdentity: CultMeshProviderIdentity = {
|
||||
providerId: "aetheria",
|
||||
serviceInstanceId: "aetheria-daemon-2",
|
||||
endpointId: "aetheria-daemon-2.rudp",
|
||||
verseId: "aetheria.local",
|
||||
};
|
||||
const secondConnection = await new CultMeshProviderRudpTransport({
|
||||
endpoint: `rudp://127.0.0.1:${server.bind.port}`,
|
||||
runtimeId: secondIdentity.serviceInstanceId,
|
||||
connectionId: 0x43554c54,
|
||||
}).connect(secondIdentity, new AbortController().signal);
|
||||
try {
|
||||
broker.enqueueCommand({
|
||||
commandId: "refresh-1",
|
||||
commandKind: "voidbot.refresh",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
payload: { force: true },
|
||||
});
|
||||
const lease = await connection.register({ identity, requestedLeaseDurationMs: 10_000 }, new AbortController().signal);
|
||||
assert.equal(lease.leaseId, "lease-1");
|
||||
const secondLease = await secondConnection.register({
|
||||
identity: secondIdentity,
|
||||
requestedLeaseDurationMs: 10_000,
|
||||
}, new AbortController().signal);
|
||||
assert.equal(secondLease.leaseId, "lease-2");
|
||||
assert.equal(broker.activeLeaseCount, 2, "the broker demultiplexes simultaneous physical provider sessions");
|
||||
|
||||
const publication: CultMeshProviderPublication = {
|
||||
publicationId: "voidbot.surface",
|
||||
documentType: "gamecult.eve.surface_state",
|
||||
schemaId: "gamecult.eve.surface_state.v1",
|
||||
recordKey: "voidbot.swarm",
|
||||
value: { generation: 3 },
|
||||
};
|
||||
await connection.publish(publication, lease);
|
||||
assert.deepEqual(accepted.get(publication.publicationId), publication.value);
|
||||
assert.deepEqual(acceptedProvenance.get(publication.publicationId), {
|
||||
runtime: identity.serviceInstanceId,
|
||||
agent: identity.providerId,
|
||||
}, "the broker derives provenance from the fenced identity");
|
||||
const secondPublication: CultMeshProviderPublication = {
|
||||
publicationId: "aetheria.surface",
|
||||
documentType: "gamecult.eve.surface_state",
|
||||
schemaId: "gamecult.eve.surface_state.v1",
|
||||
recordKey: "aetheria.pilot",
|
||||
value: { generation: 8 },
|
||||
};
|
||||
await secondConnection.publish(secondPublication, secondLease);
|
||||
await assert.rejects(secondConnection.publish({
|
||||
...secondPublication,
|
||||
publicationId: "aetheria.stolen-tuple",
|
||||
recordKey: publication.recordKey,
|
||||
}, secondLease), /already owned/);
|
||||
await assert.rejects(
|
||||
secondConnection.publish({ ...secondPublication, value: { generation: 9 } }, lease),
|
||||
/belongs to another physical session/,
|
||||
);
|
||||
|
||||
const commandArrived = deferred<CultMeshProviderCommand>();
|
||||
connection.watchCommands(command => commandArrived.resolve(command));
|
||||
const command = await commandArrived.promise;
|
||||
assert.equal(command.commandId, "refresh-1");
|
||||
await connection.publishReceipt({
|
||||
receiptId: "receipt-refresh-1",
|
||||
commandId: command.commandId,
|
||||
commandKind: command.commandKind,
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
state: "applied",
|
||||
completedAt: new Date(),
|
||||
result: { refreshed: true },
|
||||
}, lease);
|
||||
assert.deepEqual(receipts, ["receipt-refresh-1"]);
|
||||
|
||||
const renewed = await connection.renew(lease);
|
||||
assert.equal(renewed.leaseId, "lease-3");
|
||||
await assert.rejects(connection.withdrawPublication(publication.publicationId, lease), /expired|fenced/);
|
||||
await connection.withdraw({ identity, leaseId: renewed.leaseId, publicationIds: [publication.publicationId] });
|
||||
await secondConnection.withdraw({
|
||||
identity: secondIdentity,
|
||||
leaseId: secondLease.leaseId,
|
||||
publicationIds: [secondPublication.publicationId],
|
||||
});
|
||||
assert.deepEqual(deleted, [publication.publicationId, secondPublication.publicationId]);
|
||||
assert.equal(accepted.size, 0);
|
||||
assert.equal(broker.activeLeaseCount, 0);
|
||||
} finally {
|
||||
connection.close();
|
||||
secondConnection.close();
|
||||
server.close();
|
||||
broker.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 };
|
||||
}
|
||||
17
packages/cultmesh-ts/test/provider-session-wire.test.ts
Normal file
17
packages/cultmesh-ts/test/provider-session-wire.test.ts
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
import { encodeProviderSessionPayload } from "../src/provider-session-wire";
|
||||
|
||||
const REGISTRATION_GOLDEN = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03";
|
||||
|
||||
test("provider registration matches the shared C# MessagePack golden payload", () => {
|
||||
assert.equal(encodeProviderSessionPayload({
|
||||
providerId: "aetheria",
|
||||
serviceInstanceId: "aetheria-42",
|
||||
endpointId: "aetheria-public",
|
||||
verseId: "public",
|
||||
requestedLeaseDurationMs: 30_000,
|
||||
authorityLeaseId: "authority-7",
|
||||
}), REGISTRATION_GOLDEN);
|
||||
});
|
||||
487
src/GameCult.Mesh/CultMeshProviderSessionWire.cs
Normal file
487
src/GameCult.Mesh/CultMeshProviderSessionWire.cs
Normal file
|
|
@ -0,0 +1,487 @@
|
|||
using System;
|
||||
using System.Globalization;
|
||||
using GameCult.Networking;
|
||||
using MessagePack;
|
||||
|
||||
namespace GameCult.Mesh
|
||||
{
|
||||
/// <summary>
|
||||
/// Canonical CultNet service, operation, payload-schema, and status identifiers for
|
||||
/// the CultMesh provider-session control plane.
|
||||
/// </summary>
|
||||
public static class CultMeshProviderSessionWireContract
|
||||
{
|
||||
/// <summary>The CultNet service id.</summary>
|
||||
public const string ServiceId = "gamecult.mesh.provider_session";
|
||||
|
||||
/// <summary>The registration operation.</summary>
|
||||
public const string RegisterOperation = "provider.register";
|
||||
/// <summary>The lease-renewal operation.</summary>
|
||||
public const string RenewOperation = "provider.renew";
|
||||
/// <summary>The publication-put operation.</summary>
|
||||
public const string PublicationPutOperation = "provider.publication.put";
|
||||
/// <summary>The publication-delete operation.</summary>
|
||||
public const string PublicationDeleteOperation = "provider.publication.delete";
|
||||
/// <summary>The command-receipt-put operation.</summary>
|
||||
public const string ReceiptPutOperation = "provider.receipt.put";
|
||||
/// <summary>The provider-withdrawal operation.</summary>
|
||||
public const string WithdrawOperation = "provider.withdraw";
|
||||
|
||||
/// <summary>The registration payload schema.</summary>
|
||||
public const string RegistrationSchema = "gamecult.mesh.provider_registration.v1";
|
||||
/// <summary>The provider-lease payload schema.</summary>
|
||||
public const string LeaseSchema = "gamecult.mesh.provider_lease.v1";
|
||||
/// <summary>The lease-renewal payload schema.</summary>
|
||||
public const string LeaseRenewalSchema = "gamecult.mesh.provider_lease_renewal.v1";
|
||||
/// <summary>The publication-put payload schema.</summary>
|
||||
public const string PublicationPutSchema = "gamecult.mesh.provider_publication_put.v1";
|
||||
/// <summary>The publication-delete payload schema.</summary>
|
||||
public const string PublicationDeleteSchema = "gamecult.mesh.provider_publication_delete.v1";
|
||||
/// <summary>The broker-to-provider command document schema.</summary>
|
||||
public const string CommandSchema = "gamecult.mesh.provider_command.v1";
|
||||
/// <summary>The command-receipt-put payload schema.</summary>
|
||||
public const string ReceiptPutSchema = "gamecult.mesh.provider_receipt_put.v1";
|
||||
/// <summary>The provider-withdrawal payload schema.</summary>
|
||||
public const string WithdrawalSchema = "gamecult.mesh.provider_withdrawal.v1";
|
||||
/// <summary>The accepted-mutation response payload schema.</summary>
|
||||
public const string MutationAcceptanceSchema = "gamecult.mesh.provider_mutation_acceptance.v1";
|
||||
|
||||
/// <summary>The successful application status.</summary>
|
||||
public const string OkStatus = "ok";
|
||||
/// <summary>The identity or generation conflict status.</summary>
|
||||
public const string ConflictStatus = "conflict";
|
||||
/// <summary>The expired-lease status.</summary>
|
||||
public const string ExpiredStatus = "expired";
|
||||
/// <summary>The authority-denied status.</summary>
|
||||
public const string DeniedStatus = "denied";
|
||||
/// <summary>The invalid-request status.</summary>
|
||||
public const string InvalidStatus = "invalid";
|
||||
/// <summary>The CultNet inner-payload encoding.</summary>
|
||||
public const string PayloadEncoding = "messagepack-base64";
|
||||
}
|
||||
|
||||
/// <summary>Requests a fenced provider lease from the provider-session broker.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderRegistrationWire
|
||||
{
|
||||
/// <summary>Gets or sets the stable provider identity.</summary>
|
||||
[Key("providerId")] public string ProviderId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the provider process generation.</summary>
|
||||
[Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the advertised endpoint identity.</summary>
|
||||
[Key("endpointId")] public string EndpointId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the Verse identity.</summary>
|
||||
[Key("verseId")] public string VerseId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the requested lease duration in milliseconds.</summary>
|
||||
[Key("requestedLeaseDurationMs")] public int RequestedLeaseDurationMs { get; set; }
|
||||
/// <summary>Gets or sets the optional authority lease authorizing registration.</summary>
|
||||
[Key("authorityLeaseId")] public string? AuthorityLeaseId { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>Identifies the provider generation authorized by a broker lease.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderLeaseWire
|
||||
{
|
||||
/// <summary>Gets or sets the stable provider identity.</summary>
|
||||
[Key("providerId")] public string ProviderId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the authorized provider process generation.</summary>
|
||||
[Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the authorized endpoint identity.</summary>
|
||||
[Key("endpointId")] public string EndpointId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the authorized Verse identity.</summary>
|
||||
[Key("verseId")] public string VerseId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the unique fencing lease id.</summary>
|
||||
[Key("leaseId")] public string LeaseId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the RFC3339 lease validity start.</summary>
|
||||
[Key("validFromUtc")] public string ValidFromUtc { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the RFC3339 lease expiry.</summary>
|
||||
[Key("expiresAtUtc")] public string ExpiresAtUtc { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>Requests replacement of an active provider lease.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderLeaseRenewalWire
|
||||
{
|
||||
/// <summary>Gets or sets the lease being replaced.</summary>
|
||||
[Key("leaseId")] public string LeaseId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the requested replacement duration in milliseconds.</summary>
|
||||
[Key("requestedLeaseDurationMs")] public int RequestedLeaseDurationMs { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>Publishes one typed document under an active provider lease.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderPublicationPutWire
|
||||
{
|
||||
/// <summary>Gets or sets the active provider lease.</summary>
|
||||
[Key("leaseId")] public string LeaseId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the provider-owned publication identity.</summary>
|
||||
[Key("publicationId")] public string PublicationId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the exact typed CultNet document.</summary>
|
||||
[Key("document")] public CultNetRawDocumentRecord Document { get; set; } = new CultNetRawDocumentRecord();
|
||||
}
|
||||
|
||||
/// <summary>Deletes one exact provider-owned publication tuple.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderPublicationDeleteWire
|
||||
{
|
||||
/// <summary>Gets or sets the active provider lease.</summary>
|
||||
[Key("leaseId")] public string LeaseId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the provider-owned publication identity.</summary>
|
||||
[Key("publicationId")] public string PublicationId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the exact published schema id.</summary>
|
||||
[Key("schemaId")] public string SchemaId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the exact published record key.</summary>
|
||||
[Key("recordKey")] public string RecordKey { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>A retained command routed from the broker to one provider generation.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderCommandWire
|
||||
{
|
||||
/// <summary>Gets or sets the stable command identity.</summary>
|
||||
[Key("commandId")] public string CommandId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the provider-defined command kind.</summary>
|
||||
[Key("commandKind")] public string CommandKind { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the target provider identity.</summary>
|
||||
[Key("providerId")] public string ProviderId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the target provider process generation.</summary>
|
||||
[Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the typed command payload value.</summary>
|
||||
[Key("payload")] public object? Payload { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>The exactly-once outcome of one provider command.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderCommandReceiptWire
|
||||
{
|
||||
/// <summary>Gets or sets the stable receipt identity.</summary>
|
||||
[Key("receiptId")] public string ReceiptId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the completed command identity.</summary>
|
||||
[Key("commandId")] public string CommandId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the completed command kind.</summary>
|
||||
[Key("commandKind")] public string CommandKind { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the provider identity.</summary>
|
||||
[Key("providerId")] public string ProviderId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the provider process generation.</summary>
|
||||
[Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets applied, rejected, or failed.</summary>
|
||||
[Key("state")] public string State { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the RFC3339 completion time.</summary>
|
||||
[Key("completedAtUtc")] public string CompletedAtUtc { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the optional typed result value.</summary>
|
||||
[Key("result")] public object? Result { get; set; }
|
||||
/// <summary>Gets or sets the optional bounded failure detail.</summary>
|
||||
[Key("error")] public string? Error { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>Submits a durable command receipt under an active provider lease.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderReceiptPutWire
|
||||
{
|
||||
/// <summary>Gets or sets the active provider lease.</summary>
|
||||
[Key("leaseId")] public string LeaseId { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the durable command receipt.</summary>
|
||||
[Key("receipt")] public CultMeshProviderCommandReceiptWire Receipt { get; set; } = new CultMeshProviderCommandReceiptWire();
|
||||
}
|
||||
|
||||
/// <summary>Withdraws the provider generation identified by a lease.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderWithdrawalWire
|
||||
{
|
||||
/// <summary>Gets or sets the lease whose owned publications must be withdrawn.</summary>
|
||||
[Key("leaseId")] public string LeaseId { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>Confirms broker acceptance of one provider-session mutation.</summary>
|
||||
[MessagePackObject]
|
||||
public sealed class CultMeshProviderMutationAcceptanceWire
|
||||
{
|
||||
/// <summary>Gets or sets the RFC3339 broker acceptance time.</summary>
|
||||
[Key("acceptedAtUtc")] public string AcceptedAtUtc { get; set; } = string.Empty;
|
||||
/// <summary>Gets or sets the accepted lease identity, when applicable.</summary>
|
||||
[Key("leaseId")] public string? LeaseId { get; set; }
|
||||
/// <summary>Gets or sets the accepted publication identity, when applicable.</summary>
|
||||
[Key("publicationId")] public string? PublicationId { get; set; }
|
||||
/// <summary>Gets or sets the accepted command identity, when applicable.</summary>
|
||||
[Key("commandId")] public string? CommandId { get; set; }
|
||||
/// <summary>Gets or sets the accepted receipt identity, when applicable.</summary>
|
||||
[Key("receiptId")] public string? ReceiptId { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Encodes provider-session payloads inside the existing CultNet operation envelope.
|
||||
/// Transport acknowledgement is deliberately not treated as application acceptance.
|
||||
/// </summary>
|
||||
public static class CultMeshProviderSessionWire
|
||||
{
|
||||
private static readonly MessagePackSerializerOptions Options =
|
||||
MessagePackSerializerOptions.Standard.WithSecurity(MessagePackSecurity.UntrustedData);
|
||||
|
||||
/// <summary>Encodes one validated provider-session value as MessagePack base64.</summary>
|
||||
public static string EncodePayload<T>(T value)
|
||||
{
|
||||
if (value == null) throw new ArgumentNullException(nameof(value));
|
||||
Validate(value);
|
||||
return Convert.ToBase64String(MessagePackSerializer.Serialize(value, Options));
|
||||
}
|
||||
|
||||
/// <summary>Decodes and validates one MessagePack-base64 provider-session value.</summary>
|
||||
public static T DecodePayload<T>(string payload)
|
||||
{
|
||||
RequireText(payload, nameof(payload));
|
||||
byte[] bytes;
|
||||
try
|
||||
{
|
||||
bytes = Convert.FromBase64String(payload);
|
||||
}
|
||||
catch (FormatException exception)
|
||||
{
|
||||
throw new ArgumentException("Provider-session payload must be MessagePack base64.", nameof(payload), exception);
|
||||
}
|
||||
|
||||
var value = MessagePackSerializer.Deserialize<T>(bytes, Options);
|
||||
if (value == null) throw new MessagePackSerializationException("Provider-session payload decoded to null.");
|
||||
Validate(value);
|
||||
return value;
|
||||
}
|
||||
|
||||
/// <summary>Decodes a retained provider command carried as a raw CultNet document.</summary>
|
||||
public static CultMeshProviderCommandWire DecodeCommandDocument(CultNetRawDocumentRecord document)
|
||||
{
|
||||
if (document == null) throw new ArgumentNullException(nameof(document));
|
||||
if (!string.Equals(document.SchemaId, CultMeshProviderSessionWireContract.CommandSchema, StringComparison.Ordinal))
|
||||
throw new ArgumentException($"Expected provider command schema '{CultMeshProviderSessionWireContract.CommandSchema}'.", nameof(document));
|
||||
if (!string.Equals(document.PayloadEncoding, "messagepack", StringComparison.Ordinal))
|
||||
throw new ArgumentException("Provider command document must use MessagePack.", nameof(document));
|
||||
var command = MessagePackSerializer.Deserialize<CultMeshProviderCommandWire>(document.Payload, Options);
|
||||
if (command == null) throw new MessagePackSerializationException("Provider command decoded to null.");
|
||||
Validate(command);
|
||||
return command;
|
||||
}
|
||||
|
||||
/// <summary>Creates a canonical provider-session CultNet operation request.</summary>
|
||||
public static CultNetOperationRequestMessage CreateRequest<T>(
|
||||
string messageId,
|
||||
string operation,
|
||||
string payloadSchema,
|
||||
T payload,
|
||||
string? sourceRuntimeId = null,
|
||||
string? targetRuntimeId = null)
|
||||
{
|
||||
RequireText(messageId, nameof(messageId));
|
||||
ValidateOperation(operation, payloadSchema);
|
||||
return new CultNetOperationRequestMessage
|
||||
{
|
||||
MessageId = messageId,
|
||||
ServiceId = CultMeshProviderSessionWireContract.ServiceId,
|
||||
Operation = operation,
|
||||
PayloadSchema = payloadSchema,
|
||||
PayloadEncoding = CultMeshProviderSessionWireContract.PayloadEncoding,
|
||||
Payload = EncodePayload(payload),
|
||||
SourceRuntimeId = sourceRuntimeId,
|
||||
TargetRuntimeId = targetRuntimeId
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>Validates an operation request envelope and decodes its typed payload.</summary>
|
||||
public static T DecodeRequest<T>(
|
||||
CultNetOperationRequestMessage request,
|
||||
string expectedOperation,
|
||||
string expectedPayloadSchema)
|
||||
{
|
||||
if (request == null) throw new ArgumentNullException(nameof(request));
|
||||
RequireEnvelope(request.ServiceId, request.Operation, request.PayloadSchema, request.PayloadEncoding,
|
||||
expectedOperation, expectedPayloadSchema);
|
||||
return DecodePayload<T>(request.Payload);
|
||||
}
|
||||
|
||||
/// <summary>Creates a correlated application-level operation response.</summary>
|
||||
public static CultNetOperationResponseMessage CreateResponse<T>(
|
||||
CultNetOperationRequestMessage request,
|
||||
string status,
|
||||
string payloadSchema,
|
||||
T payload,
|
||||
string? sourceRuntimeId = null)
|
||||
{
|
||||
if (request == null) throw new ArgumentNullException(nameof(request));
|
||||
ValidateStatus(status);
|
||||
return new CultNetOperationResponseMessage
|
||||
{
|
||||
MessageId = RequireText(request.MessageId, nameof(request.MessageId)),
|
||||
ServiceId = CultMeshProviderSessionWireContract.ServiceId,
|
||||
Operation = RequireText(request.Operation, nameof(request.Operation)),
|
||||
Status = status,
|
||||
PayloadSchema = RequireText(payloadSchema, nameof(payloadSchema)),
|
||||
PayloadEncoding = CultMeshProviderSessionWireContract.PayloadEncoding,
|
||||
Payload = EncodePayload(payload),
|
||||
SourceRuntimeId = sourceRuntimeId
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>Validates an operation response envelope and decodes its typed payload.</summary>
|
||||
public static T DecodeResponse<T>(
|
||||
CultNetOperationResponseMessage response,
|
||||
string expectedOperation,
|
||||
string expectedPayloadSchema)
|
||||
{
|
||||
if (response == null) throw new ArgumentNullException(nameof(response));
|
||||
ValidateStatus(response.Status);
|
||||
RequireEnvelope(response.ServiceId, response.Operation, response.PayloadSchema, response.PayloadEncoding,
|
||||
expectedOperation, expectedPayloadSchema);
|
||||
return DecodePayload<T>(response.Payload);
|
||||
}
|
||||
|
||||
/// <summary>Validates the invariants of a known provider-session wire value.</summary>
|
||||
public static void Validate<T>(T value)
|
||||
{
|
||||
switch (value)
|
||||
{
|
||||
case CultMeshProviderRegistrationWire registration:
|
||||
RequireIdentity(registration.ProviderId, registration.ServiceInstanceId, registration.EndpointId, registration.VerseId);
|
||||
RequirePositive(registration.RequestedLeaseDurationMs, nameof(registration.RequestedLeaseDurationMs));
|
||||
RequireOptionalText(registration.AuthorityLeaseId, nameof(registration.AuthorityLeaseId));
|
||||
break;
|
||||
case CultMeshProviderLeaseWire lease:
|
||||
RequireIdentity(lease.ProviderId, lease.ServiceInstanceId, lease.EndpointId, lease.VerseId);
|
||||
RequireText(lease.LeaseId, nameof(lease.LeaseId));
|
||||
var validFrom = RequireTimestamp(lease.ValidFromUtc, nameof(lease.ValidFromUtc));
|
||||
var expiresAt = RequireTimestamp(lease.ExpiresAtUtc, nameof(lease.ExpiresAtUtc));
|
||||
if (expiresAt <= validFrom) throw new ArgumentException("Provider lease must expire after validFromUtc.");
|
||||
break;
|
||||
case CultMeshProviderLeaseRenewalWire renewal:
|
||||
RequireText(renewal.LeaseId, nameof(renewal.LeaseId));
|
||||
RequirePositive(renewal.RequestedLeaseDurationMs, nameof(renewal.RequestedLeaseDurationMs));
|
||||
break;
|
||||
case CultMeshProviderPublicationPutWire publication:
|
||||
RequireText(publication.LeaseId, nameof(publication.LeaseId));
|
||||
RequireText(publication.PublicationId, nameof(publication.PublicationId));
|
||||
if (publication.Document == null) throw new ArgumentException("Provider publication document is required.");
|
||||
RequireText(publication.Document.SchemaId, nameof(publication.Document.SchemaId));
|
||||
RequireText(publication.Document.RecordKey, nameof(publication.Document.RecordKey));
|
||||
if (!string.Equals(publication.Document.PayloadEncoding, "messagepack", StringComparison.Ordinal))
|
||||
throw new ArgumentException("Provider publication document must use MessagePack.");
|
||||
break;
|
||||
case CultMeshProviderPublicationDeleteWire deletion:
|
||||
RequireText(deletion.LeaseId, nameof(deletion.LeaseId));
|
||||
RequireText(deletion.PublicationId, nameof(deletion.PublicationId));
|
||||
RequireText(deletion.SchemaId, nameof(deletion.SchemaId));
|
||||
RequireText(deletion.RecordKey, nameof(deletion.RecordKey));
|
||||
break;
|
||||
case CultMeshProviderCommandWire command:
|
||||
RequireText(command.CommandId, nameof(command.CommandId));
|
||||
RequireText(command.CommandKind, nameof(command.CommandKind));
|
||||
RequireText(command.ProviderId, nameof(command.ProviderId));
|
||||
RequireText(command.ServiceInstanceId, nameof(command.ServiceInstanceId));
|
||||
break;
|
||||
case CultMeshProviderCommandReceiptWire receipt:
|
||||
RequireText(receipt.ReceiptId, nameof(receipt.ReceiptId));
|
||||
RequireText(receipt.CommandId, nameof(receipt.CommandId));
|
||||
RequireText(receipt.CommandKind, nameof(receipt.CommandKind));
|
||||
RequireText(receipt.ProviderId, nameof(receipt.ProviderId));
|
||||
RequireText(receipt.ServiceInstanceId, nameof(receipt.ServiceInstanceId));
|
||||
ValidateReceiptState(receipt.State);
|
||||
RequireTimestamp(receipt.CompletedAtUtc, nameof(receipt.CompletedAtUtc));
|
||||
RequireOptionalText(receipt.Error, nameof(receipt.Error));
|
||||
break;
|
||||
case CultMeshProviderReceiptPutWire receiptPut:
|
||||
RequireText(receiptPut.LeaseId, nameof(receiptPut.LeaseId));
|
||||
if (receiptPut.Receipt == null) throw new ArgumentException("Provider receipt is required.");
|
||||
Validate(receiptPut.Receipt);
|
||||
break;
|
||||
case CultMeshProviderWithdrawalWire withdrawal:
|
||||
RequireText(withdrawal.LeaseId, nameof(withdrawal.LeaseId));
|
||||
break;
|
||||
case CultMeshProviderMutationAcceptanceWire acceptance:
|
||||
RequireTimestamp(acceptance.AcceptedAtUtc, nameof(acceptance.AcceptedAtUtc));
|
||||
RequireOptionalText(acceptance.LeaseId, nameof(acceptance.LeaseId));
|
||||
RequireOptionalText(acceptance.PublicationId, nameof(acceptance.PublicationId));
|
||||
RequireOptionalText(acceptance.CommandId, nameof(acceptance.CommandId));
|
||||
RequireOptionalText(acceptance.ReceiptId, nameof(acceptance.ReceiptId));
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private static void RequireEnvelope(
|
||||
string serviceId,
|
||||
string operation,
|
||||
string payloadSchema,
|
||||
string payloadEncoding,
|
||||
string expectedOperation,
|
||||
string expectedPayloadSchema)
|
||||
{
|
||||
if (!string.Equals(serviceId, CultMeshProviderSessionWireContract.ServiceId, StringComparison.Ordinal))
|
||||
throw new ArgumentException($"Expected provider-session service '{CultMeshProviderSessionWireContract.ServiceId}'.");
|
||||
if (!string.Equals(operation, expectedOperation, StringComparison.Ordinal))
|
||||
throw new ArgumentException($"Expected provider-session operation '{expectedOperation}'.");
|
||||
if (!string.Equals(payloadSchema, expectedPayloadSchema, StringComparison.Ordinal))
|
||||
throw new ArgumentException($"Expected provider-session payload schema '{expectedPayloadSchema}'.");
|
||||
if (!string.Equals(payloadEncoding, CultMeshProviderSessionWireContract.PayloadEncoding, StringComparison.Ordinal))
|
||||
throw new ArgumentException("Provider-session operation payload must use MessagePack base64.");
|
||||
}
|
||||
|
||||
private static void ValidateOperation(string operation, string payloadSchema)
|
||||
{
|
||||
var expectedSchema = operation switch
|
||||
{
|
||||
CultMeshProviderSessionWireContract.RegisterOperation => CultMeshProviderSessionWireContract.RegistrationSchema,
|
||||
CultMeshProviderSessionWireContract.RenewOperation => CultMeshProviderSessionWireContract.LeaseRenewalSchema,
|
||||
CultMeshProviderSessionWireContract.PublicationPutOperation => CultMeshProviderSessionWireContract.PublicationPutSchema,
|
||||
CultMeshProviderSessionWireContract.PublicationDeleteOperation => CultMeshProviderSessionWireContract.PublicationDeleteSchema,
|
||||
CultMeshProviderSessionWireContract.ReceiptPutOperation => CultMeshProviderSessionWireContract.ReceiptPutSchema,
|
||||
CultMeshProviderSessionWireContract.WithdrawOperation => CultMeshProviderSessionWireContract.WithdrawalSchema,
|
||||
_ => throw new ArgumentException($"Unsupported provider-session operation '{operation}'.", nameof(operation))
|
||||
};
|
||||
if (!string.Equals(payloadSchema, expectedSchema, StringComparison.Ordinal))
|
||||
throw new ArgumentException($"Operation '{operation}' requires payload schema '{expectedSchema}'.", nameof(payloadSchema));
|
||||
}
|
||||
|
||||
private static void ValidateStatus(string status)
|
||||
{
|
||||
if (status != CultMeshProviderSessionWireContract.OkStatus &&
|
||||
status != CultMeshProviderSessionWireContract.ConflictStatus &&
|
||||
status != CultMeshProviderSessionWireContract.ExpiredStatus &&
|
||||
status != CultMeshProviderSessionWireContract.DeniedStatus &&
|
||||
status != CultMeshProviderSessionWireContract.InvalidStatus)
|
||||
throw new ArgumentException($"Unsupported provider-session status '{status}'.", nameof(status));
|
||||
}
|
||||
|
||||
private static void ValidateReceiptState(string state)
|
||||
{
|
||||
if (state != "applied" && state != "rejected" && state != "failed")
|
||||
throw new ArgumentException($"Unsupported provider receipt state '{state}'.", nameof(state));
|
||||
}
|
||||
|
||||
private static void RequireIdentity(string providerId, string serviceInstanceId, string endpointId, string verseId)
|
||||
{
|
||||
RequireText(providerId, nameof(providerId));
|
||||
RequireText(serviceInstanceId, nameof(serviceInstanceId));
|
||||
RequireText(endpointId, nameof(endpointId));
|
||||
RequireText(verseId, nameof(verseId));
|
||||
}
|
||||
|
||||
private static string RequireText(string value, string name)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(value)) throw new ArgumentException("Value must be non-empty.", name);
|
||||
return value;
|
||||
}
|
||||
|
||||
private static void RequireOptionalText(string? value, string name)
|
||||
{
|
||||
if (value != null && string.IsNullOrWhiteSpace(value))
|
||||
throw new ArgumentException("Optional value must be null or non-empty.", name);
|
||||
}
|
||||
|
||||
private static void RequirePositive(int value, string name)
|
||||
{
|
||||
if (value <= 0) throw new ArgumentOutOfRangeException(name, "Value must be positive.");
|
||||
}
|
||||
|
||||
private static DateTimeOffset RequireTimestamp(string value, string name)
|
||||
{
|
||||
RequireText(value, name);
|
||||
if (!DateTimeOffset.TryParse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var timestamp))
|
||||
throw new ArgumentException("Value must be an RFC3339 timestamp.", name);
|
||||
return timestamp;
|
||||
}
|
||||
}
|
||||
}
|
||||
224
tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs
Normal file
224
tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs
Normal file
|
|
@ -0,0 +1,224 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using GameCult.Networking;
|
||||
using MessagePack;
|
||||
using NUnit.Framework;
|
||||
|
||||
namespace GameCult.Mesh.Tests;
|
||||
|
||||
[TestFixture]
|
||||
public sealed class CultMeshProviderSessionWireTests
|
||||
{
|
||||
private const string RegistrationGolden = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03";
|
||||
|
||||
[Test]
|
||||
public void Registration_RoundTrips_InsideCanonicalCultNetOperationEnvelope()
|
||||
{
|
||||
var registration = new CultMeshProviderRegistrationWire
|
||||
{
|
||||
ProviderId = "aetheria",
|
||||
ServiceInstanceId = "aetheria-42",
|
||||
EndpointId = "aetheria-public",
|
||||
VerseId = "public",
|
||||
RequestedLeaseDurationMs = 30_000,
|
||||
AuthorityLeaseId = "authority-7"
|
||||
};
|
||||
|
||||
var request = CultMeshProviderSessionWire.CreateRequest(
|
||||
"register-1",
|
||||
CultMeshProviderSessionWireContract.RegisterOperation,
|
||||
CultMeshProviderSessionWireContract.RegistrationSchema,
|
||||
registration,
|
||||
sourceRuntimeId: "aetheria-42",
|
||||
targetRuntimeId: "odin");
|
||||
var envelopeBytes = CultNetSchemaMessageSerialization.Serialize(request);
|
||||
var decodedEnvelope = (CultNetOperationRequestMessage)CultNetSchemaMessageSerialization.Deserialize(envelopeBytes);
|
||||
var decoded = CultMeshProviderSessionWire.DecodeRequest<CultMeshProviderRegistrationWire>(
|
||||
decodedEnvelope,
|
||||
CultMeshProviderSessionWireContract.RegisterOperation,
|
||||
CultMeshProviderSessionWireContract.RegistrationSchema);
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(CultMeshProviderSessionWire.EncodePayload(registration), Is.EqualTo(RegistrationGolden),
|
||||
"C# and TypeScript must emit the same canonical registration payload");
|
||||
Assert.That(decodedEnvelope.ServiceId, Is.EqualTo("gamecult.mesh.provider_session"));
|
||||
Assert.That(decodedEnvelope.PayloadEncoding, Is.EqualTo("messagepack-base64"));
|
||||
Assert.That(decoded.ProviderId, Is.EqualTo("aetheria"));
|
||||
Assert.That(decoded.ServiceInstanceId, Is.EqualTo("aetheria-42"));
|
||||
Assert.That(decoded.EndpointId, Is.EqualTo("aetheria-public"));
|
||||
Assert.That(decoded.VerseId, Is.EqualTo("public"));
|
||||
Assert.That(decoded.RequestedLeaseDurationMs, Is.EqualTo(30_000));
|
||||
Assert.That(decoded.AuthorityLeaseId, Is.EqualTo("authority-7"));
|
||||
});
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void PublicationPut_RoundTrips_ExactRawDocumentTuple()
|
||||
{
|
||||
var publication = new CultMeshProviderPublicationPutWire
|
||||
{
|
||||
LeaseId = "lease-1",
|
||||
PublicationId = "surface:pilot",
|
||||
Document = new CultNetRawDocumentRecord
|
||||
{
|
||||
SchemaId = "gamecult.eve.surface.v1",
|
||||
RecordKey = "eve:surface:pilot",
|
||||
StoredAt = "2026-07-14T12:00:00Z",
|
||||
PayloadEncoding = "messagepack",
|
||||
Payload = [0x81, 0xA2, 0x69, 0x64, 0x01],
|
||||
SourceRuntimeId = "aetheria-42"
|
||||
}
|
||||
};
|
||||
|
||||
var encoded = CultMeshProviderSessionWire.EncodePayload(publication);
|
||||
var decoded = CultMeshProviderSessionWire.DecodePayload<CultMeshProviderPublicationPutWire>(encoded);
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(decoded.LeaseId, Is.EqualTo("lease-1"));
|
||||
Assert.That(decoded.PublicationId, Is.EqualTo("surface:pilot"));
|
||||
Assert.That(decoded.Document.SchemaId, Is.EqualTo("gamecult.eve.surface.v1"));
|
||||
Assert.That(decoded.Document.RecordKey, Is.EqualTo("eve:surface:pilot"));
|
||||
Assert.That(decoded.Document.Payload, Is.EqualTo(publication.Document.Payload));
|
||||
});
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CommandAndReceipt_UseCamelCaseMessagePackMaps()
|
||||
{
|
||||
var command = new CultMeshProviderCommandWire
|
||||
{
|
||||
CommandId = "cmd-7",
|
||||
CommandKind = "pilot.thrust",
|
||||
ProviderId = "aetheria",
|
||||
ServiceInstanceId = "aetheria-42",
|
||||
Payload = new Dictionary<string, object> { ["axis"] = 0.75 }
|
||||
};
|
||||
var receiptPut = new CultMeshProviderReceiptPutWire
|
||||
{
|
||||
LeaseId = "lease-1",
|
||||
Receipt = new CultMeshProviderCommandReceiptWire
|
||||
{
|
||||
ReceiptId = "receipt-7",
|
||||
CommandId = command.CommandId,
|
||||
CommandKind = command.CommandKind,
|
||||
ProviderId = command.ProviderId,
|
||||
ServiceInstanceId = command.ServiceInstanceId,
|
||||
State = "applied",
|
||||
CompletedAtUtc = "2026-07-14T12:01:02.345Z",
|
||||
Result = new Dictionary<string, object> { ["accepted"] = true }
|
||||
}
|
||||
};
|
||||
|
||||
var encodedCommand = CultMeshProviderSessionWire.EncodePayload(command);
|
||||
var encodedReceipt = CultMeshProviderSessionWire.EncodePayload(receiptPut);
|
||||
var decodedCommand = CultMeshProviderSessionWire.DecodePayload<CultMeshProviderCommandWire>(encodedCommand);
|
||||
var decodedReceipt = CultMeshProviderSessionWire.DecodePayload<CultMeshProviderReceiptPutWire>(encodedReceipt);
|
||||
var decodedCommandDocument = CultMeshProviderSessionWire.DecodeCommandDocument(new CultNetRawDocumentRecord
|
||||
{
|
||||
SchemaId = CultMeshProviderSessionWireContract.CommandSchema,
|
||||
RecordKey = "provider-command:aetheria:aetheria-42:cmd-7",
|
||||
PayloadEncoding = "messagepack",
|
||||
Payload = Convert.FromBase64String(encodedCommand)
|
||||
});
|
||||
var commandJson = MessagePackSerializer.ConvertToJson(Convert.FromBase64String(encodedCommand));
|
||||
var receiptJson = MessagePackSerializer.ConvertToJson(Convert.FromBase64String(encodedReceipt));
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(commandJson, Does.Contain("\"commandId\""));
|
||||
Assert.That(commandJson, Does.Contain("\"commandKind\""));
|
||||
Assert.That(commandJson, Does.Contain("\"serviceInstanceId\""));
|
||||
Assert.That(commandJson, Does.Contain("\"payload\""));
|
||||
Assert.That(decodedCommand.CommandKind, Is.EqualTo("pilot.thrust"));
|
||||
Assert.That(decodedCommand.Payload, Is.Not.Null);
|
||||
Assert.That(decodedCommandDocument.CommandId, Is.EqualTo("cmd-7"));
|
||||
Assert.That(receiptJson, Does.Contain("\"completedAtUtc\""));
|
||||
Assert.That(receiptJson, Does.Contain("\"receiptId\""));
|
||||
Assert.That(receiptJson, Does.Contain("\"result\""));
|
||||
Assert.That(decodedReceipt.Receipt.State, Is.EqualTo("applied"));
|
||||
Assert.That(decodedReceipt.Receipt.Result, Is.Not.Null);
|
||||
});
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void LeaseResponse_RoundTrips_WithApplicationAcceptance()
|
||||
{
|
||||
var request = CultMeshProviderSessionWire.CreateRequest(
|
||||
"register-1",
|
||||
CultMeshProviderSessionWireContract.RegisterOperation,
|
||||
CultMeshProviderSessionWireContract.RegistrationSchema,
|
||||
new CultMeshProviderRegistrationWire
|
||||
{
|
||||
ProviderId = "aetheria",
|
||||
ServiceInstanceId = "aetheria-42",
|
||||
EndpointId = "public",
|
||||
VerseId = "verse",
|
||||
RequestedLeaseDurationMs = 10_000
|
||||
});
|
||||
var response = CultMeshProviderSessionWire.CreateResponse(
|
||||
request,
|
||||
CultMeshProviderSessionWireContract.OkStatus,
|
||||
CultMeshProviderSessionWireContract.LeaseSchema,
|
||||
new CultMeshProviderLeaseWire
|
||||
{
|
||||
ProviderId = "aetheria",
|
||||
ServiceInstanceId = "aetheria-42",
|
||||
EndpointId = "public",
|
||||
VerseId = "verse",
|
||||
LeaseId = "lease-2",
|
||||
ValidFromUtc = "2026-07-14T12:00:00Z",
|
||||
ExpiresAtUtc = "2026-07-14T12:00:10Z"
|
||||
},
|
||||
sourceRuntimeId: "odin");
|
||||
|
||||
var lease = CultMeshProviderSessionWire.DecodeResponse<CultMeshProviderLeaseWire>(
|
||||
response,
|
||||
CultMeshProviderSessionWireContract.RegisterOperation,
|
||||
CultMeshProviderSessionWireContract.LeaseSchema);
|
||||
|
||||
Assert.That(response.Status, Is.EqualTo("ok"));
|
||||
Assert.That(lease.LeaseId, Is.EqualTo("lease-2"));
|
||||
Assert.That(lease.ExpiresAtUtc, Is.EqualTo("2026-07-14T12:00:10Z"));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void Validation_RejectsSplitAuthorityAndMalformedLifecycleState()
|
||||
{
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(
|
||||
() => CultMeshProviderSessionWire.CreateRequest(
|
||||
"bad-schema",
|
||||
CultMeshProviderSessionWireContract.RegisterOperation,
|
||||
CultMeshProviderSessionWireContract.PublicationPutSchema,
|
||||
new CultMeshProviderRegistrationWire()),
|
||||
Throws.ArgumentException);
|
||||
Assert.That(
|
||||
() => CultMeshProviderSessionWire.EncodePayload(new CultMeshProviderLeaseWire
|
||||
{
|
||||
ProviderId = "provider",
|
||||
ServiceInstanceId = "instance",
|
||||
EndpointId = "endpoint",
|
||||
VerseId = "verse",
|
||||
LeaseId = "lease",
|
||||
ValidFromUtc = "2026-07-14T12:00:10Z",
|
||||
ExpiresAtUtc = "2026-07-14T12:00:00Z"
|
||||
}),
|
||||
Throws.ArgumentException);
|
||||
Assert.That(
|
||||
() => CultMeshProviderSessionWire.EncodePayload(new CultMeshProviderCommandReceiptWire
|
||||
{
|
||||
ReceiptId = "receipt",
|
||||
CommandId = "command",
|
||||
CommandKind = "kind",
|
||||
ProviderId = "provider",
|
||||
ServiceInstanceId = "instance",
|
||||
State = "maybe",
|
||||
CompletedAtUtc = "2026-07-14T12:00:00Z"
|
||||
}),
|
||||
Throws.ArgumentException);
|
||||
});
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue