diff --git a/packages/cultcache-py/src/cultmesh_py/__init__.py b/packages/cultcache-py/src/cultmesh_py/__init__.py index 8c812395..95aee189 100644 --- a/packages/cultcache-py/src/cultmesh_py/__init__.py +++ b/packages/cultcache-py/src/cultmesh_py/__init__.py @@ -17,6 +17,9 @@ from .node import ( CultMeshNode, CultMeshNodeOptions, CultMeshPublicationDocumentBinding, + CultMeshReactiveDocument, + CultMeshReactiveDocumentOptions, + CultMeshReactiveDocumentReconciliation, create_node, ) from .server import CultMeshLocalServer @@ -67,6 +70,9 @@ __all__ = [ "CultMeshDocumentPublicationSource", "CultMesh", "CultMeshPublicationDocumentBinding", + "CultMeshReactiveDocument", + "CultMeshReactiveDocumentOptions", + "CultMeshReactiveDocumentReconciliation", "CultMeshGameSession", "CultMeshGameSessionOptions", "CultMeshLocalServer", diff --git a/packages/cultcache-py/src/cultmesh_py/facade.py b/packages/cultcache-py/src/cultmesh_py/facade.py index f5cbea07..115259f3 100644 --- a/packages/cultcache-py/src/cultmesh_py/facade.py +++ b/packages/cultcache-py/src/cultmesh_py/facade.py @@ -27,6 +27,8 @@ from .node import ( CultMeshNode, CultMeshNodeOptions, CultMeshPublicationDocumentBinding, + CultMeshReactiveDocument, + CultMeshReactiveDocumentOptions, create_node, ) from .server import CultMeshLocalServer @@ -298,6 +300,15 @@ class CultMesh: shard_epoch=shard_epoch, ) + @staticmethod + def reactive_document( + node: CultMeshNode, + document: DocumentDefinition[Any], + key: str, + options: CultMeshReactiveDocumentOptions | None = None, + ) -> CultMeshReactiveDocument: + return node.reactive_document(document, key, options) + @staticmethod def publication_source_from_peer_snapshot( client: CultNetRawClient, diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index 8441d514..142fe226 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -1,5 +1,6 @@ from __future__ import annotations +import threading from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path @@ -71,6 +72,22 @@ class CultMeshPublicationDocumentBinding: source: CultMeshDocumentPublicationSource | None = None +@dataclass(frozen=True) +class CultMeshReactiveDocumentReconciliation: + canonical: Any + predicted: Any + delta: dict[str, Any] + version: int + received_at: datetime + + +@dataclass(frozen=True) +class CultMeshReactiveDocumentOptions: + flush_delay_seconds: float = 0.016 + detect_local_changes: bool = True + replace_dirty_current_on_canonical_snapshot: bool = False + + @dataclass class CultMeshNode: cache: CultCache = field(default_factory=CultCache) @@ -231,6 +248,14 @@ class CultMeshNode: shard_epoch=shard_epoch, ) + def reactive_document( + self, + document: DocumentDefinition[Any], + key: str, + options: CultMeshReactiveDocumentOptions | None = None, + ) -> "CultMeshReactiveDocument": + return self.database.reactive_document(document, key, options) + def sync_document_from_publication( self, source: CultMeshDocumentPublicationSource, @@ -719,6 +744,18 @@ class CultMeshDatabase: ) return self.get_required(requested, key) + def reactive_document( + self, + document: DocumentDefinition[Any], + key: str, + options: CultMeshReactiveDocumentOptions | None = None, + ) -> "CultMeshReactiveDocument": + if not key: + raise ValueError("key must be non-empty") + if not any(self._documents_alias(document, registered) for registered in self.documents): + self.register_document(document) + return CultMeshReactiveDocument(self, document, key, options) + def sync_document_from_publication( self, source: CultMeshDocumentPublicationSource, @@ -1217,6 +1254,230 @@ def _document_matches_schema_id(document: DocumentDefinition[Any], schema_id: st return _infer_schema_name(schema_id) == entry.schema_name +class CultMeshReactiveDocument: + def __init__( + self, + database: CultMeshDatabase, + document: DocumentDefinition[Any], + key: str, + options: CultMeshReactiveDocumentOptions | None = None, + ) -> None: + self._database = database + self._document = document + self._key = key + self._options = options or CultMeshReactiveDocumentOptions() + self._lock = threading.RLock() + self._disposed = False + self._dirty = False + self._flushing = False + self._flush_queued = False + self._flush_timer: threading.Timer | None = None + self._detect_timer: threading.Timer | None = None + self._reconciliation_version = 0 + self.reconciliation: CultMeshReactiveDocumentReconciliation | None = None + self.current = self._clone(self._database.get_required(document, key)) + self._last_clean_payload = self._serialize(self.current) + self._unsubscribe = self._database.watch_record(document, key, self._apply_canonical_change) + if self._options.detect_local_changes: + self._schedule_detection() + + @property + def document(self) -> DocumentDefinition[Any]: + return self._document + + @property + def key(self) -> str: + return self._key + + @property + def is_dirty(self) -> bool: + with self._lock: + return self._dirty + + def update(self, callback: Callable[[Any], None]) -> Any: + with self._lock: + self._raise_if_disposed() + callback(self.current) + self._dirty = True + self._schedule_flush_locked() + return self.current + + def set_current(self, value: Any) -> Any: + with self._lock: + self._raise_if_disposed() + self.current = self._clone(value) + self._dirty = True + self._schedule_flush_locked() + return self.current + + def mark_dirty(self) -> None: + with self._lock: + self._raise_if_disposed() + self._dirty = True + self._schedule_flush_locked() + + def refresh(self) -> Any: + with self._lock: + self._raise_if_disposed() + self.current = self._clone(self._database.get_required(self._document, self._key)) + self._last_clean_payload = self._serialize(self.current) + self._dirty = False + self.reconciliation = None + return self.current + + def flush(self) -> None: + with self._lock: + self._raise_if_disposed() + if not self._dirty: + self._detect_local_changes_locked() + if not self._dirty: + return + if self._flushing: + self._flush_queued = True + return + if self._flush_timer is not None: + self._flush_timer.cancel() + self._flush_timer = None + self._flushing = True + self._dirty = False + predicted = self._clone(self.current) + self._last_clean_payload = self._serialize(predicted) + + try: + self._database.put(self._document, self._key, predicted) + finally: + should_flush_again = False + with self._lock: + self._flushing = False + self._detect_local_changes_locked() + should_flush_again = self._flush_queued or self._dirty + self._flush_queued = False + if should_flush_again: + self.flush() + + def clear_reconciliation(self) -> None: + with self._lock: + self._raise_if_disposed() + self.reconciliation = None + + def dispose(self) -> None: + with self._lock: + if self._disposed: + return + self._disposed = True + if self._flush_timer is not None: + self._flush_timer.cancel() + if self._detect_timer is not None: + self._detect_timer.cancel() + self._unsubscribe() + + def __enter__(self) -> "CultMeshReactiveDocument": + return self + + def __exit__(self, _exc_type: Any, _exc: Any, _tb: Any) -> None: + self.dispose() + + def _apply_canonical_change(self, change: CultMeshDatabaseChange) -> None: + if change.value is None: + return + with self._lock: + if self._disposed: + return + canonical = self._clone(change.value) + if self._dirty or self._flushing: + predicted = self._clone(self.current) + delta = self._create_delta(predicted, canonical) + if delta: + self._reconciliation_version += 1 + self.reconciliation = CultMeshReactiveDocumentReconciliation( + canonical=canonical, + predicted=predicted, + delta=delta, + version=self._reconciliation_version, + received_at=datetime.now(UTC), + ) + else: + self.reconciliation = None + if not self._options.replace_dirty_current_on_canonical_snapshot: + return + self.current = canonical + self._last_clean_payload = self._serialize(self.current) + self.reconciliation = None + + def _schedule_detection(self) -> None: + delay = max(self._options.flush_delay_seconds, 0.001) + timer = threading.Timer(delay, self._detect_timer_elapsed) + timer.daemon = True + self._detect_timer = timer + timer.start() + + def _detect_timer_elapsed(self) -> None: + with self._lock: + if self._disposed: + return + if not self._dirty and not self._flushing: + self._detect_local_changes_locked() + if self._dirty: + self._schedule_flush_locked() + if not self._disposed and self._options.detect_local_changes: + self._schedule_detection() + + def _detect_local_changes_locked(self) -> None: + if self._serialize(self.current) != self._last_clean_payload: + self._dirty = True + + 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() + + def _serialize(self, value: Any) -> bytes: + return self._document.encode_payload(value) + + def _clone(self, value: Any) -> Any: + return self._document.decode_payload(self._document.encode_payload(value)) + + @staticmethod + def _create_delta(predicted: Any, canonical: Any) -> dict[str, Any]: + delta: dict[str, Any] = {} + canonical_members = _public_document_members(canonical) + for name, predicted_value in _public_document_members(predicted).items(): + canonical_value = canonical_members.get(name) + if predicted_value == canonical_value: + continue + if isinstance(predicted_value, (int, float)) and isinstance(canonical_value, (int, float)): + delta[name] = predicted_value - canonical_value + else: + delta[name] = predicted_value + return delta + + def _raise_if_disposed(self) -> None: + if self._disposed: + raise RuntimeError("CultMeshReactiveDocument has been disposed") + + +def _public_document_members(value: Any) -> dict[str, Any]: + if hasattr(value, "__dataclass_fields__"): + return { + name: getattr(value, name) + for name in value.__dataclass_fields__ # type: ignore[attr-defined] + if not name.startswith("_") + } + if isinstance(value, dict): + return {str(key): item for key, item in value.items()} + return { + name: item + for name, item in vars(value).items() + if not name.startswith("_") + } + + def create_node( cache_path: str | Path | None = None, *, diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index e9562474..3ecb763f 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -119,6 +119,7 @@ from cultmesh_py import ( CultMeshDatabaseChange, CultMeshGameSessionOptions, CultMeshNodeOptions, + CultMeshReactiveDocumentOptions, CultMeshPeerCard, CultMeshPeerCatalog, CultMeshSimulationFact, @@ -2447,6 +2448,101 @@ class CultCacheTests(unittest.TestCase): self.assertIsNone(node.get(alias, "note:alias")) self.assertEqual(node.get_all(alias), []) + def test_cultmesh_reactive_document_coalesces_direct_alias_member_writes(self) -> None: + @dataclass + class CanonicalNote: + body: str + revision: int + + @dataclass + class UiNote: + body: str + revision: int + + document = define_database_entry_type( + "mesh.reactive_alias_note", + [("body", 0), ("revision", 1)], + cls=CanonicalNote, + schema_id="mesh.reactive_alias_note.v1", + schema_name="mesh.reactive_alias_note", + schema_version="mesh.reactive_alias_note.v1", + ) + alias = define_database_entry_type( + "mesh.reactive_alias_note.ui", + [("body", 0), ("revision", 1)], + cls=UiNote, + schema_id="mesh.reactive_alias_note.v1", + schema_name="mesh.reactive_alias_note", + schema_version="mesh.reactive_alias_note.v1", + ) + node = create_node(runtime_id="python-reactive") + node.register_document(document) + node.put(document, "note:reactive", CanonicalNote("initial", 1)) + seen: list[CultMeshDatabaseChange] = [] + node.database.watch_record(alias, "note:reactive", seen.append) + + reactive = node.reactive_document( + alias, + "note:reactive", + CultMeshReactiveDocumentOptions(flush_delay_seconds=0.02), + ) + try: + self.assertIsInstance(reactive.current, UiNote) + reactive.current.body = "first-local-edit" + reactive.current.body = "second-local-edit" + reactive.current.revision = 2 + + self._wait_until(lambda: node.get_required(alias, "note:reactive").body == "second-local-edit") + + self.assertFalse(reactive.is_dirty) + self.assertEqual(node.get_required(document, "note:reactive").revision, 2) + self.assertEqual(seen[-1].value.body, "second-local-edit") + finally: + reactive.dispose() + + def test_cultmesh_reactive_document_tracks_canonical_reconciliation_delta(self) -> None: + @dataclass + class Note: + body: str + revision: int + + document = define_database_entry_type( + "mesh.reactive_reconcile_note", + [("body", 0), ("revision", 1)], + cls=Note, + schema_id="mesh.reactive_reconcile_note.v1", + schema_name="mesh.reactive_reconcile_note", + schema_version="mesh.reactive_reconcile_note.v1", + ) + node = create_node(runtime_id="python-reactive-reconcile") + node.register_document(document) + node.put(document, "note:reconcile", Note("initial", 1)) + + reactive = CultMesh.reactive_document( + node, + document, + "note:reconcile", + CultMeshReactiveDocumentOptions(flush_delay_seconds=60), + ) + try: + reactive.update(lambda note: setattr(note, "body", "local-prediction")) + reactive.current.revision = 2 + + node.put(document, "note:reconcile", Note("authoritative", 7)) + + self.assertEqual(reactive.current.body, "local-prediction") + self.assertIsNotNone(reactive.reconciliation) + self.assertEqual(reactive.reconciliation.canonical.body, "authoritative") + self.assertEqual(reactive.reconciliation.predicted.body, "local-prediction") + self.assertEqual(reactive.reconciliation.delta["body"], "local-prediction") + self.assertEqual(reactive.reconciliation.delta["revision"], -5) + + reactive.flush() + self.assertEqual(node.get_required(document, "note:reconcile").body, "local-prediction") + self.assertIsNone(reactive.reconciliation) + finally: + reactive.dispose() + def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None: message = document_delete( message_id="delete-1",