cultcache-ts: update is conditional and retried; the blind update is gone
update and updateGlobal read the record and its content version from the store, apply to a fresh copy, commit through the store's commitIf and run again on a lost race, returning updated, declined or lostRace. A cache with no store compares the held entry. put stays the blind write. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Libby
parent
13920d5edd
commit
8bca8d831c
4 changed files with 706 additions and 92 deletions
|
|
@ -22,6 +22,8 @@ import type {
|
|||
CultCacheSchemaCatalogEntry,
|
||||
CultCacheStoreRegistration,
|
||||
CultCacheUnreadSlots,
|
||||
CultCacheUpdateOptions,
|
||||
CultCacheUpdateResult,
|
||||
} from "./types";
|
||||
|
||||
type Accessor = (value: unknown) => string | undefined;
|
||||
|
|
@ -48,6 +50,10 @@ type EntryLookups = {
|
|||
readonly indexes: ReadonlyMap<string, string>;
|
||||
};
|
||||
|
||||
// What a conditional write requires of the record at its key: the content version its store reported (null: no record), or for a
|
||||
// cache with no store the entry it held. Every write installs a new entry object, so the identity catches an A-B-A.
|
||||
type WriteCondition = { readonly version: string | null } | { readonly held: HydratedEntry | undefined };
|
||||
|
||||
type HydratedEntry = CultCacheEnvelope & {
|
||||
catalogEntry: NonNullable<CultCacheEnvelope["catalogEntry"]>;
|
||||
value: unknown;
|
||||
|
|
@ -302,85 +308,88 @@ export class CultCache {
|
|||
// 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.
|
||||
pullAllBackingStores(): Promise<void> {
|
||||
return this.#serial(async () => {
|
||||
const next = emptyHydratedState();
|
||||
const persistedGlobalKeys = new Map<string, string>();
|
||||
return this.#serial(() => this.#pullNow());
|
||||
}
|
||||
|
||||
for (const registration of this.#stores) {
|
||||
const entries = await registration.store.pullAll();
|
||||
// The pull itself, for a caller already inside a serial operation.
|
||||
async #pullNow(): Promise<void> {
|
||||
const next = emptyHydratedState();
|
||||
const persistedGlobalKeys = new Map<string, string>();
|
||||
|
||||
for (const entry of entries) {
|
||||
const registered = this.#resolveDefinitionForEnvelope(entry);
|
||||
if (!registered) {
|
||||
// A store may hold what this cache has no type for; it is carried, not served, and not a reason to refuse the load.
|
||||
next.foreign.push({
|
||||
key: entry.key,
|
||||
schemaId: schemaIdForEnvelope(entry),
|
||||
schemaName: entry.catalogEntry?.schemaName ?? entry.type,
|
||||
storedAt: entry.storedAt,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
for (const registration of this.#stores) {
|
||||
const entries = await registration.store.pullAll();
|
||||
|
||||
const type = registered.definition.type;
|
||||
const home = this.#resolveRegistration(type);
|
||||
if (home?.store !== registration.store) {
|
||||
for (const entry of entries) {
|
||||
const registered = this.#resolveDefinitionForEnvelope(entry);
|
||||
if (!registered) {
|
||||
// A store may hold what this cache has no type for; it is carried, not served, and not a reason to refuse the load.
|
||||
next.foreign.push({
|
||||
key: entry.key,
|
||||
schemaId: schemaIdForEnvelope(entry),
|
||||
schemaName: entry.catalogEntry?.schemaName ?? entry.type,
|
||||
storedAt: entry.storedAt,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
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, unreadSlots } = registered.formatter.decode(payload);
|
||||
let admitted: CultCacheEnvelope = { ...entry, type, payload };
|
||||
if (registered.global) {
|
||||
const persistedKey = persistedGlobalKeys.get(type);
|
||||
if (persistedKey !== undefined) {
|
||||
throw new Error(
|
||||
`CultCache "${type}" record "${entry.key}" was loaded from ${this.#describe(registration)}, ` +
|
||||
`but its home is ${this.#describe(home)}.`,
|
||||
`CultCache global document type "${type}" has multiple persisted entries: "${persistedKey}" and "${entry.key}".`,
|
||||
);
|
||||
}
|
||||
|
||||
const payload = this.#cloneBytes(entry.payload);
|
||||
const { value, unreadSlots } = registered.formatter.decode(payload);
|
||||
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 };
|
||||
}
|
||||
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, unreadSlots);
|
||||
const other = this.#heldUnderOtherType(next, type, hydrated.key);
|
||||
if (other !== undefined) {
|
||||
throw new Error(`Record ${hydrated.key} is held by two stores, as type "${other.type}" and as type "${type}".`);
|
||||
}
|
||||
this.#installEntry(next, registered, hydrated);
|
||||
}
|
||||
}
|
||||
|
||||
const foreignByKey = new Map<string, string>();
|
||||
for (const record of next.foreign) {
|
||||
const first = foreignByKey.get(record.key);
|
||||
if (first !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as schema "${first}" and as schema "${record.schemaName}".`);
|
||||
}
|
||||
foreignByKey.set(record.key, record.schemaName);
|
||||
const other = [...this.#definitions.keys()].map((type) => next.entries.get(this.#entryId(type, record.key))).find((held) => held !== undefined);
|
||||
const hydrated = this.#admitEntry(next, registered, admitted, value, unreadSlots);
|
||||
const other = this.#heldUnderOtherType(next, type, hydrated.key);
|
||||
if (other !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as type "${other.type}" and as schema "${record.schemaName}".`);
|
||||
throw new Error(`Record ${hydrated.key} is held by two stores, as type "${other.type}" and as type "${type}".`);
|
||||
}
|
||||
this.#installEntry(next, registered, hydrated);
|
||||
}
|
||||
}
|
||||
|
||||
const foreignByKey = new Map<string, string>();
|
||||
for (const record of next.foreign) {
|
||||
const first = foreignByKey.get(record.key);
|
||||
if (first !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as schema "${first}" and as schema "${record.schemaName}".`);
|
||||
}
|
||||
foreignByKey.set(record.key, record.schemaName);
|
||||
const other = [...this.#definitions.keys()].map((type) => next.entries.get(this.#entryId(type, record.key))).find((held) => held !== undefined);
|
||||
if (other !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as type "${other.type}" and as schema "${record.schemaName}".`);
|
||||
}
|
||||
}
|
||||
|
||||
// Judged whole, once: the refusal names the least violation of the state this load would leave, whatever the
|
||||
// order of the stores and of the entries in them.
|
||||
refuseLeastIndexSharing(function* () {
|
||||
for (const entry of next.entries.values()) {
|
||||
for (const [alias, value] of entry.lookups.indexes) {
|
||||
yield { type: entry.type, schemaName: entry.catalogEntry.schemaName, alias, value, key: entry.key };
|
||||
}
|
||||
}
|
||||
}());
|
||||
|
||||
// Judged whole, once: the refusal names the least violation of the state this load would leave, whatever the
|
||||
// order of the stores and of the entries in them.
|
||||
refuseLeastIndexSharing(function* () {
|
||||
for (const entry of next.entries.values()) {
|
||||
for (const [alias, value] of entry.lookups.indexes) {
|
||||
yield { type: entry.type, schemaName: entry.catalogEntry.schemaName, alias, value, key: entry.key };
|
||||
}
|
||||
}
|
||||
}());
|
||||
|
||||
this.#hydrated = next;
|
||||
});
|
||||
this.#hydrated = next;
|
||||
}
|
||||
|
||||
get<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
@ -513,12 +522,16 @@ export class CultCache {
|
|||
return key ? this.get(definition, key) : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* The blind write: the record at the key becomes `value`, whatever it held, and the last writer wins. `update` is the
|
||||
* conditional path, which changes a record from what the store holds now and re-runs on a lost race.
|
||||
*/
|
||||
put<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() => this.#putNow(definition, key, value));
|
||||
return this.#serial(async () => (await this.#putNow(definition, key, value)).value);
|
||||
}
|
||||
|
||||
putEnvelope<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
@ -560,31 +573,72 @@ export class CultCache {
|
|||
definition: TDefinition,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() => {
|
||||
return this.#serial(async () => {
|
||||
this.#requireGlobal(definition);
|
||||
return this.#putNow(definition, CultCache.GLOBAL_KEY, value);
|
||||
return (await this.#putNow(definition, CultCache.GLOBAL_KEY, value)).value;
|
||||
});
|
||||
}
|
||||
|
||||
update<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
/**
|
||||
* Changes the record at the key from what the store holds now. `apply` may run up to `maxAttempts` times (default 5): each
|
||||
* attempt gets a fresh copy of the record as the store holds it (undefined when absent), which it may change and return, and
|
||||
* the result is written only if the record is still that version; a lost race runs `apply` again on the newer record. It must
|
||||
* not write this cache or have side effects. Returning undefined writes nothing (`declined`). A lost race that outlasts
|
||||
* `maxAttempts` returns `lostRace` and does not throw; a refusal, or a throw from `apply`, rejects and is not retried. The
|
||||
* write is durable when the promise resolves. `document` is the cache's record after the outcome. `put` is the blind write.
|
||||
*/
|
||||
async update<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
updater: (
|
||||
apply: (
|
||||
current: CultCacheDocumentValue<TDefinition> | undefined,
|
||||
) => CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() => this.#putNow(definition, key, updater(this.get(definition, key))));
|
||||
) => CultCacheDocumentValue<TDefinition> | undefined | Promise<CultCacheDocumentValue<TDefinition> | undefined>,
|
||||
options: CultCacheUpdateOptions = {},
|
||||
): Promise<CultCacheUpdateResult<CultCacheDocumentValue<TDefinition>>> {
|
||||
if (typeof apply !== "function") {
|
||||
throw new TypeError("CultCache update needs a function to apply.");
|
||||
}
|
||||
const maxAttempts = options.maxAttempts ?? 5;
|
||||
if (!Number.isInteger(maxAttempts) || maxAttempts < 1) {
|
||||
throw new RangeError(`CultCache update maxAttempts must be an integer of at least 1, not ${String(maxAttempts)}.`);
|
||||
}
|
||||
|
||||
for (let attempt = 1; ; attempt++) {
|
||||
const { expected, copy } = await this.#serial(() => this.#readForUpdate(definition, key));
|
||||
// Outside the serial queue and the store's lock: apply may await, and may write this cache through other keys.
|
||||
const next = await apply(copy as CultCacheDocumentValue<TDefinition> | undefined);
|
||||
if (next === undefined || next === null) {
|
||||
return { outcome: "declined", document: this.get(definition, key) };
|
||||
}
|
||||
const written = await this.#serial(() => this.#putNow(definition, key, next, expected));
|
||||
if (written.outcome === "committed") {
|
||||
return { outcome: "updated", document: this.get(definition, key) as CultCacheDocumentValue<TDefinition> };
|
||||
}
|
||||
if (attempt === maxAttempts) {
|
||||
// Display only. A cache with no store has nothing to pull, and a pull would replace what it holds with nothing.
|
||||
if (this.#stores.length > 0) {
|
||||
await this.#serial(() => this.#pullNow());
|
||||
}
|
||||
return { outcome: "lostRace", document: this.get(definition, key) };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
updateGlobal<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
/** `update` of the type's global record. */
|
||||
async updateGlobal<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
updater: (
|
||||
apply: (
|
||||
current: CultCacheDocumentValue<TDefinition> | undefined,
|
||||
) => CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() =>
|
||||
this.#putNow(definition, CultCache.GLOBAL_KEY, updater(this.getGlobal(definition))),
|
||||
);
|
||||
) => CultCacheDocumentValue<TDefinition> | undefined | Promise<CultCacheDocumentValue<TDefinition> | undefined>,
|
||||
options: CultCacheUpdateOptions = {},
|
||||
): Promise<CultCacheUpdateResult<CultCacheDocumentValue<TDefinition>>> {
|
||||
this.#requireGlobal(definition);
|
||||
if (this.#hydrated.legacyGlobals.has(definition.type)) {
|
||||
throw new Error(
|
||||
`CultCache global "${definition.type}" was loaded from a legacy key; write it once with putGlobal to retire its legacy key.`,
|
||||
);
|
||||
}
|
||||
return this.update(definition, CultCache.GLOBAL_KEY, apply, options);
|
||||
}
|
||||
|
||||
delete<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
@ -724,11 +778,54 @@ export class CultCache {
|
|||
this.#stores.push(candidate);
|
||||
}
|
||||
|
||||
// One attempt's read for update: the record at the key as its home holds it now, the condition a commit must meet, and a copy of
|
||||
// the record for apply to change. A store-backed cache asks its store, which reports the record and its version from one read;
|
||||
// a cache with no store has only the held entry, whose identity is the condition and whose admitted payload is the copy.
|
||||
async #readForUpdate(
|
||||
definition: AnyCultCacheDocumentDefinition,
|
||||
key: string,
|
||||
): Promise<{ expected: WriteCondition; copy: unknown }> {
|
||||
const registered = this.#requireDefinition(definition);
|
||||
this.#requireGlobalKey(registered, key);
|
||||
const type = definition.type;
|
||||
const home = this.#homeStore(type);
|
||||
if (home === undefined) {
|
||||
const held = this.#hydrated.entries.get(this.#entryId(type, key));
|
||||
if (held === undefined && this.#heldUnderOtherType(this.#hydrated, type, key) !== undefined) {
|
||||
throw new Error(`Record ${key} is not a "${type}"; update cannot copy it.`);
|
||||
}
|
||||
return { expected: { held }, copy: held ? registered.formatter.decode(this.#cloneBytes(held.payload)).value : undefined };
|
||||
}
|
||||
if (home.read === undefined || home.commitIf === undefined) {
|
||||
throw new Error(`CultCache backing store for type "${type}" cannot write conditionally; update needs a store with read and commitIf.`);
|
||||
}
|
||||
const found = await home.read(key);
|
||||
if (found === undefined) {
|
||||
return { expected: { version: null }, copy: undefined };
|
||||
}
|
||||
if (this.#resolveDefinitionForEnvelope(found.envelope) !== registered) {
|
||||
throw new Error(`Record ${key} is not a "${type}"; update cannot copy it.`);
|
||||
}
|
||||
return { expected: { version: found.version }, copy: registered.formatter.decode(this.#cloneBytes(found.envelope.payload)).value };
|
||||
}
|
||||
|
||||
#putNow<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<{ outcome: "committed"; value: CultCacheDocumentValue<TDefinition> }>;
|
||||
#putNow<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
expected: WriteCondition,
|
||||
): Promise<{ outcome: "committed"; value: CultCacheDocumentValue<TDefinition> } | { outcome: "mismatch" }>;
|
||||
async #putNow<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
expected?: WriteCondition,
|
||||
): Promise<{ outcome: "committed"; value: CultCacheDocumentValue<TDefinition> } | { outcome: "mismatch" }> {
|
||||
const registered = this.#requireDefinition(definition);
|
||||
const payload = registered.formatter.encode(value);
|
||||
const { value: parsed, unreadSlots } = registered.formatter.decode(payload) as { value: CultCacheDocumentValue<TDefinition>; unreadSlots?: NotedSlots };
|
||||
|
|
@ -740,19 +837,44 @@ export class CultCache {
|
|||
schemaId: registered.catalogEntry.schemaId,
|
||||
catalogEntry: registered.catalogEntry,
|
||||
};
|
||||
await this.#writeNow(registered, entry, parsed, unreadSlots);
|
||||
return parsed;
|
||||
const outcome = await this.#writeNow(registered, entry, parsed, unreadSlots, expected);
|
||||
return outcome === "committed" ? { outcome, value: parsed } : { outcome };
|
||||
}
|
||||
|
||||
// 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, unreadSlots?: NotedSlots): Promise<void> {
|
||||
// With a condition, the write lands only if the record at the key is still the one the caller read: the store's version
|
||||
// (commitIf), or for a cache with no store the held entry itself. Otherwise nothing is written or installed and the answer is
|
||||
// "mismatch". Without one the write is blind, as put's.
|
||||
async #writeNow(
|
||||
registered: RegisteredDefinition,
|
||||
entry: CultCacheEnvelope,
|
||||
value: unknown,
|
||||
unreadSlots?: NotedSlots,
|
||||
expected?: WriteCondition,
|
||||
): Promise<"committed" | "mismatch"> {
|
||||
const home = this.#homeStore(entry.type);
|
||||
const retyped = this.#judge(registered, entry.key, entry);
|
||||
const hydrated = this.#admitEntry(this.#hydrated, registered, entry, value, unreadSlots);
|
||||
this.#refuseIndexSharing(this.#hydrated, entry.type, registered.catalogEntry.schemaName, entry.key, hydrated.lookups);
|
||||
const legacy = this.#hydrated.legacyGlobals.get(entry.type);
|
||||
await home?.push(entry);
|
||||
if (expected !== undefined && "held" in expected) {
|
||||
if (this.#hydrated.entries.get(this.#entryId(entry.type, entry.key)) !== expected.held) {
|
||||
return "mismatch";
|
||||
}
|
||||
}
|
||||
if (expected !== undefined && "version" in expected) {
|
||||
if (home?.commitIf === undefined) {
|
||||
throw new Error(`CultCache backing store for type "${entry.type}" cannot write conditionally; update needs a store with read and commitIf.`);
|
||||
}
|
||||
const committed = await home.commitIf(entry, expected.version);
|
||||
if (committed.outcome === "mismatch") {
|
||||
return "mismatch";
|
||||
}
|
||||
hydrated.storedAt = committed.storedAt;
|
||||
} else {
|
||||
await home?.push(entry);
|
||||
}
|
||||
if (home && legacy) {
|
||||
await this.#retireLegacyGlobal(home, legacy, entry);
|
||||
}
|
||||
|
|
@ -762,6 +884,7 @@ export class CultCache {
|
|||
this.#removeEntry(this.#hydrated, this.#definitions.get(retyped.type)!, retyped);
|
||||
}
|
||||
this.#installEntry(this.#hydrated, registered, hydrated);
|
||||
return "committed";
|
||||
}
|
||||
|
||||
// An envelope of the definition's type at the key, carrying nothing: what the judge asks about a write or a removal by that type.
|
||||
|
|
|
|||
|
|
@ -46,6 +46,21 @@ async function firstCall(path: string) {
|
|||
}
|
||||
// snippet:end
|
||||
|
||||
// The update snippet, which docs/start/README.md shows after the first call and which uses its definitions.
|
||||
// snippet:update:begin
|
||||
async function restock(path: string) {
|
||||
const cache = CultCache.builder()
|
||||
.withRegistry(registry)
|
||||
.withGenericStore(new SingleFileMessagePackBackingStore(path))
|
||||
.build();
|
||||
await cache.pullAllBackingStores();
|
||||
return cache.update(itemDocument, "item:potion", (item) => {
|
||||
const current = item ?? { name: "Potion", category: "Consumable", value: 0 };
|
||||
return { ...current, value: current.value + 10 };
|
||||
});
|
||||
}
|
||||
// snippet:update:end
|
||||
|
||||
test("define, store and read back a typed document", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "cultcache-first-call-"));
|
||||
try {
|
||||
|
|
@ -57,13 +72,35 @@ test("define, store and read back a typed document", async () => {
|
|||
}
|
||||
});
|
||||
|
||||
test("docs show the tested snippet", async () => {
|
||||
test("update with retry: restock twice reads value 20 through a second cache", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "cultcache-first-call-"));
|
||||
try {
|
||||
const path = join(dir, "restock.cc");
|
||||
assert.equal((await restock(path)).outcome, "updated");
|
||||
assert.equal((await restock(path)).outcome, "updated");
|
||||
const reopened = CultCache.builder()
|
||||
.withRegistry(registry)
|
||||
.withGenericStore(new SingleFileMessagePackBackingStore(path))
|
||||
.build();
|
||||
await reopened.pullAllBackingStores();
|
||||
assert.equal(reopened.get(itemDocument, "item:potion")?.value, 20);
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
// The docs show the region between `// snippet:<region>begin` and `end` markers verbatim, the package import aside.
|
||||
async function assertDocsShow(region: "" | "update:") {
|
||||
// Compiled to dist-test/test/first-call.test.js, so the package root is two levels up.
|
||||
const source = await readFile(join(__dirname, "..", "..", "test", "first-call.test.ts"), "utf8");
|
||||
const snippet = source
|
||||
.split("// snippet:begin\n")[1]
|
||||
.split("// snippet:end\n")[0]
|
||||
.split(`// snippet:${region}begin\n`)[1]
|
||||
.split(`// snippet:${region}end\n`)[0]
|
||||
.replace('"../src/index"', '"@gamecult/cultcache-ts"');
|
||||
const docs = (await readFile(join(__dirname, "..", "..", "..", "..", "docs", "start", "README.md"), "utf8")).replace(/\r\n/g, "\n");
|
||||
assert.ok(docs.includes("```ts\n" + snippet + "```"), "docs/start/README.md no longer shows test/first-call.test.ts");
|
||||
});
|
||||
assert.ok(docs.includes("```ts\n" + snippet + "```"), `docs/start/README.md no longer shows the ${region || "first-call "}region of test/first-call.test.ts`);
|
||||
}
|
||||
|
||||
test("docs show the tested snippet", () => assertDocsShow(""));
|
||||
|
||||
test("docs show the tested update snippet", () => assertDocsShow("update:"));
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import { z } from "zod";
|
|||
import { codecs } from "../src";
|
||||
import type { Codec } from "../src/codec";
|
||||
import { defineDocumentType, documentPayloadFromBytes } from "../src/document";
|
||||
import { recordVersion } from "../src/single-file-messagepack-backing-store";
|
||||
import type { CacheBackingStore, CultCacheEnvelope } from "../src/types";
|
||||
|
||||
// The union-surface fixture: the codecs, schema and value tests/vectors/union-parity was written from, shared by the
|
||||
|
|
@ -235,6 +236,16 @@ export function memoryStore(entries: CultCacheEnvelope[]): CacheBackingStore & {
|
|||
async pullAll() { return entries.map((entry) => ({ ...entry })); },
|
||||
async push(entry) { entries.push(entry); },
|
||||
async delete() {},
|
||||
async read(key) {
|
||||
const envelope = [...entries].reverse().find((entry) => entry.key === key);
|
||||
return envelope === undefined ? undefined : { envelope, version: recordVersion(envelope) };
|
||||
},
|
||||
async commitIf(entry, expectedVersion) {
|
||||
const current = [...entries].reverse().find((candidate) => candidate.key === entry.key);
|
||||
if ((current === undefined ? null : recordVersion(current)) !== expectedVersion) return { outcome: "mismatch" };
|
||||
entries.push(entry);
|
||||
return { outcome: "committed", storedAt: entry.storedAt };
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
443
packages/cultcache-ts/test/update.test.ts
Normal file
443
packages/cultcache-ts/test/update.test.ts
Normal file
|
|
@ -0,0 +1,443 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { mkdtemp, readFile, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import test from "node:test";
|
||||
import { setImmediate as tick } from "node:timers/promises";
|
||||
|
||||
import { decode } from "@msgpack/msgpack";
|
||||
import { z } from "zod";
|
||||
|
||||
import { CultCache } from "../src/cult-cache";
|
||||
import { defineDocumentRegistry, defineDocumentType, documentPayloadBytes } from "../src/document";
|
||||
import { SingleFileMessagePackBackingStore, readSingleFileStore } from "../src/single-file-messagepack-backing-store";
|
||||
import { SchemaConflictError } from "../src/store-format";
|
||||
import type { CacheBackingStore } from "../src/types";
|
||||
import { envelopeOf, memoryStore, read, surfaceDocument } from "./union-surface-fixture";
|
||||
|
||||
// CultCache.update: read the record and its content version from the store, apply a function to a copy, commit only if the
|
||||
// version still holds, run again on a lost race (variants:cut_spec:cut-update-with-retry-ts.r1; the C# tests in
|
||||
// tests/GameCult.Caching.Tests/UpdateTests.cs are the parity).
|
||||
|
||||
const counterDocument = defineDocumentType({
|
||||
type: "update.counter",
|
||||
schemaId: "update.counter.v1",
|
||||
schemaName: "update.counter",
|
||||
schemaVersion: "update.counter.v1",
|
||||
schema: z.object({ value: z.number().int() }),
|
||||
});
|
||||
const tallyDocument = defineDocumentType({
|
||||
type: "update.tally",
|
||||
schemaId: "update.tally.v1",
|
||||
schemaName: "update.tally",
|
||||
schemaVersion: "update.tally.v1",
|
||||
schema: z.object({ count: z.number().int() }),
|
||||
});
|
||||
const registry = defineDocumentRegistry(counterDocument, tallyDocument);
|
||||
const settingsFields = { schemaName: "update.settings", schemaVersion: "update.settings.v1", schemaId: "update.settings.v1", schema: z.object({ theme: z.string() }) };
|
||||
const settingsDocument = defineDocumentType({ type: "update.settings", ...settingsFields, global: true });
|
||||
const plainSettingsDocument = defineDocumentType({ type: "update.settings", ...settingsFields });
|
||||
|
||||
const KEY = "counter";
|
||||
|
||||
async function withDirectory<T>(body: (path: string) => Promise<T>): Promise<T> {
|
||||
const directory = await mkdtemp(join(tmpdir(), "cultcache-update-"));
|
||||
try {
|
||||
return await body(join(directory, "store.cc"));
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
type Registry = Parameters<ReturnType<typeof CultCache.builder>["withRegistry"]>[0];
|
||||
|
||||
async function open(path: string, documents: Registry = registry): Promise<CultCache> {
|
||||
const cache = CultCache.builder().withRegistry(documents).withGenericStore(new SingleFileMessagePackBackingStore(path)).build();
|
||||
await cache.pullAllBackingStores();
|
||||
return cache;
|
||||
}
|
||||
|
||||
// What the file holds at the key, read without a cache.
|
||||
function stored(path: string, key = KEY): number | undefined {
|
||||
const envelope = readSingleFileStore(path).envelopes.find((candidate) => candidate.key === key);
|
||||
return envelope === undefined ? undefined : (decode(documentPayloadBytes(envelope.payload)) as { value: number }).value;
|
||||
}
|
||||
|
||||
const plus = (n: number) => (current: { value: number } | undefined) => ({ value: (current?.value ?? 0) + n });
|
||||
|
||||
test("two writers updating lose no increment", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [left, right] = [await open(path), await open(path)];
|
||||
let calls = 0;
|
||||
const loop = async (cache: CultCache) => {
|
||||
const outcomes: string[] = [];
|
||||
for (let i = 0; i < 50; i++) {
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
calls++;
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
}, { maxAttempts: 50 });
|
||||
outcomes.push(result.outcome);
|
||||
}
|
||||
return outcomes;
|
||||
};
|
||||
const outcomes = (await Promise.all([loop(left), loop(right)])).flat();
|
||||
assert.ok(outcomes.every((outcome) => outcome === "updated"), outcomes.join());
|
||||
assert.equal(stored(path), 100);
|
||||
assert.ok(calls > 100, `the caches raced: apply ran ${calls} times`);
|
||||
});
|
||||
});
|
||||
|
||||
test("a lost attempt reruns apply on the fresh value", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const seen: Array<number | undefined> = [];
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
seen.push(current?.value);
|
||||
if (seen.length === 1) await rival.put(counterDocument, KEY, { value: 100 });
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, 100]);
|
||||
assert.equal(result.outcome, "updated");
|
||||
assert.equal(result.document?.value, 101);
|
||||
assert.equal(stored(path), 101);
|
||||
});
|
||||
});
|
||||
|
||||
test("exhausted attempts return lostRace, and the cache then shows the store", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
let calls = 0;
|
||||
const result = await cache.update(
|
||||
counterDocument,
|
||||
KEY,
|
||||
async (current) => {
|
||||
calls++;
|
||||
await rival.put(counterDocument, KEY, { value: 100 * calls });
|
||||
return { value: current!.value + 1 };
|
||||
},
|
||||
{ maxAttempts: 3 },
|
||||
);
|
||||
assert.equal(calls, 3);
|
||||
assert.equal(result.outcome, "lostRace");
|
||||
assert.equal(result.document?.value, 300);
|
||||
assert.equal(stored(path), 300);
|
||||
});
|
||||
});
|
||||
|
||||
test("five attempts by default", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
let calls = 0;
|
||||
const result = await cache.update(counterDocument, KEY, async () => {
|
||||
calls++;
|
||||
await rival.put(counterDocument, KEY, { value: calls });
|
||||
return { value: -1 };
|
||||
});
|
||||
assert.equal(result.outcome, "lostRace");
|
||||
assert.equal(calls, 5);
|
||||
});
|
||||
});
|
||||
|
||||
test("maxAttempts 1 makes one call; invalid values and a non-function reject before any call", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
let calls = 0;
|
||||
const apply = async (current: { value: number } | undefined) => {
|
||||
calls++;
|
||||
await rival.put(counterDocument, KEY, { value: 50 + calls });
|
||||
return { value: current!.value + 1 };
|
||||
};
|
||||
assert.equal((await cache.update(counterDocument, KEY, apply, { maxAttempts: 1 })).outcome, "lostRace");
|
||||
assert.equal(calls, 1);
|
||||
for (const maxAttempts of [0, -1, 1.5, Number.NaN]) {
|
||||
await assert.rejects(() => cache.update(counterDocument, KEY, apply, { maxAttempts }), RangeError);
|
||||
}
|
||||
await assert.rejects(() => cache.update(counterDocument, KEY, "no" as never), TypeError);
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("apply gets a fresh copy: changing it changes nothing held, and a lost attempt's change is not seen by the next", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const argument: Array<{ value: number } | undefined> = [];
|
||||
const seen: number[] = [];
|
||||
await cache.update(counterDocument, KEY, async (current) => {
|
||||
argument.push(current);
|
||||
assert.notEqual(current, cache.get(counterDocument, KEY));
|
||||
seen.push(current!.value);
|
||||
current!.value += 1000;
|
||||
assert.equal(cache.get(counterDocument, KEY)!.value, 1);
|
||||
if (seen.length === 1) await rival.put(counterDocument, KEY, { value: 7 });
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, 7], "the second attempt does not see the first one's change");
|
||||
assert.notEqual(argument[0], argument[1]);
|
||||
});
|
||||
});
|
||||
|
||||
test("an in-place change to the held value is neither applied nor persisted", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, fresh] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
cache.get(counterDocument, KEY)!.value = 999;
|
||||
const seen: number[] = [];
|
||||
const result = await cache.update(counterDocument, KEY, (current) => {
|
||||
seen.push(current!.value);
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1]);
|
||||
assert.equal(stored(path), 2);
|
||||
await fresh.pullAllBackingStores();
|
||||
assert.equal(fresh.get(counterDocument, KEY)!.value, 2);
|
||||
result.document!.value = 999;
|
||||
await cache.update(counterDocument, KEY, plus(1));
|
||||
assert.equal(stored(path), 3);
|
||||
});
|
||||
});
|
||||
|
||||
test("undefined from apply declines: nothing is written", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const before = await readFile(path);
|
||||
const held = await cache.update(counterDocument, KEY, () => undefined);
|
||||
assert.equal(held.outcome, "declined");
|
||||
assert.equal(held.document, cache.get(counterDocument, KEY));
|
||||
const absent = await cache.update(counterDocument, "absent", () => undefined);
|
||||
assert.deepEqual(absent, { outcome: "declined", document: undefined });
|
||||
assert.deepEqual([...(await readFile(path))], [...before]);
|
||||
});
|
||||
});
|
||||
|
||||
test("an absent key is created once under a race", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [left, right] = [await open(path), await open(path)];
|
||||
const outcomes = await Promise.all(
|
||||
[left, right].map((cache) =>
|
||||
cache.update(counterDocument, KEY, async (current) => {
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
}),
|
||||
),
|
||||
);
|
||||
assert.deepEqual(outcomes.map((result) => result.outcome), ["updated", "updated"]);
|
||||
assert.equal(stored(path), 2);
|
||||
});
|
||||
});
|
||||
|
||||
test("a rival writing and restoring the value still reruns apply", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const seen: number[] = [];
|
||||
await cache.update(counterDocument, KEY, async (current) => {
|
||||
seen.push(current!.value);
|
||||
if (seen.length === 1) {
|
||||
await rival.put(counterDocument, KEY, { value: 5 });
|
||||
await rival.put(counterDocument, KEY, { value: 1 });
|
||||
}
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, 1]);
|
||||
assert.equal(stored(path), 2);
|
||||
});
|
||||
});
|
||||
|
||||
test("a rival removal is seen as undefined", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const seen: Array<number | undefined> = [];
|
||||
await cache.update(counterDocument, KEY, async (current) => {
|
||||
seen.push(current?.value);
|
||||
if (seen.length === 1) {
|
||||
await rival.pullAllBackingStores();
|
||||
await rival.delete(counterDocument, KEY);
|
||||
}
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, undefined]);
|
||||
assert.equal(stored(path), 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("a stale cache's first attempt reads the store", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [stale, rival] = [await open(path), await open(path)];
|
||||
await rival.put(counterDocument, KEY, { value: 1 });
|
||||
await stale.pullAllBackingStores();
|
||||
await rival.put(counterDocument, KEY, { value: 7 });
|
||||
const seen: number[] = [];
|
||||
await stale.update(counterDocument, KEY, (current) => {
|
||||
seen.push(current!.value);
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [7]);
|
||||
assert.equal(stored(path), 8);
|
||||
});
|
||||
});
|
||||
|
||||
test("refusals are not retried and apply is not called for a record it cannot copy", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(tallyDocument, "tally", { count: 1 });
|
||||
const before = await readFile(path);
|
||||
let calls = 0;
|
||||
const apply = (current: { value: number } | undefined) => {
|
||||
calls++;
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
};
|
||||
// Another type at the key.
|
||||
await assert.rejects(() => cache.update(counterDocument, "tally", apply), /not a "update.counter"/);
|
||||
// A foreign record: a type this cache has no definition for.
|
||||
const narrow = await open(path, defineDocumentRegistry(counterDocument));
|
||||
await assert.rejects(() => narrow.update(counterDocument, "tally", apply), /not a "update.counter"/);
|
||||
assert.equal(calls, 0);
|
||||
assert.deepEqual([...(await readFile(path))], [...before]);
|
||||
|
||||
// A store without read and commitIf cannot take a conditional write.
|
||||
const bare: CacheBackingStore = { pullAll: async () => [], push: async () => {}, delete: async () => {} };
|
||||
const plain = CultCache.builder().withRegistry(registry).withGenericStore(bare).build();
|
||||
await plain.pullAllBackingStores();
|
||||
await assert.rejects(() => plain.update(counterDocument, KEY, apply), /read and commitIf/);
|
||||
assert.equal(calls, 0);
|
||||
|
||||
// A record held read-only is refused by the judge after apply ran once; apply that throws is not retried either.
|
||||
const surface = surfaceDocument();
|
||||
const entries = [envelopeOf(await read("extra-slot-record.msgpack"))];
|
||||
const readOnly = CultCache.builder().withDocumentType(surface).withGenericStore(memoryStore(entries)).build();
|
||||
await readOnly.pullAllBackingStores();
|
||||
let surfaceCalls = 0;
|
||||
await assert.rejects(() => readOnly.update(surface, "surface", (value) => (surfaceCalls++, value)), SchemaConflictError);
|
||||
assert.equal(surfaceCalls, 1);
|
||||
assert.equal(entries.length, 1, "nothing reached the store");
|
||||
await assert.rejects(
|
||||
() => cache.update(counterDocument, KEY, () => { calls++; throw new Error("boom"); }),
|
||||
/boom/,
|
||||
);
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("own writes cause no retry, and updated returns the held instance", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
let calls = 0;
|
||||
const apply = (current: { value: number } | undefined) => {
|
||||
calls++;
|
||||
return { value: current!.value + 1 };
|
||||
};
|
||||
await cache.update(counterDocument, KEY, apply);
|
||||
const second = await cache.update(counterDocument, KEY, apply);
|
||||
assert.equal(calls, 2);
|
||||
assert.equal(second.outcome, "updated");
|
||||
assert.equal(second.document, cache.get(counterDocument, KEY));
|
||||
assert.equal(stored(path), 3);
|
||||
// The cache holds the record as the store wrote it, the stamp the store chose included: over a record stamped in the future
|
||||
// the store writes one millisecond after it, whatever the clock says.
|
||||
const future = readSingleFileStore(path).envelopes[0]!;
|
||||
await new SingleFileMessagePackBackingStore(path).push({ ...future, storedAt: "2099-01-01T00:00:00.000Z" });
|
||||
await cache.update(counterDocument, KEY, apply);
|
||||
assert.equal(readSingleFileStore(path).envelopes[0]!.storedAt, "2099-01-01T00:00:00.001Z");
|
||||
assert.equal(cache.getEnvelope(counterDocument, KEY)!.storedAt, "2099-01-01T00:00:00.001Z");
|
||||
});
|
||||
});
|
||||
|
||||
test("apply runs outside the cache's serial queue", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
let calls = 0;
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
calls++;
|
||||
await Promise.race([
|
||||
cache.put(tallyDocument, "other", { count: 1 }),
|
||||
new Promise((_, reject) => setTimeout(() => reject(new Error("a cache call from apply did not finish")), 5000).unref()),
|
||||
]);
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
assert.equal(result.outcome, "updated");
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("a cache with no store loses no increment under interleaved updates", async () => {
|
||||
const cache = CultCache.builder().withRegistry(registry).build();
|
||||
await cache.pullAllBackingStores();
|
||||
let calls = 0;
|
||||
const loop = async () => {
|
||||
let updated = 0;
|
||||
for (let i = 0; i < 50; i++) {
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
calls++;
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
if (result.outcome === "updated") updated++;
|
||||
}
|
||||
return updated;
|
||||
};
|
||||
// The two loops run in lockstep, so one may lose every round (a lostRace is the answer, not a lost write): what is held is
|
||||
// exactly what the updated outcomes wrote.
|
||||
const updated = (await Promise.all([loop(), loop()])).reduce((sum, n) => sum + n, 0);
|
||||
assert.equal(cache.get(counterDocument, KEY)!.value, updated);
|
||||
assert.ok(calls > updated, `the loops raced: apply ran ${calls} times for ${updated} writes`);
|
||||
});
|
||||
|
||||
test("a store-less cache copies the bytes it admitted, not the held value", async () => {
|
||||
const cache = CultCache.builder().withRegistry(registry).build();
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
cache.get(counterDocument, KEY)!.value = 999;
|
||||
const seen: number[] = [];
|
||||
await cache.update(counterDocument, KEY, (current) => (seen.push(current!.value), { value: current!.value + 1 }));
|
||||
assert.deepEqual(seen, [1]);
|
||||
});
|
||||
|
||||
test("updateGlobal updates the global record and refuses one loaded from a legacy key until putGlobal retires it", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const settings = defineDocumentRegistry(settingsDocument);
|
||||
const cache = await open(path, settings);
|
||||
const first = await cache.updateGlobal(settingsDocument, (current) => ({ theme: `${current?.theme ?? ""}a` }));
|
||||
assert.equal(first.outcome, "updated");
|
||||
assert.deepEqual(cache.getGlobal(settingsDocument), { theme: "a" });
|
||||
assert.equal(readSingleFileStore(path).envelopes[0]!.key, CultCache.GLOBAL_KEY);
|
||||
await rm(path);
|
||||
|
||||
const legacy = await open(path, defineDocumentRegistry(plainSettingsDocument));
|
||||
await legacy.put(plainSettingsDocument, "old-settings", { theme: "x" });
|
||||
const loaded = await open(path, settings);
|
||||
let calls = 0;
|
||||
await assert.rejects(() => loaded.updateGlobal(settingsDocument, (current) => (calls++, current)), /putGlobal to retire its legacy key/);
|
||||
assert.equal(calls, 0);
|
||||
await loaded.putGlobal(settingsDocument, { theme: "y" });
|
||||
assert.equal((await loaded.updateGlobal(settingsDocument, (current) => ({ theme: `${current!.theme}z` }))).outcome, "updated");
|
||||
assert.deepEqual(loaded.getGlobal(settingsDocument), { theme: "yz" });
|
||||
await assert.rejects(() => cache.update(settingsDocument, "not-global", () => ({ theme: "q" })), /must use key/);
|
||||
});
|
||||
});
|
||||
|
||||
test("the ledger holds under contention: the stored value is the number of updated outcomes", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const caches = await Promise.all([open(path), open(path), open(path)]);
|
||||
const results = await Promise.all(
|
||||
caches.map(async (cache) => {
|
||||
let updated = 0;
|
||||
for (let i = 0; i < 25; i++) {
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
if (result.outcome === "updated") updated++;
|
||||
}
|
||||
return updated;
|
||||
}),
|
||||
);
|
||||
assert.equal(stored(path), results.reduce((sum, n) => sum + n, 0));
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue