From 1753dfbb92afba56a5f1275f5eaeed6b1d5ace7d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 15:56:45 +0100 Subject: [PATCH] Add CultMesh peer snapshot document handles --- packages/cultmesh-ts/README.md | 20 +++ packages/cultmesh-ts/src/index.ts | 167 +++++++++++++++++++++ packages/cultmesh-ts/test/cultmesh.test.ts | 81 ++++++++++ src/GameCult.Mesh/docs/public-api.md | 16 ++ 4 files changed, 284 insertions(+) diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index 3772ef92..9a210f33 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -265,6 +265,18 @@ const client = await CultMesh.createRudpPeerForAuthorizedPeer( ); const schemas = CultMesh.createSchemaCatalog(); await client.syncSchemaCatalog(schemas, { kinds: ["document_payload"] }); + +const health = CultMesh.documentFromPeerSnapshot( + () => client, + "gamecult.aetheria.daemon_health.v1", + "daemon:aetheria.health.v1", + { + routeHint: CultMesh.routeHint("network", endpoint.uri), + timeoutMs: 5_000, + }, +); + +const latestHealth = await health.latest(); ``` `CultMesh.createRudpPeerForPeer(...)` remains available for already trusted @@ -275,6 +287,14 @@ Discovery-first schema/catalog paths should prefer `CultMeshPeerCatalog.firstAuthorized(...)` with the authority lease catalog before using the peer-card endpoint as a dial target. +`CultMesh.documentFromPeerSnapshot(...)` is the document-level path for RUDP +snapshots. It hides the snapshot request message id, response listener, raw +payload binary normalization, and MessagePack decode behind the same +`CultMeshDocumentHandle` shape as local documents. The helper prefers an exact +schema id match and falls back to the requested record key, so runtimes that +alias the same logical document through different generated schema names can +still read the publication with one CultMesh call. + ## Streaming Mode CultMesh streaming mode is for audio, video, tensor, and opaque byte frames that diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 5ed8de31..31cb0e07 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -21,11 +21,13 @@ import { encodeCultNetMessageForWire, encodeRudpPacket, parseCultNetMessage, + type CultNetErrorMessage, type CultNetDocumentBinding, type CultNetMessage, type CultNetRawDocumentRecord, type CultNetReconnectPolicy, type CultNetSchemaCatalogOptions, + type CultNetSnapshotResponseRawMessage, type CultNetWireContract, } from "cultnet-ts"; @@ -1884,6 +1886,49 @@ export function cultMeshDocumentFromSingleFile( ); } +export function cultMeshDocumentFromPeerSnapshot( + peer: CultNetPeer | (() => CultNetPeer | Promise), + schemaId: string, + recordKey: string, + options: { + documentId?: string; + routeHint?: CultMeshRouteHint; + sourceId?: string; + timeoutMs?: number; + pollMs?: number; + messageIdPrefix?: string; + } = {}, +): CultMeshDocumentHandle { + requireNonEmpty(schemaId, "schemaId"); + requireNonEmpty(recordKey, "recordKey"); + const documentId = options.documentId ?? recordKey; + const read = async () => + decodeCultNetRawDocumentPayload( + await requestCultNetRawSnapshotDocument( + peer, + schemaId, + recordKey, + { + timeoutMs: options.timeoutMs, + messageIdPrefix: options.messageIdPrefix ?? documentId, + }, + ), + ); + + return cultMeshDocument( + documentId, + { schemaId }, + async () => read(), + { + routeHint: options.routeHint ?? cultMeshRouteHint("network", "CultNet snapshot"), + sources: [cultMeshProjectionSource(options.sourceId ?? documentId, { schemaId })], + watchDocument: cultMeshPollingDocumentWatcher(read, { + intervalMs: options.pollMs ?? 250, + }), + }, + ); +} + export function cultMeshGlobalDocumentFromCache( cache: CultCache, definition: TDefinition, @@ -3591,6 +3636,22 @@ export class CultMesh { return cultMeshDocumentFromSingleFile(path, schemaId, options); } + public static documentFromPeerSnapshot( + peer: CultNetPeer | (() => CultNetPeer | Promise), + schemaId: string, + recordKey: string, + options: { + documentId?: string; + routeHint?: CultMeshRouteHint; + sourceId?: string; + timeoutMs?: number; + pollMs?: number; + messageIdPrefix?: string; + } = {}, + ): CultMeshDocumentHandle { + return cultMeshDocumentFromPeerSnapshot(peer, schemaId, recordKey, options); + } + public static globalDocumentFromCache( cache: CultCache, definition: TDefinition, @@ -4595,6 +4656,112 @@ function normalizeRudpDocumentPut( }; } +async function requestCultNetRawSnapshotDocument( + peerOrProvider: CultNetPeer | (() => CultNetPeer | Promise), + schemaId: string, + recordKey: string, + options: { + timeoutMs?: number; + messageIdPrefix?: string; + } = {}, +): Promise { + const peer = await resolveCultNetPeer(peerOrProvider); + const messageId = `${options.messageIdPrefix ?? "cultmesh-snapshot"}:${Date.now()}:${Math.random().toString(16).slice(2)}`; + const response = await requestCultNetRawSnapshot(peer, [recordKey], { + messageId, + timeoutMs: options.timeoutMs, + }); + const candidates = response.documents.filter(candidate => candidate.recordKey === recordKey); + const document = candidates.find(candidate => candidate.schemaId === schemaId) ?? candidates[0]; + if (!document) { + throw new Error(`CultMesh peer snapshot did not return ${schemaId} at ${recordKey}.`); + } + return document; +} + +async function requestCultNetRawSnapshot( + peer: CultNetPeer, + recordKeys: readonly string[], + options: { + messageId: string; + timeoutMs?: number; + }, +): Promise { + return new Promise((resolve, reject) => { + const cleanup = (): void => { + clearTimeout(timer); + peer.off("message", onMessage); + peer.off("invalidMessage", onInvalidMessage); + peer.off("error", onError); + peer.off("close", onClose); + }; + const rejectWith = (error: Error): void => { + cleanup(); + reject(error); + }; + const onMessage = (message: CultNetMessage): void => { + if (isCultNetErrorMessage(message)) { + cleanup(); + reject(new Error(message.error)); + return; + } + if (message.schemaVersion !== "cultnet.snapshot_response_raw.v0" || + message.messageId !== options.messageId) { + return; + } + + cleanup(); + resolve(message); + }; + const onInvalidMessage = (error: Error): void => rejectWith(error); + const onError = (error: Error): void => rejectWith(error); + const onClose = (): void => rejectWith(new Error("CultNet peer closed before snapshot response.")); + const timer = setTimeout( + () => rejectWith(new Error(`Timed out waiting for CultNet snapshot ${options.messageId}.`)), + options.timeoutMs ?? 4_000, + ); + + peer.on("message", onMessage); + peer.on("invalidMessage", onInvalidMessage); + peer.on("error", onError); + peer.on("close", onClose); + peer.sendSnapshotRequest({ + schemaVersion: "cultnet.snapshot_request.v0", + messageId: options.messageId, + recordKeys: [...recordKeys], + }); + }); +} + +async function resolveCultNetPeer( + peerOrProvider: CultNetPeer | (() => CultNetPeer | Promise), +): Promise { + return typeof peerOrProvider === "function" + ? await peerOrProvider() + : peerOrProvider; +} + +function decodeCultNetRawDocumentPayload(document: CultNetRawDocumentRecord): unknown { + if (document.payloadEncoding !== "messagepack") { + throw new Error(`Unsupported CultNet raw document payload encoding ${document.payloadEncoding}.`); + } + return decode(toUint8Array(document.payload)); +} + +function isCultNetErrorMessage(message: CultNetMessage): message is CultNetErrorMessage { + return message.schemaVersion === "cultnet.error.v0"; +} + +function toUint8Array(value: unknown): Uint8Array { + if (value instanceof Uint8Array) { + return value; + } + if (Array.isArray(value)) { + return Uint8Array.from(value); + } + throw new Error("CultNet raw document payload was not binary."); +} + async function bindRudpSocket(options: CultMeshRudpSocketOptions): Promise { const socket = createSocket("udp4"); const host = options.bindHost ?? "127.0.0.1"; diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index dbb1e1d2..48fd7dd7 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -1832,6 +1832,87 @@ test("CultMesh TS publishes one registered document to a RUDP catalog", async () } }); +test("CultMesh TS reads remote RUDP snapshots through document handles", async () => { + const connectionId = 0x10203048; + const node = await CultMesh.startNode( + join(await mkdtemp(join(tmpdir(), "cultmesh-ts-rudp-snapshot-")), "node.ccmp"), + { + documents: [noteDocument], + }, + ); + await node.put(noteDocument, "note:remote", { + noteId: "note:remote", + body: "remote handles feel local", + }); + + const server = CultMesh.createRudpDocumentServer( + "cultmesh-ts-rudp-snapshot-server", + connectionId, + { + documents: new CultNetDocumentRegistry([ + defineCultNetDocumentBinding({ definition: noteDocument }), + ]), + getCache: () => node.cache, + bindHost: "127.0.0.1", + bindPort: 0, + resendDelayMs: 25, + resendPollMs: 5, + maxFragmentBytes: 1024, + maxPendingReliablePackets: 16, + }, + ); + + let peer: CultNetPeer | undefined; + try { + await server.start(); + peer = await CultMesh.createRudpPeer( + "cultmesh-ts-rudp-snapshot-client", + connectionId, + `rudp://127.0.0.1:${server.bind.port}`, + { + resendDelayMs: 25, + resendPollMs: 5, + maxFragmentBytes: 1024, + maxPendingReliablePackets: 16, + connectTimeoutMs: 1_000, + }, + ); + + const document = CultMesh.documentFromPeerSnapshot( + peer, + "cultmesh.note.v0", + "note:remote", + { + documentId: "note:remote", + timeoutMs: 1_000, + }, + ); + + assert.deepEqual(await document.latest(), { + noteId: "note:remote", + body: "remote handles feel local", + }); + + const alias = CultMesh.documentFromPeerSnapshot( + peer, + "cultmesh.note.alias.v1", + "note:remote", + { + documentId: "note:remote.alias", + timeoutMs: 1_000, + }, + ); + + assert.deepEqual(await alias.latest(), { + noteId: "note:remote", + body: "remote handles feel local", + }); + } finally { + peer?.close(); + server.close(); + } +}); + test("CultMesh TS negotiates streaming frame body transports explicitly", () => { const streams = CultMesh.createStreamCatalog(); const streamUpdates: string[] = []; diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index b18f1dd5..3f46eef7 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -330,6 +330,13 @@ unsubscribe(); const pilots = station.collection(playerShipDocument); const currentPilots = await pilots.latest(); const stopRoster = pilots.watchChanges(change => refreshRoster(change)); + +const remoteHealth = CultMesh.documentFromPeerSnapshot( + () => daemonPeer, + "gamecult.aetheria.daemon_health.v1", + "daemon:aetheria.health.v1", +); +const health = await remoteHealth.latest(); ``` Projected, remote, shared-memory, IPC, WASM, and local-cache documents all use @@ -458,6 +465,15 @@ helpers parse endpoint contact hints and choose authorized peers; `GameCult.Networking` still owns the RUDP packet codec, session state, resend, fragmentation, and channel semantics. +TypeScript runtimes can also wrap a CultNet peer snapshot as a normal document +handle with `CultMesh.documentFromPeerSnapshot(...)`. The caller supplies the +peer or peer factory, schema id, and record key; CultMesh sends the snapshot +request, waits for the matching response, decodes the raw MessagePack document, +and returns the same `latest()`/`watch()` document surface used by local cache +and single-file documents. Exact schema ids are preferred, but record-key +publications remain readable when two runtimes alias the same logical document +through different schema names. + ### Shard Authority Each shard has one primary writer for now. Non-primary writes are rejected or