diff --git a/docs/start/README.md b/docs/start/README.md index e52f7d73..6a7d6d74 100644 --- a/docs/start/README.md +++ b/docs/start/README.md @@ -167,8 +167,8 @@ fn first_call() -> anyhow::Result<()> { ## Python first call -A document is a definition with an encoder and decoder; `name=` and `indexes=` register the name lookup -and indexes. The source is +A document is a definition with an encoder and decoder; `name=` registers the name lookup, `indexes=` unique lookups and +`roles=` many-row selection aliases. The source is [`packages/cultcache-py/tests/test_first_call.py`](../../packages/cultcache-py/tests/test_first_call.py), run by `python -m unittest tests.test_first_call` in `packages/cultcache-py` with `src` on `PYTHONPATH` and `msgpack` installed (the store needs it). A second test fails if the block below stops matching the @@ -194,7 +194,6 @@ item_doc = define_document_type( encode=lambda item: asdict(item), decode=lambda raw: Item(**raw), name="name", - indexes={"category": "category"}, ) diff --git a/packages/cultcache-py/CHANGELOG.md b/packages/cultcache-py/CHANGELOG.md index b460ab05..641cc0ed 100644 --- a/packages/cultcache-py/CHANGELOG.md +++ b/packages/cultcache-py/CHANGELOG.md @@ -9,6 +9,8 @@ ### Breaking +- `indexes` are unique, as C# `[CultIndex]`: a write, load or `register_index` that leaves two records of a type holding one non-blank value under one alias is refused (`CultCacheError`) naming both keys and the alias, and writes nothing. Index text is the C# canonical text (`True`/`False`, integers as digits, floats as the exact expansion, blank values never indexed), so an empty or whitespace string is no longer a value. A new `roles` map names the many-row selection aliases (never indexed for lookup, same catalog `index_alias`). Classify every `indexes` key per the README before updating. The verdict is the contract, the text is informational. + - A single-file store written as `cultcache.store.v3` is read and written: a record that holds an element id (an owned list element with an `[CultElementId]` member, carried through every write) marks the store v3, and a v3 store is carried through a rewrite with its ids. A runtime from before this change refuses a v3 store by name. - A write over a file the reader refuses (`StoreUnreadableError`, from `push`, `push_all` and `delete`) refuses and leaves the file as it is. A flush used to replace such a file. A write set cannot merge with a disk it cannot read, so the file is repaired or removed by its owner before the store writes again. - Added `StoreUnreadableError` (a `ValueError`) for a store file this runtime cannot read, and `SchemaConflictError` (a `ValueError`) for a diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 41c3095c..ab586ff6 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -73,7 +73,7 @@ store. - `register_document_type(...)` - `register_registry(...)` - `register_name_lookup(...)` -- `register_index(...)` +- `register_index(...)` (a unique index) - `add_backing_store(...)` - `add_generic_store(...)` - `pull_all_backing_stores()` @@ -95,6 +95,33 @@ store. - `delete_global(...)` - `snapshot()` +## Indexes And Roles + +`indexes=` are unique lookups (variants:ruling:unique-indexes; C# `[CultIndex]`): within one document type and one alias +no two records hold one non-blank value, judged on the result of every write (a batch is judged whole, so a swap inside +one `put_envelopes` lands) and of every load and `register_index`. A refusal is a `CultCacheError`, +`Records and () both hold unique index ''; an index value names one record.`, keys in +ordinal order, never the value; it writes nothing, and a refused load or registration installs nothing. The contract is +the verdict; the text is informational. Blank values (absent, empty, or only C# `White_Space` characters) are never +indexed and never conflict. Two types may share an alias and a value. Values compare as the C# canonical text: strings as +is (never trimmed), `True`/`False`, integers as digits, a non-integral float as its exact decimal expansion, NaN and +infinities no value. + +`roles=` name selection aliases (C# `[CultRole]`): many records may share a value, nothing is indexed or refused, +`get_by_index` on a role answers `None`, and the member's catalog `index_alias` is the same either way, so catalog bytes +do not move. An alias is declared in one of the two. `CultMeshDatabase.watch_by_index` reads `indexes` only. + +Migrating an existing `indexes` key, per key, before bumping: + +1. Nothing looks records up by the key (`get_by_index`, `get_key_by_index`, `node.get_by_index`, `watch_by_index`): move it to `roles`. +2. Something looks it up: it stays unique only when two live records can never hold one non-blank value (the record key + is derived from that value alone, or it is an outside identity: email, sku, external id). Enums, statuses, kinds, + categories, parent or foreign ids, shared hashes and timestamps are roles; select them with `get_all` plus a filter. +3. Open each real `.cc` once with the new definitions. A refusal names both keys and the alias and writes nothing. + +Misclassifying: a role declared unique makes a store that holds a shared value refuse to open until the key is +reclassified; a unique key declared a role answers no lookups and stops refusing duplicates. + ## Persistence Model Backing stores persist envelopes, not caller domain objects. The cache decodes payloads through registered document definitions and rejects unknown type discriminators. This keeps Python's dynamic runtime from turning persistence into an open polymorphic sewer with a cheerful docstring. diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 8f4ac759..c934c33b 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -1,9 +1,11 @@ from __future__ import annotations import functools +import math import threading +from decimal import Decimal from dataclasses import dataclass, field, replace -from typing import Any, Callable, Generic, TypeVar +from typing import Any, Callable, Generic, Iterable, TypeVar from .backing_store import BackingStore, CultCacheEnvelope from .documents import DocumentDefinition, extract_value @@ -52,6 +54,52 @@ class CultCacheError(RuntimeError): pass +# Blank is no value: empty or only C# White_Space characters (the 25 chars string.IsNullOrWhiteSpace is true for). +# An explicit set, not str.isspace(): Python also counts U+001C-001F, which C# does not, and omits nothing C# counts. +_BLANK_INDEX_CHARS = frozenset( + chr(code) + for code in ( + *range(0x09, 0x0E), + 0x20, + 0x85, + 0xA0, + 0x1680, + *range(0x2000, 0x200B), + 0x2028, + 0x2029, + 0x202F, + 0x205F, + 0x3000, + ) +) + + +def _utf16_order(text: str) -> bytes: + """Sorts as C# StringComparer.Ordinal does: by UTF-16 code unit, not code point.""" + return text.encode("utf-16-be", "surrogatepass") + + +def index_text(value: Any) -> str | None: + """The canonical text of an index value, as C# IndexValue and RenderCanonicalNumber, or None when it names no + value: a string is as is (never trimmed), a boolean is `True`/`False`, an integer is its digits, a non-integral + finite float is its exact decimal expansion, NaN and infinities and blank strings are no value.""" + if value is None: + return None + if isinstance(value, bool): # before int: bool is an int subclass + return "True" if value else "False" + if isinstance(value, int): + return str(value) + if isinstance(value, float): + if not math.isfinite(value): + return None + if value.is_integer(): + return str(int(value)) # -0.0 is 0 + text = format(Decimal(value), "f") # Decimal(float) is exact + return text.rstrip("0").rstrip(".") if "." in text else text + text = value if isinstance(value, str) else str(value) + return None if all(char in _BLANK_INDEX_CHARS for char in text) else text + + @dataclass class _State: documents: dict[str, DocumentDefinition[Any]] = field(default_factory=dict) @@ -432,12 +480,20 @@ class CultCache: """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.""" type = document.type + # 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()) name_extractor = self._state.name_extractors.get(type) index_extractors = self._state.index_extractors.get(type, {}) admitted = [ (envelope, value, self._lookup_keys_for(value, name_extractor, index_extractors)) for envelope, value in records ] + self._refuse_index_sharing( + [(type, envelope.key, keys) for envelope, _, keys in admitted], + self._state.indexes, + {(type, envelope.key) for envelope, _ in records}, + ) store = self._store_for_type(type) legacy_key = self._state.legacy_global_keys.get(type) if store is not None: @@ -552,9 +608,9 @@ class CultCache: name = None if name_extractor is None else extract_value(value, name_extractor) indexes: dict[str, str] = {} for index, extractor in index_extractors.items(): - index_value = extract_value(value, extractor) + index_value = index_text(extract_value(value, extractor)) if index_value is not None: - indexes[index] = str(index_value) + indexes[index] = index_value return _LookupKeys(None if name is None else str(name), indexes) def _derive_lookups( @@ -571,17 +627,52 @@ class CultCache: name_extractor = name_extractors.get(type) type_index_extractors = index_extractors.get(type, {}) for key, value in values.items(): - keys = self._lookup_keys_for(value, name_extractor, type_index_extractors) - lookup_keys.setdefault(type, {})[key] = keys + lookup_keys.setdefault(type, {})[key] = self._lookup_keys_for(value, name_extractor, type_index_extractors) + # Judged whole, over every type at once and on the file alone (never against what the cache held): a held value can + # have been mutated in place since it was admitted, and another process may have moved a value between records. + self._refuse_index_sharing( + [(type, key, keys) for type, by_key in lookup_keys.items() for key, keys in by_key.items()], {}, () + ) + for type in lookup_keys: + for key, keys in lookup_keys.get(type, {}).items(): if keys.name is not None: names[(type, keys.name)] = key for index, index_value in keys.indexes.items(): indexes[(type, index, index_value)] = key return names, indexes, lookup_keys + def _refuse_index_sharing( + self, + landing: list[tuple[str, str, _LookupKeys]], + held: dict[tuple[str, str, str], str], + replaced: Iterable[tuple[str, str]], + ) -> None: + """The one index rule (variants:ruling:unique-indexes), in the order of C# RefuseIndexSharing: within one type and one + unique alias, no two records hold one non-blank value, judged on the result. The holders of a value are the landing + keys projecting it (in landing order) plus the held key unless this admission replaces it; two distinct holders + refuse, naming the two smallest keys (UTF-16 code-unit order) and the alias, never the value. Aliases are judged in + UTF-16 order, values in the order they landed.""" + released = set(replaced) + landed: dict[tuple[str, str, str], list[str]] = {} + for type, key, keys in landing: + for index, index_value in keys.indexes.items(): + landed.setdefault((type, index, index_value), []).append(key) + for (type, index, index_value), keys in sorted(landed.items(), key=lambda item: _utf16_order(item[0][1])): + held_key = held.get((type, index, index_value)) + if held_key is not None and (type, held_key) not in released: + keys = [*keys, held_key] + distinct = sorted(dict.fromkeys(keys), key=_utf16_order) + if len(distinct) > 1: + document = self._state.documents.get(type) + schema_name = (document.schema_name if document is not None else None) or type + raise CultCacheError( + f"Records {distinct[0]} and {distinct[1]} ({schema_name}) both hold unique index '{index}'; " + "an index value names one record." + ) + def _remove_lookups(self, type: str, key: str) -> None: - """Removes the lookups the held record produced at admission, handing a name or index value to - another held record that produced it too. Reads stored lookup keys only; runs no extractor.""" + """Removes the lookups the held record produced at admission, handing a name to another held + record that produced it too (an index value has one holder). Reads stored lookup keys only; runs no extractor.""" held = self._state.lookup_keys.get(type, {}) keys = held.get(key) if keys is None: @@ -596,9 +687,6 @@ class CultCache: if self._state.indexes.get(lookup_key) != key: continue del self._state.indexes[lookup_key] - for candidate_key, candidate in held.items(): - if candidate_key != key and candidate.indexes.get(index) == index_value: - self._state.indexes[lookup_key] = candidate_key def _assert_registered(self, document: DocumentDefinition[Any]) -> None: if self._state.documents.get(document.type) is not document: diff --git a/packages/cultcache-py/src/cultcache_py/documents.py b/packages/cultcache-py/src/cultcache_py/documents.py index 23849713..6101c136 100644 --- a/packages/cultcache-py/src/cultcache_py/documents.py +++ b/packages/cultcache-py/src/cultcache_py/documents.py @@ -157,6 +157,11 @@ class DocumentDefinition(Generic[T]): global_document: bool = False name: str | Extractor[T] | None = None indexes: dict[str, str | Extractor[T]] = field(default_factory=dict) + """Unique lookups (C# `[CultIndex]`): no two records of the type hold one non-blank value, judged at every commit and + load. `get_by_index` answers the one holder.""" + roles: dict[str, str | Extractor[T]] = field(default_factory=dict) + """Selection aliases (C# `[CultRole]`): many records may share a value. Never indexed for lookup, never refuse; the + member's catalog `index_alias` is the same either way.""" payload_encoder: Callable[[Any], bytes] = _json_payload_encoder payload_decoder: Callable[[bytes], Any] = _json_payload_decoder schema_id: str | None = None @@ -190,6 +195,14 @@ class DocumentDefinition(Generic[T]): self.validate(value) return value + def __post_init__(self) -> None: + # Judged here so every constructor passes it: the factories, the dataclass constructor and dataclasses.replace. + for alias in self.roles: + if alias in self.indexes: + raise ValueError( + f"Document type {self.type} declares '{alias}' as both an index and a role; an index already names its role." + ) + def catalog_entry(self) -> CultCacheSchemaCatalogEntry: if self._catalog_entry is not None: return self._catalog_entry @@ -241,6 +254,7 @@ def define_document_type( global_document: bool = False, name: str | Extractor[T] | None = None, indexes: dict[str, str | Extractor[T]] | None = None, + roles: dict[str, str | Extractor[T]] | None = None, payload_encoder: Callable[[Any], bytes] = _json_payload_encoder, payload_decoder: Callable[[bytes], Any] = _json_payload_decoder, schema_id: str | None = None, @@ -259,6 +273,7 @@ def define_document_type( global_document=global_document, name=name, indexes=indexes or {}, + roles=roles or {}, payload_encoder=payload_encoder, payload_decoder=payload_decoder, schema_id=schema_id, @@ -280,6 +295,7 @@ def define_database_entry_type( global_document: bool = False, name: str | Extractor[T] | None = None, indexes: dict[str, str | Extractor[T]] | None = None, + roles: dict[str, str | Extractor[T]] | None = None, schema_id: str | None = None, schema_name: str | None = None, schema_version: str | None = None, @@ -296,7 +312,7 @@ def define_database_entry_type( normalized = _normalize_fields(type, fields) member_name_lookup = name if isinstance(name, str) else None - index_lookup = indexes or {} + index_lookup = {**(roles or {}), **(indexes or {})} def members() -> tuple[CultCacheSchemaCatalogMember, ...]: return tuple( @@ -323,6 +339,7 @@ def define_database_entry_type( global_document=global_document, name=name, indexes=indexes, + roles=roles, payload_encoder=_pack_tree, payload_decoder=_messagepack_payload_decoder, schema_id=schema_id, diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 1dd41a0e..92892b51 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -103,20 +103,20 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(cache.get_key_by_name(document, "Elixir"), "item:potion") self.assertEqual(cache.get_by_index(document, "category", "Rare").name, "Elixir") - def test_incremental_lookup_delete_restores_duplicate_owner(self) -> None: + def test_incremental_lookup_delete_restores_duplicate_name_owner(self) -> None: document = item_doc() cache = CultCache() cache.register_document_type(document) cache.put(document, "item:first", Item(name="Potion", category="Consumable", value=50)) - cache.put(document, "item:second", Item(name="Potion", category="Consumable", value=60)) + cache.put(document, "item:second", Item(name="Potion", category="Rare", value=60)) self.assertEqual(cache.get_key_by_name(document, "Potion"), "item:second") - self.assertEqual(cache.get_key_by_index(document, "category", "Consumable"), "item:second") cache.delete(document, "item:second") self.assertEqual(cache.get_key_by_name(document, "Potion"), "item:first") self.assertEqual(cache.get_key_by_index(document, "category", "Consumable"), "item:first") + self.assertIsNone(cache.get_key_by_index(document, "category", "Rare")) def test_database_entry_formatter_uses_slot_indexed_messagepack_array(self) -> None: try: diff --git a/packages/cultcache-py/tests/test_first_call.py b/packages/cultcache-py/tests/test_first_call.py index f8d4b52b..10d6eb27 100644 --- a/packages/cultcache-py/tests/test_first_call.py +++ b/packages/cultcache-py/tests/test_first_call.py @@ -24,7 +24,6 @@ item_doc = define_document_type( encode=lambda item: asdict(item), decode=lambda raw: Item(**raw), name="name", - indexes={"category": "category"}, ) diff --git a/packages/cultcache-py/tests/test_unique_index.py b/packages/cultcache-py/tests/test_unique_index.py new file mode 100644 index 00000000..0dca2aa8 --- /dev/null +++ b/packages/cultcache-py/tests/test_unique_index.py @@ -0,0 +1,575 @@ +from __future__ import annotations + +import itertools +import shutil +import tempfile +import unittest +from dataclasses import asdict, dataclass +from decimal import Decimal +from pathlib import Path + +import msgpack + +from cultcache_py import ( + CultCache, + CultCacheEnvelope, + SingleFileMessagePackBackingStore, + define_database_entry_type, + define_document_type, + leaves as t, + database_entry_field, +) +from cultcache_py.cache import CultCacheError, index_text + +# variants:ruling:unique-indexes. `indexes` are unique within one type, `roles` are selection aliases. The vectors in +# tests/vectors/unique-index are written by the C# reference; the rest of the behaviour is pinned here. +VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "unique-index" + + +@dataclass +class Item: + sku: object = None + serial: object = None + group: object = None + + +def refusal(key_a: str, key_b: str, schema_name: str, alias: str) -> str: + return f"Records {key_a} and {key_b} ({schema_name}) both hold unique index '{alias}'; an index value names one record." + + +def item_doc(type: str = "item", *, indexes=None, roles=None): + return define_document_type( + type, + encode=asdict, + decode=lambda raw: Item(**raw), + indexes={"sku": "sku", "serial": "serial"} if indexes is None else indexes, + roles={"group": "group"} if roles is None else roles, + ) + + +def envelope(document, key: str, item: Item) -> CultCacheEnvelope: + return CultCacheEnvelope.create(key=key, type=document.type, payload=document.encode_payload(item)) + + +def cache_of(*documents, store=None) -> CultCache: + builder = CultCache.builder() + for document in documents: + builder.register_document_type(document) + if store is not None: + builder.add_generic_store(store) + return builder.build() + + +class RecordingStore: + """A store that counts what reaches it; the judge must refuse before any store call.""" + + def __init__(self) -> None: + self.pushed: list[str] = [] + + def pull_all(self): + return [] + + def push(self, envelope) -> None: + self.pushed.append(envelope.key) + + def push_all(self, envelopes) -> None: + self.pushed.extend(envelope.key for envelope in envelopes) + + def delete(self, type, key) -> None: + pass + + +class IndexTextTests(unittest.TestCase): + def test_canonical_text(self) -> None: + cases = [ + ("S-1", "S-1"), + (" a ", " a "), + ("", ""), + ("\x1f", "\x1f"), + ("\x1c", "\x1c"), + ("\x1d", "\x1d"), + ("\x1e", "\x1e"), + ("᠎", "᠎"), + ("​", "​"), + (True, "True"), + (False, "False"), + (7, "7"), + (-7, "-7"), + (9223372036854775807, "9223372036854775807"), + (18446744073709551615, "18446744073709551615"), + (0.0, "0"), + (-0.0, "0"), + (2.0, "2"), + (-3.0, "-3"), + (1e21, "1000000000000000000000"), + (0.1, "0.1000000000000000055511151231257827021181583404541015625"), + (-0.5, "-0.5"), + (2.5e-07, "0.00000024999999999999998868702795647156467140348468092270195484161376953125"), + ] + for value, text in cases: + with self.subTest(value=value): + self.assertEqual(index_text(value), text) + self.assertIsInstance(index_text(value), str) + + def test_no_value(self) -> None: + blank = ["", " ", "\t", "\n", "\x0b", "\x0c", "\r", "\x85", "\xa0", " ", "
", "
", " ", + " ", "　", " 　\x85 ", *map(chr, range(0x2000, 0x200B))] + for value in [None, float("nan"), float("inf"), float("-inf"), *blank]: + with self.subTest(value=repr(value)): + self.assertIsNone(index_text(value)) + + def test_blank_set_is_exactly_the_csharp_set(self) -> None: + csharp = {*range(0x09, 0x0E), 0x20, 0x85, 0xA0, 0x1680, *range(0x2000, 0x200B), 0x2028, 0x2029, 0x202F, 0x205F, 0x3000} + self.assertEqual(len(csharp), 25) + got = {code for code in range(0x110000) if 0xD800 > code or code > 0xDFFF if index_text(chr(code)) is None} + self.assertEqual(got, csharp) + # Python's isspace is the wrong test: it also counts U+001C-001F. + self.assertTrue("\x1f".isspace()) + self.assertEqual(index_text("\x1f"), "\x1f") + + def test_floats_render_exactly_not_through_str(self) -> None: + for value in (0.1, 1e-7, 123456789.123456789, 5e-324, 1.7976931348623157e308): + with self.subTest(value=value): + text = index_text(value) + self.assertNotIn("e", text.lower()) + self.assertEqual(Decimal(text), Decimal(value)) + self.assertFalse(text.endswith("0") and "." in text) + + def test_bool_is_not_an_integer(self) -> None: + self.assertEqual(index_text(True), "True") + self.assertNotEqual(index_text(True), "1") + + +class UniqueIndexTests(unittest.TestCase): + def test_a_write_of_a_held_value_at_another_key_is_refused_by_every_write_path(self) -> None: + for alias in ("sku", "serial"): # two unique indexes: each is judged + other = "serial" if alias == "sku" else "sku" + paths = { + "put": lambda cache, d, k, i: cache.put(d, k, i), + "put_envelope": lambda cache, d, k, i: cache.put_envelope(d, envelope(d, k, i)), + "put_envelopes": lambda cache, d, k, i: cache.put_envelopes(d, [envelope(d, k, i)]), + "update": lambda cache, d, k, i: cache.update(d, k, lambda _held: i), + } + for name, write in paths.items(): + with self.subTest(alias=alias, path=name): + document = item_doc() + store = RecordingStore() + cache = cache_of(document, store=store) + cache.put(document, "item:b", Item(sku="CANARY-b", serial="canary-b")) + cache.put(document, "item:a", Item(sku="CANARY-a", serial="canary-a")) + before = cache.snapshot() + pushed = list(store.pushed) + taken = "CANARY-b" if alias == "sku" else "canary-b" + landing = Item(**{alias: taken, other: "fresh"}) + with self.assertRaises(CultCacheError) as caught: + write(cache, document, "item:a", landing) + self.assertEqual(str(caught.exception), refusal("item:a", "item:b", "item", alias)) + self.assertNotIn("anary", str(caught.exception)) + self.assertEqual(cache.snapshot(), before) + self.assertEqual(store.pushed, pushed) + self.assertEqual(cache.get_key_by_index(document, alias, taken), "item:b") + self.assertIsNone(cache.get_key_by_index(document, other, "fresh")) + + def test_a_record_rewriting_its_own_key_keeps_its_value(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:a", Item(sku="S", serial="1", group="g")) + cache.put(document, "item:a", Item(sku="S", serial="2", group="h")) + self.assertEqual(cache.get_key_by_index(document, "sku", "S"), "item:a") + self.assertIsNone(cache.get_key_by_index(document, "serial", "1")) + self.assertEqual(cache.get_key_by_index(document, "serial", "2"), "item:a") + + def test_the_refusal_names_the_schema_not_the_type(self) -> None: + document = define_document_type( + "t.item", indexes={"sku": "sku"}, schema_name="schema.Item", schema_id="schema.Item.v1" + ) + cache = cache_of(document) + cache.put(document, "k:1", {"sku": "S"}) + with self.assertRaises(CultCacheError) as caught: + cache.put(document, "k:2", {"sku": "S"}) + self.assertEqual(str(caught.exception), refusal("k:1", "k:2", "schema.Item", "sku")) + + def test_the_two_smallest_keys_are_named_in_ordinal_order(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:m", Item(sku="S")) + with self.assertRaises(CultCacheError) as caught: + cache.put_envelopes(document, [envelope(document, "item:z", Item(sku="S")), envelope(document, "item:A", Item(sku="S"))]) + self.assertEqual(str(caught.exception), refusal("item:A", "item:m", "item", "sku")) + + def test_put_envelopes_holding_one_value_twice_is_refused_whole(self) -> None: + document = item_doc() + store = RecordingStore() + cache = cache_of(document, store=store) + with self.assertRaises(CultCacheError) as caught: + cache.put_envelopes( + document, + [ + envelope(document, "item:b", Item(sku="S")), + envelope(document, "item:c", Item(sku="T")), + envelope(document, "item:a", Item(sku="S")), + ], + ) + self.assertEqual(str(caught.exception), refusal("item:a", "item:b", "item", "sku")) + self.assertEqual(cache.snapshot(), {}) + self.assertEqual(store.pushed, []) + self.assertIsNone(cache.get_key_by_index(document, "sku", "T")) + + def test_verdict_does_not_depend_on_batch_order(self) -> None: + document = item_doc() + rows = [("item:a", Item(sku="S", serial="1")), ("item:b", Item(sku="T", serial="1")), ("item:c", Item(sku="U"))] + for order in itertools.permutations(rows): + with self.subTest(order=[key for key, _ in order]): + cache = cache_of(document) + with self.assertRaises(CultCacheError) as caught: + cache.put_envelopes(document, [envelope(document, key, item) for key, item in order]) + self.assertEqual(str(caught.exception), refusal("item:a", "item:b", "item", "serial")) + self.assertEqual(cache.snapshot(), {}) + + def test_put_envelopes_swapping_two_records_values_lands(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:a", Item(sku="S", serial="1")) + cache.put(document, "item:b", Item(sku="T", serial="2")) + cache.put_envelopes( + document, + [envelope(document, "item:a", Item(sku="T", serial="2")), envelope(document, "item:b", Item(sku="S", serial="1"))], + ) + self.assertEqual(cache.get_key_by_index(document, "sku", "S"), "item:b") + self.assertEqual(cache.get_key_by_index(document, "sku", "T"), "item:a") + self.assertEqual(cache.get_key_by_index(document, "serial", "1"), "item:b") + self.assertEqual(cache.get_key_by_index(document, "serial", "2"), "item:a") + + def test_a_batch_may_take_a_value_the_batch_releases(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:a", Item(sku="S")) + cache.put_envelopes(document, [envelope(document, "item:a", Item(sku="T")), envelope(document, "item:b", Item(sku="S"))]) + self.assertEqual(cache.get_key_by_index(document, "sku", "S"), "item:b") + self.assertEqual(cache.get_key_by_index(document, "sku", "T"), "item:a") + + def test_delete_frees_the_value(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:a", Item(sku="S")) + cache.delete(document, "item:a") + self.assertIsNone(cache.get_key_by_index(document, "sku", "S")) + cache.put(document, "item:b", Item(sku="S")) + self.assertEqual(cache.get_key_by_index(document, "sku", "S"), "item:b") + + def test_blank_values_never_conflict(self) -> None: + document = item_doc() + cache = cache_of(document) + blanks = ["", " ", "\x85", "　", None] + for number, blank in enumerate(blanks): + cache.put(document, f"item:blank{number}", Item(sku=blank, serial=blank)) + cache.put_envelopes(document, [envelope(document, "item:more", Item(sku=" ")), envelope(document, "item:more2", Item(sku=""))]) + for blank in ("", " ", "\x85"): + self.assertIsNone(cache.get_key_by_index(document, "sku", blank)) + # U+001F and U+FEFF are values. + cache.put(document, "item:us", Item(sku="\x1f")) + cache.put(document, "item:bom", Item(sku="")) + self.assertEqual(cache.get_key_by_index(document, "sku", "\x1f"), "item:us") + self.assertEqual(cache.get_key_by_index(document, "sku", ""), "item:bom") + with self.assertRaises(CultCacheError): + cache.put(document, "item:us2", Item(sku="\x1f")) + + def test_values_compare_as_the_canonical_text(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:a", Item(sku=1)) + for same in (1, 1.0, "1"): + with self.subTest(same=same), self.assertRaises(CultCacheError): + cache.put(document, "item:b", Item(sku=same)) + cache.put(document, "item:true", Item(sku=True)) + self.assertEqual(cache.get_key_by_index(document, "sku", "True"), "item:true") + self.assertEqual(cache.get_key_by_index(document, "sku", "1"), "item:a") + cache.put(document, "item:f", Item(sku=0.1)) + self.assertEqual(cache.get_key_by_index(document, "sku", "0.1000000000000000055511151231257827021181583404541015625"), "item:f") + + def test_two_types_may_share_an_alias_and_a_value(self) -> None: + one, two = item_doc("one"), item_doc("two") + cache = cache_of(one, two) + cache.put(one, "k:1", Item(sku="S", serial="1")) + cache.put(two, "k:2", Item(sku="S", serial="1")) + self.assertEqual(cache.get_key_by_index(one, "sku", "S"), "k:1") + self.assertEqual(cache.get_key_by_index(two, "sku", "S"), "k:2") + cache.delete(one, "k:1") + self.assertEqual(cache.get_key_by_index(two, "sku", "S"), "k:2") + + def test_roles_may_share_values_and_are_not_looked_up(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:a", Item(sku="1", group="g")) + cache.put(document, "item:b", Item(sku="2", group="g")) + self.assertIsNone(cache.get_key_by_index(document, "group", "g")) + self.assertIsNone(cache.get_by_index(document, "group", "g")) + + def test_an_alias_in_both_indexes_and_roles_is_refused(self) -> None: + with self.assertRaisesRegex(ValueError, "both an index and a role"): + define_document_type("bad", indexes={"a": "a"}, roles={"a": "a"}) + with self.assertRaisesRegex(ValueError, "both an index and a role"): + define_database_entry_type("bad", [("a", 0)], indexes={"a": "a"}, roles={"a": "a"}) + define_document_type("fine", indexes={"a": "a"}, roles={"b": "a"}) + + def test_register_index_over_held_duplicates_installs_nothing(self) -> None: + document = item_doc(indexes={}, roles={}) + cache = cache_of(document) + cache.put(document, "item:b", Item(sku="S", serial="1")) + cache.put(document, "item:a", Item(sku="S", serial="2")) + with self.assertRaises(CultCacheError) as caught: + cache.register_index(document, "sku", "sku") + self.assertEqual(str(caught.exception), refusal("item:a", "item:b", "item", "sku")) + self.assertIsNone(cache.get_key_by_index(document, "sku", "S")) + cache.register_index(document, "serial", "serial") # an installed alias is not poisoned by the refused one + self.assertEqual(cache.get_key_by_index(document, "serial", "2"), "item:a") + # Nothing was installed, so a write sharing the refused alias's value is still admitted. + cache.put(document, "item:c", Item(sku="S", serial="3")) + + def test_registration_names_an_installed_alias_a_caller_mutated_into_a_collision(self) -> None: + document = item_doc(indexes={"sku": "sku"}, roles={}) + cache = cache_of(document) + cache.put(document, "item:x", Item(sku="1", serial="s1")) + cache.put(document, "item:y", Item(sku="2", serial="s2")) + cache.get_required(document, "item:y").sku = "1" # the cache hands out its held value + with self.assertRaises(CultCacheError) as caught: + cache.register_index(document, "serial", "serial") + self.assertEqual(str(caught.exception), refusal("item:x", "item:y", "item", "sku")) + + def test_derive_names_the_ordinal_first_alias_then_value(self) -> None: + document = item_doc(indexes={}, roles={}) + cache = cache_of(document) + cache.put(document, "item:1", Item(sku="a", serial="p", group="x")) + cache.put(document, "item:2", Item(sku="a", serial="p", group="x")) + cache.put(document, "item:3", Item(sku="b", serial="q", group="y")) + cache.put(document, "item:4", Item(sku="b", serial="q", group="y")) + with self.assertRaises(CultCacheError) as caught: + cache.register_index(document, "serial", "serial") + self.assertEqual(str(caught.exception), refusal("item:1", "item:2", "item", "serial")) + + def test_a_document_in_two_unique_indexes_loads_and_registers_through_every_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "items.cc" + document = item_doc() + cache = cache_of(document, store=SingleFileMessagePackBackingStore(path)) + cache.put(document, "item:a", Item(sku="S", serial="1")) + cache.put(document, "item:b", Item(sku="T", serial="2")) + # a store written by a looser definition, then opened by this one + loose = item_doc(indexes={}, roles={}) + writer = cache_of(loose, store=SingleFileMessagePackBackingStore(path)) + writer.pull_all_backing_stores() + writer.put(loose, "item:c", Item(sku="U", serial="2")) + before = path.read_bytes() + reader = cache_of(document, store=SingleFileMessagePackBackingStore(path)) + with self.assertRaises(CultCacheError) as caught: + reader.pull_all_backing_stores() + self.assertEqual(str(caught.exception), refusal("item:b", "item:c", "item", "serial")) + self.assertEqual(path.read_bytes(), before) + self.assertEqual(reader.snapshot(), {}) + + def test_a_refused_load_keeps_the_held_state(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "items.cc" + loose = item_doc(indexes={}, roles={}) + writer = cache_of(loose, store=SingleFileMessagePackBackingStore(path)) + writer.put(loose, "item:a", Item(sku="S")) + writer.put(loose, "item:b", Item(sku="S")) + document = item_doc() + cache = cache_of(document, store=SingleFileMessagePackBackingStore(path)) + with self.assertRaises(CultCacheError): + cache.pull_all_backing_stores() + self.assertIsNone(cache.get_key_by_index(document, "sku", "S")) + self.assertEqual(cache.get_all(document), []) + + +class SimulationFactTests(unittest.TestCase): + def test_simulation_fact_catalog_entry_is_unchanged_by_indexes_to_roles(self) -> None: + import re + + from cultnet_py.cultmesh_contracts import SIMULATION_FACT_DOCUMENT_TYPE, simulation_fact_document as document + + aliases = {"shard": "shard_id", "frame": "frame", "subject": "subject_id", "claim_kind": "claim_kind"} + self.assertEqual(document.roles, aliases) + self.assertEqual(document.indexes, {}) + entry = document.catalog_entry() + snake = lambda name: re.sub(r"(? None: + from cultnet_py.cultmesh_contracts import CultMeshSimulationFact, simulation_fact_document as document + + cache = cache_of(document) + for number in range(3): + fact = CultMeshSimulationFact.from_candidate( + {"shardId": "s", "shardEpoch": 1, "frame": 4, "subjectId": "x", "claimKind": "k", "claimHash": f"h{number}", + "witnessCount": 1, "supportWeight": 1.0, "totalWeight": 1.0, "confidence": 1.0} + ) + cache.put(document, f"simulation:{number}", fact) + self.assertEqual(len(cache.get_all(document)), 3) + self.assertIsNone(cache.get_key_by_index(document, "shard", "s")) + + +class VectorTests(unittest.TestCase): + def test_values_rows(self) -> None: + document = item_doc(indexes={"sku": "sku"}, roles={}) + for label, kind, value, text in msgpack.unpackb((VECTORS / "values.msgpack").read_bytes(), raw=False): + with self.subTest(label=label): + self.assertEqual(index_text(value), text) + cache = cache_of(document) + cache.put(document, "item:1", Item(sku=value)) + if text is None: + cache.put(document, "item:2", Item(sku=value)) # blank: two records commit + self.assertEqual(len(cache.get_all(document)), 2) + else: + self.assertEqual(cache.get_key_by_index(document, "sku", text), "item:1") + with self.assertRaises(CultCacheError): + cache.put(document, "item:2", Item(sku=value)) + + def test_store_verdicts(self) -> None: + unit = define_database_entry_type + outcomes = msgpack.unpackb((VECTORS / "outcomes.msgpack").read_bytes(), raw=False) + self.assertEqual(len(outcomes), 3) + for name, alias, verdict, key_a, key_b in outcomes: + for roles_first in (False, True): # the verdict does not depend on declaration order + with self.subTest(file=name, alias=alias, roles_first=roles_first), tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / name + shutil.copyfile(VECTORS / name, path) + before = path.read_bytes() + fields = [database_entry_field("sku", 0, codec=t.string), database_entry_field("category", 1, codec=t.string)] + declared = {"indexes": {alias: "sku"} if alias else {}, "roles": {} if alias else {"Sku": "sku"}} + if roles_first: + declared = dict(reversed(list(declared.items()))) + document = unit( + "tests.unique_item", + fields, + schema_id="sha256:93bb890de83069061af696b23f94fd32fb3c2dad3c91ee645e8add56085f1882", + schema_name="tests.unique_item", + **declared, + ) + cache = cache_of(document, store=SingleFileMessagePackBackingStore(path)) + if verdict == "open": + cache.pull_all_backing_stores() + self.assertGreaterEqual(len(cache.get_all(document)), 2) + else: + with self.assertRaises(CultCacheError) as caught: + cache.pull_all_backing_stores() + self.assertEqual(str(caught.exception), refusal(key_a, key_b, "tests.unique_item", alias)) + self.assertNotIn("S-1", str(caught.exception)) + self.assertEqual(cache.snapshot(), {}) + self.assertEqual(path.read_bytes(), before) + + +class ListStore(RecordingStore): + """A store whose contents the test swaps between loads: another process rewrote the file.""" + + def __init__(self, envelopes=()) -> None: + super().__init__() + self.envelopes = list(envelopes) + + def pull_all(self): + return list(self.envelopes) + + +class BatchAndOrderTests(unittest.TestCase): + def test_a_batch_naming_a_key_twice_lands_the_last_record_and_keeps_every_held_lookup(self) -> None: + # C# CultCacheBatch stages one operation per key, the last one named. + document = item_doc() + store = RecordingStore() + cache = cache_of(document, store=store) + cache.put(document, "item:h", Item(sku="S")) + store.pushed.clear() + cache.put_envelopes( + document, [envelope(document, "item:k", Item(sku="S")), envelope(document, "item:k", Item(sku="T"))] + ) + self.assertEqual(store.pushed, ["item:k"]) + self.assertEqual(cache.get_required(document, "item:k").sku, "T") + self.assertEqual(cache.get_key_by_index(document, "sku", "S"), "item:h") + self.assertEqual(cache.get_key_by_index(document, "sku", "T"), "item:k") + with self.assertRaises(CultCacheError) as caught: + cache.put(document, "item:z", Item(sku="S")) + self.assertEqual(str(caught.exception), refusal("item:h", "item:z", "item", "sku")) + + def test_a_batch_naming_a_key_twice_is_judged_on_the_record_that_lands(self) -> None: + document = item_doc() + cache = cache_of(document) + cache.put(document, "item:h", Item(sku="S")) + # the first record for k would collide with h; only the last lands, so the batch is admitted + cache.put_envelopes( + document, [envelope(document, "item:k", Item(sku="S")), envelope(document, "item:k", Item(sku="T"))] + ) + # and the reverse order lands the colliding record, which is refused + with self.assertRaises(CultCacheError): + cache.put_envelopes( + document, [envelope(document, "item:j", Item(sku="U")), envelope(document, "item:j", Item(sku="S"))] + ) + self.assertNotIn("item:j", cache.snapshot().get("item", {})) + + def test_values_are_named_in_landing_order(self) -> None: + document = item_doc() + cache = cache_of(document) + rows = [("item:a", "B"), ("item:b", "B"), ("item:x", "A"), ("item:y", "A")] + with self.assertRaises(CultCacheError) as caught: + cache.put_envelopes(document, [envelope(document, key, Item(sku=sku)) for key, sku in rows]) + self.assertEqual(str(caught.exception), refusal("item:a", "item:b", "item", "sku")) + + def test_keys_sort_by_utf16_code_unit_not_code_point(self) -> None: + document = item_doc() + cache = cache_of(document) + astral, bmp = "k\U0001F600", "k\uff01" # code points: bmp < astral; UTF-16 units: D83D < FF01 + with self.assertRaises(CultCacheError) as caught: + cache.put_envelopes(document, [envelope(document, bmp, Item(sku="S")), envelope(document, astral, Item(sku="S"))]) + self.assertEqual(str(caught.exception), refusal(astral, bmp, "item", "sku")) + + def test_a_load_judges_every_type_at_once_aliases_in_ordinal_order(self) -> None: + first = item_doc("first", indexes={"zeta": "sku"}, roles={}) + second = item_doc("second", indexes={"alpha": "sku"}, roles={}) + loose_first, loose_second = item_doc("first", indexes={}, roles={}), item_doc("second", indexes={}, roles={}) + store = ListStore( + [envelope(loose_first, "f:1", Item(sku="S")), envelope(loose_first, "f:2", Item(sku="S"))] + + [envelope(loose_second, "s:1", Item(sku="S")), envelope(loose_second, "s:2", Item(sku="S"))] + ) + cache = cache_of(first, second, store=store) + with self.assertRaises(CultCacheError) as caught: + cache.pull_all_backing_stores() + self.assertEqual(str(caught.exception), refusal("s:1", "s:2", "second", "alpha")) + + +class LoadFreshnessTests(unittest.TestCase): + def test_a_load_is_judged_on_the_file_alone_not_against_the_held_map(self) -> None: + document = item_doc() + store = ListStore([envelope(document, "item:b", Item(sku="S")), envelope(document, "item:c", Item(sku="T"))]) + cache = cache_of(document, store=store) + cache.pull_all_backing_stores() + self.assertEqual(cache.get_key_by_index(document, "sku", "S"), "item:b") + # another process moved S from b to a, and b to T's old place + store.envelopes = [envelope(document, "item:a", Item(sku="S")), envelope(document, "item:b", Item(sku="U")), store.envelopes[1]] + cache.pull_all_backing_stores() + self.assertEqual(cache.get_key_by_index(document, "sku", "S"), "item:a") + self.assertEqual(cache.get_key_by_index(document, "sku", "U"), "item:b") + + +class DefinitionTests(unittest.TestCase): + def test_every_constructor_refuses_an_alias_in_both_indexes_and_roles(self) -> None: + from dataclasses import replace + + from cultcache_py import DocumentDefinition + + with self.assertRaisesRegex(ValueError, "both an index and a role"): + DocumentDefinition(type="item", indexes={"sku": "sku"}, roles={"sku": "sku"}) + with self.assertRaisesRegex(ValueError, "both an index and a role"): + replace(item_doc(roles={}), roles={"sku": "sku"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/cultmesh-py/README.md b/packages/cultmesh-py/README.md index 60ae4e49..020a2afc 100644 --- a/packages/cultmesh-py/README.md +++ b/packages/cultmesh-py/README.md @@ -43,9 +43,10 @@ from cultmesh_py import ( note_doc = define_database_entry_type( "mesh.note", - [("body", 0), ("name", 1), ("kind", 2)], + [("body", 0), ("name", 1), ("kind", 2), ("slug", 3)], name="name", - indexes={"kind": "kind"}, + indexes={"slug": "slug"}, # unique: no two notes hold one slug; watch_by_index and get_key_by_index look it up + roles={"kind": "kind"}, # many notes share a kind: a selection alias, never a lookup ) node = CultMesh.start_node( "mesh.cc", @@ -55,13 +56,13 @@ node = CultMesh.start_node( node.database.register_document(note_doc) unsubscribe = node.database.watch_record(note_doc, "note:1", lambda change: print(change.change_kind)) unsubscribe_named = node.database.watch_by_name(note_doc, "intro", lambda change: print(change.record_key)) -unsubscribe_kind = node.database.watch_by_index(note_doc, "kind", "guide", lambda change: print(change.record_key)) -node.database.put(note_doc, "note:1", {"body": "hello", "name": "intro", "kind": "guide"}) +unsubscribe_slug = node.database.watch_by_index(note_doc, "slug", "intro-guide", lambda change: print(change.record_key)) +node.database.put(note_doc, "note:1", {"body": "hello", "name": "intro", "kind": "guide", "slug": "intro-guide"}) live_note = node.database.get_required(note_doc, "note:1") put_message = node.database.put_raw_message( note_doc, "note:2", - {"body": "wire me", "name": "wire", "kind": "interop"}, + {"body": "wire me", "name": "wire", "kind": "interop", "slug": "wire-me"}, shard_id="interop", shard_epoch=1, ) diff --git a/packages/cultmesh-py/src/cultmesh_py/node.py b/packages/cultmesh-py/src/cultmesh_py/node.py index cee102f0..619aed36 100644 --- a/packages/cultmesh-py/src/cultmesh_py/node.py +++ b/packages/cultmesh-py/src/cultmesh_py/node.py @@ -10,7 +10,7 @@ from pathlib import Path from typing import Any, Callable from cultcache_py import CultCache, CultCacheEnvelope, SingleFileMessagePackBackingStore -from cultcache_py.cache import CultCacheError +from cultcache_py.cache import CultCacheError, index_text from cultcache_py.documents import DocumentDefinition, extract_value from cultnet_py import ( CultNetAppliedRecord, @@ -520,7 +520,7 @@ class CultMeshDatabase: raise ValueError(f"Document type {document.type!r} has no index {index!r}") def filtered(change: CultMeshDatabaseChange) -> None: - if self._change_matches_extractor(change, extractor, value): + if self._change_matches_extractor(change, extractor, value, index_text): callback(self._change_as_document(change, requested)) return self.watch(filtered, document=document) @@ -1295,10 +1295,11 @@ class CultMeshDatabase: change: CultMeshDatabaseChange, extractor: str | Callable[[Any], str | int | float | bool | None], expected: str, + render: Callable[[Any], str | None] = str, ) -> bool: return ( - self._value_matches_extractor(change.value, extractor, expected) - or self._value_matches_extractor(change.previous_value, extractor, expected) + self._value_matches_extractor(change.value, extractor, expected, render) + or self._value_matches_extractor(change.previous_value, extractor, expected, render) ) def _value_matches_extractor( @@ -1306,11 +1307,12 @@ class CultMeshDatabase: value: Any | None, extractor: str | Callable[[Any], str | int | float | bool | None], expected: str, + render: Callable[[Any], str | None] = str, ) -> bool: if value is None: return False actual = extract_value(value, extractor) - return actual is not None and str(actual) == expected + return actual is not None and render(actual) == expected def _previous_values_for_snapshot_response(self, response: dict[str, Any]) -> dict[tuple[str, str], Any]: if response.get("schemaVersion") != "cultnet.snapshot_response_raw.v0": diff --git a/packages/cultmesh-py/tests/test_unique_index_watch.py b/packages/cultmesh-py/tests/test_unique_index_watch.py new file mode 100644 index 00000000..44c606ff --- /dev/null +++ b/packages/cultmesh-py/tests/test_unique_index_watch.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import re +import unittest +from pathlib import Path + +from cultcache_py import define_database_entry_type +from cultcache_py.cache import CultCacheError +from cultmesh_py import CultMesh + +# variants:ruling:unique-indexes. `indexes` are unique lookups, `roles` are selection aliases. A watch by index reads the +# lookup's own canonical text (cultcache_py.cache.index_text), never a second rendering of the value. + + +class WatchByIndexTests(unittest.TestCase): + def test_the_watch_and_the_lookup_agree_on_the_text_of_a_value(self) -> None: + document = define_database_entry_type("mesh.priced", [("price", 0)], indexes={"price": "price"}) + node = CultMesh.create_node(runtime_id="mesh-watch-text") + node.database.register_document(document) + cases = [ + (2.0, "2", "2.0"), + (0.1, "0.1000000000000000055511151231257827021181583404541015625", "0.1"), + (-0.0, "0", "-0.0"), + ] + for index, (value, text, str_text) in enumerate(cases): + with self.subTest(value=value): + key = f"item:{index}" + heard: list[str] = [] + misheard: list[str] = [] + node.database.watch_by_index(document, "price", text, lambda change: heard.append(change.record_key)) + node.database.watch_by_index(document, "price", str_text, lambda change: misheard.append(change.record_key)) + node.database.put(document, key, {"price": value}) + self.assertEqual(node.database.get_by_index(document, "price", text), {"price": value}) + self.assertEqual(heard, [key]) + if str_text != text: + self.assertEqual(misheard, []) + self.assertIsNone(node.database.get_by_index(document, "price", str_text)) + + def test_a_blank_value_is_never_indexed_so_never_watched(self) -> None: + document = define_database_entry_type("mesh.blank", [("sku", 0)], indexes={"sku": "sku"}) + node = CultMesh.create_node(runtime_id="mesh-watch-blank") + node.database.register_document(document) + heard: list[str] = [] + node.database.watch_by_index(document, "sku", " ", lambda change: heard.append(change.record_key)) + node.database.put(document, "item:1", {"sku": " "}) + self.assertEqual(heard, []) + + def test_a_role_alias_is_not_watchable_by_index(self) -> None: + document = define_database_entry_type("mesh.roles", [("kind", 0)], roles={"kind": "kind"}) + node = CultMesh.create_node(runtime_id="mesh-watch-role") + node.database.register_document(document) + with self.assertRaisesRegex(ValueError, "has no index"): + node.database.watch_by_index(document, "kind", "guide", lambda _: None) + + +class ReadmeTests(unittest.TestCase): + def test_the_readme_note_document_admits_many_notes_of_one_kind(self) -> None: + readme = (Path(__file__).parents[1] / "README.md").read_text(encoding="utf-8") + declaration = re.search(r"^note_doc = define_database_entry_type\(\n.*?^\)\n", readme, re.S | re.M) + self.assertIsNotNone(declaration) + scope: dict = {"define_database_entry_type": define_database_entry_type} + exec(declaration.group(0), scope) # noqa: S102 - the README's own declaration + note_doc = scope["note_doc"] + node = CultMesh.create_node(runtime_id="mesh-readme") + node.database.register_document(note_doc) + node.database.put(note_doc, "note:1", {"body": "a", "name": "one", "kind": "guide", "slug": "one"}) + node.database.put(note_doc, "note:2", {"body": "b", "name": "two", "kind": "guide", "slug": "two"}) + self.assertEqual(node.database.get_by_index(note_doc, "slug", "two")["body"], "b") + with self.assertRaises(CultCacheError): + node.database.put(note_doc, "note:3", {"body": "c", "name": "three", "kind": "guide", "slug": "one"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/cultnet-py/CHANGELOG.md b/packages/cultnet-py/CHANGELOG.md index 5e4b0cfa..e7bd3afd 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) +- `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. - Decodes the write receipt frames of `contracts/cultnet/interop/write-receipt.md`: `parse_message` validates `cultnet.write_admitted.v0` and `cultnet.error.v1` against their closed schemas (unknown keys, empty or non-string ids, an unknown operation or code are refused; `error.v1` requires `messageId`, string or null; `limit_reached` is `error.v1` only). Added `CULTNET_ERROR_V1_CODES`, `CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS`, `write_content_hash` and `hello_advertises_write_receipts`; `hello` takes `supported_message_versions`. `tests/test_write_receipt.py` replays the vectors' `contentHash`, `wire` and `schemaInvalid` sections. Not implemented here: the client write tracker (`scenarios`) and the server rules (`dedupe`, `blankFields`); this package has no receipt server and no write client. - Added `message_invalid` to `CULTNET_ERROR_CODES` and `message_invalid_error(field, reason)`, byte-identical to the C# `ForMessageInvalid`. `cultmesh-py` refuses a malformed put or delete with it instead of the out-of-schema `malformed_document_put` and `malformed_document_delete`, which are removed. - Added `CultNetRudpSocketTransportServer` (`cultnet_py.rudp_server`): one UDP socket serving many RUDP peers, one session per diff --git a/packages/cultnet-py/src/cultnet_py/cultmesh_contracts.py b/packages/cultnet-py/src/cultnet_py/cultmesh_contracts.py index 91d0805b..bc5c2622 100644 --- a/packages/cultnet-py/src/cultnet_py/cultmesh_contracts.py +++ b/packages/cultnet-py/src/cultnet_py/cultmesh_contracts.py @@ -692,7 +692,7 @@ simulation_fact_document: DocumentDefinition[CultMeshSimulationFact] = define_da ], cls=CultMeshSimulationFact, name="fact_id", - indexes={"shard": "shard_id", "frame": "frame", "subject": "subject_id", "claim_kind": "claim_kind"}, + roles={"shard": "shard_id", "frame": "frame", "subject": "subject_id", "claim_kind": "claim_kind"}, schema_name=SIMULATION_FACT_DOCUMENT_TYPE, schema_version=SIMULATION_FACT_SCHEMA_VERSION, ) diff --git a/packages/cultnet-py/src/cultnet_py/schema_catalog.py b/packages/cultnet-py/src/cultnet_py/schema_catalog.py index ee6fe430..85ab799d 100644 --- a/packages/cultnet-py/src/cultnet_py/schema_catalog.py +++ b/packages/cultnet-py/src/cultnet_py/schema_catalog.py @@ -432,6 +432,8 @@ def wire_message_schema_properties(schema_version: str) -> dict[str, Any]: "document": {"$ref": "#/$defs/rawDocumentRecord"}, "schemaId": common["schemaId"], "recordKey": common["recordKey"], + "commitId": {"type": ["string", "null"], "minLength": 1}, + "commitCount": {"type": ["integer", "null"], "minimum": 2}, }, "cultnet.shard_catalog_request.v0": { **common, diff --git a/packages/cultnet-py/src/cultnet_py/subscription.py b/packages/cultnet-py/src/cultnet_py/subscription.py index 6756376b..74c3be33 100644 --- a/packages/cultnet-py/src/cultnet_py/subscription.py +++ b/packages/cultnet-py/src/cultnet_py/subscription.py @@ -15,6 +15,10 @@ class CultNetDatabaseChange: raw_document: CultNetRawDocumentRecord | None = None schema_id: str | None = None record_key: str | None = None + commit_id: str | None = None + """Set on every change of one multi-record commit, the same for all of them; None outside one (C# writes nil).""" + commit_count: int | None = None + """How many changes the commit holds (>= 2); None outside one. Carried data: Python applies no batch.""" @classmethod def from_wire(cls, message: dict[str, Any]) -> "CultNetDatabaseChange": @@ -31,6 +35,12 @@ class CultNetDatabaseChange: if raw_document is not None: schema_id = schema_id or raw_document.schema_id record_key = record_key or raw_document.record_key + commit_id = message.get("commitId") + if commit_id is not None and not (isinstance(commit_id, str) and commit_id): + raise ValueError(f"database change commitId must be a non-empty string, received {commit_id!r}") + commit_count = message.get("commitCount") + if commit_count is not None and (isinstance(commit_count, bool) or not isinstance(commit_count, int) or commit_count < 2): + raise ValueError(f"database change commitCount must be an integer of at least 2, received {commit_count!r}") return cls( message_id=str(message.get("messageId") or ""), subscription_id=str(message.get("subscriptionId") or ""), @@ -39,6 +49,8 @@ class CultNetDatabaseChange: raw_document=raw_document, schema_id=schema_id, record_key=record_key, + commit_id=commit_id, + commit_count=commit_count, ) def to_wire(self) -> dict[str, Any]: @@ -56,6 +68,10 @@ class CultNetDatabaseChange: wire["schemaId"] = self.schema_id if self.record_key is not None: wire["recordKey"] = self.record_key + if self.commit_id is not None: + wire["commitId"] = self.commit_id + if self.commit_count is not None: + wire["commitCount"] = self.commit_count return wire diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 6e8d2b71..5ece34be 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -1566,6 +1566,63 @@ class CultNetTests(unittest.TestCase): self.assertEqual(delete_change.schema_id, "schema-note") self.assertEqual(delete_change.to_wire()["recordKey"], "note:1") + def test_cultnet_database_change_commit_fields(self) -> None: + base = { + "schemaVersion": "cultnet.database_change_raw.v0", + "messageId": "change-commit", + "subscriptionId": "sub-1", + "changeKind": "removed", + "schemaId": "schema-note", + "recordKey": "note:1", + } + grouped = CultNetDatabaseChange.from_wire({**base, "commitId": "commit-7", "commitCount": 3}) + self.assertEqual((grouped.commit_id, grouped.commit_count), ("commit-7", 3)) + self.assertEqual(grouped.to_wire()["commitId"], "commit-7") + self.assertEqual(grouped.to_wire()["commitCount"], 3) + self.assertEqual(CultNetDatabaseChange.from_wire(grouped.to_wire()), grouped) + + for absent in (base, {**base, "commitId": None, "commitCount": None}): # C# writes nil outside a commit + change = CultNetDatabaseChange.from_wire(absent) + self.assertIsNone(change.commit_id) + self.assertIsNone(change.commit_count) + self.assertNotIn("commitId", change.to_wire()) + self.assertNotIn("commitCount", change.to_wire()) + + for field, bad in ( + ("commitId", ""), + ("commitId", 7), + ("commitCount", 1), + ("commitCount", 0), + ("commitCount", True), + ("commitCount", 2.0), + ("commitCount", "3"), + ): + with self.subTest(field=field, bad=bad), self.assertRaisesRegex(ValueError, field): + CultNetDatabaseChange.from_wire({**base, field: bad}) + + def test_a_two_change_commit_from_the_csharp_server_parses(self) -> None: + # CultNetDatabaseServer.DeliverCommit (CultNetDatabaseServer.cs:791-797) tags every change of a commit with its + # commitId and commitCount = the number of changes delivered, so a swap of two records arrives with commitCount 2. + # Shaped as the C# [MessagePackObject] writes it: every key present, nil where unset. + swap = [ + { + "schemaVersion": "cultnet.database_change_raw.v0", + "messageId": f"change-{index}", + "subscriptionId": "sub-1", + "changeKind": "updated", + "document": None, + "recordKey": key, + "schemaId": "schema-note", + "commitId": "9f1c0f0e4b7a4c1f9d2e", + "commitCount": 2, + } + for index, key in enumerate(("note:a", "note:b")) + ] + changes = [CultNetDatabaseChange.from_wire(message) for message in swap] + self.assertEqual([change.commit_count for change in changes], [2, 2]) + self.assertEqual({change.commit_id for change in changes}, {"9f1c0f0e4b7a4c1f9d2e"}) + self.assertEqual([change.to_wire()["commitCount"] for change in changes], [2, 2]) + def test_cultnet_raw_put_helper_carries_message_id(self) -> None: put = document_put_raw( message_id="put-1", @@ -2272,6 +2329,8 @@ class CultNetTests(unittest.TestCase): self.assertIn("document", change_schema["properties"]) self.assertIn("schemaId", change_schema["properties"]) self.assertIn("recordKey", change_schema["properties"]) + self.assertEqual(change_schema["properties"]["commitId"], {"type": ["string", "null"], "minLength": 1}) + self.assertEqual(change_schema["properties"]["commitCount"], {"type": ["integer", "null"], "minimum": 2}) log_schema = json.loads(by_version["cultnet.shard_log_response.v0"]["schemaJson"]) self.assertIn("entries", log_schema["required"])