Durable atomic writer with fsync and per-chunk download progress (closes #39)

The atomic writer fsyncs the staged temp file before rename and the directory after, and is exported for reuse. downloadFile/downloadThumbnail gain an optional per-chunk onProgress hook (non-decreasing, final equals bytesWritten; no-op when absent). Retry and TAG_FINAL checks unchanged.

Model: opus-4-8

src/download/index.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { rename, rm, writeFile } from "node:fs/promises";
+import { open, rename, rm } from "node:fs/promises";
 import { dirname, join } from "node:path";
 import {
     fromBase64,
@@ -19,12 +19,20 @@ export interface DownloadResult {
     bytesWritten: number;
 }
 
+// Fired as decrypted plaintext accumulates, with the running total of
+// plaintext bytes recovered so far. Within one download it is non-decreasing
+// and its last value equals the final `bytesWritten`. A retry restarts the
+// file from byte zero (see `fetchAndDecrypt`), so a fresh attempt begins its
+// own count from zero.
+export type ProgressCallback = (bytesDone: number) => void;
+
 const ENC_CHUNK_SIZE = STREAM_CHUNK_SIZE + STREAM_CHUNK_OVERHEAD;
 
 const streamDecrypt = async (
     stream: ReadableStream<Uint8Array>,
     header: Uint8Array,
     key: Uint8Array,
+    onProgress?: ProgressCallback,
 ): Promise<Uint8Array> => {
     const state = initStreamPull(header, key);
     const reader = stream.getReader();
@@ -51,6 +59,7 @@ const streamDecrypt = async (
             totalPlain += plaintext.length;
             chunksPulled++;
             lastTag = tag;
+            onProgress?.(totalPlain);
         }
 
         if (done) {
@@ -76,6 +85,7 @@ const streamDecrypt = async (
                 totalPlain += pulled.plaintext.length;
                 chunksPulled++;
                 lastTag = pulled.tag;
+                onProgress?.(totalPlain);
             }
             break;
         }
@@ -107,21 +117,47 @@ const streamDecrypt = async (
     return result;
 };
 
-// Write `plaintext` to `destination` atomically: stage it in a temporary
-// sibling file (same directory, so the rename cannot cross a filesystem
-// boundary) and rename it into place. Callers therefore never observe a
-// partially written destination, and a pre-existing file at that path is
-// replaced only once the new contents are complete on disk.
-const writeAtomic = async (
+// Write `plaintext` to `destination` atomically and durably: stage it in a
+// temporary sibling file (same directory, so the rename cannot cross a
+// filesystem boundary) and rename it into place. Callers therefore never
+// observe a partially written destination, and a pre-existing file at that
+// path is replaced only once the new contents are complete on disk.
+//
+// Durability against a power cut needs two fsyncs. Without them the write can
+// return while the data or the rename is still only in the kernel's page
+// cache, and a crash then resurrects an empty renamed file — exactly the
+// corruption a later backup run treats as a complete download. So the temp
+// file's contents are fsynced before the rename, and the containing directory
+// is fsynced after it, so both the bytes and the new directory entry are on
+// stable storage before this returns.
+//
+// Exported so the metadata store can reuse the same durable write.
+export const writeAtomic = async (
     destination: string,
     plaintext: Uint8Array,
 ): Promise<void> => {
+    const dir = dirname(destination);
     // The random suffix keeps concurrent downloads of the same destination
     // from stepping on each other's temporary file.
-    const tmpPath = join(dirname(destination), `.quak-${randomUUID()}.tmp`);
+    const tmpPath = join(dir, `.quak-${randomUUID()}.tmp`);
     try {
-        await writeFile(tmpPath, plaintext);
+        const handle = await open(tmpPath, "w");
+        try {
+            await handle.writeFile(plaintext);
+            await handle.sync();
+        } finally {
+            await handle.close();
+        }
         await rename(tmpPath, destination);
+        // Fsync the directory so the rename itself survives a crash: renaming
+        // over a synced temp file still leaves the new directory entry in the
+        // page cache until the directory is synced.
+        const dirHandle = await open(dir, "r");
+        try {
+            await dirHandle.sync();
+        } finally {
+            await dirHandle.close();
+        }
     } catch (err) {
         // Best-effort cleanup. A failure to remove the temporary file must
         // never replace the error that actually explains what went wrong.
@@ -150,16 +186,18 @@ const fetchAndDecrypt = async (
     openStream: () => Promise<ReadableStream<Uint8Array>>,
     header: Uint8Array,
     key: Uint8Array,
+    onProgress?: ProgressCallback,
 ): Promise<Uint8Array> =>
     withRetry(async () => {
         const stream = await openStream();
-        return streamDecrypt(stream, header, key);
+        return streamDecrypt(stream, header, key, onProgress);
     }, api.getRetryOptions());
 
 export const downloadFile = async (
     api: ApiClient,
     file: EnteFile,
     outPath?: string,
+    onProgress?: ProgressCallback,
 ): Promise<DownloadResult> => {
     const resolvedPath = outPath ?? file.metadata.title;
     const header = fromBase64(file.file.decryptionHeader);
@@ -168,6 +206,7 @@ export const downloadFile = async (
         () => api.getFileStream(file.id, { retry: false }),
         header,
         file.key,
+        onProgress,
     );
     // Outside the retry, deliberately: only the attempt that produced a
     // complete, authenticated plaintext gets to stage a temporary file, so a
@@ -181,6 +220,7 @@ export const downloadThumbnail = async (
     api: ApiClient,
     file: EnteFile,
     outPath?: string,
+    onProgress?: ProgressCallback,
 ): Promise<DownloadResult> => {
     const resolvedPath = outPath ?? `thumb_${file.metadata.title}`;
     const header = fromBase64(file.thumbnail.decryptionHeader);
@@ -189,6 +229,7 @@ export const downloadThumbnail = async (
         () => api.getThumbnailStream(file.id, { retry: false }),
         header,
         file.key,
+        onProgress,
     );
     await writeAtomic(resolvedPath, plaintext);
     return { path: resolvedPath, bytesWritten: plaintext.length };

test/download/download.test.ts

@@ -72,7 +72,11 @@ import { init, toBase64, STREAM_CHUNK_SIZE } from "../../src/crypto/index.js";
 import { ApiClient } from "../../src/api/client.js";
 import { ApiError, TruncatedStreamError } from "../../src/errors.js";
 import type { RetryOptions } from "../../src/retry.js";
-import { downloadFile, downloadThumbnail } from "../../src/download/index.js";
+import {
+    downloadFile,
+    downloadThumbnail,
+    writeAtomic,
+} from "../../src/download/index.js";
 import type { EnteFile, FileMetadata } from "../../src/model/types.js";
 
 // ---------------------------------------------------------------------------
@@ -101,17 +105,48 @@ const renameHook = vi.hoisted(() => ({
     failWith: null as Error | null,
 }));
 
+/**
+ * `open` is wrapped so the tests can observe the durability fsyncs the atomic
+ * writer performs — which are otherwise invisible: an fsync leaves no trace in
+ * the file's contents. Each `FileHandle.sync()` is recorded, and rename and
+ * sync events are appended to a single ordered `events` log so a test can pin
+ * the sequence "fsync the temp file, rename, fsync the directory" that makes a
+ * write survive a power cut. The flag the handle was opened with distinguishes
+ * the temp file (`w`) from its containing directory (`r`).
+ */
+const durabilityHook = vi.hoisted(() => ({
+    events: [] as string[],
+}));
+
 vi.mock("node:fs/promises", async (importOriginal) => {
     const actual = await importOriginal<typeof import("node:fs/promises")>();
     const { existsSync: sourceExists } = await import("node:fs");
     return {
         ...actual,
+        open: async (
+            path: Parameters<typeof actual.open>[0],
+            flags?: Parameters<typeof actual.open>[1],
+            ...rest: unknown[]
+        ): Promise<Awaited<ReturnType<typeof actual.open>>> => {
+            const handle = await actual.open(
+                path,
+                flags as Parameters<typeof actual.open>[1],
+                ...(rest as []),
+            );
+            const realSync = handle.sync.bind(handle);
+            handle.sync = async (): Promise<void> => {
+                durabilityHook.events.push(`sync:${String(flags)}:${path}`);
+                await realSync();
+            };
+            return handle;
+        },
         rename: async (from: string, to: string): Promise<void> => {
             renameHook.calls.push({
                 from,
                 to,
                 sourceExisted: sourceExists(from),
             });
+            durabilityHook.events.push(`rename:${to}`);
             if (renameHook.failWith !== null) {
                 throw renameHook.failWith;
             }
@@ -123,6 +158,7 @@ vi.mock("node:fs/promises", async (importOriginal) => {
 beforeEach(() => {
     renameHook.calls.length = 0;
     renameHook.failWith = null;
+    durabilityHook.events.length = 0;
 });
 
 let testDir: string;
@@ -1061,3 +1097,102 @@ describe("download retries: corruption is not retried", () => {
         expect(requests()).toBe(1);
     });
 });
+
+// ---------------------------------------------------------------------------
+// Durable atomic writes
+//
+// `writeAtomic` is exported so the metadata store can reuse the same
+// power-cut-safe write. Its durability is the point: the bytes and the new
+// directory entry must both be on stable storage before it returns, so a crash
+// immediately afterwards cannot resurrect an empty renamed file (#22 area 1).
+// ---------------------------------------------------------------------------
+
+describe("writeAtomic", () => {
+    it("fsyncs the temp file before the rename and the directory after", async () => {
+        const dir = mkdtempSync(join(testDir, "atomic-"));
+        const dest = join(dir, "durable.bin");
+        const bytes = patternBytes(2048, 71);
+
+        await writeAtomic(dest, bytes);
+
+        expect(readFileSync(dest)).toEqual(Buffer.from(bytes));
+        // The order is the durability contract: fsync the staged temp file so
+        // its contents are on disk, rename it into place, then fsync the
+        // directory so that new entry is on disk too. Do the directory fsync
+        // before the rename, or skip it, and a crash can lose the rename.
+        expect(durabilityHook.events).toHaveLength(3);
+        expect(durabilityHook.events[0]).toMatch(/^sync:w:.*\.tmp$/);
+        expect(durabilityHook.events[1]).toBe(`rename:${dest}`);
+        expect(durabilityHook.events[2]).toBe(`sync:r:${dir}`);
+    });
+
+    it("leaves no temp file behind when the write cannot be renamed", async () => {
+        const dir = mkdtempSync(join(testDir, "atomic-fail-"));
+        const dest = join(dir, "unrenamable.bin");
+        renameHook.failWith = new Error("simulated rename failure");
+
+        await expect(writeAtomic(dest, patternBytes(64, 72))).rejects.toThrow(
+            "simulated rename failure",
+        );
+
+        // The staged temp file was fsynced, then the rename failed; the cleanup
+        // path must remove it so a repeatedly failing write cannot fill the disk.
+        expect(existsSync(dest)).toBe(false);
+        expect(readdirSync(dir)).toEqual([]);
+    });
+});
+
+// ---------------------------------------------------------------------------
+// Per-chunk progress
+//
+// Callers streaming a large file want bytes-written as it lands, not only the
+// final total. The hook fires as decrypted plaintext accumulates; its values
+// are non-decreasing and its last value is exactly `bytesWritten`.
+// ---------------------------------------------------------------------------
+
+describe.each(entryPoints)("$name progress", ({ name, download }) => {
+    it("reports monotonic progress ending at bytesWritten", async () => {
+        // The multi-chunk fixture pulls one full 4 MiB chunk and then a small
+        // final chunk, so the callback fires more than once and monotonicity is
+        // actually observable rather than trivially true for a single fire.
+        const { api, file } = fixtureFor(
+            multiChunkKey,
+            multiChunk.header,
+            multiChunk.body,
+        );
+        const outPath = join(
+            mkdtempSync(join(testDir, `${name}-progress-`)),
+            "p.bin",
+        );
+        const seen: number[] = [];
+
+        const result = await download(api, file, outPath, (bytesDone) => {
+            seen.push(bytesDone);
+        });
+
+        expect(seen.length).toBeGreaterThan(1);
+        for (let i = 1; i < seen.length; i++) {
+            expect(seen[i]!).toBeGreaterThan(seen[i - 1]!);
+        }
+        expect(seen[seen.length - 1]).toBe(result.bytesWritten);
+        expect(result.bytesWritten).toBe(multiChunk.plaintext.length);
+    });
+
+    it("downloads normally when no progress callback is given", async () => {
+        // The callback is optional and its absence must be side-effect-free:
+        // the download succeeds exactly as it does elsewhere in this file.
+        const plaintext = patternBytes(300, 73);
+        const key = sodium.crypto_secretstream_xchacha20poly1305_keygen();
+        const { header, ciphertext } = encryptFileBody(plaintext, key);
+        const { api, file } = fixtureFor(key, header, ciphertext);
+        const outPath = join(
+            mkdtempSync(join(testDir, `${name}-noprog-`)),
+            "n.bin",
+        );
+
+        const result = await download(api, file, outPath);
+
+        expect(result.bytesWritten).toBe(plaintext.length);
+        expectSameBytes(readFileSync(outPath), plaintext);
+    });
+});