diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 0ffdd18a..f7ad1c12 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -17,6 +17,7 @@ class CultCacheError(RuntimeError): @dataclass class _State: documents: dict[str, DocumentDefinition[Any]] = field(default_factory=dict) + documents_by_schema_name: dict[str, DocumentDefinition[Any]] = field(default_factory=dict) values: dict[str, dict[str, Any]] = field(default_factory=dict) envelopes: dict[str, dict[str, CultCacheEnvelope]] = field(default_factory=dict) stores_by_type: dict[str, list[BackingStore]] = field(default_factory=dict) @@ -72,7 +73,11 @@ class CultCache: def register_document_type(self, document: DocumentDefinition[Any]) -> None: if document.type in self._state.documents: raise CultCacheError(f"Document type already registered: {document.type}") + schema_name = document.catalog_entry().schema_name + if schema_name in self._state.documents_by_schema_name: + raise CultCacheError(f"Document schema name already registered: {schema_name}") self._state.documents[document.type] = document + self._state.documents_by_schema_name[schema_name] = document if document.name is not None: self.register_name_lookup(document, document.name) for index, extractor in document.indexes.items(): @@ -105,9 +110,18 @@ class CultCache: seen_globals: set[str] = set() for store in [*self._all_specific_stores(), *self._state.generic_stores]: for envelope in store.pull_all(): - document = self._state.documents.get(envelope.type) + document = self._resolve_document_for_envelope(envelope) if document is None: raise CultCacheError(f"Unknown persisted document type: {envelope.type}") + if envelope.type != document.type: + envelope = CultCacheEnvelope( + key=envelope.key, + type=document.type, + payload=envelope.payload, + stored_at=envelope.stored_at, + schema_id=envelope.schema_id, + catalog_entry=envelope.catalog_entry, + ) if document.global_document: if envelope.type in seen_globals and envelope.key == GLOBAL_KEY: raise CultCacheError(f"Duplicate global document for type: {envelope.type}") @@ -361,3 +375,9 @@ class CultCache: self._assert_registered(document) if not document.global_document: raise CultCacheError(f"Document type is not global: {document.type}") + + def _resolve_document_for_envelope(self, envelope: CultCacheEnvelope) -> DocumentDefinition[Any] | None: + document = self._state.documents.get(envelope.type) + if document is not None: + return document + return self._state.documents_by_schema_name.get(envelope.type) diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index 3587e717..697882e0 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -178,12 +178,25 @@ def _decode_v1_snapshot(decoded: Any) -> list[CultCacheEnvelope] | None: raise ValueError("CultCache persisted records must declare storedAt") catalog_entry = catalog_by_schema_id.get(schema_id) if catalog_entry is None: - raise ValueError(f'CultCache persisted record "{key}" references missing schema id "{schema_id}"') + schema_version = _infer_schema_version_from_payload(_normalize_payload(payload)) + schema_name = _infer_schema_name(schema_version) if schema_version is not None else None + if schema_name is None: + raise ValueError(f'CultCache persisted record "{key}" references missing schema id "{schema_id}"') + catalog_entry = CultCacheSchemaCatalogEntry( + schema_id=schema_id, + schema_name=schema_name, + schema_version=schema_version, + content_hash=schema_id, + canonical_schema_json="", + compatible_schema_ids=(), + members=(), + ) + normalized_payload = _normalize_payload(payload) envelopes.append( CultCacheEnvelope( key=key, type=catalog_entry.schema_name, - payload=_normalize_payload(payload), + payload=normalized_payload, stored_at=stored_at, schema_id=schema_id, catalog_entry=catalog_entry, @@ -316,3 +329,27 @@ def _normalize_payload(payload: Any) -> bytes: if isinstance(payload, list) and all(isinstance(item, int) and 0 <= item <= 255 for item in payload): return bytes(payload) raise ValueError("CultCache record payload must be binary MessagePack bytes") + + +def _infer_schema_version_from_payload(payload: bytes) -> str | None: + try: + msgpack = SingleFileMessagePackBackingStore._msgpack() + decoded = msgpack.unpackb(payload, raw=False) + except Exception: + return None + if isinstance(decoded, list) and decoded and isinstance(decoded[0], str): + return decoded[0] + if isinstance(decoded, dict): + value = decoded.get("schemaVersion", decoded.get("schema_version")) + return value if isinstance(value, str) else None + return None + + +def _infer_schema_name(schema_version: str) -> str | None: + marker = schema_version.rfind(".v") + if marker <= 0: + return None + version = schema_version[marker + 2:] + if not version or not version.isdigit(): + return None + return schema_version[:marker] diff --git a/packages/cultcache-py/src/cultnet_py/replication.py b/packages/cultcache-py/src/cultnet_py/replication.py index 7290a434..8b18e94f 100644 --- a/packages/cultcache-py/src/cultnet_py/replication.py +++ b/packages/cultcache-py/src/cultnet_py/replication.py @@ -23,6 +23,7 @@ def schema_document_map(documents: Iterable[DocumentDefinition[Any]]) -> dict[st for document in documents: catalog_entry = document.catalog_entry() mapped[catalog_entry.schema_id] = document + mapped[catalog_entry.schema_name] = document for compatible_schema_id in catalog_entry.compatible_schema_ids: mapped[compatible_schema_id] = document return mapped @@ -35,7 +36,7 @@ def apply_raw_document_record( ) -> CultNetAppliedRecord: wire = record.to_wire() if isinstance(record, CultNetRawDocumentRecord) else record schema_id = str(wire["schemaId"]) - document = documents_by_schema_id[schema_id] + document = resolve_document_for_raw_record(documents_by_schema_id, schema_id, wire) envelope = raw_record_to_envelope(document, schema_id, wire) value = cache.put_envelope(document, envelope) return CultNetAppliedRecord( @@ -76,7 +77,7 @@ def apply_raw_snapshot( if not isinstance(record, dict): continue schema_id = str(record["schemaId"]) - document = documents_by_schema_id[schema_id] + document = resolve_document_for_raw_record(documents_by_schema_id, schema_id, record) envelope = raw_record_to_envelope(document, schema_id, record) batch = batches.setdefault(document.type, (document, [])) batch[1].append((schema_id, envelope)) @@ -120,7 +121,7 @@ def apply_shard_log_response( if change_kind in ("added", "updated") and isinstance(entry.get("put"), dict): put = entry["put"] schema_id = str(put["document"]["schemaId"]) - if schema_id not in documents_by_schema_id: + if not can_resolve_document_for_raw_record(documents_by_schema_id, schema_id, put["document"]): continue result = apply_raw_document_record(cache, documents_by_schema_id, put["document"]) applied.append(CultNetAppliedRecord(result.schema_id, result.record_key, str(change_kind), result.value)) @@ -134,3 +135,54 @@ def apply_shard_log_response( cache.delete(document, record_key) applied.append(CultNetAppliedRecord(schema_id, record_key, "removed", None)) return applied + + +def can_resolve_document_for_raw_record( + documents_by_schema_id: dict[str, DocumentDefinition[Any]], + schema_id: str, + record: dict[str, Any], +) -> bool: + try: + resolve_document_for_raw_record(documents_by_schema_id, schema_id, record) + return True + except KeyError: + return False + + +def resolve_document_for_raw_record( + documents_by_schema_id: dict[str, DocumentDefinition[Any]], + schema_id: str, + record: dict[str, Any], +) -> DocumentDefinition[Any]: + document = documents_by_schema_id.get(schema_id) + if document is not None: + 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: + return documents_by_schema_id[schema_name] + raise KeyError(schema_id) + + +def _infer_schema_version_from_payload(payload: bytes) -> str | None: + try: + import msgpack # type: ignore + decoded = msgpack.unpackb(payload, raw=False) + except Exception: + return None + if isinstance(decoded, list) and decoded and isinstance(decoded[0], str): + return decoded[0] + if isinstance(decoded, dict): + value = decoded.get("schemaVersion", decoded.get("schema_version")) + return value if isinstance(value, str) else None + return None + + +def _infer_schema_name(schema_version: str) -> str | None: + marker = schema_version.rfind(".v") + if marker <= 0: + return None + version = schema_version[marker + 2:] + if not version or not version.isdigit(): + return None + return schema_version[:marker] diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 16d72d98..1f5660ec 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -351,6 +351,53 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(raw[2][0][0], "app") self.assertEqual(raw[2][0][1], "settings") + def test_messagepack_store_recovers_schema_stamped_record_missing_catalog_entry(self) -> None: + try: + import msgpack # type: ignore + except ModuleNotFoundError: + self.skipTest("msgpack optional dependency is not installed") + + with tempfile.TemporaryDirectory() as tmp: + document = define_database_entry_type( + "runtime-policy", + [ + ("schema_version", 0), + ("name", 1), + ("value", 2), + ], + schema_id="schema-current", + schema_name="tests.schema_stamped_entry", + schema_version="tests.schema_stamped_entry.v1", + ) + store_path = Path(tmp) / "missing-catalog.msgpack" + store_path.write_bytes(msgpack.packb([ + "cultcache.store.v1", + [], + [[ + "record-1", + "sha256:stale-schema-id-from-cold-record", + "2026-06-25T12:00:00Z", + msgpack.packb([ + "tests.schema_stamped_entry.v1", + "schema-stamped", + "still readable", + ], use_bin_type=True), + ]], + ], use_bin_type=True)) + + cache = ( + CultCache.builder() + .register_document_type(document) + .add_generic_store(SingleFileMessagePackBackingStore(store_path)) + .build() + ) + + cache.pull_all_backing_stores() + self.assertEqual(cache.get_required(document, "record-1")["value"], "still readable") + envelope = cache.get_required_envelope(document, "record-1") + self.assertEqual(envelope.type, "runtime-policy") + self.assertEqual(envelope.schema_id, "sha256:stale-schema-id-from-cold-record") + def test_interop_cli_helpers_round_trip_v1_store(self) -> None: with tempfile.TemporaryDirectory() as tmp: store_path = str(Path(tmp) / "cache.msgpack") @@ -1759,6 +1806,77 @@ class CultCacheTests(unittest.TestCase): self.assertEqual([change.change_kind for change in typed_applied_log], ["updated", "removed"]) self.assertIsNone(cache.get(document, "item:1")) + def test_cultnet_replication_helpers_recover_schema_stamped_raw_records(self) -> None: + document = define_database_entry_type( + "replica.runtime-policy", + [ + ("schema_version", 0), + ("name", 1), + ("value", 2), + ], + schema_id="replica.runtime-policy.current", + schema_name="replica.runtime_policy", + schema_version="replica.runtime_policy.v1", + ) + cache = CultCache() + cache.register_document_type(document) + stale_schema_id = "sha256:stale-replica-runtime-policy" + payload = document.encode_payload({ + "schema_version": "replica.runtime_policy.v1", + "name": "policy", + "value": "still synced", + }) + record = { + "schemaId": stale_schema_id, + "recordKey": "policy:1", + "storedAt": "2026-06-25T00:00:00Z", + "payloadEncoding": "messagepack", + "payload": payload, + } + snapshot = { + "schemaVersion": "cultnet.snapshot_response_raw.v0", + "messageId": "snapshot-stale", + "documents": [record], + } + + applied_snapshot = apply_raw_snapshot(cache, [document], snapshot) + self.assertEqual(applied_snapshot[0].schema_id, stale_schema_id) + self.assertEqual(cache.get_required(document, "policy:1")["value"], "still synced") + direct_applied = apply_raw_document_record(cache, schema_document_map([document]), record) + self.assertEqual(direct_applied.schema_id, stale_schema_id) + + shard_log = { + "schemaVersion": "cultnet.shard_log_response.v0", + "messageId": "log-stale", + "shardId": "interop", + "shardEpoch": 1, + "resyncRequired": False, + "entries": [ + { + "sequence": 0, + "changeKind": "updated", + "put": { + "schemaVersion": "cultnet.document_put_raw.v0", + "messageId": "put-stale", + "document": { + **record, + "payload": document.encode_payload({ + "schema_version": "replica.runtime_policy.v1", + "name": "policy", + "value": "updated through log", + }), + }, + "shardId": "interop", + "shardEpoch": 1, + }, + } + ], + } + + applied_log = apply_shard_log_response(cache, [document], shard_log) + self.assertEqual(applied_log[0].schema_id, stale_schema_id) + self.assertEqual(cache.get_required(document, "policy:1")["value"], "updated through log") + 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/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index cb56b8a8..67e88cdf 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -252,6 +252,7 @@ struct CultCacheStoreRegistration { pub struct CultCache { definitions: BTreeMap, + schema_name_definitions: BTreeMap, entries: BTreeMap, stores: Vec, } @@ -260,6 +261,7 @@ impl CultCache { pub fn new() -> Self { Self { definitions: BTreeMap::new(), + schema_name_definitions: BTreeMap::new(), entries: BTreeMap::new(), stores: Vec::new(), } @@ -279,7 +281,18 @@ impl CultCache { T::TYPE )); } + if let Some(existing_type) = self.schema_name_definitions.get(T::SCHEMA_NAME) + && existing_type != T::TYPE + { + return Err(anyhow!( + "CultCache schema name {:?} is already registered for type {:?}", + T::SCHEMA_NAME, + existing_type + )); + } self.definitions.insert(T::TYPE.to_string(), T::SCHEMA_NAME); + self.schema_name_definitions + .insert(T::SCHEMA_NAME.to_string(), T::TYPE.to_string()); Ok(()) } @@ -310,14 +323,20 @@ impl CultCache { pub fn pull_all_backing_stores(&mut self) -> Result<()> { self.entries.clear(); let known_types: BTreeSet = self.definitions.keys().cloned().collect(); + let schema_name_definitions = self.schema_name_definitions.clone(); for registration in &mut self.stores { - for entry in registration.store.pull_all()? { - if !known_types.contains(&entry.r#type) { + for mut entry in registration.store.pull_all()? { + let Some(canonical_type) = resolve_registered_type( + &known_types, + &schema_name_definitions, + &entry.r#type, + ) else { return Err(anyhow!( "No schema is registered for persisted entry type {:?}", entry.r#type )); - } + }; + entry.r#type = canonical_type; self.entries.insert(entry_id(&entry), entry); } } @@ -522,6 +541,7 @@ impl CultCache { .filter_map(|(index, registration)| registration.types.is_empty().then_some(index)) .collect() } + } impl Default for CultCache { @@ -538,6 +558,18 @@ fn entry_id_parts(r#type: &str, key: &str) -> String { format!("{type}::{key}", type = r#type) } +fn resolve_registered_type( + known_types: &BTreeSet, + schema_name_definitions: &BTreeMap, + persisted_type: &str, +) -> Option { + if known_types.contains(persisted_type) { + return Some(persisted_type.to_string()); + } + + schema_name_definitions.get(persisted_type).cloned() +} + fn now_utc_second() -> String { chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true) } @@ -605,13 +637,25 @@ fn decode_store_snapshot(bytes: &[u8]) -> Result> { .2 .into_iter() .map(|record| { - let r#type = catalog.get(&record.1).cloned().ok_or_else(|| { - anyhow!( - "CultCache record {:?} references missing schema {:?}", - record.0, - record.1 - ) - })?; + let r#type = match catalog.get(&record.1) { + Some(schema_name) => schema_name.clone(), + None => { + let schema_version = infer_schema_version_from_payload(&record.3).ok_or_else(|| { + anyhow!( + "CultCache record {:?} references missing schema {:?}", + record.0, + record.1 + ) + })?; + infer_schema_name(&schema_version).ok_or_else(|| { + anyhow!( + "CultCache record {:?} references missing schema {:?}", + record.0, + record.1 + ) + })? + } + }; Ok(CultCacheEnvelope { key: record.0, r#type, @@ -623,6 +667,68 @@ fn decode_store_snapshot(bytes: &[u8]) -> Result> { .collect() } +fn infer_schema_version_from_payload(payload: &[u8]) -> Option { + let mut offset = 0usize; + read_array_header(payload, &mut offset)?; + read_string(payload, &mut offset) +} + +fn read_array_header(payload: &[u8], offset: &mut usize) -> Option { + let marker = *payload.get(*offset)?; + *offset += 1; + match marker { + 0x90..=0x9f => Some((marker & 0x0f) as u32), + 0xdc => { + let bytes = payload.get(*offset..(*offset + 2))?; + *offset += 2; + Some(u16::from_be_bytes([bytes[0], bytes[1]]) as u32) + } + 0xdd => { + let bytes = payload.get(*offset..(*offset + 4))?; + *offset += 4; + Some(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])) + } + _ => None, + } +} + +fn read_string(payload: &[u8], offset: &mut usize) -> Option { + let marker = *payload.get(*offset)?; + *offset += 1; + let length = match marker { + 0xa0..=0xbf => (marker & 0x1f) as usize, + 0xd9 => { + let length = *payload.get(*offset)? as usize; + *offset += 1; + length + } + 0xda => { + let bytes = payload.get(*offset..(*offset + 2))?; + *offset += 2; + u16::from_be_bytes([bytes[0], bytes[1]]) as usize + } + 0xdb => { + let bytes = payload.get(*offset..(*offset + 4))?; + *offset += 4; + u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize + } + _ => return None, + }; + + let bytes = payload.get(*offset..(*offset + length))?; + std::str::from_utf8(bytes).ok().map(str::to_string) +} + +fn infer_schema_name(schema_version: &str) -> Option { + let marker = schema_version.rfind(".v")?; + let version = schema_version.get((marker + 2)..)?; + if marker == 0 || version.is_empty() || !version.bytes().all(|value| value.is_ascii_digit()) { + return None; + } + + Some(schema_version[..marker].to_string()) +} + fn escape_json_string(value: &str) -> String { value.replace('\\', "\\\\").replace('"', "\\\"") } @@ -659,6 +765,17 @@ mod tests { body: String, } + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] + #[cultcache(type = "runtime-policy", schema = "tests.schema_stamped_entry")] + struct SchemaStamped { + #[cultcache(key = 0)] + schema_version: String, + #[cultcache(key = 1)] + name: String, + #[cultcache(key = 2)] + value: String, + } + cultcache_registry!(TestEntries { Settings, Note }); #[test] @@ -882,4 +999,40 @@ mod tests { ); Ok(()) } + + #[test] + fn messagepack_store_recovers_schema_stamped_records_missing_catalog_entries() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("missing-catalog.msgpack"); + let expected = SchemaStamped { + schema_version: "tests.schema_stamped_entry.v1".to_string(), + name: "schema-stamped".to_string(), + value: "still readable".to_string(), + }; + let snapshot = PersistedStoreSnapshot( + "cultcache.store.v1".to_string(), + Vec::new(), + vec![PersistedRecord( + "record-1".to_string(), + "sha256:stale-schema-id-from-cold-record".to_string(), + "2026-06-25T12:00:00Z".to_string(), + rmp_serde::to_vec(&expected)?, + )], + ); + fs::write(&store_path, rmp_serde::to_vec(&snapshot)?)?; + + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); + cache.pull_all_backing_stores()?; + + assert_eq!(cache.get_required::("record-1")?, expected); + assert_eq!( + cache + .get_required_envelope::("record-1")? + .schema_id, + Some("sha256:stale-schema-id-from-cold-record".to_string()) + ); + Ok(()) + } } diff --git a/packages/cultcache-ts/src/cult-cache-inspector.ts b/packages/cultcache-ts/src/cult-cache-inspector.ts index da91da41..1301fe47 100644 --- a/packages/cultcache-ts/src/cult-cache-inspector.ts +++ b/packages/cultcache-ts/src/cult-cache-inspector.ts @@ -74,13 +74,31 @@ function inspectV1Snapshot(filePath: string, fileSizeBytes: number, decoded: unk catalogBySchemaId.set(compatibleSchemaId, entry); } } + const records = recordsRaw.map((record) => decodeV1Record(record, catalogBySchemaId)); + for (const record of records) { + if (!catalog.some((entry) => entry.schemaId === record.schemaId)) { + const schemaVersion = inferSchemaVersionFromPayloadPreview(record.payloadPreview); + const schemaName = schemaVersion ? inferSchemaName(schemaVersion) : undefined; + if (schemaName) { + catalog.push({ + schemaId: record.schemaId, + schemaName, + schemaVersion: schemaVersion!, + contentHash: record.schemaId, + canonicalSchemaJson: "", + compatibleSchemaIds: [], + members: [], + }); + } + } + } return { filePath, fileSizeBytes, format: STORE_FORMAT_VERSION, catalog, - records: recordsRaw.map((record) => decodeV1Record(record, catalogBySchemaId)), + records, }; } @@ -184,8 +202,8 @@ function decodeV1Record(value: unknown, catalogBySchemaId: Map(); readonly #schemaIdDefinitions = new Map(); + readonly #schemaNameDefinitions = new Map(); readonly #entries = new Map(); readonly #typeEntryIds = new Map>(); readonly #nameToKeyMaps = new Map>(); @@ -121,6 +122,13 @@ export class CultCache { } this.#schemaIdDefinitions.set(schemaId, registered); } + const existingBySchemaName = this.#schemaNameDefinitions.get(registered.catalogEntry.schemaName); + if (existingBySchemaName && existingBySchemaName !== registered) { + throw new Error( + `CultCache schema name "${registered.catalogEntry.schemaName}" is already registered for type "${existingBySchemaName.definition.type}".`, + ); + } + this.#schemaNameDefinitions.set(registered.catalogEntry.schemaName, registered); this.#rebuildDefinitionLookups(registered); return definition; } @@ -203,6 +211,7 @@ export class CultCache { const value = registered.formatter.decode(payload); this.#applyHydratedEntry(registered, { ...entry, + type: registered.definition.type, payload, }, value, "pull"); } @@ -643,10 +652,12 @@ export class CultCache { #resolveDefinitionForEnvelope(entry: CultCacheEnvelope): RegisteredDefinition | undefined { if (entry.schemaId) { - return this.#schemaIdDefinitions.get(entry.schemaId) ?? this.#definitions.get(entry.type); + return this.#schemaIdDefinitions.get(entry.schemaId) + ?? this.#schemaNameDefinitions.get(entry.type) + ?? this.#definitions.get(entry.type); } - return this.#definitions.get(entry.type); + return this.#schemaNameDefinitions.get(entry.type) ?? this.#definitions.get(entry.type); } #resolveRoute(type: string): StoreRoute { 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 3ed89214..58dbd48d 100644 --- a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts +++ b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts @@ -52,7 +52,7 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { const snapshot = decodeSnapshot(decoded); if (snapshot) { return snapshot.records.map((record) => { - const catalogEntry = snapshot.catalogBySchemaId.get(record.schemaId); + const catalogEntry = resolveCatalogEntryForRecord(record, snapshot.catalogBySchemaId); if (!catalogEntry) { throw new Error(`CultCache persisted record "${record.key}" references missing schema id "${record.schemaId}".`); } @@ -282,6 +282,59 @@ function decodeSnapshot(decoded: unknown): DecodedSnapshot | undefined { return { catalogBySchemaId, records }; } +function resolveCatalogEntryForRecord( + record: PersistedRecord, + catalogBySchemaId: Map, +): CultCacheSchemaCatalogEntry | undefined { + const existing = catalogBySchemaId.get(record.schemaId); + if (existing) { + return existing; + } + + const schemaVersion = inferSchemaVersionFromPayload(record.payload); + if (!schemaVersion) { + return undefined; + } + + const schemaName = inferSchemaName(schemaVersion); + if (!schemaName) { + return undefined; + } + + return { + schemaId: record.schemaId, + schemaName, + schemaVersion, + contentHash: record.schemaId, + canonicalSchemaJson: "", + compatibleSchemaIds: [], + members: [], + }; +} + +function inferSchemaVersionFromPayload(payload: Uint8Array): string | undefined { + try { + const decoded = decode(payload); + if (Array.isArray(decoded) && typeof decoded[0] === "string") { + return decoded[0]; + } + + if (isObject(decoded)) { + const schemaVersion = decoded.schemaVersion ?? decoded.schema_version; + return typeof schemaVersion === "string" ? schemaVersion : undefined; + } + + return undefined; + } catch { + return undefined; + } +} + +function inferSchemaName(schemaVersion: string): string | undefined { + const match = /^(.*)\.v\d+$/.exec(schemaVersion); + return match?.[1]; +} + function decodeCatalogEntry(value: unknown): CultCacheSchemaCatalogEntry { if (!Array.isArray(value)) { throw new Error("CultCache schema catalog entries must be MessagePack arrays."); diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 1067eecd..fb206eb6 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -437,6 +437,58 @@ test("SingleFileMessagePackBackingStore reads CultCache v1 snapshots by schema i assert.equal(cache.getRequiredEnvelope(noteDocument, "record-1").schemaId, "schema-1"); }); +test("SingleFileMessagePackBackingStore recovers schema-stamped records missing catalog entries", async () => { + const stampedDocument = defineDocumentType({ + type: "runtime-policy", + schema: z.tuple([ + z.literal("tests.schema_stamped_entry.v1"), + z.string(), + z.string(), + ]), + schemaId: "schema-current", + schemaName: "tests.schema_stamped_entry", + schemaVersion: "tests.schema_stamped_entry.v1", + compatibleSchemaIds: ["schema-current"], + }); + + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-ts-")), "missing-catalog.msgpack"); + await writeFile( + storePath, + encode([ + "cultcache.store.v1", + [], + [ + [ + "record-1", + "sha256:stale-schema-id-from-cold-record", + "2026-06-25T12:00:00Z", + encode([ + "tests.schema_stamped_entry.v1", + "schema-stamped", + "still readable", + ]), + ], + ], + ]), + ); + + const cache = CultCache.builder() + .withDocumentType(stampedDocument) + .withGenericStore(new SingleFileMessagePackBackingStore(storePath)) + .build(); + + await cache.pullAllBackingStores(); + assert.deepEqual(cache.getRequired(stampedDocument, "record-1"), [ + "tests.schema_stamped_entry.v1", + "schema-stamped", + "still readable", + ]); + assert.equal( + cache.getRequiredEnvelope(stampedDocument, "record-1").schemaId, + "sha256:stale-schema-id-from-cold-record", + ); +}); + test("SingleFileMessagePackBackingStore heals legacy envelopes whose payload was persisted as an object", async () => { const noteDocument = defineDocumentType({ type: "note", @@ -604,6 +656,42 @@ test("CultCache inspector decodes v1 store catalog and record payloads from byte ]); }); +test("CultCache inspector recovers schema-stamped records missing catalog entries", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "cultcache-inspector-")); + const file = join(tempDir, "missing-catalog.cc"); + + await writeFile( + file, + encode([ + "cultcache.store.v1", + [], + [ + [ + "record-1", + "sha256:stale-schema-id-from-cold-record", + "2026-06-25T12:00:00Z", + encode([ + "tests.schema_stamped_entry.v1", + "schema-stamped", + "still readable", + ]), + ], + ], + ]), + ); + + const inspection = inspectCultCacheBytes(file, await readFile(file)); + assert.equal(inspection.catalog.length, 1); + assert.equal(inspection.catalog[0]?.schemaId, "sha256:stale-schema-id-from-cold-record"); + assert.equal(inspection.catalog[0]?.schemaName, "tests.schema_stamped_entry"); + assert.equal(inspection.records[0]?.schemaName, "tests.schema_stamped_entry"); + assert.deepEqual(inspection.records[0]?.payloadPreview, [ + "tests.schema_stamped_entry.v1", + "schema-stamped", + "still readable", + ]); +}); + interface InteropNote { schemaVersion: "cultcache.interop_note.v1"; documentId: string; diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 991037da..4792310f 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -5,6 +5,7 @@ using System.IO; using System.Linq; using System.Security.Cryptography; using System.Text; +using MessagePack; namespace GameCult.Caching.MessagePack; @@ -51,8 +52,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore var reports = new List(); foreach (var record in manifest.Records) { - reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, manifest.SchemaCatalog)); - var stored = ToStoredDocument(record, manifest.SchemaCatalog, CultDocumentMessagePackSerialization.DeserializeUntyped); + var catalog = ResolveCatalogForRecord(record, manifest.SchemaCatalog); + reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog)); + var stored = ToStoredDocument(record, catalog, CultDocumentMessagePackSerialization.DeserializeUntyped); Entries[stored.Key.Value] = stored; _dirtyKeys[stored.Key.Value] = true; EntryAdded.OnNext(stored); @@ -63,8 +65,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore foreach (var recordFile in _recordDirectory.EnumerateFiles("*.msgpack").OrderBy(file => file.Name, StringComparer.Ordinal)) { var record = CultDocumentMessagePackSerialization.DeserializePersistedRecord(File.ReadAllBytes(recordFile.FullName)); - reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, manifest.SchemaCatalog)); - var stored = ToStoredDocument(record, manifest.SchemaCatalog, CultDocumentMessagePackSerialization.DeserializeUntyped); + var catalog = ResolveCatalogForRecord(record, manifest.SchemaCatalog); + reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog)); + var stored = ToStoredDocument(record, catalog, CultDocumentMessagePackSerialization.DeserializeUntyped); Entries[stored.Key.Value] = stored; EntryAdded.OnNext(stored); } @@ -167,6 +170,79 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore }; } + private IReadOnlyCollection ResolveCatalogForRecord( + CultPersistedRecord record, + IReadOnlyCollection manifestCatalog) + { + if (manifestCatalog.Any(entry => string.Equals(entry.SchemaId, record.SchemaId, StringComparison.Ordinal))) + { + return manifestCatalog; + } + + var schemaVersion = TryReadPayloadSchemaVersion(record.Payload); + if (string.IsNullOrWhiteSpace(schemaVersion)) + { + return manifestCatalog; + } + + var schemaName = InferSchemaName(schemaVersion); + if (string.IsNullOrWhiteSpace(schemaName)) + { + return manifestCatalog; + } + + return manifestCatalog + .Append(new CultSchemaCatalogEntry + { + SchemaId = record.SchemaId, + SchemaName = schemaName, + SchemaVersion = schemaVersion, + ContentHash = record.SchemaId, + CanonicalSchemaJson = "", + CompatibleSchemaIds = Array.Empty(), + Members = Array.Empty() + }) + .ToArray(); + } + + private static string TryReadPayloadSchemaVersion(byte[] payload) + { + if (payload == null || payload.Length == 0) + { + return ""; + } + + try + { + var reader = new MessagePackReader(payload); + if (reader.NextMessagePackType != MessagePackType.Array) + { + return ""; + } + + if (reader.ReadArrayHeader() <= 0) + { + return ""; + } + + return reader.NextMessagePackType == MessagePackType.String + ? reader.ReadString() ?? "" + : ""; + } + catch + { + return ""; + } + } + + private static string InferSchemaName(string schemaVersion) + { + var versionMarker = schemaVersion.LastIndexOf(".v", StringComparison.Ordinal); + return versionMarker > 0 && versionMarker + 2 < schemaVersion.Length && char.IsDigit(schemaVersion[versionMarker + 2]) + ? schemaVersion.Substring(0, versionMarker) + : ""; + } + private string RecordPath(string key) => Path.Combine(_recordDirectory.FullName, $"{HashKey(key)}.msgpack"); private static string HashKey(string key) diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index e08a2ebc..2b0b8d1f 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -221,6 +221,67 @@ namespace GameCult.Caching.Tests } } + [Test] + public async Task DirectoryMessagePackBackingStore_Loads_Record_When_Manifest_Misses_Catalog_Entry() + { + var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); + var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); + var recordPath = Path.Combine(recordsPath, "stale-schema-record.msgpack"); + + try + { + Directory.CreateDirectory(recordsPath); + + var entry = new SchemaStampedTestEntry + { + SchemaVersion = "tests.schema_stamped_entry.v1", + Name = "schema-stamped", + Value = "still readable" + }; + var record = new CultPersistedRecord + { + Key = "record:stale", + SchemaId = "sha256:stale-schema-id-from-cold-record", + StoredAt = "2026-06-25T12:00:00Z", + Payload = CultDocumentMessagePackSerialization.SerializeUntyped(entry, typeof(SchemaStampedTestEntry)) + }; + var manifest = new CultPersistedStoreSnapshot + { + FormatVersion = "cultcache.store.v1.directory", + SchemaCatalog = Array.Empty(), + Records = Array.Empty() + }; + + File.WriteAllBytes(filePath, CultDocumentMessagePackSerialization.SerializeSnapshot(manifest)); + File.WriteAllBytes(recordPath, CultDocumentMessagePackSerialization.SerializePersistedRecord(record)); + + var readStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var readCache = new CultCache(); + readCache.AddBackingStore(readStore); + await readCache.PullAllBackingStoresAsync(); + + var loaded = readCache.Get(new CultRecordKey("record:stale")); + + Assert.That(loaded, Is.Not.Null); + Assert.That(loaded!.Value, Is.EqualTo("still readable")); + Assert.That(readStore.LastSchemaMigrationReports, Has.Count.EqualTo(1)); + Assert.That(readStore.LastSchemaMigrationReports[0].PersistedSchemaName, Is.EqualTo("tests.schema_stamped_entry")); + Assert.That(readStore.LastSchemaMigrationReports[0].Kind, Is.EqualTo(CultSchemaMigrationKind.CompatibleDrift)); + } + finally + { + if (File.Exists(filePath)) + { + File.Delete(filePath); + } + + if (Directory.Exists(recordsPath)) + { + Directory.Delete(recordsPath, recursive: true); + } + } + } + [Test] public async Task CultCache_RuntimeType_Upsert_Snapshot_And_Remove_Work_For_Editor_Tooling() { @@ -507,6 +568,20 @@ namespace GameCult.Caching.Tests public int Health; } + [CultDocument("tests.schema_stamped_entry", "tests.schema_stamped_entry.v1")] + internal sealed class SchemaStampedTestEntry + { + [Key(0)] + public string SchemaVersion = string.Empty; + + [Key(1)] + [CultName] + public string Name = string.Empty; + + [Key(2)] + public string Value = string.Empty; + } + [CultDocument("tests.named_entry", "tests.named_entry.v2")] internal sealed class NamedTestEntryAdditive {