Expand CultMesh reactive sync edge tests

This commit is contained in:
Meta Vens 2026-06-27 20:05:56 +01:00
commit 8a88ae547d
3 changed files with 293 additions and 0 deletions

View file

@ -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<typeof current>({
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<void>(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<void>(release => {
releaseFirstPrediction = release;
});
}
current = value;
},
},
);
const reactive = document.reactive<typeof current>({
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<typeof current>({
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, {

View file

@ -1338,6 +1338,8 @@ namespace GameCult.Mesh
lock (_gate)
{
_flushing = false;
if (_document.CanSet)
DetectLocalChangesLocked();
shouldFlushAgain = _flushQueued || _dirty;
_flushQueued = false;
}

View file

@ -1848,6 +1848,162 @@ public sealed class CultMeshStreamingTests
current.Revision.Should().Be(3);
}
[Test]
public async Task ReactiveDocument_DebouncesBurstUpdatesIntoOnePrediction()
{
var subject = new Subject<MeshNoteDocument>();
var current = new MeshNoteDocument
{
Schema = "tests.mesh_note.v1",
Text = "initial",
Revision = 1
};
var predictions = new List<MeshNoteDocument>();
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<MeshNoteDocument>();
var current = new MeshNoteDocument
{
Schema = "tests.mesh_note.v1",
Text = "initial",
Revision = 1
};
var allowFirstWrite = new TaskCompletionSource<bool>();
var firstWriteStarted = new TaskCompletionSource<bool>();
var predictions = new List<MeshNoteDocument>();
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<MeshNoteDocument>();
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()
{