Share one scheduler across Python reactive documents

packages/cultcache-py/src/cultmesh_py/node.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import threading
+import heapq
+import time
+import traceback
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from pathlib import Path
@@ -87,6 +90,87 @@ class CultMeshReactiveDocumentOptions:
     replace_dirty_current_on_canonical_snapshot: bool = False
 
 
+class _CultMeshScheduledCall:
+    def __init__(self, callback: Callable[[], None], wake: Callable[[], None]) -> None:
+        self._callback: Callable[[], None] | None = callback
+        self._wake = wake
+        self._lock = threading.Lock()
+
+    def cancel(self) -> None:
+        with self._lock:
+            self._callback = None
+        self._wake()
+
+    def take(self) -> Callable[[], None] | None:
+        with self._lock:
+            callback = self._callback
+            self._callback = None
+            return callback
+
+    @property
+    def cancelled(self) -> bool:
+        with self._lock:
+            return self._callback is None
+
+
+class _CultMeshReactiveScheduler:
+    """One lazy daemon worker owns delayed reactive flushes for the process."""
+
+    def __init__(self) -> None:
+        self._condition = threading.Condition()
+        self._queue: list[tuple[float, int, _CultMeshScheduledCall]] = []
+        self._sequence = 0
+        self._worker: threading.Thread | None = None
+
+    def schedule(self, delay_seconds: float, callback: Callable[[], None]) -> _CultMeshScheduledCall:
+        call = _CultMeshScheduledCall(callback, self._wake)
+        with self._condition:
+            self._sequence += 1
+            heapq.heappush(
+                self._queue,
+                (time.monotonic() + max(0.0, delay_seconds), self._sequence, call),
+            )
+            if self._worker is None:
+                self._worker = threading.Thread(
+                    target=self._run,
+                    name="cultmesh-reactive-scheduler",
+                    daemon=True,
+                )
+                self._worker.start()
+            self._condition.notify()
+        return call
+
+    def _wake(self) -> None:
+        with self._condition:
+            self._condition.notify()
+
+    def _run(self) -> None:
+        while True:
+            callback: Callable[[], None] | None = None
+            with self._condition:
+                while callback is None:
+                    while self._queue and self._queue[0][2].cancelled:
+                        heapq.heappop(self._queue)
+                    if not self._queue:
+                        self._condition.wait()
+                        continue
+                    deadline, _, call = self._queue[0]
+                    remaining = deadline - time.monotonic()
+                    if remaining > 0:
+                        self._condition.wait(remaining)
+                        continue
+                    heapq.heappop(self._queue)
+                    callback = call.take()
+            if callback is not None:
+                try:
+                    callback()
+                except Exception:
+                    traceback.print_exc()
+
+
+_REACTIVE_SCHEDULER = _CultMeshReactiveScheduler()
+
+
 @dataclass
 class CultMeshNode:
     cache: CultCache = field(default_factory=CultCache)
@@ -1373,7 +1457,7 @@ class CultMeshReactiveDocument:
         self._dirty = False
         self._flushing = False
         self._flush_queued = False
-        self._flush_timer: threading.Timer | None = None
+        self._flush_timer: _CultMeshScheduledCall | None = None
         self._reconciliation_version = 0
         self.reconciliation: CultMeshReactiveDocumentReconciliation | None = None
         self._current = self._clone(self._database.get_required(self._document, self._key))
@@ -1503,15 +1587,12 @@ class CultMeshReactiveDocument:
             self.reconciliation = None
 
     def _schedule_flush_locked(self) -> None:
-        if self._options.flush_delay_seconds <= 0:
-            threading.Thread(target=self.flush, daemon=True).start()
-            return
         if self._flush_timer is not None:
             self._flush_timer.cancel()
-        timer = threading.Timer(self._options.flush_delay_seconds, self.flush)
-        timer.daemon = True
-        self._flush_timer = timer
-        timer.start()
+        self._flush_timer = _REACTIVE_SCHEDULER.schedule(
+            self._options.flush_delay_seconds,
+            self.flush,
+        )
 
     def _clone(self, value: Any) -> Any:
         return self._document.decode_payload(self._document.encode_payload(value))

packages/cultcache-py/tests/test_cultcache.py

@@ -2595,23 +2595,27 @@ class CultCacheTests(unittest.TestCase):
 
             scheduled_timers: list[Any] = []
 
