diff --git a/packages/minerender/src/env/node/NodeCache.ts b/packages/minerender/src/env/node/NodeCache.ts index d24c68fd..54c52911 100644 --- a/packages/minerender/src/env/node/NodeCache.ts +++ b/packages/minerender/src/env/node/NodeCache.ts @@ -10,6 +10,7 @@ import { PersistentCache } from "../../cache/PersistentCache"; export class NodeCache extends PersistentCache { private initialized: Promise | undefined; + private readonly pendingWrites = new Map>(); constructor(backing: LocalStorage) { super(backing); @@ -26,7 +27,16 @@ export class NodeCache extends PersistentCache { async put(key: string, value: T): Promise { await this.ready(); - return this.backing.setItem(key, value).then(res => res.content); + const previous = this.pendingWrites.get(key); + const write = (previous ? previous.catch(() => undefined) : Promise.resolve()) + .then(() => this.backing.setItem(key, value)) + .then(() => value); + this.pendingWrites.set(key, write); + try { + return await write; + } finally { + if (this.pendingWrites.get(key) === write) this.pendingWrites.delete(key); + } } async delete(key: string): Promise { diff --git a/packages/minerender/src/env/node/NodeEnv.ts b/packages/minerender/src/env/node/NodeEnv.ts index 61722903..6fbbb6fb 100644 --- a/packages/minerender/src/env/node/NodeEnv.ts +++ b/packages/minerender/src/env/node/NodeEnv.ts @@ -38,7 +38,10 @@ export class NodeEnv implements EnvProvider { openCache(name: string, version: number): PersistentCache { return new NodeCache(NodePersist.create({ dir: name + version, - encoding: "utf8" + encoding: "utf8", + // node-persist's unref'd one-second write queue delays loads and lets Node exit before writes finish. + // NodeCache serializes writes to each key while allowing different files to write concurrently. + writeQueue: false })); } diff --git a/packages/minerender/test/node/NodeCache.test.ts b/packages/minerender/test/node/NodeCache.test.ts new file mode 100644 index 00000000..4f6f1b1b --- /dev/null +++ b/packages/minerender/test/node/NodeCache.test.ts @@ -0,0 +1,84 @@ +import test from "ava"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { NodeCache, NodeEnv, PersistentCache, shutdown } from "../../dist/node/index.js"; + +// Match node-persist's default writeQueueIntervalMs. +const WRITE_QUEUE_INTERVAL_MS = 1000; + +const cacheDirectory = mkdtempSync(join(tmpdir(), "minerender-node-cache-test-")); +class TestNodeEnv extends NodeEnv { + openCache(name: string, version: number) { + return super.openCache(join(cacheDirectory, name), version); + } +} + +test.after.always(() => { + shutdown(); + rmSync(cacheDirectory, { recursive: true, force: true }); +}); + +test("persistent puts resolve without waiting for the write queue interval", async t => { + const cache = new TestNodeEnv().openCache("cache", PersistentCache.VERSION); + const start = performance.now(); + t.is(await cache.put("first", 1), 1); + let writing = true; + const second = cache.put("second", 2).finally(() => { writing = false; }); + await new Promise(resolve => setImmediate(resolve)); + t.true(writing, "the second put is still writing when the third starts"); + const third = cache.put("third", 3); + await Promise.all([second, third]); + const elapsed = performance.now() - start; + t.true(elapsed < WRITE_QUEUE_INTERVAL_MS / 2, `three puts took ${elapsed.toFixed(1)} ms`); + t.is(await cache.get("first"), 1); + t.is(await cache.get("second"), 2); + t.is(await cache.get("third"), 3); +}); + +test("overlapping same-key puts integrate with disk storage and reopening", async t => { + const env = new TestNodeEnv(); + const cache = env.openCache("overlap", PersistentCache.VERSION); + await cache.get("initialize"); + await Promise.all([ + cache.put("same", "a".repeat(2 * 1024 * 1024)), + cache.put("same", "small-last") + ]); + const reopened = env.openCache("overlap", PersistentCache.VERSION); + t.is(await reopened.get("same"), "small-last"); +}); + +test("failed puts release queued and later writes without blocking other keys", async t => { + const failure = new Error("Write failed"); + let rejectFirst!: (error: Error) => void; + let firstStarted!: () => void; + const started = new Promise(resolve => { firstStarted = resolve; }); + const calls: string[] = []; + const cache = new NodeCache({ + init: async () => {}, + setItem: async (key: string, value: string) => { + calls.push(`${key}:${value}`); + if (value === "first") { + firstStarted(); + await new Promise((_, reject) => { rejectFirst = reject; }); + } + return { content: { key, value } }; + } + } as never); + const first = cache.put("same", "first"); + const rejected = t.throwsAsync(first, { is: failure }); + await started; + const second = cache.put("same", "second"); + try { + t.is(await cache.put("other", "independent"), "independent"); + t.deepEqual(calls, ["same:first", "other:independent"]); + } finally { + rejectFirst(failure); + await rejected; + await second; + } + t.is(await second, "second"); + t.is(await cache.put("same", "retry"), "retry"); + t.deepEqual(calls, ["same:first", "other:independent", "same:second", "same:retry"]); + t.is((cache as unknown as { pendingWrites: Map }).pendingWrites.size, 0); +});