Bind CultMesh authority identity to physical routes

This commit is contained in:
Meta Vens 2026-08-18 17:21:58 +02:00
commit 6c848bd6c0
45 changed files with 1671 additions and 194 deletions

6
package-lock.json generated
View file

@ -846,11 +846,11 @@
}
},
"packages/cultmesh-browser": {
"version": "0.2.0",
"version": "0.3.0",
"license": "MIT",
"dependencies": {
"@msgpack/msgpack": "^3.1.3",
"cultnet-ts": "^0.1.1"
"cultnet-ts": "^0.2.0"
},
"devDependencies": {
"@types/node": "^24.0.0",
@ -878,7 +878,7 @@
}
},
"packages/cultnet-ts": {
"version": "0.1.1",
"version": "0.2.0",
"license": "MIT",
"dependencies": {
"@msgpack/msgpack": "^3.1.3",

View file

@ -43,10 +43,19 @@ selection, reconnect, or resubscription. This is deliberately distinct from an
Eve surface's `providerId`, which identifies the UI provider inside the Verse.
The physical socket is a replaceable route, not either stable identity.
Odin returns exact authority-route bindings. The browser selects only routes
bound to the requested authority and `cultmesh.documents.v1`, then performs a
session-open handshake. It does not report `connected` until the peer proves
the exact Verse, authority runtime, protocol, and route generation. Operation
responses must also name that authority. Legacy catalogs are accepted only
when one authority runtime makes the endpoint binding unambiguous.
The browser handshake carries authentication through the host application's
ordinary secure cookie or equivalent upgrade credential. The C# host adapter
rejects endpoints without an authorization predicate unless anonymous local
development is explicitly enabled.
Run `scripts/verify-eve-browser-network.mjs --eve-root <path-to-Eve>` for the
real Chromium + C# headless chronology.
real Chromium + C# headless chronology. The witness publishes a lower-priority
decoy endpoint owned by another runtime and proves that neither client connects
to it before and after route rotation.

View file

@ -1,6 +1,6 @@
{
"name": "cultmesh-browser",
"version": "0.2.0",
"version": "0.3.0",
"description": "Browser-safe CultMesh document leases and typed operations over CultNet WebSocket transport.",
"type": "module",
"main": "dist/index.js",
@ -41,7 +41,7 @@
},
"dependencies": {
"@msgpack/msgpack": "^3.1.3",
"cultnet-ts": "^0.1.1"
"cultnet-ts": "^0.2.0"
},
"devDependencies": {
"@types/node": "^24.0.0",

View file

@ -12,6 +12,8 @@ import {
type CultNetSnapshotResponseRawMessage,
type CultMeshVerseCatalogRequestMessage,
type CultMeshVerseCatalogResponseMessage,
type CultMeshSessionAcceptedMessage,
type CultMeshSessionOpenMessage,
} from "cultnet-ts/contracts";
export interface CultMeshBrowserIdentity {
@ -21,7 +23,8 @@ export interface CultMeshBrowserIdentity {
export interface CultMeshBrowserRoute extends CultMeshBrowserIdentity {
endpoint: string;
generation?: string;
protocolId?: string;
generation: string;
}
export interface CultMeshBrowserRendezvous {
@ -173,6 +176,7 @@ export interface CultMeshOperationOptions {
payloadSchema: string;
payload: unknown;
idempotencyKey?: string;
/** @deprecated The connected authority is the operation target. A different value is rejected. */
targetRuntimeId?: string;
}
@ -338,6 +342,11 @@ export class CultMeshBrowserClient implements AsyncDisposable {
requireText(options.serviceId, "serviceId");
requireText(options.operation, "operation");
requireText(options.payloadSchema, "payloadSchema");
if (options.targetRuntimeId && options.targetRuntimeId !== this.identity.authorityRuntimeId) {
throw new Error(
`CultMesh operation target '${options.targetRuntimeId}' does not match connected authority '${this.identity.authorityRuntimeId}'.`,
);
}
await this.ensureConnected();
const messageId = options.idempotencyKey ?? this.#options.createId();
requireText(messageId, "operation idempotencyKey");
@ -353,7 +362,7 @@ export class CultMeshBrowserClient implements AsyncDisposable {
payloadEncoding: "messagepack-base64",
payload: bytesToBase64(encode(options.payload)),
sourceRuntimeId: this.runtimeId,
...(options.targetRuntimeId ? { targetRuntimeId: options.targetRuntimeId } : {}),
targetRuntimeId: this.identity.authorityRuntimeId,
};
const response = new Promise<CultNetOperationResponseMessage>((resolve, reject) => {
const timer = setTimeout(() => {
@ -489,13 +498,67 @@ export class CultMeshBrowserClient implements AsyncDisposable {
socket.binaryType = "arraybuffer";
try {
await new Promise<void>((resolve, reject) => {
let opened = false;
let settled = false;
const handshakeId = this.#options.createId();
const finish = (error?: Error) => {
if (settled) return;
settled = true;
clearTimeout(timer);
if (error) reject(error);
else resolve();
};
const timer = setTimeout(
() => reject(new Error(`CultMesh WebSocket open timed out for '${route.endpoint}'.`)),
() => finish(new Error(
opened
? `CultMesh authority handshake timed out for '${route.endpoint}'.`
: `CultMesh WebSocket open timed out for '${route.endpoint}'.`,
)),
this.#options.connectTimeoutMs,
);
socket.onopen = () => { clearTimeout(timer); resolve(); };
socket.onerror = () => { clearTimeout(timer); reject(new Error(`CultMesh WebSocket could not open '${route.endpoint}'.`)); };
socket.onclose = () => { clearTimeout(timer); reject(new Error(`CultMesh WebSocket closed while opening '${route.endpoint}'.`)); };
socket.onopen = () => {
opened = true;
try {
sendSocketMessage(socket, {
schemaVersion: "cultnet.hello.v0",
runtimeId: this.runtimeId,
runtimeKind: "browser",
supportedMessageVersions: [
"cultmesh.session_open.v1",
"cultmesh.session_accepted.v1",
"cultnet.database_subscribe.v0",
"cultnet.database_unsubscribe.v0",
"cultnet.database_change_raw.v0",
"cultnet.operation_request.v0",
"cultnet.operation_response.v0",
],
}, this.#options.maxFrameBytes);
const request: CultMeshSessionOpenMessage = {
schemaVersion: "cultmesh.session_open.v1",
messageId: handshakeId,
sourceRuntimeId: this.runtimeId,
verseId: this.identity.verseId,
authorityRuntimeId: this.identity.authorityRuntimeId,
protocolId: route.protocolId ?? "cultmesh.documents.v1",
routeGeneration: route.generation,
};
sendSocketMessage(socket, request, this.#options.maxFrameBytes);
} catch (error) {
finish(error instanceof Error ? error : new Error(String(error)));
}
};
socket.onmessage = event => {
void decodeSocketMessage(event, this.#options.maxFrameBytes).then(message => {
if (message.schemaVersion === "cultnet.error.v0") {
finish(new Error(message.error));
return;
}
if (message.schemaVersion !== "cultmesh.session_accepted.v1" || message.messageId !== handshakeId) return;
finish(validateSessionAcceptance(message, this.identity, route));
}).catch(error => finish(error instanceof Error ? error : new Error(String(error))));
};
socket.onerror = () => finish(new Error(`CultMesh WebSocket could not open '${route.endpoint}'.`));
socket.onclose = () => finish(new Error(`CultMesh WebSocket closed before proving authority at '${route.endpoint}'.`));
});
} catch (error) {
socket.close(1000, "CultMesh route open failed");
@ -511,18 +574,6 @@ export class CultMeshBrowserClient implements AsyncDisposable {
socket.onclose = () => this.handleClosed(generation);
this.setState("connected");
try {
this.send({
schemaVersion: "cultnet.hello.v0",
runtimeId: this.runtimeId,
runtimeKind: "browser",
supportedMessageVersions: [
"cultnet.database_subscribe.v0",
"cultnet.database_unsubscribe.v0",
"cultnet.database_change_raw.v0",
"cultnet.operation_request.v0",
"cultnet.operation_response.v0",
],
});
await Promise.all([...this.#leases.values()].map(lease => this.subscribe(lease)));
} catch (error) {
if (this.#socket === socket) this.#socket = undefined;
@ -576,6 +627,15 @@ export class CultMeshBrowserClient implements AsyncDisposable {
case "cultnet.operation_response.v0": {
const pending = this.#pendingOperations.get(message.messageId);
if (!pending) return;
if (message.sourceRuntimeId !== this.identity.authorityRuntimeId) {
this.#pendingOperations.delete(message.messageId);
clearTimeout(pending.timer);
pending.reject(new Error(
`CultMesh operation response came from '${message.sourceRuntimeId ?? "unknown"}', expected '${this.identity.authorityRuntimeId}'.`,
));
this.#socket?.close(1008, "CultMesh authority identity changed");
return;
}
this.#pendingOperations.delete(message.messageId);
clearTimeout(pending.timer);
if (message.payloadSchema === cultNetOperationFailureSchema) {
@ -733,33 +793,72 @@ function selectOdinRoute(
response: CultMeshVerseCatalogResponseMessage,
identity: CultMeshBrowserIdentity,
): CultMeshBrowserRoute {
const matchingVerses = response.verses.filter(verse =>
verse.verseId === identity.verseId &&
verse.authorityRuntimeIds.includes(identity.authorityRuntimeId));
const matchingVerses = response.verses.filter(verse => verse.verseId === identity.verseId);
if (matchingVerses.length !== 1) {
throw new Error(
`Odin returned ${matchingVerses.length} routes for Verse '${identity.verseId}' authority runtime '${identity.authorityRuntimeId}'.`,
);
}
const verse = matchingVerses[0];
const endpoints = verse.discoveryEndpoints.filter(endpoint => {
if (verse.authorityRoutes == null && verse.authorityRuntimeIds.length !== 1) {
throw new Error(
`Odin returned ambiguous legacy routes for Verse '${identity.verseId}'. Explicit authority route bindings are required.`,
);
}
if (verse.authorityRoutes == null && verse.authorityRuntimeIds[0] !== identity.authorityRuntimeId) {
throw new Error(
`Odin advertised no route for authority runtime '${identity.authorityRuntimeId}'.`,
);
}
const boundRoutes = verse.authorityRoutes ?? verse.discoveryEndpoints.map(endpoint => ({
authorityRuntimeId: verse.authorityRuntimeIds[0],
endpoint,
protocolIds: ["cultmesh.documents.v1"],
priority: 0,
generation: [verse.compatibility.transportVersion, verse.compatibility.rulesHash, endpoint].join(":"),
}));
const routes = boundRoutes.filter(route =>
route.authorityRuntimeId === identity.authorityRuntimeId &&
route.protocolIds.includes("cultmesh.documents.v1") &&
(() => {
try {
requireWebSocketEndpoint(endpoint, "Odin provider route");
requireWebSocketEndpoint(route.endpoint, "Odin provider route");
return true;
} catch {
return false;
}
});
if (endpoints.length === 0) {
})()).sort((left, right) => left.priority - right.priority || left.endpoint.localeCompare(right.endpoint));
if (routes.length === 0) {
throw new Error(`Odin advertised no browser-compatible route for authority runtime '${identity.authorityRuntimeId}'.`);
}
const route = routes[0];
return {
...identity,
endpoint: endpoints[0],
generation: [verse.compatibility.transportVersion, verse.compatibility.rulesHash, endpoints[0]].join(":"),
endpoint: route.endpoint,
protocolId: "cultmesh.documents.v1",
generation: route.generation,
};
}
function validateSessionAcceptance(
message: CultMeshSessionAcceptedMessage,
identity: CultMeshBrowserIdentity,
route: CultMeshBrowserRoute,
): Error | undefined {
if (!message.accepted) return new Error(message.error ?? "CultMesh authority rejected the session.");
const protocolId = route.protocolId ?? "cultmesh.documents.v1";
if (message.verseId !== identity.verseId ||
message.authorityRuntimeId !== identity.authorityRuntimeId ||
message.protocolId !== protocolId ||
message.routeGeneration !== route.generation) {
return new Error(
`CultMesh route proved '${message.verseId}/${message.authorityRuntimeId}/${message.protocolId}/${message.routeGeneration}', ` +
`expected '${identity.verseId}/${identity.authorityRuntimeId}/${protocolId}/${route.generation}'.`,
);
}
return undefined;
}
function requireText(value: string, field: string): void {
if (!value || value.trim().length === 0) throw new Error(`CultMesh browser ${field} is required.`);
}

View file

@ -9,6 +9,7 @@ import {
type CultNetMessage,
type CultNetOperationRequestMessage,
type CultNetRawDocumentRecord,
type CultMeshSessionOpenMessage,
} from "cultnet-ts/contracts";
import {
@ -48,6 +49,46 @@ test("Odin rendezvous resolves stable authority-runtime identity and observes ro
);
});
test("Odin rendezvous never selects another authority's faster endpoint", async () => {
const odin = new FakeOdin("ws://127.0.0.1:4050/mesh");
odin.decoyEndpoint = "ws://127.0.0.1:4049/decoy";
const rendezvous = new CultMeshBrowserOdinRendezvous({
endpoints: ["ws://127.0.0.1:4040/odin"],
runtimeId: "browser-route-binding-test",
socketFactory: () => odin.open(),
});
const route = await rendezvous.resolve({
verseId: "sample.counter",
authorityRuntimeId: "sample.counter-daemon",
});
assert.equal(route.endpoint, "ws://127.0.0.1:4050/mesh");
});
test("Odin rendezvous accepts one-authority legacy routes and rejects ambiguous legacy catalogs", async () => {
const odin = new FakeOdin("ws://127.0.0.1:4070/mesh");
odin.omitAuthorityRoutes = true;
const rendezvous = new CultMeshBrowserOdinRendezvous({
endpoints: ["ws://127.0.0.1:4040/odin"],
runtimeId: "browser-legacy-route-test",
socketFactory: () => odin.open(),
});
const route = await rendezvous.resolve({
verseId: "sample.counter",
authorityRuntimeId: "sample.counter-daemon",
});
assert.equal(route.endpoint, "ws://127.0.0.1:4070/mesh");
odin.decoyEndpoint = "ws://127.0.0.1:4069/decoy";
await assert.rejects(
rendezvous.resolve({ verseId: "sample.counter", authorityRuntimeId: "sample.counter-daemon" }),
(error: unknown) =>
error instanceof AggregateError &&
error.errors.some(inner => inner instanceof Error && /ambiguous legacy routes/.test(inner.message)),
);
});
test("browser client leases provider state, invokes an operation, and follows route replacement", async () => {
const first = new FakeProvider("ws://127.0.0.1:4101/mesh", 4);
const second = new FakeProvider("ws://127.0.0.1:4102/mesh", 9);
@ -117,18 +158,33 @@ test("browser client rejects wrong rendezvous identity and has no document write
assert.equal("setDocument" in CultMeshBrowserClient.prototype, false);
});
test("browser client rejects a route whose peer proves the wrong authority", async () => {
const provider = new FakeProvider("ws://127.0.0.1:4251/mesh", 1, false, "intruder-runtime");
await assert.rejects(
CultMeshBrowserClient.connect({
verseId: "sample.counter",
authorityRuntimeId: "sample.counter-daemon",
runtimeId: "browser-peer-proof-test",
rendezvous: { resolve: async () => ({ ...provider.route, authorityRuntimeId: "sample.counter-daemon" }) },
socketFactory: () => provider.open(),
}),
/proved .*intruder-runtime.*expected .*sample.counter-daemon/,
);
});
test("browser client bounds unanswered provider operations", async () => {
const route = {
verseId: "sample.counter",
authorityRuntimeId: "sample.counter-daemon",
endpoint: "ws://127.0.0.1:4301/mesh",
generation: "route-4301",
};
const client = await CultMeshBrowserClient.connect({
...route,
runtimeId: "browser-timeout-test",
rendezvous: { resolve: async () => route },
requestTimeoutMs: 5,
socketFactory: () => new FakeSocket(() => undefined, () => undefined),
socketFactory: () => handshakeOnlySocket(route),
});
await assert.rejects(
client.invoke({
@ -172,11 +228,39 @@ test("browser client rejects a correlated framework failure without waiting for
await client.dispose();
});
test("browser client rejects an operation response from another runtime", async () => {
const provider = new FakeProvider(
"ws://127.0.0.1:4371/mesh",
1,
false,
"sample.counter-daemon",
"intruder-runtime",
);
const client = await CultMeshBrowserClient.connect({
...provider.route,
runtimeId: "browser-response-authority-test",
rendezvous: { resolve: async () => provider.route },
socketFactory: () => provider.open(),
});
await assert.rejects(
client.invoke({
serviceId: "sample.counter",
operation: "increment",
payloadSchema: "sample.increment.v1",
payload: { amount: 1 },
}),
/response came from 'intruder-runtime'/,
);
await client.dispose();
});
test("browser client rejects oversized outbound schema messages", async () => {
const route = {
verseId: "sample.counter",
authorityRuntimeId: "sample.counter-daemon",
endpoint: "ws://127.0.0.1:4401/mesh",
generation: "route-4401",
};
await assert.rejects(
CultMeshBrowserClient.connect({
@ -220,8 +304,16 @@ class FakeProvider {
#count: number;
#subscriptions = new Map<FakeSocket, Map<string, CultNetDatabaseSubscribeMessage>>();
#rejectOperations: boolean;
#runtimeId: string;
#responseRuntimeId: string;
constructor(endpoint: string, count: number, rejectOperations = false) {
constructor(
endpoint: string,
count: number,
rejectOperations = false,
runtimeId = "sample.counter-daemon",
responseRuntimeId = runtimeId,
) {
this.route = {
verseId: "sample.counter",
authorityRuntimeId: "sample.counter-daemon",
@ -230,6 +322,8 @@ class FakeProvider {
};
this.#count = count;
this.#rejectOperations = rejectOperations;
this.#runtimeId = runtimeId;
this.#responseRuntimeId = responseRuntimeId;
}
get activeSubscriptionCount(): number {
@ -250,6 +344,9 @@ class FakeProvider {
private receive(socket: FakeSocket, wire: Uint8Array): void {
const message = parseCultNetMessage(decode(wire));
switch (message.schemaVersion) {
case "cultmesh.session_open.v1":
this.acceptSession(socket, message);
return;
case "cultnet.database_subscribe.v0":
this.subscribe(socket, message);
return;
@ -264,6 +361,18 @@ class FakeProvider {
}
}
private acceptSession(socket: FakeSocket, message: CultMeshSessionOpenMessage): void {
socket.deliver({
schemaVersion: "cultmesh.session_accepted.v1",
messageId: message.messageId,
accepted: true,
verseId: message.verseId,
authorityRuntimeId: this.#runtimeId,
protocolId: message.protocolId,
routeGeneration: message.routeGeneration,
});
}
private subscribe(socket: FakeSocket, message: CultNetDatabaseSubscribeMessage): void {
this.#subscriptions.get(socket)!.set(message.subscriptionId, message);
socket.deliver({
@ -274,6 +383,7 @@ class FakeProvider {
}
private invoke(socket: FakeSocket, message: CultNetOperationRequestMessage): void {
assert.equal(message.targetRuntimeId, this.#runtimeId);
if (this.#rejectOperations) {
socket.deliver({
schemaVersion: "cultnet.operation_response.v0",
@ -288,7 +398,7 @@ class FakeProvider {
message: "Expected payload schema 'sample.increment.v1'.",
})),
diagnostics: ["Request schema did not match."],
sourceRuntimeId: "sample.counter-daemon",
sourceRuntimeId: this.#responseRuntimeId,
});
return;
}
@ -314,7 +424,7 @@ class FakeProvider {
payloadSchema: "sample.increment_receipt.v1",
payloadEncoding: "messagepack-base64",
payload: bytesToBase64(encode({ count: this.#count })),
sourceRuntimeId: "sample.counter-daemon",
sourceRuntimeId: this.#responseRuntimeId,
});
}
@ -328,13 +438,15 @@ class FakeProvider {
storedAt: "2026-08-17T00:00:00Z",
payloadEncoding: "messagepack",
payload: encode({ count: this.#count }),
sourceRuntimeId: "sample.counter-daemon",
sourceRuntimeId: this.#runtimeId,
};
}
}
class FakeOdin {
providerEndpoint: string;
decoyEndpoint: string | undefined;
omitAuthorityRoutes = false;
constructor(providerEndpoint: string) {
this.providerEndpoint = providerEndpoint;
@ -359,7 +471,25 @@ class FakeOdin {
optionalPluginIds: [],
},
discoveryEndpoints: [this.providerEndpoint],
authorityRuntimeIds: ["sample.counter-daemon"],
authorityRuntimeIds: this.decoyEndpoint
? ["sample.counter-daemon", "sample.decoy-daemon"]
: ["sample.counter-daemon"],
...(!this.omitAuthorityRoutes ? { authorityRoutes: [
...(this.decoyEndpoint ? [{
authorityRuntimeId: "sample.decoy-daemon",
endpoint: this.decoyEndpoint,
protocolIds: ["cultmesh.documents.v1"],
priority: 0,
generation: "decoy-generation",
}] : []),
{
authorityRuntimeId: "sample.counter-daemon",
endpoint: this.providerEndpoint,
protocolIds: ["cultmesh.documents.v1"],
priority: 10,
generation: this.providerEndpoint,
},
] } : {}),
}],
});
}, () => undefined);
@ -367,6 +497,24 @@ class FakeOdin {
}
}
function handshakeOnlySocket(route: CultMeshBrowserRoute): FakeSocket {
let socket: FakeSocket;
socket = new FakeSocket(wire => {
const message = parseCultNetMessage(decode(wire));
if (message.schemaVersion !== "cultmesh.session_open.v1") return;
socket.deliver({
schemaVersion: "cultmesh.session_accepted.v1",
messageId: message.messageId,
accepted: true,
verseId: route.verseId,
authorityRuntimeId: route.authorityRuntimeId,
protocolId: message.protocolId,
routeGeneration: message.routeGeneration,
});
}, () => undefined);
return socket;
}
class FakeSocket implements CultMeshBrowserSocket {
binaryType: BinaryType = "arraybuffer";
readyState = 0;

View file

@ -0,0 +1,18 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://github.com/GameCult/cultnet-ts/contracts/cultmesh.session-accepted.schema.json",
"title": "CultMesh Session Accepted",
"type": "object",
"required": ["schemaVersion", "messageId", "accepted", "verseId", "authorityRuntimeId", "protocolId", "routeGeneration"],
"additionalProperties": false,
"properties": {
"schemaVersion": { "const": "cultmesh.session_accepted.v1" },
"messageId": { "type": "string", "minLength": 1 },
"accepted": { "type": "boolean" },
"verseId": { "type": "string", "minLength": 1 },
"authorityRuntimeId": { "type": "string", "minLength": 1 },
"protocolId": { "type": "string", "minLength": 1 },
"routeGeneration": { "type": "string" },
"error": { "type": "string", "minLength": 1 }
}
}

View file

@ -0,0 +1,17 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://github.com/GameCult/cultnet-ts/contracts/cultmesh.session-open.schema.json",
"title": "CultMesh Session Open",
"type": "object",
"required": ["schemaVersion", "messageId", "sourceRuntimeId", "verseId", "authorityRuntimeId", "protocolId", "routeGeneration"],
"additionalProperties": false,
"properties": {
"schemaVersion": { "const": "cultmesh.session_open.v1" },
"messageId": { "type": "string", "minLength": 1 },
"sourceRuntimeId": { "type": "string", "minLength": 1 },
"verseId": { "type": "string", "minLength": 1 },
"authorityRuntimeId": { "type": "string", "minLength": 1 },
"protocolId": { "type": "string", "minLength": 1 },
"routeGeneration": { "type": "string" }
}
}

View file

@ -18,6 +18,18 @@
"optionalPluginIds": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true }
}
},
"authorityRoute": {
"type": "object",
"required": ["authorityRuntimeId", "endpoint", "protocolIds", "priority", "generation"],
"additionalProperties": false,
"properties": {
"authorityRuntimeId": { "type": "string", "minLength": 1 },
"endpoint": { "type": "string", "minLength": 1 },
"protocolIds": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true },
"priority": { "type": "integer", "minimum": 0 },
"generation": { "type": "string", "minLength": 1 }
}
},
"verse": {
"type": "object",
"required": ["verseId", "displayName", "authorityModel", "compatibility", "discoveryEndpoints", "authorityRuntimeIds"],
@ -29,6 +41,7 @@
"compatibility": { "$ref": "#/$defs/compatibility" },
"discoveryEndpoints": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true },
"authorityRuntimeIds": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true },
"authorityRoutes": { "type": "array", "items": { "$ref": "#/$defs/authorityRoute" } },
"parentVerseId": { "type": "string", "minLength": 1 },
"description": { "type": "string" }
}

View file

@ -1,6 +1,6 @@
{
"name": "cultnet-ts",
"version": "0.1.1",
"version": "0.2.0",
"description": "CultNet-style typed direct-pipe messaging and state replication for TypeScript runtimes.",
"main": "dist/index.js",
"types": "dist/index.d.ts",

View file

@ -30,6 +30,8 @@ import operationRequestSchema from "../contracts/cultnet.operation-request.schem
import operationResponseSchema from "../contracts/cultnet.operation-response.schema.json";
import verseCatalogRequestSchema from "../contracts/cultmesh.verse-catalog-request.schema.json";
import verseCatalogResponseSchema from "../contracts/cultmesh.verse-catalog-response.schema.json";
import sessionOpenSchema from "../contracts/cultmesh.session-open.schema.json";
import sessionAcceptedSchema from "../contracts/cultmesh.session-accepted.schema.json";
import ghostlightAgentStateSchema from "../contracts/ghostlight.agent-state.schema.json";
export type CultNetWireContract = "cultnet.schema.v0" | "gamecult.networking.v0";
@ -59,7 +61,9 @@ export type CultNetSchemaVersion =
| "cultnet.operation_request.v0"
| "cultnet.operation_response.v0"
| "cultmesh.verse_catalog_request.v0"
| "cultmesh.verse_catalog_response.v0";
| "cultmesh.verse_catalog_response.v0"
| "cultmesh.session_open.v1"
| "cultmesh.session_accepted.v1";
export type CultNetSchemaKind = "wire_message" | "document_payload" | "shared_contract";
export type CultNetRawPayloadEncoding = "messagepack";
@ -350,10 +354,40 @@ export interface CultMeshVerseDescriptorMessage {
compatibility: CultMeshVerseCompatibilityMessage;
discoveryEndpoints: string[];
authorityRuntimeIds: string[];
authorityRoutes?: CultMeshAuthorityRouteMessage[];
parentVerseId?: string;
description?: string;
}
export interface CultMeshAuthorityRouteMessage {
authorityRuntimeId: string;
endpoint: string;
protocolIds: string[];
priority: number;
generation: string;
}
export interface CultMeshSessionOpenMessage {
schemaVersion: "cultmesh.session_open.v1";
messageId: string;
sourceRuntimeId: string;
verseId: string;
authorityRuntimeId: string;
protocolId: string;
routeGeneration: string;
}
export interface CultMeshSessionAcceptedMessage {
schemaVersion: "cultmesh.session_accepted.v1";
messageId: string;
accepted: boolean;
verseId: string;
authorityRuntimeId: string;
protocolId: string;
routeGeneration: string;
error?: string;
}
export interface CultMeshVerseCatalogRequestMessage {
schemaVersion: "cultmesh.verse_catalog_request.v0";
messageId: string;
@ -618,7 +652,9 @@ export type CultNetMessage =
| CultNetOperationRequestMessage
| CultNetOperationResponseMessage
| CultMeshVerseCatalogRequestMessage
| CultMeshVerseCatalogResponseMessage;
| CultMeshVerseCatalogResponseMessage
| CultMeshSessionOpenMessage
| CultMeshSessionAcceptedMessage;
const ajv = new Ajv2020({
allErrors: true,
@ -651,6 +687,8 @@ const CULTNET_MESSAGE_SCHEMAS = [
operationResponseSchema,
verseCatalogRequestSchema,
verseCatalogResponseSchema,
sessionOpenSchema,
sessionAcceptedSchema,
] as const;
const cultNetValidators = new Map<CultNetSchemaVersion, ValidateFunction>();
@ -1198,6 +1236,9 @@ function normalizeCultNetOptionalNulls(
}
}
return;
case "cultmesh.session_accepted.v1":
stripNullProperties(candidate, ["error"]);
return;
case "cultnet.schema_catalog_request.v0":
stripNullProperties(candidate, ["schemaIds", "kinds"]);
return;

View file

@ -83,10 +83,29 @@ test("CultNet validates the canonical CultMesh Verse catalog messages", () => {
},
discoveryEndpoints: ["wss://provider.example/cultmesh"],
authorityRuntimeIds: ["sample.counter-provider"],
authorityRoutes: [{
authorityRuntimeId: "sample.counter-provider",
endpoint: "wss://provider.example/cultmesh",
protocolIds: ["cultmesh.documents.v1"],
priority: 0,
generation: "provider-route-1",
}],
}],
});
assert.equal(response.schemaVersion, "cultmesh.verse_catalog_response.v0");
assert.equal(response.verses[0].authorityRuntimeIds[0], "sample.counter-provider");
assert.equal(response.verses[0].authorityRoutes![0].generation, "provider-route-1");
const session = parseCultNetMessage({
schemaVersion: "cultmesh.session_accepted.v1",
messageId: "session-1",
accepted: true,
verseId: "sample.counter",
authorityRuntimeId: "sample.counter-provider",
protocolId: "cultmesh.documents.v1",
routeGeneration: "provider-route-1",
});
assert.equal(session.schemaVersion, "cultmesh.session_accepted.v1");
assert.throws(() => parseCultNetMessage({
...response,

View file

@ -37,13 +37,30 @@ static async Task RunOdinAsync(Args arguments)
if (string.IsNullOrWhiteSpace(arguments.ProviderEndpoint))
throw new ArgumentException("The Odin fixture requires --provider-endpoint.");
using var catalog = new CultMeshVerseCatalog();
var routes = new List<CultMeshAuthorityRoute>
{
new(
arguments.AuthorityRuntimeId,
arguments.ProviderEndpoint,
[CultMeshProtocols.Documents.Value],
priority: 10,
generation: arguments.RouteGeneration)
};
if (!string.IsNullOrWhiteSpace(arguments.DecoyEndpoint))
{
routes.Add(new CultMeshAuthorityRoute(
arguments.DecoyAuthorityRuntimeId,
arguments.DecoyEndpoint,
[CultMeshProtocols.Documents.Value],
priority: 0,
generation: arguments.DecoyRouteGeneration));
}
catalog.Upsert(new CultMeshVerseDescriptor(
arguments.VerseId,
arguments.VerseName,
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility(arguments.TransportVersion, arguments.RulesHash),
discoveryEndpoints: [arguments.ProviderEndpoint],
authorityRuntimeIds: [arguments.AuthorityRuntimeId],
authorityRoutes: routes,
description: "Local executable Odin fixture for the browser/network onboarding witness."));
await using var schemaServer = new CultNetWebSocketSchemaServer();
@ -97,6 +114,12 @@ static async Task RunProviderAsync(Args arguments)
await node.FlushAsync();
await using var schemaServer = new CultNetWebSocketSchemaServer();
using var identity = new CultMeshSessionIdentityServer(
schemaServer,
arguments.AuthorityRuntimeId,
[arguments.VerseId],
[CultMeshProtocols.Documents.Value],
[arguments.RouteGeneration]);
using var subscriptions = new CultNetDatabaseSubscriptionServer(schemaServer, node.Database);
using var operationGate = new SemaphoreSlim(1, 1);
using var operations = new CultNetOperationServer(schemaServer, arguments.AuthorityRuntimeId)
@ -347,12 +370,16 @@ sealed class Args
public string StatePath { get; private init; } = "sample-counter.cc";
public string OdinEndpoint { get; private init; } = "";
public string ProviderEndpoint { get; private init; } = "";
public string DecoyEndpoint { get; private init; } = "";
public string Token { get; private init; } = "sample-session";
public string VerseId { get; private init; } = "sample.counter";
public string VerseName { get; private init; } = "CultMesh browser counter";
public string AuthorityRuntimeId { get; private init; } = "sample.counter-daemon";
public string DecoyAuthorityRuntimeId { get; private init; } = "sample.decoy-daemon";
public string TransportVersion { get; private init; } = "cultmesh.v0";
public string RulesHash { get; private init; } = "sample-counter-v1";
public string RouteGeneration { get; private init; } = "sample-counter-route-1";
public string DecoyRouteGeneration { get; private init; } = "sample-decoy-route-1";
public int ExpectedCount { get; private init; } = 2;
public int BenchmarkOperations { get; private init; } = 10_000;
@ -369,12 +396,16 @@ sealed class Args
StatePath = named.GetValueOrDefault("--state", "sample-counter.cc"),
OdinEndpoint = named.GetValueOrDefault("--odin", ""),
ProviderEndpoint = named.GetValueOrDefault("--provider-endpoint", ""),
DecoyEndpoint = named.GetValueOrDefault("--decoy-endpoint", ""),
Token = named.GetValueOrDefault("--token", "sample-session"),
VerseId = named.GetValueOrDefault("--verse-id", "sample.counter"),
VerseName = named.GetValueOrDefault("--verse-name", "CultMesh browser counter"),
AuthorityRuntimeId = named.GetValueOrDefault("--authority-runtime-id", "sample.counter-daemon"),
DecoyAuthorityRuntimeId = named.GetValueOrDefault("--decoy-authority-runtime-id", "sample.decoy-daemon"),
TransportVersion = named.GetValueOrDefault("--transport-version", "cultmesh.v0"),
RulesHash = named.GetValueOrDefault("--rules-hash", "sample-counter-v1"),
RouteGeneration = named.GetValueOrDefault("--route-generation", "sample-counter-route-1"),
DecoyRouteGeneration = named.GetValueOrDefault("--decoy-route-generation", "sample-decoy-route-1"),
ExpectedCount = named.TryGetValue("--expected-count", out var expectedCount)
? int.Parse(expectedCount)
: 2,

View file

@ -12,6 +12,9 @@ This is the real-process continuation of `samples/eve-two-runtime`:
without application transport or reconnect code;
- one separate local Odin fixture answers the canonical CultMesh Verse catalog
over binary CultNet WebSocket messages;
- one wrong-authority provider is deliberately advertised on a better-priority
route, and both clients must ignore it because the route is bound to another
runtime;
- the browser and C# clients speak the same binary CultNet schema-v0 WebSocket
messages through the authenticated host adapter.

View file

@ -86,7 +86,11 @@ try {
window.__sampleCount = decodeCultNetPayload<{ count: number }>(counter.current!).count;
window.__sampleReady = true;
} catch (error) {
window.__sampleError = error instanceof Error ? error.stack || error.message : String(error);
const details = error instanceof AggregateError
? error.errors.map(inner => inner instanceof Error ? inner.stack || inner.message : String(inner)).join("\n---\n")
: "";
window.__sampleError = (error instanceof Error ? error.stack || error.message : String(error))
+ (details ? `\n${details}` : "");
host.textContent = window.__sampleError;
throw error;
}

View file

@ -17,9 +17,11 @@ const eveRoot = resolve(eveFlag >= 0 ? process.argv[eveFlag + 1] : join(repoRoot
const sampleRoot = join(repoRoot, "samples", "eve-browser-network");
const workRoot = await mkdtemp(join(tmpdir(), "cultmesh-browser-network-"));
const statePath = join(workRoot, "counter.cc");
const decoyStatePath = join(workRoot, "decoy-counter.cc");
const bundlePath = join(workRoot, "bundle.js");
const token = "sample-session";
let provider;
let decoyProvider;
let replacementProvider;
let odin;
let headless;
@ -41,6 +43,8 @@ try {
alias: {
"@gamecult/eve-contracts": join(eveRoot, "packages", "eve-contracts", "src", "index.ts"),
"@gamecult/eve-browser-lowering": join(eveRoot, "packages", "eve-browser-lowering", "src", "index.ts"),
"cultmesh-browser": join(repoRoot, "packages", "cultmesh-browser", "src", "index.ts"),
"cultnet-ts/contracts": join(repoRoot, "packages", "cultnet-ts", "src", "contracts.ts"),
},
logLevel: "warning",
});
@ -54,14 +58,22 @@ try {
"--verbosity", "quiet",
"-p:NoWarn=1591%3BCS8632",
`-p:EveRoot=${eveRoot}`,
`-p:CultLibRoot=${repoRoot}`,
]);
const providerPort = await freePort();
const decoyProviderPort = await freePort();
const replacementProviderPort = await freePort();
const odinPort = await freePort();
const httpPort = await freePort();
provider = await startProvider(providerPort);
const endpoint = await provider.waitFor("PROVIDER_READY ");
odin = await startOdin(odinPort, endpoint);
decoyProvider = await startProvider(decoyProviderPort, {
authorityRuntimeId: "sample.decoy-daemon",
routeGeneration: "sample-decoy-route-1",
statePath: decoyStatePath,
});
const decoyEndpoint = await decoyProvider.waitFor("PROVIDER_READY ");
odin = await startOdin(odinPort, endpoint, decoyEndpoint);
const odinEndpoint = await odin.waitFor("ODIN_READY ");
headless = startDotnet([
"headless", "--odin", odinEndpoint, "--verse-id", "sample.counter",
@ -91,7 +103,7 @@ try {
const restartedEndpoint = await replacementProvider.waitFor("PROVIDER_READY ");
assert.notEqual(restartedEndpoint, endpoint);
await stop(odin.process);
odin = await startOdin(odinPort, restartedEndpoint);
odin = await startOdin(odinPort, restartedEndpoint, decoyEndpoint);
assert.equal(await odin.waitFor("ODIN_READY "), odinEndpoint);
await stop(provider.process);
provider = replacementProvider;
@ -122,6 +134,7 @@ try {
console.log(JSON.stringify({
providerEndpoint: endpoint,
replacementProviderEndpoint: restartedEndpoint,
rejectedWrongAuthorityEndpoint: decoyEndpoint,
odinEndpoint,
browserRuntime: "chromium",
headlessRuntime: "csharp",
@ -136,21 +149,31 @@ try {
if (browser) await browser.close().catch(() => undefined);
if (headless) await stop(headless.process);
if (provider) await stop(provider.process);
if (decoyProvider) await stop(decoyProvider.process);
if (replacementProvider) await stop(replacementProvider.process);
if (odin) await stop(odin.process);
if (httpServer) await new Promise(resolve => httpServer.close(resolve));
await rm(workRoot, { recursive: true, force: true });
}
async function startProvider(port) {
return startDotnet(["provider", "--port", String(port), "--state", statePath, "--token", token]);
async function startProvider(port, options = {}) {
return startDotnet([
"provider", "--port", String(port),
"--state", options.statePath ?? statePath,
"--authority-runtime-id", options.authorityRuntimeId ?? "sample.counter-daemon",
"--route-generation", options.routeGeneration ?? "sample-counter-route-1",
"--token", token,
]);
}
async function startOdin(port, providerEndpoint) {
async function startOdin(port, providerEndpoint, decoyEndpoint) {
return startDotnet([
"odin",
"--port", String(port),
"--provider-endpoint", providerEndpoint,
"--decoy-endpoint", decoyEndpoint,
"--decoy-authority-runtime-id", "sample.decoy-daemon",
"--decoy-route-generation", "sample-decoy-route-1",
"--token", token,
]);
}

View file

@ -42,11 +42,11 @@ namespace GameCult.Mesh.Quic.Native
public async Task<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
if (candidate == null) throw new ArgumentNullException(nameof(candidate));
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (!TryParseEndpoint(candidate.Endpoint, out var host, out var port, out var pin))
throw new NotSupportedException($"Native QUIC connector does not support '{candidate.Endpoint}'.");
@ -72,7 +72,9 @@ namespace GameCult.Mesh.Quic.Native
var transport = new CultMeshNativeQuicRealtimeTransport(
candidate.Endpoint,
handle,
_options.PollInterval);
_options.PollInterval,
target,
candidate.Generation);
try
{
await transport.WaitUntilConnectedAsync(_options.HandshakeTimeout, cancellationToken).ConfigureAwait(false);
@ -122,23 +124,42 @@ namespace GameCult.Mesh.Quic.Native
}
}
internal sealed class CultMeshNativeQuicRealtimeTransport : ICultMeshRealtimeTransport
internal sealed class CultMeshNativeQuicRealtimeTransport : ICultMeshRealtimeTransport, ICultMeshVerifiedTransport
{
private readonly object _gate = new object();
private readonly TimeSpan _pollInterval;
private IntPtr _handle;
private readonly CultMeshSessionTarget _target;
private readonly string _routeGeneration;
private bool _disposed;
public CultMeshNativeQuicRealtimeTransport(string endpoint, IntPtr handle, TimeSpan pollInterval)
public CultMeshNativeQuicRealtimeTransport(
string endpoint,
IntPtr handle,
TimeSpan pollInterval,
CultMeshSessionTarget target,
string routeGeneration)
{
Endpoint = endpoint;
_handle = handle;
_pollInterval = pollInterval;
_target = target;
_routeGeneration = routeGeneration ?? string.Empty;
}
public string TransportId => "msquic-native-realtime";
public string Endpoint { get; }
public bool IsVerifiedFor(
string verseId,
string authorityRuntimeId,
string protocolId,
string routeGeneration) =>
string.Equals(_target.VerseId, verseId, StringComparison.Ordinal) &&
string.Equals(_target.AuthorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) &&
string.Equals(CultMeshProtocols.RealtimeState.Value, protocolId, StringComparison.Ordinal) &&
string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal);
public Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default) =>
Task.FromException(new NotSupportedException(
"The Unity-compatible native QUIC connector currently owns provider-to-client state only."));

View file

@ -22,7 +22,7 @@ public sealed class CultMeshQuicRealtimeConnectorOptions
/// <summary>
/// Optional provider-aware certificate validator. Standard TLS validation is used when omitted.
/// </summary>
public Func<CultMeshEndpointId, X509Certificate2?, X509Chain?, SslPolicyErrors, bool>?
public Func<CultMeshRuntimeId, X509Certificate2?, X509Chain?, SslPolicyErrors, bool>?
ValidateProviderCertificate { get; set; }
}
@ -55,11 +55,11 @@ public sealed class CultMeshQuicRealtimeTransportConnector : ICultMeshRealtimeTr
/// <inheritdoc />
public async Task<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
if (candidate == null) throw new ArgumentNullException(nameof(candidate));
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (!QuicConnection.IsSupported)
throw new PlatformNotSupportedException("QUIC is unavailable. Install a supported MsQuic runtime.");
if (!TryParseEndpoint(candidate.Endpoint, out var host, out var port))
@ -84,14 +84,15 @@ public sealed class CultMeshQuicRealtimeTransportConnector : ICultMeshRealtimeTr
{
var providerCertificate = certificate as X509Certificate2 ??
(certificate == null ? null : new X509Certificate2(certificate));
return validator?.Invoke(endpointId, providerCertificate, chain, errors) ??
return validator?.Invoke(CultMeshRuntimeId.Parse(target.AuthorityRuntimeId), providerCertificate, chain, errors) ??
errors == SslPolicyErrors.None ||
MatchesAdvertisedCertificatePin(candidate.Endpoint, providerCertificate);
}
}
}, cancellationToken).ConfigureAwait(false);
return new CultMeshQuicRealtimeTransport(candidate.Endpoint, connection);
return new CultMeshQuicRealtimeTransport(
candidate.Endpoint, connection, target: target, routeGeneration: candidate.Generation);
}
private static bool TryParseEndpoint(string? endpoint, out string host, out int port)
@ -308,7 +309,7 @@ public sealed class CultMeshQuicRealtimeServer : IAsyncDisposable
}
}
internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport
internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport, ICultMeshVerifiedTransport
{
private readonly QuicConnection _connection;
private readonly CancellationTokenSource _shutdown = new();
@ -318,6 +319,8 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport
private readonly SemaphoreSlim _reliableSendGate = new(1, 1);
private readonly CultMeshRealtimeInbox _outbound = new();
private readonly Action<CultMeshRealtimeFrame>? _observer;
private readonly CultMeshSessionTarget? _target;
private readonly string _routeGeneration;
private readonly Task _acceptLoop;
private readonly Task _outboundLoop;
private readonly Task _completion;
@ -328,11 +331,15 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport
public CultMeshQuicRealtimeTransport(
string endpoint,
QuicConnection connection,
Action<CultMeshRealtimeFrame>? observer = null)
Action<CultMeshRealtimeFrame>? observer = null,
CultMeshSessionTarget? target = null,
string routeGeneration = "")
{
Endpoint = endpoint;
_connection = connection;
_observer = observer;
_target = target;
_routeGeneration = routeGeneration ?? string.Empty;
_acceptLoop = AcceptStreamsAsync(_shutdown.Token);
_outboundLoop = SendPublishedFramesAsync(_shutdown.Token);
_completion = Task.WhenAll(_acceptLoop, _outboundLoop);
@ -342,6 +349,17 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport
public string Endpoint { get; }
public Task Completion => _completion;
public bool IsVerifiedFor(
string verseId,
string authorityRuntimeId,
string protocolId,
string routeGeneration) =>
_target != null &&
string.Equals(_target.VerseId, verseId, StringComparison.Ordinal) &&
string.Equals(_target.AuthorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) &&
string.Equals(CultMeshProtocols.RealtimeState.Value, protocolId, StringComparison.Ordinal) &&
string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal);
internal bool TryPublishLatest(CultMeshRealtimeFrame frame)
{
if (_disposed || frame.Delivery != CultMeshRealtimeDelivery.LatestOnly) return false;

View file

@ -391,7 +391,8 @@ namespace GameCult.Mesh
PayloadSchema = requestSchema.Trim(),
PayloadEncoding = "messagepack-base64",
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(request, CultNetSchemaMessageSerialization.Options)),
SourceRuntimeId = sourceRuntimeId.Trim()
SourceRuntimeId = sourceRuntimeId.Trim(),
TargetRuntimeId = target.AuthorityRuntimeId
};
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
timeout.CancelAfter(_operationResponseTimeout);
@ -441,6 +442,11 @@ namespace GameCult.Mesh
if (!string.Equals(envelope.ServiceId, expectedService, StringComparison.Ordinal) ||
!string.Equals(envelope.Operation, expectedOperation, StringComparison.Ordinal))
throw new InvalidOperationException("CultMesh operation response did not match the requested service and operation.");
if (!string.Equals(envelope.SourceRuntimeId, target.AuthorityRuntimeId, StringComparison.Ordinal))
throw new CultMeshSessionException(new CultMeshSessionFailure(
CultMeshSessionFailureReason.Authority,
$"CultMesh operation response came from runtime '{envelope.SourceRuntimeId ?? "(missing)"}', " +
$"not selected authority '{target.AuthorityRuntimeId}'."));
if (string.Equals(envelope.PayloadSchema, CultNetOperationServer.FailureSchemaId, StringComparison.Ordinal))
{
if (!string.Equals(envelope.PayloadEncoding, "messagepack-base64", StringComparison.Ordinal))
@ -795,7 +801,7 @@ namespace GameCult.Mesh
_cache,
new CultRecordKey(recordKey),
CultMesh.Verse(target.VerseId, "cultmesh-client").Context,
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.ProviderRuntimeId));
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.AuthorityRuntimeId));
}
public CultMeshDocumentHandle<TDocument> Handle { get; }
@ -897,7 +903,7 @@ namespace GameCult.Mesh
_subscription = new CultNetDatabaseSubscriptionClient(session.OpenSchemaClient(), _cache, networkRegistry);
Handle = CultMesh.Collection<TDocument>(
_cache,
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.ProviderRuntimeId));
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.AuthorityRuntimeId));
}
public CultMeshCollectionHandle<TDocument> Handle { get; }

View file

@ -122,31 +122,56 @@ namespace GameCult.Mesh
public async Task<ICultMeshContentTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
if (!CanConnect(candidate))
throw new NotSupportedException($"Legacy RUDP content connector does not support '{candidate.Endpoint}'.");
var client = await _schemaConnector.ConnectAsync(candidate, CultMeshProtocols.Content, cancellationToken)
.ConfigureAwait(false);
return new LegacyRudpContentTransport(candidate.Endpoint, client, _options);
try
{
await CultMeshSessionIdentityClient.VerifyAsync(
client,
"cultmesh-content-client",
target,
CultMeshProtocols.Content,
candidate.Generation,
_options.ResponseTimeout,
cancellationToken).ConfigureAwait(false);
return new LegacyRudpContentTransport(
candidate.Endpoint, client, _options, target, candidate.Generation);
}
catch
{
client.Dispose();
throw;
}
}
private sealed class LegacyRudpContentTransport : ICultMeshContentTransport
private sealed class LegacyRudpContentTransport : ICultMeshContentTransport, ICultMeshVerifiedTransport
{
private readonly ICultNetSchemaClient _client;
private readonly CultMeshLegacyRudpContentTransportOptions _options;
private readonly string _verseId;
private readonly string _authorityRuntimeId;
private readonly string _routeGeneration;
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, TaskCompletionSource<CultMeshContentChunkResponseMessage>> _pending = new(StringComparer.Ordinal);
private bool _disposed;
public LegacyRudpContentTransport(
string endpoint,
ICultNetSchemaClient client,
CultMeshLegacyRudpContentTransportOptions options)
CultMeshLegacyRudpContentTransportOptions options,
CultMeshSessionTarget target,
string routeGeneration)
{
Endpoint = endpoint;
_client = client;
_options = options;
_verseId = target.VerseId;
_authorityRuntimeId = target.AuthorityRuntimeId;
_routeGeneration = routeGeneration ?? string.Empty;
_client.OnCultNet<CultMeshContentChunkResponseMessage>(OnResponse);
_client.OnCultNet<CultNetErrorMessage>(OnError);
}
@ -154,6 +179,16 @@ namespace GameCult.Mesh
public string TransportId => "legacy-rudp-content";
public string Endpoint { get; }
public bool IsVerifiedFor(
string verseId,
string authorityRuntimeId,
string protocolId,
string routeGeneration) =>
string.Equals(_verseId, verseId, StringComparison.Ordinal) &&
string.Equals(_authorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) &&
string.Equals(CultMeshProtocols.Content.Value, protocolId, StringComparison.Ordinal) &&
string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal);
public async Task CopyChunkToAsync(
CultMeshCdnChunkRef chunk,
Stream destination,

View file

@ -36,7 +36,7 @@ namespace GameCult.Mesh
bool CanConnect(CultMeshTransportCandidate candidate);
Task<ICultMeshContentTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default);
}

View file

@ -267,7 +267,8 @@ namespace GameCult.Mesh
var observations = results.SelectMany(result => result.Observations)
.Where(observation => query.VerseIds.Count == 0 || query.VerseIds.Contains(observation.Candidate.VerseId, StringComparer.Ordinal))
.Where(observation => query.AuthorityRuntimeId == null ||
observation.Candidate.AuthorityRuntimeIds.Contains(query.AuthorityRuntimeId, StringComparer.Ordinal))
observation.Candidate.AuthorityRoutes.Any(route =>
string.Equals(route.AuthorityRuntimeId, query.AuthorityRuntimeId, StringComparison.Ordinal)))
.ToArray();
foreach (var rejected in observations.Where(observation => observation.Trust == CultMeshDiscoveryTrust.Rejected))

View file

@ -59,7 +59,7 @@ namespace GameCult.Mesh
bool CanConnect(CultMeshTransportCandidate candidate);
Task<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default);
}

View file

@ -0,0 +1,213 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using GameCult.Networking;
namespace GameCult.Mesh
{
/// <summary>
/// Optional transport proof surfaced by connectors that already authenticate
/// the peer during their native handshake. Other schema clients must complete
/// the portable CultMesh handshake before use.
/// </summary>
public interface ICultMeshVerifiedSchemaClient
{
bool IsVerifiedFor(
string verseId,
string authorityRuntimeId,
string protocolId,
string routeGeneration);
}
/// <summary>
/// Proof exposed by content and realtime transports after their native or
/// portable handshake has bound the physical channel to an Odin route.
/// </summary>
public interface ICultMeshVerifiedTransport
{
bool IsVerifiedFor(
string verseId,
string authorityRuntimeId,
string protocolId,
string routeGeneration);
}
internal static class CultMeshTransportIdentity
{
public static void RequireVerified(
object transport,
CultMeshSessionTarget target,
CultMeshProtocolId protocol,
CultMeshTransportCandidate candidate)
{
if (transport is ICultMeshVerifiedTransport verified &&
verified.IsVerifiedFor(
target.VerseId,
target.AuthorityRuntimeId,
protocol.Value,
candidate.Generation))
return;
throw new CultMeshSessionException(new CultMeshSessionFailure(
CultMeshSessionFailureReason.Authority,
$"Transport did not prove CultMesh authority '{target}' for protocol '{protocol}'.",
candidate.Endpoint));
}
}
/// <summary>
/// Owns the server side of the transport-neutral CultMesh session identity
/// handshake. Application messages are not authority proof; providers attach
/// this gate to every schema transport they advertise.
/// </summary>
public sealed class CultMeshSessionIdentityServer : IDisposable
{
private readonly ICultNetSchemaServer _server;
private readonly string _authorityRuntimeId;
private readonly HashSet<string> _verseIds;
private readonly HashSet<string> _protocolIds;
private readonly HashSet<string> _routeGenerations;
private readonly Func<CultMeshSessionOpenMessage, ICultNetSchemaServerPeer, Task> _handler;
private bool _disposed;
public CultMeshSessionIdentityServer(
ICultNetSchemaServer server,
string authorityRuntimeId,
IEnumerable<string> verseIds,
IEnumerable<string> protocolIds,
IEnumerable<string> routeGenerations)
{
_server = server ?? throw new ArgumentNullException(nameof(server));
_authorityRuntimeId = Require(authorityRuntimeId, nameof(authorityRuntimeId));
_verseIds = Clean(verseIds, nameof(verseIds));
_protocolIds = Clean(protocolIds, nameof(protocolIds));
_routeGenerations = Clean(routeGenerations, nameof(routeGenerations));
_handler = HandleAsync;
_server.OnCultNet(_handler);
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_server.RemoveCultNetMessageListener<CultMeshSessionOpenMessage>(_handler);
}
private Task HandleAsync(CultMeshSessionOpenMessage request, ICultNetSchemaServerPeer peer)
{
var error = Validate(request);
peer.SendCultNet(new CultMeshSessionAcceptedMessage
{
MessageId = request.MessageId ?? string.Empty,
Accepted = error == null,
VerseId = request.VerseId ?? string.Empty,
AuthorityRuntimeId = _authorityRuntimeId,
ProtocolId = request.ProtocolId ?? string.Empty,
RouteGeneration = request.RouteGeneration ?? string.Empty,
Error = error
});
return Task.CompletedTask;
}
private string? Validate(CultMeshSessionOpenMessage request)
{
if (string.IsNullOrWhiteSpace(request.MessageId)) return "session-message-id-required";
if (string.IsNullOrWhiteSpace(request.SourceRuntimeId)) return "source-runtime-id-required";
if (!string.Equals(request.AuthorityRuntimeId, _authorityRuntimeId, StringComparison.Ordinal))
return "target-runtime-mismatch";
if (!_verseIds.Contains(request.VerseId ?? string.Empty)) return "verse-not-served";
if (!_protocolIds.Contains(request.ProtocolId ?? string.Empty)) return "protocol-not-served";
if (!_routeGenerations.Contains(request.RouteGeneration ?? string.Empty)) return "route-generation-not-served";
return null;
}
private static HashSet<string> Clean(IEnumerable<string> values, string name)
{
if (values == null) throw new ArgumentNullException(name);
var result = new HashSet<string>(
values.Where(value => !string.IsNullOrWhiteSpace(value)).Select(value => value.Trim()),
StringComparer.Ordinal);
if (result.Count == 0) throw new ArgumentException("At least one identity is required.", name);
return result;
}
private static string Require(string value, string name) =>
string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", name)
: value.Trim();
}
internal static class CultMeshSessionIdentityClient
{
public static async Task VerifyAsync(
ICultNetSchemaClient client,
string sourceRuntimeId,
CultMeshSessionTarget target,
CultMeshProtocolId protocol,
string routeGeneration,
TimeSpan timeout,
CancellationToken cancellationToken = default)
{
if (client == null) throw new ArgumentNullException(nameof(client));
if (client is ICultMeshVerifiedSchemaClient verified)
{
if (!verified.IsVerifiedFor(
target.VerseId,
target.AuthorityRuntimeId,
protocol.Value,
routeGeneration ?? string.Empty))
{
throw new CultMeshSessionException(new CultMeshSessionFailure(
CultMeshSessionFailureReason.Authority,
$"Native transport identity did not match CultMesh target '{target}' for '{protocol}'."));
}
return;
}
var messageId = Guid.NewGuid().ToString("N");
var completion = new TaskCompletionSource<CultMeshSessionAcceptedMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
client.OnCultNet<CultMeshSessionAcceptedMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
completion.TrySetResult(response);
});
client.SendCultNet(new CultMeshSessionOpenMessage
{
MessageId = messageId,
SourceRuntimeId = sourceRuntimeId,
VerseId = target.VerseId,
AuthorityRuntimeId = target.AuthorityRuntimeId,
ProtocolId = protocol.Value,
RouteGeneration = routeGeneration ?? string.Empty
});
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
deadline.CancelAfter(timeout);
var cancelled = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
using (deadline.Token.Register(() => cancelled.TrySetResult(true)))
{
if (await Task.WhenAny(completion.Task, cancelled.Task).ConfigureAwait(false) != completion.Task)
{
cancellationToken.ThrowIfCancellationRequested();
throw new CultMeshSessionException(new CultMeshSessionFailure(
CultMeshSessionFailureReason.Timeout,
$"Timed out verifying authority runtime '{target.AuthorityRuntimeId}'."));
}
}
var accepted = await completion.Task.ConfigureAwait(false);
if (!accepted.Accepted ||
!string.Equals(accepted.VerseId, target.VerseId, StringComparison.Ordinal) ||
!string.Equals(accepted.AuthorityRuntimeId, target.AuthorityRuntimeId, StringComparison.Ordinal) ||
!string.Equals(accepted.ProtocolId, protocol.Value, StringComparison.Ordinal) ||
!string.Equals(accepted.RouteGeneration, routeGeneration ?? string.Empty, StringComparison.Ordinal))
{
throw new CultMeshSessionException(new CultMeshSessionFailure(
CultMeshSessionFailureReason.Authority,
$"Route did not prove CultMesh authority '{target}' for protocol '{protocol}': " +
(accepted.Error ?? "identity-mismatch")));
}
}
}
}

View file

@ -9,14 +9,14 @@ using R3;
namespace GameCult.Mesh
{
public sealed class CultMeshEndpointId : IEquatable<CultMeshEndpointId>
public sealed class CultMeshRuntimeId : IEquatable<CultMeshRuntimeId>
{
private CultMeshEndpointId(string value) { Value = value; }
private CultMeshRuntimeId(string value) { Value = value; }
public string Value { get; }
public static CultMeshEndpointId Parse(string value) => new CultMeshEndpointId(
string.IsNullOrWhiteSpace(value) ? throw new FormatException("CultMesh endpoint identity must be non-empty.") : value.Trim());
public bool Equals(CultMeshEndpointId? other) => other != null && string.Equals(Value, other.Value, StringComparison.Ordinal);
public override bool Equals(object? obj) => Equals(obj as CultMeshEndpointId);
public static CultMeshRuntimeId Parse(string value) => new CultMeshRuntimeId(
string.IsNullOrWhiteSpace(value) ? throw new FormatException("CultMesh runtime identity must be non-empty.") : value.Trim());
public bool Equals(CultMeshRuntimeId? other) => other != null && string.Equals(Value, other.Value, StringComparison.Ordinal);
public override bool Equals(object? obj) => Equals(obj as CultMeshRuntimeId);
public override int GetHashCode() => StringComparer.Ordinal.GetHashCode(Value);
public override string ToString() => Value;
}
@ -28,31 +28,32 @@ namespace GameCult.Mesh
public sealed class CultMeshSessionTarget : IEquatable<CultMeshSessionTarget>
{
/// <summary>Creates one stable provider target inside one Verse.</summary>
public CultMeshSessionTarget(string verseId, string providerRuntimeId)
public CultMeshSessionTarget(string verseId, string authorityRuntimeId)
{
VerseId = string.IsNullOrWhiteSpace(verseId)
? throw new ArgumentException("Verse identity is required.", nameof(verseId))
: verseId.Trim();
ProviderEndpointId = CultMeshEndpointId.Parse(providerRuntimeId);
ProviderRuntimeId = ProviderEndpointId.Value;
AuthorityRuntimeId = CultMeshRuntimeId.Parse(authorityRuntimeId).Value;
}
/// <summary>Gets the stable Verse identity.</summary>
public string VerseId { get; }
/// <summary>Gets the stable provider runtime identity advertised by that Verse.</summary>
public string ProviderRuntimeId { get; }
internal CultMeshEndpointId ProviderEndpointId { get; }
internal string SessionKey => VerseId + "\u001f" + ProviderRuntimeId;
/// <summary>Gets the stable authority runtime identity advertised by that Verse.</summary>
public string AuthorityRuntimeId { get; }
/// <summary>Compatibility projection. Use <see cref="AuthorityRuntimeId"/>.</summary>
[Obsolete("Use AuthorityRuntimeId. Provider identity is not a physical endpoint.")]
public string ProviderRuntimeId => AuthorityRuntimeId;
internal string SessionKey => VerseId + "\u001f" + AuthorityRuntimeId;
public bool Equals(CultMeshSessionTarget? other) =>
other != null &&
string.Equals(VerseId, other.VerseId, StringComparison.Ordinal) &&
string.Equals(ProviderRuntimeId, other.ProviderRuntimeId, StringComparison.Ordinal);
string.Equals(AuthorityRuntimeId, other.AuthorityRuntimeId, StringComparison.Ordinal);
public override bool Equals(object? obj) => Equals(obj as CultMeshSessionTarget);
public override int GetHashCode() => HashCode.Combine(
StringComparer.Ordinal.GetHashCode(VerseId),
StringComparer.Ordinal.GetHashCode(ProviderRuntimeId));
public override string ToString() => VerseId + "/" + ProviderRuntimeId;
StringComparer.Ordinal.GetHashCode(AuthorityRuntimeId));
public override string ToString() => VerseId + "/" + AuthorityRuntimeId;
}
public sealed class CultMeshProtocolId : IEquatable<CultMeshProtocolId>
@ -80,15 +81,24 @@ namespace GameCult.Mesh
public sealed class CultMeshTransportCandidate
{
public CultMeshTransportCandidate(string endpoint, CultMeshTransportPathKind pathKind = CultMeshTransportPathKind.Direct, int priority = 0)
public CultMeshTransportCandidate(
string endpoint,
CultMeshTransportPathKind pathKind = CultMeshTransportPathKind.Direct,
int priority = 0,
string authorityRuntimeId = "",
string generation = "")
{
Endpoint = string.IsNullOrWhiteSpace(endpoint) ? throw new ArgumentException("Physical endpoint is required.", nameof(endpoint)) : endpoint;
PathKind = pathKind;
Priority = priority;
AuthorityRuntimeId = authorityRuntimeId?.Trim() ?? string.Empty;
Generation = generation?.Trim() ?? string.Empty;
}
public string Endpoint { get; }
public CultMeshTransportPathKind PathKind { get; }
public int Priority { get; }
public string AuthorityRuntimeId { get; }
public string Generation { get; }
}
public enum CultMeshSessionStatus { Connecting, Online, Degraded, Reconnecting, Offline }
@ -522,6 +532,8 @@ namespace GameCult.Mesh
public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance;
public ICultMeshDiagnosticSink Diagnostics { get; set; } = CultMeshNullDiagnosticSink.Instance;
public int MaxRacedCandidates { get; set; } = 2;
public string RuntimeId { get; set; } = "cultmesh-client";
public TimeSpan IdentityHandshakeTimeout { get; set; } = TimeSpan.FromSeconds(5);
}
public sealed class CultMeshSessionManager : IDisposable
@ -581,6 +593,8 @@ namespace GameCult.Mesh
throw new ArgumentException("Realtime transport connectors cannot contain null entries.", nameof(realtimeConnectors));
_options = options ?? new CultMeshSessionManagerOptions();
if (_options.MaxRacedCandidates <= 0) throw new ArgumentOutOfRangeException(nameof(options));
if (string.IsNullOrWhiteSpace(_options.RuntimeId)) throw new ArgumentException("Client runtime identity is required.", nameof(options));
if (_options.IdentityHandshakeTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(options));
}
public async Task<CultMeshSession> ConnectAsync(
@ -661,14 +675,10 @@ namespace GameCult.Mesh
"No streaming content connectors are configured. Register the TCP content connector or the explicit legacy RUDP connector.");
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
target.ProviderRuntimeId,
target.AuthorityRuntimeId,
new[] { target.VerseId },
target.ProviderRuntimeId)).ConfigureAwait(false);
var candidates = discovery.Candidates
.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
.Select(endpoint => new CultMeshTransportCandidate(endpoint))
.ToArray();
target.AuthorityRuntimeId)).ConfigureAwait(false);
var candidates = BoundCandidates(discovery, target, CultMeshProtocols.Content);
var routes = candidates
.SelectMany(candidate => _contentConnectors
.Where(connector => connector.CanConnect(candidate))
@ -685,7 +695,7 @@ namespace GameCult.Mesh
foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key))
{
var attempts = tier.Take(_options.MaxRacedCandidates)
.Select(route => ConnectContentCandidateAsync(target.ProviderEndpointId, route))
.Select(route => ConnectContentCandidateAsync(target, route))
.ToList();
while (attempts.Count > 0)
{
@ -709,18 +719,33 @@ namespace GameCult.Mesh
}
var last = failures.LastOrDefault();
var reason = last is CultMeshSessionException typed
? typed.Failure.Reason
: last is TimeoutException
? CultMeshSessionFailureReason.Timeout
: CultMeshSessionFailureReason.Transport;
throw Failure(
last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport,
reason,
$"No streaming content path connected for '{target}'.",
last);
}
private static async Task<ConnectedContentPath> ConnectContentCandidateAsync(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
ContentRoute route)
{
var transport = await route.Connector.ConnectAsync(route.Candidate, endpointId).ConfigureAwait(false);
return new ConnectedContentPath(route.Candidate, transport);
var transport = await route.Connector.ConnectAsync(route.Candidate, target).ConfigureAwait(false);
try
{
CultMeshTransportIdentity.RequireVerified(
transport, target, CultMeshProtocols.Content, route.Candidate);
return new ConnectedContentPath(route.Candidate, transport);
}
catch
{
transport.Dispose();
throw;
}
}
private void InvalidateContentSession(string key, CultMeshContentSession session)
@ -790,13 +815,10 @@ namespace GameCult.Mesh
"No realtime state connectors are configured. Register a QUIC connector explicitly.");
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
target.ProviderRuntimeId,
target.AuthorityRuntimeId,
new[] { target.VerseId },
target.ProviderRuntimeId)).ConfigureAwait(false);
var routes = discovery.Candidates
.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
.Select(endpoint => new CultMeshTransportCandidate(endpoint))
target.AuthorityRuntimeId)).ConfigureAwait(false);
var routes = BoundCandidates(discovery, target, CultMeshProtocols.RealtimeState)
.SelectMany(candidate => _realtimeConnectors
.Where(connector => connector.CanConnect(candidate))
.Select(connector => new RealtimeRoute(candidate, connector)))
@ -812,7 +834,7 @@ namespace GameCult.Mesh
foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key))
{
var attempts = tier.Take(_options.MaxRacedCandidates)
.Select(route => ConnectRealtimeCandidateAsync(target.ProviderEndpointId, route))
.Select(route => ConnectRealtimeCandidateAsync(target, route))
.ToList();
while (attempts.Count > 0)
{
@ -836,18 +858,33 @@ namespace GameCult.Mesh
}
var last = failures.LastOrDefault();
var reason = last is CultMeshSessionException typed
? typed.Failure.Reason
: last is TimeoutException
? CultMeshSessionFailureReason.Timeout
: CultMeshSessionFailureReason.Transport;
throw Failure(
last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport,
reason,
$"No realtime state path connected for '{target}'.",
last);
}
private static async Task<ConnectedRealtimePath> ConnectRealtimeCandidateAsync(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
RealtimeRoute route)
{
var transport = await route.Connector.ConnectAsync(route.Candidate, endpointId).ConfigureAwait(false);
return new ConnectedRealtimePath(route.Candidate, transport);
var transport = await route.Connector.ConnectAsync(route.Candidate, target).ConfigureAwait(false);
try
{
CultMeshTransportIdentity.RequireVerified(
transport, target, CultMeshProtocols.RealtimeState, route.Candidate);
return new ConnectedRealtimePath(route.Candidate, transport);
}
catch
{
transport.Dispose();
throw;
}
}
private void InvalidateRealtimeSession(string key, CultMeshRealtimeSession session)
@ -880,13 +917,10 @@ namespace GameCult.Mesh
private async Task<ConnectedPath> ResolveConnectedPathAsync(CultMeshSessionTarget target, CultMeshProtocolId protocol)
{
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
target.ProviderRuntimeId,
target.AuthorityRuntimeId,
new[] { target.VerseId },
target.ProviderRuntimeId)).ConfigureAwait(false);
var candidates = discovery.Candidates.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
.Select(endpoint => new CultMeshTransportCandidate(endpoint))
.ToArray();
target.AuthorityRuntimeId)).ConfigureAwait(false);
var candidates = BoundCandidates(discovery, target, protocol);
var routes = candidates
.SelectMany(candidate => _connectors
.Where(connector => connector.CanConnect(candidate))
@ -908,7 +942,7 @@ namespace GameCult.Mesh
foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key))
{
var attempts = tier.Take(_options.MaxRacedCandidates)
.Select(route => ConnectCandidateAsync(route, protocol))
.Select(route => ConnectCandidateAsync(target, route, protocol))
.ToList();
while (attempts.Count > 0)
{
@ -925,7 +959,12 @@ namespace GameCult.Mesh
}
}
var last = failures.LastOrDefault();
throw Failure(last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport,
var reason = last is CultMeshSessionException typed
? typed.Failure.Reason
: last is TimeoutException
? CultMeshSessionFailureReason.Timeout
: CultMeshSessionFailureReason.Transport;
throw Failure(reason,
$"No transport path connected for '{target}' and protocol '{protocol}'.", last);
}
@ -960,9 +999,9 @@ namespace GameCult.Mesh
try
{
_discovery.Invalidate(new CultMeshDiscoveryQuery(
session.Target.ProviderRuntimeId,
session.Target.AuthorityRuntimeId,
new[] { session.Target.VerseId },
session.Target.ProviderRuntimeId));
session.Target.AuthorityRuntimeId));
var replacement = await ResolveConnectedPathAsync(session.Target, session.Protocol).ConfigureAwait(false);
session.ReplacePhysicalChannel(replacement.Client);
session.Transition(new CultMeshSessionState(
@ -987,10 +1026,28 @@ namespace GameCult.Mesh
}
}
private static async Task<ConnectedPath> ConnectCandidateAsync(SchemaRoute route, CultMeshProtocolId protocol)
private async Task<ConnectedPath> ConnectCandidateAsync(
CultMeshSessionTarget target,
SchemaRoute route,
CultMeshProtocolId protocol)
{
var client = await route.Connector.ConnectAsync(route.Candidate, protocol).ConfigureAwait(false);
return new ConnectedPath(route.Candidate, client);
try
{
await CultMeshSessionIdentityClient.VerifyAsync(
client,
_options.RuntimeId,
target,
protocol,
route.Candidate.Generation,
_options.IdentityHandshakeTimeout).ConfigureAwait(false);
return new ConnectedPath(route.Candidate, client);
}
catch
{
client.Dispose();
throw;
}
}
private static async Task<T> AwaitForCallerAsync<T>(Task<T> shared, CancellationToken cancellationToken)
@ -1008,12 +1065,36 @@ namespace GameCult.Mesh
private CultMeshSessionException Failure(CultMeshSessionFailureReason reason, string message, Exception? inner = null) =>
new CultMeshSessionException(new CultMeshSessionFailure(reason, message), inner);
private static CultMeshTransportCandidate[] BoundCandidates(
CultMeshDiscoveryState discovery,
CultMeshSessionTarget target,
CultMeshProtocolId protocol)
{
return discovery.Candidates
.SelectMany(candidate => candidate.Descriptor.AuthorityRoutes)
.Where(route => string.Equals(
route.AuthorityRuntimeId,
target.AuthorityRuntimeId,
StringComparison.Ordinal))
.Where(route => route.Supports(protocol))
.GroupBy(route => route.Endpoint, StringComparer.Ordinal)
.Select(group => group.OrderBy(route => route.Priority).First())
.Select(route => new CultMeshTransportCandidate(
route.Endpoint,
priority: route.Priority,
authorityRuntimeId: route.AuthorityRuntimeId,
generation: route.Generation))
.OrderBy(candidate => candidate.Priority)
.ThenBy(candidate => candidate.Endpoint, StringComparer.Ordinal)
.ToArray();
}
private void Emit(CultMeshSessionTarget target, CultMeshProtocolId protocol, string state, string path)
{
_options.Diagnostics.Emit(new CultMeshDiagnosticEvent(
Interlocked.Increment(ref _diagnosticSequence), _options.Clock.UtcNow,
CultMeshReliabilityOrgan.Session, CultMeshDiagnosticKind.PathChanged,
target.SessionKey + ":" + protocol.Value, target.ProviderRuntimeId, state,
target.SessionKey + ":" + protocol.Value, target.AuthorityRuntimeId, state,
sourceId: "session-manager", endpoint: path));
}

