Hold PERSIST directory ownership for the life of the server (#173)

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

main.ts

@@ -46,6 +46,10 @@ 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) {
+    await PERSIST_ENGINE.acquireOwnership(() =>
+        writeLog(`Waiting for another server to release ${CONFIG.persistDir}...`)
+    );
+
     writeLog("Loading in data from persist.dat...\n");
 
     MANAGER.load();

src/persist.ts

@@ -45,6 +45,7 @@ export interface QueueStore<T = string> {
 export class FileStore<T extends JsonValue = string> implements QueueStore<T> {
     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<T extends JsonValue = string> implements QueueStore<T> {
         return this.path + ".tmp";
     }
 
+    private get ownerLockPath(): string {
+        return this.directory + "persist.lock";
+    }
+
     private ensureDirectory(): void {
         if (this.directory === "") {
             return;
@@ -253,13 +258,50 @@ export class FileStore<T extends JsonValue = string> implements QueueStore<T> {
         }
     }
 
+    // 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<void> {
+        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(/\/$/, '') + "/";
     }
 
     public close(): void {
         this.closeWriteHandle();
+        if (this.ownerHandle !== null) {
+            this.ownerHandle.close();
+            this.ownerHandle = null;
+        }
     }
 }
 

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

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/[email protected]";
+
+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<number>;
+}
+
+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<number>((resolve) => notifyListening = resolve);
+    const drain = async (stream: ReadableStream<Uint8Array>) => {
+        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<number | "waiting" | "exited"> {
+    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<never>((_, 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<Deno.CommandStatus> {
+    try { server.child.kill(signal); } catch { /* already exited */ }
+    return await server.child.status;
+}
+
+async function enqueue(port: number, payload: string): Promise<number> {
+    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<unknown> {
+    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,
+});