From 8a88ae547dbcf68c52c4dc05a44aa2da86773ab9 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 20:05:56 +0100 Subject: [PATCH] Expand CultMesh reactive sync edge tests --- packages/cultmesh-ts/test/cultmesh.test.ts | 135 +++++++++++++++ src/GameCult.Mesh/CultMeshPrimitives.cs | 2 + .../CultMeshStreamingTests.cs | 156 ++++++++++++++++++ 3 files changed, 293 insertions(+) diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 82ec5047..8266bc24 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -302,6 +302,141 @@ test("CultMesh TS reactive documents store reconciliation deltas after mispredic reactive.dispose(); }); +test("CultMesh TS reactive documents coalesce same-frame member writes", async () => { + let current = { + noteId: "note:reactive-frame", + body: "initial", + }; + const predictions: string[] = []; + const previousRequestAnimationFrame = globalThis.requestAnimationFrame; + const frameCallbacks: FrameRequestCallback[] = []; + globalThis.requestAnimationFrame = callback => { + frameCallbacks.push(callback); + return frameCallbacks.length; + }; + const document = CultMesh.document( + "cultmesh.note:reactive-frame", + noteDocument, + async () => current, + { + submitPrediction: async (_context, value) => { + predictions.push(value.body); + current = value; + }, + }, + ); + + const reactive = document.reactive({ + watch: false, + }); + try { + await reactive.ready; + reactive.current.body = "frame-1"; + reactive.current.body = "frame-2"; + reactive.current.body = "frame-3"; + + assert.equal(frameCallbacks.length, 1); + frameCallbacks[0](performance.now()); + await waitFor(() => predictions.length === 1, "same-frame prediction"); + + assert.deepEqual(predictions, ["frame-3"]); + assert.equal(current.body, "frame-3"); + } finally { + reactive.dispose(); + if (previousRequestAnimationFrame) { + globalThis.requestAnimationFrame = previousRequestAnimationFrame; + } else { + delete (globalThis as unknown as { + requestAnimationFrame?: (callback: FrameRequestCallback) => number; + }).requestAnimationFrame; + } + } +}); + +test("CultMesh TS reactive documents queue edits made while a prediction is in flight", async () => { + let current = { + noteId: "note:reactive-in-flight", + body: "initial", + }; + let releaseFirstPrediction!: () => void; + let resolveFirstPredictionStarted!: () => void; + const firstPredictionStarted = new Promise(resolve => { + resolveFirstPredictionStarted = resolve; + }); + const predictions: typeof current[] = []; + const document = CultMesh.document( + "cultmesh.note:reactive-in-flight", + noteDocument, + async () => current, + { + submitPrediction: async (_context, value) => { + predictions.push(value); + if (predictions.length === 1) { + resolveFirstPredictionStarted(); + await new Promise(release => { + releaseFirstPrediction = release; + }); + } + current = value; + }, + }, + ); + + const reactive = document.reactive({ + watch: false, + debounceMs: 0, + }); + await reactive.ready; + reactive.current.body = "first"; + await firstPredictionStarted; + reactive.current.body = "second"; + releaseFirstPrediction(); + await waitFor(() => predictions.length === 2, "in-flight queued prediction"); + + assert.deepEqual(predictions.map(value => value.body), ["first", "second"]); + assert.equal(current.body, "second"); + reactive.dispose(); +}); + +test("CultMesh TS reactive read-only documents surface mutation failures and keep watching", async () => { + let current = { + noteId: "note:reactive-readonly", + body: "initial", + }; + let watcher: ((value: typeof current) => void) | undefined; + const errors: string[] = []; + const document = CultMesh.document( + "cultmesh.note:reactive-readonly", + noteDocument, + async () => current, + { + watchDocument: (_context, callback) => { + watcher = callback; + return () => { + watcher = undefined; + }; + }, + }, + ); + + const reactive = document.reactive({ + onError: error => errors.push(String((error as Error).message ?? error)), + }); + await reactive.ready; + reactive.current.body = "local-readonly-edit"; + await waitFor(() => errors.length === 1, "read-only mutation error"); + + current = { + noteId: "note:reactive-readonly", + body: "canonical-after-error", + }; + watcher?.(current); + + assert.match(errors[0], /does not support mutation/); + assert.equal(reactive.current.body, "canonical-after-error"); + reactive.dispose(); +}); + test("CultMesh TS document handles read schema publications from single-file stores", async () => { const filePath = join(await mkdtemp(join(tmpdir(), "cultmesh-ts-publication-")), "publication.ccmp"); const node = await CultMesh.startNode(filePath, { diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index fb34c000..70ab2737 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -1338,6 +1338,8 @@ namespace GameCult.Mesh lock (_gate) { _flushing = false; + if (_document.CanSet) + DetectLocalChangesLocked(); shouldFlushAgain = _flushQueued || _dirty; _flushQueued = false; } diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 4a062e75..2cfa18b8 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -1848,6 +1848,162 @@ public sealed class CultMeshStreamingTests current.Revision.Should().Be(3); } + [Test] + public async Task ReactiveDocument_DebouncesBurstUpdatesIntoOnePrediction() + { + var subject = new Subject(); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "initial", + Revision = 1 + }; + var predictions = new List(); + var handle = CultMesh.Document( + "mesh.note.reactive.burst", + CultMesh.Verse("starbridge", "pilot-a"), + _ => Task.FromResult(current), + _ => subject, + value => + { + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }, + value => + { + predictions.Add(value); + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }); + + using var reactive = await handle.ReactiveAsync( + new CultMeshReactiveDocumentOptions + { + FlushDelay = TimeSpan.FromMilliseconds(35) + }); + + for (var i = 0; i < 5; i++) + { + reactive.Update(document => + { + document.Text = $"burst-{i}"; + document.Revision++; + }); + } + + await WaitForAsync(() => predictions.Count == 1); + + predictions[0].Text.Should().Be("burst-4"); + predictions[0].Revision.Should().Be(6); + reactive.IsDirty.Should().BeFalse(); + } + + [Test] + public async Task ReactiveDocument_DirectEditsDuringInFlightFlushQueueLatestPrediction() + { + var subject = new Subject(); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "initial", + Revision = 1 + }; + var allowFirstWrite = new TaskCompletionSource(); + var firstWriteStarted = new TaskCompletionSource(); + var predictions = new List(); + var handle = CultMesh.Document( + "mesh.note.reactive.direct-in-flight", + CultMesh.Verse("starbridge", "pilot-a"), + _ => Task.FromResult(current), + _ => subject, + value => + { + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }, + async value => + { + predictions.Add(value); + if (predictions.Count == 1) + { + firstWriteStarted.SetResult(true); + await allowFirstWrite.Task.ConfigureAwait(false); + } + + current = value; + subject.OnNext(value); + }); + + using var reactive = await handle.ReactiveAsync( + new CultMeshReactiveDocumentOptions + { + FlushDelay = TimeSpan.FromMinutes(1) + }); + reactive.Current.Text = "first-direct"; + reactive.Current.Revision = 2; + var flush = reactive.FlushAsync(); + await firstWriteStarted.Task; + + reactive.Current.Text = "second-direct"; + reactive.Current.Revision = 3; + allowFirstWrite.SetResult(true); + await flush; + + predictions.Should().HaveCount(2); + predictions[0].Text.Should().Be("first-direct"); + predictions[1].Text.Should().Be("second-direct"); + reactive.IsDirty.Should().BeFalse(); + } + + [Test] + public async Task ReactiveDocument_CanonicalEchoClearsOutstandingPredictionWithoutDelta() + { + var subject = new Subject(); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "initial", + Revision = 1 + }; + var handle = CultMesh.Document( + "mesh.note.reactive.canonical-echo", + CultMesh.Verse("starbridge", "pilot-a"), + _ => Task.FromResult(current), + _ => subject, + value => + { + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }, + value => + { + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }); + + using var reactive = await handle.ReactiveAsync( + new CultMeshReactiveDocumentOptions + { + FlushDelay = TimeSpan.FromMinutes(1) + }); + reactive.Update(document => + { + document.Text = "accepted"; + document.Revision = 2; + }); + + await reactive.FlushAsync(); + + reactive.Current.Text.Should().Be("accepted"); + reactive.Reconciliation.Should().BeNull(); + reactive.IsDirty.Should().BeFalse(); + } + [Test] public async Task DocumentHandle_ReadsSchemaPublicationsFromSingleFileStores() {