diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c807de21..80d5daf31 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/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/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..c5193fd65 --- /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 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. + 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); +});