From 892765ba97824b8de8f16b7a74ec32a2020eeda8 Mon Sep 17 00:00:00 2001 From: Libby Date: Thu, 12 Nov 2026 04:24:07 +0000 Subject: [PATCH] Python: a record with unread slots is held read-only (slot-readonly-py) (#126) --- packages/cultcache-py/CHANGELOG.md | 2 + packages/cultcache-py/README.md | 11 + .../cultcache-py/src/cultcache_py/__init__.py | 2 + .../cultcache-py/src/cultcache_py/cache.py | 113 +++- .../src/cultcache_py/documents.py | 90 +++- .../cultcache-py/src/cultcache_py/stores.py | 16 +- .../cultcache-py/tests/test_slot_readonly.py | 492 ++++++++++++++++++ .../cultcache-py/tests/test_union_parity.py | 8 +- 8 files changed, 704 insertions(+), 30 deletions(-) create mode 100644 packages/cultcache-py/tests/test_slot_readonly.py diff --git a/packages/cultcache-py/CHANGELOG.md b/packages/cultcache-py/CHANGELOG.md index 95f82042..08fb4d1f 100644 --- a/packages/cultcache-py/CHANGELOG.md +++ b/packages/cultcache-py/CHANGELOG.md @@ -9,6 +9,8 @@ ### Breaking +- A record whose payload carries a slot past its reader's declared slots, at any depth (the document, a nested record, a list element, a union arm, a unit arm), is read, held read-only, and every write or removal of its key (`put`, `put_global`, `update`, `put_envelope`, `put_envelopes`, `delete`, `delete_global`) is refused with `SchemaConflictError` before any store is touched (`cultlib-gaps:ruling:extra-trailing-slots`). Reading dropped those slots silently and the next write shed them. `SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` and the message names the type and both counts, never a slot value. A depth-0 mark is consented, and the record writable, only when the record is stored under an id the document lists in `compatible_schema_ids` and that id's catalog entry has slots past the document's, covering the mark's slot count (`cultlib-gaps:ruling:slot-mark-consent`); a nested mark, or a mark under the document's own id, keeps the record read-only whatever is declared. Missing trailing slots keep their declared defaults. An untouched marked record keeps its stored bytes through every other write. Added `CultCache.is_writable(document, key)` (advisory; the write still decides: it also answers False for a foreign-carried key, for a key held as a type the document does not own, and for a global document at a non-global key; `delete` asks the same judges, so removing a key held as another type is refused with `SchemaConflictError` unless the document owns that record's schema id (its own or one it lists compatible), in which case the removal removes that record, its file row and its lookups, as a write of the key would replace it; it returned silently and removed nothing), `DocumentDefinition.decode_payload_marked`, and `UnreadSlots`. A document with a custom `decode` (no slot codec) never marks. A store holding a variant record is still unreadable here. Implies a minor bump while the version is 0.x (0.3.0 to 0.4.0): code that wrote back a record from a newer writer now gets `SchemaConflictError`. + - A cache holds one record per key whatever its type (`variants:ruling:cache-refuses-retype`). A `put`, `put_envelope`, `put_envelopes` or `put_global` of a second type under a key the cache holds, or under a key a store carries foreign, is refused with `SchemaConflictError` naming the key, even with no store, and changes nothing; retype a key by deleting it, then putting the new type. A write may replace another registered type's record only when the written type owns its schema id (declares it or lists it compatible) and both types share one home store. A load that reads one key from two stores, whatever the types, is refused with `CultCacheError` naming the key and both types, and admits nothing. A store that holds a record no registered type claims now loads and is carried: the cache does not serve it, the store keeps its bytes through every write, `delete` or `put` at its key is refused with `SchemaConflictError`, and registering a type that declares its schema id claims it at the next load. It raised `CultCacheError` "Unknown persisted document type". Two global types share the key `__global__`, so a second global type in one cache is now refused, and a store that already holds two global types at `__global__` (a JSONL store written by caches that never pulled could produce one) no longer loads: the pull raises `CultCacheError` and admits nothing. C# refuses a second persisted global on pull the same way (`cultcache-store-composition.md`, "Globals never invented"); repair such a file by deleting one of the two records with the runtime that wrote it. - Added `foreign_records()` and the frozen `ForeignRecord(key, schema_id, schema_name, stored_at)`, sorted by key in UTF-16 code-unit order. `stores.retyped` and `stores.written_owns` lose their underscore: the cache judge and the single-file store's commit guard ask one predicate. diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 43e6c90e..72d51562 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -97,6 +97,7 @@ store. - `update_global(...)` - `delete(...)` - `delete_global(...)` +- `is_writable(document, key)` (advisory: would a write or removal of the key be refused; see below) - `snapshot()` ## Indexes And Roles @@ -180,6 +181,16 @@ The payload is a MessagePack array. Field keys are durable slot indexes: - newly added fields should use new keys - fields with defaults tolerate missing or nil slots when older payloads are read +A payload carrying a slot past the slots a reader declares, at any depth (the document, a nested record, a list element, a +union arm, a unit arm), was written by a newer build. It is read, and the record is held read-only: `put`, `update`, +`put_envelope(s)`, `delete` and their global forms refuse its key with `SchemaConflictError` before any store is touched, and +`SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` naming the +first type read with extra slots (never a slot value). `CultCache.is_writable(document, key)` answers the same question before +the write: False for such a record, for a key a store carries foreign, and for a key held as another type's record; the write +still decides. `DocumentDefinition.decode_payload_marked(payload)` returns the value and its `UnreadSlots` (or None). Declaring the +stored id in `compatible_schema_ids` consents to the loss of depth-0 slots that id's catalog entry reports; a nested mark never is. +A store keeps a read-only record's bytes through every other write. A document with a custom `decode` never marks. + That matches the Rust `#[derive(DatabaseEntry)]` formatter shape and the C# `[Key(n)]` intent: schema evolution comes from stable field slots, not source member order. diff --git a/packages/cultcache-py/src/cultcache_py/__init__.py b/packages/cultcache-py/src/cultcache_py/__init__.py index dc73ab1b..4873b6ed 100644 --- a/packages/cultcache-py/src/cultcache_py/__init__.py +++ b/packages/cultcache-py/src/cultcache_py/__init__.py @@ -9,6 +9,7 @@ from .documents import ( DatabaseEntryField, DocumentDefinition, UnionArm, + UnreadSlots, database_entry_field, define_database_entry_type, define_document_registry, @@ -38,6 +39,7 @@ __all__ = [ "SlotCodec", "StoreUnreadableError", "UnionArm", + "UnreadSlots", "leaves", "database_entry_field", "define_database_entry_type", diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 315e64ac..85b01d89 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -8,7 +8,7 @@ from dataclasses import dataclass, field, replace from typing import Any, Callable, Generic, Iterable, TypeVar from .backing_store import BackingStore, CultCacheEnvelope -from .documents import DocumentDefinition, extract_value +from .documents import DocumentDefinition, UnreadSlots, extract_value from .stores import SchemaConflictError, retyped, written_owns T = TypeVar("T") @@ -123,6 +123,10 @@ class _State: # The records the last pull read that no registered type claims, sorted by key. Derived from the pull and swapped with the # rest of the state, written nowhere else; it decides only refusals at its keys. A store keeps their bytes. foreign: list["ForeignRecord"] = field(default_factory=list) + # The records held read-only, by key (a key holds one record) with the type that holds it: the payload carried slots past the + # declared ones and nothing consents to the loss. Derived from the load or the write that admitted the record and swapped with + # the rest of the state; it decides only refusals at its keys, and any other key is absent. + unread: dict[str, tuple[str, UnreadSlots]] = field(default_factory=dict) @dataclass(frozen=True) @@ -337,6 +341,7 @@ class CultCache: routes, generic = self._state.stores_by_type, self._state.generic_store seen_globals: set[str] = set() foreign: list[ForeignRecord] = [] + loaded_unread: dict[str, tuple[str, UnreadSlots]] = {} type_at_key: dict[str, tuple[str, int]] = {} # one record per key across every store: its type (or schema name) and store stores: list[BackingStore] = [] for candidate in [*self._state.stores_by_type.values(), self._state.generic_store]: @@ -376,8 +381,10 @@ class CultCache: loaded_legacy_global_keys[envelope.type] = envelope.key envelope = replace(envelope, key=GLOBAL_KEY) self._claim_key(type_at_key, envelope.key, envelope.type, position) - envelope = self._stamped(document, envelope) - value = document.decode_payload(envelope.payload) + arrived, envelope = envelope, self._stamped(document, envelope) + value, unread = self._decode_marked(document, arrived) + if unread is not None: + loaded_unread[envelope.key] = (envelope.type, unread) loaded_values.setdefault(envelope.type, {})[envelope.key] = value read_values[(envelope.type, envelope.key)] = value loaded_envelopes.setdefault(envelope.type, {})[envelope.key] = envelope @@ -393,6 +400,7 @@ class CultCache: self._state.lookup_keys, self._state.legacy_global_keys, self._state.foreign, + self._state.unread, ) = ( loaded_values, loaded_envelopes, @@ -401,6 +409,7 @@ class CultCache: loaded_lookup_keys, loaded_legacy_global_keys, sorted(foreign, key=lambda record: _utf16_order(record.key)), + loaded_unread, ) @staticmethod @@ -491,9 +500,9 @@ class CultCache: def put_envelope(self, document: DocumentDefinition[T], envelope: CultCacheEnvelope) -> T: self._assert_registered(document) self._check_envelope(document, envelope) - envelope = self._stamped(document, envelope) - value = document.decode_payload(envelope.payload) - self._write(document, [(envelope, value)], batch=False) + arrived, envelope = envelope, self._stamped(document, envelope) + value, unread = self._decode_marked(document, envelope) + self._write(document, [(envelope, value)], batch=False, unread={envelope.key: unread}) return value @_mutating @@ -501,13 +510,33 @@ class CultCache: self._assert_registered(document) values: list[T] = [] stamped: list[CultCacheEnvelope] = [] + unread: dict[str, UnreadSlots | None] = {} for envelope in envelopes: self._check_envelope(document, envelope) stamped.append(self._stamped(document, envelope)) - values.append(document.decode_payload(envelope.payload)) - self._write(document, list(zip(stamped, values)), batch=True) + value, unread[envelope.key] = self._decode_marked(document, stamped[-1]) + values.append(value) + self._write(document, list(zip(stamped, values)), batch=True, unread=unread) return values + @staticmethod + def _decode_marked(document: DocumentDefinition[Any], held_as: CultCacheEnvelope) -> tuple[Any, UnreadSlots | None]: + """The payload decoded under the document, with the mark of any slot it carried past the declared ones, or no mark when the + loss is consented. `held_as` is the envelope under the id the record is stored under: a load's arrival, a write's stamped + envelope, so a key is writable in memory exactly when it is after a reload. A depth-0 mark is consented only when that id is + one the document lists compatible and its catalog entry has slots past the document's that cover the mark's slot count (the + loss a compatible-id declaration consents to, cultlib-gaps:ruling:slot-mark-consent). Any other mark keeps the record read-only.""" + value, slots = document.decode_payload_marked(held_as.payload) + if slots is None: + return value, None + entry, stored = document.catalog_entry(), held_as.catalog_entry + if slots.depth0_only and stored is not None and held_as.schema_id in entry.compatible_schema_ids and held_as.schema_id != entry.schema_id: + declared = {member.slot for member in entry.members} + reported = [member.slot for member in stored.members if member.slot not in declared] + if reported and slots.slot_count <= max(reported) + 1: + return value, None + return value, slots + @staticmethod def _stamped(document: DocumentDefinition[Any], envelope: CultCacheEnvelope) -> CultCacheEnvelope: """The record under its registered schema's id, whatever id the envelope arrived under, so the catalog entry a store @@ -527,7 +556,12 @@ class CultCache: raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}") def _write( - self, document: DocumentDefinition[Any], records: list[tuple[CultCacheEnvelope, Any]], *, batch: bool + self, + document: DocumentDefinition[Any], + records: list[tuple[CultCacheEnvelope, Any]], + *, + batch: bool, + unread: dict[str, UnreadSlots | None] | None = None, ) -> None: """Validates every record (home and name and index extractors included) before the store is touched; once the store accepts, applying to the cache has no fallible step.""" @@ -535,6 +569,7 @@ class CultCache: # A batch naming a key twice lands the last record for it, as C# CultCacheBatch stages one operation per key. The # judge, the store and the apply all read this one list. records = list({envelope.key: (envelope, value) for envelope, value in records}.values()) + self._refuse_unread([envelope.key for envelope, _ in records]) name_extractor = self._state.name_extractors.get(type) index_extractors = self._state.index_extractors.get(type, {}) admitted = [ @@ -566,6 +601,9 @@ class CultCache: self._remove_lookups(type, envelope.key) values[envelope.key] = value envelopes[envelope.key] = envelope + # A held mark refused this write above, so a key reaching here carries none to clear. + if (mark := (unread or {}).get(envelope.key)) is not None: + self._state.unread[envelope.key] = (type, mark) lookup_keys[envelope.key] = keys if keys.name is not None: self._state.names[(type, keys.name)] = envelope.key @@ -592,6 +630,51 @@ class CultCache: replaced.append((other, envelope.key)) return replaced + @_locked + def is_writable(self, document: DocumentDefinition[Any], key: str) -> bool: + """Whether a write or removal of the key by this document would pass the judges that run before any store is touched: False for + a record held read-only (its payload carried slots past the declared ones, at any depth, and nothing consents to the loss, so + writing it would drop data a newer writer stored), for a key a store carries foreign, and for a key held as another type's + record this document does not own. Advisory (the write still decides): it answers at call time, and a key nothing holds is + writable.""" + self._assert_registered(document) + try: + self._refuse_at(document, key) + except (SchemaConflictError, CultCacheError): + return False + return True + + def _refuse_at(self, document: DocumentDefinition[Any], key: str) -> list[tuple[str, str]]: + """The judges a write or removal of the key by this document passes before any store is touched, asked by `is_writable` and + by `delete`: a global document writes only its global key, a record held read-only is never replaced, and a key held as + another type's record or carried foreign is refused (`_refuse_retype`).""" + if document.global_document and key != GLOBAL_KEY: + raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}") + self._refuse_unread([key]) + entry = document.catalog_entry() + probe = CultCacheEnvelope.create(key=key, type=document.type, payload=b"", schema_id=entry.schema_id, catalog_entry=entry) + return self._refuse_retype(document, [probe]) + + def _refuse_unread(self, keys: list[str]) -> None: + """The one check every write and removal asks before any store is touched.""" + for key in keys: + held = self._state.unread.get(key) + if held is None: + continue + type, slots = held + raise SchemaConflictError( + self._state.envelopes[type][key].schema_id or type, + [self._state.documents[type].catalog_entry().schema_name], + key, + message=( + f"{self._state.documents[type].catalog_entry().schema_name} record {key} is read-only: {slots.type_name} was read with " + f"{slots.slot_count} slots where this build declares {slots.declared_count} ({slots.occurrences} object(s) in the " + "record). Writing it would drop data a newer writer stored; read it with types that declare those slots, or " + "write it from that build." + ), + unread_slots=slots, + ) + @staticmethod def _foreign_refusal(record: ForeignRecord) -> SchemaConflictError: return SchemaConflictError( @@ -644,16 +727,18 @@ class CultCache: def _delete(self, document: DocumentDefinition[Any], key: str) -> None: self._assert_registered(document) - if key not in self._state.envelopes.get(document.type, {}): - carried = next((record for record in self._state.foreign if record.key == key), None) - if carried is not None: - raise self._foreign_refusal(carried) + # A key held as another type's record this document owns is removed, as a write of the key would replace it (C# Remove is by key). + owned = self._refuse_at(document, key) store = self._store_for_type(document.type) legacy_key = self._state.legacy_global_keys.get(document.type) if key == GLOBAL_KEY else None if store is not None: - store.delete(document.type, key if legacy_key is None else legacy_key) + # One record per key: the owned record when this document does not hold the key, else its own, in one store call. + holder = owned[0][0] if owned else document.type + store.delete(holder, key if legacy_key is None else legacy_key) if legacy_key is not None: del self._state.legacy_global_keys[document.type] + for other, _ in owned: + self._drop(other, key) self._drop(document.type, key) def _drop(self, type: str, key: str) -> None: diff --git a/packages/cultcache-py/src/cultcache_py/documents.py b/packages/cultcache-py/src/cultcache_py/documents.py index 6101c136..4f7dfe7d 100644 --- a/packages/cultcache-py/src/cultcache_py/documents.py +++ b/packages/cultcache-py/src/cultcache_py/documents.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import threading from dataclasses import MISSING, dataclass, field from collections.abc import Mapping, Sequence, Set from itertools import chain @@ -30,6 +31,49 @@ MAX_SLOT_KEY = 65535 _STACK_FAULTS: tuple[type[BaseException], ...] = (_MsgpackStackError,) +@dataclass(frozen=True) +class UnreadSlots: + """What a payload decode found past the slots this runtime declares, at any depth: the first type read with extra slots, how + many slots it carried against how many it declares, and how many objects in the payload did. It holds no slot value, so a + refusal can name it. `depth0_only` is True when every mark sat in the document's own record (the half of consent the payload + decides). A record carrying it is read-only unless the schema comparison explains every mark (see `CultCache.is_writable`).""" + + type_name: str + slot_count: int + declared_count: int + occurrences: int + depth0_only: bool + + +class _SlotScope: + """Opened by `DocumentDefinition.decode_payload_marked`; the record codec and the union codec note into it. It is thread-local, + so no codec signature carries it, and a codec decoded outside a payload decode notes nothing.""" + + def __init__(self) -> None: + self.depth = 0 # records entered: 0 while the document's own record is read + self.first: tuple[str, int, int] | None = None + self.occurrences = 0 + self.depth0_only = True + + def note(self, type_name: str, slot_count: int, declared_count: int, depth0: bool) -> None: + if self.first is None: + self.first = (type_name, slot_count, declared_count) + self.occurrences += 1 + self.depth0_only = self.depth0_only and depth0 + + def result(self) -> UnreadSlots | None: + if self.first is None: + return None + return UnreadSlots(*self.first, self.occurrences, self.depth0_only) + + +_scope = threading.local() + + +def _current_scope() -> _SlotScope | None: + return getattr(_scope, "open", None) + + _CONTAINERS = (Mapping, Sequence, Set) _SCALAR_SEQUENCES = (str, bytes, bytearray, memoryview) @@ -185,15 +229,25 @@ class DocumentDefinition(Generic[T]): return self.payload_encoder(tree) def decode_payload(self, payload: bytes) -> T: + return self.decode_payload_marked(payload)[0] + + def decode_payload_marked(self, payload: bytes) -> tuple[T, UnreadSlots | None]: + """The value and what the decode found past the declared slots, at any depth. The payload depth check runs first: a + payload past the limit is refused, never marked.""" try: tree = self.payload_decoder(payload) except _STACK_FAULTS: raise CultCachePayloadDepthError(self.type, "read") from None _refuse_past_depth_limit(tree, self.type, "read") - value = self.decode(tree) + outer, opened = _current_scope(), _SlotScope() + _scope.open = opened + try: + value = self.decode(tree) + finally: + _scope.open = outer if self.validate: self.validate(value) - return value + return value, opened.result() def __post_init__(self) -> None: # Judged here so every constructor passes it: the factories, the dataclass constructor and dataclasses.replace. @@ -418,6 +472,9 @@ def define_union( raise ValueError(f"Union {name!r} value is malformed: expected [armKey, [armSlots]]") arm_name, codec = by_key[key] if codec is None: + scope = _current_scope() + if scope is not None and raw[1]: + scope.note(arm_name, len(raw[1]), 0, False) return UnionArm(arm_name, None) try: return UnionArm(arm_name, codec.decode(raw[1])) @@ -479,17 +536,26 @@ def _record_codec( def decode(raw: Any) -> Any: if not isinstance(raw, list): raise ValueError(f"DatabaseEntry document type {type!r} expected a MessagePack array") + scope = _current_scope() + if scope is not None and len(raw) > max_key + 1: + scope.note(type, len(raw), max_key + 1, scope.depth == 0) values: dict[str, Any] = {} - for field in normalized: - if field.key < len(raw): - try: - values[field.name] = field.codec.decode(raw[field.key]) - except ValueError as error: - raise ValueError(f"{type} slot {field.key} ({field.name}): {error}") from None - elif field.default is MISSING: - values[field.name] = None - else: - values[field.name] = field.default() if callable(field.default) else field.default + if scope is not None: + scope.depth += 1 + try: + for field in normalized: + if field.key < len(raw): + try: + values[field.name] = field.codec.decode(raw[field.key]) + except ValueError as error: + raise ValueError(f"{type} slot {field.key} ({field.name}): {error}") from None + elif field.default is MISSING: + values[field.name] = None + else: + values[field.name] = field.default() if callable(field.default) else field.default + finally: + if scope is not None: + scope.depth -= 1 if cls is None: return values return cls(**values) diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index 61841956..c15ac32e 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -13,6 +13,7 @@ from typing import Any, Callable, Iterator, TypeVar import msgpack +from .documents import UnreadSlots from .backing_store import ( CultCacheEnvelope, CultCacheSchemaCatalogEntry, @@ -46,9 +47,19 @@ class SchemaConflictError(ValueError): """A write the catalog cannot describe: records of different types under one schema id, two arrived schemas that share an id and disagree on the schema name, or a record no chosen entry publishes. Nothing is written. It names the id, the names involved (`schema_names`) and a record key (`record_key`). A cache raises it too, with an empty record key, when a - document registers a schema id another registered document already carries.""" + document registers a schema id another registered document already carries. `unread_slots` is None except for the refusal to + write or remove a record held read-only because its payload carries slots past the declared ones; its message names the type + and both slot counts and never holds a slot value.""" - def __init__(self, schema_id: str, schema_names: list[str], record_key: str, *, message: str | None = None) -> None: + def __init__( + self, + schema_id: str, + schema_names: list[str], + record_key: str, + *, + message: str | None = None, + unread_slots: UnreadSlots | None = None, + ) -> None: super().__init__( message or f"Schema id {schema_id!r} cannot describe record {record_key!r}: {', '.join(repr(name) for name in schema_names)}" @@ -56,6 +67,7 @@ class SchemaConflictError(ValueError): self.schema_id = schema_id self.schema_names = schema_names self.record_key = record_key + self.unread_slots = unread_slots class JsonLinesBackingStore: diff --git a/packages/cultcache-py/tests/test_slot_readonly.py b/packages/cultcache-py/tests/test_slot_readonly.py new file mode 100644 index 00000000..77d4a989 --- /dev/null +++ b/packages/cultcache-py/tests/test_slot_readonly.py @@ -0,0 +1,492 @@ +from __future__ import annotations + +import dataclasses +import shutil +import threading +import tempfile +import unittest +from pathlib import Path + +import msgpack +from cultcache_py.cache import CultCacheError + +from cultcache_py import ( + CultCache, + CultCachePayloadDepthError, + SlotCodec, + SchemaConflictError, + SingleFileMessagePackBackingStore, + UnionArm, + UnreadSlots, + database_entry_field, + define_database_entry_type, + define_record_codec, + leaves as t, +) +import test_union_parity as _parity +from test_union_parity import ( + VECTORS, + _surface_value, + _vector, + surface_document, + unit_outcome, +) + +SLOT = "extra-slot" # the canary: the value of every extra slot in the vectors; no refusal may carry it + +# The vectors name the C# types; this runtime names a type by its codec's name, a document by its document type. +_NAMES = {"UnionSurface": surface_document.type, "SurfaceDeclaring": surface_document.type, "Denied2": "Denied"} + +slot_unit_document = define_database_entry_type( + "tests.slot_unit", + [database_entry_field("name", 0, codec=t.string), database_entry_field("outcome", 1, codec=unit_outcome)], + name="name", +) +# The reader of extra-slot-consent.cc: the surface document's slots under a new schema id that lists the old one compatible. +declaring_document = define_database_entry_type( + "tests.union_surface", + [ + database_entry_field("id", 0, codec=t.string), + database_entry_field("version", 1, codec=t.int32), + database_entry_field("root", 2, codec=_parity._node), + database_entry_field("outcomes", 3, codec=t.list(_parity.outcome)), + database_entry_field("part", 4, codec=t.nullable(_parity.item)), + database_entry_field("crafted", 5, codec=t.nullable(_parity.crafted)), + database_entry_field("missing", 6, codec=_parity._node), + ], + name="id", + schema_id="tests.union_surface.v2", + schema_name="tests.union_surface", + compatible_schema_ids=("tests.union_surface.wide",), +) + +# file -> (type, slots carried, slots declared, every mark at depth 0): the README table. +_PAYLOADS = { + "extra-slot-record.msgpack": ("UnionSurface", 8, 7, True), + "extra-slot-nested-record.msgpack": ("Composition", 3, 2, False), + "extra-slot-arm.msgpack": ("Box", 3, 2, False), + "extra-slot-list-element.msgpack": ("Text", 3, 2, False), + "depth-skipped-at-limit.msgpack": ("UnionSurface", 8, 7, True), +} + + +def _copy_store_without_variants(source: Path, target: Path) -> None: + """This runtime holds no variant records (a five-slot record makes a store unreadable here), so the store vector is + walked with its variant row removed; every other byte is the C# writer's.""" + [snapshot] = list(msgpack.Unpacker(open(source, "rb"), raw=False)) + snapshot[2] = [record for record in snapshot[2] if len(record) <= 4] + target.write_bytes(msgpack.packb(snapshot, use_bin_type=True)) + + +def _open(path: Path, *documents) -> CultCache: + builder = CultCache.builder() + for document in documents: + builder.register_document_type(document) + cache = builder.add_generic_store(SingleFileMessagePackBackingStore(path)).build() + cache.pull_all_backing_stores() + return cache + + +def _rows(): + return msgpack.unpackb(_vector("slot-outcomes.msgpack"), raw=False) + + +class PayloadMarkTests(unittest.TestCase): + def test_every_read_only_payload_decodes_and_names_its_mark(self) -> None: + for file, (type_name, carried, declared, depth0) in _PAYLOADS.items(): + with self.subTest(file=file): + value, mark = surface_document.decode_payload_marked(_vector(file)) + self.assertEqual(value, _surface_value()) + self.assertEqual(mark, UnreadSlots(_NAMES.get(type_name, type_name), carried, declared, 1, depth0)) + self.assertEqual(surface_document.decode_payload(_vector(file)), _surface_value()) + + def test_the_outcomes_manifest_still_counts_nine_rows_and_four_read_only(self) -> None: + rows = msgpack.unpackb(_vector("outcomes.msgpack"), raw=False) + self.assertEqual(len(rows), 9) + self.assertEqual([row[0] for row in rows if row[1] == "read-only"], [f for f in _PAYLOADS if f != "depth-skipped-at-limit.msgpack"]) + + def test_a_clean_payload_and_a_short_one_carry_no_mark(self) -> None: + self.assertEqual(surface_document.decode_payload_marked(_vector("union-surface.msgpack")), (_surface_value(), None)) + slots = msgpack.unpackb(_vector("union-surface.msgpack"), raw=False) + slots[2] = [1, ["only-id"]] # a Text arm with a missing trailing slot: defaults, no mark + _, mark = surface_document.decode_payload_marked(msgpack.packb(slots, use_bin_type=True)) + self.assertIsNone(mark) + + def test_a_retired_slot_below_the_highest_key_does_not_mark(self) -> None: + slots = msgpack.unpackb(_vector("union-surface.msgpack"), raw=False) + # Image's declared slots are 0, 1, 3 (2 retired): four carried, four declared. + image = slots[2][1][1][2] + self.assertEqual(image[0], 5) + image[1][2] = "retired-slot-value" + _, mark = surface_document.decode_payload_marked(msgpack.packb(slots, use_bin_type=True)) + self.assertIsNone(mark) + + def test_the_depth_check_runs_before_the_mark(self) -> None: + with self.assertRaises(CultCachePayloadDepthError): + surface_document.decode_payload_marked(_vector("depth-skipped-past-limit.msgpack")) + + def test_the_unit_arm_with_extra_slots_is_read_past_and_marked(self) -> None: + payload = msgpack.packb(["unit-arm", [2, [SLOT]]], use_bin_type=True) + value, mark = slot_unit_document.decode_payload_marked(payload) + self.assertEqual(value, {"name": "unit-arm", "outcome": UnionArm("Denied")}) + self.assertEqual(mark, UnreadSlots("Denied", 1, 0, 1, False)) + self.assertEqual(slot_unit_document.decode_payload_marked(msgpack.packb(["ok", [2, []]]))[1], None) + + def test_a_codec_decoded_outside_a_payload_decode_notes_nothing(self) -> None: + self.assertEqual(unit_outcome.decode([2, [SLOT]]), UnionArm("Denied")) + + def test_the_scope_belongs_to_its_thread(self) -> None: + def decode_elsewhere(raw): + thread = threading.Thread(target=lambda: unit_outcome.decode([2, [SLOT]])) + thread.start() + thread.join() + return raw + + outer = define_database_entry_type("tests.threaded", [database_entry_field("a", 0, codec=SlotCodec(lambda v: v, decode_elsewhere))]) + self.assertIsNone(outer.decode_payload_marked(msgpack.packb(["x"]))[1]) + + def test_a_nested_payload_decode_restores_the_outer_scope(self) -> None: + def decode_clean_inside(raw): + surface_document.decode_payload_marked(_vector("union-surface.msgpack")) + return raw + + inner = define_record_codec("Inner", [database_entry_field("x", 0, codec=t.string)]) + outer = define_database_entry_type( + "tests.nested", + [ + database_entry_field("a", 0, codec=SlotCodec(lambda v: v, decode_clean_inside)), + database_entry_field("b", 1, codec=inner), + ], + ) + _, mark = outer.decode_payload_marked(msgpack.packb(["x", ["v", SLOT]])) + self.assertEqual(mark, UnreadSlots("Inner", 2, 1, 1, False)) + + +class StoreRowTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = Path(tempfile.mkdtemp()) + self.addCleanup(shutil.rmtree, self.tmp, True) + + def _store(self, name: str) -> tuple[Path, CultCache]: + path = self.tmp / name + _copy_store_without_variants(VECTORS / name, path) + documents = (declaring_document,) if name == "extra-slot-consent.cc" else (surface_document, slot_unit_document) + return path, _open(path, *documents) + + def _document(self, store: str, key: str): + if key == "unit-arm": + return slot_unit_document + return declaring_document if store == "extra-slot-consent.cc" else surface_document + + def test_every_row_is_writable_exactly_where_the_vector_says_and_refuses_with_its_mark(self) -> None: + walked = 0 + for store, key, writable, explained, type_name, carried, declared, occurrences, flatten in _rows(): + if flatten is not None: + continue # a variant: this runtime holds none (deviation) + walked += 1 + with self.subTest(store=store, key=key): + path, cache = self._store(store) + document = self._document(store, key) + before = path.read_bytes() + self.assertEqual(cache.is_writable(document, key), writable) + held = cache.get(document, key) + self.assertIsNotNone(held) + if writable: + continue + self.assertEqual(explained, False) + for attempt in ( + lambda: cache.put(document, key, held), + lambda: cache.delete(document, key), + lambda: cache.put_envelope(document, cache.get_required_envelope(document, key)), + lambda: cache.put_envelopes(document, [cache.get_required_envelope(document, key)]), + lambda: cache.update(document, key, lambda value: value), + ): + with self.assertRaises(SchemaConflictError) as caught: + attempt() + refusal = caught.exception + self.assertEqual( + refusal.unread_slots, + UnreadSlots(_NAMES.get(type_name, type_name), carried, declared, occurrences, refusal.unread_slots.depth0_only), + ) + self.assertEqual(refusal.record_key, key) + message = str(refusal) + self.assertNotIn(SLOT, message) + self.assertIn(refusal.unread_slots.type_name, message) + self.assertIn(f"{carried} slots", message) + self.assertIn(f"declares {declared}", message) + self.assertEqual(path.read_bytes(), before) + self.assertTrue(cache.get(document, key) is held) + self.assertEqual(walked, 7) # eight rows, the variant not walked + + def test_a_key_no_store_holds_and_another_key_are_writable_while_a_marked_one_is_held(self) -> None: + path, cache = self._store("extra-slot-store.cc") + self.assertFalse(cache.is_writable(surface_document, "marked")) + self.assertTrue(cache.is_writable(surface_document, "nothing-here")) + cache.put(surface_document, "other", cache.get_required(surface_document, "clean")) + self.assertEqual(cache.get(surface_document, "other"), _surface_value()) + + def test_consent_lands_an_explained_mark_and_refuses_the_others(self) -> None: + path, cache = self._store("extra-slot-consent.cc") + self.assertTrue(cache.is_writable(declaring_document, "consented")) + cache.put(declaring_document, "consented", cache.get_required(declaring_document, "consented")) + [stored] = [e for e in SingleFileMessagePackBackingStore(path).pull_all() if e.key == "consented"] + self.assertEqual(stored.schema_id, "tests.union_surface.v2") + self.assertEqual(len(msgpack.unpackb(stored.payload, raw=False)), 7) # the consented slot is gone + self.assertTrue(cache.is_writable(declaring_document, "consented")) + for key in ("over-consent", "nested"): + self.assertFalse(cache.is_writable(declaring_document, key), key) + with self.assertRaises(SchemaConflictError): + cache.put(declaring_document, key, cache.get_required(declaring_document, key)) + + def test_a_reload_marks_again_and_a_write_of_a_clean_envelope_over_a_marked_key_is_refused(self) -> None: + path, cache = self._store("extra-slot-store.cc") + envelope = cache.get_required_envelope(surface_document, "clean") + with self.assertRaises(SchemaConflictError): + cache.put_envelope(surface_document, type(envelope)(**{**envelope.__dict__, "key": "marked"})) + cache.pull_all_backing_stores() + self.assertFalse(cache.is_writable(surface_document, "marked")) + + def test_a_written_envelope_that_carries_extra_slots_is_held_read_only(self) -> None: + path, cache = self._store("extra-slot-store.cc") + envelope = cache.get_required_envelope(surface_document, "clean") + carrying = type(envelope)(**{**envelope.__dict__, "key": "arrived", "payload": _vector("extra-slot-arm.msgpack")}) + cache.put_envelope(surface_document, carrying) + self.assertFalse(cache.is_writable(surface_document, "arrived")) + with self.assertRaises(SchemaConflictError): + cache.delete(surface_document, "arrived") + cache.pull_all_backing_stores() + self.assertFalse(cache.is_writable(surface_document, "arrived")) + + def test_a_batch_of_envelopes_that_carry_extra_slots_is_held_read_only(self) -> None: + path, cache = self._store("extra-slot-store.cc") + envelope = cache.get_required_envelope(surface_document, "clean") + batch = [ + type(envelope)(**{**envelope.__dict__, "key": key, "payload": _vector(file)}) + for key, file in (("b1", "extra-slot-arm.msgpack"), ("b2", "union-surface.msgpack")) + ] + cache.put_envelopes(surface_document, batch) + self.assertEqual([cache.is_writable(surface_document, key) for key in ("b1", "b2")], [False, True]) + + def test_a_mark_under_an_id_the_reader_does_not_list_compatible_is_not_consented(self) -> None: + unlisted = define_database_entry_type( + "tests.union_surface", + [database_entry_field("id", 0, codec=t.string)] + [database_entry_field(n, i, codec=c) for n, i, c in ( + ("version", 1, t.int32), ("root", 2, _parity._node), ("outcomes", 3, t.list(_parity.outcome)), + ("part", 4, t.nullable(_parity.item)), ("crafted", 5, t.nullable(_parity.crafted)), ("missing", 6, _parity._node))], + name="id", + schema_id="tests.union_surface.v2", + schema_name="tests.union_surface", + ) + path = self.tmp / "unlisted.cc" + _copy_store_without_variants(VECTORS / "extra-slot-consent.cc", path) + cache = _open(path, unlisted) + self.assertFalse(cache.is_writable(unlisted, "consented")) + + def test_a_written_envelope_is_judged_on_the_id_it_is_stored_under_so_a_reload_agrees(self) -> None: + path, cache = self._store("extra-slot-consent.cc") + arrived = next(e for e in SingleFileMessagePackBackingStore(path).pull_all() if e.key == "consented") + self.assertEqual(arrived.schema_id, "tests.union_surface.wide") + cache.put_envelope(declaring_document, dataclasses.replace(arrived, key="via-envelope")) + cache.put_envelopes(declaring_document, [dataclasses.replace(arrived, key="via-batch")]) + in_memory = [cache.is_writable(declaring_document, k) for k in ("via-envelope", "via-batch")] + cache.pull_all_backing_stores() + self.assertEqual([cache.is_writable(declaring_document, k) for k in ("via-envelope", "via-batch")], in_memory) + self.assertEqual(in_memory, [False, False]) + + def test_is_writable_agrees_with_the_write_for_a_key_a_store_carries_foreign(self) -> None: + path = self.tmp / "foreign.cc" + _copy_store_without_variants(VECTORS / "extra-slot-store.cc", path) + cache = _open(path, surface_document) # tests.slot_unit is not registered: its record is carried foreign + self.assertEqual([r.key for r in cache.foreign_records()], ["unit-arm"]) + self.assertFalse(cache.is_writable(surface_document, "unit-arm")) + with self.assertRaises(SchemaConflictError): + cache.put(surface_document, "unit-arm", _surface_value()) + with self.assertRaises(SchemaConflictError): + cache.delete(surface_document, "unit-arm") + + def test_every_key_of_a_batch_is_checked_wherever_the_marked_one_stands(self) -> None: + path, cache = self._store("extra-slot-store.cc") + before = path.read_bytes() + envelope = cache.get_required_envelope(surface_document, "clean") + fresh = dataclasses.replace(envelope, key="fresh") + marked = dataclasses.replace(envelope, key="marked") + for batch in ([fresh, marked], [marked, fresh], [fresh, dataclasses.replace(envelope, key="f2"), marked]): + with self.assertRaises(SchemaConflictError) as caught: + cache.put_envelopes(surface_document, batch) + self.assertEqual(caught.exception.record_key, "marked") + self.assertEqual(path.read_bytes(), before) + self.assertIsNone(cache.get(surface_document, "fresh")) + + def test_only_the_marks_that_hold_a_record_read_only_are_kept_by_key(self) -> None: + _, store = self._store("extra-slot-store.cc") + self.assertEqual({k: v[0] for k, v in store._state.unread.items()}, {"marked": surface_document.type, "unit-arm": "tests.slot_unit"}) + _, consent = self._store("extra-slot-consent.cc") + self.assertEqual(sorted(consent._state.unread), ["nested", "over-consent"]) # the consented mark is no state + + def test_an_arrival_with_no_catalog_entry_under_a_listed_id_is_not_consented(self) -> None: + wide = next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented") + bare = dataclasses.replace(wide, catalog_entry=None) + + class Held: + def pull_all(self): + return [bare] + + def push(self, envelope): ... + + def push_all(self, envelopes): ... + + def delete(self, type, key): ... + + cache = CultCache.builder().register_document_type(declaring_document).add_generic_store(Held()).build() + cache.pull_all_backing_stores() + self.assertFalse(cache.is_writable(declaring_document, "consented")) + + def test_a_removal_is_refused_where_is_writable_says_no(self) -> None: + path, cache = self._store("extra-slot-store.cc") + before = path.read_bytes() + other = define_database_entry_type("tests.other", [database_entry_field("a", 0, codec=t.string)], name="a") + cache2 = _open(self.tmp / "extra-slot-store.cc", surface_document, slot_unit_document, other) + self.assertFalse(cache2.is_writable(other, "clean")) + with self.assertRaises(SchemaConflictError): + cache2.delete(other, "clean") + self.assertEqual(path.read_bytes(), before) + self.assertIsNotNone(cache2.get(surface_document, "clean")) + self.assertTrue(cache2.is_writable(other, "nothing-here")) + + def test_a_global_document_is_not_writable_at_a_key_other_than_the_global_one(self) -> None: + solo = define_database_entry_type("tests.solo", [database_entry_field("a", 0, codec=t.string)], global_document=True) + cache = CultCache.builder().register_document_type(solo).build() + self.assertFalse(cache.is_writable(solo, "elsewhere")) + with self.assertRaises(CultCacheError): + cache.put(solo, "elsewhere", {"a": "x"}) + with self.assertRaises(CultCacheError): + cache.delete(solo, "elsewhere") + self.assertTrue(cache.is_writable(solo, CultCache.GLOBAL_KEY)) + + def test_a_reload_replaces_the_marks_wholesale(self) -> None: + path, cache = self._store("extra-slot-store.cc") + self.assertFalse(cache.is_writable(surface_document, "marked")) + # The store is repaired by its owner: the marked record is rewritten clean. The reload must forget the old mark. + [snapshot] = list(msgpack.Unpacker(open(path, "rb"), raw=False)) + clean = next(r for r in snapshot[2] if r[0] == "clean") + for record in snapshot[2]: + if record[0] in ("marked", "unit-arm"): + record[3] = clean[3] if record[0] == "marked" else record[3] + snapshot[2] = [r for r in snapshot[2] if r[0] != "unit-arm"] + path.write_bytes(msgpack.packb(snapshot, use_bin_type=True)) + cache.pull_all_backing_stores() + self.assertTrue(cache.is_writable(surface_document, "marked")) + self.assertEqual(cache._state.unread, {}) + + def test_a_listed_id_whose_catalog_reports_no_extra_slot_does_not_consent(self) -> None: + wide = next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented") + same = dataclasses.replace(wide, catalog_entry=declaring_document.catalog_entry()) # the stored entry has no slot past the reader's + + class Held: + def pull_all(self): + return [same] + + def push(self, envelope): ... + + def push_all(self, envelopes): ... + + def delete(self, type, key): ... + + cache = CultCache.builder().register_document_type(declaring_document).add_generic_store(Held()).build() + cache.pull_all_backing_stores() + self.assertFalse(cache.is_writable(declaring_document, "consented")) + + def test_a_record_stored_under_the_readers_own_id_is_never_consented(self) -> None: + own = dataclasses.replace( + next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented"), + schema_id="tests.union_surface.v2", + ) + wider = dataclasses.replace(own.catalog_entry, schema_id="tests.union_surface.v2", compatible_schema_ids=("tests.union_surface.v2",)) + listing = dataclasses.replace( + declaring_document, schema_id="tests.union_surface.v2", compatible_schema_ids=("tests.union_surface.wide", "tests.union_surface.v2") + ) + + class Held: + def pull_all(self): + return [dataclasses.replace(own, catalog_entry=wider)] + + def push(self, envelope): ... + + def push_all(self, envelopes): ... + + def delete(self, type, key): ... + + cache = CultCache.builder().register_document_type(listing).add_generic_store(Held()).build() + cache.pull_all_backing_stores() + self.assertFalse(cache.is_writable(listing, "consented")) + + def test_a_removal_by_the_owner_of_a_key_held_as_another_type_removes_it_as_a_write_would_replace_it(self) -> None: + a = define_database_entry_type( + "tests.owned_a", [database_entry_field("a", 0, codec=t.string)], name="a", indexes={"by_a": "a"} + ) + b = define_database_entry_type( + "tests.owned_b", + [database_entry_field("a", 0, codec=t.string)], + compatible_schema_ids=(a.catalog_entry().schema_id,), + ) + path = self.tmp / "owned.cc" + writer = _open(path, a) + writer.put(a, "k", {"a": "x"}) + writer.put(a, "other", {"a": "y"}) + cache = _open(path, a, b) + self.assertTrue(cache.is_writable(b, "k")) + cache.delete(b, "k") + self.assertIsNone(cache.get(a, "k")) + self.assertIsNone(cache.get_by_index(a, "by_a", "x")) + self.assertEqual([e.key for e in SingleFileMessagePackBackingStore(path).pull_all()], ["other"]) + self.assertTrue(cache.is_writable(b, "k")) + + def test_an_owners_removal_is_one_store_call_so_the_cache_and_the_store_agree_when_it_fails(self) -> None: + a = define_database_entry_type("tests.one_a", [database_entry_field("a", 0, codec=t.string)], name="a") + b = define_database_entry_type( + "tests.one_b", [database_entry_field("a", 0, codec=t.string)], compatible_schema_ids=(a.catalog_entry().schema_id,) + ) + path = self.tmp / "one.cc" + _open(path, a).put(a, "k", {"a": "x"}) + + class SecondDeleteFails(SingleFileMessagePackBackingStore): + calls = 0 + + def delete(self, type, key): + self.calls += 1 + if self.calls > 1: + raise OSError("second delete") + super().delete(type, key) + + store = SecondDeleteFails(path) + cache = CultCache.builder().register_document_type(a).register_document_type(b).add_generic_store(store).build() + cache.pull_all_backing_stores() + cache.delete(b, "k") + self.assertEqual(store.calls, 1) + self.assertIsNone(cache.get(a, "k")) + self.assertEqual(SingleFileMessagePackBackingStore(path).pull_all(), []) + + def test_writing_another_record_and_flushing_carries_the_marked_one_unchanged(self) -> None: + path, cache = self._store("extra-slot-store.cc") + original = {e.key: e for e in SingleFileMessagePackBackingStore(path).pull_all()} + cache.put(surface_document, "clean", {**_surface_value(), "version": 9}) + after = {e.key: e for e in SingleFileMessagePackBackingStore(path).pull_all()} + for key in ("marked", "unit-arm"): + self.assertEqual(after[key].payload, original[key].payload, key) + self.assertEqual(after[key].schema_id, original[key].schema_id, key) + self.assertEqual(after[key].stored_at, original[key].stored_at, key) + self.assertNotEqual(after["clean"].payload, original["clean"].payload) + self.assertEqual(after["marked"].catalog_entry, original["marked"].catalog_entry) + + def test_the_vector_store_loads_under_the_reference_documents_and_marks_nothing_else(self) -> None: + path, cache = self._store("extra-slot-store.cc") + self.assertTrue(cache.is_writable(surface_document, "clean")) + self.assertFalse(cache.is_writable(surface_document, "marked")) + self.assertFalse(cache.is_writable(slot_unit_document, "unit-arm")) + self.assertEqual( + cache.get(slot_unit_document, "unit-arm"), {"name": "unit-arm", "outcome": UnionArm("Denied")} + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/cultcache-py/tests/test_union_parity.py b/packages/cultcache-py/tests/test_union_parity.py index 451cd030..1e19ce8d 100644 --- a/packages/cultcache-py/tests/test_union_parity.py +++ b/packages/cultcache-py/tests/test_union_parity.py @@ -261,7 +261,9 @@ class UnionSurfaceVectorTests(unittest.TestCase): else: read_only += 1 self.assertEqual(outcome_kind, "read-only", file) - self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file) + value, mark = surface_document.decode_payload_marked(payload) + self.assertEqual(value, _surface_value(), file) + self.assertIsNotNone(mark, file) # read past and marked: tests/test_slot_readonly.py pins what the mark holds self.assertEqual((refused, read_only), (5, 4)) @@ -526,7 +528,9 @@ class BoundsVectorTests(unittest.TestCase): surface_document.encode_payload(surface_document.decode_payload(payload)), _vector(detail), file ) elif outcome_kind == "read-only": - self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file) + value, mark = surface_document.decode_payload_marked(payload) + self.assertEqual(value, _surface_value(), file) + self.assertIsNotNone(mark, file) else: with self.assertRaises(ValueError, msg=file) as caught: surface_document.decode_payload(payload)