View file

@ -102,7 +102,7 @@ namespace GameCult.Mesh
CultNetDocumentRegistry? registry)
{
_sharedSession = session ?? throw new ArgumentNullException(nameof(session));
_endpoint = session.State.Path?.Endpoint ?? session.Target.ProviderRuntimeId;
_endpoint = session.State.Path?.Endpoint ?? session.Target.AuthorityRuntimeId;
_defaults = CultMesh.CloneSnapshotRequestOptions(options);
_registry = registry ?? new CultNetDocumentRegistry();
_client = session.Channel;

View file

@ -11,6 +11,21 @@ using GameCult.Networking;
namespace GameCult.Mesh
{
/// <summary>Identity served by one dedicated content listener.</summary>
public sealed class CultMeshTcpContentServerOptions
{
public string VerseId { get; set; } = string.Empty;
public string AuthorityRuntimeId { get; set; } = string.Empty;
public string RouteGeneration { get; set; } = string.Empty;
internal void Validate()
{
if (string.IsNullOrWhiteSpace(VerseId)) throw new InvalidOperationException("Content Verse identity is required.");
if (string.IsNullOrWhiteSpace(AuthorityRuntimeId)) throw new InvalidOperationException("Content authority runtime identity is required.");
if (string.IsNullOrWhiteSpace(RouteGeneration)) throw new InvalidOperationException("Content route generation is required.");
}
}
/// <summary>Creates dedicated TCP streams for immutable CultMesh content.</summary>
public sealed class CultMeshTcpContentTransportConnector : ICultMeshContentTransportConnector
{
@ -33,11 +48,11 @@ namespace GameCult.Mesh
public async Task<ICultMeshContentTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
if (candidate == null) throw new ArgumentNullException(nameof(candidate));
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (!TryParseEndpoint(candidate.Endpoint, out var host, out var port))
throw new NotSupportedException($"TCP content connector does not support '{candidate.Endpoint}'.");
@ -48,8 +63,10 @@ namespace GameCult.Mesh
{
await CultMeshTcpContentWire.AwaitAsync(client.ConnectAsync(host, port), timeout.Token)
.ConfigureAwait(false);
await CultMeshTcpContentWire.VerifyAuthorityAsync(
client.GetStream(), target, candidate, timeout.Token).ConfigureAwait(false);
Interlocked.Increment(ref _connectCount);
return new CultMeshTcpContentTransport(candidate.Endpoint, client);
return new CultMeshTcpContentTransport(candidate.Endpoint, client, target, candidate.Generation);
}
catch
{
@ -72,7 +89,7 @@ namespace GameCult.Mesh
}
}
internal sealed class CultMeshTcpContentTransport : ICultMeshContentTransport
internal sealed class CultMeshTcpContentTransport : ICultMeshContentTransport, ICultMeshVerifiedTransport
{
private readonly TcpClient _client;
private readonly NetworkStream _stream;
@ -80,12 +97,22 @@ namespace GameCult.Mesh
private readonly ConcurrentDictionary<string, PendingCopy> _pending = new(StringComparer.Ordinal);
private readonly CancellationTokenSource _shutdown = new();
private readonly Task _receiveLoop;
private readonly string _verseId;
private readonly string _authorityRuntimeId;
private readonly string _routeGeneration;
private bool _disposed;
public CultMeshTcpContentTransport(string endpoint, TcpClient client)
public CultMeshTcpContentTransport(
string endpoint,
TcpClient client,
CultMeshSessionTarget target,
string routeGeneration)
{
Endpoint = endpoint;
_client = client ?? throw new ArgumentNullException(nameof(client));
_verseId = target?.VerseId ?? throw new ArgumentNullException(nameof(target));
_authorityRuntimeId = target.AuthorityRuntimeId;
_routeGeneration = routeGeneration ?? string.Empty;
_stream = client.GetStream();
_receiveLoop = ReceiveLoopAsync(_shutdown.Token);
}
@ -93,6 +120,16 @@ namespace GameCult.Mesh
public string TransportId => "tcp-content";
public string Endpoint { get; }
public bool IsVerifiedFor(
string verseId,
string authorityRuntimeId,
string protocolId,
string routeGeneration) =>
string.Equals(_verseId, verseId, StringComparison.Ordinal) &&
string.Equals(_authorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) &&
string.Equals(CultMeshProtocols.Content.Value, protocolId, StringComparison.Ordinal) &&
string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal);
public async Task CopyChunkToAsync(
CultMeshCdnChunkRef chunk,
Stream destination,
@ -212,6 +249,7 @@ namespace GameCult.Mesh
private readonly TcpListener _listener;
private readonly Func<string, CultMeshCdnArtifactChunk?> _resolveChunk;
private readonly Func<string, bool> _canServeHash;
private readonly CultMeshTcpContentServerOptions _options;
private readonly CancellationTokenSource _shutdown = new();
private readonly ConcurrentDictionary<TcpClient, byte> _clients = new();
private readonly Task _acceptLoop;
@ -221,8 +259,9 @@ namespace GameCult.Mesh
public CultMeshTcpContentServer(
TcpListener listener,
CultCache content,
CultMeshTcpContentServerOptions options,
Func<string, bool>? canServeHash = null)
: this(listener, CreateCacheResolver(content), canServeHash)
: this(listener, CreateCacheResolver(content), options, canServeHash)
{
}
@ -234,10 +273,13 @@ namespace GameCult.Mesh
public CultMeshTcpContentServer(
TcpListener listener,
Func<string, CultMeshCdnArtifactChunk?> resolveChunk,
CultMeshTcpContentServerOptions options,
Func<string, bool>? canServeHash = null)
{
_listener = listener ?? throw new ArgumentNullException(nameof(listener));
_resolveChunk = resolveChunk ?? throw new ArgumentNullException(nameof(resolveChunk));
_options = options ?? throw new ArgumentNullException(nameof(options));
_options.Validate();
_canServeHash = canServeHash ?? (_ => true);
_listener.Start();
_acceptLoop = AcceptLoopAsync(_shutdown.Token);
@ -278,6 +320,7 @@ namespace GameCult.Mesh
try
{
using var stream = client.GetStream();
if (!await AcceptAuthorityAsync(stream, cancellationToken).ConfigureAwait(false)) return;
while (!cancellationToken.IsCancellationRequested)
{
var requestBytes = await CultMeshTcpContentWire.ReadFrameAsync(stream, cancellationToken).ConfigureAwait(false);
@ -296,6 +339,32 @@ namespace GameCult.Mesh
}
}
private async Task<bool> AcceptAuthorityAsync(Stream stream, CancellationToken cancellationToken)
{
var requestBytes = await CultMeshTcpContentWire.ReadFrameAsync(stream, cancellationToken).ConfigureAwait(false);
var request = CultNetSchemaMessageSerialization.Deserialize(requestBytes) as CultMeshSessionOpenMessage
?? throw new InvalidDataException("TCP content session did not begin with an authority handshake.");
var accepted = string.Equals(request.VerseId, _options.VerseId, StringComparison.Ordinal) &&
string.Equals(request.AuthorityRuntimeId, _options.AuthorityRuntimeId, StringComparison.Ordinal) &&
string.Equals(request.ProtocolId, CultMeshProtocols.Content.Value, StringComparison.Ordinal) &&
string.Equals(request.RouteGeneration, _options.RouteGeneration, StringComparison.Ordinal);
var response = new CultMeshSessionAcceptedMessage
{
MessageId = request.MessageId,
Accepted = accepted,
VerseId = _options.VerseId,
AuthorityRuntimeId = _options.AuthorityRuntimeId,
ProtocolId = CultMeshProtocols.Content.Value,
RouteGeneration = _options.RouteGeneration,
Error = accepted ? null : "Content session target does not match this authority route."
};
await CultMeshTcpContentWire.WriteFrameAsync(
stream,
CultNetSchemaMessageSerialization.Serialize(response),
cancellationToken).ConfigureAwait(false);
return accepted;
}
private async Task SendChunkAsync(
Stream stream,
CultMeshContentChunkRequestMessage request,
@ -391,6 +460,38 @@ namespace GameCult.Mesh
await stream.FlushAsync(cancellationToken).ConfigureAwait(false);
}
public static async Task VerifyAuthorityAsync(
Stream stream,
CultMeshSessionTarget target,
CultMeshTransportCandidate candidate,
CancellationToken cancellationToken)
{
var request = new CultMeshSessionOpenMessage
{
MessageId = Guid.NewGuid().ToString("N"),
SourceRuntimeId = "cultmesh-content-client",
VerseId = target.VerseId,
AuthorityRuntimeId = target.AuthorityRuntimeId,
ProtocolId = CultMeshProtocols.Content.Value,
RouteGeneration = candidate.Generation
};
await WriteFrameAsync(stream, CultNetSchemaMessageSerialization.Serialize(request), cancellationToken)
.ConfigureAwait(false);
var responseBytes = await ReadFrameAsync(stream, cancellationToken).ConfigureAwait(false);
var response = CultNetSchemaMessageSerialization.Deserialize(responseBytes) as CultMeshSessionAcceptedMessage
?? throw new InvalidDataException("TCP content authority handshake returned an unexpected schema.");
if (!string.Equals(response.MessageId, request.MessageId, StringComparison.Ordinal) ||
!response.Accepted ||
!string.Equals(response.VerseId, target.VerseId, StringComparison.Ordinal) ||
!string.Equals(response.AuthorityRuntimeId, target.AuthorityRuntimeId, StringComparison.Ordinal) ||
!string.Equals(response.ProtocolId, CultMeshProtocols.Content.Value, StringComparison.Ordinal) ||
!string.Equals(response.RouteGeneration, candidate.Generation, StringComparison.Ordinal))
throw new CultMeshSessionException(new CultMeshSessionFailure(
CultMeshSessionFailureReason.Authority,
response.Error ?? "TCP content endpoint did not prove the selected authority route.",
candidate.Endpoint));
}
public static async Task<byte[]> ReadFrameAsync(Stream stream, CancellationToken cancellationToken)
{
var header = new byte[HeaderBytes];

View file

@ -91,6 +91,53 @@ namespace GameCult.Mesh
}
}
/// <summary>
/// Binds one authoritative runtime to one physical route. This relationship is
/// the routing authority; the legacy endpoint/runtime lists are projections only.
/// </summary>
public sealed class CultMeshAuthorityRoute
{
public CultMeshAuthorityRoute(
string authorityRuntimeId,
string endpoint,
IEnumerable<string>? protocolIds = null,
int priority = 0,
string? generation = null)
{
AuthorityRuntimeId = RequireNonEmpty(authorityRuntimeId, nameof(authorityRuntimeId));
Endpoint = RequireNonEmpty(endpoint, nameof(endpoint));
ProtocolIds = Clean(protocolIds);
if (priority < 0) throw new ArgumentOutOfRangeException(nameof(priority));
Priority = priority;
Generation = string.IsNullOrWhiteSpace(generation)
? AuthorityRuntimeId + "@" + Endpoint
: generation!.Trim();
}
public string AuthorityRuntimeId { get; }
public string Endpoint { get; }
public IReadOnlyList<string> ProtocolIds { get; }
public int Priority { get; }
public string Generation { get; }
public bool Supports(CultMeshProtocolId protocol) =>
protocol != null &&
(ProtocolIds.Count == 0 || ProtocolIds.Contains(protocol.Value, StringComparer.Ordinal));
private static string[] Clean(IEnumerable<string>? values) =>
values?.Where(value => !string.IsNullOrWhiteSpace(value))
.Select(value => value.Trim())
.Distinct(StringComparer.Ordinal)
.OrderBy(value => value, StringComparer.Ordinal)
.ToArray()
?? Array.Empty<string>();
private static string RequireNonEmpty(string value, string paramName) =>
string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value.Trim();
}
/// <summary>
/// Describes one CultMesh Verse: a rule-bearing consensus graph.
/// </summary>
@ -107,14 +154,36 @@ namespace GameCult.Mesh
IEnumerable<string>? discoveryEndpoints = null,
IEnumerable<string>? authorityRuntimeIds = null,
string? parentVerseId = null,
string? description = null)
string? description = null,
IEnumerable<CultMeshAuthorityRoute>? authorityRoutes = null)
{
VerseId = RequireNonEmpty(verseId, nameof(verseId));
DisplayName = RequireNonEmpty(displayName, nameof(displayName));
AuthorityModel = authorityModel;
Compatibility = compatibility ?? throw new ArgumentNullException(nameof(compatibility));
DiscoveryEndpoints = Clean(discoveryEndpoints);
AuthorityRuntimeIds = Clean(authorityRuntimeIds);
var legacyEndpoints = Clean(discoveryEndpoints);
var legacyRuntimeIds = Clean(authorityRuntimeIds);
var suppliedRoutes = authorityRoutes?.ToArray();
if (suppliedRoutes?.Any(route => route == null) == true)
throw new ArgumentException("Authority routes cannot contain null.", nameof(authorityRoutes));
if (suppliedRoutes != null && (legacyEndpoints.Length > 0 || legacyRuntimeIds.Length > 0))
throw new ArgumentException(
"Supply bound authority routes or legacy endpoint/runtime lists, never both.",
nameof(authorityRoutes));
if (suppliedRoutes == null && legacyEndpoints.Length > 0 && legacyRuntimeIds.Length > 1)
throw new InvalidOperationException(
"Legacy Verse routing with multiple authority runtimes is ambiguous. " +
"Publish explicit CultMeshAuthorityRoute bindings.");
AuthorityRoutes = NormalizeRoutes(suppliedRoutes ??
(legacyRuntimeIds.Length == 1
? legacyEndpoints.Select(endpoint => new CultMeshAuthorityRoute(legacyRuntimeIds[0], endpoint))
: Array.Empty<CultMeshAuthorityRoute>()));
DiscoveryEndpoints = AuthorityRoutes.Count > 0
? AuthorityRoutes.Select(route => route.Endpoint).Distinct(StringComparer.Ordinal).ToArray()
: legacyEndpoints;
AuthorityRuntimeIds = AuthorityRoutes.Select(route => route.AuthorityRuntimeId)
.Distinct(StringComparer.Ordinal).OrderBy(value => value, StringComparer.Ordinal).ToArray();
ParentVerseId = parentVerseId;
Description = description;
}
@ -143,6 +212,8 @@ namespace GameCult.Mesh
/// Gets known authoritative runtime ids, when authority is cluster-shaped.
/// </summary>
public IReadOnlyList<string> AuthorityRuntimeIds { get; }
/// <summary>Gets exact authority-to-route bindings used for identity-sensitive resolution.</summary>
public IReadOnlyList<CultMeshAuthorityRoute> AuthorityRoutes { get; }
/// <summary>
/// Gets the parent Verse id for subscribed overlays or branches.
/// </summary>
@ -177,10 +248,29 @@ namespace GameCult.Mesh
private static string[] Clean(IEnumerable<string>? values)
{
return values?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray()
return values?.Where(value => !string.IsNullOrWhiteSpace(value))
.Select(value => value.Trim())
.Distinct(StringComparer.Ordinal)
.ToArray()
?? Array.Empty<string>();
}
private static CultMeshAuthorityRoute[] NormalizeRoutes(IEnumerable<CultMeshAuthorityRoute> routes)
{
return routes
.GroupBy(route => new
{
route.AuthorityRuntimeId,
route.Endpoint,
Protocols = string.Join("\u001f", route.ProtocolIds)
})
.Select(group => group.OrderBy(route => route.Priority).First())
.OrderBy(route => route.Priority)
.ThenBy(route => route.AuthorityRuntimeId, StringComparer.Ordinal)
.ThenBy(route => route.Endpoint, StringComparer.Ordinal)
.ToArray();
}
private static string RequireNonEmpty(string value, string paramName)
{
if (string.IsNullOrWhiteSpace(value))

View file

@ -33,6 +33,14 @@ namespace GameCult.Mesh
},
DiscoveryEndpoints = verse.DiscoveryEndpoints.ToArray(),
AuthorityRuntimeIds = verse.AuthorityRuntimeIds.ToArray(),
AuthorityRoutes = verse.AuthorityRoutes.Select(route => new CultMeshAuthorityRouteMessage
{
AuthorityRuntimeId = route.AuthorityRuntimeId,
Endpoint = route.Endpoint,
ProtocolIds = route.ProtocolIds.ToArray(),
Priority = route.Priority,
Generation = route.Generation
}).ToArray(),
ParentVerseId = verse.ParentVerseId,
Description = verse.Description
};
@ -50,6 +58,12 @@ namespace GameCult.Mesh
out var parsed)
? parsed
: CultMeshVerseAuthorityModel.SubscribedOverlay;
var routes = message.AuthorityRoutes?.Select(route => new CultMeshAuthorityRoute(
route.AuthorityRuntimeId,
route.Endpoint,
route.ProtocolIds,
route.Priority,
route.Generation)).ToArray();
return new CultMeshVerseDescriptor(
message.VerseId,
message.DisplayName,
@ -60,10 +74,11 @@ namespace GameCult.Mesh
message.Compatibility.CompatibleVerseIds,
message.Compatibility.RequiredPluginIds,
message.Compatibility.OptionalPluginIds),
message.DiscoveryEndpoints,
message.AuthorityRuntimeIds,
routes is { Length: > 0 } ? null : message.DiscoveryEndpoints,
routes is { Length: > 0 } ? null : message.AuthorityRuntimeIds,
message.ParentVerseId,
message.Description);
message.Description,
routes is { Length: > 0 } ? routes : null);
}
/// <summary>Creates the canonical filtered Verse catalog response for any CultNet transport adapter.</summary>

