diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 869839a..d9b4376 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -79,7 +79,7 @@ jobs: name: Mutation testing (Stryker) needs: mutation-mutasaurus runs-on: ubuntu-latest - timeout-minutes: 45 + timeout-minutes: 60 steps: - name: Check out repository uses: actions/checkout@v5 diff --git a/.gitignore b/.gitignore index 17174e4..2cbb2c9 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ /persist.dat /persist.dat.tmp +/persist.lock /queue /target/ /node_modules/ diff --git a/README.md b/README.md index c7066ba..4d02450 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. On startup the server takes an exclusive lock on `persist.lock` in that directory and holds it until it exits; a second `--persist` server pointed at the same directory exits with `PERSIST directory is already in use` instead of serving the same items twice. For a rolling replace, stop the old process before starting the new one. The operating system releases the lock if the process crashes, so a restart is never blocked by a stale lock file. + 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..90f6b2d 100644 --- a/main.ts +++ b/main.ts @@ -46,6 +46,13 @@ const MANAGER = new QueueManager(PERSIST_ENGINE, CONFIG.queueDepthLimit // Load up any existing queue data, if we're persisting if (PERSIST_ENGINE instanceof Persistency.FileStore) { + try { + PERSIST_ENGINE.claim(); + } catch (error) { + writeLog(`Failed to start: ${error instanceof Error ? error.message : error}`); + Deno.exit(1); + } + writeLog("Loading in data from persist.dat...\n"); MANAGER.load(); diff --git a/src/persist.ts b/src/persist.ts index bbeb34b..1246252 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -41,10 +41,18 @@ export interface QueueStore { close(): void; } +export class PersistDirectoryInUseError extends Error { + constructor(path: string) { + super(`PERSIST directory is already in use by another queue process (lock held on ${path})`); + this.name = "PersistDirectoryInUseError"; + } +} + /** Persists payloads as JSON; T must be JSON-compatible. */ export class FileStore implements QueueStore { private directory: string = ''; private writeHandle: Deno.FsFile | null = null; + private claimHandle: Deno.FsFile | null = null; private encoder = new TextEncoder(); private get path(): string { @@ -55,6 +63,10 @@ export class FileStore implements QueueStore { return this.path + ".tmp"; } + private get lockPath(): string { + return this.directory + "persist.lock"; + } + private ensureDirectory(): void { if (this.directory === "") { return; @@ -88,6 +100,13 @@ export class FileStore implements QueueStore { } } + private releaseClaim(): void { + if (this.claimHandle !== null) { + this.claimHandle.close(); + this.claimHandle = null; + } + } + private removeTempFile(): void { try { Deno.removeSync(this.tempPath); @@ -253,13 +272,32 @@ export class FileStore implements QueueStore { } } + // Each process replays the log into its own memory, so two processes on + // one directory would deliver the same items and overwrite each other's + // snapshots. The claim is an exclusive lock held until close(); the OS + // drops it if the process dies, so a crash never blocks a restart. + public claim(): void { + if (this.claimHandle !== null) { + return; + } + this.ensureDirectory(); + const handle = Deno.openSync(this.lockPath, { write: true, create: true }); + if (!handle.tryLockSync(true)) { + handle.close(); + throw new PersistDirectoryInUseError(this.lockPath); + } + this.claimHandle = handle; + } + public dir(dir: string): void { this.closeWriteHandle(); + this.releaseClaim(); this.directory = dir.replace(/\/$/, '') + "/"; } public close(): void { this.closeWriteHandle(); + this.releaseClaim(); } } diff --git a/tests/persist_test.ts b/tests/persist_test.ts index 1d7c43c..9833e02 100644 --- a/tests/persist_test.ts +++ b/tests/persist_test.ts @@ -747,3 +747,60 @@ Deno.test("FileStore.loadState skips non-event JSON records", () => { persist.close(); Deno.removeSync(tmpDir, { recursive: true }); }); + +Deno.test("FileStore.claim() refuses a directory another store has claimed until it closes", () => { + const tmpDir = Deno.makeTempDirSync(); + const owner = new Persistency.FileStore(); + const contender = new Persistency.FileStore(); + owner.dir(tmpDir); + contender.dir(tmpDir); + try { + owner.claim(); + owner.claim(); + const error = assertThrows(() => contender.claim(), Persistency.PersistDirectoryInUseError, "already in use"); + assertEquals(error.name, "PersistDirectoryInUseError"); + assertEquals(error.message.includes(tmpDir + "/persist.lock"), true); + owner.close(); + contender.claim(); + assertThrows(() => owner.claim(), Persistency.PersistDirectoryInUseError); + } finally { + owner.close(); + contender.close(); + Deno.removeSync(tmpDir, { recursive: true }); + } +}); + +Deno.test("FileStore.dir() releases the claim on the previous directory", () => { + const firstDir = Deno.makeTempDirSync(); + const secondDir = Deno.makeTempDirSync(); + const store = new Persistency.FileStore(); + const contender = new Persistency.FileStore(); + try { + store.dir(firstDir); + store.claim(); + store.dir(secondDir); + contender.dir(firstDir); + contender.claim(); + // Claims on different directories never block each other. + store.claim(); + } finally { + store.close(); + contender.close(); + Deno.removeSync(firstDir, { recursive: true }); + Deno.removeSync(secondDir, { recursive: true }); + } +}); + +Deno.test("FileStore.claim() creates a missing PERSIST directory", () => { + const tmpDir = Deno.makeTempDirSync(); + const persistDir = tmpDir + "/not/yet/created"; + const store = new Persistency.FileStore(); + store.dir(persistDir); + try { + store.claim(); + assertEquals(Deno.statSync(persistDir + "/persist.lock").isFile, true); + } finally { + store.close(); + Deno.removeSync(tmpDir, { recursive: true }); + } +}); diff --git a/tests/shared_persist_test.ts b/tests/shared_persist_test.ts new file mode 100644 index 0000000..416c448 --- /dev/null +++ b/tests/shared_persist_test.ts @@ -0,0 +1,110 @@ +import { assertEquals, assertNotEquals, assertStringIncludes } from "jsr:@std/assert@1.0"; + +// Regression coverage for #173: two `--persist` processes on one PERSIST +// directory each served the log from their own memory, so one item was +// delivered twice and a shutdown snapshot could erase the other's items. + +const TOKEN = "shared-persist-token"; +const headers = { Authorization: `Bearer ${TOKEN}`, "Content-Type": "application/json" }; + +function spawnServer(persistDir: string): Deno.ChildProcess { + return 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(); +} + +// Resolves with the bound port, or null with the output if the process +// exits before it starts listening. +async function waitForListen(child: Deno.ChildProcess): Promise<{ port: number | null; output: string }> { + const decoder = new TextDecoder(); + const reader = child.stdout.getReader(); + let output = ""; + const timeoutId = setTimeout(() => { + try { child.kill("SIGKILL"); } catch { /* ignore */ } + }, 5000); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) return { port: null, output }; + output += decoder.decode(value, { stream: true }); + const match = output.match(/Listening on (?:http:\/\/)?127\.0\.0\.1:(\d+)/); + if (match) return { port: parseInt(match[1], 10), output }; + } + } finally { + clearTimeout(timeoutId); + reader.releaseLock(); + } +} + +async function stop(child: Deno.ChildProcess): Promise { + try { await child.stdout.cancel(); } catch { /* ignore */ } + try { await child.stderr.cancel(); } catch { /* ignore */ } + try { child.kill("SIGKILL"); } catch { /* ignore */ } + try { await child.status; } catch { /* ignore */ } +} + +Deno.test("a second server on the same PERSIST directory refuses to start", async () => { + const dir = await Deno.makeTempDir(); + const first = spawnServer(dir); + let second: Deno.ChildProcess | null = null; + try { + const a = await waitForListen(first); + assertNotEquals(a.port, null, `first server did not start:\n${a.output}`); + const base = `http://127.0.0.1:${a.port}`; + + const enqueue = await fetch(`${base}/enqueue/jobs`, { + method: "POST", headers, body: JSON.stringify({ payload: "from-a" }), + }); + assertEquals(enqueue.status, 200); + await enqueue.body?.cancel(); + + second = spawnServer(dir); + const b = await waitForListen(second); + assertEquals(b.port, null, "second server bound a port on a PERSIST directory already in use"); + const status = await second.status; + const stderr = new TextDecoder().decode((await second.stderr.getReader().read()).value ?? new Uint8Array()); + assertNotEquals(status.code, 0); + assertStringIncludes(b.output + stderr, "already in use"); + + // The first server still owns the queue and delivers the item exactly once. + const dequeue = await fetch(`${base}/dequeue/jobs`, { headers }); + assertEquals(dequeue.status, 200); + assertEquals(await dequeue.json(), "from-a"); + } finally { + if (second) await stop(second); + await stop(first); + await Deno.remove(dir, { recursive: true }).catch(() => {}); + } +}); + +Deno.test("a PERSIST directory is reusable after its server stops", async () => { + const dir = await Deno.makeTempDir(); + const first = spawnServer(dir); + let second: Deno.ChildProcess | null = null; + try { + const a = await waitForListen(first); + assertNotEquals(a.port, null, `first server did not start:\n${a.output}`); + const enqueue = await fetch(`http://127.0.0.1:${a.port}/enqueue/jobs`, { + method: "POST", headers, body: JSON.stringify({ payload: "survives" }), + }); + assertEquals(enqueue.status, 200); + await enqueue.body?.cancel(); + // SIGKILL skips the shutdown flush; the OS must still release the lock. + await stop(first); + + second = spawnServer(dir); + const b = await waitForListen(second); + assertNotEquals(b.port, null, `restart on the same directory did not start:\n${b.output}`); + const dequeue = await fetch(`http://127.0.0.1:${b.port}/dequeue/jobs`, { headers }); + assertEquals(dequeue.status, 200); + assertEquals(await dequeue.json(), "survives"); + } finally { + if (second) await stop(second); + await stop(first); + await Deno.remove(dir, { recursive: true }).catch(() => {}); + } +});