feat(download): durable atomic writer and per-chunk progress hook (closes #39)

Add fsync to the atomic writer and export it, and give the download
functions an optional per-chunk progress hook.

writeAtomic now fsyncs the staged temp file before the rename and fsyncs
the containing directory after it, so a power cut immediately after the
write cannot resurrect an empty renamed file. It is exported so the
metadata store can reuse the same durable write.

downloadFile and downloadThumbnail take an optional onProgress(bytesDone)
callback that fires within streamDecrypt as decrypted plaintext
accumulates; its values are non-decreasing and the last equals
bytesWritten. Absent, the callback is a no-op. Whole-file buffering,
retry semantics, and the TAG_FINAL truncation checks are unchanged.

Covers the fsync durability gap of #22 (area 1) only; orphan reaping,
symlink/mode docs, and the remaining test coverage there stay for a
later unit, so #22 is referenced, not closed.

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
+// Write `plaintext` to `destination` atomically and durably: stage it in a
-// sibling file (same directory, so the rename cannot cross a filesystem
+// temporary sibling file (same directory, so the rename cannot cross a
-// boundary) and rename it into place. Callers therefore never observe a
+// filesystem boundary) and rename it into place. Callers therefore never
-// partially written destination, and a pre-existing file at that path is
+// observe a partially written destination, and a pre-existing file at that
-// replaced only once the new contents are complete on disk.
+// path is replaced only once the new contents are complete on disk.
-const writeAtomic = async (
+//
+// 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);
+    });
+});