diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 344c5ff3..013c0fbe 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -52,7 +52,8 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. unknown shards or stale epoch cursors; `create_shard_log_response(...)` remains the raw wire-dict compatibility projection. - Python CultNet shard replication: - Provides schema-v0 shard-log and shard-snapshot fetchers plus a pull-once + Provides schema-v0 shard-log and shard-snapshot fetchers, a schema-v0 + write-forwarder for raw put/delete messages, plus a pull-once `CultNetShardReplicator` that tracks replica cursors, applies log responses through the existing database owner, and recovers from compacted-history resync pressure by fetching a shard snapshot. Replica cursor storage can be @@ -180,6 +181,6 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js - Python does not claim the full C# shard mutation-log service surface. It can create/apply raw snapshots, create/apply package-local raw shard-log responses, consume remote shard-log responses, and run pull-once replica - catch-up with snapshot recovery and a file-backed replica cursor store. It - does not yet claim the full C# background replicator, write-forwarder, or - file-backed authoritative mutation-log store. + 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. diff --git a/packages/cultcache-py/src/cultcache_py/verify.py b/packages/cultcache-py/src/cultcache_py/verify.py index 7ca367c9..3bd890fb 100644 --- a/packages/cultcache-py/src/cultcache_py/verify.py +++ b/packages/cultcache-py/src/cultcache_py/verify.py @@ -44,9 +44,11 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = { "CultNetShardLogResponse", "CultNetFileShardReplicaCursorStore", "CultNetInMemoryShardReplicaCursorStore", + "CultNetSchemaWriteForwarder", "CultNetSchemaShardLogFetcher", "CultNetSchemaShardSnapshotFetcher", "CultNetShardLogFetcher", + "CultNetShardWriteForwarder", "CultNetShardReplicaCursor", "CultNetShardReplicaCursorStore", "CultNetShardReplicator", diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index 59058b2b..5a42c40b 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -40,9 +40,11 @@ from .shard_log import CultNetShardLogEntry, CultNetShardLogResponse from .shard_replication import ( CultNetFileShardReplicaCursorStore, CultNetInMemoryShardReplicaCursorStore, + CultNetSchemaWriteForwarder, CultNetSchemaShardLogFetcher, CultNetSchemaShardSnapshotFetcher, CultNetShardLogFetcher, + CultNetShardWriteForwarder, CultNetShardReplicaCursor, CultNetShardReplicaCursorStore, CultNetShardReplicator, @@ -73,9 +75,11 @@ __all__ = [ "CultNetShardLogResponse", "CultNetFileShardReplicaCursorStore", "CultNetInMemoryShardReplicaCursorStore", + "CultNetSchemaWriteForwarder", "CultNetSchemaShardLogFetcher", "CultNetSchemaShardSnapshotFetcher", "CultNetShardLogFetcher", + "CultNetShardWriteForwarder", "CultNetShardReplicaCursor", "CultNetShardReplicaCursorStore", "CultNetShardReplicator", diff --git a/packages/cultcache-py/src/cultnet_py/client.py b/packages/cultcache-py/src/cultnet_py/client.py index 3ee31269..6b6f7406 100644 --- a/packages/cultcache-py/src/cultnet_py/client.py +++ b/packages/cultcache-py/src/cultnet_py/client.py @@ -29,6 +29,14 @@ class CultNetRawClient: port: int timeout_seconds: float = 4.0 + def send(self, message: CultNetMessage | dict[str, Any]) -> None: + wire = message.to_wire() if isinstance(message, CultNetMessage) else message + with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection: + connection.settimeout(self.timeout_seconds) + stream = connection.makefile("rwb") + write_frame(stream, msgpack.packb(wire, use_bin_type=True)) + stream.flush() + def request(self, message: CultNetMessage | dict[str, Any], *, expected_schema_version: str) -> dict[str, Any]: wire = message.to_wire() if isinstance(message, CultNetMessage) else message with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection: diff --git a/packages/cultcache-py/src/cultnet_py/shard_replication.py b/packages/cultcache-py/src/cultnet_py/shard_replication.py index 279d13ad..7b2bcc3b 100644 --- a/packages/cultcache-py/src/cultnet_py/shard_replication.py +++ b/packages/cultcache-py/src/cultnet_py/shard_replication.py @@ -14,6 +14,14 @@ from .shard_log import CultNetShardLogResponse from .snapshot import CultNetRawSnapshotResponse +class CultNetShardWriteForwarder(Protocol): + def forward_put(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None: + ... + + def forward_delete(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None: + ... + + class CultNetShardLogFetcher(Protocol): def fetch( self, @@ -114,6 +122,32 @@ class CultNetFileShardReplicaCursorStore: ) +@dataclass(frozen=True) +class CultNetSchemaWriteForwarder: + timeout_seconds: float = 4.0 + + def forward_put(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None: + if message.get("schemaVersion") != "cultnet.document_put_raw.v0": + raise ValueError("forward_put requires a cultnet.document_put_raw.v0 message") + wire = dict(message) + wire.setdefault("shardId", shard.shard_id) + wire.setdefault("shardEpoch", shard.epoch) + self._send(shard, wire) + + def forward_delete(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None: + if message.get("schemaVersion") != "cultnet.document_delete.v0": + raise ValueError("forward_delete requires a cultnet.document_delete.v0 message") + wire = dict(message) + wire.setdefault("shardId", shard.shard_id) + wire.setdefault("shardEpoch", shard.epoch) + self._send(shard, wire) + + def _send(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None: + endpoint = _resolve_primary_endpoint(shard) + host, port = _parse_endpoint(endpoint) + CultNetRawClient(host, port, self.timeout_seconds).send(message) + + @dataclass(frozen=True) class CultNetSchemaShardLogFetcher: timeout_seconds: float = 4.0 diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 00a9250d..4907110a 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -4,10 +4,12 @@ import json import socket import tempfile import threading +import time import unittest from dataclasses import asdict, dataclass from datetime import UTC, datetime, timedelta from pathlib import Path +from typing import Callable from uuid import uuid4 from cultcache_py import ( @@ -37,10 +39,12 @@ from cultnet_py import ( CultNetShardLogResponse, CultNetFileShardReplicaCursorStore, CultNetInMemoryShardReplicaCursorStore, + CultNetSchemaWriteForwarder, CultNetSchemaShardLogFetcher, CultNetSchemaShardSnapshotFetcher, CultNetShardReplicator, CultNetShardReplicatorOptions, + CultNetShardWriteForwarder, CultNetSimulationConsensusOptions, CultNetSimulationConsensusCandidate, CultNetSimulationObservation, @@ -1556,6 +1560,60 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(resync_sequence, 2) self.assertEqual(resync_target.get_required(document, "note:1")["body"], "one") + def test_cultnet_schema_write_forwarder_sends_raw_writes_to_primary(self) -> None: + document = define_database_entry_type( + "mesh.forward_note", + [("body", 0)], + schema_id="mesh.forward_note.v1", + ) + primary = CultMesh.create_node(runtime_id="forward-primary") + primary.database.register_document(document) + server = CultMesh.serve_node(primary) + try: + shard = CultNetShardDescriptor( + shard_id="forward", + owner_runtime_id="forward-primary", + epoch=9, + is_primary=False, + schema_ids=("mesh.forward_note.v1",), + primary_endpoints=(f"cultnet://127.0.0.1:{server.port}",), + ) + forwarder: CultNetShardWriteForwarder = CultNetSchemaWriteForwarder(timeout_seconds=2.0) + put = document_put_raw( + message_id="forward-put", + key="note:forward", + schema_id=document.catalog_entry().schema_id, + stored_at="2026-06-14T00:00:00Z", + payload=document.encode_payload({"body": "forwarded"}), + ).to_wire() + delete = document_delete( + message_id="forward-delete", + schema_id=document.catalog_entry().schema_id, + record_key="note:forward", + ).to_wire() + + forwarder.forward_put(shard, put) + self._eventually(lambda: primary.database.get(document, "note:forward") is not None) + self.assertEqual(primary.database.get_required(document, "note:forward")["body"], "forwarded") + forwarder.forward_delete(shard, delete) + self._eventually(lambda: primary.database.get(document, "note:forward") is None) + + log = primary.database.build_shard_log_response(shard_id="forward", shard_epoch=9) + self.assertIsNone(primary.database.get(document, "note:forward")) + self.assertEqual([entry.change_kind for entry in log.entries], ["added", "removed"]) + self.assertEqual(log.entries[0].put["shardId"], "forward") + self.assertEqual(log.entries[1].delete["shardEpoch"], 9) + finally: + server.stop() + + def _eventually(self, predicate: Callable[[], bool], *, timeout_seconds: float = 2.0) -> None: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.01) + self.assertTrue(predicate()) + def test_cultmesh_database_watchers_observe_name_and_index_changes(self) -> None: document = define_database_entry_type( "mesh.named_watch",