Merge branch 'fix/backup-chunked-storage' into release/v1.39.2

tests/integration/backup-chunked-storage.test.ts

@@ -482,8 +482,14 @@ describe("a restore queued against a copy that is then replaced", () => {
     expect(queued.status).toBe(202);
     const { data } = (await queued.json()) as { data: { jobId: string } };
 
-    // The next weekly run writes a new copy into the same row.
-    await weeklyCopy(user.id);
+    // The weekly run leaves a copy alone while a restore of it is queued, so
+    // stand in for a replacement written anyway (an older worker, say): the
+    // copy's stream id changes.
+    await expect(weeklyCopy(user.id)).rejects.toThrow(/restore of the current/);
+    await getPrismaClient().dataBackup.update({
+      where: { id },
+      data: { chunkStreamId: "f".repeat(32) },
+    });
     const before = await measurementsOf(user.id);
     await runBackupRestoreJob(data.jobId).catch(() => undefined);
     const job = await readBackupRestoreJob(data.jobId);
@@ -492,6 +498,126 @@ describe("a restore queued against a copy that is then replaced", () => {
   });
 });
 
+describe("a copy replaced while it is being read", () => {
+  it("says it was replaced, not that it was tampered with", async () => {
+    const user = await seedAccount();
+    const prisma = getPrismaClient();
+    const { id } = await weeklyCopy(user.id);
+    const backup = await prisma.dataBackup.findUniqueOrThrow({
+      where: { id },
+      select: STORED_BACKUP_SELECT,
+    });
+    const { openStoredBackup, BackupReplacedError } =
+      await import("@/lib/export/stored-backup");
+    const source = await openStoredBackup(prisma, backup);
+
+    // The next weekly run writes a new copy into the same row.
+    await weeklyCopy(user.id);
+
+    const read = async () => {
+      for await (const chunk of source()) void chunk;
+    };
+    const err = await read().catch((e: unknown) => e);
+    expect(err).toBeInstanceOf(BackupReplacedError);
+    expect((err as Error).message).toMatch(
+      /replaced by a newer one while it was being read/,
+    );
+  });
+
+  it("is not replaced by the weekly run while a restore of it is queued or running", async () => {
+    const user = await seedAccount();
+    const prisma = getPrismaClient();
+    const first = await weeklyCopy(user.id);
+    const before = await prisma.dataBackup.findUniqueOrThrow({
+      where: { id: first.id },
+    });
+    for (const status of ["queued", "running"]) {
+      const job = await prisma.backupRestoreJob.create({
+        data: {
+          userId: user.id,
+          actorUserId: user.id,
+          backupId: first.id,
+          backupDigest: "x",
+          status,
+        },
+      });
+      const { BackupBusyError } = await import("@/lib/export/backup-blob");
+      await expect(weeklyCopy(user.id)).rejects.toBeInstanceOf(BackupBusyError);
+      const after = await prisma.dataBackup.findUniqueOrThrow({
+        where: { id: first.id },
+      });
+      expect(after.chunkStreamId).toBe(before.chunkStreamId);
+      expect(await chunksOf(first.id)).toHaveLength(first.chunks);
+      await prisma.backupRestoreJob.update({
+        where: { id: job.id },
+        data: { status: "succeeded" },
+      });
+    }
+    // Once no restore reads it, the next run replaces it as usual.
+    const next = await weeklyCopy(user.id);
+    expect(next.id).toBe(first.id);
+  });
+});
+
+describe("a row that carries both forms", () => {
+  it("is refused with a reason, whichever form might be newer", async () => {
+    // How it happens: an older release that knows only the single value
+    // writes one into a row that already has pieces, after a downgrade.
+    const user = await seedAccount();
+    const prisma = getPrismaClient();
+    const { id } = await weeklyCopy(user.id);
+    await prisma.dataBackup.update({
+      where: { id },
+      data: {
+        data: await legacyStreamedBlobFrom(async (write) => {
+          await write("{}");
+        }),
+      },
+    });
+    const backup = await prisma.dataBackup.findUniqueOrThrow({
+      where: { id },
+      select: STORED_BACKUP_SELECT,
+    });
+    const { openStoredBackup } = await import("@/lib/export/stored-backup");
+    await expect(openStoredBackup(prisma, backup)).rejects.toThrow(
+      /both a single stored value and pieces/,
+    );
+    const res = await restoreRoute(id);
+    expect(res.status).toBe(422);
+  });
+
+  it("is written in one form again by the next weekly run", async () => {
+    const user = await seedAccount();
+    const prisma = getPrismaClient();
+    const { id } = await weeklyCopy(user.id);
+    await prisma.dataBackup.update({ where: { id }, data: { data: "x" } });
+    await weeklyCopy(user.id);
+    const row = await prisma.dataBackup.findUniqueOrThrow({ where: { id } });
+    expect(row.data).toBeNull();
+    expect(await readStoredBackup(prisma, id)).toContain('"measurements"');
+  });
+});
+
+describe("two backups of the same account at the same time", () => {
+  it("run one after the other and leave one complete copy", async () => {
+    const user = await seedAccount();
+    const prisma = getPrismaClient();
+    await weeklyCopy(user.id);
+    const results = await Promise.allSettled([
+      weeklyCopy(user.id),
+      weeklyCopy(user.id),
+    ]);
+    expect(results.map((r) => r.status)).toEqual(["fulfilled", "fulfilled"]);
+    const row = await prisma.dataBackup.findFirstOrThrow({
+      where: { userId: user.id, type: "WEEKLY_AUTO" },
+    });
+    expect(await chunksOf(row.id)).toHaveLength(row.chunkCount!);
+    expect(
+      JSON.parse(await readStoredBackup(prisma, row.id)).measurements,
+    ).toHaveLength(COUNT);
+  });
+});
+
 describe("writing and removing copies", () => {
   it("keeps the previous copy whole when a run fails after writing pieces", async () => {
     const user = await seedAccount();

tests/integration/encryption-corpus-cursor-row-deleted.test.ts

@@ -14,7 +14,7 @@
  */
 import { beforeEach, describe, expect, it } from "vitest";
 
-import { encrypt } from "@/lib/crypto";
+import { newChunkStreamId, sealBackupChunk } from "@/lib/export/backup-chunks";
 import { ENCRYPTED_COLUMNS } from "@/lib/crypto/encrypted-columns";
 import {
   BLOB_ROTATION_BATCH_SIZE,
@@ -34,35 +34,50 @@ describe("encryption corpus walk", () => {
       data: { username: "cursor-walk", email: "cursor-walk@example.test" },
     });
     const count = BLOB_ROTATION_BATCH_SIZE + 5;
-    await prisma.dataBackup.createMany({
-      data: Array.from({ length: count }, (_, i) => ({
-        id: `backup-${String(i).padStart(3, "0")}`,
+    // The pieces of one stored backup: a blob column walked in small pages.
+    const backup = await prisma.dataBackup.create({
+      data: {
         userId: user.id,
-        type: `MANUAL_UPLOAD_${i}`,
-        data: encrypt(`{"n":${i}}`),
+        type: "WEEKLY_AUTO",
+        data: null,
+        chunkCount: count,
+        chunkStreamId: newChunkStreamId(),
+      },
+    });
+    const streamId = newChunkStreamId();
+    await prisma.dataBackupChunk.createMany({
+      data: Array.from({ length: count }, (_, i) => ({
+        id: `piece-${String(i).padStart(3, "0")}`,
+        backupId: backup.id,
+        seq: i,
+        data: new Uint8Array(
+          sealBackupChunk(streamId, i, i === count - 1, Buffer.from(`${i}`)),
+        ),
       })),
     });
 
     let pages = 0;
     const client = {
-      dataBackup: {
+      dataBackupChunk: {
         findMany: async (args: never) => {
-          const rows = (await prisma.dataBackup.findMany(args)) as Array<{
+          const rows = (await prisma.dataBackupChunk.findMany(args)) as Array<{
             id: string;
           }>;
           pages += 1;
           // The first page's last row disappears before the second is read.
           if (pages === 1 && rows.length > 0) {
-            await prisma.dataBackup.delete({ where: { id: rows.at(-1)!.id } });
+            await prisma.dataBackupChunk.delete({
+              where: { id: rows.at(-1)!.id },
+            });
           }
           return rows;
         },
-        update: (args: never) => prisma.dataBackup.update(args),
+        update: (args: never) => prisma.dataBackupChunk.update(args),
       },
     } as unknown as CorpusClient;
 
     const col = ENCRYPTED_COLUMNS.find(
-      (c) => c.model === "DataBackup" && c.field === "data",
+      (c) => c.model === "DataBackupChunk" && c.field === "data",
     )!;
     const scan = await scanColumn(client, col);
 

tests/integration/encryption-key-rotation-blobs.test.ts

@@ -9,11 +9,11 @@
  * pins the fix at the level the bug lived at: seed real rows under an old key,
  * rotate, and read the backup back.
  *
- * Both stored envelopes are seeded on purpose. A row written before
- * `packBackupBlob` existed is `encrypt(json)`; one written after is the
- * `HLZ1:`-prefixed gzip form. Rotation re-seals the ciphertext without ever
- * looking at the plaintext, so both come back byte-identical — and so will
- * whatever envelope a later writer introduces.
+ * Every single-value envelope is seeded on purpose: `encrypt(json)`, the
+ * `HLZ1:` gzip form and the `~hlgcm1.` stream of v1.38.6 to v1.39.1. Rotation does not
+ * re-seal them in place, which would hold a whole copy several times over; it
+ * converts each into pieces under the active key, reading the value a slice
+ * at a time, and the copy has to read back byte-identical.
  *
  * `IdempotencyKey.responseBody` rides along as the disposable case: a row
  * under a key the deployment no longer configures is DELETED rather than
@@ -37,11 +37,15 @@ import {
 } from "@/lib/crypto/encrypted-columns";
 import {
   rotateColumn,
+  scanColumn,
   type CorpusClient,
 } from "@/lib/crypto/encryption-corpus";
-import { packBackupBlob, unpackBackupBlob } from "@/lib/export/backup-blob";
+import { packBackupBlob } from "@/lib/export/backup-blob";
 import { extractKeyIdFromBytes } from "@/lib/crypto";
-import { storeBackupBlob } from "@/lib/export/store-backup-blob";
+import {
+  convertSingleValueBackup,
+  storeBackupBlob,
+} from "@/lib/export/store-backup-blob";
 import { legacyStreamedBlob } from "@/__tests__/helpers/legacy-backup-blob";
 import { getPrismaClient, truncateAllTables } from "./setup";
 import { readStoredBackup } from "./stored-backup-read";
@@ -91,101 +95,157 @@ beforeEach(async () => {
 });
 
 describe("key rotation over the non-suffixed ciphertext columns", () => {
-  it("rotates DataBackup.data in both stored envelopes and keeps it readable", async () => {
+  /**
+   * Rotation converts a single-value copy into pieces rather than re-sealing
+   * it in place. Re-sealing needs the whole value in memory several times
+   * over, and the in-app rotation runs in the same 1 GB container as
+   * everything else; converting reads the value a slice at a time.
+   */
+  async function expectConverted(id: string, createdAt: Date) {
     const prisma = getPrismaClient();
-    const gzipped = await prisma.dataBackup.create({
-      data: {
-        userId: TEST_USER_ID,
-        type: "WEEKLY_AUTO",
-        data: underKey("v1", BACKUP_JSON, true),
-      },
-    });
-    // The pre-envelope shape: `encrypt(json)` with no `HLZ1:` marker.
-    const plainEnvelope = await prisma.dataBackup.create({
-      data: {
-        userId: TEST_USER_ID,
-        type: "MANUAL_UPLOAD_1",
-        data: underKey("v1", BACKUP_JSON),
-      },
-    });
-
-    const result = await rotateColumn(
-      { dataBackup: prisma.dataBackup } as unknown as CorpusClient,
-      column("DataBackup", "data"),
-    );
-    expect(result.scanned).toBe(2);
-    expect(result.rotated).toBe(2);
-    expect(result.errors).toBe(0);
-    expect(result.dropped).toBe(0);
-
-    for (const id of [gzipped.id, plainEnvelope.id]) {
-      const row = await prisma.dataBackup.findUniqueOrThrow({ where: { id } });
-      expect(extractKeyId(row.data!)).toBe("v2");
-      expect(unpackBackupBlob(row.data!)).toBe(BACKUP_JSON);
+    const row = await prisma.dataBackup.findUniqueOrThrow({ where: { id } });
+    expect(row.data).toBeNull();
+    expect(row.chunkCount).toBeGreaterThan(0);
+    expect(row.chunkStreamId).toMatch(/^[0-9a-f]{32}$/);
+    // The copy is still the one taken on that date.
+    expect(row.createdAt).toEqual(createdAt);
+    for (const piece of await prisma.dataBackupChunk.findMany({
+      where: { backupId: id },
+    })) {
+      expect(extractKeyIdFromBytes(Buffer.from(piece.data))).toBe("v2");
     }
+    expect(await readStoredBackup(prisma, id)).toBe(BACKUP_JSON);
+  }
 
-    // Idempotent: a second pass finds nothing left to do.
-    const again = await rotateColumn(
-      { dataBackup: prisma.dataBackup } as unknown as CorpusClient,
-      column("DataBackup", "data"),
-    );
-    expect(again.scanned).toBe(2);
-    expect(again.rotated).toBe(0);
-  });
-
-  it("survives dropping the retired key once rotation has run", async () => {
-    const prisma = getPrismaClient();
-    const backup = await prisma.dataBackup.create({
-      data: {
-        userId: TEST_USER_ID,
-        type: "WEEKLY_AUTO",
-        data: underKey("v1", BACKUP_JSON, true),
-      },
-    });
-    await rotateColumn(
-      { dataBackup: prisma.dataBackup } as unknown as CorpusClient,
-      column("DataBackup", "data"),
-    );
-
-    // What the runbook tells the operator to do next: retire v1 entirely.
-    process.env.ENCRYPTION_KEYS = JSON.stringify({ v2: "2".repeat(64) });
-    _resetCryptoCacheForTests();
-
-    const row = await prisma.dataBackup.findUniqueOrThrow({
-      where: { id: backup.id },
-    });
-    expect(unpackBackupBlob(row.data!)).toBe(BACKUP_JSON);
-  });
-
-  it("rotates the single stream v1.39.1 stored, and it stays readable", async () => {
-    // The `~hlgcm1.` form carries its key id behind a marker the string
-    // codec's parser does not read, so the walk used to file it as legacy
-    // and fail to decrypt it: the one copy form every v1.39.x host had was
-    // counted as an error on every rotation.
+  it("converts DataBackup.data in every single-value envelope into pieces under the active key", async () => {
     const prisma = getPrismaClient();
     const previous = process.env.ENCRYPTION_ACTIVE_KEY_ID;
     process.env.ENCRYPTION_ACTIVE_KEY_ID = "v1";
     _resetCryptoCacheForTests();
-    const stored = legacyStreamedBlob(BACKUP_JSON);
+    // The single stream of v1.38.6 to v1.39.1, whose key id sits behind a `~hlgcm1.`
+    // marker the string codec's parser does not read.
+    const streamed = legacyStreamedBlob(BACKUP_JSON);
     process.env.ENCRYPTION_ACTIVE_KEY_ID = previous;
     _resetCryptoCacheForTests();
+    const rows = [
+      await prisma.dataBackup.create({
+        data: {
+          userId: TEST_USER_ID,
+          type: "WEEKLY_AUTO",
+          data: underKey("v1", BACKUP_JSON, true),
+        },
+      }),
+      // The pre-envelope shape: `encrypt(json)` with no `HLZ1:` marker.
+      await prisma.dataBackup.create({
+        data: {
+          userId: TEST_USER_ID,
+          type: "MANUAL_UPLOAD_1",
+          data: underKey("v1", BACKUP_JSON),
+        },
+      }),
+      await prisma.dataBackup.create({
+        data: { userId: TEST_USER_ID, type: "MANUAL_UPLOAD_2", data: streamed },
+      }),
+      // Already under the active key: converted too, so no single value is
+      // left for a reader that has to hold it whole.
+      await prisma.dataBackup.create({
+        data: {
+          userId: TEST_USER_ID,
+          type: "MANUAL_UPLOAD_3",
+          data: underKey("v2", BACKUP_JSON, true),
+        },
+      }),
+    ];
+
+    const result = await rotateColumn(
+      prisma as unknown as CorpusClient,
+      column("DataBackup", "data"),
+    );
+    expect(result).toMatchObject({
+      scanned: 4,
+      rotated: 4,
+      errors: 0,
+      dropped: 0,
+    });
+
+    // What the runbook tells the operator to do next: retire v1 entirely.
+    process.env.ENCRYPTION_KEYS = JSON.stringify({ v2: "2".repeat(64) });
+    _resetCryptoCacheForTests();
+    for (const row of rows) await expectConverted(row.id, row.createdAt);
+
+    // Idempotent: a second pass finds nothing left to do.
+    const again = await rotateColumn(
+      prisma as unknown as CorpusClient,
+      column("DataBackup", "data"),
+    );
+    expect(again).toMatchObject({ scanned: 0, rotated: 0, errors: 0 });
+  });
+
+  it("converts a single stream read in many small slices, the tag straddling one", async () => {
+    const prisma = getPrismaClient();
+    const json = JSON.stringify({
+      rows: Array.from({ length: 3_000 }, (_, i) => ({ i, r: Math.random() })),
+    });
+    const stored = legacyStreamedBlob(json);
+    const header = stored.indexOf(".", "~hlgcm1.".length) + 17;
+    const body = stored.length - header;
+    for (const slice of [64, 1_024, body - (body % 4) - 8]) {
+      const row = await prisma.dataBackup.create({
+        data: { userId: TEST_USER_ID, type: `MANUAL_${slice}`, data: stored },
+      });
+      expect(await convertSingleValueBackup(prisma, row.id, slice)).toBe(
+        "converted",
+      );
+      expect(await readStoredBackup(prisma, row.id)).toBe(json);
+    }
+  });
+
+  it("leaves a single-value copy that fails its check exactly as it was", async () => {
+    const prisma = getPrismaClient();
+    const previous = process.env.ENCRYPTION_ACTIVE_KEY_ID;
+    process.env.ENCRYPTION_ACTIVE_KEY_ID = "v1";
+    _resetCryptoCacheForTests();
+    const streamed = legacyStreamedBlob(BACKUP_JSON);
+    process.env.ENCRYPTION_ACTIVE_KEY_ID = previous;
+    _resetCryptoCacheForTests();
+    // Alter one character of the ciphertext body: the tag at the end no
+    // longer verifies, which the conversion only learns after the last slice.
+    const at = streamed.length - 40;
+    const altered =
+      streamed.slice(0, at) +
+      (streamed[at] === "A" ? "B" : "A") +
+      streamed.slice(at + 1);
     const row = await prisma.dataBackup.create({
-      data: { userId: TEST_USER_ID, type: "WEEKLY_AUTO", data: stored },
+      data: { userId: TEST_USER_ID, type: "WEEKLY_AUTO", data: altered },
     });
 
     const result = await rotateColumn(
-      { dataBackup: prisma.dataBackup } as unknown as CorpusClient,
+      prisma as unknown as CorpusClient,
       column("DataBackup", "data"),
     );
-    expect(result).toMatchObject({ scanned: 1, rotated: 1, errors: 0 });
-
-    process.env.ENCRYPTION_KEYS = JSON.stringify({ v2: "2".repeat(64) });
-    _resetCryptoCacheForTests();
+    expect(result).toMatchObject({ scanned: 1, rotated: 0, errors: 1 });
     const after = await prisma.dataBackup.findUniqueOrThrow({
       where: { id: row.id },
     });
-    expect(after.data!.startsWith("~hlgcm1.v2.")).toBe(true);
-    expect(unpackBackupBlob(after.data!)).toBe(BACKUP_JSON);
+    expect(after.data).toBe(altered);
+    expect(after.chunkStreamId).toBeNull();
+    expect(await prisma.dataBackupChunk.count()).toBe(0);
+  });
+
+  it("reports single-value copies by key id without reading them whole", async () => {
+    const prisma = getPrismaClient();
+    await prisma.dataBackup.create({
+      data: {
+        userId: TEST_USER_ID,
+        type: "WEEKLY_AUTO",
+        data: underKey("v1", BACKUP_JSON, true),
+      },
+    });
+    const scan = await scanColumn(
+      prisma as unknown as CorpusClient,
+      column("DataBackup", "data"),
+    );
+    expect(scan).toMatchObject({ total: 1, byKeyId: { v1: 1 } });
   });
 
   it("rotates every piece of a chunked copy, which reads back once the old key is gone", async () => {