View file

@ -50,8 +50,12 @@ not the brand surface.
### Verses
`CultMeshVerseDescriptor` describes a rule-bearing consensus graph. It includes
transport compatibility, rules hash, authority model, discovery endpoints,
known authority runtimes, optional parent Verse id, and plugin requirements.
transport compatibility, rules hash, authority model, exact authority-route
bindings, optional parent Verse id, and plugin requirements. Each route binds
an authority runtime to one endpoint, protocol set, priority, and generation.
The old endpoint/runtime arrays are derived projections; a legacy descriptor
with several authority runtimes is rejected because it cannot identify which
runtime owns which route.
`CultMeshVerseCatalog` is the local reactive catalog for discovered Verses.
It can publish discovery updates and find compatible transfer targets.

View file

@ -9,7 +9,8 @@ be equally good at files, realtime state, and same-machine memory access.
registered connector for each traffic class. A connector owns only its physical
connection and byte movement.
**Inputs:** stable Verse/provider identity, advertised physical candidates,
**Inputs:** stable Verse/authority-runtime identity, exact Odin-bound physical
routes with protocol and generation,
connector support and priority, caller cancellation, and locality evidence for
body negotiation.
@ -63,13 +64,22 @@ operation.
transport diagnostics. None of these grant provider authority or become content
identity.
Before a schema, content, or realtime session becomes online, the transport
must prove that the connected peer matches the selected Verse, authority
runtime, protocol, and route generation. TCP schema and content lanes use the
portable CultNet session-open handshake. QUIC uses the exact Odin-bound route
plus its TLS endpoint/certificate proof. A connector that returns bytes without
authority proof is rejected; racing every endpoint in a Verse descriptor is
not route selection.
**Forbidden writers:** applications and renderers do not choose unadvertised
endpoints, implement reconnect loops, or bypass verification. Schema messages
must not carry bulk bodies. RUDP/LiteNetLib cannot become an implicit default
for any traffic class. QUIC does not own schemas or immutable files merely
because it can transport bytes.
**Shared paths:** discovery and identity resolution remain common. Each traffic
**Shared paths:** exact route resolution and peer verification remain common.
Each traffic
class has its own connector set, connection cache, priority tiers, and failure
state. All content connectors feed the same transfer and promotion owner.

