From 446122daef11f86a3aa24d445b2ada1670e6820d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 14 Sep 2026 01:03:30 +0200 Subject: [PATCH] Build the whole next state before a pull swaps it in A pull is all-or-nothing, but TS cleared the cache and then ran user name and index accessors while re-applying entries, and Python swapped values in and then rebuilt lookups. An accessor that threw left the cache partly loaded. TS: the hydrated maps (entries, type entry ids, name and index lookups, global keys) are one HydratedState. The pull builds a fresh one, running every check and accessor, and replaces #hydrated in one assignment at the end. The reset-then-apply sequence, #resetHydratedState, the staged global-key map and the three per-map getOrCreate helpers are deleted; apply/remove take the target state. The duplicate-global check now lives only in apply. Name and index lookup rebuilds build a local map before assigning. Python: the pull derives names and indexes into local dicts and assigns values, envelopes, names and indexes together; _rebuild_indexes delegates to the same pure _derive_lookups. Rust already stages into a local map and assigns self.entries last, with no derived lookups; unchanged. Co-Authored-By: Claude Opus 5 --- .../cultcache-py/src/cultcache_py/cache.py | 39 +++- packages/cultcache-py/tests/test_cultcache.py | 47 +++++ packages/cultcache-ts/src/cult-cache.ts | 184 ++++++++---------- packages/cultcache-ts/test/cult-cache.test.ts | 47 +++++ 4 files changed, 208 insertions(+), 109 deletions(-) diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 5644b5d1..9079792d 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -146,7 +146,9 @@ class CultCache: return "the store routed to " + ", ".join(sorted(type for type, routed in routes.items() if routed is store)) def pull_all_backing_stores(self) -> None: - """Reads every store into a staged image; the cache's view changes only if all of them load.""" + """Builds the complete next state (values, envelopes, name and index lookups) in local + structures, running every check and every user extractor; the cache's state is replaced + only once that build finishes, so a refused load admits nothing.""" loaded_values: dict[str, dict[str, Any]] = {} loaded_envelopes: dict[str, dict[str, CultCacheEnvelope]] = {} routes, generic = self._state.stores_by_type, self._state.generic_store @@ -182,9 +184,14 @@ class CultCache: value = document.decode_payload(envelope.payload) loaded_values.setdefault(envelope.type, {})[envelope.key] = value loaded_envelopes.setdefault(envelope.type, {})[envelope.key] = envelope - self._state.values = loaded_values - self._state.envelopes = loaded_envelopes - self._rebuild_indexes() + loaded_names, loaded_indexes = self._derive_lookups(loaded_values) + # Plain attribute assignments: nothing fallible runs once the swap begins. + ( + self._state.values, + self._state.envelopes, + self._state.names, + self._state.indexes, + ) = (loaded_values, loaded_envelopes, loaded_names, loaded_indexes) def get(self, document: DocumentDefinition[T], key: str) -> T | None: self._assert_registered(document) @@ -365,11 +372,27 @@ class CultCache: return store def _rebuild_indexes(self) -> None: - self._state.names.clear() - self._state.indexes.clear() - for type, values in self._state.values.items(): + self._state.names, self._state.indexes = self._derive_lookups(self._state.values) + + def _derive_lookups( + self, values_by_type: dict[str, dict[str, Any]] + ) -> tuple[dict[tuple[str, str], str], dict[tuple[str, str, str], str]]: + """Builds name and index lookups for the given values in fresh dicts, touching no cache state.""" + names: dict[tuple[str, str], str] = {} + indexes: dict[tuple[str, str, str], str] = {} + for type, values in values_by_type.items(): + name_extractor = self._state.name_extractors.get(type) + index_extractors = self._state.index_extractors.get(type, {}) for key, value in values.items(): - self._add_value_indexes(type, key, value) + if name_extractor is not None: + name = extract_value(value, name_extractor) + if name is not None: + names[(type, str(name))] = key + for index, extractor in index_extractors.items(): + index_value = extract_value(value, extractor) + if index_value is not None: + indexes[(type, index, str(index_value))] = key + return names, indexes def _add_value_indexes(self, type: str, key: str, value: Any) -> None: name_extractor = self._state.name_extractors.get(type) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 2636458e..ae1080de 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -368,6 +368,53 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(cache.get_key_by_name(item, "Potion"), "item:potion") self.assertEqual(cache.get_required_global(settings)["theme"], "ash") + def test_pull_with_throwing_index_accessor_leaves_cache_unchanged(self) -> None: + def category(item: Item) -> str: + if item.name == "Bomb": + raise ValueError("category accessor refuses Bomb") + return item.category + + with tempfile.TemporaryDirectory() as tmp: + item = define_document_type( + "item", + encode=lambda value: asdict(value), + decode=lambda raw: Item(**raw), + name="name", + indexes={"category": category}, + ) + settings = define_document_type("settings", global_document=True) + store_path = Path(tmp) / "generic.cc" + cache = ( + CultCache.builder() + .register_document_type(item) + .register_document_type(settings) + .add_generic_store(SingleFileMessagePackBackingStore(store_path)) + .build() + ) + cache.put(item, "item:potion", Item(name="Potion", category="Consumable", value=50)) + cache.put_global(settings, {"theme": "ash"}) + before_envelopes = cache.snapshot_envelopes() + before_values = cache.snapshot() + + # A record whose index accessor throws lands in the store file behind the cache's back. + bomb = replace( + cache.get_required_envelope(item, "item:potion"), + key="item:bomb", + payload=item.encode_payload(Item(name="Bomb", category="Explosive", value=1)), + ) + SingleFileMessagePackBackingStore(store_path).push(bomb) + + with self.assertRaisesRegex(ValueError, "category accessor refuses Bomb"): + cache.pull_all_backing_stores() + self.assertEqual(cache.snapshot_envelopes(), before_envelopes) + self.assertEqual(cache.snapshot(), before_values) + self.assertEqual(cache.get_required(item, "item:potion").value, 50) + self.assertIsNone(cache.get(item, "item:bomb")) + self.assertEqual(cache.get_key_by_name(item, "Potion"), "item:potion") + self.assertIsNone(cache.get_key_by_name(item, "Bomb")) + self.assertEqual(cache.get_key_by_index(item, "category", "Consumable"), "item:potion") + self.assertEqual(cache.get_required_global(settings)["theme"], "ash") + def test_attach_with_empty_type_list_is_the_generic_store(self) -> None: with tempfile.TemporaryDirectory() as tmp: settings = define_database_entry_type("settings", [("theme", 0)]) diff --git a/packages/cultcache-ts/src/cult-cache.ts b/packages/cultcache-ts/src/cult-cache.ts index d4dcc37e..74e96980 100644 --- a/packages/cultcache-ts/src/cult-cache.ts +++ b/packages/cultcache-ts/src/cult-cache.ts @@ -26,6 +26,34 @@ type HydratedEntry = CultCacheEnvelope & { value: unknown; }; +// Everything a pull replaces. Built whole, then swapped in with one assignment. +type HydratedState = { + readonly entries: Map; + readonly typeEntryIds: Map>; + readonly nameToKeyMaps: Map>; + readonly indexToKeyMaps: Map>; + readonly globalKeys: Map; +}; + +function emptyHydratedState(): HydratedState { + return { + entries: new Map(), + typeEntryIds: new Map(), + nameToKeyMaps: new Map(), + indexToKeyMaps: new Map(), + globalKeys: new Map(), + }; +} + +function getOrCreate(map: Map, key: K, create: () => V): V { + let value = map.get(key); + if (value === undefined) { + value = create(); + map.set(key, value); + } + return value; +} + type BackingStoreTypeReference = string | AnyCultCacheDocumentDefinition; type AccessorInput = string | ((value: any) => unknown); @@ -70,11 +98,7 @@ export class CultCache { readonly #definitions = new Map(); readonly #schemaIdDefinitions = new Map(); readonly #schemaNameDefinitions = new Map(); - readonly #entries = new Map(); - readonly #typeEntryIds = new Map>(); - readonly #nameToKeyMaps = new Map>(); - readonly #indexToKeyMaps = new Map>(); - readonly #globalKeys = new Map(); + #hydrated = emptyHydratedState(); readonly #stores: CultCacheStoreRegistration[] = []; static builder(): CultCacheBuilder { @@ -186,7 +210,7 @@ export class CultCache { } } const candidate: CultCacheStoreRegistration = { store, types: claimed }; - for (const type of this.#typeEntryIds.keys()) { + for (const type of this.#hydrated.typeEntryIds.keys()) { const before = this.#resolveRegistration(type); this.#stores.push(candidate); const after = this.#resolveRegistration(type); @@ -205,11 +229,11 @@ export class CultCache { this.addBackingStore(store); } - // Every store is read and every record resolved, decoded and checked against its home before - // the cache's view is replaced, so a refused load admits nothing. + // The complete next state (entries, globals, name and index lookups) is built into a fresh + // HydratedState, running every check and every user accessor, and replaces the cache's state + // in one assignment only once the build finishes. A refused load admits nothing. async pullAllBackingStores(): Promise { - const loaded: Array<{ registered: RegisteredDefinition; entry: CultCacheEnvelope; value: unknown }> = []; - const globalKeys = new Map(); + const next = emptyHydratedState(); for (const registration of this.#stores) { const entries = await registration.store.pullAll(); @@ -233,24 +257,13 @@ export class CultCache { ); } - if (registered.global) { - const globalKey = globalKeys.get(type); - if (globalKey !== undefined && globalKey !== entry.key) { - throw new Error(`CultCache global document type "${type}" has multiple persisted entries.`); - } - globalKeys.set(type, entry.key); - } - const payload = this.#cloneBytes(entry.payload); const value = registered.formatter.decode(payload); - loaded.push({ registered, entry: { ...entry, type, payload }, value }); + this.#applyHydratedEntry(next, registered, { ...entry, type, payload }, value); } } - this.#resetHydratedState(); - for (const { registered, entry, value } of loaded) { - this.#applyHydratedEntry(registered, entry, value); - } + this.#hydrated = next; } get( @@ -258,7 +271,7 @@ export class CultCache { key: string, ): CultCacheDocumentValue | undefined { this.#requireDefinition(definition); - const entry = this.#entries.get(this.#entryId(definition.type, key)); + const entry = this.#hydrated.entries.get(this.#entryId(definition.type, key)); return entry ? (entry.value as CultCacheDocumentValue) : undefined; } @@ -279,7 +292,7 @@ export class CultCache { key: string, ): CultCacheEnvelope | undefined { this.#requireDefinition(definition); - const entry = this.#entries.get(this.#entryId(definition.type, key)); + const entry = this.#hydrated.entries.get(this.#entryId(definition.type, key)); return entry ? this.#toEnvelope(entry) : undefined; } @@ -303,7 +316,7 @@ export class CultCache { throw new Error(`CultCache document type "${definition.type}" is not marked as global.`); } - const globalKey = this.#globalKeys.get(definition.type); + const globalKey = this.#hydrated.globalKeys.get(definition.type); return globalKey ? this.get(definition, globalKey) : undefined; } @@ -326,7 +339,7 @@ export class CultCache { throw new Error(`CultCache document type "${definition.type}" is not marked as global.`); } - const globalKey = this.#globalKeys.get(definition.type); + const globalKey = this.#hydrated.globalKeys.get(definition.type); return globalKey ? this.getEnvelope(definition, globalKey) : undefined; } @@ -335,14 +348,14 @@ export class CultCache { ): CultCacheDocumentValue[] { this.#requireDefinition(definition); const values: CultCacheDocumentValue[] = []; - const entryIds = this.#typeEntryIds.get(definition.type); + const entryIds = this.#hydrated.typeEntryIds.get(definition.type); if (!entryIds) { return values; } for (const entryId of entryIds) { - const entry = this.#entries.get(entryId); + const entry = this.#hydrated.entries.get(entryId); if (entry) { values.push(entry.value as CultCacheDocumentValue); } @@ -360,7 +373,7 @@ export class CultCache { return undefined; } - return this.#nameToKeyMaps.get(definition.type)?.get(name); + return this.#hydrated.nameToKeyMaps.get(definition.type)?.get(name); } getIdByName( @@ -384,7 +397,7 @@ export class CultCache { value: string, ): string | undefined { this.#requireDefinition(definition); - return this.#indexToKeyMaps.get(this.#indexId(definition.type, indexName))?.get(value); + return this.#hydrated.indexToKeyMaps.get(this.#indexId(definition.type, indexName))?.get(value); } getIdByIndex( @@ -424,14 +437,14 @@ export class CultCache { const home = this.#homeStore(definition.type); if (registered.global) { - const existingGlobalKey = this.#globalKeys.get(definition.type); + const existingGlobalKey = this.#hydrated.globalKeys.get(definition.type); if (existingGlobalKey && existingGlobalKey !== key) { await this.delete(definition, existingGlobalKey); } } await home?.push(entry); - this.#applyHydratedEntry(registered, entry, parsed); + this.#applyHydratedEntry(this.#hydrated, registered, entry, parsed); return parsed; } @@ -466,14 +479,14 @@ export class CultCache { const home = this.#homeStore(definition.type); if (registered.global) { - const existingGlobalKey = this.#globalKeys.get(definition.type); + const existingGlobalKey = this.#hydrated.globalKeys.get(definition.type); if (existingGlobalKey && existingGlobalKey !== entry.key) { await this.delete(definition, existingGlobalKey); } } await home?.push(entry); - this.#applyHydratedEntry(registered, entry, parsed); + this.#applyHydratedEntry(this.#hydrated, registered, entry, parsed); return parsed; } @@ -515,7 +528,7 @@ export class CultCache { key: string, ): Promise { const registered = this.#requireDefinition(definition); - const entry = this.#entries.get(this.#entryId(definition.type, key)); + const entry = this.#hydrated.entries.get(this.#entryId(definition.type, key)); if (!entry) { return false; } @@ -524,7 +537,7 @@ export class CultCache { const envelope = this.#toEnvelope(entry); await home?.delete(envelope); - this.#removeHydratedEntry(registered, entry); + this.#removeHydratedEntry(this.#hydrated, registered, entry); return true; } @@ -536,12 +549,12 @@ export class CultCache { throw new Error(`CultCache document type "${definition.type}" is not marked as global.`); } - const globalKey = this.#globalKeys.get(definition.type); + const globalKey = this.#hydrated.globalKeys.get(definition.type); return globalKey ? this.delete(definition, globalKey) : false; } snapshot(): CultCacheEnvelope[] { - return [...this.#entries.values()].map((entry) => this.#toEnvelope(entry)); + return [...this.#hydrated.entries.values()].map((entry) => this.#toEnvelope(entry)); } #createFormatter(definition: AnyCultCacheDocumentDefinition): CultCacheDocumentFormatter { @@ -711,21 +724,22 @@ export class CultCache { } #applyHydratedEntry( + state: HydratedState, registered: RegisteredDefinition, entry: CultCacheEnvelope, value: unknown, ): void { const entryId = this.#entryId(entry.type, entry.key); - const existing = this.#entries.get(entryId); + const existing = state.entries.get(entryId); if (existing) { - this.#removeHydratedEntry(registered, existing); + this.#removeHydratedEntry(state, registered, existing); } if (registered.global) { - const currentGlobalKey = this.#globalKeys.get(entry.type); - if (currentGlobalKey && currentGlobalKey !== entry.key) { + const currentGlobalKey = state.globalKeys.get(entry.type); + if (currentGlobalKey !== undefined && currentGlobalKey !== entry.key) { throw new Error( - `CultCache global document type "${entry.type}" already has a different key "${currentGlobalKey}".`, + `CultCache global document type "${entry.type}" has multiple persisted entries: "${currentGlobalKey}" and "${entry.key}".`, ); } } @@ -737,58 +751,60 @@ export class CultCache { catalogEntry: entry.catalogEntry ?? registered.catalogEntry, value, }; - this.#entries.set(entryId, hydrated); - this.#getOrCreateTypeEntrySet(entry.type).add(entryId); + state.entries.set(entryId, hydrated); + getOrCreate(state.typeEntryIds, entry.type, () => new Set()).add(entryId); if (registered.global) { - this.#globalKeys.set(entry.type, entry.key); + state.globalKeys.set(entry.type, entry.key); } if (registered.nameAccessor) { const name = registered.nameAccessor(value); if (name !== undefined) { - this.#getOrCreateNameMap(entry.type).set(name, entry.key); + getOrCreate(state.nameToKeyMaps, entry.type, () => new Map()).set(name, entry.key); } } for (const [indexName, accessor] of registered.indexAccessors) { const indexValue = accessor(value); if (indexValue !== undefined) { - this.#getOrCreateIndexMap(entry.type, indexName).set(indexValue, entry.key); + getOrCreate(state.indexToKeyMaps, this.#indexId(entry.type, indexName), () => new Map()) + .set(indexValue, entry.key); } } } #removeHydratedEntry( + state: HydratedState, registered: RegisteredDefinition, entry: HydratedEntry, ): void { const entryId = this.#entryId(entry.type, entry.key); - this.#entries.delete(entryId); + state.entries.delete(entryId); - const typeEntryIds = this.#typeEntryIds.get(entry.type); + const typeEntryIds = state.typeEntryIds.get(entry.type); if (typeEntryIds) { typeEntryIds.delete(entryId); if (typeEntryIds.size === 0) { - this.#typeEntryIds.delete(entry.type); + state.typeEntryIds.delete(entry.type); } } - if (registered.global && this.#globalKeys.get(entry.type) === entry.key) { - this.#globalKeys.delete(entry.type); + if (registered.global && state.globalKeys.get(entry.type) === entry.key) { + state.globalKeys.delete(entry.type); } if (registered.nameAccessor) { const name = registered.nameAccessor(entry.value); if (name !== undefined) { - this.#nameToKeyMaps.get(entry.type)?.delete(name); + state.nameToKeyMaps.get(entry.type)?.delete(name); } } for (const [indexName, accessor] of registered.indexAccessors) { const value = accessor(entry.value); if (value !== undefined) { - this.#indexToKeyMaps.get(this.#indexId(entry.type, indexName))?.delete(value); + state.indexToKeyMaps.get(this.#indexId(entry.type, indexName))?.delete(value); } } } @@ -801,49 +817,51 @@ export class CultCache { } } + // Lookups are built in a local map and assigned only after every accessor has run. #rebuildNameLookup(registered: RegisteredDefinition): void { - this.#nameToKeyMaps.delete(registered.definition.type); - + const type = registered.definition.type; if (!registered.nameAccessor) { + this.#hydrated.nameToKeyMaps.delete(type); return; } - const map = this.#getOrCreateNameMap(registered.definition.type); - for (const entry of this.#entriesForType(registered.definition.type)) { + const map = new Map(); + for (const entry of this.#entriesForType(type)) { const name = registered.nameAccessor(entry.value); if (name !== undefined) { map.set(name, entry.key); } } + this.#hydrated.nameToKeyMaps.set(type, map); } #rebuildIndexLookup(registered: RegisteredDefinition, indexName: string): void { const lookupId = this.#indexId(registered.definition.type, indexName); - this.#indexToKeyMaps.delete(lookupId); - const accessor = registered.indexAccessors.get(indexName); if (!accessor) { + this.#hydrated.indexToKeyMaps.delete(lookupId); return; } - const map = this.#getOrCreateIndexMap(registered.definition.type, indexName); + const map = new Map(); for (const entry of this.#entriesForType(registered.definition.type)) { const value = accessor(entry.value); if (value !== undefined) { map.set(value, entry.key); } } + this.#hydrated.indexToKeyMaps.set(lookupId, map); } #entriesForType(type: string): HydratedEntry[] { - const entryIds = this.#typeEntryIds.get(type); + const entryIds = this.#hydrated.typeEntryIds.get(type); if (!entryIds) { return []; } const entries: HydratedEntry[] = []; for (const entryId of entryIds) { - const entry = this.#entries.get(entryId); + const entry = this.#hydrated.entries.get(entryId); if (entry) { entries.push(entry); } @@ -852,34 +870,6 @@ export class CultCache { return entries; } - #getOrCreateTypeEntrySet(type: string): Set { - let set = this.#typeEntryIds.get(type); - if (!set) { - set = new Set(); - this.#typeEntryIds.set(type, set); - } - return set; - } - - #getOrCreateNameMap(type: string): Map { - let map = this.#nameToKeyMaps.get(type); - if (!map) { - map = new Map(); - this.#nameToKeyMaps.set(type, map); - } - return map; - } - - #getOrCreateIndexMap(type: string, indexName: string): Map { - const lookupId = this.#indexId(type, indexName); - let map = this.#indexToKeyMaps.get(lookupId); - if (!map) { - map = new Map(); - this.#indexToKeyMaps.set(lookupId, map); - } - return map; - } - #toEnvelope(entry: HydratedEntry): CultCacheEnvelope { return { key: entry.key, @@ -891,14 +881,6 @@ export class CultCache { }; } - #resetHydratedState(): void { - this.#entries.clear(); - this.#typeEntryIds.clear(); - this.#nameToKeyMaps.clear(); - this.#indexToKeyMaps.clear(); - this.#globalKeys.clear(); - } - #entryId(type: string, key: string): string { return `${type}::${key}`; } diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 04dfb159..7bad08d3 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -666,6 +666,53 @@ test("pullWithDuplicateGlobalLeavesCacheUnchanged", async () => { assert.deepEqual(cache.getRequiredGlobal(settingsDocument), { theme: "ash" }); }); +test("pullWithThrowingIndexAccessorLeavesCacheUnchanged", async () => { + const itemDocument = defineDocumentType({ + type: "item", + schema: z.object({ name: z.string(), category: z.string() }), + name: "name", + indexes: { + category: (item: { name: string; category: string }) => { + if (item.name === "Bomb") { + throw new Error("category accessor refuses Bomb"); + } + return item.category; + }, + }, + }); + const settingsDocument = defineDocumentType({ + type: "settings", + schema: z.object({ theme: z.string() }), + global: true, + }); + const tempDir = await mkdtemp(join(tmpdir(), "cultcache-throwing-accessor-")); + const storePath = join(tempDir, "generic.cc"); + const cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(itemDocument, settingsDocument)) + .withGenericStore(new SingleFileMessagePackBackingStore(storePath)) + .build(); + await cache.put(itemDocument, "potion", { name: "Potion", category: "consumable" }); + await cache.putGlobal(settingsDocument, { theme: "ash" }); + const before = cache.snapshot(); + + // A record whose index accessor throws lands in the store file behind the cache's back. + const potion = cache.getRequiredEnvelope(itemDocument, "potion"); + await new SingleFileMessagePackBackingStore(storePath).push({ + ...potion, + key: "bomb", + payload: encode({ name: "Bomb", category: "explosive" }), + }); + + await assert.rejects(cache.pullAllBackingStores(), /category accessor refuses Bomb/u); + assert.deepEqual(cache.snapshot(), before); + assert.deepEqual(cache.getRequired(itemDocument, "potion"), { name: "Potion", category: "consumable" }); + assert.equal(cache.get(itemDocument, "bomb"), undefined); + assert.equal(cache.getKeyByName(itemDocument, "Potion"), "potion"); + assert.equal(cache.getKeyByName(itemDocument, "Bomb"), undefined); + assert.equal(cache.getKeyByIndex(itemDocument, "category", "consumable"), "potion"); + assert.deepEqual(cache.getRequiredGlobal(settingsDocument), { theme: "ash" }); +}); + test("CultCache with zero stores writes and deletes in memory", async () => { const settingsDocument = defineDocumentType({ type: "settings", schema: z.object({ theme: z.string() }) }); const cache = CultCache.builder().withRegistry(defineDocumentRegistry(settingsDocument)).build();