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

cultcache-py: judge consent on the stored id, index marks by key, is_writable asks the write's own judges

Browse source 
Soul s1: a written envelope is judged under the id it is stored under, so writable in memory equals writable after reload; only
marks that hold a record read-only are kept, by key (O(1) per written key); is_writable also answers False for foreign-carried
and unowned keys; tests pin batch order, thread scope, nested scope restore and the catalog-less arrival; README documents the surface.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Libby 2026-11-12 03:21:10 +00:00
commit 6d4b96fe55
4 changed files with 143 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -9,7 +9,7 @@
### 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 arrived 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), `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 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 and for a key held as a type the document does not own), `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.

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

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

90
packages/cultcache-py/src/cultcache_py/cache.py
View file

@ -123,10 +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 whose payload carried slots past the declared ones, by (type, key), with the schema comparison's verdict: a record
# held read-only. 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 an unmarked key is absent.
unread: dict[tuple[str, str], "_Unread"] = field(default_factory=dict)
# 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)
@ -140,12 +140,6 @@ class ForeignRecord:
stored_at: str
@dataclass(frozen=True)
class _Unread:
slots: UnreadSlots
explained: bool # every mark is a depth-0 slot the record's stored schema reported past the reader: what a compatible id consents to
@dataclass(frozen=True)
class _LookupKeys:
name: str | None
@ -347,7 +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[tuple[str, str], _Unread] = {}
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]:
@ -388,9 +382,9 @@ class CultCache:
envelope = replace(envelope, key=GLOBAL_KEY)
self._claim_key(type_at_key, envelope.key, envelope.type, position)
arrived, envelope = envelope, self._stamped(document, envelope)
value, unread = self._decode_marked(document, arrived, envelope)
value, unread = self._decode_marked(document, arrived)
if unread is not None:
loaded_unread[(envelope.type, envelope.key)] = unread
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
@ -507,7 +501,7 @@ class CultCache:
self._assert_registered(document)
self._check_envelope(document, envelope)
arrived, envelope = envelope, self._stamped(document, envelope)
value, unread = self._decode_marked(document, arrived, envelope)
value, unread = self._decode_marked(document, envelope)
self._write(document, [(envelope, value)], batch=False, unread={envelope.key: unread})
return value
@ -516,33 +510,32 @@ class CultCache:
self._assert_registered(document)
values: list[T] = []
stamped: list[CultCacheEnvelope] = []
unread: dict[str, _Unread | None] = {}
unread: dict[str, UnreadSlots | None] = {}
for envelope in envelopes:
self._check_envelope(document, envelope)
stamped.append(self._stamped(document, envelope))
value, unread[envelope.key] = self._decode_marked(document, envelope, stamped[-1])
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
def _decode_marked(
self, document: DocumentDefinition[Any], arrived: CultCacheEnvelope, stamped: CultCacheEnvelope
) -> tuple[Any, _Unread | None]:
"""The payload decoded under the document, with the mark of any slot it carried past the declared ones. The mark is
explained only when the record arrived under an id the document lists compatible and the schema comparison of that id's
catalog entry against the document's own reports every slot the mark carries (all at depth 0): 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(stamped.payload)
@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 = document.catalog_entry()
stored = arrived.catalog_entry
explained = False
if slots.depth0_only and stored is not None and arrived.schema_id in entry.compatible_schema_ids and arrived.schema_id != entry.schema_id:
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]
explained = bool(reported) and slots.slot_count <= max(reported) + 1
return value, _Unread(slots, explained)
if reported and slots.slot_count <= max(reported) + 1:
return value, None
return value, slots
@staticmethod
def _stamped(document: DocumentDefinition[Any], envelope: CultCacheEnvelope) -> CultCacheEnvelope:
@ -568,7 +561,7 @@ class CultCache:
records: list[tuple[CultCacheEnvelope, Any]],
*,
batch: bool,
unread: dict[str, _Unread | None] | None = None,
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."""
@ -608,11 +601,9 @@ class CultCache:
self._remove_lookups(type, envelope.key)
values[envelope.key] = value
envelopes[envelope.key] = envelope
mark = (unread or {}).get(envelope.key)
if mark is None:
self._state.unread.pop((type, envelope.key), None)
else:
self._state.unread[(type, envelope.key)] = mark
# 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
@ -641,24 +632,28 @@ class CultCache:
@_locked
def is_writable(self, document: DocumentDefinition[Any], key: str) -> bool:
"""False when the cache holds the record at the key read-only: its payload carried slots past the declared ones, at any depth,
and nothing explains the loss, so a write or removal would drop data a newer writer stored. Advisory (the write still
decides): it answers from the held record at call time, and a key the cache does not hold is writable."""
"""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)
return self._unread_at(key) is None
def _unread_at(self, key: str) -> tuple[str, _Unread] | None:
"""The unexplained mark of the record held at the key, whatever its type: a key holds one record."""
return next(((type, unread) for (type, held), unread in self._state.unread.items() if held == key and not unread.explained), None)
entry = document.catalog_entry()
probe = CultCacheEnvelope.create(key=key, type=document.type, payload=b"", schema_id=entry.schema_id, catalog_entry=entry)
try:
self._refuse_unread([key])
self._refuse_retype(document, [probe])
except SchemaConflictError:
return False
return True
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._unread_at(key)
held = self._state.unread.get(key)
if held is None:
continue
type, unread = held
slots = unread.slots
type, slots = held
raise SchemaConflictError(
self._state.envelopes[type][key].schema_id or type,
[self._state.documents[type].catalog_entry().schema_name],
@ -740,7 +735,6 @@ class CultCache:
def _drop(self, type: str, key: str) -> None:
"""Removes the held record and its lookups from the cache's maps; the store is the caller's."""
self._remove_lookups(type, key)
self._state.unread.pop((type, key), None)
for by_type in (self._state.values, self._state.envelopes, self._state.lookup_keys):
held = by_type.get(type)
if held is not None:

89
packages/cultcache-py/tests/test_slot_readonly.py
View file

@ -1,6 +1,8 @@
from __future__ import annotations
import dataclasses
import shutil
import threading
import tempfile
import unittest
from pathlib import Path
@ -10,12 +12,14 @@ import msgpack
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
@ -130,6 +134,32 @@ class PayloadMarkTests(unittest.TestCase):
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:
@ -251,6 +281,65 @@ class StoreRowTests(unittest.TestCase):
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_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()}