View file

@ -21,8 +21,10 @@ The same transport can carry many Verses:
subscribed overlay
- `Compatibility`: transport version, rules hash, compatible Verse ids, and
plugin requirements
- `DiscoveryEndpoints`: endpoints where nodes for the Verse can be found
- `AuthorityRuntimeIds`: known authoritative runtimes when cluster-shaped
- `AuthorityRoutes`: exact authority-runtime, endpoint, protocol, priority, and
generation bindings. This is the routing authority.
- `DiscoveryEndpoints`: read-only endpoint projection for legacy inspection
- `AuthorityRuntimeIds`: read-only authority projection for legacy inspection
- `ParentVerseId`: source Verse for overlays or branches
## Compatibility
@ -57,6 +59,13 @@ static config, cloud discovery, peer gossip, or mod portals. It does not care
where the descriptor came from. That keeps discovery pluggable while the Verse
model stays stable.
An endpoint and an authority id are not two independent facts. A descriptor
with explicit routes binds them. The legacy parallel arrays are accepted only
when exactly one authority runtime exists; a multi-authority legacy descriptor
is ambiguous and fails closed. Session establishment then requires the selected
physical route to prove the exact Verse, authority runtime, protocol, and route
generation before the session becomes online.
Peer exchange is the torrent-shaped layer under Verse discovery. A Verse
catalog says which graph exists; a peer catalog says which nodes can currently
serve or observe parts of that graph. Peer cards can advertise discovery,

