diff --git a/packages/cultcache-ts/src/cult-cache.ts b/packages/cultcache-ts/src/cult-cache.ts index 543356aa..9b0bc05f 100644 --- a/packages/cultcache-ts/src/cult-cache.ts +++ b/packages/cultcache-ts/src/cult-cache.ts @@ -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; }; +// 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; 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 { - return this.#serial(async () => { - const next = emptyHydratedState(); - const persistedGlobalKeys = new Map(); + 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 { + const next = emptyHydratedState(); + const persistedGlobalKeys = new Map(); - 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(); - 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(); + 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( @@ -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( definition: TDefinition, key: string, value: CultCacheDocumentValue, ): Promise> { - return this.#serial(() => this.#putNow(definition, key, value)); + return this.#serial(async () => (await this.#putNow(definition, key, value)).value); } putEnvelope( @@ -560,31 +573,72 @@ export class CultCache { definition: TDefinition, value: CultCacheDocumentValue, ): Promise> { - 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( + /** + * 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( definition: TDefinition, key: string, - updater: ( + apply: ( current: CultCacheDocumentValue | undefined, - ) => CultCacheDocumentValue, - ): Promise> { - return this.#serial(() => this.#putNow(definition, key, updater(this.get(definition, key)))); + ) => CultCacheDocumentValue | undefined | Promise | undefined>, + options: CultCacheUpdateOptions = {}, + ): Promise>> { + 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 | 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 }; + } + 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( + /** `update` of the type's global record. */ + async updateGlobal( definition: TDefinition, - updater: ( + apply: ( current: CultCacheDocumentValue | undefined, - ) => CultCacheDocumentValue, - ): Promise> { - return this.#serial(() => - this.#putNow(definition, CultCache.GLOBAL_KEY, updater(this.getGlobal(definition))), - ); + ) => CultCacheDocumentValue | undefined | Promise | undefined>, + options: CultCacheUpdateOptions = {}, + ): Promise>> { + 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( @@ -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( + definition: TDefinition, + key: string, + value: CultCacheDocumentValue, + ): Promise<{ outcome: "committed"; value: CultCacheDocumentValue }>; + #putNow( + definition: TDefinition, + key: string, + value: CultCacheDocumentValue, + expected: WriteCondition, + ): Promise<{ outcome: "committed"; value: CultCacheDocumentValue } | { outcome: "mismatch" }>; async #putNow( definition: TDefinition, key: string, value: CultCacheDocumentValue, - ): Promise> { + expected?: WriteCondition, + ): Promise<{ outcome: "committed"; value: CultCacheDocumentValue } | { outcome: "mismatch" }> { const registered = this.#requireDefinition(definition); const payload = registered.formatter.encode(value); const { value: parsed, unreadSlots } = registered.formatter.decode(payload) as { value: CultCacheDocumentValue; 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 { + // 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. diff --git a/packages/cultcache-ts/test/first-call.test.ts b/packages/cultcache-ts/test/first-call.test.ts index 45f6edd2..9ef3d308 100644 --- a/packages/cultcache-ts/test/first-call.test.ts +++ b/packages/cultcache-ts/test/first-call.test.ts @@ -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: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:")); diff --git a/packages/cultcache-ts/test/union-surface-fixture.ts b/packages/cultcache-ts/test/union-surface-fixture.ts index 1b556eb1..28089522 100644 --- a/packages/cultcache-ts/test/union-surface-fixture.ts +++ b/packages/cultcache-ts/test/union-surface-fixture.ts @@ -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 }; + }, }; } diff --git a/packages/cultcache-ts/test/update.test.ts b/packages/cultcache-ts/test/update.test.ts new file mode 100644 index 00000000..9b64e9db --- /dev/null +++ b/packages/cultcache-ts/test/update.test.ts @@ -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(body: (path: string) => Promise): Promise { + 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["withRegistry"]>[0]; + +async function open(path: string, documents: Registry = registry): Promise { + 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 = []; + 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 = []; + 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)); + }); +});