Prove reactive scheduling scales with edits

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",

packages/cultmesh-ts/src/index.ts

@@ -1140,7 +1140,7 @@ class CultMeshReactiveDocumentHandle<TDocument extends object>
   #disposed = false;
   #pendingSet: Promise<void> = Promise.resolve();
   #flushScheduled = false;
-  #flushTimer: ReturnType<typeof setTimeout> | undefined;
+  #cancelScheduledFlush: (() => void) | undefined;
   #unsubscribe: CultMeshUnsubscribe | undefined;
   #dirty = false;
   #lastPredicted: TDocument | undefined;
@@ -1238,10 +1238,8 @@ class CultMeshReactiveDocumentHandle<TDocument extends object>
 
   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<TDocument extends object>
 
     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;

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<number, FrameRequestCallback>();
+  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",

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"}'

src/GameCult.Mesh/CultMeshPrimitives.cs

@@ -1371,6 +1371,12 @@ namespace GameCult.Mesh
         /// </summary>
         public TimeSpan FlushDelay { get; set; } = TimeSpan.FromMilliseconds(16);
 
+        /// <summary>
+        /// 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.
+        /// </summary>
+        public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance;
+
         /// <summary>Gets or sets whether canonical snapshots replace dirty local predictions immediately.</summary>
         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<TDocument>.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<string, object?> CreateReconciliationDelta(

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.

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<CultMeshReactiveDocument<MeshNoteDocument>>(documentCount);
+
+        try
+        {
+            for (var index = 0; index < documentCount; index++)
+            {
+                var subject = new Subject<MeshNoteDocument>();
+                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);
+    }
+}