View file

@ -164,6 +164,16 @@ namespace GameCult.Networking
if (_disposed) return;
if (request == null) throw new ArgumentNullException(nameof(request));
if (peer == null) throw new ArgumentNullException(nameof(peer));
if (_sourceRuntimeId != null &&
!string.Equals(request.TargetRuntimeId, _sourceRuntimeId, StringComparison.Ordinal))
{
peer.SendCultNet(Failure(
request,
"denied",
"target-runtime-mismatch",
$"Operation targeted runtime '{request.TargetRuntimeId ?? "(missing)"}', not '{_sourceRuntimeId}'."));
return;
}
Route route;
try

View file

@ -72,6 +72,8 @@ namespace GameCult.Networking
CultNetSchemaVersions.SimulationConsensusCandidate => Deserialize<CultNetSimulationConsensusCandidateMessage>(payload),
CultNetSchemaVersions.VerseCatalogRequest => Deserialize<CultMeshVerseCatalogRequestMessage>(payload),
CultNetSchemaVersions.VerseCatalogResponse => Deserialize<CultMeshVerseCatalogResponseMessage>(payload),
CultNetSchemaVersions.SessionOpen => Deserialize<CultMeshSessionOpenMessage>(payload),
CultNetSchemaVersions.SessionAccepted => Deserialize<CultMeshSessionAcceptedMessage>(payload),
CultNetSchemaVersions.PeerExchangeRequest => Deserialize<CultMeshPeerExchangeRequestMessage>(payload),
CultNetSchemaVersions.PeerExchangeResponse => Deserialize<CultMeshPeerExchangeResponseMessage>(payload),
CultNetSchemaVersions.ContentChunkRequest => Deserialize<CultMeshContentChunkRequestMessage>(payload),

