From 9f543a520682ae76045a2e5b8e16bad8dfd98c29 Mon Sep 17 00:00:00 2001 From: Libby Date: Sat, 14 Nov 2026 04:37:59 +0000 Subject: [PATCH] cultnet-py: refuse a replicated record with no stamp where it becomes an envelope (py-snapshot-stampless) Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-py/src/cultmesh_py/node.py | 17 +- .../tests/test_stampless_replication.py | 194 ++++++++++++++++++ packages/cultnet-py/CHANGELOG.md | 1 + .../cultnet-py/src/cultnet_py/replication.py | 9 +- 4 files changed, 206 insertions(+), 15 deletions(-) create mode 100644 packages/cultmesh-py/tests/test_stampless_replication.py diff --git a/packages/cultmesh-py/src/cultmesh_py/node.py b/packages/cultmesh-py/src/cultmesh_py/node.py index ff6ef8f2..eb439683 100644 --- a/packages/cultmesh-py/src/cultmesh_py/node.py +++ b/packages/cultmesh-py/src/cultmesh_py/node.py @@ -12,7 +12,7 @@ from typing import Any, Callable from cultcache_py import CultCache, CultCacheEnvelope, SingleFileMessagePackBackingStore from cultcache_py.cache import CultCacheError, index_text from cultcache_py.documents import DocumentDefinition, extract_value -from cultcache_py.stores import parse_stored_at +from cultnet_py.replication import raw_record_to_envelope from cultnet_py import ( CultNetAppliedRecord, CultNetFileShardMutationLogStore, @@ -724,19 +724,8 @@ class CultMeshDatabase: document = self._raw_put_document(message) if document is None: return None - document_record = message["document"] - # A peer's put is stored under the stamp it carries, so one that is no stamp is refused: nothing is minted for it. - if parse_stored_at(document_record.get("storedAt")) is None: - raise ValueError("A raw put must carry a storedAt stamp.") - # One id: the local schema id, under which the cache stores the record. - return document, CultCacheEnvelope( - key=str(document_record["recordKey"]), - type=document.type, - schema_id=document.catalog_entry().schema_id, - payload=bytes(document_record["payload"]), - stored_at=document_record["storedAt"], - catalog_entry=document.catalog_entry(), - ) + # One id: the local schema id, under which the cache stores the record. The stamp is judged where a record becomes an envelope. + return document, raw_record_to_envelope(document, document.catalog_entry().schema_id, message["document"]) @staticmethod def _served_record(envelope: CultCacheEnvelope) -> CultNetRawDocumentRecord: diff --git a/packages/cultmesh-py/tests/test_stampless_replication.py b/packages/cultmesh-py/tests/test_stampless_replication.py new file mode 100644 index 00000000..0b2cf9e4 --- /dev/null +++ b/packages/cultmesh-py/tests/test_stampless_replication.py @@ -0,0 +1,194 @@ +"""A record a peer replicates to this node is stored under the stamp it carries, so one with no stamp is refused where it becomes an envelope. + +The stamp shape is tests/vectors/stored-at-stamp/stamps.msgpack, held to the same answers as the raw put of the server +(variants:ruling:raw-put-without-stamp-refused). A snapshot response or shard log that carries a record with no stamp is refused whole +with ValueError before anything is stored (as a record of an unregistered schema is refused whole); the refusal of a put is +message_invalid (field `document`), tested in test_remote_write_authority.py. +""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from typing import Any + +import msgpack +from cultcache_py import define_database_entry_type +from cultcache_py.stores import parse_stored_at +from cultmesh_py import CultMeshGrantAllWrites, CultMeshServerPeer, create_node +from cultmesh_py.client import CultMeshDocumentSubscription +from cultmesh_py.server import CultMeshLocalServer +from cultnet_py import CultNetDatabaseChange, document_put_raw + +SCHEMA_ID = "sha256:stampless-note" +SHARD_ID = "stampless-shard" +PEER = CultMeshServerPeer("127.0.0.1:50000", "tcp") +VECTOR = Path(__file__).resolve().parents[3] / "tests/vectors/stored-at-stamp/stamps.msgpack" +ROWS: list[tuple[str, bool]] = [(text, parses) for text, parses in msgpack.unpackb(VECTOR.read_bytes(), raw=False)] +# The vector's texts, plus the shapes a record can lack a stamp in: the field missing, and not a string at all. +REFUSED: list[tuple[str, Any]] = [(f"vector {text!r}", text) for text, parses in ROWS if not parses] + [ + ("missing", None), + ("not a string", 12345), +] + +NOTE = define_database_entry_type( + "tests.stampless_note", + [("schema_version", 0), ("text", 1)], + schema_id=SCHEMA_ID, + schema_name="tests.stampless_note", + schema_version="tests.stampless_note.v1", +) + + +def record_of(key: str, text: str, stamp: Any = "2026-11-07T00:00:00Z") -> dict[str, Any]: + record = { + "schemaId": SCHEMA_ID, + "recordKey": key, + "payloadEncoding": "messagepack", + "payload": NOTE.encode_payload({"schema_version": "tests.stampless_note.v1", "text": text}), + } + if stamp is not None: + record["storedAt"] = stamp + return record + + +def snapshot_of(*records: dict[str, Any]) -> dict[str, Any]: + return {"schemaVersion": "cultnet.snapshot_response_raw.v0", "messageId": "snapshot", "documents": list(records)} + + +def shard_log_of(*records: dict[str, Any]) -> dict[str, Any]: + entries = [ + { + "sequence": index + 1, + "committedAt": "2026-11-07T00:00:00Z", + "changeKind": "added", + "put": document_put_raw( + message_id=f"put-{index}", key=record["recordKey"], schema_id=SCHEMA_ID, stored_at="2026-11-07T00:00:00Z", + payload=record["payload"], shard_id=SHARD_ID, shard_epoch=1, + ).to_wire() | {"document": record}, + } + for index, record in enumerate(records) + ] + return { + "schemaVersion": "cultnet.shard_log_response.v0", "messageId": "log", "shardId": SHARD_ID, "shardEpoch": 1, + "entries": entries, "resyncRequired": False, + } + + +class StampVectorTests(unittest.TestCase): + def test_python_parses_every_row_of_the_vector_as_the_vector_says(self) -> None: + self.assertGreater(len(ROWS), 30) + for text, parses in ROWS: + with self.subTest(text=text): + self.assertEqual(parse_stored_at(text) is not None, parses) + + def test_every_vector_row_through_the_server_put_handler_is_stored_or_refused_as_the_vector_says(self) -> None: + with tempfile.TemporaryDirectory() as directory: + node = create_node(Path(directory) / "node.cc", runtime_id="vector-server") + node.database.register_document(NOTE) + server = CultMeshLocalServer(node=node, port=0, write_authority=CultMeshGrantAllWrites()) + sized = CultMeshLocalServer(node=node, port=0, write_authority=CultMeshGrantAllWrites(), max_snapshot_bytes=10**6) + for index, (text, parses) in enumerate(ROWS): + for label, handler, over_rudp in (("plain", server, False), ("sized", sized, False), ("rudp", server, True)): + with self.subTest(text=text, path=label): + key = f"vector:{index}:{label}" + message = document_put_raw( + message_id="wire", key=key, schema_id=SCHEMA_ID, stored_at=text, + payload=record_of(key, "peer")["payload"], shard_id=SHARD_ID, shard_epoch=1, + ).to_wire() + replies = handler._handle_connection_message(message, {}, over_rudp=over_rudp, peer=PEER) + if parses: + self.assertEqual(replies, []) + self.assertEqual(node.cache.get(NOTE, key)["text"], "peer") + else: + self.assertEqual([(r["code"], r["details"]["field"]) for r in replies], [("message_invalid", "document")]) + self.assertIsNone(node.cache.get(NOTE, key)) + + +class StamplessSnapshotTests(unittest.TestCase): + def build(self, directory: str) -> Any: + node = create_node(Path(directory) / "node.cc", runtime_id="stampless-client") + node.database.register_document(NOTE) + node.database.pull() + return node + + def test_a_snapshot_with_a_stampless_record_is_refused_whole_and_the_store_survives_a_reopen(self) -> None: + for label, stamp in REFUSED: + with self.subTest(label), tempfile.TemporaryDirectory() as directory: + node = self.build(directory) + good, bad = record_of("note:good", "good"), record_of("note:bad", "bad", stamp) + for order in ((good, bad), (bad, good)): + with self.assertRaises(ValueError): + node.database.apply_snapshot_response(snapshot_of(*order)) + self.assertIsNone(node.cache.get(NOTE, "note:good"), "a refused snapshot stores none of its records") + self.assertIsNone(node.cache.get(NOTE, "note:bad")) + # The store file reads back, and is still writable by a snapshot of stamped records. + reopened = self.build(directory) + self.assertIsNone(reopened.cache.get(NOTE, "note:bad")) + applied = reopened.database.apply_snapshot_response(snapshot_of(good)) + self.assertEqual([a.record_key for a in applied], ["note:good"]) + self.assertEqual(self.build(directory).cache.get(NOTE, "note:good")["text"], "good") + + def test_a_snapshot_of_every_stamped_vector_row_is_stored(self) -> None: + with tempfile.TemporaryDirectory() as directory: + node = self.build(directory) + rows = [(index, text) for index, (text, parses) in enumerate(ROWS) if parses] + node.database.apply_snapshot_response(snapshot_of(*[record_of(f"stamped:{i}", "ok", text) for i, text in rows])) + self.assertEqual(len(self.build(directory).cache.get_all(NOTE)), len(rows)) + for index, _ in rows: + self.assertEqual(node.cache.get(NOTE, f"stamped:{index}")["text"], "ok") + + +class StamplessShardLogTests(unittest.TestCase): + def test_a_shard_log_put_with_no_stamp_is_refused_and_the_store_stays_writable(self) -> None: + for label, stamp in REFUSED: + with self.subTest(label), tempfile.TemporaryDirectory() as directory: + node = create_node(Path(directory) / "node.cc", runtime_id="stampless-log") + node.database.register_document(NOTE) + with self.assertRaises(ValueError): + node.database.apply_shard_log_response(shard_log_of(record_of("note:bad", "bad", stamp))) + self.assertIsNone(node.cache.get(NOTE, "note:bad")) + reopened = create_node(Path(directory) / "node.cc", runtime_id="stampless-log") + reopened.database.register_document(NOTE) + reopened.database.pull() + reopened.database.apply_shard_log_response(shard_log_of(record_of("note:good", "good"))) + self.assertEqual(reopened.cache.get(NOTE, "note:good")["text"], "good") + again = create_node(Path(directory) / "node.cc", runtime_id="stampless-log") + again.database.register_document(NOTE) + again.database.pull() + self.assertEqual(again.cache.get(NOTE, "note:good")["text"], "good") + + +class _StubSubscription: + def __init__(self, change: CultNetDatabaseChange) -> None: + self._change = change + + def read_next_change(self) -> CultNetDatabaseChange: + return self._change + + +class StamplessClientReplayTests(unittest.TestCase): + def test_a_subscribed_put_with_no_stamp_is_refused_by_read_next_change_and_nothing_is_stored(self) -> None: + for label, stamp in REFUSED: + with self.subTest(label), tempfile.TemporaryDirectory() as directory: + node = create_node(Path(directory) / "node.cc", runtime_id="stampless-subscriber") + node.database.register_document(NOTE) + subscription = CultMeshDocumentSubscription(client=None, database=node.database, document=NOTE, key="note:bad") # type: ignore[arg-type] + + def deliver(record: dict[str, Any]) -> Any: + message = { + "schemaVersion": "cultnet.database_change_raw.v0", "messageId": "change", "changeKind": "added", + "schemaId": SCHEMA_ID, "recordKey": record["recordKey"], "document": record, + } + subscription._subscription = _StubSubscription(CultNetDatabaseChange.from_wire(message)) # type: ignore[assignment] + return subscription.read_next_change() + + with self.assertRaises(ValueError): + deliver(record_of("note:bad", "bad", stamp)) + self.assertIsNone(node.cache.get(NOTE, "note:bad")) + self.assertEqual(deliver(record_of("note:bad", "stamped")).value["text"], "stamped") + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/cultnet-py/CHANGELOG.md b/packages/cultnet-py/CHANGELOG.md index b7e73ae3..0652ad0a 100644 --- a/packages/cultnet-py/CHANGELOG.md +++ b/packages/cultnet-py/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased (breaking; version 0.x, so the next minor) +- `raw_record_to_envelope`, where a replicated record becomes an envelope, refuses a record whose `storedAt` is missing, empty or not a stamp (`cultcache_py.stores.parse_stored_at`, the shape pinned by `tests/vectors/stored-at-stamp`) with `ValueError`, nothing minted or stored (`variants:ruling:raw-put-without-stamp-refused`). `apply_raw_document_record`, `apply_raw_snapshot` and `apply_shard_log_response` come through it: a snapshot response with such a record is refused whole before any of its records is stored, as one naming an unregistered schema is; a shard log refuses at that entry, the entries before it already applied. It stored the record under `""`, and the next open of the store raised `CultCacheError` `Missing envelope`. `cultmesh-py`'s raw put builds its envelope through the same function, so there is one owner of the check there; its server gate still answers a peer's put `message_invalid`. A client's `read_next_change` lets the `ValueError` out. - `decode_witness_artifact_bundle_payload` (and `CultNetWitnessArtifactBundle.from_payload`) read the payload through `cultcache_py.read_payload_tree`: bytes after its first MessagePack value raise `ValueError` `cultnet.witness_artifact_bundle payload has N bytes after its first MessagePack value (refused on read).` instead of `msgpack.ExtraData`, and a payload nested past 256 raises `CultCachePayloadDepthError`. Needs `cultcache-py` 0.4.0, the release that exports `read_payload_tree`: the dependency is now `cultcache-py>=0.4.0` (it was `>=0.2.0`, which admits releases whose `import cultnet_py` fails), and `tests/test_declared_dependency_ranges.py` fails when a declared bound admits a tagged release lacking a name the package imports. The same check found `cultmesh-py` importing `cache.index_text`, which `cultcache-py` 0.2.0 lacks; its bound is `cultcache-py>=0.4.0` too (it has no changelog of its own). `resolve_document_for_raw_record` reads the schema hint from a raw record's first value whether or not bytes follow it, so a record routed by its stamp is refused by count when applied (`ValueError`, in a snapshot, a shard log and a direct apply) like one routed by a registered schema id; `apply_shard_log_response` used to skip it silently as an unknown schema. A tail on a payload whose first value names no registered schema is still `KeyError`, which a shard log skips. - `apply_raw_document_record` and `apply_raw_snapshot` raise `SchemaConflictError` when a remote record is of another type than the one the cache holds at its key (`cultcache-py` holds one record per key whatever its type); the cache is unchanged, and the caller decides whether to drop the record or the peer. `cultmesh-py`'s server answers such a put with `message_invalid`; its client's `read_next_change` lets the error out. - `CultNetDatabaseChange` carries `commit_id` and `commit_count` (`commitId`/`commitCount`; None outside a multi-record commit, as C# writes nil): `from_wire` refuses an empty or non-string id and a count that is not an integer of at least 2, `to_wire` writes them when set, and the published `database_change_raw` schema lists both. Python applies no batch. The `simulation_fact` document's `shard`, `frame`, `subject` and `claim_kind` aliases are `roles` (many facts share each); the catalog bytes do not move. diff --git a/packages/cultnet-py/src/cultnet_py/replication.py b/packages/cultnet-py/src/cultnet_py/replication.py index 345f1c62..fb4cec91 100644 --- a/packages/cultnet-py/src/cultnet_py/replication.py +++ b/packages/cultnet-py/src/cultnet_py/replication.py @@ -5,6 +5,7 @@ from typing import Any, Iterable from cultcache_py import CultCache, CultCacheEnvelope from cultcache_py.documents import DocumentDefinition +from cultcache_py.stores import parse_stored_at from .shard_log import CultNetShardLogResponse from .snapshot import CultNetRawDocumentRecord, CultNetRawSnapshotResponse @@ -55,12 +56,18 @@ def raw_record_to_envelope( schema_id: str, record: dict[str, Any], ) -> CultCacheEnvelope: + """The envelope a replicated raw record is stored as; the one place a record becomes an envelope (a direct apply, a snapshot and a + shard log all come through here). A peer's record is stored under the stamp it carries, so one whose storedAt is missing, empty or + no stamp is refused with ValueError and nothing is minted or stored (variants:ruling:raw-put-without-stamp-refused).""" + stored_at = record.get("storedAt") + if parse_stored_at(stored_at) is None: + raise ValueError("A replicated raw record must carry a storedAt stamp.") return CultCacheEnvelope( key=str(record["recordKey"]), type=document.type, schema_id=schema_id, payload=bytes(record["payload"]), - stored_at=str(record.get("storedAt") or ""), + stored_at=stored_at, catalog_entry=document.catalog_entry(), )