From 9a6d86ed2cf7cafbf48f7547b1ac4c94251c1671 Mon Sep 17 00:00:00 2001 From: Jonathan Baldie Date: Sun, 18 Oct 2026 08:55:46 +0100 Subject: [PATCH] Hold PERSIST directory ownership for the life of the server (#173) Root cause: each server loads persist.dat once and then serves its queues from memory, and the file lock was only held around a single append. A second server on the same directory therefore delivered the same item a second time, and the first server's shutdown snapshot (written from its own memory) erased items the second server had acknowledged. The server now takes an exclusive lock on persist.lock before loading the log and holds it until exit. A second server logs that it is waiting and only loads and listens once the first has exited, so a rolling replace hands over the old server's final snapshot. The lock lives on a separate file because persist.dat is replaced by rename, and the OS releases it if the process dies. Fixes #173 Co-Authored-By: Claude Opus 5.5 --- README.md | 2 + main.ts | 4 + src/persist.ts | 42 ++++++++++ tests/persist_test.ts | 28 +++++++ tests/shared_persist_dir_test.ts | 130 +++++++++++++++++++++++++++++++ 5 files changed, 206 insertions(+) create mode 100644 tests/shared_persist_dir_test.ts diff --git a/README.md b/README.md index c7066ba..f18f449 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,8 @@ If the server sees that the `persist.dat` file exists on startup, it will replay The Docker image runs as the non-root `deno` user and ships a `/data` directory owned by that user, declared as a volume and used as the default `PERSIST` path. Mount a named volume there to keep your binary log across container recreation, e.g. `-v queue-data:/data`. If you point `PERSIST` somewhere else, that directory must be writable by the `deno` user (uid 1993). +Only one server can use a `PERSIST` directory at a time. A server holds an exclusive lock on `persist.lock` in that directory until it exits. A second server started on the same directory logs that it is waiting, and only loads `persist.dat` and starts listening after the first server has stopped. So for a rolling replace, start the new server, then stop the old one: the new server picks up the old one's final snapshot. The lock is released automatically if a server crashes. + It should go without saying, but try not to edit `persist.dat`, because it might result in weird behaviour. Known persistence issues found during exploratory testing are recorded in [docs/exploratory-testing](docs/exploratory-testing/README.md). diff --git a/main.ts b/main.ts index 99ed50a..1ad5eb8 100644 --- a/main.ts +++ b/main.ts @@ -46,6 +46,10 @@ const MANAGER = new QueueManager(PERSIST_ENGINE, CONFIG.queueDepthLimit // Load up any existing queue data, if we're persisting if (PERSIST_ENGINE instanceof Persistency.FileStore) { + await PERSIST_ENGINE.acquireOwnership(() => + writeLog(`Waiting for another server to release ${CONFIG.persistDir}...`) + ); + writeLog("Loading in data from persist.dat...\n"); MANAGER.load(); diff --git a/src/persist.ts b/src/persist.ts index bbeb34b..284508f 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -45,6 +45,7 @@ export interface QueueStore { export class FileStore implements QueueStore { private directory: string = ''; private writeHandle: Deno.FsFile | null = null; + private ownerHandle: Deno.FsFile | null = null; private encoder = new TextEncoder(); private get path(): string { @@ -55,6 +56,10 @@ export class FileStore implements QueueStore { return this.path + ".tmp"; } + private get ownerLockPath(): string { + return this.directory + "persist.lock"; + } + private ensureDirectory(): void { if (this.directory === "") { return; @@ -253,6 +258,39 @@ export class FileStore implements QueueStore { } } + // Each server serves its queues from memory, so two servers sharing a + // directory would deliver the same item twice and overwrite each other's + // snapshots. Hold an exclusive lock on a separate file (persist.dat is + // replaced by rename, which would drop a lock on it) until close(); a + // second server waits here until the first exits. The OS releases the + // lock when the process dies, so a crash cannot leave it stale. + public async acquireOwnership(onWait: () => void, waitNoticeMs = 100): Promise { + if (this.ownerHandle !== null) { + return; + } + this.ensureDirectory(); + const handle = await Deno.open(this.ownerLockPath, { write: true, create: true }); + try { + const locked = handle.lock(true); + let timeoutId: number | undefined; + const notice = new Promise<"waiting">((resolve) => { + timeoutId = setTimeout(() => resolve("waiting"), waitNoticeMs); + }); + try { + if (await Promise.race([locked, notice]) === "waiting") { + onWait(); + await locked; + } + } finally { + clearTimeout(timeoutId); + } + } catch (error) { + handle.close(); + throw error; + } + this.ownerHandle = handle; + } + public dir(dir: string): void { this.closeWriteHandle(); this.directory = dir.replace(/\/$/, '') + "/"; @@ -260,6 +298,10 @@ export class FileStore implements QueueStore { public close(): void { this.closeWriteHandle(); + if (this.ownerHandle !== null) { + this.ownerHandle.close(); + this.ownerHandle = null; + } } } diff --git a/tests/persist_test.ts b/tests/persist_test.ts index 1d7c43c..d386b43 100644 --- a/tests/persist_test.ts +++ b/tests/persist_test.ts @@ -747,3 +747,31 @@ Deno.test("FileStore.loadState skips non-event JSON records", () => { persist.close(); Deno.removeSync(tmpDir, { recursive: true }); }); + +Deno.test("FileStore.acquireOwnership waits for the current owner to close (#173)", async () => { + const tmpDir = Deno.makeTempDirSync(); + const owner = new Persistency.FileStore(); + const next = new Persistency.FileStore(); + owner.dir(tmpDir); + next.dir(tmpDir); + try { + let ownerWaited = false; + await owner.acquireOwnership(() => ownerWaited = true); + assertEquals(ownerWaited, false); + + let notifyWaiting: () => void; + const waiting = new Promise((resolve) => notifyWaiting = resolve); + let acquired = false; + const takeover = next.acquireOwnership(() => notifyWaiting(), 10).then(() => acquired = true); + await waiting; + assertEquals(acquired, false); + + owner.close(); + await takeover; + assertEquals(acquired, true); + } finally { + owner.close(); + next.close(); + Deno.removeSync(tmpDir, { recursive: true }); + } +}); diff --git a/tests/shared_persist_dir_test.ts b/tests/shared_persist_dir_test.ts new file mode 100644 index 0000000..e29dc6f --- /dev/null +++ b/tests/shared_persist_dir_test.ts @@ -0,0 +1,130 @@ +// Regression coverage for #173: a second server pointed at the same PERSIST +// directory served the same queue from its own memory, so one item was +// delivered twice and the first server's shutdown snapshot erased items the +// second server had acknowledged. + +import { assertEquals } from "jsr:@std/assert@1.0"; + +const TOKEN = "shared-dir-token"; +const headers = { Authorization: `Bearer ${TOKEN}` }; + +interface Server { + child: Deno.ChildProcess; + output: () => string; + // Resolves with the port once the server logs that it is listening. + listening: Promise; +} + +function spawnServer(persistDir: string): Server { + const decoder = new TextDecoder(); + const child = new Deno.Command(Deno.execPath(), { + args: ["run", "--allow-all", "main.ts", "--persist"], + cwd: ".", + env: { HOST: "127.0.0.1", PORT: "0", PERSIST: persistDir, QUEUE_API_TOKEN: TOKEN }, + stdout: "piped", + stderr: "piped", + }).spawn(); + + let buf = ""; + let notifyListening: (port: number) => void; + const listening = new Promise((resolve) => notifyListening = resolve); + const drain = async (stream: ReadableStream) => { + for await (const chunk of stream) { + buf += decoder.decode(chunk, { stream: true }); + const match = buf.match(/Listening on (?:http:\/\/)?127\.0\.0\.1:(\d+)/); + if (match) notifyListening!(parseInt(match[1], 10)); + } + }; + drain(child.stdout).catch(() => {}); + drain(child.stderr).catch(() => {}); + return { child, output: () => buf, listening }; +} + +// Resolves with the first of: the port it listens on, "waiting" once it logs +// that another server holds the directory, or "exited". +async function startupOutcome(server: Server): Promise { + let intervalId: number; + const waiting = new Promise<"waiting">((resolve) => { + intervalId = setInterval(() => { + if (server.output().includes("Waiting for another server")) resolve("waiting"); + }, 10); + }); + let timeoutId: number; + const timeout = new Promise((_, reject) => { + timeoutId = setTimeout(() => reject(new Error(`no startup outcome:\n${server.output()}`)), 5000); + }); + try { + return await Promise.race([ + server.listening, + waiting, + server.child.status.then(() => "exited" as const), + timeout, + ]); + } finally { + clearInterval(intervalId!); + clearTimeout(timeoutId!); + } +} + +async function stop(server: Server, signal: Deno.Signal): Promise { + try { server.child.kill(signal); } catch { /* already exited */ } + return await server.child.status; +} + +async function enqueue(port: number, payload: string): Promise { + const res = await fetch(`http://127.0.0.1:${port}/enqueue/jobs`, { + method: "POST", + headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify({ payload }), + }); + await res.body?.cancel(); + return res.status; +} + +async function dequeue(port: number): Promise { + const res = await fetch(`http://127.0.0.1:${port}/dequeue/jobs`, { headers }); + if (res.status === 204) return null; + return await res.json(); +} + +Deno.test({ + name: "a second server on the same PERSIST dir waits for the first to exit, then takes over its items (#173)", + fn: async () => { + const persistDir = await Deno.makeTempDir({ prefix: "queue-shared-dir-" }); + const servers: Server[] = []; + try { + const first = spawnServer(persistDir); + servers.push(first); + const firstPort = await startupOutcome(first); + if (typeof firstPort !== "number") throw new Error(`first server did not start:\n${first.output()}`); + assertEquals(await enqueue(firstPort, "from-a"), 200); + assertEquals(await enqueue(firstPort, "from-a2"), 200); + + // Rolling replace: start the new server before stopping the old one. + const second = spawnServer(persistDir); + servers.push(second); + assertEquals(await startupOutcome(second), "waiting", second.output()); + assertEquals(await dequeue(firstPort), "from-a"); + + assertEquals((await stop(first, "SIGTERM")).code, 0, first.output()); + const secondPort = await second.listening; + assertEquals(await dequeue(secondPort), "from-a2", "the old server's state must be handed over"); + assertEquals(await enqueue(secondPort, "from-b"), 200); + + // Killed before it can flush: its acknowledged item is only in the log. + await stop(second, "SIGKILL"); + + const restarted = spawnServer(persistDir); + servers.push(restarted); + const restartedPort = await startupOutcome(restarted); + if (typeof restartedPort !== "number") throw new Error(`restart did not start:\n${restarted.output()}`); + assertEquals(await dequeue(restartedPort), "from-b"); + assertEquals(await dequeue(restartedPort), null); + } finally { + for (const server of servers) await stop(server, "SIGKILL"); + await Deno.remove(persistDir, { recursive: true }).catch(() => {}); + } + }, + sanitizeResources: false, + sanitizeOps: false, +});