From 59f9e9aedb9d5457e7c446fc306eaa24f38cff33 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 09:01:00 +0200 Subject: [PATCH] One local schema id in cultmesh-py; catalog writers merge only the same schema; a compatible entry before its owner reads as the owner cultnet-py's resolve_document_and_schema_id_for_raw_record is replaced by resolve_document_for_raw_record, which resolves the document only; cultmesh-py keys previous values by the local schema id, so a snapshot under a compatible id or the schema name reports an update with its previous value. Writers (C#, TypeScript, Python) merge same-id entries only when their schema names match; another schema's entry is dropped or the record gets a default entry under its own id, and a record's own id is claimed before an entry that lists it as compatible. The C# single-file writer prefers the entry being written now, as the directory writer does, so name, content hash and members come from the registered schema and the compatible ids are the union. Vector: compatible-before-owner-v3. A cultnet-py test's schema path no longer depends on the working directory. Co-Authored-By: Claude Sonnet 5.5 --- .../cultcache-py/src/cultcache_py/stores.py | 43 ++++++++----- packages/cultcache-py/tests/test_cultcache.py | 31 ++++++++- packages/cultcache-rs/src/lib.rs | 23 ++++--- .../single-file-messagepack-backing-store.ts | 63 +++++++++++++------ packages/cultcache-ts/test/cult-cache.test.ts | 45 +++++++++++-- packages/cultmesh-py/src/cultmesh_py/node.py | 26 +++----- packages/cultmesh-py/tests/test_cultmesh.py | 40 ++++++++++++ .../cultnet-py/src/cultnet_py/__init__.py | 4 +- .../cultnet-py/src/cultnet_py/replication.py | 15 +---- packages/cultnet-py/tests/test_cultnet.py | 2 +- src/GameCult.Caching/CultCache.cs | 13 ++-- .../StoreReadabilityTests.cs | 60 +++++++++++++++++- .../compatible-before-owner-v3.bin | 1 + .../readability/generate.py | 3 + .../readability/manifest.txt | 1 + 15 files changed, 278 insertions(+), 92 deletions(-) create mode 100644 tests/vectors/document-variants-c2a/readability/compatible-before-owner-v3.bin diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index 3c6d61c6..f06104b1 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -165,24 +165,37 @@ def _encode_snapshot(envelopes: list[CultCacheEnvelope], format_version: str) -> # own id or a compatible one), else a default entry under the record's id. The catalog is keyed by the entry's own id, so a # record read under a compatible id is written back beside the entry that lists it. catalog_by_schema_id: dict[str, CultCacheSchemaCatalogEntry] = {} + + def place(envelope: CultCacheEnvelope, candidates: list[CultCacheSchemaCatalogEntry]) -> None: + for candidate in candidates: + existing = catalog_by_schema_id.get(candidate.schema_id) + if existing is None: + catalog_by_schema_id[candidate.schema_id] = candidate + return + # Entries that share an id and a schema name are one schema seen by different writers: the entry written lists every id + # any of them lists. An entry of another schema is never merged: the record gets a default entry under its own id. + if existing.schema_name == candidate.schema_name: + catalog_by_schema_id[candidate.schema_id] = replace( + existing, + compatible_schema_ids=tuple(dict.fromkeys((*existing.compatible_schema_ids, *candidate.compatible_schema_ids))), + ) + return + raise ValueError(f"CultCache schema id {_schema_id_for(envelope)!r} cannot identify two different schemas") + + # A record's own id is claimed first, so an entry that lists it as compatible cannot take it from the schema that owns it, + # whatever order the records arrive in. + listed: list[tuple[CultCacheEnvelope, CultCacheSchemaCatalogEntry]] = [] for envelope in envelopes: schema_id = _schema_id_for(envelope) supplied = envelope.catalog_entry - published = ( - supplied - if supplied is not None and (supplied.schema_id == schema_id or schema_id in supplied.compatible_schema_ids) - else _default_catalog_entry(envelope) - ) - # Entries that share an id are one schema seen by different writers: the entry written lists every id any of them lists. - existing = catalog_by_schema_id.get(published.schema_id) - catalog_by_schema_id[published.schema_id] = ( - published - if existing is None - else replace( - existing, - compatible_schema_ids=tuple(dict.fromkeys((*existing.compatible_schema_ids, *published.compatible_schema_ids))), - ) - ) + if supplied is not None and supplied.schema_id != schema_id and schema_id in supplied.compatible_schema_ids: + listed.append((envelope, supplied)) + elif supplied is not None and supplied.schema_id == schema_id: + place(envelope, [supplied]) + else: + place(envelope, [_default_catalog_entry(envelope)]) + for envelope, supplied in listed: + place(envelope, [supplied, _default_catalog_entry(envelope)]) catalog = [ _encode_catalog_entry(entry) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 4596c5ab..ffbc2706 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -347,8 +347,35 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(len(SingleFileMessagePackBackingStore(path).pull_all()), 2) def test_an_id_an_entry_owns_names_that_entry_not_one_that_lists_it_as_compatible(self) -> None: - store = SingleFileMessagePackBackingStore(self._C2A_VECTORS / "readability" / "own-id-over-compatible-v3.bin") - self.assertEqual([envelope.type for envelope in store.pull_all()], ["vectors.item", "vectors.item"]) + for vector in ("own-id-over-compatible-v3.bin", "compatible-before-owner-v3.bin"): + store = SingleFileMessagePackBackingStore(self._C2A_VECTORS / "readability" / vector) + self.assertEqual([envelope.type for envelope in store.pull_all()], ["vectors.item", "vectors.item"], vector) + + # Entries of different schemas that share an id are not merged: the record gets a default entry under its own id, so no record + # is read as another schema, whichever order the records arrive in. + def test_the_store_writer_never_merges_an_entry_of_another_schema(self) -> None: + import msgpack # type: ignore + from cultcache_py import CultCacheSchemaCatalogEntry + from cultcache_py.stores import _decode_snapshot, _encode_snapshot + + def entry(name: str, compatible: tuple[str, ...]) -> CultCacheSchemaCatalogEntry: + return CultCacheSchemaCatalogEntry( + schema_id="tests.retype.x", schema_name=name, schema_version=name + ".v1", + content_hash="tests.retype.x", canonical_schema_json="", compatible_schema_ids=compatible, members=(), + ) + + def record(key: str, type: str, schema_id: str, catalog_entry: CultCacheSchemaCatalogEntry) -> CultCacheEnvelope: + return CultCacheEnvelope( + key=key, type=type, payload=msgpack.packb({"key": key}, use_bin_type=True), + stored_at="2026-09-30T00:00:00Z", schema_id=schema_id, catalog_entry=catalog_entry, + ) + + a = record("a", "tests.retype.mine", "tests.retype.x", entry("tests.retype.mine", ("tests.retype.x",))) + b = record("b", "tests.retype.theirs", "tests.retype.y", entry("tests.retype.theirs", ("tests.retype.x", "tests.retype.y"))) + for order in ([a, b], [b, a]): + decoded = msgpack.unpackb(msgpack.packb(_encode_snapshot(order, "cultcache.store.v1"), use_bin_type=True), raw=False) + _, envelopes = _decode_snapshot(decoded) + self.assertEqual(sorted((e.key, e.type) for e in envelopes), [("a", "tests.retype.mine"), ("b", "tests.retype.theirs")]) def test_a_record_loaded_under_a_compatible_schema_id_is_written_back_under_the_id_its_catalog_entry_carries(self) -> None: import msgpack # type: ignore diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 2fa649b0..0be95fed 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -4826,16 +4826,19 @@ mod tests { // An id one entry owns and a later entry lists as compatible names the entry that owns it. #[test] fn an_id_an_entry_owns_names_that_entry_not_one_that_lists_it_as_compatible() -> Result<()> { - let temp = tempfile::tempdir()?; - let path = temp.path().join("store.msgpack"); - std::fs::copy( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")) - .join("../../tests/vectors/document-variants-c2a/readability/own-id-over-compatible-v3.bin"), - &path, - )?; - let envelopes = SingleFileMessagePackBackingStore::new(&path).pull_all()?; - assert_eq!(envelopes.len(), 2); - assert!(envelopes.iter().all(|envelope| envelope.r#type == "vectors.item"), "{envelopes:?}"); + for vector in ["own-id-over-compatible-v3.bin", "compatible-before-owner-v3.bin"] { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.msgpack"); + std::fs::copy( + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../tests/vectors/document-variants-c2a/readability") + .join(vector), + &path, + )?; + let envelopes = SingleFileMessagePackBackingStore::new(&path).pull_all()?; + assert_eq!(envelopes.len(), 2, "{vector}"); + assert!(envelopes.iter().all(|envelope| envelope.r#type == "vectors.item"), "{vector}: {envelopes:?}"); + } Ok(()) } diff --git a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts index 46af9bbc..b8227ba2 100644 --- a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts +++ b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts @@ -220,13 +220,9 @@ function encodeSnapshot(entries: CultCacheEnvelope[], format: StoreFormat): unkn // record read under a compatible id is written back beside the entry that lists it. function catalogEntriesFor(entries: CultCacheEnvelope[]): Map { const catalog = new Map(); - for (const entry of entries) { + const defaultEntryFor = (entry: CultCacheEnvelope): CultCacheSchemaCatalogEntry => { const schemaId = schemaIdForEnvelope(entry); - const supplied = entry.catalogEntry; - const publishes = - supplied !== undefined && - (supplied.schemaId === schemaId || (supplied.compatibleSchemaIds ?? []).includes(schemaId)); - const published: CultCacheSchemaCatalogEntry = publishes ? supplied : { + return { schemaId, schemaName: entry.type, schemaVersion: `${entry.type}.v1`, @@ -239,19 +235,48 @@ function catalogEntriesFor(entries: CultCacheEnvelope[]): Map { + for (const candidate of candidates) { + const existing = catalog.get(candidate.schemaId); + if (existing === undefined) { + catalog.set(candidate.schemaId, candidate); + return; + } + + // Entries that share an id and a schema name are one schema seen by different writers: the entry written lists every id any + // of them lists. An entry of another schema is never merged: the record gets a default entry under its own id. + if (existing.schemaName === candidate.schemaName) { + catalog.set(candidate.schemaId, { + ...existing, + compatibleSchemaIds: [ + ...new Set([...(existing.compatibleSchemaIds ?? [existing.schemaId]), ...(candidate.compatibleSchemaIds ?? [candidate.schemaId])]), + ], + }); + return; + } + } + + throw new Error(`CultCache schema id "${schemaIdForEnvelope(entry)}" cannot identify two different schemas.`); + }; + + // A record's own id is claimed first, so an entry that lists it as compatible cannot take it from the schema that owns it, + // whatever order the records arrive in. + const listed: Array<[CultCacheEnvelope, CultCacheSchemaCatalogEntry]> = []; + for (const entry of entries) { + const schemaId = schemaIdForEnvelope(entry); + const supplied = entry.catalogEntry; + if (supplied !== undefined && supplied.schemaId !== schemaId && (supplied.compatibleSchemaIds ?? []).includes(schemaId)) { + listed.push([entry, supplied]); + } else if (supplied !== undefined && supplied.schemaId === schemaId) { + place(entry, [supplied]); + } else { + place(entry, [defaultEntryFor(entry)]); + } + } + + for (const [entry, supplied] of listed) { + place(entry, [supplied, defaultEntryFor(entry)]); } return catalog; diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 951b3896..21b37e1b 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -1383,13 +1383,48 @@ test("the store writer merges the compatible ids of entries that share a schema assert.equal((await new SingleFileMessagePackBackingStore(file).pullAll()).length, 2); }); +// Entries of different schemas that share an id are not merged: the record gets a default entry under its own id, so no record is +// read as another schema, whichever order the records arrive in. +test("the store writer never merges an entry of another schema into a record's catalog entry", async () => { + const entry = (schemaName: string, compatibleSchemaIds: string[]) => ({ + schemaId: "tests.retype.x", + schemaName, + schemaVersion: `${schemaName}.v1`, + contentHash: "tests.retype.x", + canonicalSchemaJson: "", + compatibleSchemaIds, + members: [], + }); + const record = (key: string, type: string, schemaId: string, catalogEntry: ReturnType) => ({ + key, + type, + schemaId, + payload: encode({ key }), + storedAt: "2026-09-30T00:00:00.0000000Z", + catalogEntry, + }); + // a is schema "mine" under x; b is schema "theirs" under y, carrying an entry with x's id that lists y. + const a = record("a", "tests.retype.mine", "tests.retype.x", entry("tests.retype.mine", ["tests.retype.x"])); + const b = record("b", "tests.retype.theirs", "tests.retype.y", entry("tests.retype.theirs", ["tests.retype.x", "tests.retype.y"])); + const dir = await mkdtemp(join(tmpdir(), "cultcache-retype-")); + + for (const [name, order] of [["ab", [a, b]], ["ba", [b, a]]] as const) { + const file = join(dir, `${name}.msgpack`); + await new SingleFileMessagePackBackingStore(file).pushAll([...order]); + const back = await new SingleFileMessagePackBackingStore(file).pullAll(); + assert.deepEqual(back.map((envelope) => [envelope.key, envelope.type]).sort(), [["a", "tests.retype.mine"], ["b", "tests.retype.theirs"]], name); + } +}); + test("an id an entry owns names that entry, not one that lists it as compatible", async () => { const dir = await mkdtemp(join(tmpdir(), "cultcache-own-id-")); - const file = await copyVector(dir, join(c2aVectors, "readability", "own-id-over-compatible-v3.bin"), "own.msgpack"); - const envelopes = await new SingleFileMessagePackBackingStore(file).pullAll(); - assert.deepEqual(envelopes.map((envelope) => envelope.type), ["vectors.item", "vectors.item"]); - const inspection = inspectCultCacheBytes("own.msgpack", await readFile(file)); - assert.deepEqual(inspection.records.map((record) => record.schemaName), ["vectors.item", "vectors.item"]); + for (const vector of ["own-id-over-compatible-v3.bin", "compatible-before-owner-v3.bin"]) { + const file = await copyVector(dir, join(c2aVectors, "readability", vector), "own.msgpack"); + const envelopes = await new SingleFileMessagePackBackingStore(file).pullAll(); + assert.deepEqual(envelopes.map((envelope) => envelope.type), ["vectors.item", "vectors.item"], vector); + const inspection = inspectCultCacheBytes("own.msgpack", await readFile(file)); + assert.deepEqual(inspection.records.map((record) => record.schemaName), ["vectors.item", "vectors.item"], vector); + } }); // The inspector reads what the reader reads: a record under an id its catalog lists only as compatible is inspected. diff --git a/packages/cultmesh-py/src/cultmesh_py/node.py b/packages/cultmesh-py/src/cultmesh_py/node.py index 09ac2012..9af9f624 100644 --- a/packages/cultmesh-py/src/cultmesh_py/node.py +++ b/packages/cultmesh-py/src/cultmesh_py/node.py @@ -26,7 +26,7 @@ from cultnet_py import ( apply_shard_log_response as apply_cultnet_shard_log_response, document_delete, document_put_raw, - resolve_document_and_schema_id_for_raw_record, + resolve_document_for_raw_record, schema_document_map, ) @@ -657,13 +657,11 @@ class CultMeshDatabase: schema_id = str(document_record.get("schemaId") or "") documents_by_schema_id = schema_document_map(self.documents) try: - document, resolved_schema_id = resolve_document_and_schema_id_for_raw_record( - documents_by_schema_id, - schema_id, - document_record, - ) + document = resolve_document_for_raw_record(documents_by_schema_id, schema_id, document_record) except KeyError: return None + # One id: the local schema id, under which the cache stores the record and previous values are keyed. + resolved_schema_id = document.catalog_entry().schema_id record_key = str(document_record["recordKey"]) previous = self.cache.get(document, record_key) envelope = CultCacheEnvelope( @@ -1142,12 +1140,8 @@ class CultMeshDatabase: documents_by_schema_id = schema_document_map(self.documents) if raw_record is not None: try: - _, resolved_schema_id = resolve_document_and_schema_id_for_raw_record( - documents_by_schema_id, - schema_id, - raw_record, - ) - return resolved_schema_id, record_key + document = resolve_document_for_raw_record(documents_by_schema_id, schema_id, raw_record) + return document.catalog_entry().schema_id, record_key except KeyError: pass document = documents_by_schema_id.get(schema_id) @@ -1307,15 +1301,11 @@ class CultMeshDatabase: continue schema_id = str(record.get("schemaId")) try: - document, resolved_schema_id = resolve_document_and_schema_id_for_raw_record( - documents_by_schema_id, - schema_id, - record, - ) + document = resolve_document_for_raw_record(documents_by_schema_id, schema_id, record) except KeyError: continue record_key = str(record.get("recordKey")) - previous[(resolved_schema_id, record_key)] = self.cache.get(document, record_key) + previous[(document.catalog_entry().schema_id, record_key)] = self.cache.get(document, record_key) return previous @staticmethod diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index 9a697370..5e2befcf 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -338,6 +338,46 @@ class CultMeshTests(unittest.TestCase): self.assertEqual(observed[0].value["value"], "new") self.assertEqual(node.cache.get_required_envelope(document, "policy:1").schema_id, local_schema_id) + # A record arriving under a compatible id or the schema name is the same record: one id, the local one, so an existing + # record is updated with its previous value. + def test_cultmesh_node_reports_an_update_when_a_record_arrives_under_a_compatible_id_or_the_schema_name(self) -> None: + document = define_database_entry_type( + "mesh.compatible-policy", + [("schema_version", 0), ("name", 1), ("value", 2)], + schema_id="mesh.compatible-policy.current", + schema_name="mesh.compatible_policy", + schema_version="mesh.compatible_policy.v1", + compatible_schema_ids=["mesh.compatible-policy.current", "mesh.compatible-policy.older"], + ) + for wire_schema_id in ("mesh.compatible-policy.older", "mesh.compatible_policy"): + node = create_node(runtime_id="python-node") + node.register_document(document) + node.put(document, "policy:1", {"schema_version": "mesh.compatible_policy.v1", "name": "policy", "value": "old"}) + observed: list[CultMeshDatabaseChange] = [] + unsubscribe = node.database.watch_record(document, "policy:1", observed.append) + try: + applied = node.database.apply_snapshot_response({ + "schemaVersion": "cultnet.snapshot_response_raw.v0", + "messageId": "compatible-snapshot", + "documents": [{ + "schemaId": wire_schema_id, + "recordKey": "policy:1", + "storedAt": "2026-06-25T00:00:00Z", + "payloadEncoding": "messagepack", + "payload": document.encode_payload({ + "schema_version": "mesh.compatible_policy.v1", "name": "policy", "value": "new", + }), + }], + }) + finally: + unsubscribe() + + local_schema_id = document.catalog_entry().schema_id + self.assertEqual(applied[0].schema_id, local_schema_id, wire_schema_id) + self.assertEqual([change.change_kind for change in observed], ["updated"], wire_schema_id) + self.assertEqual(observed[0].previous_value["value"], "old", wire_schema_id) + self.assertEqual(observed[0].value["value"], "new", wire_schema_id) + def test_cultmesh_node_syncs_snapshot_and_shard_log_through_cultnet_client(self) -> None: import msgpack # type: ignore from cultnet_py import read_frame, write_frame diff --git a/packages/cultnet-py/src/cultnet_py/__init__.py b/packages/cultnet-py/src/cultnet_py/__init__.py index b1734f25..61c81668 100644 --- a/packages/cultnet-py/src/cultnet_py/__init__.py +++ b/packages/cultnet-py/src/cultnet_py/__init__.py @@ -65,7 +65,7 @@ from .replication import ( apply_raw_document_record, apply_raw_snapshot, apply_shard_log_response, - resolve_document_and_schema_id_for_raw_record, + resolve_document_for_raw_record, schema_document_map, ) from .schema_catalog import ( @@ -223,7 +223,7 @@ __all__ = [ "apply_raw_document_record", "apply_raw_snapshot", "apply_shard_log_response", - "resolve_document_and_schema_id_for_raw_record", + "resolve_document_for_raw_record", "compute_simulation_claim_hash", "compute_reconnect_delay_ms", "create_reconnect_policy", diff --git a/packages/cultnet-py/src/cultnet_py/replication.py b/packages/cultnet-py/src/cultnet_py/replication.py index bb26a8c8..531c0012 100644 --- a/packages/cultnet-py/src/cultnet_py/replication.py +++ b/packages/cultnet-py/src/cultnet_py/replication.py @@ -154,28 +154,19 @@ def can_resolve_document_for_raw_record( return False +# routes a replicated raw record to a local binding; the store always persists the local schema id; this is not store-reader recovery. def resolve_document_for_raw_record( documents_by_schema_id: dict[str, DocumentDefinition[Any]], schema_id: str, record: dict[str, Any], ) -> DocumentDefinition[Any]: - return resolve_document_and_schema_id_for_raw_record(documents_by_schema_id, schema_id, record)[0] - - -# routes a replicated raw record to a local binding; the store always persists the local schema id; this is not store-reader recovery. -def resolve_document_and_schema_id_for_raw_record( - documents_by_schema_id: dict[str, DocumentDefinition[Any]], - schema_id: str, - record: dict[str, Any], -) -> tuple[DocumentDefinition[Any], str]: document = documents_by_schema_id.get(schema_id) if document is not None: - return document, schema_id + return document schema_version = _infer_schema_version_from_payload(bytes(record["payload"])) schema_name = _infer_schema_name(schema_version) if schema_version is not None else None if schema_name is not None and schema_name in documents_by_schema_id: - document = documents_by_schema_id[schema_name] - return document, document.catalog_entry().schema_id + return documents_by_schema_id[schema_name] raise KeyError(schema_id) diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index b8aad39d..45315e1c 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -2190,7 +2190,7 @@ class CultNetTests(unittest.TestCase): runtime_kind="python", display_name="Python Interop Test", agent_id="python-interop-test-agent", - schema_path=str(Path("contracts/cultnet/interop/cultnet.interop-note.schema.json")), + schema_path=str(Path(__file__).resolve().parents[3] / "contracts" / "cultnet" / "interop" / "cultnet.interop-note.schema.json"), ) binding = state.bindings["note"] logged_value = { diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index ff753ce8..28884e10 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -29,15 +29,18 @@ namespace GameCult.Caching public CultSchemaMemberCatalogEntry[] Members { get; set; } = Array.Empty(); - // One entry per schema id, whatever order the candidates come in: entries that share an id are one schema seen by different - // writers, so the entry written lists every id any of them lists, and its other fields are the first's (or the last's). + // One entry per schema id, whatever order the candidates come in. Entries that share an id and a schema name are one schema seen + // by different writers: the merged entry takes its name, content hash and members from the entry being written now (the + // last with preferLast, else the first) and lists every id any of them lists. An entry of another schema under the same id is + // dropped, never merged: its ids are not this schema's. internal static IEnumerable MergeById(IEnumerable entries, bool preferLast = false) { foreach (var group in entries.GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)) { var members = group.ToArray(); var chosen = preferLast ? members[^1] : members[0]; - if (members.Length == 1) + var sameSchema = members.Where(entry => string.Equals(entry.SchemaName, chosen.SchemaName, StringComparison.Ordinal)).ToArray(); + if (sameSchema.Length == 1) { yield return chosen; continue; @@ -50,7 +53,7 @@ namespace GameCult.Caching SchemaVersion = chosen.SchemaVersion, ContentHash = chosen.ContentHash, CanonicalSchemaJson = chosen.CanonicalSchemaJson, - CompatibleSchemaIds = members.SelectMany(entry => entry.CompatibleSchemaIds).Distinct(StringComparer.Ordinal).ToArray(), + CompatibleSchemaIds = sameSchema.SelectMany(entry => entry.CompatibleSchemaIds).Distinct(StringComparer.Ordinal).ToArray(), Members = chosen.Members }; } @@ -3664,7 +3667,7 @@ namespace GameCult.Caching var snapshot = new CultPersistedStoreSnapshot { FormatVersion = HeaderFor(holdsElementIds, existingHeader, wholeStore, directoryStore: false, holdsVariants: ordered.Any(record => record.Variant != null)), - SchemaCatalog = CultSchemaCatalogEntry.MergeById(catalog) + SchemaCatalog = CultSchemaCatalogEntry.MergeById(catalog, preferLast: true) // An entry stays while it publishes a schema id some record carries, as its own id or a compatible one. .Where(entry => used.Contains(entry.SchemaId) || entry.CompatibleSchemaIds.Any(used.Contains)) .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal) diff --git a/tests/GameCult.Caching.Tests/StoreReadabilityTests.cs b/tests/GameCult.Caching.Tests/StoreReadabilityTests.cs index d6931b69..abbc3864 100644 --- a/tests/GameCult.Caching.Tests/StoreReadabilityTests.cs +++ b/tests/GameCult.Caching.Tests/StoreReadabilityTests.cs @@ -156,11 +156,12 @@ namespace GameCult.Caching.Tests } // An id one entry owns and a later entry lists as compatible names the entry that owns it. - [Test] - public void AnIdAnEntryOwnsNamesThatEntryNotOneThatListsItAsCompatible() + [TestCase("own-id-over-compatible-v3.bin")] + [TestCase("compatible-before-owner-v3.bin")] + public void AnIdAnEntryOwnsNamesThatEntryNotOneThatListsItAsCompatible(string vector) { var path = Path.Combine(_directory, "own-id.cc"); - File.WriteAllBytes(path, File.ReadAllBytes(Path.Combine(VectorRoot(), "own-id-over-compatible-v3.bin"))); + File.WriteAllBytes(path, File.ReadAllBytes(Path.Combine(VectorRoot(), vector))); using var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry }); Assert.That(cache.Get(new CultRecordKey("alpha")), Is.Not.Null, "the record read as the schema that owns its id"); @@ -200,6 +201,59 @@ namespace GameCult.Caching.Tests var rewritten = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(path)); Assert.That(rewritten.SchemaCatalog.Single().CompatibleSchemaIds, Does.Contain("vectors.old.id")); + Assert.That(rewritten.SchemaCatalog.Single().ContentHash, Is.Not.EqualTo("stale"), "the merged entry's content hash is the schema being written's"); + } + + // The single-file writer merges the same way: a commit onto a file keeps the ids the file's entry lists, and the entry's + // content hash is the registered schema's, not the file's stale one. + [Test] + public void ASingleFileCommitMergesTheFilesEntryWithTheSchemaBeingWritten() + { + var path = Seed("merged-single.cc"); + var manifest = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(path)); + var entry = manifest.SchemaCatalog.Single(); + manifest.Records.Single().SchemaId = "vectors.old.id"; + entry.ContentHash = "stale"; + entry.CompatibleSchemaIds = new[] { entry.SchemaId, "vectors.old.id" }; + File.WriteAllBytes(path, CultDocumentMessagePackSerialization.SerializeSnapshot(manifest)); + + using (var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry })) + { + Assert.That(cache.Commit(batch => + { + batch.Expect(new CultRecordKey("e"), null); + batch.Upsert(typeof(IdDeck), new IdDeck { Name = "e" }, new CultRecordKey("e")); + }), Is.True); + } + + var rewritten = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(path)); + Assert.That(rewritten.SchemaCatalog.Single().CompatibleSchemaIds, Does.Contain("vectors.old.id")); + Assert.That(rewritten.SchemaCatalog.Single().ContentHash, Is.Not.EqualTo("stale")); + } + + // Entries that share an id merge only when they are the same schema: another schema's entry is dropped, its ids never + // listed, and the entry written is the one being written now. + [Test] + public void EntriesOfDifferentSchemasThatShareAnIdAreNotMerged() + { + CultSchemaCatalogEntry Entry(string name, string hash, params string[] compatible) => new() + { + SchemaId = "x", SchemaName = name, SchemaVersion = name + ".v1", ContentHash = hash, CompatibleSchemaIds = compatible + }; + var mine = Entry("mine", "h1", "x"); + var theirs = Entry("theirs", "h2", "x", "z"); + + foreach (var order in new[] { new[] { mine, theirs }, new[] { theirs, mine } }) + { + var first = CultSchemaCatalogEntry.MergeById(order, preferLast: false).Single(); + var last = CultSchemaCatalogEntry.MergeById(order, preferLast: true).Single(); + Assert.That(first, Is.SameAs(order[0])); + Assert.That(last, Is.SameAs(order[1])); + } + + var sameSchema = CultSchemaCatalogEntry.MergeById(new[] { Entry("mine", "old", "x", "y"), Entry("mine", "new", "x") }, preferLast: true).Single(); + Assert.That(sameSchema.ContentHash, Is.EqualTo("new")); + Assert.That(sameSchema.CompatibleSchemaIds, Is.EquivalentTo(new[] { "x", "y" })); } // Entries that share an id are merged whatever order they arrive in: every id any of them lists survives. diff --git a/tests/vectors/document-variants-c2a/readability/compatible-before-owner-v3.bin b/tests/vectors/document-variants-c2a/readability/compatible-before-owner-v3.bin new file mode 100644 index 00000000..2c32e0bb --- /dev/null +++ b/tests/vectors/document-variants-c2a/readability/compatible-before-owner-v3.bin @@ -0,0 +1 @@ +“²cultcache.store.v3’—±vectors.item.next­vectors.other°vectors.other.v1±vectors.item.next¢{}‘¬vectors.item—¬vectors.item¬vectors.item¯vectors.item.v1¬vectors.itemÙL{"schemaName":"vectors.item","schemaVersion":"vectors.item.v1","members":[]}’”¥alpha¬vectors.item¹2026-09-29T00:00:00+00:00Ä’¥alpha”¤beta¬vectors.item¹2026-09-29T00:00:01+00:00Ä’¤beta \ No newline at end of file diff --git a/tests/vectors/document-variants-c2a/readability/generate.py b/tests/vectors/document-variants-c2a/readability/generate.py index 2b455225..f44d2a20 100644 --- a/tests/vectors/document-variants-c2a/readability/generate.py +++ b/tests/vectors/document-variants-c2a/readability/generate.py @@ -43,6 +43,9 @@ vectors = { "compatible-id-only-v3": pack([base[0], [[base[1][0][0] + ".next"] + base[1][0][1:]], base[2]]), # An id that one entry owns and a later entry lists as compatible names the entry that owns it (here, vectors.item, which does not list itself as compatible). "own-id-over-compatible-v3": pack([base[0], [base[1][0][:5] + [[]] + base[1][0][6:]] + [["vectors.item.next", "vectors.other", "vectors.other.v1", "vectors.item.next", "{}", ["vectors.item"], []]], base[2]]), + # The same, with the entry that lists the id as compatible before the entry that owns it: a reader that takes the first entry + # naming an id in one pass reads the record as vectors.other. + "compatible-before-owner-v3": pack([base[0], [["vectors.item.next", "vectors.other", "vectors.other.v1", "vectors.item.next", "{}", ["vectors.item"], []], base[1][0][:5] + [[]] + base[1][0][6:]], base[2]]), # v3-base plus a variant of alpha: a fifth record slot, [baseKey, overrides[]], over an empty payload, overriding the name. Only C# reads it. "variant-slot-v3": pack([base[0], base[1], base[2] + [["gamma"] + base[2][0][1:3] + [b"", ["alpha", [[0, [[0, ""]], "", "gamma"]]]]]]), # v3-base with a fourth top-level slot after the records. diff --git a/tests/vectors/document-variants-c2a/readability/manifest.txt b/tests/vectors/document-variants-c2a/readability/manifest.txt index 841c4ec4..7d9b4224 100644 --- a/tests/vectors/document-variants-c2a/readability/manifest.txt +++ b/tests/vectors/document-variants-c2a/readability/manifest.txt @@ -25,3 +25,4 @@ variant-slot-v3.bin reads refuses refuses refuses missing-schema-versioned-v3.bin refuses refuses refuses refuses compatible-id-only-v3.bin reads reads reads reads own-id-over-compatible-v3.bin reads reads reads reads +compatible-before-owner-v3.bin reads reads reads reads