diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index e968d7e0..37b6b554 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -2549,6 +2549,16 @@ export function cultMeshDocumentFromPeerSnapshot( { timeoutMs: options.timeoutMs, messageIdPrefix: options.messageIdPrefix ?? documentId, + accepts: typeof schemaOrDefinition === "string" + ? undefined + : record => { + try { + schemaOrDefinition.schema.parse(decodeCultNetRawDocumentPayload(record)); + return true; + } catch { + return false; + } + }, }, ), ), @@ -5718,6 +5728,7 @@ async function requestCultNetRawSnapshotDocument( options: { timeoutMs?: number; messageIdPrefix?: string; + accepts?: (document: CultNetRawDocumentRecord) => boolean; } = {}, ): Promise { const peer = await resolveCultNetPeer(peerOrProvider); @@ -5727,7 +5738,10 @@ async function requestCultNetRawSnapshotDocument( timeoutMs: options.timeoutMs, }); const candidates = response.documents.filter(candidate => candidate.recordKey === recordKey); - const document = candidates.find(candidate => candidate.schemaId === schemaId) ?? candidates[0]; + const document = + candidates.find(candidate => candidate.schemaId === schemaId) ?? + (options.accepts ? candidates.find(options.accepts) : undefined) ?? + candidates[0]; if (!document) { throw new Error(`CultMesh peer snapshot did not return ${schemaId} at ${recordKey}.`); } diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 28484033..2924a81c 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -34,6 +34,24 @@ const noteAliasDocument = defineDocumentType({ name: "noteId", }); +const foreignNoteDocument = defineDocumentType({ + type: "runtime.generated.cultmesh.note", + schemaId: "runtime.generated.cultmesh.note.ui.42", + schema: z.object({ + noteId: z.string(), + body: z.string(), + }), + name: "noteId", +}); + +const incompatibleNoteDocument = defineDocumentType({ + type: "runtime.generated.incompatible-note", + schemaId: "runtime.generated.incompatible-note.v1", + schema: z.object({ + nope: z.string(), + }), +}); + async function waitFor(predicate: () => boolean, description: string): Promise { const startedAt = Date.now(); while (!predicate()) { @@ -2370,6 +2388,76 @@ test("CultMesh TS reads remote RUDP snapshots through document handles", async ( } }); +test("CultMesh TS peer snapshot handles choose compatible foreign schema payloads", async () => { + const connectionId = 0x10203049; + const node = await CultMesh.startNode( + join(await mkdtemp(join(tmpdir(), "cultmesh-ts-rudp-foreign-snapshot-")), "node.ccmp"), + { + documents: [incompatibleNoteDocument, foreignNoteDocument], + }, + ); + await node.put(incompatibleNoteDocument, "note:foreign", { + nope: "not the requested shape", + }); + await node.put(foreignNoteDocument, "note:foreign", { + noteId: "note:foreign", + body: "foreign schema id, local shape", + }); + + const server = CultMesh.createRudpDocumentServer( + "cultmesh-ts-rudp-foreign-snapshot-server", + connectionId, + { + documents: new CultNetDocumentRegistry([ + defineCultNetDocumentBinding({ definition: incompatibleNoteDocument }), + defineCultNetDocumentBinding({ definition: foreignNoteDocument }), + ]), + 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-foreign-snapshot-client", + connectionId, + `rudp://127.0.0.1:${server.bind.port}`, + { + resendDelayMs: 25, + resendPollMs: 5, + maxFragmentBytes: 1024, + maxPendingReliablePackets: 16, + connectTimeoutMs: 1_000, + }, + ); + + const alias = CultMesh.documentFromPeerSnapshot( + peer, + noteAliasDocument, + "note:foreign", + { + documentId: "note:foreign.alias", + timeoutMs: 1_000, + }, + ); + + assert.deepEqual(await alias.latest(), { + noteId: "note:foreign", + body: "foreign schema id, local shape", + }); + } finally { + peer?.close(); + server.close(); + } +}); + test("CultMesh TS negotiates streaming frame body transports explicitly", () => { const streams = CultMesh.createStreamCatalog(); const streamUpdates: string[] = []; diff --git a/packages/cultnet-ts/src/replication.ts b/packages/cultnet-ts/src/replication.ts index 600a21e3..9925fd1d 100644 --- a/packages/cultnet-ts/src/replication.ts +++ b/packages/cultnet-ts/src/replication.ts @@ -238,11 +238,14 @@ export class CultNetDocumentRegistry { cache: CultCache, message: CultNetDocumentPutRawMessage, ): Promise { - const binding = this.#requireSchemaBinding(message.document.schemaId); + const resolution = this.#resolveRawDocumentBinding(message.document); + const { binding } = resolution; return cache.putEnvelope(binding.definition, { key: message.document.recordKey, type: binding.definition.type, - schemaId: message.document.schemaId, + schemaId: resolution.preserveIncomingSchemaId + ? message.document.schemaId + : schemaIdForBinding(binding), payload: new Uint8Array(message.document.payload), storedAt: message.document.storedAt, }); @@ -292,6 +295,35 @@ export class CultNetDocumentRegistry { return binding; } + #resolveRawDocumentBinding(document: CultNetRawDocumentRecord): { + binding: CultNetDocumentBinding; + preserveIncomingSchemaId: boolean; + } { + const exact = this.getBySchemaId(document.schemaId); + if (exact) { + return { binding: exact, preserveIncomingSchemaId: true }; + } + + const payload = new Uint8Array(document.payload); + for (const binding of this.#uniqueBindings()) { + try { + decodeDocumentValue(binding.definition, payload); + return { binding, preserveIncomingSchemaId: false }; + } catch { + // Keep looking; a foreign schema id is acceptable only when the payload validates locally. + } + } + + throw new Error(`No CultNet document binding is registered for schema "${document.schemaId}".`); + } + + #uniqueBindings(): CultNetDocumentBinding[] { + return [...new Set([ + ...this.#schemaBindings.values(), + ...this.#bindings.values(), + ])]; + } + #createRawDocumentRecord(envelope: CultCacheEnvelope): CultNetRawDocumentRecord { const binding = this.#requireBinding(envelope.type); return { diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index c2a8a8d8..838dfbb2 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1215,6 +1215,57 @@ test("CultNet raw replication preserves CultCache payload bytes for bit-compatib } }); +test("CultNet raw replication applies compatible foreign-schema snapshots", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "cultnetts-foreign-schema-raw-")); + + try { + const documentDefinition = defineDocumentType({ + type: "cultnet.note", + schemaId: "cultnet.note.v1", + schemaName: "cultnet.note", + schemaVersion: "cultnet.note.v1", + schema: z.object({ + schema_version: z.string(), + noteId: z.string(), + body: z.string(), + }), + name: "noteId", + }); + const registry = new CultNetDocumentRegistry([ + defineCultNetDocumentBinding({ definition: documentDefinition }), + ]); + const originCache = CultCache.builder() + .withDocumentType(documentDefinition) + .withGenericStore(new SingleFileMessagePackBackingStore(join(tempDir, "origin.msgpack"))) + .build(); + const targetCache = CultCache.builder() + .withDocumentType(documentDefinition) + .withGenericStore(new SingleFileMessagePackBackingStore(join(tempDir, "target.msgpack"))) + .build(); + + await originCache.put(documentDefinition, "note:foreign-schema", { + schema_version: "cultnet.note.v1", + noteId: "note:foreign-schema", + body: "accepted by payload shape", + }); + const rawSnapshot = registry.createRawSnapshotResponse(originCache, "foreign-schema-raw"); + rawSnapshot.documents[0]!.schemaId = "runtime.generated.cultnet.note.ui.99"; + + await registry.applyRawSnapshotResponse(targetCache, rawSnapshot); + + assert.equal( + targetCache.getRequired(documentDefinition, "note:foreign-schema").body, + "accepted by payload shape", + ); + assert.equal( + targetCache.getRequiredEnvelope(documentDefinition, "note:foreign-schema").schemaId, + "cultnet.note.v1", + ); + } finally { + rmSync(tempDir, { recursive: true, force: true }); + } +}); + test("CultNet interop slot compatibility defaults missing trailing fields and rejects mismatched slots", () => { const note: InteropNote = { schemaVersion: INTEROP_SCHEMA_VERSION,