diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 35e6be480..932209f27 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -22852,8 +22852,9 @@ paths: - $ref: "#/components/schemas/HealthAdminResponse" - $ref: "#/components/schemas/HealthPublicResponse" "503": - description: "Degraded — the database did not answer, or a role this process runs is not up. Same body shape as the 200, - with `status: \"degraded\"`. A restart loop reports this rather than failing to answer, so a monitor can + description: "Degraded — the database did not answer within two seconds, or a role this process runs is not up. Same + body shape as the 200, with `status: \"degraded\"`, and `reason: \"starting\"` while the background role is + still coming up after a start. A restart loop reports this rather than failing to answer, so a monitor can tell 'starting' from 'gone'." content: application/json: @@ -84210,14 +84211,18 @@ components: when the process serves web, and the worker loop is running when the process runs the worker. `degraded` on any of those failing." HealthReason: - description: "Present only when the process refuses to serve for a reason a person has to fix. - `encryption_key_mismatch`: the configured encryption key cannot open the data in this database (checked at - boot). Every other route then answers 503 with `meta.errorCode` `encryption.key_mismatch`, and background jobs - are not started. Fix: restore the original key, point the server at the matching database, or start with an - empty one; then restart." + description: "Why a `degraded` process is degraded, when it is something other than a component failure. + `encryption_key_mismatch`: the process refuses to serve for a reason a person has to fix; the configured + encryption key cannot open the data in this database (checked at boot). Every other route then answers 503 with + `meta.errorCode` `encryption.key_mismatch`, and background jobs are not started. Fix: restore the original key, + point the server at the matching database, or start with an empty one; then restart. `starting`: the database + answers but the background role (the pg-boss worker, or the producer of a web-only process) is still coming up + after a start. Transient; nothing to fix, poll again. The probe answers at once while it lasts rather than + waiting for the start to finish." type: string enum: - encryption_key_mismatch + - starting HealthPublicResponse: type: object properties: diff --git a/src/__tests__/measurement-write-plausibility-guard.test.ts b/src/__tests__/measurement-write-plausibility-guard.test.ts index 9b3cf83e0..4ff24ab5c 100644 --- a/src/__tests__/measurement-write-plausibility-guard.test.ts +++ b/src/__tests__/measurement-write-plausibility-guard.test.ts @@ -73,6 +73,10 @@ const WRITERS: Record = { validatedIn: "lib/validations/measurement.ts", }, "app/api/import/route.ts": { kind: "validator" }, + // v1.44 — the set-based insert writes rows that already passed the + // route's `validateMeasurementRange` per entry; everything else goes + // through the reconciler, which gates itself. + "app/api/measurements/batch/route.ts": { kind: "validator" }, "lib/mcp/writes.ts": { kind: "validator" }, "lib/measurements/create-from-telegram.ts": { kind: "validator" }, "lib/measurements/import-apple-health-export.ts": { kind: "validator" }, diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index da045a6ed..6242c8a5b 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -44,6 +44,7 @@ const WRITE_SITES: Record = { "app/api/measurements/bulk-delete/route.ts": TAIL, "app/api/measurements/by-external-ids/route.ts": TAIL, "app/api/measurements/restore/route.ts": TAIL, + "app/api/measurements/batch/route.ts": TAIL, "app/api/measurements/route.ts": TAIL, "app/api/mental-health/assessments/route.ts": TAIL, "lib/insights/score-row.ts": TAIL, diff --git a/src/app/api/health/__tests__/route.test.ts b/src/app/api/health/__tests__/route.test.ts index d3865a248..9c2ce8a16 100644 --- a/src/app/api/health/__tests__/route.test.ts +++ b/src/app/api/health/__tests__/route.test.ts @@ -163,3 +163,53 @@ describe("GET /api/health under ENCRYPTION_KEY_CHECK=warn", () => { } }); }); + +describe("GET /api/health while the background role starts", () => { + it("says `starting` at once, and `ok` when the role is up", async () => { + const { markStartup } = await import("@/lib/boot/startup-state"); + vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web"); + markStartup("starting"); + try { + const starting = await health(); + expect(starting.response.status).toBe(503); + expect(starting.body).toEqual({ status: "degraded", reason: "starting" }); + + getGlobalBoss.mockReturnValue({}); + markStartup("ready"); + const ready = await health(); + expect(ready.response.status).toBe(200); + expect(ready.body).toEqual({ status: "ok" }); + } finally { + markStartup("ready"); + } + }); + + it("does not call a failed start `starting`", async () => { + const { markStartup } = await import("@/lib/boot/startup-state"); + vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web"); + markStartup("failed"); + try { + const { response, body } = await health(); + expect(response.status).toBe(503); + expect(body).toEqual({ status: "degraded" }); + } finally { + markStartup("ready"); + } + }); + + it("does not wait on a database that does not answer", async () => { + vi.useFakeTimers(); + try { + vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web"); + getGlobalBoss.mockReturnValue({}); + queryRaw.mockImplementation(() => new Promise(() => undefined)); + const pending = health(); + await vi.advanceTimersByTimeAsync(2_000); + const { response, body } = await pending; + expect(response.status).toBe(503); + expect(body).toEqual({ status: "degraded" }); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/src/app/api/health/route.ts b/src/app/api/health/route.ts index 2e7764036..43b8ff1f8 100644 --- a/src/app/api/health/route.ts +++ b/src/app/api/health/route.ts @@ -11,19 +11,41 @@ import { isKeyMismatch, KEY_MISMATCH_HEALTH_REASON, } from "@/lib/boot/key-mismatch-state"; +import { startupPhase } from "@/lib/boot/startup-state"; + +/** + * How long the probe waits for `SELECT 1`. During a boot storm the pool can + * be saturated by the worker's start-up work; a probe that waits its turn + * held a monitor for seconds. Past this the database counts as not answering + * and the probe says `degraded` at once. + */ +const DB_PROBE_TIMEOUT_MS = 2_000; + +/** Reported while the background role is still coming up. */ +const STARTING_HEALTH_REASON = "starting"; + +async function probeDatabase(): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + prisma.$queryRaw`SELECT 1`.then(() => true), + new Promise((resolve) => { + timer = setTimeout(() => resolve(false), DB_PROBE_TIMEOUT_MS); + }), + ]); + } catch { + return false; + } finally { + if (timer) clearTimeout(timer); + } +} export const dynamic = "force-dynamic"; export const GET = apiHandler(async () => { annotate({ action: { name: "health.check" } }); - let dbOk = true; - - try { - await prisma.$queryRaw`SELECT 1`; - } catch { - dbOk = false; - } + const dbOk = await probeDatabase(); const worker = getWorkerStatus(); const producerReady = getGlobalBoss() !== null; @@ -31,20 +53,23 @@ export const GET = apiHandler(async () => { // other route; the probe says so in a word every caller may see, because // it is the one place a catalog install's status page reads. const keyMismatch = isKeyMismatch(); - const reason = keyMismatch ? KEY_MISMATCH_HEALTH_REASON : undefined; + // Otherwise, while the background role is still starting, the 503 below + // says so: a transient state, not a fault, and not a hang. + const roleDown = + (shouldRunWeb() && !producerReady) || + (shouldRunWorker() && !worker.running); + const reason = keyMismatch + ? KEY_MISMATCH_HEALTH_REASON + : dbOk && roleDown && startupPhase() === "starting" + ? STARTING_HEALTH_REASON + : undefined; // Under ENCRYPTION_KEY_CHECK=warn the same finding does not refuse; it is // named as a warning and leaves the status alone. const warning = !keyMismatch && getKeyMismatchWarning() ? KEY_MISMATCH_HEALTH_REASON : undefined; - const status = - !keyMismatch && - dbOk && - (!shouldRunWeb() || producerReady) && - (!shouldRunWorker() || worker.running) - ? "ok" - : "degraded"; + const status = !keyMismatch && dbOk && !roleDown ? "ok" : "degraded"; const statusCode = status === "ok" ? 200 : 503; const cacheHeaders = { diff --git a/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts b/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts index 09ed7da58..7ce59b9f5 100644 --- a/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts +++ b/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts @@ -9,6 +9,11 @@ * prefilter's read answering nothing, which is the pre-v1.42 path where every * row went through the reconciler. Statuses, reasons and counters must match * exactly; only the transaction count may differ. + * + * v1.44 — the golden run is also blind to the set-based insert: its + * `INSERT … ON CONFLICT DO NOTHING` stand-in lands nothing, so every row + * reaches the reconciler one at a time, the sequential path. The shipped run + * takes the set insert, and the two must still leave the same table. */ import { beforeEach, describe, expect, it, vi } from "vitest"; import { NextRequest } from "next/server"; @@ -29,6 +34,8 @@ const store = vi.hoisted(() => ({ rows: [] as StoredRow[], /** Golden mode: the prefilter's read answers nothing. */ prefilterBlind: false, + /** Golden mode: the set-based insert lands nothing. */ + setInsertBlind: false, nextId: 1, })); @@ -81,12 +88,70 @@ vi.mock("@/lib/db", () => { }, ), }, - $transaction: vi.fn(async (fn: (tx: unknown) => unknown) => fn({})), + $transaction: vi.fn(async (fn: (tx: unknown) => unknown) => + fn({ + measurement: { + // `INSERT … ON CONFLICT DO NOTHING` over both unique identities, + // in VALUES order, returning the rows that landed. + createManyAndReturn: async (args: { + data: Array<{ + userId: string; + type: string; + source: string; + externalId: string; + measuredAt: Date; + sleepStage: string | null; + value: number; + }>; + }) => { + if (store.setInsertBlind) return []; + const landed: Array<{ + type: string; + source: string; + externalId: string; + }> = []; + for (const d of args.data) { + const scope = store.rows.filter( + (r) => + r.userId === d.userId && + r.type === d.type && + r.source === d.source, + ); + const conflict = scope.some( + (r) => + r.externalId === d.externalId || + (sameInstant(r.measuredAt, d.measuredAt) && + r.sleepStage === (d.sleepStage ?? null)), + ); + if (conflict) continue; + store.rows.push({ + id: `m${store.nextId++}`, + userId: d.userId, + type: d.type, + source: d.source, + externalId: d.externalId, + measuredAt: d.measuredAt, + sleepStage: d.sleepStage ?? null, + value: d.value, + deletedAt: null, + }); + landed.push({ + type: d.type, + source: d.source, + externalId: d.externalId, + }); + } + return landed; + }, + }, + }), + ), }; return { prisma }; }); vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + lockMeasurementUser: async () => undefined, reconcileExternalMeasurement: async ( _tx: unknown, desired: { @@ -283,6 +348,7 @@ async function golden(setup: () => void, entries: Entry[]) { store.nextId = 1; setup(); store.prefilterBlind = true; + store.setInsertBlind = true; vi.mocked(prisma.$transaction).mockClear(); const old = await post(entries); const oldTransactions = vi.mocked(prisma.$transaction).mock.calls.length; @@ -292,6 +358,7 @@ async function golden(setup: () => void, entries: Entry[]) { store.nextId = 1; setup(); store.prefilterBlind = false; + store.setInsertBlind = false; vi.mocked(prisma.$transaction).mockClear(); const next = await post(entries); const transactions = vi.mocked(prisma.$transaction).mock.calls.length; @@ -304,6 +371,7 @@ beforeEach(() => { vi.clearAllMocks(); store.rows = []; store.prefilterBlind = false; + store.setInsertBlind = false; vi.mocked(getSession).mockResolvedValue({ session: { id: "s", expiresAt: new Date(NOW + HOUR) }, user: { id: USER, username: "u", role: "USER" as const }, @@ -334,7 +402,9 @@ describe("exact-duplicate prefilter (v1.42)", () => { entries, ); expect(next.duplicates).toBe(500); - expect(oldTransactions).toBe(500); + // Sequentially: two set-insert statements that land nothing, then one + // reconciler transaction per row. + expect(oldTransactions).toBe(502); expect(transactions).toBe(0); }); @@ -366,7 +436,8 @@ describe("exact-duplicate prefilter (v1.42)", () => { weight("uuid-move", new Date(NOW - DAY)), ]); expect(next.entries[0].status).toBe("updated"); - expect(transactions).toBe(1); + // The set insert skips it on the external identity, the reconciler moves it. + expect(transactions).toBe(2); }); it("4: the same id with another sleep stage still reconciles", async () => { @@ -389,7 +460,7 @@ describe("exact-duplicate prefilter (v1.42)", () => { sleepStage: 3, }; const { transactions } = await golden(setup, [entry]); - expect(transactions).toBe(1); + expect(transactions).toBe(2); }); it("5: `stats:` rows are never prefiltered; a second snapshot is superseded", async () => { @@ -615,3 +686,88 @@ describe("folded_window guard (v1.42)", () => { expect(data.entries[0].status).toBe("inserted"); }); }); + +describe("set-based insert (v1.44)", () => { + it("500 fresh samples land in two statements, as the reconciler would have", async () => { + const entries = Array.from({ length: 500 }, (_, i) => + pulse(`fresh-${i}`, new Date(NOW - DAY - i * 60_000), 60 + (i % 30)), + ); + const { next, oldTransactions, transactions } = await golden( + () => undefined, + entries, + ); + expect(next.inserted).toBe(500); + expect(oldTransactions).toBe(502); + expect(transactions).toBe(2); + }); + + it("a mix of fresh, moved and resurrected rows leaves the same table", async () => { + const at = (h: number) => new Date(NOW - DAY - h * HOUR); + const setup = () => { + // Moves: the stored row sits at another instant. + seed({ type: "WEIGHT", externalId: "moved", measuredAt: at(9) }); + // Resurrects: a tombstone at a different instant under the same id. + seed({ + type: "WEIGHT", + externalId: "back", + measuredAt: at(8), + deletedAt: new Date(NOW - HOUR), + }); + // Natural-identity collision under another id: reconciled, not inserted. + seed({ type: "WEIGHT", externalId: "other", measuredAt: at(3) }); + }; + const { next, transactions } = await golden(setup, [ + weight("fresh-a", at(1)), + weight("moved", at(2)), + weight("collides", at(3)), + weight("fresh-b", at(4)), + weight("back", at(5)), + weight("fresh-c", at(6)), + ]); + expect(next.entries.map((e) => e.status)).toEqual([ + "inserted", + "updated", + "updated", + "inserted", + "updated", + "inserted", + ]); + // One set statement, then the three rows it skipped. + expect(transactions).toBe(4); + }); + + it("a row sharing an identity in the batch flushes the buffer first", async () => { + const at = new Date(NOW - DAY); + const { next } = await golden( + () => undefined, + [ + weight("solo", new Date(at.getTime() - HOUR)), + weight("twice", at), + weight("twice", at), + ], + ); + expect(next.entries.map((e) => e.status)).toEqual([ + "inserted", + "inserted", + "duplicate", + ]); + }); + + it("keeps the order when a later reconcile would claim a buffered row's slot", async () => { + // Stored X at t1. C moves X to t5; D and E share t1, so neither is + // buffered. Sequentially C frees t1 before D arrives: D inserts, E + // merges into D. Reconciling D before C would instead take X's row for + // D and leave C to insert, a different table with the same rows. + const t1 = new Date(NOW - DAY - HOUR); + const t5 = new Date(NOW - DAY - 5 * HOUR); + const { next } = await golden( + () => seed({ type: "WEIGHT", externalId: "X", measuredAt: t1 }), + [weight("X", t5), weight("Y", t1), weight("Z", t1)], + ); + expect(next.entries.map((e) => e.status)).toEqual([ + "updated", + "inserted", + "updated", + ]); + }); +}); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 501b03fea..cda75930f 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -66,7 +66,11 @@ import { oppositeMergeSource, type MergeCandidate, } from "@/lib/measurements/cross-source-merge"; -import { reconcileExternalMeasurement } from "@/lib/measurements/reconcile-external-measurement"; +import { + lockMeasurementUser, + reconcileExternalMeasurement, + type MeasurementReconciliationVerdict, +} from "@/lib/measurements/reconcile-external-measurement"; import { classifyRangeRejection } from "@/lib/measurements/plausibility-gate"; import { FOLDED_WINDOW_REASON, @@ -134,6 +138,12 @@ const MAX_BATCH_ENTRIES = 500; // The cap is a DoS ceiling, not a tight bound. const MAX_BODY_BYTES = 4 * 1024 * 1024; +/** + * Rows per set-based insert statement. 500 entries fit two statements; each + * row binds about fifteen parameters, far under PostgreSQL's 65 535. + */ +const SET_INSERT_CHUNK = 250; + // v1.4.25 (security H-2): cap batch ingest at 60 // batches per user per minute. Healthy iOS sync drains its // HealthKit observer queue in well under one batch per minute (the @@ -635,6 +645,8 @@ async function postBatch(request: NextRequest): Promise { let updatedCount = 0; let duplicateCount = 0; let failedCount = 0; + // v1.44 — subset of `insertedCount` written by the set-based insert. + let setInsertedCount = 0; // v1.11.4 (iOS #2) — subset of `duplicateCount` collapsed by the // MANUAL↔APPLE_HEALTH same-reading merge (as opposed to a plain // composite-key duplicate). Surfaced as a dedicated wide-event count so @@ -855,6 +867,188 @@ async function postBatch(request: NextRequest): Promise { [...foldedIndexes].map((index) => prepared[index]), ); + /** The reconciler's input for one prepared row. */ + const desiredFor = (p: Prepared) => { + const statsRow = isStatsExternalId(p.row.externalId as string); + return { + userId: user.id, + type: p.row.type as MeasurementType, + value: p.row.value as number, + valueMin: p.row.valueMin as number | null, + valueMax: p.row.valueMax as number | null, + unit: p.row.unit as string, + source: p.row + .source as Prisma.MeasurementUncheckedCreateInput["source"], + measuredAt: p.row.measuredAt as Date, + externalId: p.row.externalId as string, + externalSourceVersion: p.row.externalSourceVersion as string | null, + glucoseContext: p.row + .glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"], + sleepStage: p.row + .sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"], + rhythmClassification: p.row + .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"], + deviceType: p.row.deviceType as string | null, + aggregationProvenance: statsRow + ? ("HEALTHKIT_STATISTICS" as const) + : null, + }; + }; + + const pushMergeCandidate = (p: Prepared) => { + if (isMergeableSource(p.row.source as string)) { + inBatchCandidates.push({ + type: p.row.type as MeasurementType, + source: p.row.source as string, + value: p.row.value as number, + measuredAt: p.row.measuredAt as Date, + }); + } + }; + + /** + * Record one row's verdict. `candidate` is false for a row whose merge + * candidate was already pushed when it was buffered (see below). + */ + const applyVerdict = ( + p: Prepared, + verdict: Pick & + Partial<{ + dirtyIdentities: Array<{ type: MeasurementType; measuredAt: Date }>; + error: { code?: string }; + }>, + candidate: boolean, + ) => { + switch (verdict.status) { + case "inserted": + results[p.index] = { index: p.index, status: "inserted" }; + insertedCount++; + insertedPrepared.push(p); + writtenIdentities.push({ + type: p.row.type as MeasurementType, + measuredAt: p.row.measuredAt as Date, + }); + break; + case "updated": + case "resurrected": + results[p.index] = { index: p.index, status: "updated" }; + updatedCount++; + writtenIdentities.push( + { + type: p.row.type as MeasurementType, + measuredAt: p.row.measuredAt as Date, + }, + ...(verdict.dirtyIdentities ?? []), + ); + break; + case "duplicate": + results[p.index] = { index: p.index, status: "duplicate" }; + duplicateCount++; + break; + case "failed": + results[p.index] = { + index: p.index, + status: "failed", + reason: verdict.error?.code ?? "write_failed", + }; + failedCount++; + return; + case "rejected_range": + // The reconciler carries the same plausibility gate this route + // already applied above, so reaching here means the two disagreed — + // report it exactly as the earlier guard would have, and let the + // wide-event tally record which end refused. + results[p.index] = outOfRangeResult( + p.index, + p.row.type as string, + p.row.value as number, + ); + return; + } + if (candidate) pushMergeCandidate(p); + }; + + const reconcileRow = (p: Prepared) => + prisma.$transaction((tx) => + reconcileExternalMeasurement(tx, desiredFor(p), { + exactExternalMatch: isStatsExternalId(p.row.externalId as string) + ? "update" + : "duplicate", + }), + ); + + // v1.44 — set-based inserts. A fresh sample used to cost one interactive + // transaction of about ten round trips (savepoint, three advisory locks, + // two candidate reads, the insert), so a 500-entry batch of new samples + // was 500 of those in a row: the 2–15 s outliers. + // + // The reconciler inserts exactly when no row, live or tombstoned, holds + // the entry's external identity `(type, source, externalId)` or its + // natural identity `(type, measuredAt, source, sleepStage)`. Both are + // full unique indexes (the natural one NULLS NOT DISTINCT), so + // `INSERT … ON CONFLICT DO NOTHING` inserts exactly those rows and skips + // every other; a skipped row then goes through the reconciler as before + // and gets its own verdict (update, resurrect, duplicate). The insert + // runs under the same per-account advisory lock the reconciler takes, so + // the two serialise like two reconcilers would. + // + // Only a row whose external and natural key each occur once in the batch + // is buffered, and the buffer is flushed before any other row reaches the + // reconciler. Such a row shares no identity with another entry, so no + // other entry's reconcile can create, move or retire a row at its + // identity, and inserting it out of turn changes nothing. Rows the insert + // skips are reconciled in their original order. A buffered row's merge + // candidate is pushed when it is buffered, as the sequential loop pushed + // it after a successful write; the only case that differs is a write that + // fails outright, where the candidate stays. + const pendingInserts: Prepared[] = []; + const flushPendingInserts = async () => { + if (pendingInserts.length === 0) return; + const batch = pendingInserts.splice(0, pendingInserts.length); + const landed = new Set(); + try { + for (let i = 0; i < batch.length; i += SET_INSERT_CHUNK) { + const slice = batch.slice(i, i + SET_INSERT_CHUNK); + const rows = await prisma.$transaction(async (tx) => { + await lockMeasurementUser(tx, user.id); + return tx.measurement.createManyAndReturn({ + data: slice.map((p) => ({ + ...desiredFor(p), + sleepStage: (p.row.sleepStage ?? + null) as Prisma.MeasurementUncheckedCreateInput["sleepStage"], + deletedAt: null, + })), + skipDuplicates: true, + select: { type: true, source: true, externalId: true }, + }); + }); + const keys = new Set( + rows.map((r) => `${r.type}::${r.source}::${r.externalId}`), + ); + for (const p of slice) { + if (keys.has(`${p.row.type}::${p.row.source}::${p.row.externalId}`)) + landed.add(p); + } + } + } catch (err) { + // Whatever did not land goes through the reconciler below, which + // turns a real database fault into a per-entry `failed` verdict. + annotate({ + action: { name: "measurement.batch.set_insert_fallback" }, + meta: { + error_class: err instanceof Error ? err.name : "unknown", + landed: landed.size, + pending: batch.length, + }, + }); + } + setInsertedCount += landed.size; + for (const p of batch) { + if (landed.has(p)) applyVerdict(p, { status: "inserted" }, false); + else applyVerdict(p, await reconcileRow(p), false); + } + }; + for (const p of prepared) { const externalId = p.row.externalId as string; const statsRow = isStatsExternalId(externalId); @@ -895,101 +1089,29 @@ async function postBatch(request: NextRequest): Promise { : { index: p.index, status: "duplicate" }; duplicateCount++; if (folded) foldedWindowCount++; - if (isMergeableSource(p.row.source as string)) { - inBatchCandidates.push({ - type: p.row.type as MeasurementType, - source: p.row.source as string, - value: p.row.value as number, - measuredAt: p.row.measuredAt as Date, - }); - } + pushMergeCandidate(p); continue; } - const verdict = await prisma.$transaction((tx) => - reconcileExternalMeasurement( - tx, - { - userId: user.id, - type: p.row.type as MeasurementType, - value: p.row.value as number, - valueMin: p.row.valueMin as number | null, - valueMax: p.row.valueMax as number | null, - unit: p.row.unit as string, - source: p.row - .source as Prisma.MeasurementUncheckedCreateInput["source"], - measuredAt: p.row.measuredAt as Date, - externalId, - externalSourceVersion: p.row.externalSourceVersion as string | null, - glucoseContext: p.row - .glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"], - sleepStage: p.row - .sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"], - rhythmClassification: p.row - .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"], - deviceType: p.row.deviceType as string | null, - aggregationProvenance: statsRow ? "HEALTHKIT_STATISTICS" : null, - }, - { exactExternalMatch: statsRow ? "update" : "duplicate" }, - ), + const naturalKey = naturalKeyOf( + p.row.type, + p.row.source, + p.row.measuredAt as Date, + p.row.sleepStage, ); - - switch (verdict.status) { - case "inserted": - results[p.index] = { index: p.index, status: "inserted" }; - insertedCount++; - insertedPrepared.push(p); - writtenIdentities.push({ - type: p.row.type as MeasurementType, - measuredAt: p.row.measuredAt as Date, - }); - break; - case "updated": - case "resurrected": - results[p.index] = { index: p.index, status: "updated" }; - updatedCount++; - writtenIdentities.push( - { - type: p.row.type as MeasurementType, - measuredAt: p.row.measuredAt as Date, - }, - ...(verdict.dirtyIdentities ?? []), - ); - break; - case "duplicate": - results[p.index] = { index: p.index, status: "duplicate" }; - duplicateCount++; - break; - case "failed": - results[p.index] = { - index: p.index, - status: "failed", - reason: verdict.error.code ?? "write_failed", - }; - failedCount++; - continue; - case "rejected_range": - // The reconciler carries the same plausibility gate this route - // already applied above, so reaching here means the two disagreed — - // report it exactly as the earlier guard would have, and let the - // wide-event tally record which end refused. - results[p.index] = outOfRangeResult( - p.index, - p.row.type as string, - p.row.value as number, - ); - continue; + if ( + externalKeyCounts.get(key) === 1 && + naturalKeyCounts.get(naturalKey) === 1 + ) { + pendingInserts.push(p); + pushMergeCandidate(p); + continue; } - if (isMergeableSource(p.row.source as string)) { - inBatchCandidates.push({ - type: p.row.type as MeasurementType, - source: p.row.source as string, - value: p.row.value as number, - measuredAt: p.row.measuredAt as Date, - }); - } + await flushPendingInserts(); + applyVerdict(p, await reconcileRow(p), true); } + await flushPendingInserts(); } // v1.43 — the hours, after the measurements. One write per slot: the last @@ -1342,6 +1464,9 @@ async function postBatch(request: NextRequest): Promise { // v1.42 — duplicates refused because a live `stats:` row already // covers the sample's hour or day (`folded-window.ts`). folded_window: foldedWindowCount, + // v1.44 — inserts that took the set-based path rather than one + // reconciler transaction each. + set_inserted: setInsertedCount, }, }); diff --git a/src/instrumentation.ts b/src/instrumentation.ts index 01de8ed80..3f9cca282 100644 --- a/src/instrumentation.ts +++ b/src/instrumentation.ts @@ -79,57 +79,76 @@ export async function register() { // A diagnostic must never break boot. } - const { shouldRunWorker } = await import("@/lib/process-type"); + // The background role starts after the server is up. Awaiting it here + // held every request, `/api/health` included, for the seconds pg-boss + // needs to start, provision its queues and run the boot discovery; the + // probe now answers `starting` meanwhile. A request that enqueues while + // the boss is not bound yet takes the same path as during a producer + // reconnect: a garnish skips its enqueue (the owning cron catches up), + // and a route that needs the worker answers 503 until it is up. + const [{ shouldRunWorker }, { markStartup }] = await Promise.all([ + import("@/lib/process-type"), + import("@/lib/boot/startup-state"), + ]); + markStartup("starting"); if (!shouldRunWorker()) { // Split deployment: web requests still produce queue jobs, but this // process owns no schedules, supervision, or consumers. - try { - const { startGlobalBossProducer } = - await import("@/lib/jobs/boss-instance"); - const boss = await startGlobalBossProducer(process.env.DATABASE_URL); - if (boss) { - const { registerShutdownHandlers } = - await import("@/lib/jobs/reminder/worker-lifecycle"); - registerShutdownHandlers(boss); + void (async () => { + try { + const { startGlobalBossProducer } = + await import("@/lib/jobs/boss-instance"); + const boss = await startGlobalBossProducer(process.env.DATABASE_URL); + if (boss) { + const { registerShutdownHandlers } = + await import("@/lib/jobs/reminder/worker-lifecycle"); + registerShutdownHandlers(boss); + } + markStartup(boss ? "ready" : "failed"); + } catch (err) { + markStartup("failed"); + // Queue-backed garnishes degrade independently; never make the web + // server unavailable because its producer connection failed. + console.error( + "[instrumentation] Failed to start pg-boss producer:", + err, + ); } - } catch (err) { - // Queue-backed garnishes degrade independently; never make the web - // server unavailable because its producer connection failed. - console.error( - "[instrumentation] Failed to start pg-boss producer:", - err, - ); - } + })(); return; } - const { WideEventBuilder } = await import("@/lib/logging/event-builder"); - const { emitIfSampled } = await import("@/lib/logging/transports"); + void (async () => { + const { WideEventBuilder } = await import("@/lib/logging/event-builder"); + const { emitIfSampled } = await import("@/lib/logging/transports"); - // Emit startup event - const startupEvt = new WideEventBuilder("background"); - startupEvt.setBackground({ task_name: "startup" }); + // Emit startup event + const startupEvt = new WideEventBuilder("background"); + startupEvt.setBackground({ task_name: "startup" }); - try { - const { startReminderWorker } = - await import("@/lib/jobs/reminder-worker"); - await startReminderWorker(); + try { + const { startReminderWorker } = + await import("@/lib/jobs/reminder-worker"); + await startReminderWorker(); + markStartup("ready"); - startupEvt.addMeta("reminder_worker", "started"); - startupEvt.finish(); - emitIfSampled(startupEvt.toJSON()); - } catch (err) { - startupEvt.setError(err); - startupEvt.addMeta("reminder_worker", "failed"); - startupEvt.finish(); - emitIfSampled(startupEvt.toJSON()); + startupEvt.addMeta("reminder_worker", "started"); + startupEvt.finish(); + emitIfSampled(startupEvt.toJSON()); + } catch (err) { + markStartup("failed"); + startupEvt.setError(err); + startupEvt.addMeta("reminder_worker", "failed"); + startupEvt.finish(); + emitIfSampled(startupEvt.toJSON()); - // Still log to stderr for container health checks - console.error( - "[instrumentation] CRITICAL: Failed to start reminder worker:", - err, - ); - } + // Still log to stderr for container health checks + console.error( + "[instrumentation] CRITICAL: Failed to start reminder worker:", + err, + ); + } + })(); } } diff --git a/src/lib/boot/startup-state.ts b/src/lib/boot/startup-state.ts new file mode 100644 index 000000000..59cd80a22 --- /dev/null +++ b/src/lib/boot/startup-state.ts @@ -0,0 +1,23 @@ +/** + * Where this process is in bringing up its background role. + * + * The pg-boss worker (or, in a web-only process, the send-only producer) + * starts after the HTTP server is already serving: starting it inside + * `register()` held every request, `/api/health` included, for the seconds + * pg-boss needs to start, provision its queues and run the boot discovery. + * The probe reads this to say `starting` while that is under way rather than + * hanging, and the operator sees a 503 that names a transient state. + * + * `null` means no background role was started in this process at all. + */ +export type StartupPhase = "starting" | "ready" | "failed"; + +const KEY = Symbol.for("healthlog.boot.startupPhase"); + +export function markStartup(phase: StartupPhase): void { + (globalThis as Record)[KEY] = phase; +} + +export function startupPhase(): StartupPhase | null { + return ((globalThis as Record)[KEY] as StartupPhase) ?? null; +} diff --git a/src/lib/measurements/reconcile-external-measurement.ts b/src/lib/measurements/reconcile-external-measurement.ts index d4fb50784..0fdd7b773 100644 --- a/src/lib/measurements/reconcile-external-measurement.ts +++ b/src/lib/measurements/reconcile-external-measurement.ts @@ -140,7 +140,13 @@ function lockKeys(desired: ExternalMeasurementWrite): string[] { ].sort(); } -async function lockUser( +/** + * The per-account advisory lock every cooperative measurement writer takes + * first. Exported for the batch route's set-based insert, which takes it + * around its `INSERT … ON CONFLICT DO NOTHING` so it serialises with the + * reconciler exactly as a second reconciler would. + */ +export async function lockMeasurementUser( tx: PrismaTypes.TransactionClient, userId: string, ): Promise { @@ -259,7 +265,7 @@ export async function reconcileExternalMeasurement( await tx.$executeRawUnsafe(`SAVEPOINT ${SAVEPOINT}`); try { - await lockUser(tx, desired.userId); + await lockMeasurementUser(tx, desired.userId); for (const key of lockKeys(desired)) { await tx.$queryRaw( Prisma.sql` diff --git a/src/lib/openapi/routes/health.ts b/src/lib/openapi/routes/health.ts index 2540e8fe6..90d9221ba 100644 --- a/src/lib/openapi/routes/health.ts +++ b/src/lib/openapi/routes/health.ts @@ -26,11 +26,14 @@ const healthStatusEnum = z.enum(["ok", "degraded"]).meta({ "`ok` when the database answers AND every role this process runs holds up: the pg-boss producer is bound when the process serves web, and the worker loop is running when the process runs the worker. `degraded` on any of those failing.", }); -const healthReason = z.enum(["encryption_key_mismatch"]).optional().meta({ - id: "HealthReason", - description: - "Present only when the process refuses to serve for a reason a person has to fix. `encryption_key_mismatch`: the configured encryption key cannot open the data in this database (checked at boot). Every other route then answers 503 with `meta.errorCode` `encryption.key_mismatch`, and background jobs are not started. Fix: restore the original key, point the server at the matching database, or start with an empty one; then restart.", -}); +const healthReason = z + .enum(["encryption_key_mismatch", "starting"]) + .optional() + .meta({ + id: "HealthReason", + description: + "Why a `degraded` process is degraded, when it is something other than a component failure. `encryption_key_mismatch`: the process refuses to serve for a reason a person has to fix; the configured encryption key cannot open the data in this database (checked at boot). Every other route then answers 503 with `meta.errorCode` `encryption.key_mismatch`, and background jobs are not started. Fix: restore the original key, point the server at the matching database, or start with an empty one; then restart. `starting`: the database answers but the background role (the pg-boss worker, or the producer of a web-only process) is still coming up after a start. Transient; nothing to fix, poll again. The probe answers at once while it lasts rather than waiting for the start to finish.", + }); const healthPublicResponse = z .object({ status: healthStatusEnum, reason: healthReason }) @@ -92,7 +95,7 @@ export const healthPaths: NonNullable = { }, "503": { description: - "Degraded — the database did not answer, or a role this process runs is not up. Same body shape as the 200, with `status: \"degraded\"`. A restart loop reports this rather than failing to answer, so a monitor can tell 'starting' from 'gone'.", + "Degraded — the database did not answer within two seconds, or a role this process runs is not up. Same body shape as the 200, with `status: \"degraded\"`, and `reason: \"starting\"` while the background role is still coming up after a start. A restart loop reports this rather than failing to answer, so a monitor can tell 'starting' from 'gone'.", content: { "application/json": { schema: z.union([healthAdminResponse, healthPublicResponse]), diff --git a/tests/integration/measurements-batch-perf.test.ts b/tests/integration/measurements-batch-perf.test.ts new file mode 100644 index 000000000..900a4e4d1 --- /dev/null +++ b/tests/integration/measurements-batch-perf.test.ts @@ -0,0 +1,255 @@ +/** + * `POST /api/measurements/batch` under a realistic large batch. + * + * Production saw this route take 2 s and more, up to 15 s, on full + * 500-entry uploads. The cost was one interactive reconciler transaction + * (about ten round trips) per fresh sample. Since v1.44 fresh samples land + * through a set-based `INSERT … ON CONFLICT DO NOTHING` and only rows that + * hit an existing identity reach the reconciler. + * + * This suite pins the shape of the work rather than a machine's speed: + * - a fresh 500-entry batch opens a handful of transactions, not 500; + * - a re-post of the same batch opens none (the duplicate prefilter); + * - the rollup leg runs once per request, not once per row; + * - the wall-clock bound is generous, a tripwire for a regression back to + * per-row work, not a benchmark. + */ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { cookieJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const TEST_USER_ID = "user-batch-perf-test"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const rollupCalls = vi.hoisted(() => ({ count: 0 })); +vi.mock("@/lib/rollups/after-measurement-mutation", async () => { + const actual = await vi.importActual< + typeof import("@/lib/rollups/after-measurement-mutation") + >("@/lib/rollups/after-measurement-mutation"); + return { + ...actual, + afterMeasurementMutation: ( + ...args: Parameters + ) => { + rollupCalls.count += 1; + return actual.afterMeasurementMutation(...args); + }, + }; +}); + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + cookieJar.clear(); + rollupCalls.count = 0; + await getPrismaClient().user.create({ + data: { + id: TEST_USER_ID, + username: "batch-perf", + email: "batch-perf@example.test", + timezone: "Europe/Berlin", + }, + }); + const session = await getPrismaClient().session.create({ + data: { + userId: TEST_USER_ID, + expiresAt: new Date(Date.now() + 60 * 60 * 1000), + }, + }); + cookieJar.set("healthlog_session", session.id); +}); + +/** + * What a phone sends after a day offline: dense heart rate, a few HRV and + * oxygen readings, a night of sleep stages, and the day's cumulative totals. + */ +function realisticBatch(dayOffset = 2): Array> { + const day = new Date(Date.now() - dayOffset * 86_400_000); + day.setUTCHours(0, 0, 0, 0); + const at = (minutes: number) => + new Date(day.getTime() + minutes * 60_000).toISOString(); + const entries: Array> = []; + for (let i = 0; i < 400; i++) { + entries.push({ + hkIdentifier: "HKQuantityTypeIdentifierHeartRate", + value: 55 + (i % 40), + unit: "count/min", + startDate: at(i * 3), + endDate: at(i * 3), + externalId: `perf-hr-${dayOffset}-${i}`, + }); + } + for (let i = 0; i < 40; i++) { + entries.push({ + hkIdentifier: "HKQuantityTypeIdentifierHeartRateVariabilitySDNN", + value: 30 + i, + unit: "ms", + startDate: at(i * 30 + 1), + endDate: at(i * 30 + 1), + externalId: `perf-hrv-${dayOffset}-${i}`, + }); + entries.push({ + hkIdentifier: "HKQuantityTypeIdentifierOxygenSaturation", + value: 0.96, + unit: "%", + startDate: at(i * 30 + 2), + endDate: at(i * 30 + 2), + externalId: `perf-spo2-${dayOffset}-${i}`, + }); + } + for (let i = 0; i < 16; i++) { + entries.push({ + hkIdentifier: "HKCategoryTypeIdentifierSleepAnalysis", + value: 30, + unit: "min", + startDate: at(-480 + i * 30), + endDate: at(-450 + i * 30), + externalId: `perf-sleep-${dayOffset}-${i}`, + sleepStage: 3 + (i % 3), + }); + } + const dayKey = day.toISOString().slice(0, 10); + for (const [hk, unit, value] of [ + ["HKQuantityTypeIdentifierStepCount", "count", 9_431], + ["HKQuantityTypeIdentifierActiveEnergyBurned", "kcal", 512], + ["HKQuantityTypeIdentifierAppleExerciseTime", "min", 42], + ["HKQuantityTypeIdentifierAppleStandTime", "min", 180], + ] as const) { + entries.push({ + hkIdentifier: hk, + value, + unit, + startDate: at(0), + endDate: at(18 * 60), + externalId: `stats:${hk}:${dayKey}`, + }); + } + return entries; +} + +async function post(entries: Array>) { + const { POST } = await import("@/app/api/measurements/batch/route"); + const res = await POST( + new NextRequest("http://localhost/api/measurements/batch", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ entries }), + }), + ); + expect(res.status).toBe(200); + return ( + (await res.json()) as { + data: { + inserted: number; + updated: number; + duplicates: number; + failed: number; + skipped: number; + }; + } + ).data; +} + +describe("POST /api/measurements/batch — large-batch cost (real Postgres)", () => { + it("writes a fresh 500-entry batch in a handful of transactions", async () => { + const { prisma } = await import("@/lib/db"); + const entries = realisticBatch(); + expect(entries).toHaveLength(500); + + const transaction = vi.spyOn(prisma, "$transaction"); + const started = performance.now(); + const result = await post(entries); + const elapsedMs = performance.now() - started; + const transactions = transaction.mock.calls.length; + transaction.mockRestore(); + + expect(result.failed).toBe(0); + expect(result.inserted).toBe(500); + // Two set statements (250 rows each) plus the hour-bucket and checkpoint + // writes. Before v1.44 this was one transaction per row: 500. + expect(transactions).toBeLessThanOrEqual(10); + expect(rollupCalls.count).toBe(1); + expect( + await getPrismaClient().measurement.count({ + where: { userId: TEST_USER_ID }, + }), + ).toBe(500); + // A tripwire, not a benchmark: per-row work on a local container sits + // well above this, the set path well below. + expect(elapsedMs).toBeLessThan(10_000); + console.info( + `[batch-perf] fresh 500: ${Math.round(elapsedMs)} ms, ${transactions} transactions`, + ); + }); + + it("answers a re-post of the same batch without a transaction", async () => { + const { prisma } = await import("@/lib/db"); + const entries = realisticBatch(); + await post(entries); + rollupCalls.count = 0; + + const transaction = vi.spyOn(prisma, "$transaction"); + const started = performance.now(); + const result = await post(entries); + const elapsedMs = performance.now() - started; + const transactions = transaction.mock.calls.length; + transaction.mockRestore(); + + // The four `stats:` rows carry overwrite semantics and reconcile; every + // sample is an exact duplicate and never reaches a transaction. + expect(result.inserted).toBe(0); + expect(result.duplicates + result.updated).toBe(500); + expect(transactions).toBeLessThanOrEqual(8); + expect(elapsedMs).toBeLessThan(10_000); + console.info( + `[batch-perf] re-post 500: ${Math.round(elapsedMs)} ms, ${transactions} transactions`, + ); + }); + + it("keeps the set path's verdicts when the batch overlaps stored rows", async () => { + const entries = realisticBatch(); + await post(entries.slice(0, 250)); + // Move half of the stored heart-rate rows: same id, another instant. + const moved = entries.slice(0, 100).map((e) => ({ + ...e, + startDate: new Date( + Date.parse(e.endDate as string) + 1_000, + ).toISOString(), + endDate: new Date(Date.parse(e.endDate as string) + 1_000).toISOString(), + })); + const result = await post([...moved, ...entries.slice(250)]); + expect(result.failed).toBe(0); + expect(result.updated).toBeGreaterThanOrEqual(100); + expect(result.inserted).toBe(250); + expect( + await getPrismaClient().measurement.count({ + where: { userId: TEST_USER_ID, deletedAt: null }, + }), + ).toBe(500); + }); +});