-            class FakeTimer:
-                def __init__(self, interval: float, callback: Callable[[], None]) -> None:
-                    self.interval = interval
+            class FakeScheduledCall:
+                def __init__(self, delay_seconds: float, callback: Callable[[], None]) -> None:
+                    self.delay_seconds = delay_seconds
                     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
 
+            class FakeScheduler:
+                def schedule(
+                    self,
+                    delay_seconds: float,
+                    callback: Callable[[], None],
+                ) -> FakeScheduledCall:
+                    call = FakeScheduledCall(delay_seconds, callback)
+                    scheduled_timers.append(call)
+                    return call
+
             reactive_documents: list[Any] = []
-            with patch("cultmesh_py.node.threading.Timer", FakeTimer):
+            with patch("cultmesh_py.node._REACTIVE_SCHEDULER", FakeScheduler()):
                 try:
                     for index in range(document_count):
                         reactive_documents.append(
@@ -2634,7 +2638,6 @@ class CultCacheTests(unittest.TestCase):
                         )
 
                     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,
@@ -2645,6 +2648,46 @@ class CultCacheTests(unittest.TestCase):
 
             self.assertTrue(all(timer.cancelled for timer in scheduled_timers))
 
+    def test_cultmesh_reactive_documents_share_one_lazy_scheduler_thread(self) -> None:
+        document = define_document_type(
+            "mesh.reactive_shared_scheduler_note",
+            encode=lambda value: value,
+            decode=lambda value: value,
+        )
+        node = create_node(runtime_id="python-reactive-shared-scheduler")
+        node.register_document(document)
+        reactive_documents: list[Any] = []
+        try:
+            for index in range(100):
+                node.put(document, f"note:{index}", {"body": "initial", "revision": 1})
+                reactive_documents.append(
+                    node.authoritative_writer(document, f"note:{index}").reactive(
+                        CultMeshReactiveDocumentOptions(flush_delay_seconds=60.0)
+                    )
+                )
+
+            before = sum(
+                thread.name == "cultmesh-reactive-scheduler"
+                for thread in threading.enumerate()
+            )
+            for index, reactive in enumerate(reactive_documents):
+                reactive.update(
+                    lambda draft, changed=index: draft.update(
+                        body=f"changed-{changed}",
+                        revision=2,
+                    )
+                )
+            after = sum(
+                thread.name == "cultmesh-reactive-scheduler"
+                for thread in threading.enumerate()
+            )
+
+            self.assertLessEqual(before, 1)
+            self.assertEqual(after, 1)
+        finally:
+            for reactive in reactive_documents:
+                reactive.dispose()
+
     def test_cultmesh_observed_document_cannot_publish_borrowed_mutation(self) -> None:
         document = define_document_type(
             "mesh.observed_note",

scripts/verify-reactive-document-scaling.ps1

@@ -1,12 +1,22 @@
 param(
-    [string] $PythonPath = ""
+    [string] $PythonPath = "",
+    [string] $NodePath = ""
 )
 
 $ErrorActionPreference = "Stop"
 
 $root = Split-Path -Parent $PSScriptRoot
 $dotnet = Get-Command dotnet -ErrorAction Stop
-$node = Get-Command node -ErrorAction Stop
+$nodeExecutable = if ([string]::IsNullOrWhiteSpace($NodePath)) {
+    $node = Get-Command node -ErrorAction SilentlyContinue
+    if ($node) { $node.Source } else { Join-Path $env:ProgramFiles "nodejs\node.exe" }
+}
+else {
+    [IO.Path]::GetFullPath($NodePath)
+}
+if (-not (Test-Path -LiteralPath $nodeExecutable -PathType Leaf)) {
+    throw "Node.js executable not found: $nodeExecutable"
+}
 $pythonExecutable = if ([string]::IsNullOrWhiteSpace($PythonPath)) {
     (Get-Command python -ErrorAction Stop).Source
 }
@@ -31,18 +41,18 @@ if ($LASTEXITCODE -ne 0) {
 
 Push-Location (Join-Path $root "packages\cultmesh-ts")
 try {
-    & $node.Source $typescript -p tsconfig.json --pretty false
+    & $nodeExecutable $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
+    & $nodeExecutable $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
+    & $nodeExecutable --test @testFiles
     if ($LASTEXITCODE -ne 0) {
         throw "CultMesh TypeScript reactive scaling verification failed."
     }