From cce337ed8e1f4ecd44ad11528c704dc534c6c29a Mon Sep 17 00:00:00 2001 From: Jonathan Baldie Date: Sat, 10 Oct 2026 05:06:50 +0100 Subject: [PATCH] Fix short writes on a full PERSIST volume losing acknowledged items (#160) Root cause: FileStore ignored the byte count returned by Deno.FsFile.writeSync(). On a full volume the OS performs a short write and returns fewer bytes without an error, so a partial log line was reported as success. The next append was glued onto it and replay dropped both; the shutdown snapshot was fsynced and renamed over a complete persist.dat while truncated. - Write every byte or throw (EFBIG/ENOSPC surfaces on the retry). - Roll a failed append back to the previous end of the log (truncate and seek), so the log stays line-aligned. - A snapshot that cannot be fully written throws before the rename, keeping the old persist.dat; shutdown logs the error and exits 1. - Manager logs enqueue/dequeue before mutating memory, so a request that cannot be persisted fails (500) without changing queue state. Regression tests use `ulimit -f` to force real short writes without needing root to mount a tiny volume. Co-Authored-By: Claude Opus 5.5 --- main.ts | 9 ++- src/manager.ts | 25 ++++--- src/persist.ts | 56 ++++++++++----- tests/short_write_test.ts | 147 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 209 insertions(+), 28 deletions(-) create mode 100644 tests/short_write_test.ts diff --git a/main.ts b/main.ts index b258096..99ed50a 100644 --- a/main.ts +++ b/main.ts @@ -69,7 +69,14 @@ async function shutdown(signal: string): Promise { // 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(); diff --git a/src/manager.ts b/src/manager.ts index 8bce202..b48187c 100644 --- a/src/manager.ts +++ b/src/manager.ts @@ -110,15 +110,17 @@ export default class Manager { 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 { 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; } diff --git a/src/persist.ts b/src/persist.ts index 05d425f..bbeb34b 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -108,30 +108,52 @@ export class FileStore implements QueueStore { } 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>): 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>): 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 implements QueueStore { temp.lockSync(true); try { for (const event of events) { - temp.writeSync(this.serialize(event)); + this.writeAll(temp, this.serialize(event)); } temp.syncSync(); } finally { diff --git a/tests/short_write_test.ts b/tests/short_write_test.ts new file mode 100644 index 0000000..625a1d7 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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(() => {}); + } +});