From 13a9aacf206610b3068323eb712445c87321c9e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 5 Sep 2026 17:05:43 +0200 Subject: [PATCH 1/2] fix(backup): stream the nightly off-host upload instead of building it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The off-host job was said to benefit from the streaming work the weekly in-database backup got, and it did not. Only the payload builder was shared: `runOffhostBackup` called `buildFullBackupJson` for the whole document, `gzipSync` over that whole string, a whole-buffer `createCipheriv` pass over the result, and handed the finished buffer to a single `PutObject`. Four full copies of the record alive at once, on the same arithmetic the weekly pass had already been fixed for. Configuring an S3 target on a long-lived record therefore restarted the container. On an account of 445 000 measurements the JSON alone is 242 MB, and against a 1 GB container's 524 MB heap the run died of heap exhaustion seventeen seconds in — taking every signed-in session with it, because the job shares the app process. The write path is a stream end to end now: `streamFullBackupJson` produces the document a page at a time, gzip and an incremental AES-256-GCM writer consume it as it arrives, and `@aws-sdk/lib-storage` puts it up in two 8 MB parts. What the process holds is fixed by the pipeline's shape rather than by the record. Measured on that same 445 000-measurement account under `--max-old-space-size=450`, with the heap pre-loaded to what a warm server holds: the old path dies with `Ineffective mark-compacts near heap limit`, the new one finishes with a peak 79 MB above its baseline and writes a 9.1 MB object. The stored envelope gains version 3, which moves the GCM tag from in front of the ciphertext to the end. That is the only change the stream required — the tag exists only once the last block is in, so a leading one means the whole object must exist before its first byte can be sent. Nothing about the authentication moves: the tag still covers every ciphertext byte and is verified before a byte of plaintext comes back. Versions 1 and 2 still read, so every object already in a bucket restores through the same reader with no flag to tell them apart. The incremental writer is the crypto module's, split into its byte-level half so the base64 form the in-database blob uses and the raw form an object needs are one implementation under two framings. The off-host key handling is unchanged: a separate `BACKUP_ENCRYPTION_KEY`, 64 hex or 32-byte base64, passed in rather than looked up. A run that uploaded nothing for anybody is a failed job now, not `ok: true` with an empty bucket behind it. Wrong credentials, a missing bucket and an unreachable endpoint all fail every account rather than one, and the target's own sentence rides out as the cause. A run where some account got a copy still succeeds; retrying the whole cohort over one object would re-upload everybody's. The one ceiling left is structural rather than a memory bound, and it is counted rather than discovered: a multipart upload carries 10 000 parts, so an object past 80 GB is refused for that account with a clear message instead of failing halfway through with an SDK error about part numbers. `tests/integration/offhost-backup-streaming-memory.test.ts` pins the budget from both sides — what the streaming uploader holds, and that the materialising path does not fit the same budget on the same fixture in the same process, so the number cannot pass by measuring nothing. --- CHANGELOG.md | 36 ++ docs/ops/backup-restore.md | 80 +++- package.json | 1 + pnpm-lock.yaml | 128 ++++-- src/lib/crypto.ts | 99 ++++- src/lib/jobs/__tests__/offhost-backup.test.ts | 190 ++++++++- src/lib/jobs/__tests__/restore-drill.test.ts | 9 + src/lib/jobs/job-outcome.ts | 1 + src/lib/jobs/offhost-backup.ts | 387 ++++++++++++++++-- .../__tests__/offhost-backup-handler.test.ts | 141 +++++++ src/lib/jobs/reminder/backup-handlers.ts | 39 +- .../admin-backups-canonical-roundtrip.test.ts | 6 + .../offhost-backup-streaming-memory.test.ts | 335 +++++++++++++++ 13 files changed, 1340 insertions(+), 112 deletions(-) create mode 100644 src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts create mode 100644 tests/integration/offhost-backup-streaming-memory.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 02eb12e26..111e061b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,41 @@ # Changelog +## [Unreleased] + +### Fixed + +- **The nightly off-host backup no longer takes the instance down with it.** + Configuring an S3 target and waiting for 02:30 was, on a long-lived record, + a way to restart the container: the uploader built the whole backup as one + JSON string, gzipped that whole string, encrypted the whole result and + handed the finished buffer to a single upload, so four full copies of the + record were alive at once. On an account of 445 000 measurements the JSON + alone is 242 MB, and the run died of heap exhaustion seventeen seconds in — + taking every signed-in session on the instance with it, because the job runs + inside the app process. It streams now, end to end: the JSON is produced a + page at a time, gzip and the cipher consume it as it arrives, and the object + goes up in 8 MB parts. What the process holds no longer depends on how much + you have recorded. + +- **A nightly run that uploaded nothing for anybody now reports failure.** It + used to say it was fine. Wrong credentials, a bucket that does not exist and + an unreachable endpoint all fail every account rather than one, and all + three used to leave the jobs page reading healthy over an empty bucket. The + run now fails with the storage provider's own sentence as the cause. A run + where some account got a copy still succeeds, with the rest counted, because + retrying the whole cohort over one object would re-upload everybody's. + +### Changed + +- **Off-host objects carry their authentication tag at the end.** The tag can + only be produced once the last block of ciphertext is in, so a leading one + meant the whole object had to exist before its first byte could be sent — + which is the thing that could not be streamed. Nothing about the encryption + changes; the tag still covers every byte and is still verified before any + plaintext comes back. Objects already in your bucket, in either older + layout, restore exactly as before and the restore script needs no flag to + tell them apart. + ## [1.38.9] — 2026-09-05 The date order you choose now reaches every date on screen, a plain-HTTP diff --git a/docs/ops/backup-restore.md b/docs/ops/backup-restore.md index 7465207fb..92991852b 100644 --- a/docs/ops/backup-restore.md +++ b/docs/ops/backup-restore.md @@ -16,13 +16,21 @@ container (queue `data-backup-offhost`). Object key layout: ``` magic = "HLBK" (4 bytes, ASCII) -version = 0x01 (1 byte) +version = 0x03 (1 byte) iv = 12 random bytes (AES-GCM nonce) -authTag = 16 bytes (AES-GCM tag) -ciphertext = N bytes (AES-256-GCM, key = BACKUP_ENCRYPTION_KEY) -plaintext = JSON dump (UTF-8) +ciphertext = N bytes (AES-256-GCM over gzip(JSON dump), key = BACKUP_ENCRYPTION_KEY) +authTag = 16 bytes (AES-GCM tag, trailing) ``` +Three versions exist in the wild and all three restore. `0x01` encrypted the +JSON directly and `0x02` gzipped it first; both carry the tag in FRONT of the +ciphertext, which is what made them impossible to write a piece at a time — +GCM only produces the tag once the last block is in, so a leading tag means the +whole object has to exist before its first byte can be sent. `0x03` moves the tag to the end and changes nothing else: it still +covers every ciphertext byte, and the reader still verifies it before returning +a single byte of plaintext. Objects already in your bucket stay readable, and +the restore script needs no flag to tell them apart. + ## Required env vars | Var | Required | Notes | @@ -35,6 +43,15 @@ plaintext = JSON dump (UTF-8) | `BACKUP_S3_REGION` | no | defaults to `auto` (Cloudflare R2) | | `BACKUP_RETENTION_DAYS` | no | defaults to `30` | +## Bucket permissions + +The worker needs `PutObject`, `GetObject` and `AbortMultipartUpload`. It never +calls `DeleteObject` on a backup key, so a compromised worker cannot wipe the +history; `AbortMultipartUpload` only reaches an upload that same worker started +and is what clears the parts of a run that failed halfway. Without it, a failed +upload leaves parts that are billed and do not show in a bucket listing. On +Cloudflare R2 the **Object Read & Write** token already covers all three. + ## Bucket lifecycle (recommended) The worker prunes objects older than `BACKUP_RETENTION_DAYS`, but the @@ -95,6 +112,15 @@ The script downloads the object, decrypts it, and writes the JSON dump to disk. Importing the JSON back into a HealthLog instance is left to the operator (use `prisma db seed` or a custom script). +Restoring is not streamed and does not need to be. It holds the whole document, +because the next thing anyone does with a backup is parse it as one JSON +object, and handing back plaintext the auth tag has not yet covered would trade +the authentication for memory. It runs on your machine rather than in the +container, so give it room: a record of several hundred thousand measurements +decompresses to a few hundred megabytes, and +`NODE_OPTIONS=--max-old-space-size=2048` in front of the command is enough for +a 445 000-measurement account. + ### What a backup deliberately does not carry Every credential-shaped row is left out, and this is not an oversight to fix: @@ -122,6 +148,52 @@ The full per-model reasoning lives in `src/lib/export/backup-plan.ts`, where every excluded model carries a written verdict and a structural test refuses to let a new model land without one. +## Container memory (the nightly off-host job) + +This is the part that bites, and it bit the nightly job a release after it bit +the weekly one. The job runs inside the app process, so V8's heap limit is the +app's heap limit, and a container capped at 1 GB gives Node a 524 MB old-space +limit by default. A long-lived Next.js server is already holding a large share +of that before the job starts. + +The uploader used to build the whole backup JSON as one string, gzipped +that whole string, ran a whole-buffer cipher pass over the result and handed +the finished buffer to a single `PutObject` — four full copies of the record +alive at once. On an account of 445 000 measurements the JSON alone is 242 MB, +and the first configured run took the container down seventeen seconds in with +`FATAL ERROR: Reached heap limit`. Because the job shares the app process, one +account's size restarted the instance for everybody on it. + +It streams now. The JSON is produced a page at a time, gzip and the cipher +consume it as it arrives, and the object goes up as a multipart upload that +holds two 8 MB parts. What the process holds is fixed by that pipeline's shape +rather than by the size of the record going through it: measured on the same +445 000-measurement account under `--max-old-space-size=450`, the old path +died and the new one finished holding tens of megabytes, writing a 9.1 MB +object that restores to the identical record. + +One ceiling remains, and it is structural rather than a memory bound: a +multipart upload carries 10 000 parts, so 80 GB is the largest object one +account can produce. Past it the account's backup fails with a clear refusal, +is counted in the run's `offhost_backup_oversized` meta, and the pass carries +on with everybody else. + +### Reading a failed run + +A nightly run that could not upload for **anybody** now fails the pg-boss job +instead of reporting success. This is the case wrong credentials, a missing +bucket and an unreachable endpoint all land in, and the target's own sentence +rides out as the failure cause — `The request signature we calculated does not +match the signature you provided`, `The specified bucket does not exist`, +`connect ECONNREFUSED`. Check `offhost_backup_uploaded` against +`offhost_backup_total_users`: before this change a run where every single +upload failed still read `ok: true`, so a bucket could stay empty while the +jobs page looked healthy. + +A run where SOME account got a copy still succeeds, with the rest counted in +`offhost_backup_failed`. Failing the whole queue over one account's object +would re-upload everybody's on every retry. + ## The weekly in-database backup (`data-backup`) Separate from the off-host job above, and easy to confuse with it. A second diff --git a/package.json b/package.json index 71be99776..a939de0bc 100644 --- a/package.json +++ b/package.json @@ -54,6 +54,7 @@ }, "dependencies": { "@aws-sdk/client-s3": "^3.1075.0", + "@aws-sdk/lib-storage": "3.1075.0", "@dnd-kit/core": "^6.3.1", "@dnd-kit/sortable": "^10.0.0", "@dnd-kit/utilities": "^3.2.2", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 43fb9d4c3..0422f5883 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -33,6 +33,9 @@ importers: '@aws-sdk/client-s3': specifier: ^3.1075.0 version: 3.1075.0 + '@aws-sdk/lib-storage': + specifier: 3.1075.0 + version: 3.1075.0(@aws-sdk/client-s3@3.1075.0) '@dnd-kit/core': specifier: ^6.3.1 version: 6.3.1(react-dom@19.2.7(react@19.2.7))(react@19.2.7) @@ -345,6 +348,12 @@ packages: resolution: {integrity: sha512-g2BoECD1q01kTPByi56+VLVvdWDzMkKIcr77qixpqH0okw2t0U5CoPv+6S8v/D1Y2Wa6QKKtn6XAtDzP+Kfpvg==} engines: {node: '>=20.0.0'} + '@aws-sdk/lib-storage@3.1075.0': + resolution: {integrity: sha512-Npdac3/Fv994iCdFFeqloPel+95Zv6dRiRUpfey6W/CqpB/QdnYP60U4YBhBgJj6V5020Pm3ad0HV4WzCE5hHQ==} + engines: {node: '>=20.0.0'} + peerDependencies: + '@aws-sdk/client-s3': ^3.1075.0 + '@aws-sdk/middleware-flexible-checksums@3.974.33': resolution: {integrity: sha512-qMgQSPemQq2/eW/e/0+SpY4kYR5L7dUgBiVdEc5bd+ztHNv07ZMYiI+sTiir3TgKndFfglSw/VFi7oZJ6bZ63g==} engines: {node: '>=20.0.0'} @@ -2641,6 +2650,10 @@ packages: resolution: {integrity: sha512-mLUktFAn+Pa2agl1J7VgtYNFWCX8/b4GMJSK1hCu4YCvtBfM6F8Os3EP4ry+DFFlXOf3wyvlgXhuUdFoy52D3g==} engines: {node: '>=18.0.0'} + '@smithy/core@3.33.3': + resolution: {integrity: sha512-CsOeKq/9kA3y6VJHt+/+VTCtBaxJ4OTFpgrjIUhPpDIKxBci1k2bJaQASF2h/ELWrulGp+t97DZ0mevfAD8idg==} + engines: {node: '>=18.0.0'} + '@smithy/credential-provider-imds@4.4.2': resolution: {integrity: sha512-18UMDMyrAbDcpmL1gLUA7ww0fRTcdCrSjSJOi2Sbld+tVjwD/pW+OAwjlScFLR7vvBnhZrIPQ7kVuTf1mnJLug==} engines: {node: '>=18.0.0'} @@ -2665,6 +2678,10 @@ packages: resolution: {integrity: sha512-Z5TAOxygoFvybJV3igo5SloFflSokHx2hu1eFA+DxDTcn+FtKxUSui+rbTRG1pAafMA888Z3MVvCWUuvCrTXjg==} engines: {node: '>=18.0.0'} + '@smithy/types@4.18.0': + resolution: {integrity: sha512-CgB6HHWer/vrKps24ulRIbpcpb7K4xAU7SkZ7YHzBPlwHsvsrCJFEXK421s+cJzX+ZrqtA/TuU5w1HzI7k9N8A==} + engines: {node: '>=18.0.0'} + '@smithy/util-buffer-from@2.2.0': resolution: {integrity: sha512-IJdWBbTcMQ6DA0gdNhh/BwrLkDR+ADW5Kr1aZmd4k3DIF6ezMV4R2NIAmT08wQJ3yUK82thHWmC/TnK/wpMMIA==} engines: {node: '>=14.0.0'} @@ -3442,6 +3459,9 @@ packages: buffer-equal-constant-time@1.0.1: resolution: {integrity: sha512-zRpUiDwd/xk6ADqPMATG8vc9VPrkck7T07OIx0gnjmJAnHnTVXNQG3vfvWNuiZIkwu9KrKdA1iJKfsfTVxE6NA==} + buffer@5.6.0: + resolution: {integrity: sha512-/gDYp/UtU0eA1ys8bOs9J6a+E/KWIY+DZ+Q2WESNUA0jFRsJOc0SNUO6xJ5SGA1xueg3NL65W6s+NY5l9cunuw==} + buffer@5.7.1: resolution: {integrity: sha512-EHcyIPBQ4BSGlvjB16k5KgAJ27CIsHY/2JBmCRReo48y9rQ3MaUzWX3KVlBa4U7MyX02HdVj0K7C3WaB3ju7FQ==} @@ -6158,6 +6178,9 @@ packages: resolution: {integrity: sha512-eLoXW/DHyl62zxY4SCaIgnRhuMr6ri4juEYARS8E6sCEqzKpOiE521Ucofdx+KnDZl5xmvGYaaKCk5FEOxJCoQ==} engines: {node: '>= 0.4'} + stream-browserify@3.0.0: + resolution: {integrity: sha512-H73RAHsVBapbim0tU2JwwOiXUj+fikfiaoYAKHF3VJfA0pe2BCzkhAHBlLG6REzE+2WNZcxOXjK7lkso+9euLA==} + streamx@2.28.0: resolution: {integrity: sha512-1Yowhzjf0ivGMrTIkY9hav5TxobO9qIVqUE41fiCGMGgc3CLlf4MY+9AHmZqBWgDTue0fY9zWjYFVyf6Diuobw==} @@ -6867,8 +6890,8 @@ snapshots: '@aws-crypto/util': 5.2.0 '@aws-sdk/core': 3.974.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/client-s3@3.1075.0': @@ -6893,9 +6916,9 @@ snapshots: '@aws-sdk/types': 3.973.13 '@aws-sdk/xml-builder': 3.972.31 '@aws/lambda-invoke-store': 0.2.4 - '@smithy/core': 3.26.0 + '@smithy/core': 3.33.3 '@smithy/signature-v4': 5.5.2 - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 bowser: 2.14.1 tslib: 2.8.1 @@ -6903,18 +6926,18 @@ snapshots: dependencies: '@aws-sdk/core': 3.974.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/credential-provider-http@3.972.51': dependencies: '@aws-sdk/core': 3.974.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 + '@smithy/core': 3.33.3 '@smithy/fetch-http-handler': 5.5.2 '@smithy/node-http-handler': 4.8.2 - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/credential-provider-ini@3.972.56': @@ -6928,9 +6951,9 @@ snapshots: '@aws-sdk/credential-provider-web-identity': 3.972.55 '@aws-sdk/nested-clients': 3.997.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 + '@smithy/core': 3.33.3 '@smithy/credential-provider-imds': 4.4.2 - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/credential-provider-login@3.972.55': @@ -6938,8 +6961,8 @@ snapshots: '@aws-sdk/core': 3.974.23 '@aws-sdk/nested-clients': 3.997.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/credential-provider-node@3.972.58': @@ -6951,17 +6974,17 @@ snapshots: '@aws-sdk/credential-provider-sso': 3.972.55 '@aws-sdk/credential-provider-web-identity': 3.972.55 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 + '@smithy/core': 3.33.3 '@smithy/credential-provider-imds': 4.4.2 - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/credential-provider-process@3.972.49': dependencies: '@aws-sdk/core': 3.974.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/credential-provider-sso@3.972.55': @@ -6970,8 +6993,8 @@ snapshots: '@aws-sdk/nested-clients': 3.997.23 '@aws-sdk/token-providers': 3.1074.0 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/credential-provider-web-identity@3.972.55': @@ -6979,8 +7002,18 @@ snapshots: '@aws-sdk/core': 3.974.23 '@aws-sdk/nested-clients': 3.997.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 + tslib: 2.8.1 + + '@aws-sdk/lib-storage@3.1075.0(@aws-sdk/client-s3@3.1075.0)': + dependencies: + '@aws-sdk/client-s3': 3.1075.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 + buffer: 5.6.0 + events: 3.3.0 + stream-browserify: 3.0.0 tslib: 2.8.1 '@aws-sdk/middleware-flexible-checksums@3.974.33': @@ -6993,8 +7026,8 @@ snapshots: '@aws-sdk/core': 3.974.23 '@aws-sdk/signature-v4-multi-region': 3.996.35 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/nested-clients@3.997.23': @@ -7004,17 +7037,17 @@ snapshots: '@aws-sdk/core': 3.974.23 '@aws-sdk/signature-v4-multi-region': 3.996.35 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 + '@smithy/core': 3.33.3 '@smithy/fetch-http-handler': 5.5.2 '@smithy/node-http-handler': 4.8.2 - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/signature-v4-multi-region@3.996.35': dependencies: '@aws-sdk/types': 3.973.13 '@smithy/signature-v4': 5.5.2 - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/token-providers@3.1074.0': @@ -7022,13 +7055,13 @@ snapshots: '@aws-sdk/core': 3.974.23 '@aws-sdk/nested-clients': 3.997.23 '@aws-sdk/types': 3.973.13 - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/types@3.973.13': dependencies: - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws-sdk/util-locate-window@3.965.5': @@ -7037,7 +7070,7 @@ snapshots: '@aws-sdk/xml-builder@3.972.31': dependencies: - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@aws/lambda-invoke-store@0.2.4': {} @@ -9186,19 +9219,24 @@ snapshots: '@smithy/core@3.26.0': dependencies: '@aws-crypto/crc32': 5.2.0 - '@smithy/types': 4.15.0 + '@smithy/types': 4.18.0 + tslib: 2.8.1 + + '@smithy/core@3.33.3': + dependencies: + '@smithy/types': 4.18.0 tslib: 2.8.1 '@smithy/credential-provider-imds@4.4.2': dependencies: - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@smithy/fetch-http-handler@5.5.2': dependencies: - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@smithy/is-array-buffer@2.2.0': @@ -9207,20 +9245,24 @@ snapshots: '@smithy/node-http-handler@4.8.2': dependencies: - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@smithy/signature-v4@5.5.2': dependencies: - '@smithy/core': 3.26.0 - '@smithy/types': 4.15.0 + '@smithy/core': 3.33.3 + '@smithy/types': 4.18.0 tslib: 2.8.1 '@smithy/types@4.15.0': dependencies: tslib: 2.8.1 + '@smithy/types@4.18.0': + dependencies: + tslib: 2.8.1 + '@smithy/util-buffer-from@2.2.0': dependencies: '@smithy/is-array-buffer': 2.2.0 @@ -10002,6 +10044,11 @@ snapshots: buffer-equal-constant-time@1.0.1: {} + buffer@5.6.0: + dependencies: + base64-js: 1.5.1 + ieee754: 1.2.1 + buffer@5.7.1: dependencies: base64-js: 1.5.1 @@ -12954,6 +13001,11 @@ snapshots: es-errors: 1.3.0 internal-slot: 1.1.0 + stream-browserify@3.0.0: + dependencies: + inherits: 2.0.4 + readable-stream: 3.6.2 + streamx@2.28.0: dependencies: events-universal: 1.0.1 diff --git a/src/lib/crypto.ts b/src/lib/crypto.ts index 5d8506387..11d381643 100644 --- a/src/lib/crypto.ts +++ b/src/lib/crypto.ts @@ -429,6 +429,76 @@ export function extractKeyId(encoded: string): string | null { /** Marker prefix of a streamed ciphertext. Disjoint from every other format. */ const STREAM_CODEC_PREFIX = "~hlgcm1."; +/** + * Incremental AES-256-GCM writer over raw bytes. + * + * The byte-level half of the codec, split out from the base64 one below + * because not every destination is a text column. `iv | update()* | final()` + * concatenated in order is the same `iv | ciphertext | authTag` the base64 + * form encodes, so the two write the same bytes and only differ in how they + * are framed on the way out. + * + * The key is passed in rather than looked up. Every application row goes + * through `createStreamEncryptor()` under the active `ENCRYPTION_KEYS` entry, + * but the off-host backup is deliberately encrypted under a SEPARATE key + * (`BACKUP_ENCRYPTION_KEY`) so a leak of one does not expose the other — and + * it needs this exact writer, not a second scheme. + */ +export interface RawStreamEncryptor { + /** The 12-byte IV. Belongs in front of the ciphertext. */ + readonly iv: Buffer; + /** Encrypt one plaintext chunk. May return an empty buffer. */ + update(chunk: Buffer): Buffer; + /** Flush the cipher and append the auth tag. No further calls after this. */ + final(): Buffer; +} + +/** Open a byte-level streaming encryptor under an explicit 32-byte key. */ +export function createRawStreamEncryptor(key: Buffer): RawStreamEncryptor { + if (key.byteLength !== 32) { + throw new Error("Stream encryption key must be 32 bytes"); + } + const iv = randomBytes(IV_LENGTH); + const cipher = createCipheriv(ALGORITHM, key, iv); + let done = false; + + return { + iv, + update(chunk: Buffer): Buffer { + if (done) throw new Error("Stream encryptor already finalised"); + return cipher.update(chunk); + }, + final(): Buffer { + if (done) throw new Error("Stream encryptor already finalised"); + done = true; + // The tag is plaintext-independent trailing data, so it simply joins + // the byte stream. + return Buffer.concat([cipher.final(), cipher.getAuthTag()]); + }, + }; +} + +/** + * Verify and decrypt `iv | ciphertext | authTag` under an explicit key. + * + * Deliberately not incremental: the tag covers every ciphertext byte and is + * checked by `final()` before a single plaintext byte is returned, and a + * caller that reads a backup parses it as one document anyway. Releasing + * unverified plaintext to save a copy would trade the authentication for + * memory. + */ +export function decryptRawStream(packed: Buffer, key: Buffer): Buffer { + if (packed.byteLength < IV_LENGTH + AUTH_TAG_LENGTH) { + throw new Error("Streamed ciphertext is truncated"); + } + const iv = packed.subarray(0, IV_LENGTH); + const tag = packed.subarray(packed.byteLength - AUTH_TAG_LENGTH); + const ct = packed.subarray(IV_LENGTH, packed.byteLength - AUTH_TAG_LENGTH); + const dec = createDecipheriv(ALGORITHM, key, iv); + dec.setAuthTag(tag); + return Buffer.concat([dec.update(ct), dec.final()]); +} + /** Incremental AES-256-GCM writer. Emits base64 pieces; concatenate in order. */ export interface StreamEncryptor { /** The header, including base64(iv). Emit before any `update()` output. */ @@ -440,7 +510,7 @@ export interface StreamEncryptor { } /** - * Open a streaming encryptor under the ACTIVE key. + * Open a streaming encryptor under the ACTIVE key, framed as base64. * * The 12-byte IV is exactly four base64 groups, so it encodes standalone and * the ciphertext continues on a group boundary — which is what lets the rest @@ -448,11 +518,9 @@ export interface StreamEncryptor { */ export function createStreamEncryptor(): StreamEncryptor { const { id, key } = getActiveKey(); - const iv = randomBytes(IV_LENGTH); - const cipher = createCipheriv(ALGORITHM, key, iv); + const raw = createRawStreamEncryptor(key); // Bytes that did not fill a 3-byte base64 group in the previous chunk. let carry = Buffer.alloc(0); - let done = false; const emit = (bytes: Buffer): string => { const buf = carry.byteLength > 0 ? Buffer.concat([carry, bytes]) : bytes; @@ -462,17 +530,13 @@ export function createStreamEncryptor(): StreamEncryptor { }; return { - header: `${STREAM_CODEC_PREFIX}${id}.${iv.toString("base64")}`, + header: `${STREAM_CODEC_PREFIX}${id}.${raw.iv.toString("base64")}`, update(chunk: Buffer): string { - if (done) throw new Error("Stream encryptor already finalised"); - return emit(cipher.update(chunk)); + return emit(raw.update(chunk)); }, final(): string { - if (done) throw new Error("Stream encryptor already finalised"); - done = true; - // The tag is plaintext-independent trailing data, so it simply joins the - // byte stream; the last group is padded exactly once, here. - const tail = Buffer.concat([carry, cipher.final(), cipher.getAuthTag()]); + // The last group is padded exactly once, here. + const tail = Buffer.concat([carry, raw.final()]); carry = Buffer.alloc(0); return tail.toString("base64"); }, @@ -511,16 +575,7 @@ export function decryptStream(stored: string): Buffer { `ENCRYPTION_KEYS before decrypting rows written under that key.`, ); } - const packed = Buffer.from(rest.slice(dot + 1), "base64"); - if (packed.byteLength < IV_LENGTH + AUTH_TAG_LENGTH) { - throw new Error("Streamed ciphertext is truncated"); - } - const iv = packed.subarray(0, IV_LENGTH); - const tag = packed.subarray(packed.byteLength - AUTH_TAG_LENGTH); - const ct = packed.subarray(IV_LENGTH, packed.byteLength - AUTH_TAG_LENGTH); - const dec = createDecipheriv(ALGORITHM, key, iv); - dec.setAuthTag(tag); - return Buffer.concat([dec.update(ct), dec.final()]); + return decryptRawStream(Buffer.from(rest.slice(dot + 1), "base64"), key); } /** The key id a streamed ciphertext was written under, or null when unparsable. */ diff --git a/src/lib/jobs/__tests__/offhost-backup.test.ts b/src/lib/jobs/__tests__/offhost-backup.test.ts index de9dbe17f..f507cb1f5 100644 --- a/src/lib/jobs/__tests__/offhost-backup.test.ts +++ b/src/lib/jobs/__tests__/offhost-backup.test.ts @@ -1,4 +1,5 @@ import { createCipheriv, randomBytes } from "node:crypto"; +import type { Readable } from "node:stream"; import { describe, it, expect, beforeEach, vi } from "vitest"; import { @@ -10,20 +11,14 @@ const mocks = vi.hoisted(() => ({ buildFullBackupPayload: vi.fn(), })); -// The uploader takes the already-serialised form, so the stand-in for -// `buildFullBackupJson` serialises whatever the payload mock was told to -// return. Every case below keeps stubbing the PAYLOAD, which is the thing -// these tests are actually about. +// Only the payload builder is stubbed. The REAL streaming writer runs on top +// of it, so every case below keeps stubbing the PAYLOAD — which is what these +// tests are about — while the framing that turns it into object bytes is the +// framing the job actually uses. `isDeferredRows` rides along because the +// writer asks it about every section; nothing here defers. vi.mock("@/lib/export/full-backup-payload", () => ({ buildFullBackupPayload: mocks.buildFullBackupPayload, - buildFullBackupJson: async (...args: unknown[]) => - JSON.stringify( - ( - (await mocks.buildFullBackupPayload(...args)) as { - payload: unknown; - } - ).payload, - ), + isDeferredRows: () => false, })); import { encryptBackup, @@ -31,6 +26,7 @@ import { loadOffhostConfig, runOffhostBackup, runOffhostRoundtripTest, + uploadEncryptedBackup, } from "../offhost-backup"; const ENC_KEY = @@ -114,6 +110,14 @@ function makeS3Mock() { const store = new Map(); return { store, + // Consumes what it is given rather than storing the stream: the upload + // path is what applies backpressure to the producer, so a double that did + // not read would deadlock instead of failing. + putStream: vi.fn(async (k: string, body: Readable) => { + const chunks: Buffer[] = []; + for await (const c of body) chunks.push(Buffer.from(c as Uint8Array)); + store.set(k, Buffer.concat(chunks)); + }), putObject: vi.fn(async (k: string, b: Buffer) => { store.set(k, Buffer.from(b)); }), @@ -214,7 +218,14 @@ describe("runOffhostBackup", () => { expect(mocks.buildFullBackupPayload).toHaveBeenCalledWith(prisma, "u1", { purpose: "disaster-recovery", exportedAt: now, + // The three unbounded tables are declared, not read. Without this the + // writer is streaming a payload that was materialised first, which is + // the arrangement that took the container down. + deferBulk: true, }); + // Nothing went through the whole-buffer arm. + expect(s3.putObject).not.toHaveBeenCalled(); + expect(s3.putStream).toHaveBeenCalledTimes(2); }); it("uploads the canonical builder output without reshaping it", async () => { @@ -252,8 +263,11 @@ describe("runOffhostBackup", () => { }, ], }; + // A copy, because the streaming writer releases each section from the + // payload as it goes — that destructive walk is what keeps its peak at one + // section, and handing it the fixture itself would empty the expectation. mocks.buildFullBackupPayload.mockResolvedValueOnce({ - payload: canonicalPayload, + payload: structuredClone(canonicalPayload), counts: {}, }); @@ -298,6 +312,156 @@ describe("runOffhostBackup", () => { expect(report.uploaded).toBe(1); expect(report.failed).toBe(1); }); + + it("refuses an account whose object outgrows one multipart upload", async () => { + const s3 = makeS3Mock(); + const prisma = { + user: { findMany: vi.fn().mockResolvedValue([{ id: "u1" }]) }, + }; + mocks.buildFullBackupPayload.mockResolvedValue({ + payload: { + schemaVersion: BACKUP_SCHEMA_VERSION, + exportedAt: "2026-05-08T00:00:00.000Z", + userId: "u1", + // Incompressible, so the limit is crossed by real object bytes rather + // than by gzip failing to shrink a repetitive fixture. + measurements: Array.from({ length: 400 }, (_, at) => ({ + id: `m-${at}`, + note: randomBytes(64).toString("hex"), + })), + }, + counts: {}, + }); + + const report = await runOffhostBackup( + prisma as never, + s3, + new Date("2026-05-08T00:00:00Z"), + { maxBytes: 1024 }, + ); + + expect(report.uploaded).toBe(0); + expect(report.failed).toBe(1); + expect(report.oversized).toBe(1); + expect(report.failures[0]?.message).toContain("a single object may occupy"); + // The refusal is the account's, not the process's, and it leaves nothing + // half-written behind it. + expect(s3.store.size).toBe(0); + }); + + it("reports the largest object it wrote", async () => { + const s3 = makeS3Mock(); + const prisma = { + user: { findMany: vi.fn().mockResolvedValue([{ id: "u1" }]) }, + }; + const report = await runOffhostBackup( + prisma as never, + s3, + new Date("2026-05-08T00:00:00Z"), + ); + expect(report.largestObjectBytes).toBe( + s3.store.get("2026-05-08/user-u1.json.enc")!.byteLength, + ); + expect(report.oversized).toBe(0); + }); +}); + +describe("uploadEncryptedBackup", () => { + const key = Buffer.from(ENC_KEY, "hex"); + + it("writes a version-3 object that reads back byte for byte", async () => { + const s3 = makeS3Mock(); + const document = JSON.stringify({ + hello: "world", + rows: Array.from({ length: 5_000 }, (_, at) => ({ at })), + }); + + const bytes = await uploadEncryptedBackup(s3, "k", key, async (write) => { + // In pieces, because that is how the real producer arrives. + for (let at = 0; at < document.length; at += 997) { + await write(document.slice(at, at + 997)); + } + }); + + const stored = s3.store.get("k")!; + expect(stored.byteLength).toBe(bytes); + expect(stored.subarray(0, 5).toString("binary")).toBe("HLBK\x03"); + expect(decryptBackup(stored, key)).toBe(document); + }); + + it("restores an object of the old shape and one of the new one alike", async () => { + const s3 = makeS3Mock(); + const document = JSON.stringify({ userId: "u1", n: 7 }); + + // A genuine version-2 object: the writer that produced every object + // already sitting in an operator's bucket. + const legacy = encryptBackup(document, key); + await s3.putObject("old", legacy); + await uploadEncryptedBackup(s3, "new", key, (write) => write(document)); + + const oldBytes = s3.store.get("old")!; + const newBytes = s3.store.get("new")!; + expect(oldBytes.subarray(0, 5).toString("binary")).toBe("HLBK\x02"); + expect(newBytes.subarray(0, 5).toString("binary")).toBe("HLBK\x03"); + // Different framing, same record, one reader. + expect(decryptBackup(oldBytes, key)).toBe(document); + expect(decryptBackup(newBytes, key)).toBe(document); + }); + + it("rejects a tampered version-3 object rather than returning a partial one", async () => { + const s3 = makeS3Mock(); + await uploadEncryptedBackup(s3, "k", key, (write) => + write(JSON.stringify({ userId: "u1" })), + ); + const stored = Buffer.from(s3.store.get("k")!); + // One flipped bit in the ciphertext body, well clear of the trailing tag. + stored[20] ^= 0x01; + expect(() => decryptBackup(stored, key)).toThrow(); + }); + + it("fails the upload rather than the process when the producer throws", async () => { + const s3 = makeS3Mock(); + await expect( + uploadEncryptedBackup(s3, "k", key, async (write) => { + await write("{"); + throw new Error("db gone"); + }), + ).rejects.toThrow("db gone"); + expect(s3.store.has("k")).toBe(false); + }); + + it("stops the producer when the bucket refuses the upload mid-write", async () => { + const s3 = makeS3Mock(); + // Refused after the producer is already going, and without reading a byte + // — the shape a rejected CreateMultipartUpload has. The producer is by + // then waiting on the compressor to drain, and the reader that would have + // drained it is gone. + s3.putStream.mockImplementationOnce(async () => { + await new Promise((resolve) => setTimeout(resolve, 50)); + throw new Error("The specified bucket does not exist"); + }); + + await expect( + uploadEncryptedBackup(s3, "k", key, async (write) => { + // Well past the compressor's buffer, so the wait is real. + for (let at = 0; at < 200; at++) { + await write(randomBytes(4096).toString("hex")); + } + }), + ).rejects.toThrow("The specified bucket does not exist"); + }); + + it("surfaces the target's own message when the upload is refused", async () => { + const s3 = makeS3Mock(); + s3.putStream.mockRejectedValueOnce( + new Error( + "The request signature we calculated does not match the signature you provided.", + ), + ); + await expect( + uploadEncryptedBackup(s3, "k", key, (write) => write("{}")), + ).rejects.toThrow("The request signature we calculated"); + }); }); describe("runOffhostRoundtripTest", () => { diff --git a/src/lib/jobs/__tests__/restore-drill.test.ts b/src/lib/jobs/__tests__/restore-drill.test.ts index 46cb5a13b..e5367f857 100644 --- a/src/lib/jobs/__tests__/restore-drill.test.ts +++ b/src/lib/jobs/__tests__/restore-drill.test.ts @@ -1,3 +1,4 @@ +import type { Readable } from "node:stream"; import { describe, it, expect, beforeEach, vi } from "vitest"; import { encryptBackup } from "../offhost-backup"; import { @@ -29,6 +30,14 @@ function makeS3Mock(initial: Record = {}) { const store = new Map(Object.entries(initial)); return { store, + // Consumes what it is given rather than storing the stream: the upload + // path is what applies backpressure to the producer, so a double that did + // not read would deadlock instead of failing. + putStream: vi.fn(async (k: string, body: Readable) => { + const chunks: Buffer[] = []; + for await (const c of body) chunks.push(Buffer.from(c as Uint8Array)); + store.set(k, Buffer.concat(chunks)); + }), putObject: vi.fn(async (k: string, b: Buffer | Uint8Array) => { store.set(k, Buffer.from(b)); }), diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 4419ba044..7d372364a 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -138,6 +138,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "notified", "offhost_backup_configured", "offhost_backup_failed", + "offhost_backup_oversized", "offhost_backup_total_users", "offhost_backup_uploaded", "outcome", diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index 6d070e7e7..9f868623e 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -14,13 +14,19 @@ * Retention: the worker NEVER calls DeleteObject on backup keys. Operators * MUST configure a bucket-level lifecycle rule (e.g. expire after * `BACKUP_RETENTION_DAYS`). This keeps the IAM grant for the worker - * limited to PutObject + GetObject, so a compromised worker cannot wipe - * the backup history. See docs/ops/backup-restore.md. + * limited to PutObject + GetObject + AbortMultipartUpload, so a compromised + * worker cannot wipe the backup history. The abort is what cleans up a run + * that failed partway rather than leaving billed, unlistable parts behind; + * `AbortMultipartUpload` can only touch an upload this worker started, never + * a finished object. See docs/ops/backup-restore.md. */ +import { Buffer } from "node:buffer"; import { createCipheriv, createDecipheriv, randomBytes } from "node:crypto"; -import { gunzipSync, gzipSync } from "node:zlib"; +import { Transform, type Readable } from "node:stream"; +import { createGzip, gunzipSync, gzipSync } from "node:zlib"; import type { PrismaClient } from "@/generated/prisma/client"; -import { buildFullBackupJson } from "@/lib/export/full-backup-payload"; +import { createRawStreamEncryptor, decryptRawStream } from "@/lib/crypto"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; import { getEvent } from "@/lib/logging/context"; const ALGORITHM = "aes-256-gcm"; @@ -82,24 +88,45 @@ export function loadOffhostConfig(): OffhostBackupConfig | null { } /** - * Encrypt a JSON payload with the dedicated backup key. + * The envelope one off-host object is written in. * - * Wire format (binary): - * magic(4)="HLBK" || version(1) || iv(12) || tag(16) || ciphertext + * Wire format (binary), by version byte: + * 1: magic(4)="HLBK" || 0x01 || iv(12) || tag(16) || ciphertext(json) + * 2: magic(4)="HLBK" || 0x02 || iv(12) || tag(16) || ciphertext(gzip(json)) + * 3: magic(4)="HLBK" || 0x03 || iv(12) || ciphertext(gzip(json)) || tag(16) * - * Version 1 encrypts the JSON directly. Version 2 gzips it first, which is - * what the version byte was there for: a large account's dump is hundreds of - * megabytes of extremely repetitive JSON, and encrypting it whole means the - * string, the cipher's input copy, the ciphertext and the concatenation are - * all resident at once — the same arithmetic that made the in-database weekly - * backup unfinishable. Gzip takes an order of magnitude off every one of them - * and off the object in the bucket. Objects written under version 1 stay - * readable; both the restore drill and `scripts/restore-backup.ts` come - * through `decryptBackup`, so neither needs to know which it got. + * Version 1 encrypted the JSON directly; version 2 gzipped it first. Both put + * the tag in front of the ciphertext, and that is precisely what could not be + * written a piece at a time: GCM only produces the tag once the last block is + * in, so a leading tag means the whole object has to exist before its first + * byte can be emitted. Version 3 moves the tag to the end and changes nothing + * else about the authentication — it still covers every ciphertext byte, and + * `decryptBackup` still verifies it before returning a single byte of + * plaintext. It is the same move `~hlgcm1.` made for the in-database blob, and + * it uses the same writer. + * + * Every version reads. An operator's bucket holds objects written by whichever + * release was running that night, and the newest usable copy is exactly the one + * that must not need a matching binary; `decryptBackup` takes all three and + * neither `scripts/restore-backup.ts` nor the monthly restore drill needs to + * know which it got. */ const BACKUP_ENVELOPE_PLAIN = 0x01; const BACKUP_ENVELOPE_GZIP = 0x02; +const BACKUP_ENVELOPE_STREAM = 0x03; +const MAGIC = "HLBK"; +/** magic(4) + version(1). Where the per-version body begins. */ +const PREAMBLE_LENGTH = 5; +/** + * Write a whole JSON string as a version-2 object. + * + * The job does not use this any more — it streams, and a streaming writer + * cannot produce a leading tag. It stays because version 2 is the shape + * sitting in every operator's bucket today, and the test that proves both + * shapes restore has to write a genuine old object rather than a hand-built + * byte string that only looks like one. + */ export function encryptBackup(plaintext: string, key: Buffer): Buffer { const iv = randomBytes(IV_LENGTH); const cipher = createCipheriv(ALGORITHM, key, iv); @@ -109,7 +136,7 @@ export function encryptBackup(plaintext: string, key: Buffer): Buffer { ]); const tag = cipher.getAuthTag(); const header = Buffer.from([ - ...Buffer.from("HLBK", "binary"), + ...Buffer.from(MAGIC, "binary"), BACKUP_ENVELOPE_GZIP, ]); return Buffer.concat([header, iv, tag, ct]); @@ -117,16 +144,27 @@ export function encryptBackup(plaintext: string, key: Buffer): Buffer { export function decryptBackup(buf: Buffer, key: Buffer): string { const magic = buf.subarray(0, 4).toString("binary"); - const version = buf[4]; + const version = buf[PREAMBLE_LENGTH - 1]; if ( - magic !== "HLBK" || - (version !== BACKUP_ENVELOPE_PLAIN && version !== BACKUP_ENVELOPE_GZIP) + magic !== MAGIC || + (version !== BACKUP_ENVELOPE_PLAIN && + version !== BACKUP_ENVELOPE_GZIP && + version !== BACKUP_ENVELOPE_STREAM) ) { throw new Error("Invalid backup envelope (bad magic or version)"); } - const iv = buf.subarray(5, 5 + IV_LENGTH); - const tag = buf.subarray(5 + IV_LENGTH, 5 + IV_LENGTH + TAG_LENGTH); - const ct = buf.subarray(5 + IV_LENGTH + TAG_LENGTH); + if (version === BACKUP_ENVELOPE_STREAM) { + // iv | ciphertext | tag, exactly what the streaming writer emits and what + // the shared reader verifies whole before it hands back a byte. + const plaintext = decryptRawStream(buf.subarray(PREAMBLE_LENGTH), key); + return gunzipSync(plaintext).toString("utf8"); + } + const iv = buf.subarray(PREAMBLE_LENGTH, PREAMBLE_LENGTH + IV_LENGTH); + const tag = buf.subarray( + PREAMBLE_LENGTH + IV_LENGTH, + PREAMBLE_LENGTH + IV_LENGTH + TAG_LENGTH, + ); + const ct = buf.subarray(PREAMBLE_LENGTH + IV_LENGTH + TAG_LENGTH); const dec = createDecipheriv(ALGORITHM, key, iv); dec.setAuthTag(tag); const plaintext = Buffer.concat([dec.update(ct), dec.final()]); @@ -135,8 +173,251 @@ export function decryptBackup(buf: Buffer, key: Buffer): string { : plaintext.toString("utf8"); } +/** + * How much of one object the upload holds at a time, and how many of those + * windows are in flight. `@aws-sdk/lib-storage` buffers `partSize` bytes per + * queued part, so this pair — not the object — is the upload's footprint: + * 16 MB, whatever the record turns out to be. + */ +const UPLOAD_PART_BYTES = 8 * 1024 * 1024; +const UPLOAD_CONCURRENCY = 2; + +/** + * The largest object one multipart upload can carry: S3 and every compatible + * target cap a multipart upload at 10 000 parts. + * + * This is the only ceiling the write path still has. Nothing here grows with + * the record any more — the JSON is produced a page at a time, gzip and the + * cipher consume it as it arrives, and the upload holds two parts — so there + * is no memory bound left to state, and inventing one would be theatre. What + * remains is structural: past 10 000 parts the SDK fails the upload partway + * through with an error about part numbers, having already written most of the + * object. Counting the bytes as they are produced turns that into one clear + * refusal, and the count is what the test drives against a small limit. + */ +const MAX_MULTIPART_PARTS = 10_000; + +/** Default cap for one uploaded object, in bytes. */ +export function defaultOffhostObjectLimit(): number { + return UPLOAD_PART_BYTES * MAX_MULTIPART_PARTS; +} + +/** Bytes as an operator reads them. Kilobytes below a megabyte. */ +function size(bytes: number): string { + const mb = bytes / 1024 / 1024; + if (bytes < 1024 * 1024) return `${Math.round(bytes / 1024)} KB`; + return mb < 100 ? `${mb.toFixed(1)} MB` : `${Math.round(mb)} MB`; +} + +/** + * Thrown when one account's encrypted object outgrows what a single multipart + * upload can carry. That account's backup fails; every other account's still + * runs, and nothing partial is left in the bucket. + */ +export class OffhostBackupTooLargeError extends Error { + readonly bytes: number; + readonly limitBytes: number; + + constructor(bytes: number, limitBytes: number) { + super( + `Off-host backup stopped after ${size(bytes)} of encrypted backup for ` + + `one account, over the ${size(limitBytes)} a single object may ` + + `occupy (${MAX_MULTIPART_PARTS} parts of ${size(UPLOAD_PART_BYTES)}). ` + + `Nothing was uploaded for this account.`, + ); + this.name = "OffhostBackupTooLargeError"; + this.bytes = bytes; + this.limitBytes = limitBytes; + } +} + +/** + * gzip bytes in, framed version-3 object bytes out, counted as they go. + * + * The header is emitted lazily so it rides in front of the first ciphertext + * piece rather than needing a separate write, and on `flush` when there was no + * plaintext at all — an empty object is still a well-formed envelope. + */ +function createEnvelopeStream( + key: Buffer, + limitBytes: number, +): { stream: Transform; bytes: () => number } { + const encryptor = createRawStreamEncryptor(key); + const header = Buffer.concat([ + Buffer.from(MAGIC, "binary"), + Buffer.from([BACKUP_ENVELOPE_STREAM]), + encryptor.iv, + ]); + let written = 0; + let headerEmitted = false; + + const take = (piece: Buffer): Buffer => { + written += piece.byteLength; + if (written > limitBytes) { + throw new OffhostBackupTooLargeError(written, limitBytes); + } + return piece; + }; + + const preamble = (into: Buffer[]): void => { + if (headerEmitted) return; + headerEmitted = true; + into.push(take(header)); + }; + + const stream = new Transform({ + transform(chunk: Buffer, _encoding, callback): void { + try { + const out: Buffer[] = []; + preamble(out); + const piece = encryptor.update(chunk); + if (piece.byteLength > 0) out.push(take(piece)); + callback(null, Buffer.concat(out)); + } catch (err) { + callback(err as Error); + } + }, + flush(callback): void { + try { + const out: Buffer[] = []; + preamble(out); + out.push(take(encryptor.final())); + callback(null, Buffer.concat(out)); + } catch (err) { + callback(err as Error); + } + }, + }); + + return { stream, bytes: () => written }; +} + +/** Produces the backup JSON in pieces. Every piece is written in order. */ +export type BackupJsonProducer = ( + write: (chunk: string) => Promise, +) => Promise; + +export interface UploadBackupOptions { + /** + * Largest object this call may upload, in bytes. Defaults to + * `defaultOffhostObjectLimit()`. Tests pass an explicit value; nothing else + * should need to. + */ + maxBytes?: number; +} + +/** + * Produce one account's backup JSON and put it in the bucket, holding none of + * it. + * + * JSON piece → gzip → AES-256-GCM → multipart upload, with backpressure the + * whole way: the gzip stream's `write` tells the producer when to wait, the + * envelope only ever holds one chunk, and the uploader holds + * `UPLOAD_CONCURRENCY` parts. What the process holds is therefore fixed by + * this pipeline's shape rather than by the size of the record going through + * it — which is the entire difference from what this job did before, where the + * JSON string, the gzip buffer, the ciphertext and the request body were all + * resident at once. + * + * Answers the number of object bytes written. + */ +export async function uploadEncryptedBackup( + s3: S3Like, + objectKey: string, + encryptionKey: Buffer, + produce: BackupJsonProducer, + options: UploadBackupOptions = {}, +): Promise { + const limitBytes = options.maxBytes ?? defaultOffhostObjectLimit(); + const gzip = createGzip(); + const { stream: envelope, bytes } = createEnvelopeStream( + encryptionKey, + limitBytes, + ); + + let failure: unknown = null; + // Both directions, or one end's failure hangs the other: a gzip error has to + // reach the uploader, and the envelope refusing an oversized object has to + // stop the producer. + gzip.on("error", (err: Error) => { + failure ??= err; + envelope.destroy(err); + }); + envelope.on("error", (err: Error) => { + failure ??= err; + // WITH the error, not bare. A bare destroy leaves a producer that is + // waiting on `drain` waiting forever — the refusal would hang the account + // it was supposed to fail, which is a worse outcome than the size it was + // refusing. + gzip.destroy(err); + }); + gzip.pipe(envelope); + + // Started before the producer runs: the uploader is what drains the + // envelope, and without a reader the first part's worth of backpressure + // would stall the producer forever. + const uploaded = s3.putStream(objectKey, envelope).then( + () => null, + (err: unknown) => { + // A refused upload takes the reader away, and a producer that is + // waiting on `drain` would wait for a reader that is never coming + // back. Tearing the pipeline down here is what turns "the bucket said + // no" into a failed account rather than a job that never returns. + failure ??= err; + gzip.destroy(err instanceof Error ? err : new Error(String(err))); + return err; + }, + ); + + const write = async (chunk: string): Promise => { + if (failure) throw failure; + if (gzip.write(chunk, "utf8")) return; + await new Promise((resolve, reject) => { + const onDrain = (): void => { + gzip.off("error", onError); + resolve(); + }; + const onError = (err: Error): void => { + gzip.off("drain", onDrain); + reject(err); + }; + gzip.once("drain", onDrain); + gzip.once("error", onError); + }); + }; + + try { + await produce(write); + gzip.end(); + } catch (err) { + gzip.destroy(); + envelope.destroy(); + // Settled, not ignored: tearing the pipeline down makes the uploader + // reject too, and an unawaited rejection would surface later with nothing + // around it. What it says is only an echo — the producer's own failure is + // the one that explains the run, and when the envelope refused the object + // the producer already rethrew that refusal verbatim. + await uploaded; + throw err; + } + + const uploadError = await uploaded; + if (failure) throw failure; + if (uploadError) throw uploadError; + return bytes(); +} + export interface S3Like { putObject(key: string, body: Buffer | Uint8Array): Promise; + /** + * Put an object whose body arrives as a stream, without buffering it. + * + * Separate from `putObject` rather than an overload of it: the one-byte + * health check wants a plain PUT and a test double wants a value it can + * assert on, while this arm has to be a multipart upload and has to consume + * what it is given. + */ + putStream(key: string, body: Readable): Promise; getObject(key: string): Promise; headObject(key: string): Promise; listObjects( @@ -170,6 +451,33 @@ export async function getS3Client(cfg: OffhostBackupConfig): Promise { }; return { + putStream: async (key, body) => { + // Dynamic for the same reason as the client above: an environment + // without the SDK must still be able to import this module. + const storage = (await import("@aws-sdk/lib-storage").catch((err) => { + throw new Error( + `@aws-sdk/lib-storage is not installed (${(err as Error).message}). ` + + `Run: pnpm add @aws-sdk/lib-storage`, + ); + })) as typeof import("@aws-sdk/lib-storage"); + + const upload = new storage.Upload({ + client, + params: { + Bucket: cfg.bucket, + Key: key, + Body: body, + ContentType: "application/octet-stream", + }, + queueSize: UPLOAD_CONCURRENCY, + partSize: UPLOAD_PART_BYTES, + // A failed upload leaves nothing behind. Orphaned parts are billed + // and are invisible in a bucket listing, so an operator would never + // find them. + leavePartsOnError: false, + }); + await upload.done(); + }, putObject: async (key, body) => { await client.send( new mod.PutObjectCommand({ @@ -219,12 +527,19 @@ interface BackupRunReport { failed: number; failures: Array<{ userId: string; message: string }>; totalUsers: number; + /** The biggest object this run wrote. Tracks the record over time. */ + largestObjectBytes: number; + /** Accounts refused for size rather than failed for a reason. */ + oversized: number; } +export type RunOffhostBackupOptions = UploadBackupOptions; + export async function runOffhostBackup( prisma: PrismaClient, s3Override?: S3Like, now: Date = new Date(), + options: RunOffhostBackupOptions = {}, ): Promise { const cfg = loadOffhostConfig(); if (!cfg) { @@ -238,22 +553,32 @@ export async function runOffhostBackup( const users = await prisma.user.findMany({ select: { id: true } }); let uploaded = 0; let failed = 0; + let oversized = 0; + let largestObjectBytes = 0; const failures: Array<{ userId: string; message: string }> = []; const evt = getEvent(); for (const user of users) { try { - const ciphertext = encryptBackup( - await buildFullBackupJson(prisma, user.id, { - purpose: "disaster-recovery", - exportedAt: now, - }), + const objectBytes = await uploadEncryptedBackup( + s3, + `${dateKey}/user-${user.id}.json.enc`, cfg.encryptionKey, + // The same writer the weekly in-database pass uses. The payload + // builder was always shared; everything after it was not, which is why + // this job kept dying on a record the weekly one had learned to + // survive. + (write) => + streamFullBackupJson(prisma, user.id, write, { + purpose: "disaster-recovery", + exportedAt: now, + }), + options, ); - const key = `${dateKey}/user-${user.id}.json.enc`; - await s3.putObject(key, ciphertext); + largestObjectBytes = Math.max(largestObjectBytes, objectBytes); uploaded++; } catch (err) { failed++; + if (err instanceof OffhostBackupTooLargeError) oversized++; const message = (err as Error).message ?? "unknown"; failures.push({ userId: user.id, message: message.slice(0, 200) }); // Surface per-user failure detail so an operator can tell WHICH user @@ -274,6 +599,8 @@ export async function runOffhostBackup( failed, failures, totalUsers: users.length, + largestObjectBytes, + oversized, }; } diff --git a/src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts b/src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts new file mode 100644 index 000000000..41bfb935b --- /dev/null +++ b/src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts @@ -0,0 +1,141 @@ +/** + * What the nightly off-host pass tells the queue about itself. + * + * The rule under test is the one the weekly in-database pass already carries: + * a run that put nothing off-host protected nobody, and `ok: true` about that + * is how a bucket stays empty while the jobs page reads healthy. Wrong + * credentials, a bucket that does not exist and a target that refuses the + * signature all fail every account rather than one, so they land in exactly + * that arm — with the target's own sentence as the cause, not a stack. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + runOffhostBackup: vi.fn(), + getWorkerPrisma: vi.fn(() => ({})), +})); + +vi.mock("@/lib/jobs/offhost-backup", async () => { + const actual = await vi.importActual< + typeof import("@/lib/jobs/offhost-backup") + >("@/lib/jobs/offhost-backup"); + return { + // The real error class, so the handler's `instanceof` arm is the one that + // runs rather than a look-alike. + OffhostBackupNotConfiguredError: actual.OffhostBackupNotConfiguredError, + runOffhostBackup: mocks.runOffhostBackup, + }; +}); + +vi.mock("@/lib/logging/background", () => ({ + withBackgroundEvent: vi.fn( + async (_name: string, run: (event: object) => Promise) => + run({ + addMeta: vi.fn(), + addWarning: vi.fn(), + setBackground: vi.fn(), + setError: vi.fn(), + }), + ), +})); + +vi.mock("../shared", () => ({ getWorkerPrisma: mocks.getWorkerPrisma })); + +import { OffhostBackupNotConfiguredError } from "@/lib/jobs/offhost-backup"; +import { handleOffhostBackup } from "../backup-handlers"; + +function report(over: Partial> = {}) { + return { + config: { endpoint: "https://r2.example", bucket: "hl", region: "auto" }, + uploaded: 2, + failed: 0, + failures: [], + totalUsers: 2, + largestObjectBytes: 9_000, + oversized: 0, + ...over, + }; +} + +describe("handleOffhostBackup", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("succeeds when every account reached the bucket", async () => { + mocks.runOffhostBackup.mockResolvedValue(report()); + const outcome = await handleOffhostBackup([]); + expect(outcome.ok).toBe(true); + expect(outcome.ok && outcome.did).toEqual({ + offhost_backup_uploaded: 2, + offhost_backup_failed: 0, + offhost_backup_total_users: 2, + offhost_backup_oversized: 0, + }); + }); + + it("fails the run when no account could be uploaded", async () => { + mocks.runOffhostBackup.mockResolvedValue( + report({ + uploaded: 0, + failed: 2, + failures: [ + { + userId: "u1", + message: + "The request signature we calculated does not match the signature you provided.", + }, + { userId: "u2", message: "same" }, + ], + largestObjectBytes: 0, + }), + ); + + const outcome = await handleOffhostBackup([]); + + expect(outcome.ok).toBe(false); + expect(!outcome.ok && outcome.reason).toBe("no account could be uploaded"); + // The target's own words, so an operator reads what to fix. + expect(!outcome.ok && outcome.cause).toContain( + "The request signature we calculated", + ); + expect(!outcome.ok && outcome.did).toMatchObject({ + offhost_backup_uploaded: 0, + offhost_backup_failed: 2, + }); + }); + + it("still succeeds when some account got a copy", async () => { + mocks.runOffhostBackup.mockResolvedValue( + report({ + uploaded: 1, + failed: 1, + failures: [{ userId: "u2", message: "db gone" }], + }), + ); + const outcome = await handleOffhostBackup([]); + // Fanning the whole cohort out again over one object would re-upload + // everybody's. + expect(outcome.ok).toBe(true); + }); + + it("does not fail a host that has no accounts at all", async () => { + mocks.runOffhostBackup.mockResolvedValue( + report({ uploaded: 0, failed: 0, totalUsers: 0, largestObjectBytes: 0 }), + ); + const outcome = await handleOffhostBackup([]); + // Absence of work is not failure. + expect(outcome.ok).toBe(true); + }); + + it("stays quiet on the self-hosts that never configured a bucket", async () => { + mocks.runOffhostBackup.mockRejectedValue( + new OffhostBackupNotConfiguredError("nope"), + ); + const outcome = await handleOffhostBackup([]); + expect(outcome.ok).toBe(true); + expect(outcome.ok && outcome.did).toEqual({ + offhost_backup_configured: false, + }); + }); +}); diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 97b6f5432..48c6ae397 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -40,6 +40,12 @@ export async function handleOffhostBackup( evt.addMeta("offhost_backup_total_users", report.totalUsers); evt.addMeta("offhost_backup_endpoint", report.config.endpoint); evt.addMeta("offhost_backup_bucket", report.config.bucket); + // The uploaded size is the one number that says whether this pass is + // heading back towards the wall it hit before: it tracks the record. + evt.addMeta( + "offhost_backup_largest_object_bytes", + report.largestObjectBytes, + ); // Per-user failure detail is also emitted as warnings inside // runOffhostBackup; echo a structured digest for at-a-glance triage. if (report.failures.length > 0) { @@ -48,14 +54,37 @@ export async function handleOffhostBackup( JSON.stringify(report.failures.slice(0, 10)), ); } - // Per-user upload failures ride out as `offhost_backup_failed`: the run - // itself uploaded what it could, and failing the queue over one user's - // object would re-upload the whole cohort on every retry. - return jobDone({ + + const did = { offhost_backup_uploaded: report.uploaded, offhost_backup_failed: report.failed, offhost_backup_total_users: report.totalUsers, - }); + offhost_backup_oversized: report.oversized, + }; + + // A run that uploaded nothing for anybody put nothing off-host, and + // `ok: true` about that is how a bucket stays empty while the job page + // reads healthy — the same rule the weekly pass already carries. Wrong + // credentials, a bucket that does not exist and a target that refuses + // the signature all land here, because they fail every account rather + // than one. Per-account failures still ride out as counts when SOME + // account got a copy: that is the fan-out rule, and retrying the whole + // cohort over one object would re-upload everybody's. + if (report.totalUsers > 0 && report.uploaded === 0) { + // The SDK's own words, not a stack: `runJob` puts the cause message in + // the reported meta, and "SignatureDoesNotMatch" is the sentence an + // operator can act on. + return jobFailed( + "no account could be uploaded", + report.failures[0]?.message, + did, + ); + } + + // Per-user upload failures ride out as `offhost_backup_failed`: the run + // itself uploaded what it could, and failing the queue over one user's + // object would re-upload the whole cohort on every retry. + return jobDone(did); } catch (err) { // Not configured ⇒ skip silently with a warning, not an error: most // self-hosts never set the S3 credentials, and a nightly failed job on diff --git a/tests/integration/admin-backups-canonical-roundtrip.test.ts b/tests/integration/admin-backups-canonical-roundtrip.test.ts index 7a3818ac4..7eabf894c 100644 --- a/tests/integration/admin-backups-canonical-roundtrip.test.ts +++ b/tests/integration/admin-backups-canonical-roundtrip.test.ts @@ -1,4 +1,5 @@ import { Buffer } from "node:buffer"; +import type { Readable } from "node:stream"; import { beforeEach, describe, expect, it, vi } from "vitest"; import type { Prisma } from "@/generated/prisma/client"; @@ -93,6 +94,11 @@ function makeS3Store(): S3Like & { objects: Map } { const objects = new Map(); return { objects, + putStream: vi.fn(async (key: string, body: Readable) => { + const chunks: Buffer[] = []; + for await (const c of body) chunks.push(Buffer.from(c as Uint8Array)); + objects.set(key, Buffer.concat(chunks)); + }), putObject: vi.fn(async (key, body) => { objects.set(key, Buffer.from(body)); }), diff --git a/tests/integration/offhost-backup-streaming-memory.test.ts b/tests/integration/offhost-backup-streaming-memory.test.ts new file mode 100644 index 000000000..4bb6a7bc4 --- /dev/null +++ b/tests/integration/offhost-backup-streaming-memory.test.ts @@ -0,0 +1,335 @@ +/** + * The nightly off-host backup, run against a record big enough to have killed + * the process, with its live memory measured rather than assumed. + * + * What this pins. The weekly in-database pass learned to stream in v1.38.6; + * the off-host one did not. It shared the payload BUILDER and nothing after + * it, so it still made the whole JSON as one string, gzipped that whole + * string, ran a whole-buffer cipher pass over the result and handed the + * finished buffer to `PutObject` — four full copies of the record alive at + * once. On the live instance that was `FATAL ERROR: Reached heap limit` + * seventeen seconds into the first run, and because the job shares the app + * process, one account's size restarted the instance for everybody on it. + * + * Why it measures the way it does. `process.memoryUsage().heapUsed` on its own + * is not a measurement: V8 lets garbage float in proportion to the heap limit, + * and a test fork's limit is several times a container's, so the same code + * "peaks" at wildly different numbers depending on who is running it. Every + * reading below is taken after a forced collection, so what is compared is + * what each writer HOLDS. + * + * The two halves are the whole point. One says the streaming uploader stays + * inside a budget; the other says the materialising path does not fit that + * budget on the same fixture in the same process. Without the second, the + * budget could be any number at all and the first would still pass — green + * because nothing was measured rather than because something was proved. + * + * The uploader here is an in-process stand-in that counts bytes and throws + * them away, which is what a socket does. A run against a real bucket lives in + * `docs/ops/backup-restore.md`; what this file can prove on every gate run is + * the arithmetic, and the arithmetic is what killed the container. + */ +import { Buffer } from "node:buffer"; +import type { Readable } from "node:stream"; +import v8 from "node:v8"; +import vm from "node:vm"; + +import { beforeAll, afterAll, describe, expect, it } from "vitest"; + +import { buildFullBackupPayload } from "@/lib/export/full-backup-payload"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; +import { + OffhostBackupTooLargeError, + decryptBackup, + encryptBackup, + uploadEncryptedBackup, + type S3Like, +} from "@/lib/jobs/offhost-backup"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const OWNER_ID = "offhost-streaming-owner"; +const MEASUREMENT_ROWS = 120_000; +const MOOD_ROWS = 8_000; +const INTAKE_ROWS = 20_000; + +/** The dedicated off-host key, separate from `ENCRYPTION_KEYS` by design. */ +const BACKUP_KEY = Buffer.from( + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + "hex", +); + +/** + * What the streaming uploader may hold on top of the process's own baseline. + * + * Measured, not chosen: on this fixture the uploader holds a page of rows, one + * gzip window and the object bytes still in flight, and the materialising path + * holds the JSON string and the object at once. The budget sits with a wide + * margin either side of that pair, which is what makes it a test rather than a + * coin toss. + */ +const STREAM_BUDGET_BYTES = 48 * 1024 * 1024; + +const prisma = getPrismaClient(); + +/** + * A collection this process can ask for, without the runner having to be + * started with `--expose-gc`. + */ +const forceGc = ((): (() => void) => { + v8.setFlagsFromString("--expose-gc"); + const gc = vm.runInNewContext("gc") as () => void; + v8.setFlagsFromString("--no-expose-gc"); + return gc; +})(); + +/** Heap held after a forced collection. Garbage is not a measurement. */ +function liveHeapBytes(): number { + forceGc(); + forceGc(); + return process.memoryUsage().heapUsed; +} + +/** + * The one number this file exists to produce, on stderr where a gate run keeps + * it. Peaks are the evidence, and evidence that only prints on failure is not + * evidence. + */ +function reportPeak(label: string, heldBytes: number): void { + const mb = (bytes: number): number => Math.round(bytes / 1024 / 1024); + process.stderr.write( + `[offhost-memory] ${label}: peak ${mb(heldBytes)} MB held by the backup, ` + + `${mb(v8.getHeapStatistics().heap_size_limit)} MB heap limit\n`, + ); +} + +/** + * A bucket that behaves like a socket: it consumes what it is handed and keeps + * only the byte count, plus — for the one key the restore assertion reads — + * the object itself, which on this fixture is a couple of megabytes next to a + * 48 MB budget. + */ +function makeCountingS3(keep?: string): S3Like & { + bytes: number; + kept: Buffer | null; + onChunk?: () => void; +} { + const sink = { + bytes: 0, + kept: null as Buffer | null, + onChunk: undefined as (() => void) | undefined, + putStream: async (key: string, body: Readable): Promise => { + const held: Buffer[] = []; + for await (const chunk of body) { + const piece = Buffer.from(chunk as Uint8Array); + sink.bytes += piece.byteLength; + if (key === keep) held.push(piece); + sink.onChunk?.(); + } + if (key === keep) sink.kept = Buffer.concat(held); + }, + putObject: async (): Promise => { + throw new Error("the nightly pass must never take the whole-buffer arm"); + }, + getObject: async (): Promise => { + throw new Error("not used"); + }, + headObject: async (): Promise => false, + listObjects: async (): Promise> => [], + deleteObject: async (): Promise => {}, + }; + return sink; +} + +async function seedLargeRecord(): Promise { + await prisma.user.create({ + data: { id: OWNER_ID, username: "offhost-streaming-owner" }, + }); + await prisma.medication.create({ + data: { + id: "offhost-med", + userId: OWNER_ID, + name: "Seeded", + dose: "10 mg", + }, + }); + // One statement, one round trip. Notes on one row in forty and ciphertext on + // one in twenty-five so the base64 arm of the serialiser is exercised at + // scale, and one in two hundred is a tombstone because a backup that drops + // them resurrects deleted readings on the next device sync. + await prisma.$executeRawUnsafe( + `INSERT INTO measurements ( + id, user_id, type, value, unit, source, measured_at, notes, + notes_encrypted, external_id, created_at, updated_at, sync_version, + deleted_at) + SELECT + 'ox' || lpad(g::text, 10, '0'), + $1, + 'PULSE'::measurement_type, + 60 + (g % 40), + 'bpm', + 'APPLE_HEALTH'::measurement_source, + timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'), + CASE WHEN g % 40 = 0 THEN 'a note recorded with reading ' || g END, + CASE WHEN g % 25 = 0 + THEN decode(md5(g::text) || md5((g + 1)::text), 'hex') END, + 'offhost-stream-' || g, + timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'), + timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'), + 1, + CASE WHEN g % 200 = 0 + THEN timestamp '2026-01-01 00:00:00' + (g * interval '1 second') END + FROM generate_series(1, ${MEASUREMENT_ROWS}) AS g`, + OWNER_ID, + ); + await prisma.$executeRawUnsafe( + `INSERT INTO mood_entries ( + id, user_id, date, mood, score, source, mood_logged_at, synced_at, + created_at, updated_at, tz, note, sync_version, deleted_at) + SELECT + 'oy' || lpad(g::text, 10, '0'), + $1, + to_char(timestamp '2019-01-01' + (g * interval '1 hour'), 'YYYY-MM-DD'), + 'okay', 3, 'MOODLOG', + timestamp '2019-01-01' + (g * interval '1 hour'), + now(), now(), now(), 'Europe/Berlin', + CASE WHEN g % 3 = 0 THEN 'a journal line about day ' || g END, + 1, + CASE WHEN g % 150 = 0 THEN now() END + FROM generate_series(1, ${MOOD_ROWS}) AS g`, + OWNER_ID, + ); + await prisma.$executeRawUnsafe( + `INSERT INTO medication_intake_events ( + id, user_id, medication_id, scheduled_for, taken_at, skipped, source, + created_at, updated_at, sync_version, deleted_at, dose_taken) + SELECT + 'oz' || lpad(g::text, 10, '0'), + $1, + 'offhost-med', + timestamp '2019-01-01' + (g * interval '10 minutes'), + CASE WHEN g % 7 <> 0 + THEN timestamp '2019-01-01' + (g * interval '10 minutes') END, + (g % 7 = 0), + 'WEB'::intake_source, + timestamp '2019-01-01' + (g * interval '10 minutes'), + timestamp '2019-01-01' + (g * interval '10 minutes'), + 1, + CASE WHEN g % 300 = 0 THEN now() END, + '10 mg' + FROM generate_series(1, ${INTAKE_ROWS}) AS g`, + OWNER_ID, + ); +} + +describe("off-host backup under a memory budget", () => { + beforeAll(async () => { + expect( + typeof forceGc, + "without a real collection every reading in this file measures " + + "uncollected garbage and nothing here can fail", + ).toBe("function"); + await truncateAllTables(prisma); + await seedLargeRecord(); + }, 240_000); + + afterAll(async () => { + await truncateAllTables(prisma); + }); + + it("uploads a restorable object while holding a bounded amount of the record", async () => { + const objectKey = "2026-09-05/user-offhost-streaming-owner.json.enc"; + const s3 = makeCountingS3(objectKey); + const baseline = liveHeapBytes(); + let peakHeld = 0; + let samples = 0; + // Sampled rather than continuous: a forced collection per chunk would + // dominate the runtime, and one in twenty still lands inside every phase + // of the walk. + s3.onChunk = () => { + if (samples++ % 20 === 0) { + peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); + } + }; + + const objectBytes = await uploadEncryptedBackup( + s3, + objectKey, + BACKUP_KEY, + (write) => + // The purpose the nightly job asks for: tombstones and ciphertext ride + // verbatim, which is the arm that has to fit in memory. + streamFullBackupJson(prisma, OWNER_ID, write, { + purpose: "disaster-recovery", + }), + ); + peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); + reportPeak("large record", peakHeld); + + expect(objectBytes).toBe(s3.bytes); + expect( + peakHeld, + `the streaming uploader held ${Math.round(peakHeld / 1024 / 1024)} MB ` + + "of a record it is supposed to pass through a page at a time", + ).toBeLessThan(STREAM_BUDGET_BYTES); + + // An object that writes but does not read is worse than none. + const restored = JSON.parse(decryptBackup(s3.kept!, BACKUP_KEY)) as { + measurements: Array<{ deletedAt: string | null }>; + moodEntries: unknown[]; + intakeEvents: unknown[]; + }; + expect(restored.measurements).toHaveLength(MEASUREMENT_ROWS); + expect(restored.moodEntries).toHaveLength(MOOD_ROWS); + expect(restored.intakeEvents).toHaveLength(INTAKE_ROWS); + // Tombstones ride along, or the next device sync resurrects them. + expect( + restored.measurements.filter((row) => row.deletedAt !== null).length, + ).toBe(Math.floor(MEASUREMENT_ROWS / 200)); + }, 300_000); + + it("would not fit that budget if the object were materialised", async () => { + const baseline = liveHeapBytes(); + // Exactly what the job did before, with the step `buildFullBackupJson` + // takes internally spelled out: the payload graph, the JSON string it is + // stringified into, then a whole-buffer gzip-and-encrypt pass, then a + // single put of the finished buffer. + const { payload } = await buildFullBackupPayload(prisma, OWNER_ID, { + purpose: "disaster-recovery", + }); + const json = JSON.stringify(payload); + const object = encryptBackup(json, BACKUP_KEY); + // All three are deliberately still reachable at the reading: holding the + // graph, the document and the finished object at the same time is exactly + // the shape that exhausted the container. + const held = liveHeapBytes() - baseline; + expect(payload.measurements).toHaveLength(MEASUREMENT_ROWS); + expect(json.length).toBeGreaterThan(0); + expect(object.byteLength).toBeGreaterThan(0); + reportPeak("materialised", held); + expect( + held, + "materialising this fixture no longer costs what the budget in this " + + "file assumes; re-measure the budget rather than widening it", + ).toBeGreaterThan(STREAM_BUDGET_BYTES * 2); + }, 300_000); + + it("fails as a job rather than as a process when one object does not fit", async () => { + // The only ceiling the write path still has is structural: a multipart + // upload carries 10 000 parts and no more. An account past it is now a + // failed backup for that account instead of a confusing SDK error halfway + // through an object that is already mostly written. + const s3 = makeCountingS3(); + await expect( + uploadEncryptedBackup( + s3, + "k", + BACKUP_KEY, + (write) => + streamFullBackupJson(prisma, OWNER_ID, write, { + purpose: "disaster-recovery", + }), + { maxBytes: 64 * 1024 }, + ), + ).rejects.toBeInstanceOf(OffhostBackupTooLargeError); + }, 120_000); +}); From 3ee3461f922f9bb8c8fa1dbaaba234b841a715d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 5 Sep 2026 20:23:08 +0200 Subject: [PATCH 2/2] chore(backup): drop the whole-string builder the streaming path replaced Nothing calls it any more: the weekly pass and the off-host job both walk the payload through the streaming writer. Keeping an exported function that builds the entire record as one string invites the next caller to reintroduce the memory profile this branch removes. --- src/lib/export/full-backup-payload.ts | 37 ------------------- src/lib/export/full-backup-stream.ts | 2 +- .../offhost-backup-streaming-memory.test.ts | 2 +- 3 files changed, 2 insertions(+), 39 deletions(-) diff --git a/src/lib/export/full-backup-payload.ts b/src/lib/export/full-backup-payload.ts index 6f1328b98..9aa30beb9 100644 --- a/src/lib/export/full-backup-payload.ts +++ b/src/lib/export/full-backup-payload.ts @@ -661,43 +661,6 @@ export function isDeferredRows(value: unknown): value is DeferredRows { (value as Record)[DEFERRED_ROWS] === true ); } - -/** - * The same payload, already serialised, with the object graph released. - * - * Built through the streaming writer rather than by stringifying a finished - * payload, so the object graph and the string never coexist: the three - * unbounded tables go past a page at a time and every other section is - * released as soon as its JSON exists. What is left is the answer itself, - * which a caller that wants a string necessarily holds. - * - * A caller that does NOT need the string — the weekly job, which only wants - * the compressed ciphertext — should go through `packBackupBlobStreaming` and - * `streamFullBackupJson` directly and never let the whole document exist at - * all. This function is for the callers that genuinely hand a JSON body on. - */ -export async function buildFullBackupJson( - prisma: PrismaClient, - userId: string, - options: FullBackupOptions = {}, -): Promise { - // Imported here rather than at the top of the file: the writer imports this - // module for `buildFullBackupPayload`, and a static edge in both directions - // is a cycle. The call is once per backup, so the cost is nil. - const { streamFullBackupJson } = - await import("@/lib/export/full-backup-stream"); - const pieces: string[] = []; - await streamFullBackupJson( - prisma, - userId, - (chunk) => { - pieces.push(chunk); - }, - options, - ); - return pieces.join(""); -} - /** * Build the canonical full-backup payload for `userId`. Portable exports omit * tombstones and document ciphertext; disaster-recovery payloads preserve diff --git a/src/lib/export/full-backup-stream.ts b/src/lib/export/full-backup-stream.ts index 16de554dd..c5b364aa0 100644 --- a/src/lib/export/full-backup-stream.ts +++ b/src/lib/export/full-backup-stream.ts @@ -1,7 +1,7 @@ /** * The full-backup payload, written out incrementally instead of built. * - * Why it exists. `buildFullBackupJson` has to hold two things at once that + * Why it exists. the old whole-string builder (removed with the streaming path) had to hold two things at once that * both scale with the record: the payload object graph, and the JSON string * `JSON.stringify` makes of it. On a seeded account of 445 000 measurements, * 30 000 mood entries and 60 000 intake events, that pair alone exhausts a diff --git a/tests/integration/offhost-backup-streaming-memory.test.ts b/tests/integration/offhost-backup-streaming-memory.test.ts index 4bb6a7bc4..c5193fd65 100644 --- a/tests/integration/offhost-backup-streaming-memory.test.ts +++ b/tests/integration/offhost-backup-streaming-memory.test.ts @@ -289,7 +289,7 @@ describe("off-host backup under a memory budget", () => { it("would not fit that budget if the object were materialised", async () => { const baseline = liveHeapBytes(); - // Exactly what the job did before, with the step `buildFullBackupJson` + // Exactly what the job did before, with the step the old whole-string builder took // takes internally spelled out: the payload graph, the JSON string it is // stringified into, then a whole-buffer gzip-and-encrypt pass, then a // single put of the finished buffer.