mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 04:32:42 +00:00
Merge pull request #927 from MBombeck/fix/offhost-backup-streaming
fix(backup): stream the off-host copy so it fits a small container
This commit is contained in:
15 files changed
+1341
-150
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
Generated
+90
-38
@@ -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
|
||||
|
||||
+77
-22
@@ -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. */
|
||||
|
||||
@@ -661,43 +661,6 @@ export function isDeferredRows(value: unknown): value is DeferredRows {
|
||||
(value as Record<symbol, unknown>)[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<string> {
|
||||
// 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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<string, Buffer>();
|
||||
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", () => {
|
||||
|
||||
@@ -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<string, Buffer> = {}) {
|
||||
const store = new Map<string, Buffer>(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));
|
||||
}),
|
||||
|
||||
@@ -138,6 +138,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
|
||||
"notified",
|
||||
"offhost_backup_configured",
|
||||
"offhost_backup_failed",
|
||||
"offhost_backup_oversized",
|
||||
"offhost_backup_total_users",
|
||||
"offhost_backup_uploaded",
|
||||
"outcome",
|
||||
|
||||
+357
-30
@@ -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<void>,
|
||||
) => Promise<unknown>;
|
||||
|
||||
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<number> {
|
||||
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<void> => {
|
||||
if (failure) throw failure;
|
||||
if (gzip.write(chunk, "utf8")) return;
|
||||
await new Promise<void>((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<void>;
|
||||
/**
|
||||
* 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<void>;
|
||||
getObject(key: string): Promise<Buffer>;
|
||||
headObject(key: string): Promise<boolean>;
|
||||
listObjects(
|
||||
@@ -170,6 +451,33 @@ export async function getS3Client(cfg: OffhostBackupConfig): Promise<S3Like> {
|
||||
};
|
||||
|
||||
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<BackupRunReport> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -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<unknown>) =>
|
||||
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<Record<string, unknown>> = {}) {
|
||||
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,
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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
|
||||
|
||||
@@ -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<string, Buffer> } {
|
||||
const objects = new Map<string, Buffer>();
|
||||
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));
|
||||
}),
|
||||
|
||||
@@ -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<void> => {
|
||||
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<void> => {
|
||||
throw new Error("the nightly pass must never take the whole-buffer arm");
|
||||
},
|
||||
getObject: async (): Promise<Buffer> => {
|
||||
throw new Error("not used");
|
||||
},
|
||||
headObject: async (): Promise<boolean> => false,
|
||||
listObjects: async (): Promise<Array<{ key: string }>> => [],
|
||||
deleteObject: async (): Promise<void> => {},
|
||||
};
|
||||
return sink;
|
||||
}
|
||||
|
||||
async function seedLargeRecord(): Promise<void> {
|
||||
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);
|
||||
});
|
||||
Reference in new issue
Block a user