GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Persist Python authoritative shard logs

Browse source
This commit is contained in:
Meta Vens 2026-06-14 11:28:30 +01:00
commit fae72cc951
7 changed files with 225 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/cultcache-py/PARITY.md
View file

@ -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.

2
packages/cultcache-py/README.md
View file

@ -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))

2
packages/cultcache-py/src/cultcache_py/verify.py
View file

@ -43,11 +43,13 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = {
"CultNetShardLogEntry",
"CultNetShardLogResponse",
"CultNetFileShardReplicaCursorStore",
"CultNetFileShardMutationLogStore",
"CultNetInMemoryShardReplicaCursorStore",
"CultNetSchemaWriteForwarder",
"CultNetSchemaShardLogFetcher",
"CultNetSchemaShardSnapshotFetcher",
"CultNetShardLogFetcher",
"CultNetShardMutationLogStore",
"CultNetShardWriteForwarder",
"CultNetShardReplicaCursor",
"CultNetShardReplicaCursorStore",

72
packages/cultcache-py/src/cultmesh_py/node.py
View file

@ -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()

4
packages/cultcache-py/src/cultnet_py/__init__.py
View file

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

119
packages/cultcache-py/src/cultnet_py/shard_replication.py
View file

@ -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()

33
packages/cultcache-py/tests/test_cultcache.py
View file

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