Fix short writes on a full PERSIST volume losing acknowledged items

main.ts

@@ -69,7 +69,14 @@ async function shutdown(signal: string): Promise<void> {
     // If we're using file persistency, save all current state to persistant storage
     if (PERSIST_ENGINE instanceof Persistency.FileStore) {
         writeLog("Flushing data to persist.dat...\n");
-        MANAGER.save();
+        try {
+            MANAGER.save();
+        } catch (error) {
+            // The previous persist.dat is left in place; it still holds every logged item.
+            writeLog(`Failed to flush data to persist.dat: ${error instanceof Error ? error.message : error}`);
+            PERSIST_ENGINE.close();
+            Deno.exit(1);
+        }
     }
 
     PERSIST_ENGINE.close();

src/manager.ts

@@ -110,15 +110,17 @@ export default class Manager<T = string> {
         if (queue.length >= this.queueDepthLimit) {
             throw new Error("Queue depth limit reached");
         }
+        // Log before mutating, so an item that cannot be persisted is
+        // rejected rather than held only in memory.
+        if (this.persistEnabled) {
+            this.store.saveEvent(name, payload, true);
+        }
+
         if (!existing) {
             this.register(name, queue);
         }
         queue.push(payload);
 
-        if (this.persistEnabled) {
-            this.store.saveEvent(name, payload, true);
-        }
-
         return this;
     }
 
@@ -129,17 +131,20 @@ export default class Manager<T = string> {
             return undefined;
         }
 
-        const wasNonEmpty = queue.length > 0;
-        const payload = queue.shift();
-
-        if (wasNonEmpty && queue.length === 0) {
-            this.queues.delete(name);
+        const payload = queue.peek();
+        if (payload === undefined) {
+            return undefined;
         }
 
-        if (payload !== undefined && this.persistEnabled) {
+        if (this.persistEnabled) {
             this.store.saveEvent(name, payload, false);
         }
 
+        queue.shift();
+        if (queue.length === 0) {
+            this.queues.delete(name);
+        }
+
         return payload;
     }
 

src/persist.ts

@@ -108,30 +108,52 @@ export class FileStore<T extends JsonValue = string> implements QueueStore<T> {
     }
 
     public saveEvent(queueName: string, payload: T, isEnqueue: boolean): void {
-        this.ensureOpen();
-        this.writeHandle!.lockSync(true);
-        try {
-            this.writeHandle!.writeSync(this.serialize({
-                queue: queueName,
-                payload: payload,
-                enqueue: isEnqueue,
-                dequeue: !isEnqueue,
-            }));
-        } finally {
-            this.writeHandle!.unlockSync();
-        }
+        this.append([{
+            queue: queueName,
+            payload: payload,
+            enqueue: isEnqueue,
+            dequeue: !isEnqueue,
+        }]);
     }
 
     public saveBatch(events: Array<QueueEvent<T>>): void {
         if (events.length === 0) return;
+        this.append(events);
+    }
+
+    // writeSync() may write fewer bytes than requested (e.g. on a full
+    // volume) without throwing, so keep writing until every byte is down
+    // or the OS reports the error.
+    private writeAll(file: Deno.FsFile, bytes: Uint8Array): void {
+        let offset = 0;
+        while (offset < bytes.length) {
+            const written = file.writeSync(bytes.subarray(offset));
+            if (written === 0) {
+                throw new Error("Failed to write to " + this.path + ": wrote 0 bytes");
+            }
+            offset += written;
+        }
+    }
+
+    // A failed append is rolled back to the previous end of the log, so a
+    // partial line can never be glued onto the next appended event.
+    private append(events: Array<QueueEvent<T>>): void {
         this.ensureOpen();
-        this.writeHandle!.lockSync(true);
+        const handle = this.writeHandle!;
+        handle.lockSync(true);
         try {
-            for (const event of events) {
-                this.writeHandle!.writeSync(this.serialize(event));
+            const end = handle.statSync().size;
+            try {
+                for (const event of events) {
+                    this.writeAll(handle, this.serialize(event));
+                }
+            } catch (error) {
+                handle.truncateSync(end);
+                handle.seekSync(end, Deno.SeekMode.Start);
+                throw error;
             }
         } finally {
-            this.writeHandle!.unlockSync();
+            handle.unlockSync();
         }
     }
 
@@ -144,7 +166,7 @@ export class FileStore<T extends JsonValue = string> implements QueueStore<T> {
             temp.lockSync(true);
             try {
                 for (const event of events) {
-                    temp.writeSync(this.serialize(event));
+                    this.writeAll(temp, this.serialize(event));
                 }
                 temp.syncSync();
             } finally {

tests/short_write_test.ts

@@ -0,0 +1,147 @@
+import { assertEquals, assertNotEquals } from "jsr:@std/assert@1.0";
+
+// Regression tests for #160: short writes to persist.dat on a full PERSIST
+// volume. `ulimit -f` caps the size of any file the server writes, which
+// makes the kernel perform a real short write (then fail with EFBIG) exactly
+// as a full disk does, without needing root to mount a tiny volume.
+// SIGXFSZ is ignored so the write fails instead of killing the process.
+
+const TOKEN = "short-write-token";
+const AUTH = { "Authorization": `Bearer ${TOKEN}` };
+// 100 blocks is 50-100 KiB depending on the shell's block size; either way
+// it is far below LARGE_PAYLOAD and far above the small items.
+const FILE_SIZE_LIMIT_BLOCKS = 100;
+const LARGE_PAYLOAD = "x".repeat(200 * 1024);
+
+function serverCommand(persistDir: string, limitFileSize: boolean): Deno.Command {
+    const env = { HOST: "127.0.0.1", PORT: "0", PERSIST: persistDir, QUEUE_API_TOKEN: TOKEN };
+    if (!limitFileSize) {
+        return new Deno.Command(Deno.execPath(), {
+            args: ["run", "--allow-all", "main.ts", "--persist"],
+            env,
+            stdout: "piped",
+            stderr: "piped",
+        });
+    }
+    return new Deno.Command("sh", {
+        args: [
+            "-c",
+            `trap '' XFSZ; ulimit -f ${FILE_SIZE_LIMIT_BLOCKS}; exec "$0" run --allow-all main.ts --persist`,
+            Deno.execPath(),
+        ],
+        env,
+        stdout: "piped",
+        stderr: "piped",
+    });
+}
+
+async function startServer(persistDir: string, limitFileSize: boolean): Promise<{ child: Deno.ChildProcess; port: number }> {
+    const child = serverCommand(persistDir, limitFileSize).spawn();
+    const reader = child.stdout.getReader();
+    const decoder = new TextDecoder();
+    let output = "";
+    const deadline = Date.now() + 10000;
+    while (Date.now() < deadline) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        output += decoder.decode(value, { stream: true });
+        const match = output.match(/Listening on (?:http:\/\/)?127\.0\.0\.1:(\d+)/);
+        if (match) {
+            reader.releaseLock();
+            return { child, port: parseInt(match[1], 10) };
+        }
+    }
+    reader.releaseLock();
+    await stopServer(child);
+    throw new Error(`Server did not start. stdout: ${output}`);
+}
+
+async function stopServer(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 */ }
+}
+
+async function enqueue(port: number, queue: string, payload: string): Promise<number> {
+    const res = await fetch(`http://127.0.0.1:${port}/enqueue/${queue}`, {
+        method: "POST",
+        headers: { ...AUTH, "Content-Type": "application/json" },
+        body: JSON.stringify({ payload }),
+    });
+    await res.text();
+    return res.status;
+}
+
+async function length(port: number, queue: string): Promise<string> {
+    const res = await fetch(`http://127.0.0.1:${port}/length/${queue}`, { headers: AUTH });
+    return await res.text();
+}
+
+async function drain(port: number, queue: string): Promise<unknown[]> {
+    const items: unknown[] = [];
+    while (true) {
+        const res = await fetch(`http://127.0.0.1:${port}/dequeue/${queue}`, { headers: AUTH });
+        if (res.status === 204) {
+            await res.body?.cancel();
+            return items;
+        }
+        assertEquals(res.status, 200);
+        items.push(await res.json());
+    }
+}
+
+function logLine(queue: string, payload: string): string {
+    return JSON.stringify({ queue, payload, enqueue: true, dequeue: false }) + "\n";
+}
+
+Deno.test("e2e: an enqueue that cannot be fully logged is rejected and does not corrupt later items (#160)", async () => {
+    const tempDir = await Deno.makeTempDir();
+    let limited: Deno.ChildProcess | undefined;
+    let recovered: Deno.ChildProcess | undefined;
+    try {
+        const started = await startServer(tempDir, true);
+        limited = started.child;
+        assertEquals(await enqueue(started.port, "q", "before-full"), 200);
+        assertNotEquals(await enqueue(started.port, "q", LARGE_PAYLOAD), 200);
+        assertEquals(await enqueue(started.port, "q", "after-full"), 200);
+        assertEquals(await length(started.port, "q"), "2");
+        limited.kill("SIGKILL");
+        await limited.status;
+
+        const restarted = await startServer(tempDir, false);
+        recovered = restarted.child;
+        assertEquals(await drain(restarted.port, "q"), ["before-full", "after-full"]);
+    } finally {
+        if (limited) await stopServer(limited);
+        if (recovered) await stopServer(recovered);
+        await Deno.remove(tempDir, { recursive: true }).catch(() => {});
+    }
+});
+
+Deno.test("e2e: a snapshot that cannot be fully written keeps the complete persist.dat (#160)", async () => {
+    const tempDir = await Deno.makeTempDir();
+    const persistFile = `${tempDir}/persist.dat`;
+    const original = logLine("q", "small-1") + logLine("q", LARGE_PAYLOAD);
+    await Deno.writeTextFile(persistFile, original);
+    let recovered: Deno.ChildProcess | undefined;
+    try {
+        // Startup compacts the log via the same snapshot path as SIGTERM.
+        // The snapshot cannot fit under the file size limit.
+        const child = serverCommand(tempDir, true).spawn();
+        const timeoutId = setTimeout(() => child.kill("SIGKILL"), 10000);
+        const { code, stdout } = await child.output();
+        clearTimeout(timeoutId);
+        const output = new TextDecoder().decode(stdout);
+        assertEquals(output.includes("Listening on"), false, `server started without a complete snapshot:\n${output}`);
+        assertNotEquals(code, 0);
+        assertEquals(await Deno.readTextFile(persistFile), original);
+
+        const restarted = await startServer(tempDir, false);
+        recovered = restarted.child;
+        assertEquals(await drain(restarted.port, "q"), ["small-1", LARGE_PAYLOAD]);
+    } finally {
+        if (recovered) await stopServer(recovered);
+        await Deno.remove(tempDir, { recursive: true }).catch(() => {});
+    }
+});