From 8f575550af2e071c4da3f8456d86986eeba7f77b Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 22 Sep 2026 12:01:06 +0200 Subject: [PATCH] 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 | 61 ++++++++++++--- test/download/download.test.ts | 137 ++++++++++++++++++++++++++++++++- 2 files changed, 187 insertions(+), 11 deletions(-) diff --git a/src/download/index.ts b/src/download/index.ts index bda3ba7..d0be2bd 100644 --- a/src/download/index.ts +++ b/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, header: Uint8Array, key: Uint8Array, + onProgress?: ProgressCallback, ): Promise => { 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 => { + 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>, header: Uint8Array, key: Uint8Array, + onProgress?: ProgressCallback, ): Promise => 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 => { 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 => { 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 }; diff --git a/test/download/download.test.ts b/test/download/download.test.ts index 0608381..95bcdf2 100644 --- a/test/download/download.test.ts +++ b/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(); const { existsSync: sourceExists } = await import("node:fs"); return { ...actual, + open: async ( + path: Parameters[0], + flags?: Parameters[1], + ...rest: unknown[] + ): Promise>> => { + const handle = await actual.open( + path, + flags as Parameters[1], + ...(rest as []), + ); + const realSync = handle.sync.bind(handle); + handle.sync = async (): Promise => { + durabilityHook.events.push(`sync:${String(flags)}:${path}`); + await realSync(); + }; + return handle; + }, rename: async (from: string, to: string): Promise => { 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); + }); +});