View file

@ -135,6 +135,10 @@ namespace GameCult.Networking
/// CultMesh Verse catalog response contract identifier.
/// </summary>
public const string VerseCatalogResponse = "cultmesh.verse_catalog_response.v0";
/// <summary>CultMesh session identity request contract identifier.</summary>
public const string SessionOpen = "cultmesh.session_open.v1";
/// <summary>CultMesh verified session identity response contract identifier.</summary>
public const string SessionAccepted = "cultmesh.session_accepted.v1";
/// <summary>
/// CultMesh peer exchange request contract identifier.
/// </summary>
@ -1179,6 +1183,33 @@ namespace GameCult.Networking
}
}
/// <summary>Requests an identity-bound CultMesh session before application traffic begins.</summary>
[MessagePackObject]
public class CultMeshSessionOpenMessage : ICultNetSchemaMessage
{
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.SessionOpen;
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
[Key("sourceRuntimeId")] public string SourceRuntimeId { get; set; } = string.Empty;
[Key("verseId")] public string VerseId { get; set; } = string.Empty;
[Key("authorityRuntimeId")] public string AuthorityRuntimeId { get; set; } = string.Empty;
[Key("protocolId")] public string ProtocolId { get; set; } = string.Empty;
[Key("routeGeneration")] public string RouteGeneration { get; set; } = string.Empty;
}
/// <summary>Accepts or rejects a CultMesh session with the server's actual identity.</summary>
[MessagePackObject]
public class CultMeshSessionAcceptedMessage : ICultNetSchemaMessage
{
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.SessionAccepted;
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
[Key("accepted")] public bool Accepted { get; set; }
[Key("verseId")] public string VerseId { get; set; } = string.Empty;
[Key("authorityRuntimeId")] public string AuthorityRuntimeId { get; set; } = string.Empty;
[Key("protocolId")] public string ProtocolId { get; set; } = string.Empty;
[Key("routeGeneration")] public string RouteGeneration { get; set; } = string.Empty;
[Key("error")] public string? Error { get; set; }
}
/// <summary>
/// Requests known CultMesh Verses from a peer.
/// </summary>
@ -1254,6 +1285,11 @@ namespace GameCult.Networking
/// </summary>
[Key("authorityRuntimeIds")] public string[] AuthorityRuntimeIds { get; set; } = Array.Empty<string>();
/// <summary>
/// Gets or sets exact authoritative-runtime route bindings. This is the
/// routing authority; the parallel legacy arrays are projections only.
/// </summary>
[Key("authorityRoutes")] public CultMeshAuthorityRouteMessage[] AuthorityRoutes { get; set; } = Array.Empty<CultMeshAuthorityRouteMessage>();
/// <summary>
/// Gets or sets the parent Verse id for overlays or branches.
/// </summary>
[Key("parentVerseId")] public string? ParentVerseId { get; set; }
@ -1263,6 +1299,17 @@ namespace GameCult.Networking
[Key("description")] public string? Description { get; set; }
}
/// <summary>Wire representation of one authority-bound CultMesh route.</summary>
[MessagePackObject]
public class CultMeshAuthorityRouteMessage
{
[Key("authorityRuntimeId")] public string AuthorityRuntimeId { get; set; } = string.Empty;
[Key("endpoint")] public string Endpoint { get; set; } = string.Empty;
[Key("protocolIds")] public string[] ProtocolIds { get; set; } = Array.Empty<string>();
[Key("priority")] public int Priority { get; set; }
[Key("generation")] public string Generation { get; set; } = string.Empty;
}
/// <summary>
/// Wire representation of CultMesh Verse compatibility.
/// </summary>

View file

