From c83fdb6773e6c9a35f005f3b9a8829c50a5cc8a4 Mon Sep 17 00:00:00 2001 From: Jonathan Ng Date: Sun, 18 Oct 2026 11:06:47 -0700 Subject: [PATCH 1/2] Keep pod biometrics history off the pod so Watch-labelled nights outlive the pod's retention --- docs/biometrics-datasets.md | 4 + package.json | 1 + scripts/biometrics-archive/README.md | 197 +++++++++++++++ scripts/biometrics-archive/archive.test.ts | 271 ++++++++++++++++++++ scripts/biometrics-archive/cli.ts | 53 ++++ scripts/biometrics-archive/download.ts | 235 +++++++++++++++++ scripts/biometrics-archive/merge.test.ts | 118 +++++++++ scripts/biometrics-archive/merge.ts | 253 +++++++++++++++++++ scripts/biometrics-archive/pull.ts | 280 +++++++++++++++++++++ scripts/biometrics-archive/store.ts | 119 +++++++++ 10 files changed, 1531 insertions(+) create mode 100644 scripts/biometrics-archive/README.md create mode 100644 scripts/biometrics-archive/archive.test.ts create mode 100644 scripts/biometrics-archive/cli.ts create mode 100644 scripts/biometrics-archive/download.ts create mode 100644 scripts/biometrics-archive/merge.test.ts create mode 100644 scripts/biometrics-archive/merge.ts create mode 100644 scripts/biometrics-archive/pull.ts create mode 100644 scripts/biometrics-archive/store.ts diff --git a/docs/biometrics-datasets.md b/docs/biometrics-datasets.md index 1e0a456d..0be8608c 100644 --- a/docs/biometrics-datasets.md +++ b/docs/biometrics-datasets.md @@ -320,3 +320,7 @@ share it. ```sh mkdir -p snapshot && curl -fsS "http://POD:3000/api/biometrics/analysis-export?start=2026-10-15T09:16:00Z&end=2026-10-15T15:24:00Z" | tar -xf - -C snapshot/ ``` + +To keep history beyond the pod's retention, `pnpm biometrics:archive pull` +pulls this export on a schedule into a long-term local store with a merged +database (`scripts/biometrics-archive/README.md`). diff --git a/package.json b/package.json index fdace762..97ed10d9 100644 --- a/package.json +++ b/package.json @@ -31,6 +31,7 @@ "build:biometrics-worker": "node scripts/build/biometrics-worker.mjs", "biometrics:experiment": "tsx scripts/algorithm-experiments/cli.ts", "biometrics:local-fit": "tsx scripts/biometrics-local-fit/cli.ts", + "biometrics:archive": "tsx scripts/biometrics-archive/cli.ts", "test:browser": "node tests/browser/run.mjs" }, "lint-staged": { diff --git a/scripts/biometrics-archive/README.md b/scripts/biometrics-archive/README.md new file mode 100644 index 00000000..3263bdd9 --- /dev/null +++ b/scripts/biometrics-archive/README.md @@ -0,0 +1,197 @@ +# Off-pod biometrics archive + +Keep a pod's biometrics history off the pod for good. A pod keeps about four +weeks of RAW sensor archives (the pruner deletes the oldest at 80% disk) and +90 days of vitals. Every night with Apple Watch data is a labelled night for +tuning pod biometrics, so this tool pulls the pod's read-only analysis export +into a long-term store on a laptop or NAS. Run it daily or weekly. + +## Pull + +```sh +pnpm biometrics:archive pull --pod 192.168.1.88 --dest /Volumes/nas/sleepypod +pnpm biometrics:archive pull --pod pod.local:3000 --dest ~/sleepypod-archive --since 2026-10-01T00:00:00Z --until 2026-10-08T00:00:00Z +pnpm biometrics:archive pull --pod 192.168.1.88 --dest ~/sleepypod-archive --no-raw # database only +``` + +Each pull calls `GET http://POD:3000/api/biometrics/analysis-export?start=&end=&raw=1` +([format](../../docs/biometrics-datasets.md#analysis-snapshot-export)). It's read-only on the pod. + +- **Window.** Without `--since`, a pull continues from the cursor: the largest + served window end among this pod's successful pulls. The first pull covers the + last 14 days. `--until` defaults to now. Ranges longer than 14 days are + split into 14-day windows, oldest first, and each window is committed on its own. +- **429** (another export is streaming): retried up to 6 times. The wait is the + pod's `Retry-After`, or else 30 s doubling each time, at most 5 min. +- **413** (the window's RAW is over 2 GiB): the window is halved, oldest half + first, down to 1 h. Below that the pull fails and tells you to use `--no-raw`. +- **Verification.** The tar streams into `/.staging/`, hashing each entry + as it arrives. Nothing is stored until the export checks out: the manifest + is valid `analysis-export-v1`, it lists exactly the files received with + matching sizes and SHA-256s, the window is the one requested, and + `Content-Length` matches. A corrupt, partial or refused pull leaves the store + unchanged and doesn't move the cursor. +- **Commit order.** RAW blobs first, then the database merge in one + transaction, then the `archive/` links, then the pull record, which is the + cursor. If a pull dies part way, the next one repeats the same window, and + every step is idempotent. +- **Lock.** `/.biometrics-archive.lock` allows one pull per destination. + A lock left by a dead process on the same host is taken over. Otherwise the + pull fails and names the holder. + +## Layout + +``` +/ + .biometrics-archive.lock while a pull runs + / hashed pod id from the manifest; the hostname if it is null (dev machine), with a warning + biometrics.db long-term merged database (pod schema + archive_* tables) + raw/.RAW.gz content-addressed, written once + archive/.RAW.gz hard link to raw/ under the pod's original name + pulls/.json one per successful window +``` + +`archive/` names never get overwritten. If the pod reuses a name for different +bytes (counter reset, reinstall), or a file the archiver didn't write is +already there, the new one becomes `-.RAW.gz`. That's +listed under `raw.renamed` in the pull record and in the `archive_raw_files` +table. On filesystems without hard links the file is copied. + +A pull record holds the pod's manifest verbatim and the requested window. It +also has `raw` counts (new, already stored, linked, copied, renamed), per-table +`tables` counts (snapshot rows, inserted, reassigned, ignored, updated, new +conflicts), the `rows` totals, schema tags, warnings and `cursor`. + +`/` works directly as the data folder for +`analysis/watch-deviation`: it expects `biometrics.db` and `archive/*.RAW.gz`. +Add `watch.json` yourself, then +`SP_BIOMETRICS_DIR=/ analysis/watch-deviation/run.sh …`. The +archive database uses a rollback journal, not WAL, so it can be opened from a +read-only mount. + +## Merging the database + +Each export carries the pod's whole `biometrics.db`, not just the window. The +archive database is created and migrated with this checkout's +`src/db/biometrics-migrations`, so it has the pod's schema plus +`archive_uq_` unique indexes where the pod has no natural key. Before a +merge it's migrated to the newest migration the checkout knows. A snapshot +from a newer pod build fails with "update the checkout" and nothing is stored. +An older snapshot still merges: columns it lacks take their defaults. + +Rules (`TABLES` in `merge.ts`): + +- **Nothing is deleted.** Rows the pod has pruned (90-day retention, deleted + sleep records, deleted reference imports) stay in the archive. +- **Integer `id` surrogates** keep the pod's id when it's free. Ids are + monotonic on one pod. A row whose id is already taken by a different key + gets a new archive id and counts as `reassigned`. That only happens after the + pod's database was reset, since the pod id survives reinstalls. + `vitals_quality.vitals_id` is remapped to the archive id of its vitals row. +- **append**: a row is written once. Insert-or-ignore on the natural key. If a + key matches but the content differs, the archive keeps its row and the + pod's copy goes to `archive_conflicts`, recorded once. +- **state**: the pod updates rows in place. The newest snapshot (by manifest + `generatedAtMs`) overwrites the non-key columns. An older snapshot only + inserts. +- A row the archive's constraints refuse is kept in `archive_conflicts` with + reason `rejected`. A table the checkout has no rule for is merged + append-only on its primary key, with a warning. + +| Tables | Natural key | Mode | +|---|---|---| +| `vitals`, `vitals_quality`, `cap_sense_frames` | side, timestamp | append | +| `movement` | side, timestamp (added) | append | +| `bed_temp`, `freezer_temp`, `ambient_light`, `water_level_readings`, `flow_readings` | timestamp | append | +| `prime_events` | timestamp (added) | append | +| `thermal_state` | side, timestamp, is_powered (added) | append | +| `flash_write_samples` | boot_id, timestamp (added) | append | +| `calibration_runs` | side, sensor_type, created_at (added) | append | +| `reference_measurements` | import_id, record_key | append | +| `biometric_runtime_configs` | side, generation | append; after a pod reset reuses a generation, the new config is a conflict | +| `biometric_holdout_accesses` | id (text) | append | +| `biometric_shadow_samples` | run_id, sequence | append | +| `sleep_records` | side, entered_bed_at, left_bed_at (added) | state; editing a record's times on the pod adds a second row | +| `water_level_alerts` | type, started_at (added) | state (dismissed_at) | +| `pump_alerts` | timestamp, type, side (added) | state (acknowledged/dismissed) | +| `health_runs` | check_id, started_at (added) | state (last_seen_at) | +| `calibration_profiles` | side, sensor_type | state: the current profile; history is in `calibration_runs` | +| `reference_stages` | side, start, source | state: a corrected import replaces the label | +| `reference_vitals` | side, timestamp, source | state | +| `reference_imports` | id (text) | state (deleted_at_ms) | +| `biometric_processing_runs`, `biometric_experiments`, `biometric_backtest_jobs`, `biometric_raw_sources`, `biometric_shadow_runs`, `biometric_dataset_raw_artifacts` | id (text) | state | +| `biometric_backtest_results` | job_id | state | +| `biometric_raw_artifacts` | content_hash | state | +| `biometric_runtime_receipts` | side, component, generation | state | +| `biometric_datasets` | id (text) | state, but `snapshot` and `byte_length` keep their first value: releasing a dataset on the pod replaces them with a tombstone | + +Bookkeeping tables: `archive_snapshots` (each merged snapshot's SHA-256, +`generatedAtMs` and migration), `archive_raw_files` (RAW hash → original name +→ `archive/` name) and `archive_conflicts`. A test fails when a migration adds +a table without a rule here. + +## Storage interface and S3 + +`store.ts` defines the seam. Only `LocalStore` is implemented. + +```ts +interface Store { + put(key, stream, { sha256, size, mtimeMs? }) // atomic replace; rejects bytes that don't match + get(key): Readable + exists(key): boolean + list(prefix): string[] // recursive, sorted + link(from, to): 'linked' | 'copied' // fails if `to` exists +} +``` + +Keys are `/raw/.RAW.gz`, `/archive/` and +`/pulls/.json`. The caller enforces write-once +(`exists` → skip). The SQLite merge needs a local file, so `biometrics.db` +stays outside the interface, at `//biometrics.db`. The +staging directory and the lock file are local too. + +An S3 backend would implement `put` as a multipart upload with +`x-amz-checksum-sha256`, `exists` as `HeadObject`, `list` as `ListObjectsV2` +and `link` as `CopyObject` (`'copied'`). `pull.ts` would take a local work +directory for the database, staging and lock, plus the bucket store for +everything else. After each merge the database would be uploaded as +`/biometrics.db` (versioned bucket), and on a fresh machine it would be +downloaded first. The cursor already comes from `pulls/` in the store. + +## Scheduling + +Nothing is installed for you. Pick a time after the night's last archive is +written (about 45 min after you get up). + +cron (Linux/NAS): + +```cron +30 12 * * * cd /path/to/sleepypod-core && pnpm -s biometrics:archive pull --pod 192.168.1.88 --dest /srv/sleepypod-archive >> /var/log/sleepypod-archive.log 2>&1 +``` + +launchd (macOS), `~/Library/LaunchAgents/com.sleepypod.biometrics-archive.plist`, then +`launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.sleepypod.biometrics-archive.plist`: + +```xml + + + + + Labelcom.sleepypod.biometrics-archive + ProgramArguments + + /bin/zsh-lc + cd ~/src/sleepypod-core && pnpm -s biometrics:archive pull --pod 192.168.1.88 --dest /Volumes/nas/sleepypod + + StartCalendarIntervalHour12Minute30 + StandardOutPath/tmp/sleepypod-archive.log + StandardErrorPath/tmp/sleepypod-archive.log + + +``` + +A missed run (laptop asleep, pod offline) is harmless: the next run picks up at the cursor. Only RAW older than the pod's +retention (about four weeks) is lost, so don't go longer than that between runs. + +**Privacy:** the archive has the full biometrics database for both sides of the +bed plus raw sensor waveforms. Keep it on trusted machines. diff --git a/scripts/biometrics-archive/archive.test.ts b/scripts/biometrics-archive/archive.test.ts new file mode 100644 index 00000000..a5f657f4 --- /dev/null +++ b/scripts/biometrics-archive/archive.test.ts @@ -0,0 +1,271 @@ +// @vitest-environment node +import { createHash } from 'node:crypto' +import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, utimesSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import Database from 'better-sqlite3' +import { drizzle } from 'drizzle-orm/better-sqlite3' +import { migrate } from 'drizzle-orm/better-sqlite3/migrator' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { runArchiveCli } from './cli' +import { ExportIntegrityError } from './download' +import { ArchiveLockedError, findCursor, LOCK_FILE, pull, type PullOptions } from './pull' +import { LocalStore } from './store' + +const state = vi.hoisted(() => ({ archive: '', source: '' })) +vi.mock('@/src/lib/podStorage', () => ({ + get ARCHIVE_DIR() { + return state.archive + }, +})) +vi.mock('@/src/db', () => ({ biometricsDb: { $client: { get name() { + return state.source +} } } })) +vi.mock('@/src/homekit/storage', () => ({ readSeed: () => ({ source: 'mmc-cid', value: 'cid-0123' }) })) +vi.mock('@/src/hardware/dacMonitor.instance', () => ({ getDacMonitorIfRunning: () => null })) + +const DAY = 24 * 60 * 60 * 1000 +const NOW = Date.UTC(2026, 9, 18, 12) +const POD = 'pod.local' + +let root: string +let dest: string +let pod: Database.Database +let route: { GET: (request: Request) => Promise } +const calls: string[] = [] + +const podFetch: typeof fetch = async (input) => { + calls.push(String(input)) + return route.GET(new Request(String(input))) +} + +beforeEach(async () => { + root = mkdtempSync(join(tmpdir(), 'biometrics-archive-test-')) + state.archive = join(root, 'biometrics-archive') + state.source = join(root, 'biometrics.db') + dest = join(root, 'dest') + mkdirSync(state.archive) + mkdirSync(join(root, 'tmpfs')) + vi.stubEnv('TMPDIR', join(root, 'tmpfs')) + pod = new Database(state.source) + pod.pragma('journal_mode = WAL') + migrate(drizzle(pod), { migrationsFolder: 'src/db/biometrics-migrations' }) + calls.length = 0 + vi.resetModules() + route = await import('@/app/api/biometrics/analysis-export/route') +}) +afterEach(() => { + pod.close() + vi.unstubAllEnvs() + rmSync(root, { recursive: true, force: true }) +}) + +const seconds = (ms: number) => Math.floor(ms / 1000) +function vitals(fromMs: number, count: number, side = 'left') { + const insert = pod.prepare('INSERT INTO vitals (side, timestamp, heart_rate) VALUES (?, ?, ?)') + const quality = pod.prepare('INSERT INTO vitals_quality (vitals_id, side, timestamp, quality_score, created_at) VALUES (?, ?, ?, 0.9, ?)') + for (let index = 0; index < count; index++) { + const ts = seconds(fromMs) + index * 60 + const id = insert.run(side, ts, 55 + index).lastInsertRowid + quality.run(id, side, ts, ts) + } +} +function rawArchive(name: string, mtimeMs: number, body = `frame ${name} ${mtimeMs}`) { + const path = join(state.archive, name) + writeFileSync(path, body) + utimesSync(path, mtimeMs / 1000, mtimeMs / 1000) + return createHash('sha256').update(body).digest('hex') +} + +const options = (extra: Partial = {}): PullOptions => ({ pod: POD, dest, includeRaw: true, fetch: podFetch, now: () => NOW, sleep: async () => {}, ...extra }) +const podDir = () => readdirSync(dest).filter(name => !name.startsWith('.')) +function archiveDb(read: (db: Database.Database) => T): T { + const db = new Database(join(dest, podDir()[0], 'biometrics.db'), { readonly: true }) + try { + return read(db) + } + finally { + db.close() + } +} +const count = (table: string) => archiveDb(db => (db.prepare(`SELECT COUNT(*) AS n FROM ${table}`).get() as { n: number }).n) +function listing(directory: string): string[] { + return readdirSync(directory, { withFileTypes: true }).flatMap(entry => entry.isDirectory() + ? listing(join(directory, entry.name)).map(name => `${entry.name}/${name}`) + : [`${entry.name}:${createHash('sha256').update(readFileSync(join(directory, entry.name))).digest('hex')}`]) +} + +describe('biometrics archive pull', () => { + it('stores a first pull: content-addressed RAW, archive links, merged database and a pull record', async () => { + vitals(NOW - 3 * DAY, 5) + const a = rawArchive('A0000001.RAW.gz', NOW - 3 * DAY) + const b = rawArchive('A0000002.RAW.gz', NOW - DAY) + + const [record] = await pull(options()) + expect(calls).toEqual([`http://pod.local:3000/api/biometrics/analysis-export?start=${NOW - 14 * DAY}&end=${NOW}&raw=1`]) + const [dir] = podDir() + expect(dir).toMatch(/^[0-9a-f]{32}$/) + expect(readdirSync(join(dest, dir, 'raw')).sort()).toEqual([`${a}.RAW.gz`, `${b}.RAW.gz`].sort()) + const linked = join(dest, dir, 'archive', 'A0000001.RAW.gz') + expect(statSync(linked).ino).toBe(statSync(join(dest, dir, 'raw', `${a}.RAW.gz`)).ino) + expect(statSync(linked).mtimeMs).toBe(NOW - 3 * DAY) + expect(count('vitals')).toBe(5) + expect(count('vitals_quality')).toBe(5) + expect(record).toMatchObject({ podHost: 'pod.local:3000', podDir: dir, cursor: { endMs: NOW }, raw: { new: 2, skipped: 0, linked: 2, renamed: [] } }) + expect(record.tables.vitals).toMatchObject({ snapshot: 5, inserted: 5, reassigned: 0, ignored: 0, conflicts: 0 }) + const pulls = readdirSync(join(dest, dir, 'pulls')) + expect(pulls).toHaveLength(1) + expect(JSON.parse(readFileSync(join(dest, dir, 'pulls', pulls[0]), 'utf8')).manifest.formatVersion).toBe('analysis-export-v1') + expect(existsSync(join(dest, LOCK_FILE))).toBe(false) + expect(existsSync(join(dest, '.staging'))).toBe(false) + }) + + it('is idempotent: the same window again adds nothing, and the cursor makes the default pull a no-op', async () => { + vitals(NOW - 2 * DAY, 4) + rawArchive('A0000001.RAW.gz', NOW - 2 * DAY) + await pull(options()) + const [again] = await pull(options({ sinceMs: NOW - 14 * DAY, untilMs: NOW })) + expect(again.raw).toMatchObject({ new: 0, skipped: 1, linked: 0, copied: 0 }) + expect(again.rows).toEqual({ inserted: 0, reassigned: 0, ignored: again.rows.ignored, updated: 0, conflicts: 0 }) + expect(again.rows.ignored).toBeGreaterThan(0) + expect(count('vitals')).toBe(4) + + calls.length = 0 + expect(await pull(options())).toEqual([]) + expect(calls).toEqual([]) + }) + + it('merges overlapping windows without duplicating RAW or rows', async () => { + const t0 = NOW - 10 * DAY + vitals(t0, 3) + rawArchive('A0000001.RAW.gz', t0 + 12 * 60 * 60 * 1000) + rawArchive('A0000002.RAW.gz', t0 + 36 * 60 * 60 * 1000) + await pull(options({ sinceMs: t0, untilMs: t0 + 2 * DAY })) + vitals(t0 + 2 * DAY, 2) + rawArchive('A0000003.RAW.gz', t0 + 60 * 60 * 60 * 1000) + const [second] = await pull(options({ sinceMs: t0 + DAY, untilMs: t0 + 3 * DAY })) + expect(second.raw).toMatchObject({ new: 1, skipped: 1 }) + expect(second.tables.vitals).toMatchObject({ inserted: 2, ignored: 3 }) + expect(count('vitals')).toBe(5) + expect(readdirSync(join(dest, podDir()[0], 'archive')).sort()).toEqual(['A0000001.RAW.gz', 'A0000002.RAW.gz', 'A0000003.RAW.gz']) + }) + + it('keeps history the pod has since pruned', async () => { + vitals(NOW - 13 * DAY, 3) + await pull(options({ now: () => NOW - 5 * DAY })) + pod.exec(`DELETE FROM vitals_quality; DELETE FROM vitals WHERE timestamp < ${seconds(NOW - 10 * DAY)}`) + vitals(NOW - 2 * DAY, 2) + const [second] = await pull(options()) + expect(second.window).toMatchObject({ startMs: NOW - 5 * DAY, endMs: NOW }) + expect(count('vitals')).toBe(5) + expect(count('vitals_quality')).toBe(5) + }) + + it.each([ + ['a flipped RAW byte', (bytes: Buffer) => { + const at = bytes.indexOf('frame A0000009') + bytes[at] ^= 0xff + return bytes + }, 'SHA-256 mismatch: raw/A0000009.RAW.gz'], + ['a cut-short download', (bytes: Buffer) => bytes.subarray(0, bytes.length - 2048), 'truncated'], + ])('rejects %s without changing the store or the cursor', async (_label, corrupt, message) => { + vitals(NOW - 3 * DAY, 2) + await pull(options({ now: () => NOW - DAY })) + const before = listing(dest) + vitals(NOW - DAY / 2, 2) + rawArchive('A0000009.RAW.gz', NOW - DAY / 2) + const corrupting: typeof fetch = async (input) => { + const response = await podFetch(input) + return new Response(new Uint8Array(corrupt(Buffer.from(await response.arrayBuffer()))), { status: 200, headers: response.headers }) + } + await expect(pull(options({ fetch: corrupting }))).rejects.toThrow(message) + await expect(pull(options({ fetch: corrupting }))).rejects.toBeInstanceOf(ExportIntegrityError) + expect(listing(dest)).toEqual(before) + expect(await findCursor(new LocalStore(dest), dest, 'pod.local:3000', () => {})).toBe(NOW - DAY) + }) + + it('retries 429 with backoff, honouring Retry-After', async () => { + vitals(NOW - DAY, 1) + let busy = 2 + const delays: number[] = [] + const fetchImpl: typeof fetch = async (input) => { + if (busy-- > 0) return Response.json({ error: 'Another analysis export is in progress, retry later' }, { status: 429, headers: busy ? { 'Retry-After': '60' } : {} }) + return podFetch(input) + } + const records = await pull(options({ fetch: fetchImpl, sleep: async (ms) => { + delays.push(ms) + }, retry: { attempts: 3, baseDelayMs: 1000, maxDelayMs: 45_000 } })) + expect(delays).toEqual([45_000, 2000]) + expect(records).toHaveLength(1) + + busy = 5 + await expect(pull(options({ fetch: fetchImpl, sinceMs: NOW - DAY, untilMs: NOW, retry: { attempts: 2, baseDelayMs: 1, maxDelayMs: 1 } }))).rejects.toThrow('HTTP 429') + }) + + it('halves the window on 413 down to the floor, oldest half first', async () => { + vitals(NOW - 12 * DAY, 2) + const fetchImpl: typeof fetch = async (input) => { + const query = new URL(String(input)).searchParams + if (Number(query.get('end')) - Number(query.get('start')) > 4 * DAY) return Response.json({ error: 'over the limit' }, { status: 413 }) + return podFetch(input) + } + const records = await pull(options({ fetch: fetchImpl })) + expect(records.map(record => [record.window.startMs, record.window.endMs])).toEqual([ + [NOW - 14 * DAY, NOW - 10.5 * DAY], [NOW - 10.5 * DAY, NOW - 7 * DAY], [NOW - 7 * DAY, NOW - 3.5 * DAY], [NOW - 3.5 * DAY, NOW], + ]) + expect(count('vitals')).toBe(2) + + const always413: typeof fetch = async () => Response.json({ error: 'over the limit' }, { status: 413 }) + await expect(pull(options({ fetch: always413, sinceMs: NOW - DAY, untilMs: NOW, minWindowMs: 6 * 60 * 60 * 1000 }))) + .rejects.toThrow(/over the limit.*cannot be halved below 6 h; pull it with --no-raw/) + }) + + it('splits ranges longer than 14 days into chunks, oldest first', async () => { + await pull(options({ sinceMs: NOW - 30 * DAY, includeRaw: false })) + expect(calls.map(call => new URL(call).searchParams.toString())).toEqual([ + `start=${NOW - 30 * DAY}&end=${NOW - 16 * DAY}&raw=0`, `start=${NOW - 16 * DAY}&end=${NOW - 2 * DAY}&raw=0`, `start=${NOW - 2 * DAY}&end=${NOW}&raw=0`, + ]) + }) + + it('lets only one pull into a destination run at a time', async () => { + mkdirSync(dest) + writeFileSync(join(dest, LOCK_FILE), JSON.stringify({ pid: process.pid, host: (await import('node:os')).hostname(), startedAt: 'earlier' })) + await expect(pull(options())).rejects.toBeInstanceOf(ArchiveLockedError) + expect(calls).toEqual([]) + + // A lock left by a process that no longer exists is taken over. + writeFileSync(join(dest, LOCK_FILE), JSON.stringify({ pid: 999_999, host: (await import('node:os')).hostname(), startedAt: 'earlier' })) + await expect(pull(options())).resolves.toHaveLength(1) + + const results = await Promise.allSettled([pull(options({ sinceMs: NOW - DAY })), pull(options({ sinceMs: NOW - DAY }))]) + expect(results.map(result => result.status).sort()).toEqual(['fulfilled', 'rejected']) + expect((results.find(result => result.status === 'rejected') as PromiseRejectedResult).reason).toBeInstanceOf(ArchiveLockedError) + }) + + it('never overwrites an archive name the pod reused for different bytes', async () => { + const first = rawArchive('A0000001.RAW.gz', NOW - 5 * DAY, 'before the counter reset') + await pull(options({ now: () => NOW - 4 * DAY })) + rmSync(join(state.archive, 'A0000001.RAW.gz')) + const second = rawArchive('A0000001.RAW.gz', NOW - DAY, 'after the counter reset') + const [record] = await pull(options()) + const dir = join(dest, podDir()[0]) + const renamed = `A0000001-${second.slice(0, 12)}.RAW.gz` + expect(record.raw.renamed).toEqual([{ name: 'A0000001.RAW.gz', archiveName: renamed, sha256: second }]) + expect(readFileSync(join(dir, 'archive', 'A0000001.RAW.gz'), 'utf8')).toBe('before the counter reset') + expect(readFileSync(join(dir, 'archive', renamed), 'utf8')).toBe('after the counter reset') + expect(archiveDb(db => db.prepare('SELECT sha256, name, archive_name AS archiveName FROM archive_raw_files ORDER BY mtime_ms').all())).toEqual([ + { sha256: first, name: 'A0000001.RAW.gz', archiveName: 'A0000001.RAW.gz' }, + { sha256: second, name: 'A0000001.RAW.gz', archiveName: renamed }, + ]) + }) + + it('validates CLI arguments before contacting the pod', async () => { + await expect(runArchiveCli(['pull', '--dest', dest], { fetch: podFetch })).rejects.toThrow('--pod is required') + await expect(runArchiveCli(['pull', '--pod', 'pod local', '--dest', dest], { fetch: podFetch })).rejects.toThrow('--pod is HOST or HOST:PORT') + await expect(runArchiveCli(['pull', '--pod', POD, '--dest', dest, '--since', 'yesterday'], { fetch: podFetch })).rejects.toThrow('--since must be an ISO 8601 timestamp') + await expect(runArchiveCli(['pull', '--pod', POD, '--dest', dest, '--since', '2026-10-10T00:00:00Z', '--until', '2026-10-01T00:00:00Z'], { fetch: podFetch })) + .rejects.toThrow('--since must be before --until') + await expect(runArchiveCli(['push'])).rejects.toThrow('Usage') + expect(calls).toEqual([]) + }) +}) diff --git a/scripts/biometrics-archive/cli.ts b/scripts/biometrics-archive/cli.ts new file mode 100644 index 00000000..2c3b71d9 --- /dev/null +++ b/scripts/biometrics-archive/cli.ts @@ -0,0 +1,53 @@ +import { resolve } from 'node:path' +import { pathToFileURL } from 'node:url' +import { z } from 'zod' +import { pull, type PullOptions } from './pull' + +const usage = `Usage: + pull --pod HOST[:PORT] --dest DIR [--since ISO] [--until ISO] [--no-raw] +Pulls the pod's analysis export (GET /api/biometrics/analysis-export) into DIR//. +Without --since, continues from the last successful pull (first run: the last 14 days).` + +const instant = (flag: string) => z.string().refine(value => Number.isFinite(Date.parse(value)), `${flag} must be an ISO 8601 timestamp`) + .transform(value => Date.parse(value)) + +const pullArgsSchema = z.object({ + pod: z.string({ error: '--pod is required' }).regex(/^(\[[0-9A-Fa-f:.]+\]|[A-Za-z0-9.-]+)(:\d{1,5})?$/, '--pod is HOST or HOST:PORT'), + dest: z.string({ error: '--dest is required' }).min(1), + since: instant('--since').optional(), + until: instant('--until').optional(), + noRaw: z.boolean(), +}).refine(args => args.since === undefined || args.until === undefined || args.since < args.until, { message: '--since must be before --until' }) + +function options(args: string[]) { + const values: Record = {} + for (let index = 0; index < args.length; index++) { + const arg = args[index] + if (arg === '--no-raw') values.noRaw = true + else if (['--pod', '--dest', '--since', '--until'].includes(arg) && index + 1 < args.length) values[arg.slice(2)] = args[++index] + else throw new Error(`Unknown or incomplete option: ${arg}\n${usage}`) + } + return { noRaw: false, ...values } +} + +export async function runArchiveCli(args: string[], overrides: Partial = {}) { + const [command, ...rest] = args + if (command !== 'pull') throw new Error(usage) + const parsed = pullArgsSchema.safeParse(options(rest)) + if (!parsed.success) throw new Error(`${parsed.error.issues.map(issue => issue.message).join('; ')}\n${usage}`) + const { pod, dest, since, until, noRaw } = parsed.data + const records = await pull({ pod, dest: resolve(dest), sinceMs: since, untilMs: until, includeRaw: !noRaw, + log: line => process.stderr.write(`${line}\n`), ...overrides }) + process.stdout.write(`${JSON.stringify({ pulls: records.length, cursor: records.at(-1)?.cursor.end ?? null, + newRaw: records.reduce((total, record) => total + record.raw.new, 0), + rowsAdded: records.reduce((total, record) => total + record.rows.inserted + record.rows.reassigned, 0), + conflicts: records.reduce((total, record) => total + record.rows.conflicts, 0) })}\n`) + return records +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + runArchiveCli(process.argv.slice(2)).catch((error: unknown) => { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`) + process.exitCode = 1 + }) +} diff --git a/scripts/biometrics-archive/download.ts b/scripts/biometrics-archive/download.ts new file mode 100644 index 00000000..211758af --- /dev/null +++ b/scripts/biometrics-archive/download.ts @@ -0,0 +1,235 @@ +import { createHash } from 'node:crypto' +import { createWriteStream } from 'node:fs' +import { mkdir } from 'node:fs/promises' +import { join } from 'node:path' +import { Readable } from 'node:stream' +import { pipeline } from 'node:stream/promises' +import type { ReadableStream as WebReadableStream } from 'node:stream/web' +import { z } from 'zod' +import { ANALYSIS_EXPORT_FORMAT, ANALYSIS_EXPORT_PATH } from '@/src/biometrics/analysis-export' +import { parseTarHeader } from '@/src/biometrics/backtest/bundle-tar' + +const DATABASE_ENTRY = 'biometrics.db' +const MANIFEST_ENTRY = 'manifest.json' +const RAW_ENTRY = /^raw\/([A-Za-z0-9_-]+\.RAW\.gz)$/ +const MAX_MANIFEST_BYTES = 16 * 1024 * 1024 + +const sha256 = z.string().regex(/^[0-9a-f]{64}$/) +const size = z.number().int().nonnegative() + +/** manifest.json of analysis-export-v1 (src/biometrics/analysis-export-bundle.ts). */ +export const exportManifestSchema = z.object({ + formatVersion: z.literal(ANALYSIS_EXPORT_FORMAT), + generatedAt: z.string(), + generatedAtMs: size, + build: z.object({ branch: z.string(), commitHash: z.string(), commitTitle: z.string(), buildDate: z.string() }), + pod: z.object({ + id: z.string().regex(/^[0-9a-f]{32}$/).nullable(), + idSource: z.string(), + hostname: z.string().min(1), + model: z.object({ podVersion: z.string(), name: z.string(), hardwareRevision: z.string().nullable() }).nullable(), + }), + schema: z.object({ + biometrics: z.object({ migrationHash: z.string(), migrationCreatedAt: z.number(), migrationTag: z.string().nullable(), appliedMigrations: size }).nullable(), + }), + window: z.object({ + requested: z.object({ startMs: size, endMs: size, start: z.string(), end: z.string() }), + rawSelection: z.object({ includeRaw: z.boolean(), mtimeFromMs: z.number(), mtimeToMs: z.number() }), + served: z.object({ archiveCount: size, rawMtimeMinMs: z.number().nullable(), rawMtimeMaxMs: z.number().nullable() }), + }), + files: z.array(z.discriminatedUnion('role', [ + z.object({ role: z.literal('database'), path: z.literal(DATABASE_ENTRY), bytes: size, sha256 }), + z.object({ role: z.literal('raw'), path: z.string().regex(RAW_ENTRY), name: z.string(), mtimeMs: z.number(), bytes: size, sha256 }), + ])), +}) +export type ExportManifest = z.infer +export type ManifestRawFile = Extract + +export interface ExportWindow { startMs: number, endMs: number, includeRaw: boolean } + +/** A verified export in a staging directory: every file matched the manifest. */ +export interface DownloadedExport { + manifest: ExportManifest + /** manifest.json as served, kept verbatim in the pull record. */ + manifestJson: unknown + databasePath: string + raw: Array +} + +/** A response other than 200; 413 and 429 drive window halving and retry. */ +export class ExportHttpError extends Error { + constructor(readonly status: number, message: string, readonly retryAfterMs: number | null) { + super(message) + } +} + +export class ExportIntegrityError extends Error {} + +export function exportUrl(pod: string, window: ExportWindow) { + return `http://${pod}${ANALYSIS_EXPORT_PATH}?start=${window.startMs}&end=${window.endMs}&raw=${window.includeRaw ? 1 : 0}` +} + +/** Byte-exact reads over a chunked stream. */ +class ByteReader { + private chunks: Buffer[] = [] + private buffered = 0 + private ended = false + total = 0 + constructor(private readonly source: AsyncIterator) {} + + private async fill(bytes: number) { + while (this.buffered < bytes && !this.ended) { + const next = await this.source.next() + if (next.done) this.ended = true + else { + const chunk = Buffer.from(next.value) + this.chunks.push(chunk) + this.buffered += chunk.length + this.total += chunk.length + } + } + } + + private take(bytes: number) { + const first = this.chunks[0] + if (first.length <= bytes) { + this.chunks.shift() + this.buffered -= first.length + return first + } + this.chunks[0] = first.subarray(bytes) + this.buffered -= bytes + return first.subarray(0, bytes) + } + + async read(bytes: number) { + await this.fill(bytes) + if (this.buffered < bytes) throw new ExportIntegrityError('Export stream ended early (truncated tar)') + const parts: Buffer[] = [] + let remaining = bytes + while (remaining > 0) { + const part = this.take(remaining) + parts.push(part) + remaining -= part.length + } + return Buffer.concat(parts, bytes) + } + + async* stream(bytes: number) { + let remaining = bytes + while (remaining > 0) { + if (this.buffered === 0) await this.fill(1) + if (this.buffered === 0) throw new ExportIntegrityError('Export stream ended early (truncated tar)') + const part = this.take(remaining) + remaining -= part.length + yield part + } + } + + async atEnd() { + await this.fill(1) + return this.buffered === 0 + } +} + +async function errorMessage(response: Response) { + const text = await response.text().catch(() => '') + try { + const parsed = JSON.parse(text) as { error?: unknown } + if (typeof parsed.error === 'string') return parsed.error + } + catch { /* not JSON */ } + return text.slice(0, 500) || response.statusText +} + +/** + * Download one export into `staging`, hashing every entry as it streams, then + * check the manifest: the exact file set, sizes, SHA-256s and the requested + * window. Throws before anything leaves `staging`. + */ +export async function downloadExport(fetchImpl: typeof fetch, pod: string, window: ExportWindow, staging: string): Promise { + const response = await fetchImpl(exportUrl(pod, window)).catch((error: unknown) => { + const cause = error instanceof Error && error.cause instanceof Error ? error.cause.message : String(error) + throw new Error(`Could not reach ${pod}: ${cause}`) + }) + if (response.status !== 200) { + const retryAfter = Number(response.headers.get('Retry-After') ?? Number.NaN) + throw new ExportHttpError(response.status, `HTTP ${response.status} from ${pod}: ${await errorMessage(response)}`, + Number.isFinite(retryAfter) && retryAfter >= 0 ? retryAfter * 1000 : null) + } + if (!response.body) throw new ExportIntegrityError('Export response has no body') + await mkdir(join(staging, 'raw'), { recursive: true }) + const reader = new ByteReader(Readable.fromWeb(response.body as WebReadableStream)[Symbol.asyncIterator]()) + const received = new Map() + let manifestText: string | undefined + for (;;) { + const header = parseTarHeader(await reader.read(512)) + if (!header) break + if (manifestText !== undefined) throw new ExportIntegrityError(`Unexpected entry after manifest.json: ${header.name}`) + if (received.has(header.name)) throw new ExportIntegrityError(`Duplicate export entry: ${header.name}`) + const padding = (512 - (header.bytes % 512)) % 512 + if (header.name === MANIFEST_ENTRY) { + if (header.bytes > MAX_MANIFEST_BYTES) throw new ExportIntegrityError('manifest.json is too large') + manifestText = (await reader.read(header.bytes)).toString('utf8') + } + else { + const raw = RAW_ENTRY.exec(header.name) + if (header.name !== DATABASE_ENTRY && !raw) throw new ExportIntegrityError(`Unexpected export entry: ${header.name}`) + const localPath = raw ? join(staging, 'raw', raw[1]) : join(staging, DATABASE_ENTRY) + const digest = createHash('sha256') + await pipeline(Readable.from((async function* () { + for await (const chunk of reader.stream(header.bytes)) { + digest.update(chunk) + yield chunk + } + })()), createWriteStream(localPath, { flags: 'wx', mode: 0o600 })) + received.set(header.name, { bytes: header.bytes, sha256: digest.digest('hex'), localPath }) + } + await reader.read(padding) + } + while (!(await reader.atEnd())) { + if ((await reader.read(512)).some(byte => byte !== 0)) throw new ExportIntegrityError('Export tar has data after its end marker') + } + const declared = Number(response.headers.get('Content-Length')) + if (Number.isFinite(declared) && declared > 0 && reader.total !== declared) { + throw new ExportIntegrityError(`Export was ${reader.total} bytes, Content-Length said ${declared}`) + } + if (manifestText === undefined) throw new ExportIntegrityError('Export has no manifest.json (download cut short?)') + + let manifestJson: unknown + try { + manifestJson = JSON.parse(manifestText) + } + catch { + throw new ExportIntegrityError('manifest.json is not valid JSON') + } + const parsed = exportManifestSchema.safeParse(manifestJson) + if (!parsed.success) throw new ExportIntegrityError(`manifest.json does not match ${ANALYSIS_EXPORT_FORMAT}: ${parsed.error.issues.map(issue => `${issue.path.join('.')}: ${issue.message}`).join('; ')}`) + const manifest = parsed.data + const { requested, rawSelection } = manifest.window + if (requested.startMs !== window.startMs || requested.endMs !== window.endMs || rawSelection.includeRaw !== window.includeRaw) { + throw new ExportIntegrityError(`Pod served window ${requested.startMs}–${requested.endMs} (raw=${rawSelection.includeRaw}), not the one requested`) + } + + const listed = new Set() + for (const file of manifest.files) { + if (listed.has(file.path)) throw new ExportIntegrityError(`manifest.json lists ${file.path} twice`) + listed.add(file.path) + if (file.role === 'raw' && file.path !== `raw/${file.name}`) throw new ExportIntegrityError(`manifest.json raw entry ${file.path} names ${file.name}`) + const got = received.get(file.path) + if (!got) throw new ExportIntegrityError(`Export is missing ${file.path}`) + if (got.bytes !== file.bytes) throw new ExportIntegrityError(`Size mismatch: ${file.path} is ${got.bytes} bytes, manifest says ${file.bytes}`) + if (got.sha256 !== file.sha256) throw new ExportIntegrityError(`SHA-256 mismatch: ${file.path}`) + } + for (const name of received.keys()) if (!listed.has(name)) throw new ExportIntegrityError(`Export entry ${name} is not in manifest.json`) + if (!listed.has(DATABASE_ENTRY)) throw new ExportIntegrityError('Export has no biometrics.db') + + // Every listed file was received (checked above), so the lookups below hit. + const localPath = (path: string) => received.get(path)?.localPath ?? '' + return { + manifest, + manifestJson, + databasePath: localPath(DATABASE_ENTRY), + raw: manifest.files.filter((file): file is ManifestRawFile => file.role === 'raw').map(file => ({ ...file, localPath: localPath(file.path) })), + } +} diff --git a/scripts/biometrics-archive/merge.test.ts b/scripts/biometrics-archive/merge.test.ts new file mode 100644 index 00000000..783b0ed0 --- /dev/null +++ b/scripts/biometrics-archive/merge.test.ts @@ -0,0 +1,118 @@ +// @vitest-environment node +import { randomBytes } from 'node:crypto' +import { cpSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import Database from 'better-sqlite3' +import { drizzle } from 'drizzle-orm/better-sqlite3' +import { migrate } from 'drizzle-orm/better-sqlite3/migrator' +import { afterEach, beforeEach, expect, it } from 'vitest' +import type { ExportManifest } from './download' +import { checkSnapshotSchema, mergeSnapshot, migrationJournal, MIGRATIONS_FOLDER, openArchiveDatabase, TABLES } from './merge' + +let root: string +let archive: Database.Database +const open: Database.Database[] = [] +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'biometrics-archive-merge-')) + archive = openArchiveDatabase(join(root, 'archive.db')) +}) +afterEach(() => { + archive.close() + for (const db of open.splice(0)) db.close() + rmSync(root, { recursive: true, force: true }) +}) + +const journal = migrationJournal() +let snapshots = 0 +function snapshot(fill: (db: Database.Database) => void, migrationsFolder = MIGRATIONS_FOLDER) { + const path = join(root, `snapshot-${++snapshots}.db`) + const db = new Database(path) + migrate(drizzle(db), { migrationsFolder }) + fill(db) + db.close() + return path +} +function manifest(generatedAtMs: number, entry = journal[journal.length - 1]) { + return { + generatedAtMs, generatedAt: new Date(generatedAtMs).toISOString(), + files: [{ role: 'database', path: 'biometrics.db', bytes: 0, sha256: randomBytes(32).toString('hex') }], + schema: { biometrics: { migrationHash: 'h', migrationCreatedAt: entry.when, migrationTag: entry.tag, appliedMigrations: 1 } }, + } as unknown as ExportManifest +} +const merge = (path: string, generatedAtMs: number, entry?: (typeof journal)[number]) => + mergeSnapshot(archive, path, manifest(generatedAtMs, entry), { pullKey: `pod/pulls/${generatedAtMs}.json`, nowMs: generatedAtMs, occupiedArchiveNames: new Set() }) +const rows = (sql: string) => archive.prepare(sql).all() +function vitals(db: Database.Database, side: string, timestamps: number[], heartRate = 60) { + for (const ts of timestamps) { + const id = db.prepare('INSERT INTO vitals (side, timestamp, heart_rate) VALUES (?, ?, ?)').run(side, ts, heartRate).lastInsertRowid + db.prepare('INSERT INTO vitals_quality (vitals_id, side, timestamp, quality_score, created_at) VALUES (?, ?, ?, 0.9, ?)').run(id, side, ts, ts) + } +} + +it('has a merge rule, with a unique key in the archive, for every table the migrations create', () => { + const tables = (rows('SELECT name FROM sqlite_master WHERE type = \'table\' AND name NOT LIKE \'sqlite_%\' AND name NOT LIKE \'archive_%\' AND name != \'__drizzle_migrations\'') as Array<{ name: string }>) + .map(row => row.name).sort() + expect(TABLES.map(spec => spec.table).sort()).toEqual(tables) + const indexed = rows('SELECT name FROM sqlite_master WHERE type = \'index\' AND name LIKE \'archive_uq_%\'') as Array<{ name: string }> + expect(indexed.map(row => row.name).sort()).toEqual(['calibration_runs', 'flash_write_samples', 'health_runs', 'movement', 'prime_events', 'pump_alerts', + 'sleep_records', 'thermal_state', 'vitals_quality', 'water_level_alerts'].map(name => `archive_uq_${name}`)) +}) + +it('gives rows new ids after a pod database reset and keeps vitals_quality pointing at its vitals row', () => { + merge(snapshot(db => vitals(db, 'left', [100, 160, 220])), 1) + const result = merge(snapshot(db => vitals(db, 'left', [10_000, 10_060])), 2) + expect(result.tables.vitals).toMatchObject({ snapshot: 2, inserted: 0, reassigned: 2, conflicts: 0 }) + expect(rows('SELECT v.timestamp, q.timestamp AS qualityTimestamp FROM vitals_quality q JOIN vitals v ON v.id = q.vitals_id ORDER BY v.timestamp')) + .toEqual([100, 160, 220, 10_000, 10_060].map(ts => ({ timestamp: ts, qualityTimestamp: ts }))) +}) + +it('keeps the first copy of an append-only row and records a changed copy once', () => { + merge(snapshot(db => vitals(db, 'left', [100], 60)), 1) + const changed = snapshot(db => vitals(db, 'left', [100], 72)) + expect(merge(changed, 2).tables.vitals).toMatchObject({ inserted: 0, ignored: 1, conflicts: 1 }) + expect(merge(changed, 3).tables.vitals).toMatchObject({ conflicts: 0 }) + expect(rows('SELECT heart_rate AS hr FROM vitals')).toEqual([{ hr: 60 }]) + const [conflict] = rows('SELECT table_name, reason, key, archive_row, snapshot_row FROM archive_conflicts') as Array> + expect(conflict).toMatchObject({ table_name: 'vitals', reason: 'changed', key: '{"side":"left","timestamp":100}' }) + expect(JSON.parse(conflict.archive_row).heart_rate).toBe(60) + expect(JSON.parse(conflict.snapshot_row).heart_rate).toBe(72) +}) + +it('takes the newest snapshot\'s values for state tables, but keeps a released dataset\'s content', () => { + const dataset = (db: Database.Database, snapshotText: string, releasedAtMs: number | null) => db.prepare(`INSERT INTO biometric_datasets + (id, side, subject_id, idempotency_key, request_hash, content_hash, manifest, snapshot, byte_length, created_at_ms, released_at_ms) + VALUES ('ds-1', 'left', 'me', 'k', 'r', 'c', '{}', ?, ?, 1, ?)`).run(snapshotText, snapshotText.length, releasedAtMs) + const health = (db: Database.Database, lastSeenAt: number) => db.prepare('INSERT INTO health_runs (check_id, status, started_at, last_seen_at) VALUES (\'disk\', \'ok\', 10, ?)').run(lastSeenAt) + merge(snapshot((db) => { + dataset(db, '{"nights":[1,2]}', null) + health(db, 20) + }), 1000) + const released = snapshot((db) => { + dataset(db, '{"released":true}', 5000) + health(db, 90) + }) + const older = snapshot(db => health(db, 50)) + expect(merge(released, 2000).tables.health_runs).toMatchObject({ updated: 1 }) + expect(merge(older, 1500).tables.health_runs).toMatchObject({ updated: 0, conflicts: 0 }) + expect(rows('SELECT last_seen_at AS lastSeenAt FROM health_runs')).toEqual([{ lastSeenAt: 90 }]) + expect(rows('SELECT snapshot, byte_length AS bytes, released_at_ms AS releasedAtMs FROM biometric_datasets')) + .toEqual([{ snapshot: '{"nights":[1,2]}', bytes: 16, releasedAtMs: 5000 }]) +}) + +it('merges a snapshot from an older schema and refuses one newer than the checkout', () => { + const folder = join(root, 'migrations') + cpSync(MIGRATIONS_FOLDER, folder, { recursive: true }) + const old = journal[19] + writeFileSync(join(folder, 'meta', '_journal.json'), JSON.stringify({ version: '7', dialect: 'sqlite', entries: journal.slice(0, 20) })) + const path = snapshot(db => vitals(db, 'right', [500, 560]), folder) + expect(checkSnapshotSchema(manifest(1, old).schema.biometrics)).toMatchObject({ known: true }) + const result = merge(path, 1, old) + expect(result.tables.vitals).toMatchObject({ inserted: 2 }) + expect(result.tables.biometric_datasets).toBeUndefined() + expect(rows('SELECT side, processing_run_id AS run FROM vitals')).toEqual([{ side: 'right', run: null }, { side: 'right', run: null }]) + + const latest = journal[journal.length - 1] + expect(() => checkSnapshotSchema({ migrationHash: 'h', migrationCreatedAt: latest.when + 1, migrationTag: '0099_future', appliedMigrations: 99 })) + .toThrow('The pod\'s biometrics schema (0099_future) is newer than this checkout\'s') +}) diff --git a/scripts/biometrics-archive/merge.ts b/scripts/biometrics-archive/merge.ts new file mode 100644 index 00000000..e8557366 --- /dev/null +++ b/scripts/biometrics-archive/merge.ts @@ -0,0 +1,253 @@ +import { readFileSync } from 'node:fs' +import { join } from 'node:path' +import { fileURLToPath } from 'node:url' +import Database from 'better-sqlite3' +import { drizzle } from 'drizzle-orm/better-sqlite3' +import { migrate } from 'drizzle-orm/better-sqlite3/migrator' +import type { ExportManifest, ManifestRawFile } from './download' + +export const MIGRATIONS_FOLDER = fileURLToPath(new URL('../../src/db/biometrics-migrations', import.meta.url)) + +/** + * How a pod table merges into the archive. `key` is the natural key, backed by + * a unique index (the archive adds `archive_uq_
` where the pod has none). + * - append: rows are written once. A key match with other content is recorded + * in archive_conflicts and the archive keeps its row. + * - state: the pod updates rows in place. The newest snapshot's non-key values + * win, except `keep` columns, which keep what the archive saw first. + * Integer `id` surrogates keep the pod's id when it is free; a row whose id is + * taken by a different key (the pod's database was reset) gets a new id. + */ +export interface TableSpec { + table: string + key: string[] + mode: 'append' | 'state' + keep?: string[] + /** SQL for a column's value, over the snapshot row `s`. */ + remap?: Record +} + +export const TABLES: TableSpec[] = [ + { table: 'vitals', key: ['side', 'timestamp'], mode: 'append' }, + // vitals_id follows the vitals row to its archive id. + { table: 'vitals_quality', key: ['side', 'timestamp'], mode: 'append', remap: { + vitals_id: 'COALESCE((SELECT m.id FROM main.vitals m JOIN snap.vitals v ON m.side IS v.side AND m.timestamp IS v.timestamp WHERE v.id = s.vitals_id), s.vitals_id)', + } }, + { table: 'movement', key: ['side', 'timestamp'], mode: 'append' }, + { table: 'sleep_records', key: ['side', 'entered_bed_at', 'left_bed_at'], mode: 'state' }, + { table: 'cap_sense_frames', key: ['side', 'timestamp'], mode: 'append' }, + { table: 'bed_temp', key: ['timestamp'], mode: 'append' }, + { table: 'freezer_temp', key: ['timestamp'], mode: 'append' }, + { table: 'ambient_light', key: ['timestamp'], mode: 'append' }, + { table: 'water_level_readings', key: ['timestamp'], mode: 'append' }, + { table: 'flow_readings', key: ['timestamp'], mode: 'append' }, + { table: 'thermal_state', key: ['side', 'timestamp', 'is_powered'], mode: 'append' }, + { table: 'prime_events', key: ['timestamp'], mode: 'append' }, + { table: 'flash_write_samples', key: ['boot_id', 'timestamp'], mode: 'append' }, + { table: 'water_level_alerts', key: ['type', 'started_at'], mode: 'state' }, + { table: 'pump_alerts', key: ['timestamp', 'type', 'side'], mode: 'state' }, + { table: 'health_runs', key: ['check_id', 'started_at'], mode: 'state' }, + { table: 'calibration_profiles', key: ['side', 'sensor_type'], mode: 'state' }, + { table: 'calibration_runs', key: ['side', 'sensor_type', 'created_at'], mode: 'append' }, + { table: 'reference_imports', key: ['id'], mode: 'state' }, + { table: 'reference_measurements', key: ['import_id', 'record_key'], mode: 'append' }, + { table: 'reference_stages', key: ['side', 'start', 'source'], mode: 'state' }, + { table: 'reference_vitals', key: ['side', 'timestamp', 'source'], mode: 'state' }, + { table: 'biometric_processing_runs', key: ['id'], mode: 'state' }, + { table: 'biometric_runtime_configs', key: ['side', 'generation'], mode: 'append' }, + { table: 'biometric_runtime_receipts', key: ['side', 'component', 'generation'], mode: 'state' }, + // Releasing a dataset on the pod replaces its snapshot with a tombstone; the archive keeps the content. + { table: 'biometric_datasets', key: ['id'], mode: 'state', keep: ['snapshot', 'byte_length'] }, + { table: 'biometric_experiments', key: ['id'], mode: 'state' }, + { table: 'biometric_backtest_jobs', key: ['id'], mode: 'state' }, + { table: 'biometric_backtest_results', key: ['job_id'], mode: 'state' }, + { table: 'biometric_holdout_accesses', key: ['id'], mode: 'append' }, + { table: 'biometric_raw_artifacts', key: ['content_hash'], mode: 'state' }, + { table: 'biometric_dataset_raw_artifacts', key: ['id'], mode: 'state' }, + { table: 'biometric_raw_sources', key: ['id'], mode: 'state' }, + { table: 'biometric_shadow_runs', key: ['id'], mode: 'state' }, + { table: 'biometric_shadow_samples', key: ['run_id', 'sequence'], mode: 'append' }, +] + +const INTERNAL_TABLE = /^(sqlite_|__drizzle_migrations$|archive_)/ + +const ARCHIVE_TABLES = ` +CREATE TABLE IF NOT EXISTS archive_snapshots ( + sha256 TEXT PRIMARY KEY NOT NULL, generated_at_ms INTEGER NOT NULL, migration_tag TEXT, + migration_created_at INTEGER, pull_key TEXT NOT NULL, merged_at_ms INTEGER NOT NULL); +CREATE TABLE IF NOT EXISTS archive_raw_files ( + sha256 TEXT PRIMARY KEY NOT NULL, name TEXT NOT NULL, archive_name TEXT NOT NULL UNIQUE, + mtime_ms INTEGER NOT NULL, bytes INTEGER NOT NULL, pull_key TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS archive_conflicts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, table_name TEXT NOT NULL, reason TEXT NOT NULL, key TEXT NOT NULL, + archive_row TEXT, snapshot_row TEXT NOT NULL, pull_key TEXT NOT NULL, UNIQUE (table_name, key, snapshot_row));` + +export interface TableCounts { snapshot: number, inserted: number, reassigned: number, ignored: number, updated: number, conflicts: number } +export interface ArchiveName { sha256: string, name: string, archiveName: string, renamed: boolean } +export interface MergeResult { tables: Record, archiveNames: ArchiveName[], warnings: string[] } + +const quote = (name: string) => `"${name.replaceAll('"', '""')}"` + +interface Column { name: string, type: string, pk: number } + +function columns(db: Database.Database, schema: 'main' | 'snap', table: string) { + return db.prepare('SELECT name, type, pk FROM pragma_table_info(?, ?)').all(table, schema) as Column[] +} + +function tables(db: Database.Database, schema: 'main' | 'snap') { + return (db.prepare(`SELECT name FROM ${schema}.sqlite_master WHERE type = 'table'`).all() as Array<{ name: string }>) + .map(row => row.name).filter(name => !INTERNAL_TABLE.test(name)) +} + +function hasUniqueKey(db: Database.Database, table: string, key: string[]) { + const pk = columns(db, 'main', table).filter(column => column.pk > 0).map(column => column.name) + if (pk.length === key.length && pk.every(name => key.includes(name))) return true + const indexes = db.prepare('SELECT name, "unique" AS isUnique, partial FROM pragma_index_list(?)').all(table) as Array<{ name: string, isUnique: number, partial: number }> + return indexes.some((index) => { + if (!index.isUnique || index.partial) return false + const indexed = (db.prepare('SELECT name FROM pragma_index_info(?)').all(index.name) as Array<{ name: string }>).map(row => row.name) + return indexed.length === key.length && indexed.every(name => key.includes(name)) + }) +} + +interface JournalEntry { when: number, tag: string } +export function migrationJournal(folder = MIGRATIONS_FOLDER): JournalEntry[] { + return (JSON.parse(readFileSync(join(folder, 'meta', '_journal.json'), 'utf8')) as { entries: JournalEntry[] }).entries +} + +/** + * The archive database is migrated to this checkout's newest migration, which + * must be at least the snapshot's. An older snapshot still merges: columns it + * lacks take their defaults. + */ +export function checkSnapshotSchema(schema: ExportManifest['schema']['biometrics'], journal = migrationJournal()) { + if (!schema) throw new Error('The snapshot has no applied biometrics migrations; refusing to merge it') + const latest = journal[journal.length - 1] + if (schema.migrationCreatedAt > latest.when) { + throw new Error(`The pod's biometrics schema (${schema.migrationTag ?? schema.migrationCreatedAt}) is newer than this checkout's (${latest.tag}). Update the checkout, then pull again.`) + } + return { known: journal.some(entry => entry.when === schema.migrationCreatedAt), checkoutTag: latest.tag } +} + +/** Open (or create) the archive database: migrate it, then add the merge's unique keys and bookkeeping tables. */ +export function openArchiveDatabase(path: string, migrationsFolder = MIGRATIONS_FOLDER) { + const db = new Database(path) + // A single file, so the archive reads from read-only and network mounts without -shm. + db.pragma('journal_mode = DELETE') + db.pragma('busy_timeout = 5000') + migrate(drizzle(db), { migrationsFolder }) + // Snapshots are internally consistent and the archive never deletes, so + // table order can't break references; enforcement would only block merges. + db.pragma('foreign_keys = OFF') + db.exec(ARCHIVE_TABLES) + const existing = new Set(tables(db, 'main')) + for (const spec of TABLES) { + if (!existing.has(spec.table) || hasUniqueKey(db, spec.table, spec.key)) continue + db.exec(`CREATE UNIQUE INDEX IF NOT EXISTS ${quote(`archive_uq_${spec.table}`)} ON ${quote(spec.table)} (${spec.key.map(quote).join(', ')})`) + } + return db +} + +function mergeTable(db: Database.Database, spec: TableSpec, newest: boolean, pullKey: string): TableCounts { + const table = quote(spec.table) + const archiveColumns = columns(db, 'main', spec.table) + const snapshotNames = new Set(columns(db, 'snap', spec.table).map(column => column.name)) + const common = archiveColumns.map(column => column.name).filter(name => snapshotNames.has(name)) + const missingKey = spec.key.filter(name => !common.includes(name)) + if (missingKey.length) throw new Error(`${spec.table}: key column(s) ${missingKey.join(', ')} missing from the snapshot or archive`) + const surrogate = !spec.key.includes('id') && archiveColumns.some(column => column.name === 'id' && column.pk === 1 && column.type.toUpperCase() === 'INTEGER') + const value = (name: string) => spec.remap?.[name] ?? `s.${quote(name)}` + const keyMatch = spec.key.map(name => `m.${quote(name)} IS s.${quote(name)}`).join(' AND ') + const json = (alias: string, names: string[]) => `json_object(${names.map(name => `'${name}', ${alias}.${quote(name)}`).join(', ')})` + const counts: TableCounts = { snapshot: 0, inserted: 0, reassigned: 0, ignored: 0, updated: 0, conflicts: 0 } + counts.snapshot = (db.prepare(`SELECT COUNT(*) AS n FROM snap.${table}`).get() as { n: number }).n + + const insert = (names: string[], where = '') => + db.prepare(`INSERT OR IGNORE INTO main.${table} (${names.map(quote).join(', ')}) SELECT ${names.map(value).join(', ')} FROM snap.${table} s ${where}`).run().changes + counts.inserted = insert(common) + if (surrogate) counts.reassigned = insert(common.filter(name => name !== 'id'), `WHERE NOT EXISTS (SELECT 1 FROM main.${table} m WHERE ${keyMatch})`) + + const recordConflicts = (reason: string, archiveRow: string, from: string, where: string) => + db.prepare(`INSERT OR IGNORE INTO archive_conflicts (table_name, reason, key, archive_row, snapshot_row, pull_key) + SELECT ?, ?, ${json('s', spec.key)}, ${archiveRow}, ${json('s', common)}, ? FROM ${from} WHERE ${where}`).run(spec.table, reason, pullKey).changes + // A row the archive's constraints refused (for example a second active reference + // import with the same content after a pod reset) is kept aside, not lost. + counts.conflicts += recordConflicts('rejected', 'NULL', `snap.${table} s`, `NOT EXISTS (SELECT 1 FROM main.${table} m WHERE ${keyMatch})`) + + const compared = common.filter(name => !spec.key.includes(name) && !(surrogate && name === 'id') && !spec.remap?.[name] && !spec.keep?.includes(name)) + if (compared.length) { + const differs = `NOT (${compared.map(name => `m.${quote(name)} IS s.${quote(name)}`).join(' AND ')})` + if (spec.mode === 'append') { + counts.conflicts += recordConflicts('changed', json('m', common), `snap.${table} s JOIN main.${table} m ON ${keyMatch}`, differs) + } + else if (newest) { + counts.updated = db.prepare(`UPDATE main.${table} AS m SET ${compared.map(name => `${quote(name)} = s.${quote(name)}`).join(', ')} + FROM snap.${table} AS s WHERE ${keyMatch} AND ${differs}`).run().changes + } + } + counts.ignored = counts.snapshot - counts.inserted - counts.reassigned + return counts +} + +function assignArchiveNames(db: Database.Database, raw: readonly ManifestRawFile[], pullKey: string, occupied: ReadonlySet): ArchiveName[] { + const known = db.prepare('SELECT archive_name AS archiveName FROM archive_raw_files WHERE sha256 = ?') + const taken = db.prepare('SELECT 1 FROM archive_raw_files WHERE archive_name = ?') + const add = db.prepare('INSERT INTO archive_raw_files (sha256, name, archive_name, mtime_ms, bytes, pull_key) VALUES (?, ?, ?, ?, ?, ?)') + return raw.map((file) => { + const existing = known.get(file.sha256) as { archiveName: string } | undefined + if (existing) return { sha256: file.sha256, name: file.name, archiveName: existing.archiveName, renamed: existing.archiveName !== file.name } + // The pod reused a name for different bytes (counter reset, reinstall), or a + // file the archive didn't write sits there: never overwrite. + const renamed = Boolean(taken.get(file.name)) || occupied.has(file.name) + const archiveName = renamed ? file.name.replace(/\.RAW\.gz$/, `-${file.sha256.slice(0, 12)}.RAW.gz`) : file.name + add.run(file.sha256, file.name, archiveName, Math.floor(file.mtimeMs), file.bytes, pullKey) + return { sha256: file.sha256, name: file.name, archiveName, renamed } + }) +} + +/** + * Merge one verified snapshot into the archive database in a single + * transaction, together with the RAW name index and the snapshot record. + * Nothing is ever deleted: rows the pod has since pruned stay. + */ +export function mergeSnapshot(db: Database.Database, snapshotPath: string, manifest: ExportManifest, + options: { pullKey: string, nowMs: number, occupiedArchiveNames: ReadonlySet }): MergeResult { + const { pullKey, nowMs } = options + db.prepare('ATTACH DATABASE ? AS snap').run(snapshotPath) + try { + return db.transaction(() => { + const warnings: string[] = [] + const latest = (db.prepare('SELECT MAX(generated_at_ms) AS ms FROM archive_snapshots').get() as { ms: number | null }).ms + const newest = latest === null || manifest.generatedAtMs >= latest + if (!newest) warnings.push(`Snapshot from ${manifest.generatedAt} is older than one already merged; state tables keep their newer values`) + const archiveTables = new Set(tables(db, 'main')) + const snapshotTables = tables(db, 'snap') + const classified = new Set(TABLES.map(spec => spec.table)) + const specs = [...TABLES.filter(spec => snapshotTables.includes(spec.table))] + for (const name of snapshotTables.filter(table => !classified.has(table)).sort()) { + if (!archiveTables.has(name)) { + warnings.push(`Table ${name} is not in this checkout's schema; skipped`) + continue + } + const pk = columns(db, 'main', name).filter(column => column.pk > 0).sort((a, b) => a.pk - b.pk).map(column => column.name) + if (!pk.length) { + warnings.push(`Table ${name} has no merge rule or primary key; skipped`) + continue + } + warnings.push(`Table ${name} has no merge rule; merged on its primary key (${pk.join(', ')})`) + specs.push({ table: name, key: pk, mode: 'append' }) + } + const tableCounts: Record = {} + for (const spec of specs) tableCounts[spec.table] = mergeTable(db, spec, newest, pullKey) + const archiveNames = assignArchiveNames(db, manifest.files.filter((file): file is ManifestRawFile => file.role === 'raw'), pullKey, options.occupiedArchiveNames) + const database = manifest.files.find(file => file.role === 'database') + const schema = manifest.schema.biometrics + db.prepare('INSERT OR IGNORE INTO archive_snapshots (sha256, generated_at_ms, migration_tag, migration_created_at, pull_key, merged_at_ms) VALUES (?, ?, ?, ?, ?, ?)') + .run(database?.sha256 ?? '', manifest.generatedAtMs, schema?.migrationTag ?? null, schema?.migrationCreatedAt ?? null, pullKey, nowMs) + return { tables: tableCounts, archiveNames, warnings } + })() + } + finally { + db.prepare('DETACH DATABASE snap').run() + } +} diff --git a/scripts/biometrics-archive/pull.ts b/scripts/biometrics-archive/pull.ts new file mode 100644 index 00000000..3089ae85 --- /dev/null +++ b/scripts/biometrics-archive/pull.ts @@ -0,0 +1,280 @@ +import { randomBytes } from 'node:crypto' +import { createReadStream } from 'node:fs' +import { mkdir, readdir, readFile, rm, writeFile } from 'node:fs/promises' +import { hostname } from 'node:os' +import { join } from 'node:path' +import { z } from 'zod' +import { MAX_ANALYSIS_WINDOW_MS } from '@/src/biometrics/analysis-export' +import { downloadExport, ExportHttpError, type DownloadedExport, type ExportWindow } from './download' +import { checkSnapshotSchema, mergeSnapshot, openArchiveDatabase, type TableCounts } from './merge' +import { LocalStore, putBuffer, readText, type Store } from './store' + +export const PULL_RECORD_FORMAT = 'biometrics-archive-pull-v1' +export const LOCK_FILE = '.biometrics-archive.lock' +const STAGING_DIR = '.staging' +const HOUR_MS = 60 * 60 * 1000 + +export interface PullOptions { + /** host[:port]; port defaults to 3000. */ + pod: string + dest: string + sinceMs?: number + untilMs?: number + includeRaw: boolean + fetch?: typeof fetch + sleep?: (ms: number) => Promise + now?: () => number + log?: (line: string) => void + /** 429 handling: attempts per window and exponential backoff bounds. */ + retry?: { attempts: number, baseDelayMs: number, maxDelayMs: number } + /** 413 halves the window down to this length, then fails. */ + minWindowMs?: number +} + +export interface PullRecord { + formatVersion: typeof PULL_RECORD_FORMAT + pulledAt: string + podHost: string + podDir: string + podId: string | null + window: ExportWindow + /** The served (requested) window end; the next default pull starts here. */ + cursor: { endMs: number, end: string } + raw: { new: number, skipped: number, linked: number, copied: number, renamed: Array<{ name: string, archiveName: string, sha256: string }> } + tables: Record + rows: { inserted: number, reassigned: number, ignored: number, updated: number, conflicts: number } + schema: { snapshotTag: string | null, knownToCheckout: boolean, checkoutTag: string } + warnings: string[] + manifest: unknown +} + +export class ArchiveLockedError extends Error {} + +const pullRecordCursorSchema = z.object({ + formatVersion: z.literal(PULL_RECORD_FORMAT), + podHost: z.string(), + cursor: z.object({ endMs: z.number().int().nonnegative() }), +}) + +export function normalizePod(pod: string) { + return /:\d+$/.test(pod) ? pod : `${pod}:3000` +} + +/** Directory name for a pod: its stable id, else its hostname made path-safe. */ +export function podDirectory(pod: DownloadedExport['manifest']['pod']) { + return pod.id ?? (pod.hostname.replace(/[^A-Za-z0-9._-]/g, '_').replace(/^\.+/, '') || 'unknown-pod') +} + +function isAlive(pid: number) { + try { + process.kill(pid, 0) + return true + } + catch (error) { + return (error as NodeJS.ErrnoException).code === 'EPERM' + } +} + +/** One pull per destination. A lock left by a dead process on this host is taken over. */ +export async function acquireLock(dest: string) { + const path = join(dest, LOCK_FILE) + const body = JSON.stringify({ pid: process.pid, host: hostname(), startedAt: new Date().toISOString() }) + for (let attempt = 0; ; attempt++) { + try { + await writeFile(path, body, { flag: 'wx', mode: 0o600 }) + return async () => { + await rm(path, { force: true }) + } + } + catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error + let holder: { pid?: unknown, host?: unknown, startedAt?: unknown } = {} + try { + holder = JSON.parse(await readFile(path, 'utf8')) as typeof holder + } + catch { /* unreadable: treat as held */ } + const stale = holder.host === hostname() && typeof holder.pid === 'number' && !isAlive(holder.pid) + if (!stale || attempt > 0) { + throw new ArchiveLockedError(`Another pull into ${dest} is running (pid ${String(holder.pid)} on ${String(holder.host)} since ${String(holder.startedAt)}). If none is, delete ${path}.`) + } + await rm(path, { force: true }) + } + } +} + +/** + * The next default start: the largest served end among this pod's successful + * pulls, so an explicit backfill of an older range never moves it back. Pods + * are matched by --pod; if none match and the destination holds exactly one + * pod with pulls, that pod's cursor is used. + */ +export async function findCursor(store: Store, dest: string, podHost: string, log: (line: string) => void) { + const directories = (await readdir(dest, { withFileTypes: true }).catch(() => [])) + .filter(entry => entry.isDirectory() && !entry.name.startsWith('.')).map(entry => entry.name) + const perPod = new Map() + for (const directory of directories) { + const cursors = { matched: null as number | null, any: null as number | null } + for (const key of await store.list(`${directory}/pulls/`)) { + if (!key.endsWith('.json')) continue + const parsed = pullRecordCursorSchema.safeParse(await readText(store, key).then(text => JSON.parse(text) as unknown).catch(() => null)) + if (!parsed.success) continue + const end = parsed.data.cursor.endMs + cursors.any = Math.max(cursors.any ?? end, end) + if (parsed.data.podHost === podHost) cursors.matched = Math.max(cursors.matched ?? end, end) + } + if (cursors.any !== null) perPod.set(directory, cursors) + } + const matched = [...perPod.values()].map(cursor => cursor.matched).filter(end => end !== null) + if (matched.length) return Math.max(...matched) + if (perPod.size === 1) { + const [[directory, cursor]] = [...perPod] + log(`No earlier pull from ${podHost}; continuing from the only pod in ${dest} (${directory})`) + return cursor.any + } + return null +} + +export function planWindows(sinceMs: number, untilMs: number) { + const windows: Array<{ startMs: number, endMs: number }> = [] + for (let start = sinceMs; start < untilMs; start += MAX_ANALYSIS_WINDOW_MS) windows.push({ startMs: start, endMs: Math.min(start + MAX_ANALYSIS_WINDOW_MS, untilMs) }) + return windows +} + +const sum = (counts: Record, field: keyof TableCounts) => Object.values(counts).reduce((total, row) => total + row[field], 0) + +/** Store one verified export: RAW blobs, the merged database, archive/ links, then the pull record (the cursor). */ +async function commit(store: Store, dest: string, podHost: string, download: DownloadedExport, window: ExportWindow, nowMs: number, log: (line: string) => void): Promise { + const { manifest } = download + const warnings: string[] = [] + const podDir = podDirectory(manifest.pod) + if (!manifest.pod.id) { + warnings.push(`Pod id is null (no hardware seed); storing under hostname ${podDir}`) + log(`warning: ${warnings[0]}`) + } + const schema = checkSnapshotSchema(manifest.schema.biometrics) + if (!schema.known) warnings.push(`Snapshot migration ${manifest.schema.biometrics?.migrationTag ?? manifest.schema.biometrics?.migrationCreatedAt} is not in this checkout's journal; merged on common columns`) + + const stamp = new Date(nowMs).toISOString().replace(/:/g, '-') + let pullKey = `${podDir}/pulls/${stamp}.json` + for (let suffix = 1; await store.exists(pullKey); suffix++) pullKey = `${podDir}/pulls/${stamp}-${suffix}.json` + + const raw: PullRecord['raw'] = { new: 0, skipped: 0, linked: 0, copied: 0, renamed: [] } + for (const file of download.raw) { + const key = `${podDir}/raw/${file.sha256}.RAW.gz` + if (await store.exists(key)) raw.skipped++ + else { + await store.put(key, createReadStream(file.localPath), { sha256: file.sha256, size: file.bytes, mtimeMs: file.mtimeMs }) + raw.new++ + } + } + + await mkdir(join(dest, podDir), { recursive: true }) + const occupied = new Set() + for (const file of download.raw) if (await store.exists(`${podDir}/archive/${file.name}`)) occupied.add(file.name) + const db = openArchiveDatabase(join(dest, podDir, 'biometrics.db')) + let merged + try { + merged = mergeSnapshot(db, download.databasePath, manifest, { pullKey, nowMs, occupiedArchiveNames: occupied }) + } + finally { + db.close() + } + warnings.push(...merged.warnings) + + for (const name of merged.archiveNames) { + const archiveKey = `${podDir}/archive/${name.archiveName}` + if (name.renamed) raw.renamed.push({ name: name.name, archiveName: name.archiveName, sha256: name.sha256 }) + if (await store.exists(archiveKey)) continue + raw[await store.link(`${podDir}/raw/${name.sha256}.RAW.gz`, archiveKey) === 'linked' ? 'linked' : 'copied']++ + } + + const record: PullRecord = { + formatVersion: PULL_RECORD_FORMAT, + pulledAt: new Date(nowMs).toISOString(), + podHost, + podDir, + podId: manifest.pod.id, + window, + cursor: { endMs: manifest.window.requested.endMs, end: manifest.window.requested.end }, + raw, + tables: merged.tables, + rows: { inserted: sum(merged.tables, 'inserted'), reassigned: sum(merged.tables, 'reassigned'), ignored: sum(merged.tables, 'ignored'), + updated: sum(merged.tables, 'updated'), conflicts: sum(merged.tables, 'conflicts') }, + schema: { snapshotTag: manifest.schema.biometrics?.migrationTag ?? null, knownToCheckout: schema.known, checkoutTag: schema.checkoutTag }, + warnings, + manifest: download.manifestJson, + } + await putBuffer(store, pullKey, Buffer.from(`${JSON.stringify(record, null, 2)}\n`)) + return record +} + +/** + * Pull [since, until] from the pod in ≤14-day windows, oldest first. Each + * window is downloaded and verified in staging before anything is stored; a + * failed window leaves the store and the cursor as they were. + */ +export async function pull(options: PullOptions): Promise { + const fetchImpl = options.fetch ?? fetch + const sleep = options.sleep ?? (ms => new Promise(resolve => setTimeout(resolve, ms))) + const now = options.now ?? Date.now + const log = options.log ?? (() => {}) + const retry = options.retry ?? { attempts: 6, baseDelayMs: 30_000, maxDelayMs: 5 * 60_000 } + const minWindowMs = options.minWindowMs ?? HOUR_MS + const podHost = normalizePod(options.pod) + + await mkdir(options.dest, { recursive: true }) + const release = await acquireLock(options.dest) + const records: PullRecord[] = [] + try { + await rm(join(options.dest, STAGING_DIR), { recursive: true, force: true }) + const store = new LocalStore(options.dest) + const untilMs = options.untilMs ?? now() + const sinceMs = options.sinceMs ?? (await findCursor(store, options.dest, podHost, log)) ?? untilMs - MAX_ANALYSIS_WINDOW_MS + if (sinceMs >= untilMs) { + log(`Nothing to pull: ${new Date(sinceMs).toISOString()} is not before ${new Date(untilMs).toISOString()}`) + return records + } + const pending = planWindows(sinceMs, untilMs).reverse() + for (let span = pending.pop(); span; span = pending.pop()) { + const window: ExportWindow = { ...span, includeRaw: options.includeRaw } + const label = `${new Date(span.startMs).toISOString()} → ${new Date(span.endMs).toISOString()}` + const staging = join(options.dest, STAGING_DIR, `${process.pid}-${randomBytes(4).toString('hex')}`) + try { + let download: DownloadedExport | undefined + for (let attempt = 1; !download; attempt++) { + try { + download = await downloadExport(fetchImpl, podHost, window, staging) + } + catch (error) { + if (!(error instanceof ExportHttpError) || error.status !== 429 || attempt >= retry.attempts) throw error + const delay = Math.min(retry.maxDelayMs, error.retryAfterMs ?? retry.baseDelayMs * 2 ** (attempt - 1)) + log(`${label}: pod busy with another export (429), retrying in ${Math.round(delay / 1000)} s`) + await rm(staging, { recursive: true, force: true }) + await sleep(delay) + } + } + const record = await commit(store, options.dest, podHost, download, window, now(), log) + records.push(record) + log(`${label}: ${record.raw.new} new RAW (${record.raw.skipped} already stored), ${record.rows.inserted + record.rows.reassigned} rows added, ` + + `${record.rows.updated} updated, ${record.rows.conflicts} conflicts → ${record.podDir}/pulls/`) + } + catch (error) { + if (!(error instanceof ExportHttpError) || error.status !== 413) throw error + const half = Math.floor((span.endMs - span.startMs) / 2) + if (half < minWindowMs) { + throw new Error(`${error.message}. The window ${label} cannot be halved below ${minWindowMs / HOUR_MS} h; pull it with --no-raw, or fetch its RAW archives another way.`) + } + log(`${label}: RAW over the export limit (413), halving the window`) + pending.push({ startMs: span.startMs + half, endMs: span.endMs }, { startMs: span.startMs, endMs: span.startMs + half }) + } + finally { + await rm(staging, { recursive: true, force: true }) + } + } + return records + } + finally { + await rm(join(options.dest, STAGING_DIR), { recursive: true, force: true }) + await release() + } +} diff --git a/scripts/biometrics-archive/store.ts b/scripts/biometrics-archive/store.ts new file mode 100644 index 00000000..5e654efc --- /dev/null +++ b/scripts/biometrics-archive/store.ts @@ -0,0 +1,119 @@ +import { createHash, randomBytes } from 'node:crypto' +import { constants, createReadStream, createWriteStream } from 'node:fs' +import { copyFile, link, mkdir, readdir, rename, rm, stat, utimes } from 'node:fs/promises' +import { dirname, join, resolve, sep } from 'node:path' +import { Readable } from 'node:stream' +import { pipeline } from 'node:stream/promises' + +/** `mtimeMs` is the source file's modification time, kept where the backend can. */ +export interface ObjectMeta { sha256: string, size: number, mtimeMs?: number } + +/** + * Where archived objects live. Keys are relative POSIX paths such as + * `/raw/.RAW.gz`. `put` replaces atomically and rejects bytes + * that don't match `meta`; write-once is the caller's job (`exists` → skip). + * `link` makes `to` name the same bytes as `from` and fails if `to` exists; + * a backend without links copies. + */ +export interface Store { + put(key: string, body: Readable, meta: ObjectMeta): Promise + get(key: string): Promise + exists(key: string): Promise + /** Every key under `prefix`, recursively, sorted. */ + list(prefix: string): Promise + link(from: string, to: string): Promise<'linked' | 'copied'> +} + +export class StoreIntegrityError extends Error {} + +const KEY = /^[A-Za-z0-9._-]+(\/[A-Za-z0-9._-]+)*$/ + +export class LocalStore implements Store { + readonly root: string + constructor(root: string) { + this.root = resolve(root) + } + + private path(key: string) { + if (!KEY.test(key) || key.split('/').some(part => part === '.' || part === '..')) throw new Error(`Invalid store key: ${key}`) + return join(this.root, ...key.split('/')) + } + + async put(key: string, body: Readable, meta: ObjectMeta) { + const target = this.path(key) + await mkdir(dirname(target), { recursive: true }) + const temporary = `${target}.tmp-${process.pid}-${randomBytes(4).toString('hex')}` + const digest = createHash('sha256') + let size = 0 + try { + await pipeline(body, async function* (source: AsyncIterable) { + for await (const chunk of source) { + digest.update(chunk) + size += chunk.length + yield chunk + } + }, createWriteStream(temporary, { flags: 'wx', mode: 0o600 })) + const sha256 = digest.digest('hex') + if (size !== meta.size || sha256 !== meta.sha256) throw new StoreIntegrityError(`${key}: expected ${meta.size} bytes with SHA-256 ${meta.sha256}, got ${size} bytes with ${sha256}`) + if (meta.mtimeMs !== undefined) await utimes(temporary, new Date(), new Date(meta.mtimeMs)) + await rename(temporary, target) + } + finally { + await rm(temporary, { force: true }) + } + } + + async get(key: string) { + const path = this.path(key) + await stat(path) + return createReadStream(path) + } + + async exists(key: string) { + return stat(this.path(key)).then(info => info.isFile(), () => false) + } + + async list(prefix: string) { + const trimmed = prefix.replace(/\/+$/, '') + const start = trimmed ? this.path(trimmed) : this.root + const keys: string[] = [] + const walk = async (directory: string): Promise => { + const entries = await readdir(directory, { withFileTypes: true }).catch((error: NodeJS.ErrnoException) => { + if (error.code === 'ENOENT' || error.code === 'ENOTDIR') return [] + throw error + }) + for (const entry of entries) { + const path = join(directory, entry.name) + if (entry.isDirectory()) await walk(path) + else if (entry.isFile() && !entry.name.includes('.tmp-')) keys.push(path.slice(this.root.length + 1).split(sep).join('/')) + } + } + await walk(start) + return keys.filter(key => key.startsWith(prefix)).sort() + } + + async link(from: string, to: string) { + const source = this.path(from), target = this.path(to) + await mkdir(dirname(target), { recursive: true }) + try { + await link(source, target) + return 'linked' as const + } + catch (error) { + // Filesystems without hard links (some NAS/SMB mounts) get a copy. + if (!['EPERM', 'ENOTSUP', 'EXDEV', 'ENOSYS'].includes((error as NodeJS.ErrnoException).code ?? '')) throw error + await copyFile(source, target, constants.COPYFILE_EXCL) + return 'copied' as const + } + } +} + +export async function putBuffer(store: Store, key: string, bytes: Buffer) { + await store.put(key, Readable.from([bytes]), { sha256: createHash('sha256').update(bytes).digest('hex'), size: bytes.length }) +} + +export async function readText(store: Store, key: string) { + const chunks: Buffer[] = [] + for await (const chunk of await store.get(key)) chunks.push(chunk as Buffer) + return Buffer.concat(chunks).toString('utf8') +} From 718e08b41f3ee678ec9cf25ed5b0d8d1895c3c70 Mon Sep 17 00:00:00 2001 From: Jonathan Ng Date: Sun, 18 Oct 2026 11:27:40 -0700 Subject: [PATCH 2/2] fix(biometrics): keep database-only pulls from advancing the archiver's RAW cursor --- scripts/biometrics-archive/archive.test.ts | 9 +++++++++ scripts/biometrics-archive/pull.ts | 12 ++++++++---- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/scripts/biometrics-archive/archive.test.ts b/scripts/biometrics-archive/archive.test.ts index a5f657f4..383b3e54 100644 --- a/scripts/biometrics-archive/archive.test.ts +++ b/scripts/biometrics-archive/archive.test.ts @@ -184,6 +184,15 @@ describe('biometrics archive pull', () => { expect(await findCursor(new LocalStore(dest), dest, 'pod.local:3000', () => {})).toBe(NOW - DAY) }) + it('does not let a database-only pull advance the cursor of RAW pulls', async () => { + vitals(NOW - 3 * DAY, 2) + await pull(options({ now: () => NOW - 2 * DAY })) + await pull(options({ now: () => NOW - DAY, includeRaw: false })) + const store = new LocalStore(dest) + expect(await findCursor(store, dest, 'pod.local:3000', () => {}, true)).toBe(NOW - 2 * DAY) + expect(await findCursor(store, dest, 'pod.local:3000', () => {}, false)).toBe(NOW - DAY) + }) + it('retries 429 with backoff, honouring Retry-After', async () => { vitals(NOW - DAY, 1) let busy = 2 diff --git a/scripts/biometrics-archive/pull.ts b/scripts/biometrics-archive/pull.ts index 3089ae85..ec8e9a36 100644 --- a/scripts/biometrics-archive/pull.ts +++ b/scripts/biometrics-archive/pull.ts @@ -54,6 +54,8 @@ const pullRecordCursorSchema = z.object({ formatVersion: z.literal(PULL_RECORD_FORMAT), podHost: z.string(), cursor: z.object({ endMs: z.number().int().nonnegative() }), + // Records written before the mode was stored always included RAW. + window: z.object({ includeRaw: z.boolean().default(true) }).default({ includeRaw: true }), }) export function normalizePod(pod: string) { @@ -106,9 +108,11 @@ export async function acquireLock(dest: string) { * The next default start: the largest served end among this pod's successful * pulls, so an explicit backfill of an older range never moves it back. Pods * are matched by --pod; if none match and the destination holds exactly one - * pod with pulls, that pod's cursor is used. + * pod with pulls, that pod's cursor is used. A pull that includes RAW only + * counts earlier RAW pulls, so a --no-raw pull never skips archives it did not + * fetch; a database-only pull counts every record. */ -export async function findCursor(store: Store, dest: string, podHost: string, log: (line: string) => void) { +export async function findCursor(store: Store, dest: string, podHost: string, log: (line: string) => void, includeRaw = true) { const directories = (await readdir(dest, { withFileTypes: true }).catch(() => [])) .filter(entry => entry.isDirectory() && !entry.name.startsWith('.')).map(entry => entry.name) const perPod = new Map() @@ -117,7 +121,7 @@ export async function findCursor(store: Store, dest: string, podHost: string, lo for (const key of await store.list(`${directory}/pulls/`)) { if (!key.endsWith('.json')) continue const parsed = pullRecordCursorSchema.safeParse(await readText(store, key).then(text => JSON.parse(text) as unknown).catch(() => null)) - if (!parsed.success) continue + if (!parsed.success || (includeRaw && !parsed.data.window.includeRaw)) continue const end = parsed.data.cursor.endMs cursors.any = Math.max(cursors.any ?? end, end) if (parsed.data.podHost === podHost) cursors.matched = Math.max(cursors.matched ?? end, end) @@ -229,7 +233,7 @@ export async function pull(options: PullOptions): Promise { await rm(join(options.dest, STAGING_DIR), { recursive: true, force: true }) const store = new LocalStore(options.dest) const untilMs = options.untilMs ?? now() - const sinceMs = options.sinceMs ?? (await findCursor(store, options.dest, podHost, log)) ?? untilMs - MAX_ANALYSIS_WINDOW_MS + const sinceMs = options.sinceMs ?? (await findCursor(store, options.dest, podHost, log, options.includeRaw)) ?? untilMs - MAX_ANALYSIS_WINDOW_MS if (sinceMs >= untilMs) { log(`Nothing to pull: ${new Date(sinceMs).toISOString()} is not before ${new Date(untilMs).toISOString()}`) return records