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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
c8326ce3dc
commit
59f9e9aedb
15 changed files with 278 additions and 92 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(())
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string, CultCacheSchemaCatalogEntry> {
|
||||
const catalog = new Map<string, CultCacheSchemaCatalogEntry>();
|
||||
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<string, CultCacheS
|
|||
compatibleSchemaIds: [schemaId],
|
||||
members: [],
|
||||
};
|
||||
// Entries that share an id are one schema seen by different writers: the entry written lists every id any of them lists.
|
||||
const existing = catalog.get(published.schemaId);
|
||||
catalog.set(
|
||||
published.schemaId,
|
||||
existing === undefined
|
||||
? published
|
||||
: {
|
||||
...existing,
|
||||
compatibleSchemaIds: [
|
||||
...new Set([...(existing.compatibleSchemaIds ?? [existing.schemaId]), ...(published.compatibleSchemaIds ?? [published.schemaId])]),
|
||||
],
|
||||
},
|
||||
);
|
||||
};
|
||||
const place = (entry: CultCacheEnvelope, candidates: CultCacheSchemaCatalogEntry[]): void => {
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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<typeof entry>) => ({
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
|
|
@ -29,15 +29,18 @@ namespace GameCult.Caching
|
|||
|
||||
public CultSchemaMemberCatalogEntry[] Members { get; set; } = Array.Empty<CultSchemaMemberCatalogEntry>();
|
||||
|
||||
// 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<CultSchemaCatalogEntry> MergeById(IEnumerable<CultSchemaCatalogEntry> 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)
|
||||
|
|
|
|||
|
|
@ -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<VectorItem>(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.
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
“²cultcache.store.v3’—±vectors.item.nextvectors.other°vectors.other.v1±vectors.item.next¢{}‘¬vectors.item<65>—¬vectors.item¬vectors.item¯vectors.item.v1¬vectors.itemÙL{"schemaName":"vectors.item","schemaVersion":"vectors.item.v1","members":[]}<7D><>’”¥alpha¬vectors.item¹2026-09-29T00:00:00+00:00Ä’¥alpha”¤beta¬vectors.item¹2026-09-29T00:00:01+00:00Ä’¤beta
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue