Add TS typed peer snapshot node sync

This commit is contained in:
Meta Vens 2026-06-27 23:14:18 +01:00
commit 743979eca7
3 changed files with 157 additions and 0 deletions

View file

@ -316,6 +316,18 @@ const health = CultMesh.documentFromPeerSnapshot(
);
const latestHealth = await health.latest();
const syncedHealth = await station.syncDocumentFromPeerSnapshot(
() => client,
daemonHealthUiDocument,
"daemon:aetheria.health.v1",
{
timeoutMs: 5_000,
},
);
const localHealth = await station
.document(daemonHealthUiDocument, "daemon:aetheria.health.v1")
.latest();
```
`CultMesh.createRudpPeerForPeer(...)` remains available for already trusted
@ -336,6 +348,14 @@ still read the publication with one CultMesh call. Pass a CultCache document
definition instead of a raw schema id when you want a typed handle whose
`latest()` result is parsed through that definition.
Use `node.syncDocumentFromPeerSnapshot(...)` or
`CultMesh.syncDocumentFromPeerSnapshot(...)` when a runtime should hydrate its
local node from the remote snapshot. The helper requests the raw snapshot,
applies it through the node's document registry, and returns the requested
typed definition. Same-schema aliases stay local after the call: the caller can
keep using `node.document(aliasDefinition, key)` or `node.reactiveDocument(...)`
without re-threading the RUDP peer.
The same typed-definition overload is available for
`CultMesh.documentFromStore(...)` and `CultMesh.documentFromSingleFile(...)`, so
local schema publications can be read as typed reactive documents without

View file

@ -3815,6 +3815,46 @@ export class CultMeshNode {
.reactive<CultCacheDocumentValue<TDefinition> & object>(reactiveOptions);
}
public async syncDocumentFromPeerSnapshot<TDefinition extends AnyCultCacheDocumentDefinition>(
peer: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),
definition: TDefinition,
key: string,
options: {
timeoutMs?: number;
messageIdPrefix?: string;
} = {},
): Promise<CultCacheDocumentValue<TDefinition>> {
const registered = this.resolveCacheDefinition(definition);
const schema = cultMeshSchemaFromDefinition(definition);
const schemaId = schema.schemaId ?? schema.type;
if (!schemaId) {
throw new Error("CultMesh synced document requires a schema id or document type.");
}
const document = await requestCultNetRawSnapshotDocument(
peer,
schemaId,
key,
{
timeoutMs: options.timeoutMs,
messageIdPrefix: options.messageIdPrefix ?? key,
accepts: record => {
try {
definition.schema.parse(decodeCultNetRawDocumentPayload(record));
return true;
} catch {
return false;
}
},
},
);
await this.documents.applyRawDocumentPutMessage(this.cache, {
schemaVersion: "cultnet.document_put_raw.v0",
messageId: options.messageIdPrefix ?? `sync:${key}`,
document,
});
return this.parseDocumentValue(definition, this.cache.getRequired(registered, key));
}
public globalDocument<TDefinition extends AnyCultCacheDocumentDefinition>(
definition: TDefinition,
options: {
@ -4792,6 +4832,19 @@ export class CultMesh {
return node.reactiveDocument(definition, key, options);
}
public static syncDocumentFromPeerSnapshot<TDefinition extends AnyCultCacheDocumentDefinition>(
node: CultMeshNode,
peer: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),
definition: TDefinition,
key: string,
options: {
timeoutMs?: number;
messageIdPrefix?: string;
} = {},
): Promise<CultCacheDocumentValue<TDefinition>> {
return node.syncDocumentFromPeerSnapshot(peer, definition, key, options);
}
public static documents(
...documents: readonly CultMeshDocumentHandle<any>[]
): CultMeshDocumentCatalog {

View file

@ -2530,6 +2530,90 @@ test("CultMesh TS reads remote RUDP snapshots through document handles", async (
}
});
test("CultMesh TS syncs remote RUDP snapshots into a local node with same-schema aliases", async () => {
const connectionId = 0x1020304a;
const source = await CultMesh.startNode(
join(await mkdtemp(join(tmpdir(), "cultmesh-ts-rudp-sync-source-")), "node.ccmp"),
{
documents: [noteDocument],
},
);
await source.put(noteDocument, "note:sync-remote", {
noteId: "note:sync-remote",
body: "synced once, read as local",
});
const target = await CultMesh.startNode(
join(await mkdtemp(join(tmpdir(), "cultmesh-ts-rudp-sync-target-")), "node.ccmp"),
{
documents: [noteDocument],
},
);
const server = CultMesh.createRudpDocumentServer(
"cultmesh-ts-rudp-sync-server",
connectionId,
{
documents: new CultNetDocumentRegistry([
defineCultNetDocumentBinding({ definition: noteDocument }),
]),
getCache: () => source.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-sync-client",
connectionId,
`rudp://127.0.0.1:${server.bind.port}`,
{
resendDelayMs: 25,
resendPollMs: 5,
maxFragmentBytes: 1024,
maxPendingReliablePackets: 16,
connectTimeoutMs: 1_000,
},
);
const synced = await target.syncDocumentFromPeerSnapshot(
peer,
noteAliasDocument,
"note:sync-remote",
{
timeoutMs: 1_000,
},
);
const facadeSynced = await CultMesh.syncDocumentFromPeerSnapshot(
target,
peer,
noteAliasDocument,
"note:sync-remote",
{
timeoutMs: 1_000,
},
);
assert.deepEqual(synced, {
noteId: "note:sync-remote",
body: "synced once, read as local",
});
assert.deepEqual(facadeSynced, synced);
assert.equal(target.getRequired(noteDocument, "note:sync-remote").body, "synced once, read as local");
assert.equal(target.getRequired(noteAliasDocument, "note:sync-remote").body, "synced once, read as local");
assert.equal((await target.document(noteAliasDocument, "note:sync-remote").latest()).body, "synced once, read as local");
} finally {
peer?.close();
server.close();
}
});
test("CultMesh TS peer snapshot handles choose compatible foreign schema payloads", async () => {
const connectionId = 0x10203049;
const node = await CultMesh.startNode(