diff --git a/packages/cultcache-py/src/cultmesh_py/__init__.py b/packages/cultcache-py/src/cultmesh_py/__init__.py index 95aee189..54b68343 100644 --- a/packages/cultcache-py/src/cultmesh_py/__init__.py +++ b/packages/cultcache-py/src/cultmesh_py/__init__.py @@ -14,8 +14,10 @@ from .node import ( CultMeshDatabase, CultMeshDatabaseChange, CultMeshDocumentPublicationSource, + CultMeshDocumentWriter, CultMeshNode, CultMeshNodeOptions, + CultMeshObservedDocument, CultMeshPublicationDocumentBinding, CultMeshReactiveDocument, CultMeshReactiveDocumentOptions, @@ -68,8 +70,10 @@ __all__ = [ "CultMeshDatabase", "CultMeshDatabaseChange", "CultMeshDocumentPublicationSource", + "CultMeshDocumentWriter", "CultMesh", "CultMeshPublicationDocumentBinding", + "CultMeshObservedDocument", "CultMeshReactiveDocument", "CultMeshReactiveDocumentOptions", "CultMeshReactiveDocumentReconciliation", diff --git a/packages/cultcache-py/src/cultmesh_py/facade.py b/packages/cultcache-py/src/cultmesh_py/facade.py index 115259f3..4c9c34a0 100644 --- a/packages/cultcache-py/src/cultmesh_py/facade.py +++ b/packages/cultcache-py/src/cultmesh_py/facade.py @@ -24,11 +24,11 @@ from cultnet_py import ( from .client import CultMeshDocumentSubscription, CultMeshPeerExchangeClient, CultMeshVerseDiscoveryClient from .node import ( CultMeshDocumentPublicationSource, + CultMeshDocumentWriter, CultMeshNode, CultMeshNodeOptions, + CultMeshObservedDocument, CultMeshPublicationDocumentBinding, - CultMeshReactiveDocument, - CultMeshReactiveDocumentOptions, create_node, ) from .server import CultMeshLocalServer @@ -301,13 +301,20 @@ class CultMesh: ) @staticmethod - def reactive_document( + def authoritative_writer( node: CultMeshNode, document: DocumentDefinition[Any], key: str, - options: CultMeshReactiveDocumentOptions | None = None, - ) -> CultMeshReactiveDocument: - return node.reactive_document(document, key, options) + ) -> CultMeshDocumentWriter: + return node.authoritative_writer(document, key) + + @staticmethod + def observe_document( + node: CultMeshNode, + document: DocumentDefinition[Any], + key: str, + ) -> CultMeshObservedDocument: + return node.observe_document(document, key) @staticmethod def publication_source_from_peer_snapshot( diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index 142fe226..3afddd24 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -84,7 +84,6 @@ class CultMeshReactiveDocumentReconciliation: @dataclass(frozen=True) class CultMeshReactiveDocumentOptions: flush_delay_seconds: float = 0.016 - detect_local_changes: bool = True replace_dirty_current_on_canonical_snapshot: bool = False @@ -248,13 +247,19 @@ class CultMeshNode: shard_epoch=shard_epoch, ) - def reactive_document( + def authoritative_writer( self, document: DocumentDefinition[Any], key: str, - options: CultMeshReactiveDocumentOptions | None = None, - ) -> "CultMeshReactiveDocument": - return self.database.reactive_document(document, key, options) + ) -> "CultMeshDocumentWriter": + return self.database.authoritative_writer(document, key) + + def observe_document( + self, + document: DocumentDefinition[Any], + key: str, + ) -> "CultMeshObservedDocument": + return self.database.observe_document(document, key) def sync_document_from_publication( self, @@ -744,17 +749,27 @@ class CultMeshDatabase: ) return self.get_required(requested, key) - def reactive_document( + def authoritative_writer( self, document: DocumentDefinition[Any], key: str, - options: CultMeshReactiveDocumentOptions | None = None, - ) -> "CultMeshReactiveDocument": + ) -> "CultMeshDocumentWriter": 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) + return CultMeshDocumentWriter(self, document, key) + + def observe_document( + self, + document: DocumentDefinition[Any], + key: str, + ) -> "CultMeshObservedDocument": + 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 CultMeshObservedDocument(self, document, key) def sync_document_from_publication( self, @@ -1254,17 +1269,104 @@ def _document_matches_schema_id(document: DocumentDefinition[Any], schema_id: st return _infer_schema_name(schema_id) == entry.schema_name -class CultMeshReactiveDocument: +class CultMeshDocumentWriter: def __init__( self, database: CultMeshDatabase, document: DocumentDefinition[Any], key: str, + ) -> None: + self.database = database + self.document = document + self.key = key + + def write(self, value: Any) -> None: + self.database.put(self.document, self.key, self._clone(value)) + + def update(self, callback: Callable[[Any], Any | None]) -> Any: + draft = self._clone(self.database.get_required(self.document, self.key)) + replacement = callback(draft) + next_value = draft if replacement is None else replacement + self.write(next_value) + return self._clone(next_value) + + def reactive( + self, options: CultMeshReactiveDocumentOptions | None = None, + ) -> "CultMeshReactiveDocument": + return CultMeshReactiveDocument(self, options) + + def _clone(self, value: Any) -> Any: + return self.document.decode_payload(self.document.encode_payload(value)) + + +class CultMeshObservedDocument: + def __init__( + self, + database: CultMeshDatabase, + document: DocumentDefinition[Any], + key: str, ) -> None: self._database = database self._document = document self._key = key + self._lock = threading.RLock() + self._disposed = False + self._current = self._clone(database.get_required(document, key)) + self._unsubscribe = database.watch_record(document, key, self._apply_canonical_change) + + @property + def current(self) -> Any: + with self._lock: + self._raise_if_disposed() + return self._current + + def refresh(self) -> Any: + with self._lock: + self._raise_if_disposed() + self._current = self._clone( + self._database.get_required(self._document, self._key) + ) + return self._current + + def dispose(self) -> None: + with self._lock: + if self._disposed: + return + self._disposed = True + self._unsubscribe() + + def __enter__(self) -> "CultMeshObservedDocument": + 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 not self._disposed: + self._current = self._clone(change.value) + + def _clone(self, value: Any) -> Any: + return self._document.decode_payload(self._document.encode_payload(value)) + + def _raise_if_disposed(self) -> None: + if self._disposed: + raise RuntimeError("CultMeshObservedDocument has been disposed") + + +class CultMeshReactiveDocument: + def __init__( + self, + writer: CultMeshDocumentWriter, + options: CultMeshReactiveDocumentOptions | None = None, + ) -> None: + self.writer = writer + self._database = writer.database + self._document = writer.document + self._key = writer.key self._options = options or CultMeshReactiveDocumentOptions() self._lock = threading.RLock() self._disposed = False @@ -1272,14 +1374,14 @@ class CultMeshReactiveDocument: 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() + self._current = self._clone(self._database.get_required(self._document, self._key)) + self._unsubscribe = self._database.watch_record( + self._document, + self._key, + self._apply_canonical_change, + ) @property def document(self) -> DocumentDefinition[Any]: @@ -1294,42 +1396,43 @@ class CultMeshReactiveDocument: with self._lock: return self._dirty - def update(self, callback: Callable[[Any], None]) -> Any: + @property + def snapshot(self) -> Any: with self._lock: self._raise_if_disposed() - callback(self.current) - self._dirty = True - self._schedule_flush_locked() - return self.current + return self._clone(self._current) - def set_current(self, value: Any) -> Any: + def update(self, callback: Callable[[Any], Any | None]) -> Any: with self._lock: self._raise_if_disposed() - self.current = self._clone(value) + draft = self._clone(self._current) + replacement = callback(draft) + self._current = self._clone(draft if replacement is None else replacement) self._dirty = True self._schedule_flush_locked() - return self.current + return self._clone(self._current) - def mark_dirty(self) -> None: + def replace_local(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._clone(self._current) 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._current = self._clone( + self._database.get_required(self._document, self._key) + ) self._dirty = False self.reconciliation = None - return self.current + return self._clone(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: @@ -1340,16 +1443,14 @@ class CultMeshReactiveDocument: self._flush_timer = None self._flushing = True self._dirty = False - predicted = self._clone(self.current) - self._last_clean_payload = self._serialize(predicted) + predicted = self._clone(self._current) try: - self._database.put(self._document, self._key, predicted) + self.writer.write(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: @@ -1367,8 +1468,6 @@ class CultMeshReactiveDocument: 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": @@ -1385,7 +1484,7 @@ class CultMeshReactiveDocument: return canonical = self._clone(change.value) if self._dirty or self._flushing: - predicted = self._clone(self.current) + predicted = self._clone(self._current) delta = self._create_delta(predicted, canonical) if delta: self._reconciliation_version += 1 @@ -1400,32 +1499,9 @@ class CultMeshReactiveDocument: 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._current = canonical 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() @@ -1437,9 +1513,6 @@ class CultMeshReactiveDocument: 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)) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 3ecb763f..d00a05ab 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -9,6 +9,7 @@ import tempfile import threading import time import unittest +from unittest.mock import patch import hashlib import hmac import io @@ -2448,7 +2449,7 @@ 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: + def test_cultmesh_reactive_document_coalesces_explicit_alias_updates(self) -> None: @dataclass class CanonicalNote: body: str @@ -2481,16 +2482,14 @@ class CultCacheTests(unittest.TestCase): 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), + reactive = node.authoritative_writer(alias, "note:reactive").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.assertIsInstance(reactive.snapshot, UiNote) + reactive.update(lambda note: setattr(note, "body", "first-local-edit")) + reactive.update(lambda note: setattr(note, "body", "second-local-edit")) + reactive.update(lambda note: setattr(note, "revision", 2)) self._wait_until(lambda: node.get_required(alias, "note:reactive").body == "second-local-edit") @@ -2518,19 +2517,18 @@ class CultCacheTests(unittest.TestCase): node.register_document(document) node.put(document, "note:reconcile", Note("initial", 1)) - reactive = CultMesh.reactive_document( + reactive = CultMesh.authoritative_writer( node, document, "note:reconcile", - CultMeshReactiveDocumentOptions(flush_delay_seconds=60), - ) + ).reactive(CultMeshReactiveDocumentOptions(flush_delay_seconds=60)) try: reactive.update(lambda note: setattr(note, "body", "local-prediction")) - reactive.current.revision = 2 + reactive.update(lambda note: setattr(note, "revision", 2)) node.put(document, "note:reconcile", Note("authoritative", 7)) - self.assertEqual(reactive.current.body, "local-prediction") + self.assertEqual(reactive.snapshot.body, "local-prediction") self.assertIsNotNone(reactive.reconciliation) self.assertEqual(reactive.reconciliation.canonical.body, "authoritative") self.assertEqual(reactive.reconciliation.predicted.body, "local-prediction") @@ -2543,6 +2541,65 @@ class CultCacheTests(unittest.TestCase): finally: reactive.dispose() + def test_cultmesh_reactive_document_is_idle_until_explicit_nested_update(self) -> None: + document = define_document_type( + "mesh.nested_loadout", + encode=lambda value: value, + decode=lambda value: value, + ) + node = create_node(runtime_id="python-reactive-nested") + node.register_document(document) + node.put( + document, + "ship:nested", + {"loadout": {"weapon": {"tuning": {"damage": 10}}}}, + ) + + with patch("cultmesh_py.node.threading.Timer") as timer: + reactive = node.authoritative_writer(document, "ship:nested").reactive() + timer.assert_not_called() + + try: + borrowed = reactive.snapshot + borrowed["loadout"]["weapon"]["tuning"]["damage"] = 99 + self.assertEqual( + node.get_required(document, "ship:nested")["loadout"]["weapon"]["tuning"]["damage"], + 10, + ) + + reactive.update( + lambda draft: draft["loadout"]["weapon"]["tuning"].__setitem__("damage", 12) + ) + reactive.flush() + + self.assertEqual(reactive.snapshot["loadout"]["weapon"]["tuning"]["damage"], 12) + self.assertEqual( + node.get_required(document, "ship:nested")["loadout"]["weapon"]["tuning"]["damage"], + 12, + ) + finally: + reactive.dispose() + + def test_cultmesh_observed_document_cannot_publish_borrowed_mutation(self) -> None: + document = define_document_type( + "mesh.observed_note", + encode=lambda value: value, + decode=lambda value: value, + ) + node = create_node(runtime_id="python-observed") + node.register_document(document) + node.put(document, "note:observed", {"body": "initial"}) + + observed = node.observe_document(document, "note:observed") + try: + observed.current["body"] = "borrowed-local-edit" + self.assertEqual(node.get_required(document, "note:observed")["body"], "initial") + + node.put(document, "note:observed", {"body": "canonical"}) + self.assertEqual(observed.current["body"], "canonical") + finally: + observed.dispose() + def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None: message = document_delete( message_id="delete-1",