Refuse re-entrant TS cache calls and adopt legacy-key globals on load
The serial chain now advances on its own promise and returns the operation's promise untouched, so an unawaited failing put is an unhandled rejection again instead of being swallowed by the chain's catch. A queued call started from inside a store call, decoder, accessor or updater of the same cache's running operation throws at entry instead of waiting on itself. Detection is an AsyncLocalStorage list of running operations, so independent concurrent callers are still queued. A pull adopts a global persisted under a key other than __global__ as that type's global without writing. The first write or delete of it removes the legacy record in the same call; a failed removal takes the new record back out. Two globals of one type on disk, under any keys, are still refused. concurrentGlobalPutsCannotBothReachDisk now pauses inside push and fails when the chain is replaced by Promise.resolve().then(operation). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
9845103085
commit
4158a38a34
2 changed files with 232 additions and 15 deletions
|
|
@ -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<string, Map<string, string>>;
|
||||
readonly indexToKeyMaps: Map<string, Map<string, string>>;
|
||||
readonly globalKeys: Map<string, string>;
|
||||
// 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<string, CultCacheEnvelope>;
|
||||
};
|
||||
|
||||
// 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<readonly RunningOperation[]>();
|
||||
|
||||
// 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<string, RegisteredDefinition>();
|
||||
#hydrated = emptyHydratedState();
|
||||
readonly #stores: CultCacheStoreRegistration[] = [];
|
||||
#tail: Promise<unknown> = Promise.resolve();
|
||||
#tail: Promise<void> = 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<T>(operation: () => T | Promise<T>): Promise<T> {
|
||||
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<void>((resolve) => {
|
||||
advance = resolve;
|
||||
});
|
||||
return previous.then(() =>
|
||||
runningOperations.run([...outer, current], async () => {
|
||||
current.running = true;
|
||||
try {
|
||||
return await operation();
|
||||
} finally {
|
||||
current.running = false;
|
||||
advance();
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
registerDocumentType<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
@ -232,6 +270,7 @@ export class CultCache {
|
|||
pullAllBackingStores(): Promise<void> {
|
||||
return this.#serial(async () => {
|
||||
const next = emptyHydratedState();
|
||||
const persistedGlobalKeys = new Map<string, string>();
|
||||
|
||||
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<void> {
|
||||
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<void> {
|
||||
try {
|
||||
await home.delete(legacy);
|
||||
} catch (error) {
|
||||
await home.delete(written).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async #deleteNow<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void>((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<void>((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 () => {
|
||||
|
|
|
|||
Loading…
Reference in a new issue