GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Recover schema-stamped cache records across runtimes

Browse source
This commit is contained in:
Meta Vens 2026-06-27 14:34:53 +01:00
commit a0b31b8f2a
11 changed files with 773 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

22
packages/cultcache-py/src/cultcache_py/cache.py
View file

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

41
packages/cultcache-py/src/cultcache_py/stores.py
View file

@ -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]

58
packages/cultcache-py/src/cultnet_py/replication.py
View file

@ -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]

118
packages/cultcache-py/tests/test_cultcache.py
View file

@ -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

173
packages/cultcache-rs/src/lib.rs
View file

@ -252,6 +252,7 @@ struct CultCacheStoreRegistration {
pub struct CultCache {
definitions: BTreeMap<String, &'static str>,
schema_name_definitions: BTreeMap<String, String>,
entries: BTreeMap<String, CultCacheEnvelope>,
stores: Vec<CultCacheStoreRegistration>,
}
@ -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<String> = 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<String>,
schema_name_definitions: &BTreeMap<String, String>,
persisted_type: &str,
) -> Option<String> {
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<Vec<CultCacheEnvelope>> {
.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<Vec<CultCacheEnvelope>> {
.collect()
}
fn infer_schema_version_from_payload(payload: &[u8]) -> Option<String> {
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<u32> {
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<String> {
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<String> {
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::<SchemaStamped>()?;
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path));
cache.pull_all_backing_stores()?;
assert_eq!(cache.get_required::<SchemaStamped>("record-1")?, expected);
assert_eq!(
cache
.get_required_envelope::<SchemaStamped>("record-1")?
.schema_id,
Some("sha256:stale-schema-id-from-cold-record".to_string())
);
Ok(())
}
}

69
packages/cultcache-ts/src/cult-cache-inspector.ts
View file

@ -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<string, Inspected
throw new Error("CultCache persisted records must declare key, schemaId, and storedAt.");
}
const catalogEntry = catalogBySchemaId.get(schemaId);
const payloadBytes = normalizePayloadBytes(payload);
const catalogEntry = catalogBySchemaId.get(schemaId) ?? recoverCatalogEntry(schemaId, payloadBytes);
return {
key,
schemaId,
@ -196,6 +214,53 @@ function decodeV1Record(value: unknown, catalogBySchemaId: Map<string, Inspected
};
}
function recoverCatalogEntry(schemaId: string, payload: Uint8Array): InspectedCatalogEntry | undefined {
let payloadPreview: unknown;
try {
payloadPreview = decode(payload);
} catch {
return undefined;
}
const schemaVersion = inferSchemaVersionFromPayloadPreview(payloadPreview);
if (!schemaVersion) {
return undefined;
}
const schemaName = inferSchemaName(schemaVersion);
if (!schemaName) {
return undefined;
}
return {
schemaId,
schemaName,
schemaVersion,
contentHash: schemaId,
canonicalSchemaJson: "",
compatibleSchemaIds: [],
members: [],
};
}
function inferSchemaVersionFromPayloadPreview(payloadPreview: unknown): string | undefined {
if (Array.isArray(payloadPreview) && typeof payloadPreview[0] === "string") {
return payloadPreview[0];
}
if (isRecord(payloadPreview)) {
const schemaVersion = payloadPreview.schemaVersion ?? payloadPreview.schema_version;
return typeof schemaVersion === "string" ? schemaVersion : undefined;
}
return undefined;
}
function inferSchemaName(schemaVersion: string): string | undefined {
const match = /^(.*)\.v\d+$/.exec(schemaVersion);
return match?.[1];
}
function decodeLegacyEnvelope(value: unknown): InspectedRecord {
if (!isRecord(value)) {
throw new Error("CultCache legacy envelopes must be MessagePack maps.");

15
packages/cultcache-ts/src/cult-cache.ts
View file

@ -74,6 +74,7 @@ export class CultCache {
readonly #definitions = new Map<string, RegisteredDefinition>();
readonly #schemaIdDefinitions = new Map<string, RegisteredDefinition>();
readonly #schemaNameDefinitions = new Map<string, RegisteredDefinition>();
readonly #entries = new Map<string, HydratedEntry>();
readonly #typeEntryIds = new Map<string, Set<string>>();
readonly #nameToKeyMaps = new Map<string, Map<string, string>>();
@ -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 {

55
packages/cultcache-ts/src/single-file-messagepack-backing-store.ts
View file

@ -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<string, CultCacheSchemaCatalogEntry>,
): 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.");

88
packages/cultcache-ts/test/cult-cache.test.ts
View file

@ -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;

84
src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs
View file

@ -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<CultSchemaMigrationReport>();
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<CultSchemaCatalogEntry> ResolveCatalogForRecord(
CultPersistedRecord record,
IReadOnlyCollection<CultSchemaCatalogEntry> 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<string>(),
Members = Array.Empty<CultSchemaMemberCatalogEntry>()
})
.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)

75
tests/GameCult.Caching.Tests/BackingStoreTests.cs
View file

@ -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<CultSchemaCatalogEntry>(),
Records = Array.Empty<CultPersistedRecord>()
};
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<SchemaStampedTestEntry>(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
{