@ -22,7 +22,7 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests
var connector = new CultMeshNativeQuicRealtimeTransportConnector();
using var client = await connector.ConnectAsync(
new CultMeshTransportCandidate(endpoint!),
CultMeshEndpointId.Parse("aetheria.local"));
new CultMeshSessionTarget("aetheria", "aetheria.local"));
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var frame = await client.ReceiveAsync(timeout.Token);
@ -49,7 +49,7 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests
using var client = await connector.ConnectAsync(
new CultMeshTransportCandidate(endpoint),
CultMeshEndpointId.Parse("aetheria.local"));
new CultMeshSessionTarget("aetheria", "aetheria.local"));
var expected = new CultMeshRealtimeFrame
{
ChannelId = "aetheria.entities",
@ -92,7 +92,7 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests
var connect = () => connector.ConnectAsync(
new CultMeshTransportCandidate(endpoint),
CultMeshEndpointId.Parse("aetheria.local"));
new CultMeshSessionTarget("aetheria", "aetheria.local"));
await connect.Should().ThrowAsync<IOException>()
.WithMessage("*certificate*pin*");
@ -114,10 +114,10 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests
var connector = new CultMeshNativeQuicRealtimeTransportConnector();
var departedClient = await connector.ConnectAsync(
new CultMeshTransportCandidate(endpoint),
CultMeshEndpointId.Parse("aetheria.local"));
new CultMeshSessionTarget("aetheria", "aetheria.local"));
using var healthyClient = await connector.ConnectAsync(
new CultMeshTransportCandidate(endpoint),
CultMeshEndpointId.Parse("aetheria.local"));
new CultMeshSessionTarget("aetheria", "aetheria.local"));
await WaitUntilAsync(() => server.ConnectionCount == 2);
await server.BroadcastAsync(Frame(sequence: 1));
using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));

View file

@ -40,7 +40,7 @@ public sealed class CultMeshQuicRealtimeTransportTests
});
using var client = await connector.ConnectAsync(
new CultMeshTransportCandidate($"cultmesh-state+quic://127.0.0.1:{listener.LocalEndPoint.Port}"),
CultMeshEndpointId.Parse("service:aetheria.daemon"));
new CultMeshSessionTarget("aetheria", "service:aetheria.daemon"));
await using var serverConnection = await accept;
client.TransportId.Should().Be("msquic-realtime");
@ -62,7 +62,7 @@ public sealed class CultMeshQuicRealtimeTransportTests
using var client = await connector.ConnectAsync(
new CultMeshTransportCandidate(endpoint),
CultMeshEndpointId.Parse("service:aetheria.daemon"));
new CultMeshSessionTarget("aetheria", "service:aetheria.daemon"));
client.TransportId.Should().Be("msquic-realtime");
await WaitUntilAsync(() => server.ConnectionCount == 1);
@ -78,8 +78,8 @@ public sealed class CultMeshQuicRealtimeTransportTests
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", endpointId.Value);
var runtimeId = CultMeshRuntimeId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", runtimeId.Value);
using var sessions = CreateSessions(server, out var validatedIdentity);
CultMeshRealtimeSession session;
try
@ -105,7 +105,7 @@ public sealed class CultMeshQuicRealtimeTransportTests
new[] { firstAtServer.Sequence, secondAtServer.Sequence }.Should().Equal(1, 2);
new[] { firstAtClient.Sequence, secondAtClient.Sequence }.Should().Equal(3, 4);
firstAtClient.Payload.ToArray().Should().Equal(3, 4, 5);
validatedIdentity.Value.Should().Be(endpointId.Value);
validatedIdentity.Value.Should().Be(runtimeId.Value);
}
[Test]
@ -118,8 +118,8 @@ public sealed class CultMeshQuicRealtimeTransportTests
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", endpointId.Value);
var runtimeId = CultMeshRuntimeId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", runtimeId.Value);
using var sessions = CreateSessions(server, out _);
var session = await sessions.ConnectRealtimeAsync(target);
await WaitUntilAsync(() => server.ConnectionCount == 1);
@ -148,8 +148,8 @@ public sealed class CultMeshQuicRealtimeTransportTests
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", endpointId.Value);
var runtimeId = CultMeshRuntimeId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", runtimeId.Value);
using var sessions = CreateSessions(server, out _);
var session = await sessions.ConnectRealtimeAsync(target);

View file

@ -53,6 +53,13 @@ public sealed class CultMeshBodySessionTests
RuntimeId = "cultmesh-body-rudp-test", Socket = socket, MaxFragmentBytes = 1024,
MaxPendingReliablePackets = 8192
});
var endpoint = $"rudp://127.0.0.1:{server.LocalEndPoint.Port}";
using var identityServer = new CultMeshSessionIdentityServer(
server,
"aetheria.daemon",
new[] { "aetheria" },
new[] { CultMeshProtocols.Bodies.Value },
new[] { "aetheria.daemon@" + endpoint });
using var bodyServer = new CultMeshBodyServer(server, store);
using var pumpCancellation = new CancellationTokenSource();
var pump = Task.Run(async () =>
@ -64,7 +71,6 @@ public sealed class CultMeshBodySessionTests
if (progress.TransportItemsConsumed == 0) await Task.Delay(1, pumpCancellation.Token);
}
});
var endpoint = $"rudp://127.0.0.1:{server.LocalEndPoint.Port}";
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
using var sessions = new CultMeshSessionManager(discovery,
new ICultMeshTransportConnector[] { new CultMeshSchemaTransportConnector() });
@ -179,7 +185,7 @@ public sealed class CultMeshBodySessionTests
((Func<T, ICultNetSchemaServerPeer, Task>)_handlers[typeof(T)])(message, peer);
}
private sealed class LoopbackClient : ICultNetSchemaClient
private sealed class LoopbackClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
{
private readonly LoopbackServer _server;
private readonly Dictionary<Type, List<Delegate>> _handlers = new();
@ -208,6 +214,8 @@ public sealed class CultMeshBodySessionTests
foreach (var value in values.ToArray()) ((Action<T>)value)(message);
}
public void Dispose() { }
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria.daemon";
}
private sealed class LoopbackPeer : ICultNetSchemaServerPeer

View file

@ -303,7 +303,7 @@ public sealed class CultMeshClientTests
result.MessageId.Should().Be("pilot-command-1");
result.Status.Should().Be("accepted");
result.SourceRuntimeId.Should().Be("provider-daemon");
result.SourceRuntimeId.Should().Be("aetheria-daemon");
result.Value.ReceiptId.Should().Be("receipt:pilot-command-1");
result.Value.Count.Should().Be(2);
connector.Clients.Should().ContainSingle();
@ -370,6 +370,31 @@ public sealed class CultMeshClientTests
connector.Clients[0].Operations.Should().ContainSingle();
}
[Test]
public async Task InvokeOperation_RejectsReceiptFromAnotherRuntime()
{
var connector = new DocumentConnector { ResponseSourceRuntimeId = "intruder-daemon" };
using var mesh = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { "rudp://odin:3076" },
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
Func<Task> invoke = async () => await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
Target,
"tests.counter",
"tests.counter.increment",
"tests.counter.increment_request.v1",
"tests.counter.increment_receipt.v1",
new ClientOperationRequest { Amount = 2 },
sourceRuntimeId: "pilot-one",
idempotencyKey: "pilot-command-wrong-source");
var failure = await invoke.Should().ThrowAsync<CultMeshSessionException>();
failure.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority);
}
[Test]
public async Task Dispose_TerminatesDocumentWaitingForInitialSnapshot()
{
@ -438,13 +463,15 @@ public sealed class CultMeshClientTests
}
}
private sealed class ConnectedClient : ICultNetSchemaClient
private sealed class ConnectedClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
{
public bool Connected => true;
public void Connect(string host, int port) { }
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage { }
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage { }
public void Dispose() { }
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
}
private sealed class DocumentConnector : ICultMeshTransportConnector
@ -454,6 +481,7 @@ public sealed class CultMeshClientTests
public bool SuppressFirstSubscriptionSnapshot { get; set; }
public bool RejectOperations { get; set; }
public bool FailFirstOperationBeforeResponse { get; set; }
public string ResponseSourceRuntimeId { get; set; } = "aetheria-daemon";
public string ConnectorId => "document-test";
public int Priority => 0;
public int SubscribeCount => Clients.Sum(client => client.SubscribeCount);
@ -469,13 +497,14 @@ public sealed class CultMeshClientTests
? int.MaxValue
: SuppressFirstSubscriptionSnapshot && Clients.Count == 0 ? 1 : 0,
rejectOperations: RejectOperations,
failOperationBeforeResponse: FailFirstOperationBeforeResponse && Clients.Count == 0);
failOperationBeforeResponse: FailFirstOperationBeforeResponse && Clients.Count == 0,
responseSourceRuntimeId: ResponseSourceRuntimeId);
Clients.Add(client);
return Task.FromResult<ICultNetSchemaClient>(client);
}
}
private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth
private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth, ICultMeshVerifiedSchemaClient
{
private readonly List<Action<CultNetSnapshotResponseRawMessage>> _snapshots = new();
private readonly List<Action<CultNetOperationResponseMessage>> _operationResponses = new();
@ -484,17 +513,20 @@ public sealed class CultMeshClientTests
private readonly TaskCompletionSource<Exception> _failure = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly bool _rejectOperations;
private bool _failOperationBeforeResponse;
private readonly string _responseSourceRuntimeId;
private int _snapshotsToSuppress;
public DocumentClient(
string text,
int snapshotsToSuppress = 0,
bool rejectOperations = false,
bool failOperationBeforeResponse = false)
bool failOperationBeforeResponse = false,
string responseSourceRuntimeId = "aetheria-daemon")
{
_text = text;
_snapshotsToSuppress = snapshotsToSuppress;
_rejectOperations = rejectOperations;
_failOperationBeforeResponse = failOperationBeforeResponse;
_responseSourceRuntimeId = responseSourceRuntimeId;
var cache = CultMesh.CreateCultCacheDocumentRegistry(typeof(ClientTestDocument));
_documents = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(ClientTestDocument) }, cache);
}
@ -514,6 +546,7 @@ public sealed class CultMeshClientTests
}
if (message is CultNetOperationRequestMessage operation)
{
operation.TargetRuntimeId.Should().Be("aetheria-daemon");
Operations.Add(operation);
if (_failOperationBeforeResponse)
{
@ -539,7 +572,7 @@ public sealed class CultMeshClientTests
failure,
CultNetSchemaMessageSerialization.Options)),
Diagnostics = new[] { failure.Message },
SourceRuntimeId = "provider-daemon"
SourceRuntimeId = _responseSourceRuntimeId
};
foreach (var handler in _operationResponses.ToArray()) handler(rejected);
return;
@ -562,7 +595,7 @@ public sealed class CultMeshClientTests
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
receipt,
CultNetSchemaMessageSerialization.Options)),
SourceRuntimeId = "provider-daemon"
SourceRuntimeId = _responseSourceRuntimeId
};
foreach (var handler in _operationResponses.ToArray()) handler(operationResponse);
return;
@ -596,6 +629,8 @@ public sealed class CultMeshClientTests
}
public void Dispose() { }
public void Fail(Exception error) => _failure.TrySetResult(error);
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
}
private static async Task WaitUntilAsync(Func<bool> condition)

View file

@ -17,6 +17,7 @@ namespace GameCult.Mesh.Tests;
[TestFixture]
public sealed class CultMeshContentSessionTests
{
private const string RouteGeneration = "content-test-route-1";
private string _directory = null!;
[SetUp]
@ -41,7 +42,13 @@ public sealed class CultMeshContentSessionTests
StringComparer.Ordinal);
using var contentServer = new CultMeshTcpContentServer(
new TcpListener(IPAddress.Loopback, 0),
hash => providerChunks.TryGetValue(hash, out var chunk) ? chunk : null);
hash => providerChunks.TryGetValue(hash, out var chunk) ? chunk : null,
new CultMeshTcpContentServerOptions
{
VerseId = "aetheria",
AuthorityRuntimeId = "aetheria.daemon",
RouteGeneration = RouteGeneration
});
var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}";
var connector = new CultMeshTcpContentTransportConnector();
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
@ -73,6 +80,34 @@ public sealed class CultMeshContentSessionTests
"all chunk requests borrow one identity-first content session");
}
[Test]
public async Task TcpContentSessionRejectsEndpointServingAnotherAuthority()
{
using var content = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk)));
using var contentServer = new CultMeshTcpContentServer(
new TcpListener(IPAddress.Loopback, 0),
content,
new CultMeshTcpContentServerOptions
{
VerseId = "aetheria",
AuthorityRuntimeId = "intruder.daemon",
RouteGeneration = RouteGeneration
});
var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}";
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
using var sessions = new CultMeshSessionManager(
discovery,
Array.Empty<ICultMeshTransportConnector>(),
new ICultMeshContentTransportConnector[] { new CultMeshTcpContentTransportConnector() });
Func<Task> connect = async () => await sessions.ConnectContentAsync(
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
var error = await connect.Should().ThrowAsync<CultMeshSessionException>();
error.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority);
}
[Test]
public async Task RudpContentSessionCarriesConcurrentChunkWindowWithoutPerFragmentSchedulerDelay()
{
@ -91,6 +126,12 @@ public sealed class CultMeshContentSessionTests
MaxPendingReliablePackets = 8192
});
using var contentServer = new CultMeshLegacyRudpContentServer(wireServer, providerCache);
using var identityServer = new CultMeshSessionIdentityServer(
wireServer,
"aetheria.daemon",
new[] { "aetheria" },
new[] { CultMeshProtocols.Content.Value },
new[] { RouteGeneration });
using var pumpCancellation = new CancellationTokenSource();
var pump = Task.Run(async () =>
{
@ -196,8 +237,14 @@ public sealed class CultMeshContentSessionTests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "rules"),
new[] { _endpoint },
new[] { "aetheria.daemon" }),
authorityRoutes: new[]
{
new CultMeshAuthorityRoute(
"aetheria.daemon",
_endpoint,
new[] { CultMeshProtocols.Content.Value },
generation: RouteGeneration)
}),
SourceId,
now,
now.AddMinutes(1),
@ -241,7 +288,7 @@ public sealed class CultMeshContentSessionTests
((Func<TMessage, ICultNetSchemaServerPeer, Task>)_handlers[typeof(TMessage)])(message, peer);
}
private sealed class LoopbackSchemaClient : ICultNetSchemaClient
private sealed class LoopbackSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
{
private readonly LoopbackSchemaServer _server;
private readonly Dictionary<Type, List<Delegate>> _handlers = new();
@ -270,6 +317,8 @@ public sealed class CultMeshContentSessionTests
handlers.Add(callback);
}
public void Dispose() { }
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria.daemon";
public void Emit<T>(T message) where T : ICultNetSchemaMessage
{
if (!_handlers.TryGetValue(typeof(T), out var handlers)) return;

View file

@ -295,7 +295,7 @@ public sealed class CultMeshDiscoveryServiceTests
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
new[] { endpoint },
new[] { "odin:aetheria", "odin:games", "aetheria" }),
new[] { "aetheria" }),
sourceId,
clock.UtcNow,
clock.UtcNow + ttl,

View file

