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, +});