From 0010ee92a2515ecf2bf726afad26a36b8143de2a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 20 Oct 2026 14:50:18 +0200 Subject: [PATCH] cultcache-ts: single-file store runs every write under the .lock cross-process lock #enqueue takes withStoreWriteLock; the legacy heal moves out of pullAll's read into a queued, locked operation that re-reads; push and delete read through #readDisk so they never enqueue from inside the locked queue (a self-deadlock). Builds. Co-Authored-By: Claude Opus 5.5 --- .../single-file-messagepack-backing-store.ts | 40 +++-- packages/cultcache-ts/test/store-file.test.ts | 152 +++++++++++++++++- 2 files changed, 178 insertions(+), 14 deletions(-) diff --git a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts index 604c9a0d..f10b146b 100644 --- a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts +++ b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts @@ -6,7 +6,7 @@ import { z } from "zod"; import { documentPayloadBytes, documentPayloadFromBytes } from "./document"; import { checkPayloadDepth } from "./payload-depth"; -import { commitStoreFile } from "./store-file"; +import { commitStoreFile, withStoreWriteLock } from "./store-file"; import { STORE_FORMAT_VERSION, isV1Snapshot, requireV1RecordSlots } from "./store-format"; import type { CacheBackingStore, @@ -44,17 +44,32 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { } async pullAll(): Promise { + const { envelopes, healLegacy } = await this.#readDisk(); + if (healLegacy) { + // The heal decides from its own read, taken under the store lock. + await this.#enqueue(async () => { + const again = await this.#readDisk(); + if (again.healLegacy) { + await this.#writeAll(again.envelopes); + } + }); + } + + return envelopes; + } + + async #readDisk(): Promise<{ envelopes: CultCacheEnvelope[]; healLegacy: boolean }> { // An empty file is not a store, since no CultCache writer leaves one, so it is // decoded and refused. Only a read that finds nothing at the path is an empty store. const data = await readStore(this.filePath); if (data === undefined) { - return []; + return { envelopes: [], healLegacy: false }; } const decoded = decode(data); const snapshot = decodeSnapshot(decoded); if (snapshot) { - return snapshot.records.map((record) => { + const envelopes = snapshot.records.map((record) => { const catalogEntry = resolveCatalogEntryForRecord(record, snapshot.catalogBySchemaId); if (!catalogEntry) { throw new Error(`CultCache persisted record "${record.key}" references missing schema id "${record.schemaId}".`); @@ -69,6 +84,7 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { storedAt: record.storedAt, }; }); + return { envelopes, healLegacy: false }; } const legacy = decodeLegacyEnvelopeArray(decoded); @@ -90,11 +106,7 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { payload: documentPayloadFromBytes(entry.payload), })); - if (repairedLegacyPayload) { - await this.#writeAll(parsed); - } - - return parsed; + return { envelopes: parsed, healLegacy: repairedLegacyPayload }; } throw new Error(`CultCache file ${this.filePath} is not a recognized CultCache MessagePack store.`); @@ -102,7 +114,7 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { async push(entry: CultCacheEnvelope): Promise { await this.#enqueue(async () => { - const existing = await this.pullAll(); + const existing = (await this.#readDisk()).envelopes; const filtered = existing.filter( (candidate) => !(candidate.type === entry.type && candidate.key === entry.key), ); @@ -113,7 +125,7 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { async delete(entry: CultCacheEnvelope): Promise { await this.#enqueue(async () => { - const existing = await this.pullAll(); + const existing = (await this.#readDisk()).envelopes; const filtered = existing.filter( (candidate) => !(candidate.type === entry.type && candidate.key === entry.key), ); @@ -134,9 +146,11 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { async #enqueue(operation: () => Promise): Promise { let result!: T; - const next = this.#writeQueue.then(async () => { - result = await operation(); - }); + const next = this.#writeQueue.then(() => + withStoreWriteLock(this.filePath, async () => { + result = await operation(); + }), + ); this.#writeQueue = next.then( () => undefined, diff --git a/packages/cultcache-ts/test/store-file.test.ts b/packages/cultcache-ts/test/store-file.test.ts index 25874f4e..46c9d3cd 100644 --- a/packages/cultcache-ts/test/store-file.test.ts +++ b/packages/cultcache-ts/test/store-file.test.ts @@ -1,8 +1,16 @@ import assert from "node:assert/strict"; +import { execFile } from "node:child_process"; import { mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { join, resolve } from "node:path"; import test from "node:test"; +import { promisify } from "node:util"; + +import { decode, encode } from "@msgpack/msgpack"; + +import { documentPayloadFromBytes } from "../src/document"; +import { SingleFileMessagePackBackingStore } from "../src/single-file-messagepack-backing-store"; +import type { CultCacheEnvelope } from "../src/types"; import { MOVEFILE_REPLACE_WRITE_THROUGH, @@ -11,9 +19,12 @@ import { nodeStoreFileOps, replaceStoreFile, replaceStoreFileWindows, + withStoreWriteLock, } from "../src/store-file"; import type { ReplaceKernel, StoreFileOps } from "../src/store-file"; +const execFileAsync = promisify(execFile); + async function scratch(): Promise { return mkdtemp(join(tmpdir(), "cultcache-store-file-")); } @@ -286,3 +297,142 @@ test( } }, ); + +// --------------------------------------------------------------------------------------------- +// The cross-process write lock (TypeScript against TypeScript; C# and Rust are in cult-cache.test.ts). + +const entryOf = (key: string): CultCacheEnvelope => ({ + key, + type: "t", + payload: documentPayloadFromBytes(Uint8Array.of(0x90)), + storedAt: "2026-10-20T00:00:00Z", +}); + +function sleepFor(milliseconds: number): Promise { + return new Promise((resolveSleep) => setTimeout(resolveSleep, milliseconds)); +} + +function within(promise: Promise, milliseconds: number, what: string): Promise { + let timer: NodeJS.Timeout | undefined; + const expired = new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`${what} did not finish within ${milliseconds} ms (deadlock)`)), milliseconds); + }); + return Promise.race([promise, expired]).finally(() => clearTimeout(timer)); +} + +test("a second writer in this process waits for the store lock and runs after release", async () => { + const directory = await scratch(); + try { + const store = join(directory, "store.cc"); + const order: string[] = []; + let second: Promise | undefined; + await withStoreWriteLock(store, async () => { + second = withStoreWriteLock(store, async () => { + order.push("second"); + }); + await sleepFor(150); + order.push("first-done"); + }); + await second; + assert.deepEqual(order, ["first-done", "second"]); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + +test("the store lock gives up after its wait bound and names the lock path", async () => { + const directory = await scratch(); + try { + const store = join(directory, "store.cc"); + await withStoreWriteLock(store, async () => { + await assert.rejects( + withStoreWriteLock(store, async () => undefined, { timeoutMs: 200 }), + (error: Error) => error.message.includes(`${store}.lock`) && /200 ms/.test(error.message), + ); + }); + // The failed waiter released nothing it did not hold: the lock is free again. + assert.equal(await withStoreWriteLock(store, async () => "free", { timeoutMs: 200 }), "free"); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + +test("the store lock is released when the operation throws", async () => { + const directory = await scratch(); + try { + const store = join(directory, "store.cc"); + await assert.rejects(withStoreWriteLock(store, async () => Promise.reject(new Error("boom"))), /boom/); + assert.equal(await withStoreWriteLock(store, async () => "free", { timeoutMs: 200 }), "free"); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + +const legacyObjectPayload = (): Uint8Array => + encode([{ key: "note:hello", type: "note", payload: { title: "Hello" }, storedAt: "2026-10-20T00:00:00Z" }]); + +test("an open that heals a legacy store waits for the store lock and heals from its own read", async () => { + const directory = await scratch(); + try { + const path = join(directory, "legacy.cc"); + const legacy = legacyObjectPayload(); + await writeFile(path, legacy); + const store = new SingleFileMessagePackBackingStore(path); + let healed: Promise | undefined; + let settled = false; + await withStoreWriteLock(path, async () => { + healed = store.pullAll().then(() => { + settled = true; + }); + await sleepFor(300); + assert.equal(settled, false, "the heal must wait for the lock"); + assert.deepEqual([...(await readFile(path))], [...legacy], "the file keeps its legacy bytes while the lock is held elsewhere"); + }); + await within(healed!, 5000, "the heal"); + assert.equal((decode(await readFile(path)) as unknown[])[0], "cultcache.store.v1"); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + +test("push and delete over a legacy store that needs healing finish instead of waiting on their own lock", async () => { + const directory = await scratch(); + try { + const path = join(directory, "legacy.cc"); + await writeFile(path, legacyObjectPayload()); + const store = new SingleFileMessagePackBackingStore(path); + await within(store.push(entryOf("a")), 5000, "push"); + await writeFile(path, legacyObjectPayload()); + await within(store.delete(entryOf("a")), 5000, "delete"); + const keys = (await store.pullAll()).map((entry) => entry.key); + assert.deepEqual(keys, ["note:hello"]); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + +test("two TypeScript processes pushing distinct keys to one store lose none", async () => { + const directory = await scratch(); + try { + const path = join(directory, "store.cc"); + const dist = resolve(__dirname, "..", "..", "dist"); + const script = ` + const { SingleFileMessagePackBackingStore } = require(${JSON.stringify(join(dist, "single-file-messagepack-backing-store.js"))}); + const { documentPayloadFromBytes } = require(${JSON.stringify(join(dist, "document.js"))}); + (async () => { + const store = new SingleFileMessagePackBackingStore(${JSON.stringify(path)}); + for (let i = 0; i < 40; i++) { + await store.push({ key: process.argv[1] + "-" + i, type: "t", payload: documentPayloadFromBytes(Uint8Array.of(0x90)), storedAt: "2026-10-20T00:00:00Z" }); + } + })().catch((error) => { console.error(error); process.exit(1); }); + `; + await Promise.all( + ["p", "q"].map((name) => execFileAsync(process.execPath, ["-e", script, name], { timeout: 120_000 })), + ); + const keys = (await new SingleFileMessagePackBackingStore(path).pullAll()).map((entry) => entry.key).sort(); + assert.equal(keys.length, 80); + assert.equal(new Set(keys).size, 80); + } finally { + await rm(directory, { recursive: true, force: true }); + } +});