@ -79,6 +79,81 @@ public sealed class CultMeshSessionManagerTests
fastClient.DisposeCount.Should().Be(0);
}
[Test]
public async Task Connect_NeverAttemptsRouteBoundToAnotherAuthority()
{
var clock = new ManualSessionClock();
var descriptor = new CultMeshVerseDescriptor(
"aetheria",
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
authorityRoutes: new[]
{
new CultMeshAuthorityRoute("decoy-daemon", "rudp://decoy:3076"),
new CultMeshAuthorityRoute("aetheria-daemon", "rudp://authority:3076")
});
using var discovery = new CultMeshDiscoveryService(
new[] { new StaticRouteSource(clock, descriptor) },
new CultMeshDiscoveryServiceOptions { Clock = clock });
var attempts = new List<string>();
var connector = new FakeConnector((candidate, _) =>
{
attempts.Add(candidate.Endpoint);
return Task.FromResult<ICultNetSchemaClient>(new FakeSchemaClient());
});
using var manager = new CultMeshSessionManager(discovery, new[] { connector });
var session = await manager.ConnectAsync(Target, CultMeshProtocols.Documents);
session.State.Path!.Endpoint.Should().Be("rudp://authority:3076");
attempts.Should().Equal("rudp://authority:3076");
}
[Test]
public void Connect_RejectsRouteWhoseNativeIdentityDoesNotMatchTarget()
{
var clock = new ManualSessionClock();
using var discovery = Discovery(clock, () => new[] { "rudp://decoy:3076" });
var connector = new FakeConnector((_, _) =>
Task.FromResult<ICultNetSchemaClient>(new NativeMismatchClient()));
using var manager = new CultMeshSessionManager(discovery, new[] { connector });
var error = Assert.ThrowsAsync<CultMeshSessionException>(() =>
manager.ConnectAsync(Target, CultMeshProtocols.Documents));
error!.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority);
}
[Test]
public void Connect_RejectsPortableHandshakeFromWrongRuntime()
{
var clock = new ManualSessionClock();
using var discovery = Discovery(clock, () => new[] { "rudp://decoy:3076" });
var connector = new FakeConnector((_, _) =>
Task.FromResult<ICultNetSchemaClient>(new HandshakeSchemaClient("decoy-daemon")));
using var manager = new CultMeshSessionManager(discovery, new[] { connector });
var error = Assert.ThrowsAsync<CultMeshSessionException>(() =>
manager.ConnectAsync(Target, CultMeshProtocols.Documents));
error!.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority);
}
[Test]
public void Descriptor_RejectsAmbiguousLegacyMultiAuthorityRoutes()
{
var action = () => new CultMeshVerseDescriptor(
"aetheria",
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
discoveryEndpoints: new[] { "rudp://shared:3076" },
authorityRuntimeIds: new[] { "authority-a", "authority-b" });
action.Should().Throw<InvalidOperationException>().WithMessage("*ambiguous*");
}
[Test]
public async Task Connect_CallerCancellationDoesNotCancelSharedAttempt()
{
@ -308,6 +383,30 @@ public sealed class CultMeshSessionManagerTests
CultMeshDiscoveryTrust.Signed);
}
private sealed class StaticRouteSource : ICultMeshLookupSource
{
private readonly ManualSessionClock _clock;
private readonly CultMeshVerseDescriptor _descriptor;
public StaticRouteSource(ManualSessionClock clock, CultMeshVerseDescriptor descriptor)
{
_clock = clock;
_descriptor = descriptor;
}
public string SourceId => "odin";
public Task<IReadOnlyList<CultMeshDiscoveryObservation>> LookupAsync(
CultMeshDiscoveryQuery query,
CancellationToken cancellationToken = default) =>
Task.FromResult<IReadOnlyList<CultMeshDiscoveryObservation>>(new[]
{
new CultMeshDiscoveryObservation(
_descriptor,
SourceId,
_clock.UtcNow,
_clock.UtcNow.AddMinutes(1),
CultMeshDiscoveryTrust.Signed)
});
}
private sealed class FakeConnector : ICultMeshTransportConnector
{
private readonly Func<CultMeshTransportCandidate, CancellationToken, Task<ICultNetSchemaClient>> _connect;
@ -324,7 +423,7 @@ public sealed class CultMeshSessionManagerTests
}
}
private sealed class FakeSchemaClient : ICultNetSchemaClient, ICultNetSchemaClientHealth
private sealed class FakeSchemaClient : ICultNetSchemaClient, ICultNetSchemaClientHealth, ICultMeshVerifiedSchemaClient
{
private readonly TaskCompletionSource<Exception> _failure = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly List<Action<CultNetErrorMessage>> _handlers = new();
@ -340,6 +439,8 @@ public sealed class CultMeshSessionManagerTests
_handlers.Add(message => callback((T)(object)message));
}
public void Dispose() => Interlocked.Increment(ref _disposeCount);
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
public void Fail(Exception error) => _failure.TrySetResult(error);
public void Emit(CultNetErrorMessage message) { foreach (var handler in _handlers.ToArray()) handler(message); }
}
@ -360,7 +461,46 @@ public sealed class CultMeshSessionManagerTests
public void Dispose() { }
}
private sealed class RespondingSchemaClient : ICultNetSchemaClient
private sealed class NativeMismatchClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
{
public bool Connected => true;
public void Connect(string host, int port) { }
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage { }
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage { }
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => false;
public void Dispose() { }
}
private sealed class HandshakeSchemaClient : ICultNetSchemaClient
{
private readonly string _actualRuntimeId;
private readonly List<Action<CultMeshSessionAcceptedMessage>> _handlers = new();
public HandshakeSchemaClient(string actualRuntimeId) => _actualRuntimeId = actualRuntimeId;
public bool Connected => true;
public void Connect(string host, int port) { }
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
{
if (message is not CultMeshSessionOpenMessage request) return;
var response = new CultMeshSessionAcceptedMessage
{
MessageId = request.MessageId,
Accepted = true,
VerseId = request.VerseId,
AuthorityRuntimeId = _actualRuntimeId,
ProtocolId = request.ProtocolId,
RouteGeneration = request.RouteGeneration
};
foreach (var handler in _handlers.ToArray()) handler(response);
}
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
{
if (typeof(T) == typeof(CultMeshSessionAcceptedMessage))
_handlers.Add(message => callback((T)(object)message));
}
public void Dispose() { }
}
private sealed class RespondingSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
{
private readonly List<Action<CultNetSnapshotResponseRawMessage>> _handlers = new();
private int _disposeCount;
@ -385,9 +525,11 @@ public sealed class CultMeshSessionManagerTests
_handlers.Add(message => callback((T)(object)message));
}
public void Dispose() => Interlocked.Increment(ref _disposeCount);
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
}
private sealed class PeerExchangeSchemaClient : ICultNetSchemaClient
private sealed class PeerExchangeSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
{
private readonly List<Action<CultMeshPeerExchangeResponseMessage>> _handlers = new();
public bool Connected => true;
@ -413,9 +555,11 @@ public sealed class CultMeshSessionManagerTests
_handlers.Add(message => callback((T)(object)message));
}
public void Dispose() { }
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
}
private sealed class LeaseSchemaClient : ICultNetSchemaClient
private sealed class LeaseSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
{
private readonly List<Action<CultNetErrorMessage>> _handlers = new();
private int _disposeCount;
@ -430,6 +574,8 @@ public sealed class CultMeshSessionManagerTests
}
public void Emit(CultNetErrorMessage message) { foreach (var handler in _handlers.ToArray()) handler(message); }
public void Dispose() => Interlocked.Increment(ref _disposeCount);
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
}
private sealed class ManualSessionClock : ICultMeshClock

View file

@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
@ -15,10 +16,17 @@ namespace GameCult.Mesh.Tests;
[TestFixture]
public sealed class CultMeshTransportModularityTests
{
private const string RouteGeneration = "transport-modularity-route-1";
[Test]
public async Task SchemaSessionPrefersTcpTierWithoutTouchingLegacyDatagrams()
{
using var server = new TcpFramedCultNetSchemaServer(new TcpListener(IPAddress.Loopback, 0));
using var identity = new CultMeshSessionIdentityServer(
server,
"aetheria.daemon",
new[] { "aetheria" },
new[] { CultMeshProtocols.Documents.Value },
new[] { RouteGeneration });
server.OnCultNet<CultNetHelloMessage>((message, peer) =>
{
peer.SendCultNet(new CultNetErrorMessage { Error = "tcp:" + message.RuntimeId });
@ -61,7 +69,15 @@ public sealed class CultMeshTransportModularityTests
using var content = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk)));
await CultMeshCdn.PublishAsync(content, artifact);
using var server = new CultMeshTcpContentServer(new TcpListener(IPAddress.Loopback, 0), content);
using var server = new CultMeshTcpContentServer(
new TcpListener(IPAddress.Loopback, 0),
content,
new CultMeshTcpContentServerOptions
{
VerseId = "aetheria",
AuthorityRuntimeId = "aetheria.daemon",
RouteGeneration = RouteGeneration
});
var tcpEndpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{server.LocalEndPoint.Port}";
var legacy = new RejectingLegacyContentConnector();
using var discovery = new CultMeshDiscoveryService(new[]
@ -156,6 +172,26 @@ public sealed class CultMeshTransportModularityTests
.WithMessage("*No realtime state connectors are configured*");
}
[Test]
public async Task RealtimeSessionRejectsConnectedTransportWithoutAuthorityProof()
{
using var discovery = new CultMeshDiscoveryService(new[]
{
new RouteSource("cultmesh-state+quic://127.0.0.1:3076")
});
using var sessions = new CultMeshSessionManager(
discovery,
Array.Empty<ICultMeshTransportConnector>(),
Array.Empty<ICultMeshContentTransportConnector>(),
new ICultMeshRealtimeTransportConnector[] { new UnverifiedRealtimeConnector() });
Func<Task> connect = async () => await sessions.ConnectRealtimeAsync(
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
var error = await connect.Should().ThrowAsync<CultMeshSessionException>();
error.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority);
}
private sealed class RouteSource : ICultMeshLookupSource
{
private readonly string[] _endpoints;
@ -176,8 +212,11 @@ public sealed class CultMeshTransportModularityTests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "rules"),
_endpoints,
new[] { "aetheria.daemon" }),
authorityRoutes: _endpoints.Select(endpoint => new CultMeshAuthorityRoute(
"aetheria.daemon",
endpoint,
protocolIds: null,
generation: RouteGeneration))),
SourceId,
now,
now.AddMinutes(1),
@ -214,7 +253,7 @@ public sealed class CultMeshTransportModularityTests
candidate.Endpoint.StartsWith("rudp://", StringComparison.OrdinalIgnoreCase);
public Task<ICultMeshContentTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
Interlocked.Increment(ref _connectCount);
@ -232,24 +271,70 @@ public sealed class CultMeshTransportModularityTests
candidate.Endpoint.StartsWith("cultmesh-state+quic://", StringComparison.OrdinalIgnoreCase);
public Task<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
Interlocked.Increment(ref _connectCount);
return Task.FromResult<ICultMeshRealtimeTransport>(new LoopbackRealtimeTransport(candidate.Endpoint));
return Task.FromResult<ICultMeshRealtimeTransport>(
new LoopbackRealtimeTransport(candidate.Endpoint, target, candidate.Generation));
}
}
private sealed class LoopbackRealtimeTransport : ICultMeshRealtimeTransport
private sealed class UnverifiedRealtimeConnector : ICultMeshRealtimeTransportConnector
{
public string ConnectorId => "unverified-realtime";
public int Priority => 0;
public bool CanConnect(CultMeshTransportCandidate candidate) => true;
public Task<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default) =>
Task.FromResult<ICultMeshRealtimeTransport>(new UnverifiedRealtimeTransport(candidate.Endpoint));
}
private sealed class UnverifiedRealtimeTransport : ICultMeshRealtimeTransport
{
public UnverifiedRealtimeTransport(string endpoint) => Endpoint = endpoint;
public string TransportId => "unverified-realtime";
public string Endpoint { get; }
public Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default) =>
Task.CompletedTask;
public Task<CultMeshRealtimeFrame> ReceiveAsync(CancellationToken cancellationToken = default) =>
Task.FromException<CultMeshRealtimeFrame>(new InvalidOperationException());
public void Dispose() { }
}
private sealed class LoopbackRealtimeTransport : ICultMeshRealtimeTransport, ICultMeshVerifiedTransport
{
private readonly Queue<CultMeshRealtimeFrame> _frames = new();
private readonly SemaphoreSlim _available = new(0);
private bool _disposed;
public LoopbackRealtimeTransport(string endpoint) => Endpoint = endpoint;
private readonly CultMeshSessionTarget _target;
private readonly string _routeGeneration;
public LoopbackRealtimeTransport(
string endpoint,
CultMeshSessionTarget target,
string routeGeneration)
{
Endpoint = endpoint;
_target = target;
_routeGeneration = routeGeneration;
}
public string TransportId => "test-quic-state";
public string Endpoint { get; }
public bool IsVerifiedFor(
string verseId,
string authorityRuntimeId,
string protocolId,
string routeGeneration) =>
string.Equals(_target.VerseId, verseId, StringComparison.Ordinal) &&
string.Equals(_target.AuthorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) &&
string.Equals(CultMeshProtocols.RealtimeState.Value, protocolId, StringComparison.Ordinal) &&
string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal);
public Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default)
{
if (_disposed) throw new ObjectDisposedException(nameof(LoopbackRealtimeTransport));

View file

@ -77,6 +77,41 @@ namespace GameCult.Networking.Tests
Is.EqualTo(4));
}
[Test]
public async Task RejectsRequestAddressedToAnotherRuntimeBeforeApplicationHandler()
{
var transport = new FakeServer();
var peer = new FakePeer();
var calls = 0;
using var operations = new CultNetOperationServer(transport, "sample.provider")
.Register<AddRequest, AddReceipt>(
"sample.counter",
"sample.counter.add",
"sample.add.v1",
"sample.add_receipt.v1",
context =>
{
calls++;
return Task.FromResult(new AddReceipt { Count = context.Value.Amount });
});
var request = Request(new AddRequest { Amount = 3 });
request.TargetRuntimeId = "other.provider";
await transport.DispatchAsync(request, peer);
Assert.That(calls, Is.Zero);
var response = peer.Messages.OfType<CultNetOperationResponseMessage>().Single();
var failure = MessagePackSerializer.Deserialize<CultNetOperationFailure>(
Convert.FromBase64String(response.Payload),
CultNetSchemaMessageSerialization.Options);
Assert.Multiple(() =>
{
Assert.That(response.Status, Is.EqualTo("denied"));
Assert.That(failure.Code, Is.EqualTo("target-runtime-mismatch"));
Assert.That(response.SourceRuntimeId, Is.EqualTo("sample.provider"));
});
}
[Test]
public async Task RejectsWrongSchemaBeforeApplicationHandler()
{
@ -157,7 +192,8 @@ namespace GameCult.Networking.Tests
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
value,
CultNetSchemaMessageSerialization.Options)),
SourceRuntimeId = "sample.browser"
SourceRuntimeId = "sample.browser",
TargetRuntimeId = "sample.provider"
};
[MessagePackObject]

View file

@ -3627,8 +3627,19 @@ namespace GameCult.Networking.Tests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", vanillaRules),
discoveryEndpoints: ["cultmesh://aetheria.example.test:3075"],
authorityRuntimeIds: ["gc-us-east", "gc-eu-west"]);
authorityRoutes:
[
new CultMeshAuthorityRoute(
"gc-us-east",
"cultmesh://us-east.aetheria.example.test:3075",
[CultMeshProtocols.Documents.Value],
generation: "east-1"),
new CultMeshAuthorityRoute(
"gc-eu-west",
"cultmesh://eu-west.aetheria.example.test:3075",
[CultMeshProtocols.Documents.Value],
generation: "west-1")
]);
var modded = new CultMeshVerseDescriptor(
"aetheria-skylands",
"Aetheria: Skylands",

View file

@ -339,6 +339,12 @@ static async Task<Result> MeasureCultMeshRudpAsync(byte[] payload)
MaxPendingReliablePackets = 8192
});
using var contentServer = new CultMeshLegacyRudpContentServer(wireServer, providerCache);
using var identityServer = new CultMeshSessionIdentityServer(
wireServer,
"service:parity.provider",
["parity"],
[CultMeshProtocols.Content.Value],
["parity-route-1"]);
using var pumpCancellation = new CancellationTokenSource();
var pump = Task.Run(async () =>
{
@ -359,7 +365,7 @@ static async Task<Result> MeasureCultMeshRudpAsync(byte[] payload)
[new CultMeshLegacyRudpContentTransportConnector(
new CultMeshLegacyRudpContentTransportOptions { ResponseTimeout = TimeSpan.FromMinutes(2) })]);
var provider = new CultMeshSessionContentProvider(
"parity.provider", sessions, CultMeshEndpointId.Parse("service:parity.provider"));
"parity.provider", sessions, new CultMeshSessionTarget("parity", "service:parity.provider"));
var directory = Path.Combine(Path.GetTempPath(), "cultmesh-transport-parity", Guid.NewGuid().ToString("N"));
using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshContentTransferStateDocument)));
@ -392,7 +398,14 @@ static async Task<Result> MeasureCultMeshTcpAsync(byte[] payload)
await CultMeshCdn.PublishAsync(providerCache, artifact);
using var contentServer = new CultMeshTcpContentServer(
new TcpListener(IPAddress.Loopback, 0), providerCache);
new TcpListener(IPAddress.Loopback, 0),
providerCache,
new CultMeshTcpContentServerOptions
{
VerseId = "parity",
AuthorityRuntimeId = "service:parity.provider",
RouteGeneration = "parity-route-1"
});
var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}";
using var discovery = new CultMeshDiscoveryService([new RouteSource(endpoint)]);
using var sessions = new CultMeshSessionManager(
@ -400,7 +413,7 @@ static async Task<Result> MeasureCultMeshTcpAsync(byte[] payload)
Array.Empty<ICultMeshTransportConnector>(),
[new CultMeshTcpContentTransportConnector()]);
var provider = new CultMeshSessionContentProvider(
"parity.provider", sessions, CultMeshEndpointId.Parse("service:parity.provider"));
"parity.provider", sessions, new CultMeshSessionTarget("parity", "service:parity.provider"));
var directory = Path.Combine(Path.GetTempPath(), "cultmesh-transport-parity", Guid.NewGuid().ToString("N"));
using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshContentTransferStateDocument)));
@ -479,8 +492,14 @@ internal sealed class RouteSource(string endpoint) : ICultMeshLookupSource
new(
new CultMeshVerseDescriptor(
"parity", "Transport parity", CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "parity"), [endpoint],
["service:parity.provider"]),
new CultMeshVerseCompatibility("cultmesh.v1", "parity"),
authorityRoutes:
[
new CultMeshAuthorityRoute(
"service:parity.provider",
endpoint,
generation: "parity-route-1")
]),
SourceId, now, now.AddMinutes(1), CultMeshDiscoveryTrust.Signed)
];
return Task.FromResult(observations);