diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index badfadb5..9c402968 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -317,7 +317,7 @@ class CultCache: structures, running every check and every user extractor; the cache's state is replaced only once that build finishes, so a refused load admits nothing.""" loaded_values: dict[str, dict[str, Any]] = {} - read_order: dict[tuple[str, str], Any] = {} # (type, key) -> value, in the order the stores yielded them + read_values: dict[tuple[str, str], Any] = {} # (type, key) -> value, one per key, the last read wins; the judge is order-free loaded_envelopes: dict[str, dict[str, CultCacheEnvelope]] = {} loaded_legacy_global_keys: dict[str, str] = {} routes, generic = self._state.stores_by_type, self._state.generic_store @@ -356,10 +356,10 @@ class CultCache: envelope = self._stamped(document, envelope) value = document.decode_payload(envelope.payload) loaded_values.setdefault(envelope.type, {})[envelope.key] = value - read_order[(envelope.type, envelope.key)] = value + read_values[(envelope.type, envelope.key)] = value loaded_envelopes.setdefault(envelope.type, {})[envelope.key] = envelope loaded_names, loaded_indexes, loaded_lookup_keys = self._derive_lookups( - [(type, key, value) for (type, key), value in read_order.items()], self._state.name_extractors, self._state.index_extractors + [(type, key, value) for (type, key), value in read_values.items()], self._state.name_extractors, self._state.index_extractors ) # Plain attribute assignments: nothing fallible runs once the swap begins. ( @@ -627,7 +627,7 @@ class CultCache: name_extractors: dict[str, str | Any], index_extractors: dict[str, dict[str, str | Any]], ) -> tuple[_Names, _Indexes, _LookupKeysByType]: - """Builds name and index lookups for the given (type, key, value) records, judged in the order given, in fresh dicts, touching no cache state.""" + """Builds name and index lookups for the given (type, key, value) records, judged whole and order-free, in fresh dicts, touching no cache state.""" names: _Names = {} indexes: _Indexes = {} lookup_keys: _LookupKeysByType = {} @@ -636,7 +636,7 @@ class CultCache: keys = self._lookup_keys_for(value, name_extractors.get(type), index_extractors.get(type, {})) lookup_keys.setdefault(type, {})[key] = keys landing.append((type, key, keys)) - # Judged whole, over every type at once and on the file alone (never against what the cache held): a held value can + # Judged whole and order-free, 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(landing, {}, ()) for type in lookup_keys: @@ -653,28 +653,35 @@ class CultCache: 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.""" + """The one index rule (variants:ruling:unique-indexes): within one type and one unique alias, no two records hold one + non-blank value, judged on the state the admission would leave and never on staging, file or pull order. The holders + of a value are the landing keys projecting it plus the held key unless this admission replaces it. Of every + (type, alias, value) with two or more distinct holders, the refusal names the least by (alias, its least key, its + second-least key, schema name), each compared by UTF-16 code unit, never the value.""" 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])): + violations: list[tuple[bytes, bytes, bytes, bytes, str, str, str, str]] = [] + for (type, index, index_value), keys in landed.items(): 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) + distinct = sorted(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." + violations.append( + (_utf16_order(index), _utf16_order(distinct[0]), _utf16_order(distinct[1]), _utf16_order(schema_name), + index, distinct[0], distinct[1], schema_name) ) + if violations: + *_, index, first, second, schema_name = min(violations, key=lambda violation: violation[:4]) + raise CultCacheError( + f"Records {first} and {second} ({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 to another held diff --git a/packages/cultcache-py/tests/test_unique_index.py b/packages/cultcache-py/tests/test_unique_index.py index 2d988e78..9c19999d 100644 --- a/packages/cultcache-py/tests/test_unique_index.py +++ b/packages/cultcache-py/tests/test_unique_index.py @@ -470,6 +470,74 @@ class VectorTests(unittest.TestCase): self.assertEqual(path.read_bytes(), before) + def test_refusals(self) -> None: + ids = { + "tests.unique_item": "sha256:93bb890de83069061af696b23f94fd32fb3c2dad3c91ee645e8add56085f1882", + "tests.uq_pair": "sha256:2538cceafb6dcbce22ef11b932ecb804af5f96bc3aaedff692a52d3640433e60", + "tests.uq_alias": "sha256:77a77a394ebd8bf71e14b9b6f2cea4978c25d4e705fd12364f660edbf6c63b1f", + "tests.uq_alpha": "sha256:53aeaf2ea9b25e85ade6b15fc04c7653e7b206c87e258f1aa74dd32a840a1389", + "tests.uq_beta": "sha256:45ed62a64c91f6d815d3ec975c4a4a1b2ce07b6e7262a32d828673cf10899586", + } + aliases = { + "tests.uq_pair": {"Sku": "sku", "Alt": "second"}, + "tests.uq_alias": {"Sku": "sku", "alt": "second"}, + } + documents = { + name: define_database_entry_type( + name, + [database_entry_field("sku", 0, codec=t.string), database_entry_field("second", 1, codec=t.string)], + indexes=aliases.get(name, {"Sku": "sku"}), + roles={}, + schema_id=schema_id, + schema_name=name, + ) + for name, schema_id in ids.items() + } + rows = msgpack.unpackb((VECTORS / "refusals.msgpack").read_bytes(), raw=False) + walked = 0 + for label, ops, expected in rows: + if any(op[0] == "commit" and len({schema for _, schema, _, _ in op[1]}) > 1 for op in ops): + continue # put_envelopes takes one type: no batch across types (rows 3 and 26) + with self.subTest(label=label), tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "store.cc" + state: dict[str, CultCache] = {} + + def perform(op) -> None: + if op[0] == "open": + if op[1] is not None: + shutil.copyfile(VECTORS / op[1], path) + state["cache"] = cache_of(*documents.values(), store=SingleFileMessagePackBackingStore(path)) + state["cache"].pull_all_backing_stores() + elif op[0] == "pull": + shutil.copyfile(VECTORS / op[1], path) + state["cache"].pull_all_backing_stores() + else: + document = documents[op[1][0][1]] + state["cache"].put_envelopes( + document, [envelope(document, key, {"sku": sku, "second": second}) for key, _, sku, second in op[1]] + ) + + for op in ops[:-1]: + perform(op) + if expected is None: + perform(ops[-1]) + else: + last = ops[-1] + if last[0] == "open": + before = (VECTORS / last[1]).read_bytes() + elif last[0] == "pull": + before = (VECTORS / last[1]).read_bytes() + else: + before = path.read_bytes() if path.exists() else b"" + k1, k2, schema, alias = expected + with self.assertRaises(CultCacheError) as caught: + perform(last) + self.assertEqual(str(caught.exception), refusal(k1, k2, schema, alias)) + self.assertEqual(path.read_bytes() if path.exists() else b"", before) + walked += 1 + self.assertEqual(walked, 31) + + class ListStore(RecordingStore): """A store whose contents the test swaps between loads: another process rewrote the file.""" @@ -515,22 +583,6 @@ class BatchAndOrderTests(unittest.TestCase): ) 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={}) @@ -553,7 +605,7 @@ class BatchAndOrderTests(unittest.TestCase): cache.put_envelopes(document, [envelope(document, f"item:{n}", row) for n, row in enumerate(rows)]) self.assertEqual(str(caught.exception), refusal("item:0", "item:1", "item", astral)) - def test_a_load_judges_records_in_the_order_the_file_yields_them(self) -> None: + def test_a_load_names_the_least_violation_whatever_order_the_store_yields(self) -> None: alpha, beta = item_doc("alpha", indexes={"k": "sku"}, roles={}), item_doc("beta", indexes={"k": "sku"}, roles={}) loose_alpha, loose_beta = item_doc("alpha", indexes={}, roles={}), item_doc("beta", indexes={}, roles={}) store = ListStore( @@ -568,7 +620,7 @@ class BatchAndOrderTests(unittest.TestCase): cache = cache_of(alpha, beta, store=store) with self.assertRaises(CultCacheError) as caught: cache.pull_all_backing_stores() - self.assertEqual(str(caught.exception), refusal("b:1", "b:2", "beta", "k")) + self.assertEqual(str(caught.exception), refusal("a:1", "a:2", "alpha", "k")) def test_register_index_refuses_an_alias_the_definition_declares_as_a_role(self) -> None: document = item_doc()