Refuse a second server on a PERSIST directory already in use (#173)

.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

.gitignore

@@ -1,5 +1,6 @@
 /persist.dat
 /persist.dat.tmp
+/persist.lock
 /queue
 /target/
 /node_modules/

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).

main.ts

@@ -46,6 +46,13 @@ const MANAGER = new QueueManager<Payload>(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();

src/persist.ts

@@ -41,10 +41,18 @@ export interface QueueStore<T = string> {
     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<T extends JsonValue = string> implements QueueStore<T> {
     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<T extends JsonValue = string> implements QueueStore<T> {
         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<T extends JsonValue = string> implements QueueStore<T> {
         }
     }
 
+    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<T extends JsonValue = string> implements QueueStore<T> {
         }
     }
 
+    // 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();
     }
 }
 

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

tests/shared_persist_test.ts

@@ -0,0 +1,110 @@
+import { assertEquals, assertNotEquals, assertStringIncludes } from "jsr:@std/[email protected]";
+
+// 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<void> {
+    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(() => {});
+    }
+});