From ba680020b9e645f979a74fcadc1eb138f67afbef Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 18 Aug 2026 00:44:41 +0200 Subject: [PATCH] Prove reactive scheduling scales with edits --- packages/cultcache-py/tests/test_cultcache.py | 67 +++++++++++++++- packages/cultmesh-ts/src/index.ts | 29 ++++--- packages/cultmesh-ts/test/cultmesh.test.ts | 76 ++++++++++++++++++ scripts/verify-reactive-document-scaling.ps1 | 75 ++++++++++++++++++ src/GameCult.Mesh/CultMeshPrimitives.cs | 56 +++++++++++-- .../docs/document-write-authority.md | 14 ++++ .../CultMeshStreamingTests.cs | 79 +++++++++++++++++++ 7 files changed, 378 insertions(+), 18 deletions(-) create mode 100644 scripts/verify-reactive-document-scaling.ps1 diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index d00a05ab..f753b12e 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -16,7 +16,7 @@ import io from dataclasses import asdict, dataclass from datetime import UTC, datetime, timedelta from pathlib import Path -from typing import Callable +from typing import Any, Callable from uuid import uuid4 from cultcache_py import ( @@ -2580,6 +2580,71 @@ class CultCacheTests(unittest.TestCase): finally: reactive.dispose() + def test_cultmesh_reactive_scheduling_scales_with_changed_documents_only(self) -> None: + document = define_document_type( + "mesh.reactive_scale_note", + encode=lambda value: value, + decode=lambda value: value, + ) + + for document_count in (1, 100, 1000): + node = create_node(runtime_id=f"python-reactive-scale-{document_count}") + node.register_document(document) + for index in range(document_count): + node.put(document, f"note:{index}", {"body": "initial", "revision": 1}) + + scheduled_timers: list[Any] = [] + + class FakeTimer: + def __init__(self, interval: float, callback: Callable[[], None]) -> None: + self.interval = interval + self.callback = callback + self.daemon = False + self.cancelled = False + self.started = False + scheduled_timers.append(self) + + def start(self) -> None: + self.started = True + + def cancel(self) -> None: + self.cancelled = True + + reactive_documents: list[Any] = [] + with patch("cultmesh_py.node.threading.Timer", FakeTimer): + try: + for index in range(document_count): + reactive_documents.append( + node.authoritative_writer(document, f"note:{index}").reactive() + ) + + self.assertEqual( + len(scheduled_timers), + 0, + f"{document_count} idle documents must schedule no timers", + ) + + changed_document_count = max(1, document_count // 100) + for index in range(changed_document_count): + reactive_documents[index].update( + lambda draft, changed=index: draft.update( + body=f"changed-{changed}", + revision=2, + ) + ) + + self.assertEqual(len(scheduled_timers), changed_document_count) + self.assertTrue(all(timer.started for timer in scheduled_timers)) + self.assertEqual( + sum(reactive.is_dirty for reactive in reactive_documents), + changed_document_count, + ) + finally: + for reactive in reactive_documents: + reactive.dispose() + + self.assertTrue(all(timer.cancelled for timer in scheduled_timers)) + def test_cultmesh_observed_document_cannot_publish_borrowed_mutation(self) -> None: document = define_document_type( "mesh.observed_note", diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 5d81425e..7300ad01 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -1140,7 +1140,7 @@ class CultMeshReactiveDocumentHandle #disposed = false; #pendingSet: Promise = Promise.resolve(); #flushScheduled = false; - #flushTimer: ReturnType | undefined; + #cancelScheduledFlush: (() => void) | undefined; #unsubscribe: CultMeshUnsubscribe | undefined; #dirty = false; #lastPredicted: TDocument | undefined; @@ -1238,10 +1238,8 @@ class CultMeshReactiveDocumentHandle public dispose(): void { this.#disposed = true; - if (this.#flushTimer !== undefined) { - clearTimeout(this.#flushTimer); - this.#flushTimer = undefined; - } + this.#cancelScheduledFlush?.(); + this.#cancelScheduledFlush = undefined; this.#flushScheduled = false; this.#unsubscribe?.(); this.#unsubscribe = undefined; @@ -1294,28 +1292,35 @@ class CultMeshReactiveDocumentHandle const debounceMs = this.#debounceMs; if (debounceMs > 0) { - this.#flushTimer = setTimeout(flush, debounceMs); + const timer = setTimeout(flush, debounceMs); + this.#cancelScheduledFlush = () => clearTimeout(timer); return; } const requestFrame = (globalThis as { requestAnimationFrame?: (callback: FrameRequestCallback) => number; + cancelAnimationFrame?: (handle: number) => void; }).requestAnimationFrame; if (typeof requestFrame === "function") { - requestFrame(() => flush()); + const frame = requestFrame(() => flush()); + const cancelFrame = (globalThis as { + cancelAnimationFrame?: (handle: number) => void; + }).cancelAnimationFrame; + this.#cancelScheduledFlush = typeof cancelFrame === "function" + ? () => cancelFrame(frame) + : undefined; return; } - this.#flushTimer = setTimeout(flush, 0); + const timer = setTimeout(flush, 0); + this.#cancelScheduledFlush = () => clearTimeout(timer); } private flushNow(): void { if (!this.#flushScheduled && !this.#dirty) return; this.#flushScheduled = false; - if (this.#flushTimer !== undefined) { - clearTimeout(this.#flushTimer); - this.#flushTimer = undefined; - } + this.#cancelScheduledFlush?.(); + this.#cancelScheduledFlush = undefined; if (this.#disposed || !this.#dirty) return; const snapshot = cloneCultMeshDocument(this.#current); this.#dirty = false; diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 709dc5b8..18e40a32 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -572,6 +572,82 @@ test("CultMesh TS reactive documents coalesce same-frame explicit updates", asyn } }); +test("CultMesh TS reactive scheduling scales with changed documents only", async () => { + const previousRequestAnimationFrame = globalThis.requestAnimationFrame; + const previousCancelAnimationFrame = globalThis.cancelAnimationFrame; + const frameCallbacks = new Map(); + let nextFrame = 1; + globalThis.requestAnimationFrame = callback => { + const frame = nextFrame++; + frameCallbacks.set(frame, callback); + return frame; + }; + globalThis.cancelAnimationFrame = frame => { + frameCallbacks.delete(frame); + }; + + try { + for (const documentCount of [1, 100, 1000]) { + const reactiveDocuments = []; + for (let index = 0; index < documentCount; index++) { + let current = { + noteId: `note:scale:${index}`, + body: "initial", + }; + const document = CultMesh.document( + `cultmesh.note:scale:${index}`, + noteDocument, + async () => current, + { + submitPrediction: async (_context, value) => { + current = value; + }, + }, + ); + reactiveDocuments.push(document.predictionWriter().reactive({ watch: false })); + } + + await Promise.all(reactiveDocuments.map(document => document.ready)); + assert.equal(frameCallbacks.size, 0, `${documentCount} idle documents scheduled no work`); + + const changedDocumentCount = Math.max(1, Math.floor(documentCount / 100)); + for (let index = 0; index < changedDocumentCount; index++) { + reactiveDocuments[index].update(value => { + value.body = `changed-${index}`; + }); + } + + assert.equal( + frameCallbacks.size, + changedDocumentCount, + `${documentCount} documents scheduled only the ${changedDocumentCount} changed documents`, + ); + assert.equal( + reactiveDocuments.filter(document => document.dirty).length, + changedDocumentCount, + ); + + for (const reactive of reactiveDocuments) reactive.dispose(); + assert.equal(frameCallbacks.size, 0, "dispose cancels every scheduled frame flush"); + } + } finally { + if (previousRequestAnimationFrame) { + globalThis.requestAnimationFrame = previousRequestAnimationFrame; + } else { + delete (globalThis as unknown as { + requestAnimationFrame?: (callback: FrameRequestCallback) => number; + }).requestAnimationFrame; + } + if (previousCancelAnimationFrame) { + globalThis.cancelAnimationFrame = previousCancelAnimationFrame; + } else { + delete (globalThis as unknown as { + cancelAnimationFrame?: (handle: number) => void; + }).cancelAnimationFrame; + } + } +}); + test("CultMesh TS reactive documents queue edits made while a prediction is in flight", async () => { let current = { noteId: "note:reactive-in-flight", diff --git a/scripts/verify-reactive-document-scaling.ps1 b/scripts/verify-reactive-document-scaling.ps1 new file mode 100644 index 00000000..1729ed4d --- /dev/null +++ b/scripts/verify-reactive-document-scaling.ps1 @@ -0,0 +1,75 @@ +param( + [string] $PythonPath = "" +) + +$ErrorActionPreference = "Stop" + +$root = Split-Path -Parent $PSScriptRoot +$dotnet = Get-Command dotnet -ErrorAction Stop +$node = Get-Command node -ErrorAction Stop +$pythonExecutable = if ([string]::IsNullOrWhiteSpace($PythonPath)) { + (Get-Command python -ErrorAction Stop).Source +} +else { + [IO.Path]::GetFullPath($PythonPath) +} +if (-not (Test-Path -LiteralPath $pythonExecutable)) { + throw "Python executable not found: $pythonExecutable" +} +$typescript = Join-Path $root "node_modules\typescript\bin\tsc" +if (-not (Test-Path -LiteralPath $typescript)) { + throw "TypeScript compiler not found: $typescript" +} + +& $dotnet.Source test (Join-Path $root "tests\GameCult.Mesh.Tests\GameCult.Mesh.Tests.csproj") ` + --filter "FullyQualifiedName~ReactiveDocument_SchedulingScalesWithChangedDocumentsOnly" ` + --nologo ` + --verbosity minimal +if ($LASTEXITCODE -ne 0) { + throw "CultMesh C# reactive scaling verification failed." +} + +Push-Location (Join-Path $root "packages\cultmesh-ts") +try { + & $node.Source $typescript -p tsconfig.json --pretty false + if ($LASTEXITCODE -ne 0) { + throw "CultMesh TypeScript source build failed." + } + & $node.Source $typescript -p tsconfig.test.json --pretty false + if ($LASTEXITCODE -ne 0) { + throw "CultMesh TypeScript test build failed." + } + $testFiles = Get-ChildItem -LiteralPath .\dist-test\test -Filter "*.test.js" | + Sort-Object Name | + ForEach-Object FullName + & $node.Source --test @testFiles + if ($LASTEXITCODE -ne 0) { + throw "CultMesh TypeScript reactive scaling verification failed." + } +} +finally { + Pop-Location +} + +$pythonPath = Join-Path $root "packages\cultcache-py\src" +$previousPythonPath = $env:PYTHONPATH +$env:PYTHONPATH = if ([string]::IsNullOrWhiteSpace($previousPythonPath)) { + $pythonPath +} +else { + $pythonPath + [IO.Path]::PathSeparator + $previousPythonPath +} +try { + & $pythonExecutable -m unittest discover ` + -s (Join-Path $root "packages\cultcache-py\tests") ` + -p "test_cultcache.py" ` + -k "reactive_scheduling_scales_with_changed_documents_only" + if ($LASTEXITCODE -ne 0) { + throw "CultMesh Python reactive scaling verification failed." + } +} +finally { + $env:PYTHONPATH = $previousPythonPath +} + +Write-Output '{"runtimes":["csharp","typescript","python"],"documentCounts":[1,100,1000],"changedFraction":0.01,"idleSchedules":0,"invariant":"scheduling is proportional to explicitly changed documents"}' diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index 00ba60b7..d31f8397 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -1371,6 +1371,12 @@ namespace GameCult.Mesh /// public TimeSpan FlushDelay { get; set; } = TimeSpan.FromMilliseconds(16); + /// + /// Gets or sets the clock that owns delayed flush scheduling. Tests and deterministic hosts may + /// replace the system clock without introducing a second scheduler abstraction. + /// + public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance; + /// Gets or sets whether canonical snapshots replace dirty local predictions immediately. public bool ReplaceDirtyCurrentOnCanonicalSnapshot { get; set; } } @@ -1425,7 +1431,7 @@ namespace GameCult.Mesh private readonly CultMeshReactiveDocumentOptions _options; private readonly object _gate = new(); private IDisposable? _subscription; - private Timer? _flushTimer; + private CancellationTokenSource? _flushDelayCancellation; private TDocument _current; private bool _dirty; private bool _flushQueued; @@ -1542,7 +1548,7 @@ namespace GameCult.Mesh return; } - _flushTimer?.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); + CancelScheduledFlushLocked(); _flushing = true; _dirty = false; predicted = CultMeshDocumentHandle.CloneDocument(_current); @@ -1582,10 +1588,10 @@ namespace GameCult.Mesh if (_disposed) return; _disposed = true; + CancelScheduledFlushLocked(); } _subscription?.Dispose(); - _flushTimer?.Dispose(); } private void ApplyCanonicalSnapshot(TDocument canonical) @@ -1634,8 +1640,48 @@ namespace GameCult.Mesh return; } - _flushTimer ??= new Timer(_ => _ = Task.Run(FlushAsync)); - _flushTimer.Change(_options.FlushDelay, Timeout.InfiniteTimeSpan); + CancelScheduledFlushLocked(); + var cancellation = new CancellationTokenSource(); + _flushDelayCancellation = cancellation; + _ = FlushAfterDelayAsync(cancellation); + } + + private async Task FlushAfterDelayAsync(CancellationTokenSource cancellation) + { + try + { + await _options.Clock.DelayAsync(_options.FlushDelay, cancellation.Token).ConfigureAwait(false); + lock (_gate) + { + if (_disposed || cancellation.IsCancellationRequested) + return; + } + + await FlushAsync().ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + } + finally + { + lock (_gate) + { + if (ReferenceEquals(_flushDelayCancellation, cancellation)) + _flushDelayCancellation = null; + } + + cancellation.Dispose(); + } + } + + private void CancelScheduledFlushLocked() + { + var cancellation = _flushDelayCancellation; + _flushDelayCancellation = null; + if (cancellation == null) + return; + + cancellation.Cancel(); } private static IReadOnlyDictionary CreateReconciliationDelta( diff --git a/src/GameCult.Mesh/docs/document-write-authority.md b/src/GameCult.Mesh/docs/document-write-authority.md index 19cc4bb3..0d8d8a68 100644 --- a/src/GameCult.Mesh/docs/document-write-authority.md +++ b/src/GameCult.Mesh/docs/document-write-authority.md @@ -69,4 +69,18 @@ Cut line: - nested edits made through `Update` are coalesced and published; - direct mutation of a returned snapshot cannot publish; - an idle reactive document creates no periodic detection work; +- scheduling 1, 100, or 1,000 reactive documents creates zero idle work, and + editing one percent schedules only that one percent; - canonical reconciliation and in-flight edits remain deterministic. + +Run the cross-runtime scheduling gate from a CultLib checkout: + +```powershell +powershell -ExecutionPolicy Bypass -File scripts/verify-reactive-document-scaling.ps1 +``` + +The gate runs the C#, TypeScript, and Python implementations against the same +1/100/1,000-document cases. It proves ownership and scheduling proportionality; +it does not claim allocation, payload-size, or end-to-end latency budgets. Those +need a separate measured benchmark against representative document bodies and +real transports. diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 6fafbf16..273d1e87 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -3,6 +3,7 @@ using System.Buffers; using System.Collections.Generic; using System.IO; using System.Linq; +using System.Threading; using System.Threading.Tasks; using FluentAssertions; using GameCult.Caching; @@ -1991,6 +1992,69 @@ public sealed class CultMeshStreamingTests reactive.IsDirty.Should().BeFalse(); } + [TestCase(1)] + [TestCase(100)] + [TestCase(1000)] + public async Task ReactiveDocument_SchedulingScalesWithChangedDocumentsOnly(int documentCount) + { + var clock = new BlockingCountClock(); + var reactiveDocuments = new List>(documentCount); + + try + { + for (var index = 0; index < documentCount; index++) + { + var subject = new Subject(); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = $"initial-{index}", + Revision = 1 + }; + var handle = CultMesh.Document( + $"mesh.note.reactive.scale.{index}", + CultMesh.Verse("scale", "test"), + _ => Task.FromResult(current), + _ => subject, + value => + { + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }); + + reactiveDocuments.Add(await handle.AuthoritativeWriter().ReactiveAsync( + new CultMeshReactiveDocumentOptions + { + Clock = clock, + FlushDelay = TimeSpan.FromSeconds(1) + })); + } + + clock.DelayCount.Should().Be(0, "idle reactive documents must not schedule background work"); + + var changedDocumentCount = Math.Max(1, documentCount / 100); + for (var index = 0; index < changedDocumentCount; index++) + { + reactiveDocuments[index].Update(document => + { + document.Text = $"changed-{index}"; + document.Revision++; + }); + } + + clock.DelayCount.Should().Be( + changedDocumentCount, + "flush scheduling must be proportional to explicitly changed documents"); + reactiveDocuments.Count(document => document.IsDirty).Should().Be(changedDocumentCount); + } + finally + { + foreach (var reactiveDocument in reactiveDocuments) + reactiveDocument.Dispose(); + } + } + [Test] public async Task ReactiveDocument_ReplaceLocalFlushesWholeDocument() { @@ -4001,3 +4065,18 @@ internal sealed class FailedBackgroundSchemaClient : ICultNetSchemaClient, ICult { } } + +internal sealed class BlockingCountClock : ICultMeshClock +{ + private int _delayCount; + + public DateTimeOffset UtcNow => DateTimeOffset.UnixEpoch; + + public int DelayCount => Volatile.Read(ref _delayCount); + + public Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _delayCount); + return Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + } +}