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

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()