diff --git a/packages/cultcache-ts/README.md b/packages/cultcache-ts/README.md index 2b85c6df..d1730958 100644 --- a/packages/cultcache-ts/README.md +++ b/packages/cultcache-ts/README.md @@ -180,11 +180,19 @@ const playerDocument = defineDocumentType({ You can also register them later if you feel like changing the machine after boot: ```ts -cache.registerNameLookup(playerDocument, "displayName"); -cache.registerIndex(playerDocument, "faction", "faction"); +await cache.registerNameLookup(playerDocument, "displayName"); +await cache.registerIndex(playerDocument, "faction", "faction"); ``` -Global documents are singleton-style per type: +Registration, `addBackingStore`, `pullAllBackingStores` and every write and +delete run one at a time through a per-cache queue and return promises; reads +are synchronous. A write validates the whole record (home store, global key, +name and index accessors) before any store is touched, so a refused write +changes neither store nor cache. Registering an accessor that throws on a held +record installs nothing. + +Global documents are singleton-style per type, always stored under +`CultCache.GLOBAL_KEY` (`__global__`); writing one under any other key throws: ```ts await cache.putGlobal(settingsDocument, { theme: "ash", retries: 3 }); diff --git a/packages/cultcache-ts/src/cult-cache.ts b/packages/cultcache-ts/src/cult-cache.ts index 74e96980..5d70a0d1 100644 --- a/packages/cultcache-ts/src/cult-cache.ts +++ b/packages/cultcache-ts/src/cult-cache.ts @@ -13,17 +13,27 @@ import type { CultCacheStoreRegistration, } from "./types"; +type Accessor = (value: unknown) => string | undefined; + type RegisteredDefinition = { readonly definition: AnyCultCacheDocumentDefinition; readonly global: boolean; readonly formatter: CultCacheDocumentFormatter; readonly catalogEntry: CultCacheSchemaCatalogEntry; - nameAccessor?: (value: unknown) => string | undefined; - readonly indexAccessors: Map string | undefined>; + nameAccessor?: Accessor; + indexAccessors: ReadonlyMap; +}; + +// The name and index values a record produced when it was admitted. Removing the record +// reads these and runs no accessor, so removal has no fallible step. +type EntryLookups = { + readonly name?: string; + readonly indexes: ReadonlyMap; }; type HydratedEntry = CultCacheEnvelope & { value: unknown; + lookups: EntryLookups; }; // Everything a pull replaces. Built whole, then swapped in with one assignment. @@ -35,6 +45,13 @@ type HydratedState = { readonly globalKeys: Map; }; +// Lookups for every held record of one type under candidate accessors, derived before anything is installed. +type DerivedTypeLookups = { + readonly entryLookups: Array<[HydratedEntry, EntryLookups]>; + readonly nameMap?: Map; + readonly indexMaps: Map>; +}; + function emptyHydratedState(): HydratedState { return { entries: new Map(), @@ -57,11 +74,17 @@ function getOrCreate(map: Map, key: K, create: () => V): V { type BackingStoreTypeReference = string | AnyCultCacheDocumentDefinition; type AccessorInput = string | ((value: any) => unknown); +// Assigned in CultCache's static block. The builder's cache is unreachable until build(), so no +// queued operation can be in flight and registration and attach may run synchronously. +let registerNow!: (cache: CultCache, definition: AnyCultCacheDocumentDefinition) => void; +let attachNow!: (cache: CultCache, store: CacheBackingStore, types: BackingStoreTypeReference[]) => void; + export class CultCacheBuilder { readonly #cache = new CultCache(); + #built = false; withDocumentType(definition: TDefinition): this { - this.#cache.registerDocumentType(definition); + registerNow(this.#unbuilt(), definition); return this; } @@ -70,7 +93,10 @@ export class CultCacheBuilder { | CultCacheDocumentRegistry | Iterable, ): this { - this.#cache.registerRegistry(registry); + const cache = this.#unbuilt(); + for (const definition of registryDefinitions(registry)) { + registerNow(cache, definition); + } return this; } @@ -78,192 +104,166 @@ export class CultCacheBuilder { store: CacheBackingStore, ...types: BackingStoreTypeReference[] ): this { - this.#cache.addBackingStore(store, ...types); + attachNow(this.#unbuilt(), store, types); return this; } withGenericStore(store: CacheBackingStore): this { - this.#cache.addGenericBackingStore(store); + attachNow(this.#unbuilt(), store, []); return this; } build(): CultCache { + this.#built = true; + return this.#cache; + } + + #unbuilt(): CultCache { + if (this.#built) { + throw new Error("CultCacheBuilder has already built its cache; configure the cache through its own methods."); + } return this.#cache; } } +function registryDefinitions( + registry: CultCacheDocumentRegistry | Iterable, +): Iterable { + return Symbol.iterator in Object(registry) && !("definitions" in Object(registry)) + ? registry as Iterable + : (registry as CultCacheDocumentRegistry).definitions; +} + export class CultCache { static readonly GLOBAL_KEY = "__global__"; + static { + registerNow = (cache, definition) => cache.#registerNow(definition); + attachNow = (cache, store, types) => cache.#attachNow(store, types); + } + readonly #definitions = new Map(); readonly #schemaIdDefinitions = new Map(); readonly #schemaNameDefinitions = new Map(); #hydrated = emptyHydratedState(); readonly #stores: CultCacheStoreRegistration[] = []; + #tail: Promise = Promise.resolve(); static builder(): CultCacheBuilder { return new CultCacheBuilder(); } + // Attach, pull, registration and every write and delete run one at a time through this chain, + // so validation, the store call and applying to the cache never interleave with another + // mutation of this cache. Reads are not queued. Observer delivery is not scheduled here. + #serial(operation: () => T | Promise): Promise { + const run = this.#tail.then(operation); + this.#tail = run.catch(() => undefined); + return run; + } + registerDocumentType( definition: TDefinition, - ): TDefinition { - const existing = this.#definitions.get(definition.type); - if (existing && existing.definition !== definition) { - throw new Error(`CultCache already has a different definition registered for type "${definition.type}".`); - } - - const registered = existing ?? { - definition, - global: definition.global === true, - formatter: this.#createFormatter(definition), - catalogEntry: this.#createCatalogEntry(definition), - nameAccessor: undefined, - indexAccessors: new Map string | undefined>(), - }; - - registered.nameAccessor = definition.name - ? this.#compileAccessor(definition.type, "name", definition.name) - : undefined; - registered.indexAccessors.clear(); - for (const [indexName, accessor] of this.#enumerateDefinitionIndexes(definition)) { - registered.indexAccessors.set( - indexName, - this.#compileAccessor(definition.type, `index "${indexName}"`, accessor), - ); - } - - this.#definitions.set(definition.type, registered); - for (const schemaId of registered.catalogEntry.compatibleSchemaIds ?? [registered.catalogEntry.schemaId]) { - const existingBySchema = this.#schemaIdDefinitions.get(schemaId); - if (existingBySchema && existingBySchema !== registered) { - throw new Error(`CultCache schema id "${schemaId}" is already registered for type "${existingBySchema.definition.type}".`); - } - 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; + ): Promise { + return this.#serial(() => { + this.#registerNow(definition); + return definition; + }); } registerRegistry( registry: | CultCacheDocumentRegistry | Iterable, - ): this { - const definitions = Symbol.iterator in Object(registry) && !("definitions" in Object(registry)) - ? registry as Iterable - : (registry as CultCacheDocumentRegistry).definitions; - - for (const definition of definitions) { - this.registerDocumentType(definition); - } - - return this; + ): Promise { + return this.#serial(() => { + for (const definition of registryDefinitions(registry)) { + this.#registerNow(definition); + } + }); } registerNameLookup( definition: TDefinition, accessor: CultCacheDocumentAccessor>, - ): TDefinition { - const registered = this.#requireDefinition(definition); - registered.nameAccessor = this.#compileAccessor(definition.type, "name", accessor); - this.#rebuildNameLookup(registered); - return definition; + ): Promise { + return this.#serial(() => { + const registered = this.#requireDefinition(definition); + const nameAccessor = this.#compileAccessor(definition.type, "name", accessor); + const derived = this.#deriveTypeLookups(definition.type, nameAccessor, registered.indexAccessors); + this.#installAccessors(registered, nameAccessor, registered.indexAccessors, derived); + return definition; + }); } registerIndex( definition: TDefinition, indexName: string, accessor: CultCacheDocumentAccessor>, - ): TDefinition { - if (!indexName || indexName.trim().length === 0) { - throw new Error(`CultCache index names for type "${definition.type}" must be non-empty.`); - } + ): Promise { + return this.#serial(() => { + if (!indexName || indexName.trim().length === 0) { + throw new Error(`CultCache index names for type "${definition.type}" must be non-empty.`); + } - const registered = this.#requireDefinition(definition); - registered.indexAccessors.set( - indexName, - this.#compileAccessor(definition.type, `index "${indexName}"`, accessor), - ); - this.#rebuildIndexLookup(registered, indexName); - return definition; + const registered = this.#requireDefinition(definition); + const indexAccessors = new Map(registered.indexAccessors); + indexAccessors.set(indexName, this.#compileAccessor(definition.type, `index "${indexName}"`, accessor)); + const derived = this.#deriveTypeLookups(definition.type, registered.nameAccessor, indexAccessors); + this.#installAccessors(registered, registered.nameAccessor, indexAccessors, derived); + return definition; + }); } addBackingStore( store: CacheBackingStore, ...types: BackingStoreTypeReference[] - ): void { - const claimed = [...new Set(types.map((value) => (typeof value === "string" ? value : value.type)))]; - if (claimed.length === 0 && this.#stores.some((registration) => registration.types.length === 0)) { - throw new Error("Backing store would be a second generic store; name the types it is home to."); - } - for (const type of claimed) { - if (this.#stores.some((registration) => registration.types.includes(type))) { - throw new Error(`CultCache type "${type}" is already routed to another backing store; it cannot also route to this one.`); - } - } - const candidate: CultCacheStoreRegistration = { store, types: claimed }; - for (const type of this.#hydrated.typeEntryIds.keys()) { - const before = this.#resolveRegistration(type); - this.#stores.push(candidate); - const after = this.#resolveRegistration(type); - this.#stores.pop(); - if (before !== after) { - throw new Error( - `Attaching this backing store would move "${type}" from ${this.#describe(before)} to ${this.#describe(after)}; ` + - "attach routed stores before the generic store.", - ); - } - } - this.#stores.push(candidate); + ): Promise { + return this.#serial(() => this.#attachNow(store, types)); } - addGenericBackingStore(store: CacheBackingStore): void { - this.addBackingStore(store); + addGenericBackingStore(store: CacheBackingStore): Promise { + return this.addBackingStore(store); } // 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 next = emptyHydratedState(); + pullAllBackingStores(): Promise { + return this.#serial(async () => { + const next = emptyHydratedState(); - for (const registration of this.#stores) { - const entries = await registration.store.pullAll(); + for (const registration of this.#stores) { + const entries = await registration.store.pullAll(); - for (const entry of entries) { - const registered = this.#resolveDefinitionForEnvelope(entry); - if (!registered) { - throw new Error( - entry.schemaId - ? `No schema is registered for persisted schema id "${entry.schemaId}".` - : `No schema is registered for persisted document type "${entry.type}".`, - ); + for (const entry of entries) { + const registered = this.#resolveDefinitionForEnvelope(entry); + if (!registered) { + throw new Error( + entry.schemaId + ? `No schema is registered for persisted schema id "${entry.schemaId}".` + : `No schema is registered for persisted document type "${entry.type}".`, + ); + } + + const type = registered.definition.type; + const home = this.#resolveRegistration(type); + if (home?.store !== registration.store) { + throw new Error( + `CultCache "${type}" record "${entry.key}" was loaded from ${this.#describe(registration)}, ` + + `but its home is ${this.#describe(home)}.`, + ); + } + + const payload = this.#cloneBytes(entry.payload); + const value = registered.formatter.decode(payload); + const hydrated = this.#admitEntry(next, registered, { ...entry, type, payload }, value); + this.#installEntry(next, registered, hydrated); } - - const type = registered.definition.type; - const home = this.#resolveRegistration(type); - if (home?.store !== registration.store) { - throw new Error( - `CultCache "${type}" record "${entry.key}" was loaded from ${this.#describe(registration)}, ` + - `but its home is ${this.#describe(home)}.`, - ); - } - - const payload = this.#cloneBytes(entry.payload); - const value = registered.formatter.decode(payload); - this.#applyHydratedEntry(next, registered, { ...entry, type, payload }, value); } - } - this.#hydrated = next; + this.#hydrated = next; + }); } get( @@ -311,11 +311,7 @@ export class CultCache { getGlobal( definition: TDefinition, ): CultCacheDocumentValue | undefined { - const registered = this.#requireDefinition(definition); - if (!registered.global) { - throw new Error(`CultCache document type "${definition.type}" is not marked as global.`); - } - + this.#requireGlobal(definition); const globalKey = this.#hydrated.globalKeys.get(definition.type); return globalKey ? this.get(definition, globalKey) : undefined; } @@ -334,11 +330,7 @@ export class CultCache { getGlobalEnvelope( definition: TDefinition, ): CultCacheEnvelope | undefined { - const registered = this.#requireDefinition(definition); - if (!registered.global) { - throw new Error(`CultCache document type "${definition.type}" is not marked as global.`); - } - + this.#requireGlobal(definition); const globalKey = this.#hydrated.globalKeys.get(definition.type); return globalKey ? this.getEnvelope(definition, globalKey) : undefined; } @@ -347,21 +339,7 @@ export class CultCache { definition: TDefinition, ): CultCacheDocumentValue[] { this.#requireDefinition(definition); - const values: CultCacheDocumentValue[] = []; - const entryIds = this.#hydrated.typeEntryIds.get(definition.type); - - if (!entryIds) { - return values; - } - - for (const entryId of entryIds) { - const entry = this.#hydrated.entries.get(entryId); - if (entry) { - values.push(entry.value as CultCacheDocumentValue); - } - } - - return values; + return this.#entriesForType(definition.type).map((entry) => entry.value as CultCacheDocumentValue); } getKeyByName( @@ -417,12 +395,180 @@ export class CultCache { return key ? this.get(definition, key) : undefined; } - async put( + put( + definition: TDefinition, + key: string, + value: CultCacheDocumentValue, + ): Promise> { + return this.#serial(() => this.#putNow(definition, key, value)); + } + + putEnvelope( + definition: TDefinition, + envelope: CultCacheEnvelope, + ): Promise> { + return this.#serial(async () => { + const registered = this.#requireDefinition(definition); + if (envelope.type !== definition.type) { + throw new Error( + `CultCache envelope type "${envelope.type}" does not match definition "${definition.type}".`, + ); + } + if (!envelope.key || envelope.key.trim().length === 0) { + throw new Error(`CultCache envelope key for type "${definition.type}" must be non-empty.`); + } + if (!envelope.storedAt || envelope.storedAt.trim().length === 0) { + throw new Error(`CultCache envelope storedAt for type "${definition.type}" must be non-empty.`); + } + this.#requireGlobalKey(registered, envelope.key); + + const payload = this.#cloneBytes(envelope.payload); + const parsed = registered.formatter.decode(payload) as CultCacheDocumentValue; + const entry: CultCacheEnvelope = { + key: envelope.key, + type: envelope.type, + payload, + storedAt: envelope.storedAt, + schemaId: envelope.schemaId ?? registered.catalogEntry.schemaId, + catalogEntry: envelope.catalogEntry ?? registered.catalogEntry, + }; + await this.#writeNow(registered, entry, parsed); + return parsed; + }); + } + + putGlobal( + definition: TDefinition, + value: CultCacheDocumentValue, + ): Promise> { + return this.#serial(() => { + this.#requireGlobal(definition); + return this.#putNow(definition, CultCache.GLOBAL_KEY, value); + }); + } + + update( + definition: TDefinition, + key: string, + updater: ( + current: CultCacheDocumentValue | undefined, + ) => CultCacheDocumentValue, + ): Promise> { + return this.#serial(() => this.#putNow(definition, key, updater(this.get(definition, key)))); + } + + updateGlobal( + definition: TDefinition, + updater: ( + current: CultCacheDocumentValue | undefined, + ) => CultCacheDocumentValue, + ): Promise> { + return this.#serial(() => + this.#putNow(definition, CultCache.GLOBAL_KEY, updater(this.getGlobal(definition))), + ); + } + + delete( + definition: TDefinition, + key: string, + ): Promise { + return this.#serial(() => this.#deleteNow(definition, key)); + } + + deleteGlobal( + definition: TDefinition, + ): Promise { + return this.#serial(async () => { + this.#requireGlobal(definition); + const globalKey = this.#hydrated.globalKeys.get(definition.type); + return globalKey ? this.#deleteNow(definition, globalKey) : false; + }); + } + + snapshot(): CultCacheEnvelope[] { + return [...this.#hydrated.entries.values()].map((entry) => this.#toEnvelope(entry)); + } + + #registerNow(definition: AnyCultCacheDocumentDefinition): void { + const existing = this.#definitions.get(definition.type); + if (existing && existing.definition !== definition) { + throw new Error(`CultCache already has a different definition registered for type "${definition.type}".`); + } + + const registered: RegisteredDefinition = existing ?? { + definition, + global: definition.global === true, + formatter: this.#createFormatter(definition), + catalogEntry: this.#createCatalogEntry(definition), + nameAccessor: undefined, + indexAccessors: new Map(), + }; + + const nameAccessor = definition.name + ? this.#compileAccessor(definition.type, "name", definition.name) + : undefined; + const indexAccessors = new Map(); + for (const [indexName, accessor] of this.#enumerateDefinitionIndexes(definition)) { + indexAccessors.set(indexName, this.#compileAccessor(definition.type, `index "${indexName}"`, accessor)); + } + + const schemaIds = registered.catalogEntry.compatibleSchemaIds ?? [registered.catalogEntry.schemaId]; + for (const schemaId of schemaIds) { + const existingBySchema = this.#schemaIdDefinitions.get(schemaId); + if (existingBySchema && existingBySchema !== registered) { + throw new Error(`CultCache schema id "${schemaId}" is already registered for type "${existingBySchema.definition.type}".`); + } + } + 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}".`, + ); + } + const derived = this.#deriveTypeLookups(definition.type, nameAccessor, indexAccessors); + + // Every check and accessor has run; install. + this.#definitions.set(definition.type, registered); + for (const schemaId of schemaIds) { + this.#schemaIdDefinitions.set(schemaId, registered); + } + this.#schemaNameDefinitions.set(registered.catalogEntry.schemaName, registered); + this.#installAccessors(registered, nameAccessor, indexAccessors, derived); + } + + #attachNow(store: CacheBackingStore, types: BackingStoreTypeReference[]): void { + const claimed = [...new Set(types.map((value) => (typeof value === "string" ? value : value.type)))]; + if (claimed.length === 0 && this.#stores.some((registration) => registration.types.length === 0)) { + throw new Error("Backing store would be a second generic store; name the types it is home to."); + } + for (const type of claimed) { + if (this.#stores.some((registration) => registration.types.includes(type))) { + throw new Error(`CultCache type "${type}" is already routed to another backing store; it cannot also route to this one.`); + } + } + const candidate: CultCacheStoreRegistration = { store, types: claimed }; + for (const type of this.#hydrated.typeEntryIds.keys()) { + const before = this.#resolveRegistration(type); + this.#stores.push(candidate); + const after = this.#resolveRegistration(type); + this.#stores.pop(); + if (before !== after) { + throw new Error( + `Attaching this backing store would move "${type}" from ${this.#describe(before)} to ${this.#describe(after)}; ` + + "attach routed stores before the generic store.", + ); + } + } + this.#stores.push(candidate); + } + + async #putNow( definition: TDefinition, key: string, value: CultCacheDocumentValue, ): Promise> { const registered = this.#requireDefinition(definition); + this.#requireGlobalKey(registered, key); const payload = registered.formatter.encode(value); const parsed = registered.formatter.decode(payload) as CultCacheDocumentValue; const entry: CultCacheEnvelope = { @@ -433,97 +579,20 @@ export class CultCache { schemaId: registered.catalogEntry.schemaId, catalogEntry: registered.catalogEntry, }; - - const home = this.#homeStore(definition.type); - - if (registered.global) { - const existingGlobalKey = this.#hydrated.globalKeys.get(definition.type); - if (existingGlobalKey && existingGlobalKey !== key) { - await this.delete(definition, existingGlobalKey); - } - } - - await home?.push(entry); - this.#applyHydratedEntry(this.#hydrated, registered, entry, parsed); + await this.#writeNow(registered, entry, parsed); return parsed; } - async putEnvelope( - definition: TDefinition, - envelope: CultCacheEnvelope, - ): Promise> { - const registered = this.#requireDefinition(definition); - if (envelope.type !== definition.type) { - throw new Error( - `CultCache envelope type "${envelope.type}" does not match definition "${definition.type}".`, - ); - } - if (!envelope.key || envelope.key.trim().length === 0) { - throw new Error(`CultCache envelope key for type "${definition.type}" must be non-empty.`); - } - if (!envelope.storedAt || envelope.storedAt.trim().length === 0) { - throw new Error(`CultCache envelope storedAt for type "${definition.type}" must be non-empty.`); - } - - const payload = this.#cloneBytes(envelope.payload); - const parsed = registered.formatter.decode(payload) as CultCacheDocumentValue; - const entry: CultCacheEnvelope = { - key: envelope.key, - type: envelope.type, - payload, - storedAt: envelope.storedAt, - schemaId: envelope.schemaId ?? registered.catalogEntry.schemaId, - catalogEntry: envelope.catalogEntry ?? registered.catalogEntry, - }; - - const home = this.#homeStore(definition.type); - - if (registered.global) { - const existingGlobalKey = this.#hydrated.globalKeys.get(definition.type); - if (existingGlobalKey && existingGlobalKey !== entry.key) { - await this.delete(definition, existingGlobalKey); - } - } - + // Validates the whole write (home, global singleton, name and index accessors) before the store + // is touched; once the store accepts, applying to the cache has no fallible step. + async #writeNow(registered: RegisteredDefinition, entry: CultCacheEnvelope, value: unknown): Promise { + const home = this.#homeStore(entry.type); + const hydrated = this.#admitEntry(this.#hydrated, registered, entry, value); await home?.push(entry); - this.#applyHydratedEntry(this.#hydrated, registered, entry, parsed); - return parsed; + this.#installEntry(this.#hydrated, registered, hydrated); } - async putGlobal( - definition: TDefinition, - value: CultCacheDocumentValue, - ): Promise> { - const registered = this.#requireDefinition(definition); - if (!registered.global) { - throw new Error(`CultCache document type "${definition.type}" is not marked as global.`); - } - - return this.put(definition, CultCache.GLOBAL_KEY, value); - } - - async update( - definition: TDefinition, - key: string, - updater: ( - current: CultCacheDocumentValue | undefined, - ) => CultCacheDocumentValue, - ): Promise> { - const current = this.get(definition, key); - return this.put(definition, key, updater(current)); - } - - async updateGlobal( - definition: TDefinition, - updater: ( - current: CultCacheDocumentValue | undefined, - ) => CultCacheDocumentValue, - ): Promise> { - const current = this.getGlobal(definition); - return this.putGlobal(definition, updater(current)); - } - - async delete( + async #deleteNow( definition: TDefinition, key: string, ): Promise { @@ -534,27 +603,26 @@ export class CultCache { } const home = this.#homeStore(definition.type); - - const envelope = this.#toEnvelope(entry); - await home?.delete(envelope); - this.#removeHydratedEntry(this.#hydrated, registered, entry); + await home?.delete(this.#toEnvelope(entry)); + this.#removeEntry(this.#hydrated, registered, entry); return true; } - async deleteGlobal( - definition: TDefinition, - ): Promise { + #requireGlobal(definition: AnyCultCacheDocumentDefinition): RegisteredDefinition { const registered = this.#requireDefinition(definition); if (!registered.global) { throw new Error(`CultCache document type "${definition.type}" is not marked as global.`); } - - const globalKey = this.#hydrated.globalKeys.get(definition.type); - return globalKey ? this.delete(definition, globalKey) : false; + return registered; } - snapshot(): CultCacheEnvelope[] { - return [...this.#hydrated.entries.values()].map((entry) => this.#toEnvelope(entry)); + // A global is only ever stored under GLOBAL_KEY; any other key is refused before a store is touched. + #requireGlobalKey(registered: RegisteredDefinition, key: string): void { + if (registered.global && key !== CultCache.GLOBAL_KEY) { + throw new Error( + `CultCache global document type "${registered.definition.type}" must use key "${CultCache.GLOBAL_KEY}", not "${key}".`, + ); + } } #createFormatter(definition: AnyCultCacheDocumentDefinition): CultCacheDocumentFormatter { @@ -639,7 +707,7 @@ export class CultCache { type: string, label: string, accessor: AccessorInput, - ): (value: unknown) => string | undefined { + ): Accessor { if (typeof accessor === "function") { return (value: unknown) => this.#normalizeIndexValue(type, label, accessor(value)); } @@ -723,18 +791,29 @@ export class CultCache { : `the store routed to ${registration.types.join(", ")}`; } - #applyHydratedEntry( + #lookupsFor( + nameAccessor: Accessor | undefined, + indexAccessors: ReadonlyMap, + value: unknown, + ): EntryLookups { + const name = nameAccessor?.(value); + const indexes = new Map(); + for (const [indexName, accessor] of indexAccessors) { + const indexValue = accessor(value); + if (indexValue !== undefined) { + indexes.set(indexName, indexValue); + } + } + return { name, indexes }; + } + + // Fallible half of admission: the global singleton check and every user accessor. Touches no state. + #admitEntry( state: HydratedState, registered: RegisteredDefinition, entry: CultCacheEnvelope, value: unknown, - ): void { - const entryId = this.#entryId(entry.type, entry.key); - const existing = state.entries.get(entryId); - if (existing) { - this.#removeHydratedEntry(state, registered, existing); - } - + ): HydratedEntry { if (registered.global) { const currentGlobalKey = state.globalKeys.get(entry.type); if (currentGlobalKey !== undefined && currentGlobalKey !== entry.key) { @@ -744,37 +823,40 @@ export class CultCache { } } - const hydrated: HydratedEntry = { + return { ...entry, payload: this.#cloneBytes(entry.payload), schemaId: entry.schemaId ?? registered.catalogEntry.schemaId, catalogEntry: entry.catalogEntry ?? registered.catalogEntry, value, + lookups: this.#lookupsFor(registered.nameAccessor, registered.indexAccessors, value), }; + } + + // Infallible half of admission: replaces any held record at the same key and installs the admitted one. + #installEntry(state: HydratedState, registered: RegisteredDefinition, hydrated: HydratedEntry): void { + const entryId = this.#entryId(hydrated.type, hydrated.key); + const existing = state.entries.get(entryId); + if (existing) { + this.#removeEntry(state, registered, existing); + } + state.entries.set(entryId, hydrated); - getOrCreate(state.typeEntryIds, entry.type, () => new Set()).add(entryId); - + getOrCreate(state.typeEntryIds, hydrated.type, () => new Set()).add(entryId); if (registered.global) { - state.globalKeys.set(entry.type, entry.key); + state.globalKeys.set(hydrated.type, hydrated.key); } - - if (registered.nameAccessor) { - const name = registered.nameAccessor(value); - if (name !== undefined) { - getOrCreate(state.nameToKeyMaps, entry.type, () => new Map()).set(name, entry.key); - } + if (hydrated.lookups.name !== undefined) { + getOrCreate(state.nameToKeyMaps, hydrated.type, () => new Map()).set(hydrated.lookups.name, hydrated.key); } - - for (const [indexName, accessor] of registered.indexAccessors) { - const indexValue = accessor(value); - if (indexValue !== undefined) { - getOrCreate(state.indexToKeyMaps, this.#indexId(entry.type, indexName), () => new Map()) - .set(indexValue, entry.key); - } + for (const [indexName, indexValue] of hydrated.lookups.indexes) { + getOrCreate(state.indexToKeyMaps, this.#indexId(hydrated.type, indexName), () => new Map()) + .set(indexValue, hydrated.key); } } - #removeHydratedEntry( + // Reads the lookups stored at admission; runs no accessor. + #removeEntry( state: HydratedState, registered: RegisteredDefinition, entry: HydratedEntry, @@ -794,63 +876,72 @@ export class CultCache { state.globalKeys.delete(entry.type); } - if (registered.nameAccessor) { - const name = registered.nameAccessor(entry.value); - if (name !== undefined) { - state.nameToKeyMaps.get(entry.type)?.delete(name); + if (entry.lookups.name !== undefined) { + const names = state.nameToKeyMaps.get(entry.type); + if (names?.get(entry.lookups.name) === entry.key) { + names.delete(entry.lookups.name); } } - - for (const [indexName, accessor] of registered.indexAccessors) { - const value = accessor(entry.value); - if (value !== undefined) { - state.indexToKeyMaps.get(this.#indexId(entry.type, indexName))?.delete(value); + for (const [indexName, indexValue] of entry.lookups.indexes) { + const index = state.indexToKeyMaps.get(this.#indexId(entry.type, indexName)); + if (index?.get(indexValue) === entry.key) { + index.delete(indexValue); } } } - #rebuildDefinitionLookups(registered: RegisteredDefinition): void { - this.#rebuildNameLookup(registered); - - for (const indexName of registered.indexAccessors.keys()) { - this.#rebuildIndexLookup(registered, indexName); - } - } - - // Lookups are built in a local map and assigned only after every accessor has run. - #rebuildNameLookup(registered: RegisteredDefinition): void { - const type = registered.definition.type; - if (!registered.nameAccessor) { - this.#hydrated.nameToKeyMaps.delete(type); - return; + // Runs the candidate accessors over every held record of the type into local maps. + #deriveTypeLookups( + type: string, + nameAccessor: Accessor | undefined, + indexAccessors: ReadonlyMap, + ): DerivedTypeLookups { + const entryLookups: Array<[HydratedEntry, EntryLookups]> = []; + const nameMap = nameAccessor ? new Map() : undefined; + const indexMaps = new Map>(); + for (const indexName of indexAccessors.keys()) { + indexMaps.set(indexName, new Map()); } - 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); + const lookups = this.#lookupsFor(nameAccessor, indexAccessors, entry.value); + entryLookups.push([entry, lookups]); + if (lookups.name !== undefined) { + nameMap?.set(lookups.name, entry.key); + } + for (const [indexName, indexValue] of lookups.indexes) { + indexMaps.get(indexName)?.set(indexValue, entry.key); } } - this.#hydrated.nameToKeyMaps.set(type, map); + + return { entryLookups, nameMap, indexMaps }; } - #rebuildIndexLookup(registered: RegisteredDefinition, indexName: string): void { - const lookupId = this.#indexId(registered.definition.type, indexName); - const accessor = registered.indexAccessors.get(indexName); - if (!accessor) { - this.#hydrated.indexToKeyMaps.delete(lookupId); - return; + // Installs accessors and their derived lookups together. Nothing here can throw. + #installAccessors( + registered: RegisteredDefinition, + nameAccessor: Accessor | undefined, + indexAccessors: ReadonlyMap, + derived: DerivedTypeLookups, + ): void { + const type = registered.definition.type; + for (const indexName of registered.indexAccessors.keys()) { + this.#hydrated.indexToKeyMaps.delete(this.#indexId(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); - } + registered.nameAccessor = nameAccessor; + registered.indexAccessors = indexAccessors; + for (const [entry, lookups] of derived.entryLookups) { + entry.lookups = lookups; + } + if (derived.nameMap) { + this.#hydrated.nameToKeyMaps.set(type, derived.nameMap); + } else { + this.#hydrated.nameToKeyMaps.delete(type); + } + for (const [indexName, map] of derived.indexMaps) { + this.#hydrated.indexToKeyMaps.set(this.#indexId(type, indexName), map); } - this.#hydrated.indexToKeyMaps.set(lookupId, map); } #entriesForType(type: string): HydratedEntry[] { diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 7bad08d3..991849db 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -163,8 +163,8 @@ test("CultCache can register name and index lookups after entries already exist" body: "world", }); - cache.registerNameLookup(noteDocument, "title"); - cache.registerIndex(noteDocument, "author", "author"); + await cache.registerNameLookup(noteDocument, "title"); + await cache.registerIndex(noteDocument, "author", "author"); assert.equal(cache.getIdByName(noteDocument, "Hello"), "note:hello"); assert.equal(cache.getIdByIndex(noteDocument, "author", "ari"), "note:hello"); @@ -533,24 +533,24 @@ test("SingleFileMessagePackBackingStore heals legacy envelopes whose payload was assert.ok(records[0]?.[3] instanceof Uint8Array); }); -test("CultCache rejects a second generic backing store", () => { +test("CultCache rejects a second generic backing store", async () => { const cache = new CultCache(); - cache.addGenericBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-a.cc"))); - assert.throws( - () => cache.addGenericBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-b.cc"))), + await cache.addGenericBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-a.cc"))); + await assert.rejects( + cache.addGenericBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-b.cc"))), /second generic store/u, ); }); -test("CultCache rejects a type registered to two stores", () => { +test("CultCache rejects a type registered to two stores", async () => { const cache = new CultCache(); - cache.addBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-a.cc")), "item"); - assert.throws( - () => cache.addBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-b.cc")), "settings", "item"), + await cache.addBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-a.cc")), "item"); + await assert.rejects( + cache.addBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-b.cc")), "settings", "item"), /"item" is already routed/u, ); // A refused registration claims nothing, so "settings" is still free. - cache.addBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-c.cc")), "settings"); + await cache.addBackingStore(new SingleFileMessagePackBackingStore(join(tmpdir(), "unused-c.cc")), "settings"); }); test("CultCache routes each type to its home store", async () => { @@ -600,8 +600,8 @@ test("CultCache refuses an attach that would move a held record's home", async ( .build(); await cache.put(settingsDocument, "app", { theme: "v1-generic" }); - assert.throws( - () => cache.addBackingStore(new SingleFileMessagePackBackingStore(settingsPath), settingsDocument), + await assert.rejects( + async () => cache.addBackingStore(new SingleFileMessagePackBackingStore(settingsPath), settingsDocument), /would move "settings" from the generic store to the store routed to settings/u, ); // Nothing attached: the next write still lands in the generic store, and the refused store stays untouched. @@ -629,14 +629,201 @@ test("CultCache refuses at load a record delivered by a store that is not its ho .withBackingStore(new SingleFileMessagePackBackingStore(settingsPath), settingsDocument) .withGenericStore(new SingleFileMessagePackBackingStore(genericPath)) .build(); + // The cache already holds a record, in its home store, before the refused load. + await cache.put(settingsDocument, "held", { theme: "home" }); + const before = cache.snapshot(); + const settingsBytes = await readFile(settingsPath); await assert.rejects( cache.pullAllBackingStores(), /"settings" record "app" was loaded from the generic store, but its home is the store routed to settings/u, ); assert.equal(cache.get(settingsDocument, "app"), undefined); - assert.deepEqual(cache.snapshot(), []); + assert.deepEqual(cache.snapshot(), before); + assert.deepEqual(cache.getRequired(settingsDocument, "held"), { theme: "home" }); assert.deepEqual(await readFile(genericPath), genericBytes); + assert.deepEqual(await readFile(settingsPath), settingsBytes); +}); + +async function storeState(path: string): Promise { + if (!existsSync(path)) { + return []; + } + return (await new SingleFileMessagePackBackingStore(path).pullAll()) + .map((entry) => `${entry.type}:${entry.key}:${Buffer.from(entry.payload).toString("hex")}`) + .sort(); +} + +test("putWithThrowingAccessorChangesNeitherStoreNorCache", async () => { + const itemDocument = defineDocumentType({ + type: "item", + schema: z.object({ n: z.string() }), + name: (value: { n: string }) => { + if (value.n === "bad") { + throw new Error("accessor refuses bad"); + } + return value.n; + }, + }); + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-put-accessor-")), "generic.cc"); + const cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(itemDocument)) + .withGenericStore(new SingleFileMessagePackBackingStore(storePath)) + .build(); + await cache.put(itemDocument, "k", { n: "good" }); + const storeBefore = await storeState(storePath); + const cacheBefore = cache.snapshot(); + + await assert.rejects(cache.put(itemDocument, "k", { n: "bad" }), /accessor refuses bad/u); + assert.deepEqual(await storeState(storePath), storeBefore); + assert.deepEqual(cache.snapshot(), cacheBefore); + assert.deepEqual(cache.getRequired(itemDocument, "k"), { n: "good" }); + assert.equal(cache.getKeyByName(itemDocument, "good"), "k"); +}); + +test("overwriteUnderRegisteredAccessorThatNowThrowsChangesNeitherStoreNorCache", async () => { + const itemDocument = defineDocumentType({ type: "item", schema: z.object({ c: z.string() }) }); + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-overwrite-accessor-")), "generic.cc"); + const cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(itemDocument)) + .withGenericStore(new SingleFileMessagePackBackingStore(storePath)) + .build(); + await cache.put(itemDocument, "k", { c: "old" }); + let refusing = false; + await cache.registerIndex(itemDocument, "category", (value: { c: string }) => { + if (refusing) { + throw new Error("accessor refuses now"); + } + return value.c; + }); + refusing = true; + const storeBefore = await storeState(storePath); + const cacheBefore = cache.snapshot(); + + await assert.rejects(cache.put(itemDocument, "k", { c: "new" }), /accessor refuses now/u); + assert.deepEqual(await storeState(storePath), storeBefore); + assert.deepEqual(cache.snapshot(), cacheBefore); + assert.equal(cache.getKeyByIndex(itemDocument, "category", "old"), "k"); + // Deleting runs no accessor: it uses the lookup values stored when the record was admitted. + assert.equal(await cache.delete(itemDocument, "k"), true); + assert.equal(cache.getKeyByIndex(itemDocument, "category", "old"), undefined); +}); + +test("registeringAccessorThatThrowsOnHeldRecordInstallsNothing", async () => { + const itemDocument = defineDocumentType({ type: "item", schema: z.object({ n: z.string() }) }); + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-register-accessor-")), "generic.cc"); + const cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(itemDocument)) + .withGenericStore(new SingleFileMessagePackBackingStore(storePath)) + .build(); + await cache.put(itemDocument, "old", { n: "bad" }); + await assert.rejects( + async () => cache.registerIndex(itemDocument, "i", (value: { n: string }) => { + if (value.n === "bad") { + throw new Error("index refuses bad"); + } + return value.n; + }), + /index refuses bad/u, + ); + await cache.put(itemDocument, "old", { n: "fine" }); + assert.deepEqual(cache.getRequired(itemDocument, "old"), { n: "fine" }); + assert.equal(cache.getKeyByIndex(itemDocument, "i", "fine"), undefined); +}); + +test("globalReplaceWithFailingStorePushKeepsOldGlobal", async () => { + const settingsDocument = defineDocumentType({ + type: "settings", + schema: z.object({ t: z.string() }), + global: true, + }); + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-global-push-")), "generic.cc"); + const inner = new SingleFileMessagePackBackingStore(storePath); + let failPush = false; + const store: CacheBackingStore = { + pullAll: () => inner.pullAll(), + delete: (entry) => inner.delete(entry), + pushAll: (entries, options) => inner.pushAll(entries, options), + push: async (entry) => { + if (failPush) { + throw new Error("disk full"); + } + return inner.push(entry); + }, + }; + const cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(settingsDocument)) + .withGenericStore(store) + .build(); + await cache.putGlobal(settingsDocument, { t: "base" }); + const storeBefore = await storeState(storePath); + const cacheBefore = cache.snapshot(); + failPush = true; + + await assert.rejects(cache.putGlobal(settingsDocument, { t: "new" }), /disk full/u); + await assert.rejects(cache.put(settingsDocument, "other", { t: "new" }), /must use key "__global__"/u); + assert.deepEqual(await storeState(storePath), storeBefore); + assert.deepEqual(cache.snapshot(), cacheBefore); + assert.deepEqual(cache.getRequiredGlobal(settingsDocument), { t: "base" }); +}); + +test("concurrentGlobalPutsCannotBothReachDisk", async () => { + const settingsDocument = defineDocumentType({ + type: "settings", + schema: z.object({ t: z.string() }), + global: true, + }); + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-global-race-")), "generic.cc"); + const build = () => CultCache.builder() + .withRegistry(defineDocumentRegistry(settingsDocument)) + .withGenericStore(new SingleFileMessagePackBackingStore(storePath)) + .build(); + const cache = build(); + await cache.putGlobal(settingsDocument, { t: "base" }); + + await Promise.allSettled([ + cache.put(settingsDocument, "a", { t: "a" }), + cache.put(settingsDocument, "b", { t: "b" }), + cache.putGlobal(settingsDocument, { t: "c" }), + cache.putGlobal(settingsDocument, { t: "d" }), + ]); + assert.deepEqual( + (await new SingleFileMessagePackBackingStore(storePath).pullAll()).map((entry) => entry.key), + [CultCache.GLOBAL_KEY], + ); + assert.deepEqual(cache.getRequiredGlobal(settingsDocument), { t: "d" }); + const reloaded = build(); + await reloaded.pullAllBackingStores(); + assert.deepEqual(reloaded.getRequiredGlobal(settingsDocument), { t: "d" }); +}); + +test("concurrentPutAndAttachCannotLandRecordInNonHomeStore", async () => { + const settingsDocument = defineDocumentType({ type: "settings", schema: z.object({ t: z.string() }) }); + const tempDir = await mkdtemp(join(tmpdir(), "cultcache-put-attach-race-")); + const genericPath = join(tempDir, "generic.cc"); + const settingsPath = join(tempDir, "settings.cc"); + const cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(settingsDocument)) + .withGenericStore(new SingleFileMessagePackBackingStore(genericPath)) + .build(); + + const pending = cache.put(settingsDocument, "app", { t: "x" }); + const attach = Promise.resolve().then(() => + cache.addBackingStore(new SingleFileMessagePackBackingStore(settingsPath), settingsDocument), + ); + await pending; + await assert.rejects(attach, /would move "settings" from the generic store/u); + + assert.deepEqual( + (await new SingleFileMessagePackBackingStore(genericPath).pullAll()).map((entry) => entry.key), + ["app"], + ); assert.equal(existsSync(settingsPath), false); + const reloaded = CultCache.builder() + .withRegistry(defineDocumentRegistry(settingsDocument)) + .withGenericStore(new SingleFileMessagePackBackingStore(genericPath)) + .build(); + await reloaded.pullAllBackingStores(); + assert.deepEqual(reloaded.getRequired(settingsDocument, "app"), { t: "x" }); }); test("pullWithDuplicateGlobalLeavesCacheUnchanged", async () => { diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 5cf26466..e9fbb8be 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -5677,7 +5677,7 @@ export class CultMesh { const documents = new CultNetDocumentRegistry(options.bindings); for (const definition of options.documents ?? []) { - cache.registerDocumentType(definition); + await cache.registerDocumentType(definition); if (!documents.get(definition.type)) { documents.register(defineCultNetDocumentBinding({ definition })); }