Test: lib.backup() works on up to the content pool's limit of originals at once

Against the fake Ente server, with each download held until the test lets
it go: a library with the default pools has five downloads in flight and a
pool of two has two; with verify, only the originals whose downloads are
in flight have been hashed; a 404 for one photo does not stop the rest;
and two runs end the same with a pool of one as with the default pools.
Today the backup downloads one original at a time, so the first four fail.

Model: opus-5-5

test/cli/backup-concurrency.test.ts

@@ -0,0 +1,465 @@
+/**
+ * How many originals `lib.backup()` works on at once.
+ *
+ * The backup puts up to the content pool's limit of originals at their save
+ * paths at the same time: 5 for a library opened with the default
+ * `RequestPools`, as `quak backup` opens it, or the `contentConcurrency` of the
+ * pools a library was opened with. Everything the backup does for one original
+ * counts against that limit, hashing it for `verify` included, so however large
+ * the account is, the backup has no more than that many originals open at once.
+ *
+ * As in `backup-end-to-end.test.ts`, the real `Client` and `Library` run
+ * against `buildMetaFetch` from `test/ente-server.ts`, a fake of the Ente HTTP
+ * API serving data this file encrypts with libsodium. Here a test can also hold
+ * each original's download until it lets it go, to see how many the backup has
+ * in flight at once.
+ *
+ * The account, user 42, has one album of its own, Holiday, of eight photos,
+ * all taken at noon on 1 March 2024, each with an original of its own.
+ */
+
+import {
+    existsSync,
+    lstatSync,
+    mkdtempSync,
+    readdirSync,
+    readFileSync,
+    readlinkSync,
+    rmSync,
+    writeFileSync,
+} from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import sodium from "libsodium-wrappers-sumo";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+import type { BackupOptions } from "../../src/backup.js";
+import { Client, type ClientSnapshot } from "../../src/client.js";
+import { init, toBase64 } from "../../src/crypto/index.js";
+import { Library } from "../../src/library/index.js";
+import { RequestPools } from "../../src/library/pools.js";
+import {
+    buildAccount,
+    buildMetaFetch,
+    encryptSecretbox,
+    encryptStreamBlob,
+} from "../ente-server.js";
+import { blake2b } from "../live-photo.js";
+
+const EMAIL = "backup@example.com";
+const PASSWORD = "backup-password";
+// The fake server logs every account in as user 42.
+const USER_ID = 42;
+
+// Noon on 1 March 2024 in the time zone of the machine running the test, in
+// microseconds, as Ente records times.
+const TAKEN = new Date(2024, 2, 1, 12).getTime() * 1000;
+
+// The eight photos' IDs. Photo `id` is titled `photo-<id>.jpg`, and its
+// original is a few bytes naming it.
+const PHOTO_IDS = [101, 102, 103, 104, 105, 106, 107, 108];
+const originalOf = (id: number): Uint8Array =>
+    new TextEncoder().encode(`the photo ${id}`);
+
+// Where a backup to `dir` saves photo `id`'s original.
+const savedAt = (dir: string, id: number): string =>
+    join(dir, "2024/2024-03/2024-03-01", `2024-03-01.${id}.jpg`);
+
+// The fake server, the session it logs the account in with, and the
+// directory every backup below is made in.
+let server: typeof globalThis.fetch;
+let session: ClientSnapshot;
+let root: string;
+
+// The fake server, holding the account encrypted as Ente's clients upload it.
+const buildServer = async () => {
+    const account = await buildAccount(PASSWORD);
+
+    // The album as `/collections/v2` lists it: its key encrypted with the
+    // account's master key, its name with its own key.
+    const albumKey = sodium.crypto_secretbox_keygen();
+    const encryptedAlbumKey = encryptSecretbox(albumKey, account.masterKey);
+    const encryptedName = encryptSecretbox(
+        new TextEncoder().encode("Holiday"),
+        albumKey,
+    );
+    const album = {
+        id: 10,
+        owner: { id: USER_ID },
+        encryptedKey: toBase64(encryptedAlbumKey.ciphertext),
+        keyDecryptionNonce: toBase64(encryptedAlbumKey.nonce),
+        encryptedName: toBase64(encryptedName.ciphertext),
+        nameDecryptionNonce: toBase64(encryptedName.nonce),
+        type: "album",
+        updationTime: TAKEN,
+    };
+
+    // Each photo as the album's `/collections/v2/diff` lists it: its key
+    // encrypted with the album's key, and its metadata, holding the hash of
+    // its original, encrypted with its own key. Its original, encrypted with
+    // its own key too, is what the files CDN serves for it.
+    const ciphertexts: Record<number, Uint8Array> = {};
+    const photos = PHOTO_IDS.map((id) => {
+        const key = sodium.crypto_secretstream_xchacha20poly1305_keygen();
+        const original = encryptStreamBlob(originalOf(id), key);
+        ciphertexts[id] = original.ciphertext;
+        const encryptedKey = encryptSecretbox(key, albumKey);
+        const metadata = encryptStreamBlob(
+            new TextEncoder().encode(
+                JSON.stringify({
+                    title: `photo-${id}.jpg`,
+                    fileType: 0,
+                    hash: blake2b(originalOf(id)),
+                    creationTime: TAKEN,
+                    modificationTime: TAKEN,
+                }),
+            ),
+            key,
+        );
+        return {
+            id,
+            collectionID: 10,
+            ownerID: USER_ID,
+            encryptedKey: toBase64(encryptedKey.ciphertext),
+            keyDecryptionNonce: toBase64(encryptedKey.nonce),
+            metadata: {
+                encryptedData: toBase64(metadata.ciphertext),
+                decryptionHeader: toBase64(metadata.header),
+            },
+            file: { decryptionHeader: toBase64(original.header) },
+            thumbnail: {
+                decryptionHeader: toBase64(sodium.randombytes_buf(24)),
+            },
+            updationTime: TAKEN,
+        };
+    });
+
+    return buildMetaFetch({
+        verifier: account.verifier,
+        srpAttributes: account.srpAttributes,
+        keyAttributes: account.keyAttributes,
+        encryptedToken: account.encryptedToken,
+        collections: [album],
+        filesByCollection: { 10: photos },
+        encryptedMLData: {},
+        fileCiphertexts: ciphertexts,
+    });
+};
+
+beforeAll(async () => {
+    await init();
+    await sodium.ready;
+    root = mkdtempSync(join(tmpdir(), "quak-backup-concurrency-"));
+    server = await buildServer();
+    const client = await Client.login({
+        email: EMAIL,
+        password: PASSWORD,
+        apiOptions: { fetch: server },
+    });
+    session = client.toJSON();
+});
+
+afterAll(() => {
+    rmSync(root, { recursive: true, force: true });
+});
+
+// The ID of the photo whose original a request downloads, or undefined for any
+// other request. quak downloads an original from
+// `https://files.ente.io/?fileID=<id>`.
+const downloadOf = (input: RequestInfo | URL): number | undefined => {
+    const url = new URL(String(input));
+    return url.hostname === "files.ente.io"
+        ? Number(url.searchParams.get("fileID"))
+        : undefined;
+};
+
+// The fake server, except that it answers 404 to every download of photo
+// `failing`'s original.
+const failingFor =
+    (failing: number): typeof globalThis.fetch =>
+    async (input, init) =>
+        downloadOf(input) === failing
+            ? new Response("not found", { status: 404 })
+            : server(input, init);
+
+// The fake server, holding each download of an original until the test lets
+// it go. `held` lists the downloads waiting, the first to arrive first;
+// `most()` is the largest number that ever waited at once. A held download of
+// photo `failing`'s original is answered 404 once it is let go.
+const holdingServer = (failing?: number) => {
+    const held: { id: number; release: () => void }[] = [];
+    let most = 0;
+    const fetch: typeof globalThis.fetch = async (input, init) => {
+        const id = downloadOf(input);
+        if (id !== undefined) {
+            await new Promise<void>((release) => {
+                held.push({ id, release });
+                most = Math.max(most, held.length);
+            });
+            if (id === failing) {
+                return new Response("not found", { status: 404 });
+            }
+        }
+        return server(input, init);
+    };
+    return { fetch, held, most: () => most };
+};
+
+// Let the held downloads go one at a time, the first to arrive first, each
+// once `limit` of them wait, or all the downloads still to come when there
+// are fewer. Each of the eight photos is downloaded once. With a backup that
+// downloads fewer than `limit` at once, this fails waiting for them.
+const releaseOneAtATime = async (
+    held: { release: () => void }[],
+    limit: number,
+): Promise<void> => {
+    for (let left = PHOTO_IDS.length; left > 0; left--) {
+        await vi.waitFor(
+            () => expect(held).toHaveLength(Math.min(limit, left)),
+            { interval: 5 },
+        );
+        held.shift()!.release();
+    }
+};
+
+// A library of the account opened as `quak backup` opens one, with no
+// precache and no background refresh, sending its requests through `fetch`.
+// It backs up to `dir` and keeps its cache beside it. Its request pools are
+// `pools`, or by default the library's own.
+const openLibrary = (
+    dir: string,
+    fetch: typeof globalThis.fetch,
+    pools?: RequestPools,
+): Promise<Library> =>
+    Library.open({
+        client: Client.fromJSON(session, { fetch }),
+        downloadDirectory: dir,
+        cacheDirectory: `${dir}-cache`,
+        refreshIntervalSeconds: Infinity,
+        precacheThumbnails: false,
+        precacheOriginals: false,
+        pools,
+    });
+
+// Every entry under `dir`, relative to it, sorted, a symlink as
+// `name -> target`.
+const listTree = (dir: string): string[] =>
+    readdirSync(dir, { recursive: true })
+        .sort()
+        .map((name) => {
+            const path = join(dir, name);
+            return lstatSync(path).isSymbolicLink()
+                ? `${name} -> ${readlinkSync(path)}`
+                : name;
+        });
+
+const expectStored = (dir: string, ids: number[]): void => {
+    for (const id of ids) {
+        expect(readFileSync(savedAt(dir, id))).toEqual(
+            Buffer.from(originalOf(id)),
+        );
+    }
+};
+
+describe("lib.backup()", () => {
+    // A library opened with the default pools, as `quak backup` opens one,
+    // downloads up to five originals at once. As soon as one finishes, the
+    // backup starts on the next photo, so five stay in flight until fewer
+    // than five are left.
+    it("downloads five originals at once, and never more, with the default pools", async () => {
+        const dir = mkdtempSync(join(root, "backup-"));
+        const holding = holdingServer();
+        const lib = await openLibrary(dir, holding.fetch);
+        try {
+            const backup = lib.backup();
+            await releaseOneAtATime(holding.held, 5);
+            const result = await backup;
+
+            expect(holding.most()).toBe(5);
+            expect(result).toMatchObject({
+                totalFiles: 8,
+                downloaded: 8,
+                failed: 0,
+            });
+            expectStored(dir, PHOTO_IDS);
+        } finally {
+            await lib.close();
+        }
+    });
+
+    // The limit is the content pool's, so a library opened with pools of
+    // another `contentConcurrency` backs up that many at once.
+    it("downloads as many originals at once as the content pool of the pools it was opened with allows", async () => {
+        const dir = mkdtempSync(join(root, "backup-"));
+        const holding = holdingServer();
+        const lib = await openLibrary(
+            dir,
+            holding.fetch,
+            new RequestPools({ contentConcurrency: 2 }),
+        );
+        try {
+            const backup = lib.backup();
+            await releaseOneAtATime(holding.held, 2);
+            const result = await backup;
+
+            expect(holding.most()).toBe(2);
+            expect(result).toMatchObject({ downloaded: 8, failed: 0 });
+            expectStored(dir, PHOTO_IDS);
+        } finally {
+            await lib.close();
+        }
+    });
+
+    // With `verify`, hashing an original already stored is part of the work
+    // on it, and counts against the limit too: a large backup never has more
+    // than five originals open for hashing. Here every stored original has
+    // changed on disk, so each is hashed, found not to match, and downloaded
+    // again. While five downloads wait, the backup has hashed only those five
+    // photos' originals; it opens the next one once a download finishes.
+    it("with verify, hashes no more originals ahead of their downloads than the limit", async () => {
+        const dir = mkdtempSync(join(root, "backup-"));
+        const first = await openLibrary(dir, server);
+        try {
+            await first.backup();
+        } finally {
+            await first.close();
+        }
+        for (const id of PHOTO_IDS) {
+            writeFileSync(savedAt(dir, id), "not the photo any more");
+        }
+
+        const holding = holdingServer();
+        const lib = await openLibrary(dir, holding.fetch);
+        const lines: string[] = [];
+        const mismatches = () =>
+            lines.filter((line) => line.startsWith("MISMATCH"));
+        try {
+            const backup = lib.backup({
+                verify: true,
+                onProgress: (line) => lines.push(line),
+            });
+            await vi.waitFor(() => expect(holding.held).toHaveLength(5), {
+                interval: 5,
+            });
+            expect(mismatches()).toHaveLength(5);
+
+            await releaseOneAtATime(holding.held, 5);
+            const result = await backup;
+
+            expect(holding.most()).toBe(5);
+            expect(mismatches()).toHaveLength(8);
+            expect(result).toMatchObject({
+                downloaded: 8,
+                verified: 0,
+                mismatched: 8,
+                failed: 0,
+            });
+            expectStored(dir, PHOTO_IDS);
+        } finally {
+            await lib.close();
+        }
+    });
+
+    // The server answers 404 for photo 103's original. A 404 is not retried,
+    // so that download fails at once, while other downloads are still in
+    // flight. The backup records the failure and goes on: the downloads
+    // beside it finish, the photos after it are downloaded, and only photo
+    // 103 is missing from the backup and in failures.json.
+    it("goes on with the other originals when one fails", async () => {
+        const dir = mkdtempSync(join(root, "backup-"));
+        const holding = holdingServer(103);
+        const lib = await openLibrary(dir, holding.fetch);
+        try {
+            const backup = lib.backup();
+            await releaseOneAtATime(holding.held, 5);
+            const result = await backup;
+
+            expect(result).toMatchObject({
+                totalFiles: 8,
+                downloaded: 7,
+                failed: 1,
+            });
+            expect(result.errors.map((e) => e.fileID)).toEqual([103]);
+            expectStored(
+                dir,
+                PHOTO_IDS.filter((id) => id !== 103),
+            );
+            expect(existsSync(savedAt(dir, 103))).toBe(false);
+            const failures = JSON.parse(
+                readFileSync(join(dir, "failures.json"), "utf-8"),
+            );
+            expect(Object.keys(failures.files)).toEqual(["103"]);
+        } finally {
+            await lib.close();
+        }
+    });
+
+    // Working on several originals at once changes only how long a backup
+    // takes. The same two runs, with a content pool of one, which backs up
+    // one original at a time, and with the default pools, end the same way:
+    // the same result, the same progress lines (in another order), the same
+    // failures.json, and the same tree. The first run stores every photo but
+    // 103, whose download is answered 404. Then two stored originals change
+    // on disk, and a run with `verify` puts them back and tries 103 again.
+    it("ends as a backup of one original at a time does", async () => {
+        const twoRuns = async (pools: RequestPools) => {
+            const dir = mkdtempSync(join(root, "backup-"));
+            const lines: string[] = [];
+            const backup = async (opts: BackupOptions) => {
+                const lib = await openLibrary(dir, failingFor(103), pools);
+                try {
+                    return await lib.backup({
+                        ...opts,
+                        onProgress: (line) => lines.push(line),
+                    });
+                } finally {
+                    await lib.close();
+                }
+            };
+            const first = await backup({});
+            writeFileSync(savedAt(dir, 101), "not the photo any more");
+            writeFileSync(savedAt(dir, 105), "not the photo any more");
+            const second = await backup({ verify: true });
+
+            // Each entry of failures.json without the time it was last
+            // tried, which differs from run to run.
+            const { files } = JSON.parse(
+                readFileSync(join(dir, "failures.json"), "utf-8"),
+            ) as { files: Record<string, { lastTriedAt?: number }> };
+            for (const entry of Object.values(files)) delete entry.lastTriedAt;
+
+            return {
+                first,
+                second,
+                lines: lines.sort(),
+                failures: files,
+                tree: listTree(dir),
+            };
+        };
+
+        const oneAtATime = await twoRuns(
+            new RequestPools({ contentConcurrency: 1 }),
+        );
+        const fiveAtATime = await twoRuns(new RequestPools());
+
+        expect(oneAtATime.first).toMatchObject({
+            totalFiles: 8,
+            downloaded: 7,
+            skipped: 0,
+            failed: 1,
+        });
+        expect(oneAtATime.second).toMatchObject({
+            totalFiles: 8,
+            downloaded: 2,
+            skipped: 5,
+            verified: 5,
+            mismatched: 2,
+            unchecked: 0,
+            failed: 1,
+        });
+        expect(oneAtATime.failures).toEqual({
+            "103": expect.objectContaining({ fileID: 103, attempts: 2 }),
+        });
+        expect(fiveAtATime).toEqual(oneAtATime);
+    });
+});