diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 013c0fbe..3968eef1 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -183,4 +183,6 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js responses, consume remote shard-log responses, and run pull-once replica catch-up with snapshot recovery, a file-backed replica cursor store, and schema-v0 write forwarding. It does not yet claim the full C# background - replicator or file-backed authoritative mutation-log store. + replicator. Its authoritative shard-log store is file-backed and supports + compaction watermarks, but it is still package-local rather than the full C# + daemon/server stack. diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index d402a994..c8e32c00 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -226,6 +226,7 @@ from cultcache_py import define_database_entry_type from cultnet_py import ( CultNetClientAuthorityScope, CultNetFileShardReplicaCursorStore, + CultNetFileShardMutationLogStore, CultNetRawClient, CultNetSchemaShardLogFetcher, CultNetSchemaShardSnapshotFetcher, @@ -243,6 +244,7 @@ note_doc = define_database_entry_type( ) node = CultMesh.start_node("mesh.cc", runtime_id="python-runtime") node.database.register_document(note_doc) +node.database.use_shard_mutation_log_store(CultNetFileShardMutationLogStore("mesh.shard-logs")) unsubscribe = node.database.watch_record(note_doc, "note:1", lambda change: print(change.change_kind)) unsubscribe_named = node.database.watch_by_name(note_doc, "intro", lambda change: print(change.record_key)) unsubscribe_kind = node.database.watch_by_index(note_doc, "kind", "guide", lambda change: print(change.record_key)) diff --git a/packages/cultcache-py/src/cultcache_py/verify.py b/packages/cultcache-py/src/cultcache_py/verify.py index 3bd890fb..7bb5eb28 100644 --- a/packages/cultcache-py/src/cultcache_py/verify.py +++ b/packages/cultcache-py/src/cultcache_py/verify.py @@ -43,11 +43,13 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = { "CultNetShardLogEntry", "CultNetShardLogResponse", "CultNetFileShardReplicaCursorStore", + "CultNetFileShardMutationLogStore", "CultNetInMemoryShardReplicaCursorStore", "CultNetSchemaWriteForwarder", "CultNetSchemaShardLogFetcher", "CultNetSchemaShardSnapshotFetcher", "CultNetShardLogFetcher", + "CultNetShardMutationLogStore", "CultNetShardWriteForwarder", "CultNetShardReplicaCursor", "CultNetShardReplicaCursorStore", diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index 223fc4d2..156123d9 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -15,6 +15,7 @@ from cultnet_py import ( CultNetRawSnapshotResponse, CultNetShardLogEntry, CultNetShardLogResponse, + CultNetShardMutationLogStore, apply_raw_snapshot as apply_cultnet_raw_snapshot, apply_shard_log_response as apply_cultnet_shard_log_response, document_delete, @@ -224,6 +225,7 @@ class CultMeshDatabase: self._node = node self._subscribers: list[tuple[DocumentDefinition[Any] | None, str | None, Callable[[CultMeshDatabaseChange], None]]] = [] self._shard_logs: dict[str, list[dict[str, Any]]] = {} + self._shard_log_store: CultNetShardMutationLogStore | None = None @property def cache(self) -> CultCache: @@ -316,6 +318,9 @@ class CultMeshDatabase: self.cache.register_document_type(document) self.documents.append(document) + def use_shard_mutation_log_store(self, store: CultNetShardMutationLogStore) -> None: + self._shard_log_store = store + def put(self, document: DocumentDefinition[Any], key: str, value: Any) -> None: previous = self.cache.get(document, key) self.cache.put(document, key, value) @@ -451,11 +456,13 @@ class CultMeshDatabase: return self._latest_shard_epoch(shard_id) or 0 def shard_ids(self) -> list[str]: + if self._shard_log_store is not None: + return sorted(set(self._shard_logs) | set(self._shard_log_store.shard_ids())) return sorted(self._shard_logs) def shard_schema_ids(self, shard_id: str) -> list[str]: schema_ids: set[str] = set() - for entry in self._shard_logs.get(shard_id, []): + for entry in self._shard_log_entries(shard_id): put = entry.get("put") if isinstance(put, dict) and isinstance(put.get("document"), dict): schema_id = put["document"].get("schemaId") @@ -615,15 +622,22 @@ class CultMeshDatabase: resync_required=True, reason="stale_epoch", ) + compacted_through = self._shard_log_compacted_through(shard_id) + if after_sequence < compacted_through: + return CultNetShardLogResponse( + message_id=message_id, + shard_id=shard_id, + shard_epoch=latest_epoch, + entries=(), + resync_required=True, + reason="compacted_history", + compacted_through=compacted_through, + ) entries = [ CultNetShardLogEntry.from_wire(entry) - for entry in self._shard_logs.get(shard_id, []) + for entry in self._shard_log_entries(shard_id, after_sequence=after_sequence, limit=limit) if int(entry.get("sequence") or 0) > after_sequence ] - if limit is not None: - if limit < 0: - raise ValueError("limit must be non-negative") - entries = entries[:limit] return CultNetShardLogResponse( message_id=message_id, shard_id=shard_id, @@ -729,12 +743,20 @@ class CultMeshDatabase: def _append_shard_log_entry(self, shard_id: str, entry: dict[str, Any]) -> None: log = self._shard_logs.setdefault(shard_id, []) - entry["sequence"] = len(log) + 1 + latest_sequence = max( + [int(item.get("sequence") or 0) for item in log] + + [int(item.get("sequence") or 0) for item in self._stored_shard_log_entries(shard_id)] + + [self._shard_log_compacted_through(shard_id)] + ) + entry["sequence"] = latest_sequence + 1 entry["committedAt"] = datetime.now(UTC).isoformat().replace("+00:00", "Z") - log.append(entry) + if self._shard_log_store is not None: + self._shard_log_store.append(shard_id, entry) + else: + log.append(entry) def _latest_shard_epoch(self, shard_id: str) -> int | None: - log = self._shard_logs.get(shard_id, []) + log = self._shard_log_entries(shard_id) for entry in reversed(log): put = entry.get("put") if isinstance(put, dict) and isinstance(put.get("shardEpoch"), int): @@ -746,7 +768,7 @@ class CultMeshDatabase: def _live_shard_record_keys(self, shard_id: str) -> set[tuple[str, str]]: records: set[tuple[str, str]] = set() - for entry in self._shard_logs.get(shard_id, []): + for entry in self._shard_log_entries(shard_id): put = entry.get("put") if isinstance(put, dict) and isinstance(put.get("document"), dict): document = put["document"] @@ -762,6 +784,36 @@ class CultMeshDatabase: records.discard((str(schema_id), str(record_key))) return records + def _shard_log_entries( + self, + shard_id: str, + *, + after_sequence: int = 0, + limit: int | None = None, + ) -> list[dict[str, Any]]: + if self._shard_log_store is not None: + return self._shard_log_store.read(shard_id, after_sequence=after_sequence, limit=limit) + entries = [ + entry + for entry in self._shard_logs.get(shard_id, []) + if int(entry.get("sequence") or 0) > after_sequence + ] + if limit is not None: + if limit < 0: + raise ValueError("limit must be non-negative") + entries = entries[:limit] + return entries + + def _stored_shard_log_entries(self, shard_id: str) -> list[dict[str, Any]]: + if self._shard_log_store is None: + return [] + return self._shard_log_store.read(shard_id, after_sequence=0) + + def _shard_log_compacted_through(self, shard_id: str) -> int: + if self._shard_log_store is None: + return 0 + return self._shard_log_store.get_compacted_through(shard_id) + def _document_for_schema(self, schema_id: str) -> DocumentDefinition[Any] | None: for document in self.documents: entry = document.catalog_entry() diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index 5a42c40b..144c21a6 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -39,11 +39,13 @@ from .shard_catalog import CultNetShardCatalog, CultNetShardDescriptor from .shard_log import CultNetShardLogEntry, CultNetShardLogResponse from .shard_replication import ( CultNetFileShardReplicaCursorStore, + CultNetFileShardMutationLogStore, CultNetInMemoryShardReplicaCursorStore, CultNetSchemaWriteForwarder, CultNetSchemaShardLogFetcher, CultNetSchemaShardSnapshotFetcher, CultNetShardLogFetcher, + CultNetShardMutationLogStore, CultNetShardWriteForwarder, CultNetShardReplicaCursor, CultNetShardReplicaCursorStore, @@ -74,11 +76,13 @@ __all__ = [ "CultNetShardLogEntry", "CultNetShardLogResponse", "CultNetFileShardReplicaCursorStore", + "CultNetFileShardMutationLogStore", "CultNetInMemoryShardReplicaCursorStore", "CultNetSchemaWriteForwarder", "CultNetSchemaShardLogFetcher", "CultNetSchemaShardSnapshotFetcher", "CultNetShardLogFetcher", + "CultNetShardMutationLogStore", "CultNetShardWriteForwarder", "CultNetShardReplicaCursor", "CultNetShardReplicaCursorStore", diff --git a/packages/cultcache-py/src/cultnet_py/shard_replication.py b/packages/cultcache-py/src/cultnet_py/shard_replication.py index 7b2bcc3b..84279a17 100644 --- a/packages/cultcache-py/src/cultnet_py/shard_replication.py +++ b/packages/cultcache-py/src/cultnet_py/shard_replication.py @@ -2,6 +2,7 @@ from __future__ import annotations from dataclasses import dataclass, field from datetime import UTC, datetime +import hashlib from pathlib import Path from typing import Any, Protocol from urllib.parse import urlparse @@ -22,6 +23,23 @@ class CultNetShardWriteForwarder(Protocol): ... +class CultNetShardMutationLogStore(Protocol): + def shard_ids(self) -> list[str]: + ... + + def read(self, shard_id: str, *, after_sequence: int = 0, limit: int | None = None) -> list[dict[str, Any]]: + ... + + def append(self, shard_id: str, entry: dict[str, Any]) -> None: + ... + + def get_compacted_through(self, shard_id: str) -> int: + ... + + def compact_through(self, shard_id: str, sequence: int) -> None: + ... + + class CultNetShardLogFetcher(Protocol): def fetch( self, @@ -122,6 +140,103 @@ class CultNetFileShardReplicaCursorStore: ) +@dataclass(frozen=True) +class CultNetFileShardMutationLogStore: + root_path: str | Path + + def __post_init__(self) -> None: + Path(self.root_path).mkdir(parents=True, exist_ok=True) + + def shard_ids(self) -> list[str]: + shards = [] + for path in Path(self.root_path).glob("*.meta.mpack"): + value = msgpack.unpackb(path.read_bytes(), raw=False) + if isinstance(value, dict) and value.get("shardId"): + shards.append(str(value["shardId"])) + return sorted(set(shards)) + + def read(self, shard_id: str, *, after_sequence: int = 0, limit: int | None = None) -> list[dict[str, Any]]: + _require_non_empty(shard_id, "shard_id") + if after_sequence < 0: + raise ValueError("after_sequence must be non-negative") + entries = [ + entry + for entry in self._read_entries(shard_id) + if int(entry.get("sequence") or 0) > after_sequence + ] + entries.sort(key=lambda entry: int(entry.get("sequence") or 0)) + if limit is not None: + if limit < 0: + raise ValueError("limit must be non-negative") + entries = entries[:limit] + return entries + + def append(self, shard_id: str, entry: dict[str, Any]) -> None: + _require_non_empty(shard_id, "shard_id") + sequence = int(entry.get("sequence") or 0) + if sequence <= 0: + raise ValueError("shard log entry sequence must be positive") + entries = [ + existing + for existing in self._read_entries(shard_id) + if int(existing.get("sequence") or 0) != sequence + ] + entries.append(dict(entry)) + entries.sort(key=lambda value: int(value.get("sequence") or 0)) + self._entry_path(shard_id).write_bytes(msgpack.packb(entries, use_bin_type=True)) + self._write_metadata(shard_id, self._read_metadata(shard_id)) + + def get_compacted_through(self, shard_id: str) -> int: + _require_non_empty(shard_id, "shard_id") + metadata = self._read_metadata(shard_id) + return int(metadata.get("compactedThrough") or 0) + + def compact_through(self, shard_id: str, sequence: int) -> None: + _require_non_empty(shard_id, "shard_id") + if sequence < 0: + raise ValueError("sequence must be non-negative") + current = self.get_compacted_through(shard_id) + if sequence <= current: + return + retained = [ + entry + for entry in self._read_entries(shard_id) + if int(entry.get("sequence") or 0) > sequence + ] + self._entry_path(shard_id).write_bytes(msgpack.packb(retained, use_bin_type=True)) + self._write_metadata(shard_id, {"compactedThrough": sequence}) + + def _read_entries(self, shard_id: str) -> list[dict[str, Any]]: + path = self._entry_path(shard_id) + if not path.exists() or path.stat().st_size == 0: + return [] + value = msgpack.unpackb(path.read_bytes(), raw=False) + if not isinstance(value, list): + raise ValueError("CultNet shard mutation log file must contain a MessagePack array") + return [dict(entry) for entry in value if isinstance(entry, dict)] + + def _read_metadata(self, shard_id: str) -> dict[str, Any]: + path = self._metadata_path(shard_id) + if not path.exists() or path.stat().st_size == 0: + return {} + value = msgpack.unpackb(path.read_bytes(), raw=False) + if not isinstance(value, dict): + raise ValueError("CultNet shard mutation log metadata must be a MessagePack map") + return dict(value) + + def _write_metadata(self, shard_id: str, metadata: dict[str, Any]) -> None: + payload = dict(metadata) + payload["shardId"] = shard_id + payload.setdefault("compactedThrough", 0) + self._metadata_path(shard_id).write_bytes(msgpack.packb(payload, use_bin_type=True)) + + def _entry_path(self, shard_id: str) -> Path: + return Path(self.root_path) / f"{_hash_shard_id(shard_id)}.mpack" + + def _metadata_path(self, shard_id: str) -> Path: + return Path(self.root_path) / f"{_hash_shard_id(shard_id)}.meta.mpack" + + @dataclass(frozen=True) class CultNetSchemaWriteForwarder: timeout_seconds: float = 4.0 @@ -264,3 +379,7 @@ def _parse_endpoint(endpoint: str) -> tuple[str, int]: def _require_non_empty(value: str, name: str) -> None: if not value or not value.strip(): raise ValueError(f"{name} must be non-empty") + + +def _hash_shard_id(shard_id: str) -> str: + return hashlib.sha256(shard_id.encode("utf-8")).hexdigest() diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 4907110a..d60356c0 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -38,12 +38,14 @@ from cultnet_py import ( CultNetShardLogEntry, CultNetShardLogResponse, CultNetFileShardReplicaCursorStore, + CultNetFileShardMutationLogStore, CultNetInMemoryShardReplicaCursorStore, CultNetSchemaWriteForwarder, CultNetSchemaShardLogFetcher, CultNetSchemaShardSnapshotFetcher, CultNetShardReplicator, CultNetShardReplicatorOptions, + CultNetShardMutationLogStore, CultNetShardWriteForwarder, CultNetSimulationConsensusOptions, CultNetSimulationConsensusCandidate, @@ -1461,6 +1463,37 @@ class CultCacheTests(unittest.TestCase): self.assertEqual([record.change_kind for record in typed_applied], ["updated", "removed"]) self.assertIsNone(typed_target.database.get(document, "note:1")) + def test_cultnet_file_shard_mutation_log_store_persists_authoritative_entries(self) -> None: + document = define_database_entry_type( + "mesh.durable_log_note", + [("body", 0)], + schema_id="mesh.durable_log_note.v1", + ) + with tempfile.TemporaryDirectory() as tmp: + store: CultNetShardMutationLogStore = CultNetFileShardMutationLogStore(Path(tmp) / "logs") + source = CultMesh.create_node(runtime_id="durable-primary") + source.database.register_document(document) + source.database.use_shard_mutation_log_store(store) + source.database.put_raw_message(document, "note:1", {"body": "first"}, shard_id="durable", shard_epoch=4) + source.database.put_raw_message(document, "note:2", {"body": "second"}, shard_id="durable", shard_epoch=4) + + reopened = CultMesh.create_node(runtime_id="durable-primary") + reopened.database.register_document(document) + reopened.database.use_shard_mutation_log_store(CultNetFileShardMutationLogStore(Path(tmp) / "logs")) + response = reopened.database.build_shard_log_response(shard_id="durable", shard_epoch=4) + self.assertEqual(reopened.database.shard_ids(), ["durable"]) + self.assertEqual(reopened.database.shard_schema_ids("durable"), ["mesh.durable_log_note.v1"]) + self.assertEqual([entry.sequence for entry in response.entries], [1, 2]) + self.assertEqual(response.entries[0].raw_document.record_key, "note:1") + + store.compact_through("durable", 1) + compacted = reopened.database.build_shard_log_response(shard_id="durable", shard_epoch=4, after_sequence=0) + retained = reopened.database.build_shard_log_response(shard_id="durable", shard_epoch=4, after_sequence=1) + self.assertTrue(compacted.resync_required) + self.assertEqual(compacted.reason, "compacted_history") + self.assertEqual(compacted.compacted_through, 1) + self.assertEqual([entry.sequence for entry in retained.entries], [2]) + def test_cultnet_shard_replicator_pulls_logs_and_resyncs_from_snapshot(self) -> None: document = define_database_entry_type( "mesh.replica_note",