diff --git a/packages/cultcache-ts/src/cult-cache.ts b/packages/cultcache-ts/src/cult-cache.ts index 5d70a0d1..8b5e87d3 100644 --- a/packages/cultcache-ts/src/cult-cache.ts +++ b/packages/cultcache-ts/src/cult-cache.ts @@ -1,3 +1,5 @@ +import { AsyncLocalStorage } from "node:async_hooks"; + import { decode, encode } from "@msgpack/msgpack"; import type { @@ -43,8 +45,18 @@ type HydratedState = { readonly nameToKeyMaps: Map>; readonly indexToKeyMaps: Map>; readonly globalKeys: Map; + // Load compatibility shim: a global persisted under a key other than GLOBAL_KEY is held as that + // type's global under GLOBAL_KEY, and its persisted envelope is kept here so the first write or + // delete of that global removes it from the home store. Remove once no store holds a global under + // a key other than GLOBAL_KEY. + readonly legacyGlobals: Map; }; +// The operations running in the current async flow. A queued call from inside one of this cache's +// own operations would wait on itself, so it is refused at entry. +type RunningOperation = { readonly cache: CultCache; running: boolean }; +const runningOperations = new AsyncLocalStorage(); + // Lookups for every held record of one type under candidate accessors, derived before anything is installed. type DerivedTypeLookups = { readonly entryLookups: Array<[HydratedEntry, EntryLookups]>; @@ -59,6 +71,7 @@ function emptyHydratedState(): HydratedState { nameToKeyMaps: new Map(), indexToKeyMaps: new Map(), globalKeys: new Map(), + legacyGlobals: new Map(), }; } @@ -147,7 +160,7 @@ export class CultCache { readonly #schemaNameDefinitions = new Map(); #hydrated = emptyHydratedState(); readonly #stores: CultCacheStoreRegistration[] = []; - #tail: Promise = Promise.resolve(); + #tail: Promise = Promise.resolve(); static builder(): CultCacheBuilder { return new CultCacheBuilder(); @@ -156,10 +169,35 @@ export class CultCache { // 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. + // The chain advances on its own promise; the caller gets the operation's promise, untouched, so + // an unawaited failure is still an unhandled rejection. Stores, decoders, accessors and updaters + // must not call a queued method of the cache they are serving: that call throws. #serial(operation: () => T | Promise): Promise { - const run = this.#tail.then(operation); - this.#tail = run.catch(() => undefined); - return run; + const outer = runningOperations.getStore() ?? []; + if (outer.some((candidate) => candidate.cache === this && candidate.running)) { + throw new Error( + "CultCache refuses a re-entrant call: an attach, pull, registration, write or delete was started from " + + "inside a store call, decoder, accessor or updater of an operation on the same cache, and would wait on itself.", + ); + } + + const current: RunningOperation = { cache: this, running: false }; + let advance!: () => void; + const previous = this.#tail; + this.#tail = new Promise((resolve) => { + advance = resolve; + }); + return previous.then(() => + runningOperations.run([...outer, current], async () => { + current.running = true; + try { + return await operation(); + } finally { + current.running = false; + advance(); + } + }), + ); } registerDocumentType( @@ -232,6 +270,7 @@ export class CultCache { pullAllBackingStores(): Promise { return this.#serial(async () => { const next = emptyHydratedState(); + const persistedGlobalKeys = new Map(); for (const registration of this.#stores) { const entries = await registration.store.pullAll(); @@ -257,7 +296,21 @@ export class CultCache { const payload = this.#cloneBytes(entry.payload); const value = registered.formatter.decode(payload); - const hydrated = this.#admitEntry(next, registered, { ...entry, type, payload }, value); + let admitted: CultCacheEnvelope = { ...entry, type, payload }; + if (registered.global) { + const persistedKey = persistedGlobalKeys.get(type); + if (persistedKey !== undefined) { + throw new Error( + `CultCache global document type "${type}" has multiple persisted entries: "${persistedKey}" and "${entry.key}".`, + ); + } + persistedGlobalKeys.set(type, entry.key); + if (entry.key !== CultCache.GLOBAL_KEY) { + next.legacyGlobals.set(type, admitted); + admitted = { ...admitted, key: CultCache.GLOBAL_KEY }; + } + } + const hydrated = this.#admitEntry(next, registered, admitted, value); this.#installEntry(next, registered, hydrated); } } @@ -588,10 +641,26 @@ export class CultCache { async #writeNow(registered: RegisteredDefinition, entry: CultCacheEnvelope, value: unknown): Promise { const home = this.#homeStore(entry.type); const hydrated = this.#admitEntry(this.#hydrated, registered, entry, value); + const legacy = this.#hydrated.legacyGlobals.get(entry.type); await home?.push(entry); + if (home && legacy) { + await this.#retireLegacyGlobal(home, legacy, entry); + } + this.#hydrated.legacyGlobals.delete(entry.type); this.#installEntry(this.#hydrated, registered, hydrated); } + // The first write of a global loaded from a legacy key removes that record in the same write. If the + // removal fails, the new record is taken back out, so the store keeps only the legacy record. + async #retireLegacyGlobal(home: CacheBackingStore, legacy: CultCacheEnvelope, written: CultCacheEnvelope): Promise { + try { + await home.delete(legacy); + } catch (error) { + await home.delete(written).catch(() => undefined); + throw error; + } + } + async #deleteNow( definition: TDefinition, key: string, @@ -603,7 +672,8 @@ export class CultCache { } const home = this.#homeStore(definition.type); - await home?.delete(this.#toEnvelope(entry)); + await home?.delete(this.#hydrated.legacyGlobals.get(definition.type) ?? this.#toEnvelope(entry)); + this.#hydrated.legacyGlobals.delete(definition.type); this.#removeEntry(this.#hydrated, registered, entry); return true; } diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 991849db..aedcf74b 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -773,27 +773,174 @@ test("concurrentGlobalPutsCannotBothReachDisk", async () => { global: true, }); const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-global-race-")), "generic.cc"); + const inner = new SingleFileMessagePackBackingStore(storePath); + // The first push after arming writes, then pauses until released. A second write that reaches the + // store during that pause is recorded: serialized writes cannot. + let armed = false; + let paused = false; + let reachedStoreWhilePaused = false; + let release!: () => void; + let reportPaused!: () => void; + const pausedSignal = new Promise((resolve) => { + reportPaused = resolve; + }); + const store: CacheBackingStore = { + pullAll: () => inner.pullAll(), + delete: (entry) => inner.delete(entry), + push: async (entry) => { + reachedStoreWhilePaused ||= paused; + await inner.push(entry); + if (armed) { + armed = false; + paused = true; + await new Promise((resolve) => { + release = resolve; + reportPaused(); + }); + paused = false; + } + }, + }; + const build = (backing: CacheBackingStore) => CultCache.builder() + .withRegistry(defineDocumentRegistry(settingsDocument)) + .withGenericStore(backing) + .build(); + const cache = build(store); + await cache.putGlobal(settingsDocument, { t: "base" }); + + armed = true; + const first = cache.putGlobal(settingsDocument, { t: "first" }); + await pausedSignal; + const second = cache.putGlobal(settingsDocument, { t: "second" }); + await new Promise((resolve) => setTimeout(resolve, 50)); + release(); + await Promise.all([first, second]); + + assert.equal(reachedStoreWhilePaused, false); + const onDisk = await inner.pullAll(); + assert.deepEqual(onDisk.map((entry) => entry.key), [CultCache.GLOBAL_KEY]); + assert.deepEqual( + Buffer.from(onDisk[0].payload), + Buffer.from(cache.getRequiredEnvelope(settingsDocument, CultCache.GLOBAL_KEY).payload), + ); + assert.deepEqual(cache.getRequiredGlobal(settingsDocument), { t: "second" }); + const reloaded = build(new SingleFileMessagePackBackingStore(storePath)); + await reloaded.pullAllBackingStores(); + assert.deepEqual(reloaded.getRequiredGlobal(settingsDocument), { t: "second" }); +}); + +test("unawaitedFailingPutIsAnUnhandledRejection", async () => { + const itemDocument = defineDocumentType({ type: "item", schema: z.object({ n: z.string() }) }); + const store: CacheBackingStore = { + pullAll: async () => [], + push: async () => { + throw new Error("disk full"); + }, + delete: async () => undefined, + }; + const cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(itemDocument)) + .withGenericStore(store) + .build(); + + const runnerListeners = process.listeners("unhandledRejection"); + process.removeAllListeners("unhandledRejection"); + const unhandled: unknown[] = []; + const listener = (reason: unknown) => { + unhandled.push(reason); + }; + process.on("unhandledRejection", listener); + try { + void cache.put(itemDocument, "k", { n: "x" }); + await new Promise((resolve) => setTimeout(resolve, 50)); + } finally { + process.off("unhandledRejection", listener); + for (const runnerListener of runnerListeners) { + process.on("unhandledRejection", runnerListener); + } + } + + assert.equal(unhandled.length, 1); + assert.match(String((unhandled[0] as Error).message), /disk full/u); + // The chain survived the rejection: the next operation still runs. + await assert.rejects(cache.put(itemDocument, "k", { n: "y" }), /disk full/u); +}); + +test("storePushThatAwaitsItsOwnCacheIsRefusedInsteadOfHanging", async () => { + const itemDocument = defineDocumentType({ type: "item", schema: z.object({ n: z.string() }) }); + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-reentrant-")), "generic.cc"); + const inner = new SingleFileMessagePackBackingStore(storePath); + let cache!: CultCache; + const store: CacheBackingStore = { + pullAll: () => inner.pullAll(), + delete: (entry) => inner.delete(entry), + push: async (entry) => { + await cache.put(itemDocument, "side", { n: "side" }); + await inner.push(entry); + }, + }; + cache = CultCache.builder() + .withRegistry(defineDocumentRegistry(itemDocument)) + .withGenericStore(store) + .build(); + + const outcome = await Promise.race([ + cache.put(itemDocument, "k", { n: "x" }).then(() => "resolved", (error: unknown) => error), + new Promise((resolve) => setTimeout(() => resolve("timed out"), 2000)), + ]); + + assert.ok(outcome instanceof Error, `expected a refusal, got ${String(outcome)}`); + assert.match(outcome.message, /re-entrant call/u); + assert.deepEqual(await inner.pullAll(), []); + assert.deepEqual(cache.snapshot(), []); + // Concurrent callers outside the operation are still queued, not refused. + const plain = CultCache.builder().withRegistry(defineDocumentRegistry(itemDocument)).build(); + await Promise.all([plain.put(itemDocument, "a", { n: "a" }), plain.put(itemDocument, "b", { n: "b" })]); + assert.equal(plain.getAll(itemDocument).length, 2); +}); + +test("legacyKeyGlobalLoadsWithoutWritingAndFirstWriteLeavesOneGlobal", async () => { + const settingsDocument = defineDocumentType({ + type: "settings", + schema: z.object({ t: z.string() }), + global: true, + }); + const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-legacy-global-")), "generic.cc"); const build = () => CultCache.builder() .withRegistry(defineDocumentRegistry(settingsDocument)) .withGenericStore(new SingleFileMessagePackBackingStore(storePath)) .build(); - const cache = build(); - await cache.putGlobal(settingsDocument, { t: "base" }); + const seed = CultCache.builder().withRegistry(defineDocumentRegistry(settingsDocument)).build(); + await seed.putGlobal(settingsDocument, { t: "old" }); + const legacy = { ...seed.getRequiredEnvelope(settingsDocument, CultCache.GLOBAL_KEY), key: "legacy" }; + await new SingleFileMessagePackBackingStore(storePath).push(legacy); + const bytes = await readFile(storePath); - 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" }), - ]); + const cache = build(); + await cache.pullAllBackingStores(); + assert.deepEqual(cache.getRequiredGlobal(settingsDocument), { t: "old" }); + assert.deepEqual(cache.get(settingsDocument, CultCache.GLOBAL_KEY), { t: "old" }); + assert.deepEqual(await readFile(storePath), bytes); + + await cache.putGlobal(settingsDocument, { t: "new" }); 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" }); + assert.deepEqual(reloaded.getRequiredGlobal(settingsDocument), { t: "new" }); + + // Deleting an adopted global removes the legacy record. + await new SingleFileMessagePackBackingStore(storePath).pushAll([legacy]); + const deleting = build(); + await deleting.pullAllBackingStores(); + assert.equal(await deleting.deleteGlobal(settingsDocument), true); + assert.deepEqual(await new SingleFileMessagePackBackingStore(storePath).pullAll(), []); + + // Two globals of one type on disk, under any keys, are refused. + await new SingleFileMessagePackBackingStore(storePath).pushAll([legacy, { ...legacy, key: CultCache.GLOBAL_KEY }]); + await assert.rejects(build().pullAllBackingStores(), /has multiple persisted entries/u); }); test("concurrentPutAndAttachCannotLandRecordInNonHomeStore", async () => {