quak/test/cli/backup-concurrency.test.ts

/**
 * 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);
    });
});