diff --git a/Dockerfile b/Dockerfile index 2630d356c..625feff6d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -212,11 +212,13 @@ COPY --from=builder /app/src/generated ./src/generated # application graph (`@/…`, the generated Prisma client) run from a source # checkout instead. `src/__tests__/container-scripts-ship-guard.test.ts` holds # the runbooks and the entrypoint against this list. `pending-migrations.mjs` -# is the entrypoint's own: it logs the pending migration count before -# `migrate deploy`. Root-owned on purpose: the app user can run them but not +# and `migrate-deploy.mjs` are the entrypoint's own: the first logs the +# pending migration count, the second runs `migrate deploy` with its +# lock-timeout retry. Root-owned on purpose: the app user can run them but not # rewrite them. COPY --from=builder /app/scripts/reset-password.mjs ./scripts/reset-password.mjs COPY --from=builder /app/scripts/pending-migrations.mjs ./scripts/pending-migrations.mjs +COPY --from=builder /app/scripts/migrate-deploy.mjs ./scripts/migrate-deploy.mjs COPY --from=builder /app/src/lib/auth/argon2-params.mjs ./src/lib/auth/argon2-params.mjs COPY --from=builder /app/scripts/repair-intake-anomalies.ts ./scripts/repair-intake-anomalies.ts # Prove the reset CLI's imports resolve in this tree: with every import found @@ -224,7 +226,8 @@ COPY --from=builder /app/scripts/repair-intake-anomalies.ts ./scripts/repair-int RUN node scripts/reset-password.mjs 2>&1 | grep -q "reset-password: usage:" # The boot notice parses, and the `pg` it loads on a real boot resolves here. RUN env -u DATABASE_URL node scripts/pending-migrations.mjs && \ - node --input-type=module -e "await import('pg')" + node --input-type=module -e "await import('pg')" && \ + node --input-type=module -e "await import('./scripts/migrate-deploy.mjs')" # Install the migration CLI, its config dependency, and a pinned launcher for # the maintenance scripts already shipped in the standalone tree. The Prisma diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh index 8eff71a6b..b33d52002 100644 --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -88,9 +88,12 @@ echo "HealthLog: Database is reachable." # Say how many migrations this image is about to apply, with a reminder to # back up first. Reporting only: it never stops the boot. node scripts/pending-migrations.mjs || true +# `migrate-deploy.mjs` runs `prisma migrate deploy` and retries only a +# migration that timed out on a lock and is marked safe to re-run; any other +# failure stops the boot with Prisma's exit code (docs/ops/deploy.md). echo "HealthLog: Running database migrations..." NODE_PATH=/opt/prisma-cli/node_modules \ - node /opt/prisma-cli/node_modules/prisma/build/index.js migrate deploy + node scripts/migrate-deploy.mjs echo "HealthLog: Migrations complete." # ── Start application ──────────────────────────────────────── diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 4629570e2..a967cc72b 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -21437,8 +21437,9 @@ paths: per day with a count, rather than one per request. `retentionDays` rides the response because audit history is purged on a schedule the operator configures: a view that showed the rows without the window would imply a completeness it does not have, and clients must state the number the server sends rather than assume the 365-day - default. Whether the list stopped at its row ceiling is `meta.truncated` (it was a `truncated` field in `data` - before v1.43.7). Not delegable: a grant is not an introduction to the other people in the household." + default. Whether the list stopped at its row ceiling is `meta.truncated`; the `truncated` field in `data` + carries the same value and is deprecated. Not delegable: a grant is not an introduction to the other people in + the household." responses: "200": description: The activity, and how far back it can see. @@ -81393,9 +81394,15 @@ components: description: How far back this feed can see, in days. Resolved from the instance's audit retention setting (365 by default, operator-configurable), NOT a constant — an instance running a 90-day window sends 90. The query is bounded to this window, so it is the real limit of the list and not only a caption. Show it beside the rows. + truncated: + type: boolean + deprecated: true + description: "Deprecated: read `meta.truncated`, which carries the same value. Whether the list stopped at its row + ceiling; kept in `data` for clients built against v1.43.6 and earlier." required: - entries - retentionDays + - truncated additionalProperties: false description: The owner's record of what somebody else did in their record, bounded by the instance's audit retention. RecordActivityEntry: diff --git a/docs/integrations/google-health.md b/docs/integrations/google-health.md index 0b187be94..554f1ed1f 100644 --- a/docs/integrations/google-health.md +++ b/docs/integrations/google-health.md @@ -245,8 +245,10 @@ reading: upper-bound filter Google refused. - `stopReason`: why it ended early. `time_budget` and `page_budget` are the normal end of a slice, `rate_limited` means Google would not take more - requests for now, `job_budget` that the job itself ran out of time. - `null` means the slice read everything it started. + requests for now, `job_budget` that the job itself ran out of time, + `lane_wanted` that the slice made room for another heavy job (a nightly + run, a backup, an upload) that was waiting; the next slice starts two + minutes later. `null` means the slice read everything it started. - `collectionsDone` of `collectionsTotal`: data types read to the end so far. - `collections`: one entry per data type the slice touched, with its diff --git a/docs/ops/deploy.md b/docs/ops/deploy.md index 1bf418cdc..89ac3d380 100644 --- a/docs/ops/deploy.md +++ b/docs/ops/deploy.md @@ -40,6 +40,55 @@ not reachable". The entrypoint also runs the migrations on every boot, so step 2 is optional; it is there for when you want to see the migration result before the new container takes traffic. +### A migration that timed out on a lock + +Some migrations bound how long they wait for a table lock, so that a +long-running holder (most often an autovacuum that is preventing +transaction-ID wraparound, which does not step aside) cannot queue every +write to that table behind the migration. Migration `0412` waits at most ten +seconds for `measurements`. When the wait runs out, the boot log shows +Prisma's report followed by a line from the entrypoint: + +``` +Error: P3018 +... +Migration name: 0412_measurement_hot_updates + +Database error code: 55P03 +... +HealthLog: migration 0412_measurement_hot_updates timed out waiting for a lock (55P03) and changed nothing; marking it rolled back and retrying in 15s (retry 1 of 5). +``` + +What happens on its own: such a migration is written so that a timed-out +attempt changes nothing, and it says so in its file +(`-- @retry-on-lock-timeout`). The entrypoint then marks that one migration +rolled back (`prisma migrate resolve --rolled-back`) and runs +`migrate deploy` again after 15, 30, 60, 120 and 240 seconds. The app starts +as soon as an attempt gets the lock. A later boot that finds the attempt still +on record (Prisma's `P3009`) reads the cause from `_prisma_migrations` and +does the same. Nothing else is retried: any other failure, and a lock timeout +in a migration that does not carry the marker, stops the boot exactly as +before. + +If all five retries time out, the log ends with +`HealthLog: ERROR — migration … timed out on a lock 6 times in a row; giving +up.` and the container exits. Something has held the table for about eight +minutes. Find it: + +```sh +docker compose exec db psql -U healthlog -c " + SELECT pid, backend_type, state, now() - xact_start AS running, query + FROM pg_stat_activity + WHERE pid IN (SELECT pid FROM pg_locks + WHERE relation = 'measurements'::regclass)" +``` + +An `autovacuum: VACUUM … (to prevent wraparound)` row finishes on its own; +wait for it to disappear and start the container again (a restart policy does +that for you, and each boot retries). Anything else is a session you can +judge: a stuck client may be ended with `SELECT pg_terminate_backend()`. +You never need to run `migrate resolve` by hand for this case. + ### v1.11.0 notes Migrations `0110`–`0114` are additive (new tables / columns for the diff --git a/prisma/migrations/0411_encrypt_insights_cached_text/migration.sql b/prisma/migrations/0411_encrypt_insights_cached_text/migration.sql index dfd6ff1e1..6718709f4 100644 --- a/prisma/migrations/0411_encrypt_insights_cached_text/migration.sql +++ b/prisma/migrations/0411_encrypt_insights_cached_text/migration.sql @@ -16,7 +16,7 @@ -- -- Additive only: no table rewrite, no data change. -ALTER TABLE "users" ADD COLUMN "insights_cached_text_encrypted" BYTEA; +ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "insights_cached_text_encrypted" BYTEA; -- Partial index on exactly the backfill's discovery predicate, so the boot -- scan is index-only and shrinks to nothing as the backfill converges (the diff --git a/prisma/migrations/0412_measurement_hot_updates/migration.sql b/prisma/migrations/0412_measurement_hot_updates/migration.sql index 6846cee4f..d49f488c7 100644 --- a/prisma/migrations/0412_measurement_hot_updates/migration.sql +++ b/prisma/migrations/0412_measurement_hot_updates/migration.sql @@ -28,13 +28,22 @@ -- `ALTER TABLE ... SET` SHARE UPDATE EXCLUSIVE; both wait for an autovacuum -- that is preventing wraparound, which does not step aside, and while they -- wait every write to `measurements` queues behind them, including the ones --- the running release still serves during a deploy. `migrate deploy` runs the --- whole file as one transaction, so `SET LOCAL` bounds the wait for this file --- alone: past ten seconds the migration fails and rolls back whole, the --- writes go on, and the next deploy tries again once `prisma migrate resolve --- --rolled-back 0412_measurement_hot_updates` has cleared the failed attempt. - -SET LOCAL lock_timeout = '10s'; +-- the running release still serves during a deploy. The wait is bounded at +-- ten seconds; past that the attempt fails with SQLSTATE 55P03 and the queued +-- writes go on. +-- +-- `migrate deploy` does not run a migration file inside one transaction: each +-- statement commits on its own, so a top-level `SET LOCAL` does nothing and a +-- file that fails half-way stays half-applied. The two statements that take a +-- lock therefore run inside one `DO` block, which is a single statement and +-- so commits or rolls back whole, and the timeout is set inside it with +-- `set_config(..., true)`, scoped to that statement. The function before it +-- is `CREATE OR REPLACE` and harmless to run again. A lock timeout leaves the +-- file safe to apply once more, which is what the marker below declares: +-- `scripts/migrate-deploy.mjs` marks the attempt rolled back and retries it +-- with a backoff (docs/ops/deploy.md, "A migration that timed out on a lock"). +-- +-- @retry-on-lock-timeout CREATE OR REPLACE FUNCTION "measurements_keep_stamps_on_noop"() RETURNS trigger LANGUAGE plpgsql AS $$ @@ -48,13 +57,19 @@ BEGIN END; $$; -CREATE TRIGGER "measurements_noop_update_keeps_stamps" - BEFORE UPDATE ON "measurements" - FOR EACH ROW - EXECUTE FUNCTION "measurements_keep_stamps_on_noop"(); +DO $migration$ +BEGIN + PERFORM set_config('lock_timeout', '10s', true); -ALTER TABLE "measurements" SET ( - fillfactor = 85, - autovacuum_vacuum_scale_factor = 0.05, - autovacuum_analyze_scale_factor = 0.05 -); + CREATE TRIGGER "measurements_noop_update_keeps_stamps" + BEFORE UPDATE ON "measurements" + FOR EACH ROW + EXECUTE FUNCTION "measurements_keep_stamps_on_noop"(); + + ALTER TABLE "measurements" SET ( + fillfactor = 85, + autovacuum_vacuum_scale_factor = 0.05, + autovacuum_analyze_scale_factor = 0.05 + ); +END +$migration$; diff --git a/prisma/migrations/0413_strava_backfill_resume/migration.sql b/prisma/migrations/0413_strava_backfill_resume/migration.sql index 821446acc..8d11102ef 100644 --- a/prisma/migrations/0413_strava_backfill_resume/migration.sql +++ b/prisma/migrations/0413_strava_backfill_resume/migration.sql @@ -9,4 +9,4 @@ -- Nullable and unset for every existing row: an unfinished backfill starts -- once more from the newest page and records its place from then on. -ALTER TABLE "users" ADD COLUMN "strava_backfill_resume_before" TIMESTAMP(3); +ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "strava_backfill_resume_before" TIMESTAMP(3); diff --git a/scripts/__tests__/migrate-deploy.test.ts b/scripts/__tests__/migrate-deploy.test.ts new file mode 100644 index 000000000..a63edc96b --- /dev/null +++ b/scripts/__tests__/migrate-deploy.test.ts @@ -0,0 +1,283 @@ +/** + * The boot's `migrate deploy` wrapper (`scripts/migrate-deploy.mjs`): which + * failures it retries, and the loop it runs around Prisma. + * + * The report texts below are what Prisma 7's `migrate deploy` printed against + * a real Postgres 16 for a migration that hit its `lock_timeout` (P3018), and + * for the boot after that (P3009). The loop is driven end to end through a + * stand-in CLI that plays those reports back, so no database is needed here; + * `tests/integration/migrate-deploy-lock-retry.test.ts` runs the real CLI. + */ +import { spawnSync } from "node:child_process"; +import { + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; + +import { + decideLockRetry, + lockTimeoutInReport, + retryDelays, +} from "../migrate-deploy.mjs"; + +const P3018_LOCK = `Applying migration \`0412_measurement_hot_updates\` +Error: P3018 + +A migration failed to apply. New migrations cannot be applied before the error is recovered from. Read more about how to resolve migration issues in a production database: https://pris.ly/d/migrate-resolve + +Migration name: 0412_measurement_hot_updates + +Database error code: 55P03 + +Database error: +ERROR: canceling statement due to lock timeout +`; + +const P3018_UNIQUE = P3018_LOCK.replace("55P03", "23505").replace( + "canceling statement due to lock timeout", + 'duplicate key value violates unique constraint "x"', +); + +const P3009 = `Error: P3009 + +migrate found failed migrations in the target database, new migrations will not be applied. Read more about how to resolve migration issues in a production database: https://pris.ly/d/migrate-resolve +The \`0412_measurement_hot_updates\` migration started at 2026-10-20 08:38:22.114155 UTC failed +`; + +const MARKED = "-- @retry-on-lock-timeout\nDO $$ BEGIN END $$;\n"; +const UNMARKED = "ALTER TABLE x ADD COLUMN y int;\n"; + +const reader = + (sql: string | null) => + (_name: string): string | null => + sql; + +describe("which failure is retried", () => { + it("reads the failed migration and its SQLSTATE from Prisma's report", () => { + expect(lockTimeoutInReport(P3018_LOCK)).toBe( + "0412_measurement_hot_updates", + ); + expect(lockTimeoutInReport(P3018_UNIQUE)).toBeNull(); + expect(lockTimeoutInReport(P3009)).toBeNull(); + expect(lockTimeoutInReport(null)).toBeNull(); + }); + + it("retries a lock timeout of a migration marked safe to re-run", () => { + expect( + decideLockRetry({ + output: P3018_LOCK, + failedRows: null, + readMigration: reader(MARKED), + }), + ).toEqual({ retry: true, migration: "0412_measurement_hot_updates" }); + }); + + it("does not retry a lock timeout of an unmarked migration", () => { + const decision = decideLockRetry({ + output: P3018_LOCK, + failedRows: null, + readMigration: reader(UNMARKED), + }); + expect(decision.retry).toBe(false); + expect(decision).toMatchObject({ + reason: expect.stringContaining("not marked safe to re-run"), + }); + }); + + it("does not retry any other database error", () => { + expect( + decideLockRetry({ + output: P3018_UNIQUE, + failedRows: null, + readMigration: reader(MARKED), + }), + ).toEqual({ retry: false, reason: "not a lock timeout" }); + }); + + it("does not retry a migration the image does not ship", () => { + expect( + decideLockRetry({ + output: P3018_LOCK, + failedRows: null, + readMigration: reader(null), + }).retry, + ).toBe(false); + }); + + it("on the boot after a failure, reads the cause from the stored row", () => { + const failedRows = [ + { migration_name: "0412_measurement_hot_updates", logs: P3018_LOCK }, + ]; + expect( + decideLockRetry({ + output: P3009, + failedRows, + readMigration: reader(MARKED), + }), + ).toEqual({ retry: true, migration: "0412_measurement_hot_updates" }); + // A stored failure of another kind stays fatal. + expect( + decideLockRetry({ + output: P3009, + failedRows: [{ ...failedRows[0], logs: P3018_UNIQUE }], + readMigration: reader(MARKED), + }).retry, + ).toBe(false); + // Two failed rows, or rows that could not be read: not a state to guess at. + expect( + decideLockRetry({ + output: P3009, + failedRows: [failedRows[0], failedRows[0]], + readMigration: reader(MARKED), + }).retry, + ).toBe(false); + expect( + decideLockRetry({ + output: P3009, + failedRows: null, + readMigration: reader(MARKED), + }).retry, + ).toBe(false); + }); + + it("refuses a migration name that is not a directory name", () => { + expect( + decideLockRetry({ + output: P3018_LOCK.replace( + "Migration name: 0412_measurement_hot_updates", + "Migration name: ../../etc", + ), + failedRows: null, + readMigration: reader(MARKED), + }).retry, + ).toBe(false); + }); + + it("reads the backoff from the environment and falls back on nonsense", () => { + expect(retryDelays(undefined)).toEqual([15, 30, 60, 120, 240]); + expect(retryDelays("0, 1")).toEqual([0, 1]); + expect(retryDelays("a,1")).toEqual([15, 30, 60, 120, 240]); + }); +}); + +describe("the retry loop", () => { + const dirs: string[] = []; + afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true }); + }); + + /** + * A stand-in Prisma CLI: each `migrate deploy` prints the next scripted + * report and exits with its code; every call is appended to `calls.log`. + */ + function setup(reports: Array<[number, string]>, sql: string) { + const dir = mkdtempSync(join(tmpdir(), "hl-migrate-deploy-")); + dirs.push(dir); + const migrations = join(dir, "migrations"); + mkdirSync(join(migrations, "0412_measurement_hot_updates"), { + recursive: true, + }); + writeFileSync( + join(migrations, "0412_measurement_hot_updates", "migration.sql"), + sql, + ); + writeFileSync(join(dir, "reports.json"), JSON.stringify(reports)); + const cli = join(dir, "fake-prisma.mjs"); + writeFileSync( + cli, + `import { appendFileSync, readFileSync, writeFileSync, existsSync } from "node:fs"; +const dir = ${JSON.stringify(dir)}; +const args = process.argv.slice(2); +appendFileSync(dir + "/calls.log", args.join(" ") + "\\n"); +if (args[0] === "migrate" && args[1] === "deploy") { + const n = existsSync(dir + "/n") ? Number(readFileSync(dir + "/n", "utf8")) : 0; + writeFileSync(dir + "/n", String(n + 1)); + const [code, text] = JSON.parse(readFileSync(dir + "/reports.json", "utf8"))[n]; + process.stderr.write(text); + process.exit(code); +} +process.exit(0); +`, + ); + const run = () => + spawnSync( + process.execPath, + [join(process.cwd(), "scripts/migrate-deploy.mjs"), "--schema", "x"], + { + encoding: "utf8", + env: { + ...process.env, + DATABASE_URL: "", + HEALTHLOG_PRISMA_CLI: cli, + HEALTHLOG_MIGRATIONS_DIR: migrations, + HEALTHLOG_MIGRATE_LOCK_RETRY_DELAYS: "0,0", + }, + }, + ); + const calls = () => + existsSync(join(dir, "calls.log")) + ? readFileSync(join(dir, "calls.log"), "utf8").trim().split("\n") + : []; + return { run, calls }; + } + + it("marks the timed-out migration rolled back and deploys again", () => { + const { run, calls } = setup( + [ + [1, P3018_LOCK], + [0, "All migrations have been successfully applied."], + ], + MARKED, + ); + const result = run(); + expect(result.status).toBe(0); + expect(calls()).toEqual([ + "migrate deploy --schema x", + "migrate resolve --rolled-back 0412_measurement_hot_updates --schema x", + "migrate deploy --schema x", + ]); + expect(result.stdout).toContain("timed out waiting for a lock (55P03)"); + }); + + it("gives up after the last delay with Prisma's exit code", () => { + const { run, calls } = setup( + [ + [1, P3018_LOCK], + [1, P3018_LOCK], + [1, P3018_LOCK], + ], + MARKED, + ); + const result = run(); + expect(result.status).toBe(1); + expect(calls().filter((c) => c.startsWith("migrate deploy"))).toHaveLength( + 3, + ); + expect(calls().filter((c) => c.startsWith("migrate resolve"))).toHaveLength( + 2, + ); + expect(result.stderr).toContain("giving up"); + }); + + it("stops at the first failure that is not a lock timeout", () => { + const { run, calls } = setup([[1, P3018_UNIQUE]], MARKED); + const result = run(); + expect(result.status).toBe(1); + expect(calls()).toEqual(["migrate deploy --schema x"]); + }); + + it("stops at a lock timeout of a migration that is not marked", () => { + const { run, calls } = setup([[1, P3018_LOCK]], UNMARKED); + const result = run(); + expect(result.status).toBe(1); + expect(calls()).toEqual(["migrate deploy --schema x"]); + expect(result.stderr).toContain("not marked safe to re-run"); + }); +}); diff --git a/scripts/migrate-deploy.mjs b/scripts/migrate-deploy.mjs new file mode 100644 index 000000000..23abb5f9a --- /dev/null +++ b/scripts/migrate-deploy.mjs @@ -0,0 +1,239 @@ +/** + * `prisma migrate deploy` for the container boot, with one narrow retry: a + * migration that gave up waiting for a lock. + * + * Run by `docker-entrypoint.sh` as `node scripts/migrate-deploy.mjs`. + * + * A migration that bounds its lock wait (`lock_timeout`) fails with SQLSTATE + * 55P03 when something holds the table longer, typically an autovacuum that + * is preventing wraparound and does not step aside. Prisma then records the + * attempt as failed, and every later `migrate deploy` stops at P3009 until an + * operator runs `prisma migrate resolve --rolled-back`: under a restart + * policy that is a container that never comes up. + * + * `migrate deploy` does not wrap a migration file in a transaction, so a + * failed file is not necessarily rolled back: statements before the one that + * failed stay applied. Marking such an attempt rolled back and running it + * again is only safe when the file was written for it. A migration declares + * that with a `-- @retry-on-lock-timeout` line, which its author adds only + * when every statement before the lock-taking one is harmless to repeat and + * the lock-taking statements form one statement (a `DO` block) that rolls + * back whole. `src/__tests__/migration-lock-timeout-guard.test.ts` holds the + * migrations to that. + * + * So: on a 55P03 failure of a migration carrying the marker, mark that one + * migration rolled back and run `migrate deploy` again after a backoff (15, + * 30, 60, 120, 240 seconds). Any other failure, a lock timeout of a migration + * without the marker, or a fifth lock timeout in a row, ends the boot with + * Prisma's exit code exactly as before. The failure is read from Prisma's + * own report (`Migration name:` and `Database error code:`), or, on a boot + * after a failed one (P3009), from the `logs` Prisma stored on the failed + * row in `_prisma_migrations`. + * + * Environment: + * HEALTHLOG_PRISMA_CLI Prisma's CLI entry (default: the image's copy) + * HEALTHLOG_MIGRATIONS_DIR where the migration files are (default: + * prisma/migrations under the working directory) + * HEALTHLOG_MIGRATE_LOCK_RETRY_DELAYS + * comma-separated backoff in seconds + * Arguments are passed through to every Prisma call (`--schema …`). + */ +import { spawn } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { pathToFileURL } from "node:url"; + +const DEFAULT_RETRY_DELAYS_S = [15, 30, 60, 120, 240]; + +/** The line a migration carries when a lock-timeout retry is safe for it. */ +export const RETRY_MARKER = /^--\s*@retry-on-lock-timeout\s*$/m; + +const LOCK_TIMEOUT = "55P03"; +const MIGRATION_NAME = /^[A-Za-z0-9_]+$/; + +/** + * The migration a P3018 report says failed on a lock timeout, or null. + * Prisma prints `Migration name: ` and `Database error code: ` + * for a migration that failed to apply. + */ +export function lockTimeoutInReport(text) { + if (typeof text !== "string") return null; + const name = /^Migration name: (\S+)\s*$/m.exec(text)?.[1]; + const code = /^Database error code: (\S+)\s*$/m.exec(text)?.[1]; + return name && code === LOCK_TIMEOUT ? name : null; +} + +/** Whether Prisma refused to start because a failed migration is on record. */ +function isFailedMigrationOnRecord(output) { + return /\bP3009\b/.test(output); +} + +/** + * Decide whether a failed `migrate deploy` is the one failure retried here. + * + * @param {object} input + * @param {string} input.output everything Prisma printed + * @param {Array<{migration_name: string, logs: string | null}> | null} input.failedRows + * the unfinished, not-rolled-back rows of `_prisma_migrations`, or null + * when they were not read + * @param {(name: string) => string | null} input.readMigration + * the migration's SQL, or null when the image does not ship it + * @returns {{ retry: true, migration: string } | { retry: false, reason: string }} + */ +export function decideLockRetry({ output, failedRows, readMigration }) { + let migration = lockTimeoutInReport(output); + if (!migration && isFailedMigrationOnRecord(output) && failedRows) { + // More than one failed row is not a state this boot can reason about. + if (failedRows.length === 1) { + migration = lockTimeoutInReport(failedRows[0].logs); + } + } + if (!migration) { + return { retry: false, reason: "not a lock timeout" }; + } + if (!MIGRATION_NAME.test(migration)) { + return { retry: false, reason: `unexpected migration name ${migration}` }; + } + const sql = readMigration(migration); + if (sql === null) { + return { retry: false, reason: `${migration} is not in this image` }; + } + if (!RETRY_MARKER.test(sql)) { + return { + retry: false, + reason: `${migration} timed out on a lock but is not marked safe to re-run`, + }; + } + return { retry: true, migration }; +} + +/** The backoff schedule, from the environment or the default. */ +export function retryDelays(raw) { + if (!raw) return DEFAULT_RETRY_DELAYS_S; + const delays = raw.split(",").map((part) => Number(part.trim())); + return delays.every((d) => Number.isFinite(d) && d >= 0) + ? delays + : DEFAULT_RETRY_DELAYS_S; +} + +function runPrisma(cli, args) { + return new Promise((resolve) => { + const child = spawn(process.execPath, [cli, ...args], { + stdio: ["ignore", "pipe", "pipe"], + }); + let output = ""; + child.stdout.on("data", (chunk) => { + output += chunk; + process.stdout.write(chunk); + }); + child.stderr.on("data", (chunk) => { + output += chunk; + process.stderr.write(chunk); + }); + child.on("error", (error) => { + output += String(error); + resolve({ code: 1, output }); + }); + child.on("close", (code) => resolve({ code: code ?? 1, output })); + }); +} + +/** The failed rows of `_prisma_migrations`, or null when they cannot be read. */ +async function readFailedRows(connectionString) { + if (!connectionString) return null; + try { + const { default: pg } = await import("pg"); + const url = new URL(connectionString); + const schema = url.searchParams.get("schema") || "public"; + url.searchParams.delete("schema"); + const client = new pg.Client({ + connectionString: url.toString(), + connectionTimeoutMillis: 5_000, + statement_timeout: 5_000, + }); + await client.connect(); + try { + const { rows } = await client.query( + `SELECT migration_name, logs FROM ${client.escapeIdentifier(schema)}._prisma_migrations + WHERE finished_at IS NULL AND rolled_back_at IS NULL`, + ); + return rows; + } finally { + await client.end().catch(() => {}); + } + } catch { + return null; + } +} + +function migrationReader(dir) { + return (name) => { + try { + return readFileSync(join(dir, name, "migration.sql"), "utf8"); + } catch { + return null; + } + }; +} + +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function main({ + cli = process.env.HEALTHLOG_PRISMA_CLI || + "/opt/prisma-cli/node_modules/prisma/build/index.js", + migrationsDir = process.env.HEALTHLOG_MIGRATIONS_DIR || + join(process.cwd(), "prisma", "migrations"), + delays = retryDelays(process.env.HEALTHLOG_MIGRATE_LOCK_RETRY_DELAYS), + extraArgs = process.argv.slice(2), +} = {}) { + const readMigration = migrationReader(migrationsDir); + for (let attempt = 0; ; attempt++) { + const deploy = await runPrisma(cli, ["migrate", "deploy", ...extraArgs]); + if (deploy.code === 0) return 0; + + const failedRows = isFailedMigrationOnRecord(deploy.output) + ? await readFailedRows(process.env.DATABASE_URL) + : null; + const decision = decideLockRetry({ + output: deploy.output, + failedRows, + readMigration, + }); + if (!decision.retry) { + if (decision.reason !== "not a lock timeout") { + console.error( + `HealthLog: not retrying the migration: ${decision.reason}.`, + ); + } + return deploy.code; + } + if (attempt >= delays.length) { + console.error( + `HealthLog: ERROR — migration ${decision.migration} timed out on a lock ` + + `${attempt + 1} times in a row; giving up. Something holds the table ` + + "for minutes: see docs/ops/deploy.md, " + + '"A migration that timed out on a lock".', + ); + return deploy.code; + } + const delay = delays[attempt]; + console.log( + `HealthLog: migration ${decision.migration} timed out waiting for a lock ` + + `(55P03) and changed nothing; marking it rolled back and retrying in ` + + `${delay}s (retry ${attempt + 1} of ${delays.length}).`, + ); + const resolve = await runPrisma(cli, [ + "migrate", + "resolve", + "--rolled-back", + decision.migration, + ...extraArgs, + ]); + if (resolve.code !== 0) return resolve.code; + await sleep(delay * 1000); + } +} + +if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) { + process.exitCode = await main(); +} diff --git a/src/__tests__/helpers/session-secret.ts b/src/__tests__/helpers/session-secret.ts new file mode 100644 index 000000000..d1db58feb --- /dev/null +++ b/src/__tests__/helpers/session-secret.ts @@ -0,0 +1,11 @@ +/** + * A session cookie secret of the exact shape `createSession` mints (`hls_` + * and 32 random bytes in hex). The session lookup refuses any other shape + * before it asks the database, so a fixture that invents a readable secret + * tests the refusal, not the lookup. + */ +import { randomBytes } from "node:crypto"; + +export function fixtureSessionSecret(): string { + return `hls_${randomBytes(32).toString("hex")}`; +} diff --git a/src/__tests__/job-queue-runtime-guard.test.ts b/src/__tests__/job-queue-runtime-guard.test.ts index 71dd3a2fe..48da4c47d 100644 --- a/src/__tests__/job-queue-runtime-guard.test.ts +++ b/src/__tests__/job-queue-runtime-guard.test.ts @@ -438,6 +438,28 @@ describe("heavy jobs share the lane", () => { expect(inverted).toEqual([]); }); + it("declares every heavy queue long, so its expiry, stop and exclusivity are proven", () => { + // A heavy job is the one most likely to run past pg-boss's default + // expiry, and the retry that follows an expiry runs it a second time. + // `long` is the declaration the checks above hold to an explicit expiry, + // a stop on the budget and a lock (or a named marker). A heavy queue + // that is genuinely bounded far below the default may be named here + // with its reason instead. + const HEAVY_BUT_SHORT: Record = {}; + const notLong = (table: Readonly>) => + Object.keys(HEAVY_QUEUES) + .filter((q) => table[q]?.runtime !== "long" && !(q in HEAVY_BUT_SHORT)) + .sort(); + expect(notLong(QUEUE_RUNTIME)).toEqual([]); + // The planted case: the fold as it was declared before, `short`. + expect( + notLong({ + ...QUEUE_RUNTIME, + "rollup-full-backfill": { runtime: "short", why: "four statements" }, + }), + ).toEqual(["rollup-full-backfill"]); + }); + it("defers only on queues where a job can go back to created", () => { // A deferred job returns from active to created. Under `short` (one // queued job per key) and `stately` (one per state) a twin already diff --git a/src/__tests__/list-cap-reader-guard.test.ts b/src/__tests__/list-cap-reader-guard.test.ts index 3c6752832..94c5517fa 100644 --- a/src/__tests__/list-cap-reader-guard.test.ts +++ b/src/__tests__/list-cap-reader-guard.test.ts @@ -6,8 +6,8 @@ * / `meta.limit`). The web client read only `data` (`apiGet`), so the signal * reached nobody: twelve routes sent it and no surface showed it. A capped * route is now bound to a reader, a production file outside the API tree - * that fetches the route's path through `apiGetList` / `apiPostList`, the - * one client reader of the cap, which a surface pairs with `ListCapNotice`. + * that fetches the route's path through `apiGetList`, the one client reader + * of the cap, which a surface pairs with `ListCapNotice`. * * The routes are derived from the tree (every route file that answers with * `apiListSuccess`), never listed by hand, so a newly capped route fails here @@ -48,7 +48,7 @@ function clientPathPattern(route: string): RegExp { ) .join("/"); return new RegExp( - `\\bapi(?:Get|Post)List\\s*<[\\s\\S]{0,200}?>\\s*\\(\\s*["\`]${body}(?:[?"\`])|\\bapi(?:Get|Post)List\\s*\\(\\s*["\`]${body}(?:[?"\`])`, + `\\bapiGetList\\s*<[\\s\\S]{0,200}?>\\s*\\(\\s*["\`]${body}(?:[?"\`])|\\bapiGetList\\s*\\(\\s*["\`]${body}(?:[?"\`])`, ); } diff --git a/src/__tests__/migration-lock-timeout-guard.test.ts b/src/__tests__/migration-lock-timeout-guard.test.ts new file mode 100644 index 000000000..6afe05a6e --- /dev/null +++ b/src/__tests__/migration-lock-timeout-guard.test.ts @@ -0,0 +1,128 @@ +/** + * Structural guard: how a migration bounds its lock wait, and when the boot + * may retry it. + * + * `prisma migrate deploy` runs a migration file statement by statement, each + * committing on its own; it does not wrap the file in a transaction. Two + * consequences were written into this repository the other way round: + * + * - A top-level `SET LOCAL lock_timeout` does nothing (there is no + * transaction for it to be local to), so the migration waits as long as + * the lock is held, and every write to the table queues behind it. + * - A file that fails half-way stays half-applied, so marking it rolled back + * and running it again can fail on what already exists, or apply a change + * twice. + * + * So a lock wait is bounded inside a `DO` block with + * `set_config('lock_timeout', …, true)`, which is one statement and rolls back + * whole; and the `-- @retry-on-lock-timeout` marker, which lets + * `scripts/migrate-deploy.mjs` retry a 55P03 failure on boot, is allowed only + * on a file whose every other statement is harmless to run again. + */ +import { readdirSync, readFileSync, existsSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; + +import { RETRY_MARKER } from "../../scripts/migrate-deploy.mjs"; + +const MIGRATIONS_DIR = join(process.cwd(), "prisma", "migrations"); + +function migrations(): Array<{ name: string; sql: string }> { + return readdirSync(MIGRATIONS_DIR, { withFileTypes: true }) + .filter((entry) => entry.isDirectory()) + .map((entry) => join(MIGRATIONS_DIR, entry.name, "migration.sql")) + .filter((file) => existsSync(file)) + .map((file) => ({ + name: file.split("/").at(-2) ?? file, + sql: readFileSync(file, "utf8"), + })); +} + +/** The statements of a file, comments removed, dollar-quoted bodies blanked. */ +function topLevelStatements(sql: string): string[] { + return sql + .replace(/--[^\n]*/g, "") + .replace(/\$([A-Za-z_]*)\$[\s\S]*?\$\1\$/g, "$BODY$") + .split(";") + .map((statement) => statement.trim().replace(/\s+/g, " ")) + .filter((statement) => statement.length > 0); +} + +/** Statements that leave the database as they found it when run twice. */ +const REPEATABLE = [ + /^DO \$BODY\$$/i, + /^CREATE OR REPLACE (FUNCTION|VIEW|PROCEDURE) /i, + /^CREATE (UNIQUE )?INDEX IF NOT EXISTS /i, + /^CREATE TABLE IF NOT EXISTS /i, + /^ALTER TABLE "?\w+"? ADD COLUMN IF NOT EXISTS /i, + /^DROP (INDEX|TRIGGER|FUNCTION|TABLE) IF EXISTS /i, +]; + +/** A session setting changed outside a transaction-scoped `DO` body. */ +const TOP_LEVEL_TIMEOUT = + /^SET (LOCAL |SESSION )?(lock_timeout|statement_timeout)\b|^SET LOCAL\b/i; + +function violations(name: string, sql: string): string[] { + const found: string[] = []; + const statements = topLevelStatements(sql); + for (const statement of statements) { + if (TOP_LEVEL_TIMEOUT.test(statement)) { + found.push(`${name}: top-level "${statement}" has no effect`); + } + } + const boundsLockWait = /set_config\(\s*'lock_timeout'/i.test(sql); + const marked = RETRY_MARKER.test(sql); + if (boundsLockWait && !marked) { + found.push( + `${name}: bounds its lock wait but lacks -- @retry-on-lock-timeout`, + ); + } + if (marked) { + for (const statement of statements) { + if (!REPEATABLE.some((pattern) => pattern.test(statement))) { + found.push( + `${name}: marked retry-safe, but "${statement.slice(0, 60)}" is not repeatable`, + ); + } + } + } + return found; +} + +describe("migration lock waits", () => { + it("bounds lock waits where they take effect and marks only repeatable files", () => { + const found = migrations().flatMap(({ name, sql }) => + violations(name, sql), + ); + expect(found, found.join("\n")).toEqual([]); + }); + + it("sees the one migration that bounds its wait today", () => { + const marked = migrations() + .filter(({ sql }) => RETRY_MARKER.test(sql)) + .map(({ name }) => name); + expect(marked).toContain("0412_measurement_hot_updates"); + }); + + it("catches each shape it exists to refuse", () => { + expect(violations("a", "SET LOCAL lock_timeout = '10s';")).toHaveLength(1); + expect( + violations( + "b", + "DO $x$ BEGIN PERFORM set_config('lock_timeout', '1s', true); END $x$;", + ), + ).toHaveLength(1); + expect( + violations( + "c", + "-- @retry-on-lock-timeout\nCREATE TRIGGER t BEFORE UPDATE ON m FOR EACH ROW EXECUTE FUNCTION f();\nDO $x$ BEGIN PERFORM set_config('lock_timeout', '1s', true); END $x$;", + ), + ).toEqual([expect.stringContaining("is not repeatable")]); + expect( + violations( + "d", + "-- @retry-on-lock-timeout\nCREATE OR REPLACE FUNCTION f() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RETURN NEW; END; $$;\nDO $x$ BEGIN PERFORM set_config('lock_timeout', '1s', true); END $x$;", + ), + ).toEqual([]); + }); +}); diff --git a/src/__tests__/provider-rounding-guard.test.ts b/src/__tests__/provider-rounding-guard.test.ts index 6c4b279c2..697af441e 100644 --- a/src/__tests__/provider-rounding-guard.test.ts +++ b/src/__tests__/provider-rounding-guard.test.ts @@ -14,22 +14,15 @@ import { readFileSync } from "node:fs"; import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { walkSourceFiles } from "./helpers/source-files"; +import { PROVIDER_DIRS } from "./helpers/provider-dirs"; +import { stripComments, walkSourceFiles } from "./helpers/source-files"; const LIB = join(process.cwd(), "src", "lib"); -const PROVIDER_DIRS = [ - "apple-health", - "dawarich", - "fitbit", - "google-health", - "nightscout", - "oura", - "polar", - "strava", - "whoop", - "withings", - "import", -]; +/** + * Every provider the product connects to, plus the file importers (Health + * Connect, the export parsers), which are not a connection of their own. + */ +const ROUNDING_DIRS = [...PROVIDER_DIRS, "import"]; const ROUNDING = /\.toFixed\s*\(|\b(?:function|const|let)\s+round2\b|Math\.round\s*\([^;]*\*\s*(10+)\s*\)\s*\/\s*\1\b/; @@ -41,16 +34,10 @@ const ALLOWED: Record = { "Prints a compression ratio into an error message.", }; -function strip(text: string): string { - return text - .replace(/\/\*[\s\S]*?\*\//g, "") - .replace(/(^|[^:])\/\/.*$/gm, "$1"); -} - describe("provider clients round through importValue", () => { const files = walkSourceFiles(LIB, { floor: 1000 }).filter( (f) => - PROVIDER_DIRS.some((d) => f.startsWith(`${d}/`)) && + ROUNDING_DIRS.some((d) => f.startsWith(`${d}/`)) && !f.includes("__tests__") && !f.endsWith(".test.ts"), ); @@ -63,7 +50,7 @@ describe("provider clients round through importValue", () => { const offenders = files.filter( (f) => !(f in ALLOWED) && - ROUNDING.test(strip(readFileSync(join(LIB, f), "utf8"))), + ROUNDING.test(stripComments(readFileSync(join(LIB, f), "utf8"))), ); expect(offenders).toEqual([]); }); diff --git a/src/app/api/account/activity/route.ts b/src/app/api/account/activity/route.ts index d72b9db82..365584600 100644 --- a/src/app/api/account/activity/route.ts +++ b/src/app/api/account/activity/route.ts @@ -153,6 +153,10 @@ export const GET = apiHandler(async () => { // The bound the view states out loud. Resolved here so the UI never has to // hardcode a number the operator can change. retentionDays, + // Deprecated copy of `meta.truncated`. It was a required field of `data` + // through v1.43.6, and a client built against that contract must keep + // decoding it; it goes in a minor release. The web reads `meta`. + truncated: cap.truncated, }, // Whether the list is everything inside that window, or its most recent // hundred rows, rides the envelope like every capped list's (`meta`). diff --git a/src/app/api/admin/import-apple-health-export/route.ts b/src/app/api/admin/import-apple-health-export/route.ts index ab4b03e51..798061698 100644 --- a/src/app/api/admin/import-apple-health-export/route.ts +++ b/src/app/api/admin/import-apple-health-export/route.ts @@ -55,6 +55,10 @@ export const POST = apiHandler(async (request: NextRequest) => { throw new HttpError(429, "Too many import uploads, try again later"); } + // A refused upload is an audit row here, not the signal the user-facing + // import routes emit for the same refusals: this is an administrator + // acting on another person's account, and every attempt at that, refused + // ones included, belongs in the trail an operator reviews. const declaredBytes = Number(request.headers.get("content-length") ?? 0); if (declaredBytes > MAX_UPLOAD_BYTES) { await auditLog("admin.import-apple-health.kickoff.denied", { diff --git a/src/app/api/auth/native/complete/__tests__/route.test.ts b/src/app/api/auth/native/complete/__tests__/route.test.ts index 540b12105..e82eeed96 100644 --- a/src/app/api/auth/native/complete/__tests__/route.test.ts +++ b/src/app/api/auth/native/complete/__tests__/route.test.ts @@ -45,6 +45,7 @@ import { GET } from "../route"; import { checkAuthSurfaceRateLimit } from "@/lib/rate-limit"; import { mintNativeHandoff } from "@/lib/auth/native-handoff"; import { prisma } from "@/lib/db"; +import { fixtureSessionSecret } from "@/__tests__/helpers/session-secret"; import { NATIVE_HANDOFF_STATE_COOKIE, encodeNativeHandoffState, @@ -71,7 +72,7 @@ function makeRequest(opts: { ); } if (opts.withSession !== false) { - r.cookies.set("healthlog_session", "hls_sessionsecret"); + r.cookies.set("healthlog_session", fixtureSessionSecret()); } return r; } diff --git a/src/app/api/auth/oidc/callback/__tests__/route.test.ts b/src/app/api/auth/oidc/callback/__tests__/route.test.ts index 370884fba..a01fa9212 100644 --- a/src/app/api/auth/oidc/callback/__tests__/route.test.ts +++ b/src/app/api/auth/oidc/callback/__tests__/route.test.ts @@ -101,6 +101,7 @@ import { verifyIdToken, } from "@/lib/auth/oidc"; import { mintNativeHandoff } from "@/lib/auth/oidc-native-handoff"; +import { fixtureSessionSecret } from "@/__tests__/helpers/session-secret"; const CONFIG = { issuerUrl: "https://idp.example.com", @@ -616,7 +617,8 @@ describe("GET /api/auth/oidc/callback — duplicate-callback idempotency", () => const req = new NextRequest( "http://localhost/api/auth/oidc/callback?code=abc&state=STATE", ); - if (withSession) req.cookies.set("healthlog_session", "hls_valid_secret"); + if (withSession) + req.cookies.set("healthlog_session", fixtureSessionSecret()); return req; } diff --git a/src/app/api/strava/credentials/__tests__/route.test.ts b/src/app/api/strava/credentials/__tests__/route.test.ts index f57870da6..a2e6b53d3 100644 --- a/src/app/api/strava/credentials/__tests__/route.test.ts +++ b/src/app/api/strava/credentials/__tests__/route.test.ts @@ -27,7 +27,10 @@ const { storeMock, clearMock } = vi.hoisted(() => ({ storeMock: vi.fn(), clearMock: vi.fn(), })); -vi.mock("@/lib/strava/credentials", () => ({ +vi.mock("@/lib/strava/credentials", async (importOriginal) => ({ + STRAVA_DISCONNECTED: ( + await importOriginal() + ).STRAVA_DISCONNECTED, storeStravaClientCredentials: storeMock, clearStravaClientCredentials: clearMock, })); @@ -80,14 +83,18 @@ describe("DELETE /api/strava/credentials", () => { expect(markDisconnected).toHaveBeenCalledWith("u1", "strava"); }); - it("clears the athlete id alongside both tokens", async () => { + it("clears the athlete id and the backfill state alongside both tokens", async () => { await del(); + // The backfill markers belong to the account that was connected; a later + // connection must start its own history walk from the newest activity. expect(userUpdate).toHaveBeenCalledWith( expect.objectContaining({ data: { stravaAccessTokenEncrypted: null, stravaRefreshTokenEncrypted: null, stravaAthleteId: null, + stravaBackfillCompletedAt: null, + stravaBackfillResumeBefore: null, }, }), ); diff --git a/src/app/api/strava/credentials/route.ts b/src/app/api/strava/credentials/route.ts index 7605775ef..30607cc1a 100644 --- a/src/app/api/strava/credentials/route.ts +++ b/src/app/api/strava/credentials/route.ts @@ -12,6 +12,7 @@ import { markDisconnected } from "@/lib/integrations/status"; import { storeStravaClientCredentials, clearStravaClientCredentials, + STRAVA_DISCONNECTED, } from "@/lib/strava/credentials"; import { stravaCredentialsSchema } from "@/lib/validations/strava"; import { NextRequest } from "next/server"; @@ -98,11 +99,7 @@ export const DELETE = apiHandler(async () => { await prisma.user.update({ where: { id: user.id }, - data: { - stravaAccessTokenEncrypted: null, - stravaRefreshTokenEncrypted: null, - stravaAthleteId: null, - }, + data: STRAVA_DISCONNECTED, }); await clearStravaClientCredentials(user.id); diff --git a/src/app/api/strava/disconnect/__tests__/route.test.ts b/src/app/api/strava/disconnect/__tests__/route.test.ts new file mode 100644 index 000000000..ee03d9dd0 --- /dev/null +++ b/src/app/api/strava/disconnect/__tests__/route.test.ts @@ -0,0 +1,63 @@ +/** + * `POST /api/strava/disconnect` leaves nothing of the connected account behind + * that a later connection would inherit: not the tokens, and not the backfill + * markers. A resume point left standing made the next account's first walk + * start below the old account's oldest activity, so its newer history was + * never asked for. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/api-handler", () => ({ + apiHandler: unknown>(fn: T) => fn, + requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), +})); +vi.mock("@/lib/db", () => ({ prisma: { user: { update: vi.fn() } } })); +vi.mock("@/lib/logging/context", () => ({ + annotate: vi.fn(), + getEvent: () => null, +})); +vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() })); +vi.mock("@/lib/integrations/status", () => ({ markDisconnected: vi.fn() })); +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(async () => ({ allowed: true })), + rateLimitHeaders: vi.fn(() => ({})), +})); +vi.mock("@/lib/strava/client", () => ({ deauthorize: vi.fn() })); +vi.mock("@/lib/strava/credentials", async (importOriginal) => ({ + STRAVA_DISCONNECTED: ( + await importOriginal() + ).STRAVA_DISCONNECTED, + getStravaConnection: vi.fn(async () => ({ accessToken: "tok" })), +})); +vi.mock("@/lib/api-response", () => ({ + apiSuccess: (data: unknown) => ({ data, error: null, status: 200 }), + apiError: (error: string, status: number) => ({ data: null, error, status }), +})); + +import { POST } from "../route"; +import { prisma } from "@/lib/db"; + +const userUpdate = prisma.user.update as ReturnType; + +beforeEach(() => { + vi.clearAllMocks(); + userUpdate.mockResolvedValue({}); +}); + +describe("POST /api/strava/disconnect", () => { + it("clears the backfill markers with the tokens", async () => { + const res = (await (POST as unknown as () => Promise<{ data: unknown }>)()) + .data; + expect(res).toEqual({ disconnected: true }); + expect(userUpdate).toHaveBeenCalledWith({ + where: { id: "u1" }, + data: { + stravaAccessTokenEncrypted: null, + stravaRefreshTokenEncrypted: null, + stravaAthleteId: null, + stravaBackfillCompletedAt: null, + stravaBackfillResumeBefore: null, + }, + }); + }); +}); diff --git a/src/app/api/strava/disconnect/route.ts b/src/app/api/strava/disconnect/route.ts index b41c4a75e..e3634b330 100644 --- a/src/app/api/strava/disconnect/route.ts +++ b/src/app/api/strava/disconnect/route.ts @@ -6,7 +6,10 @@ import { annotate, getEvent } from "@/lib/logging/context"; import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; import { markDisconnected } from "@/lib/integrations/status"; import { deauthorize } from "@/lib/strava/client"; -import { getStravaConnection } from "@/lib/strava/credentials"; +import { + getStravaConnection, + STRAVA_DISCONNECTED, +} from "@/lib/strava/credentials"; /** * v1.28.x — disconnect Strava for the current user. @@ -42,11 +45,7 @@ export const POST = apiHandler(async () => { await prisma.user.update({ where: { id: user.id }, - data: { - stravaAccessTokenEncrypted: null, - stravaRefreshTokenEncrypted: null, - stravaAthleteId: null, - }, + data: STRAVA_DISCONNECTED, }); await auditLog("strava.disconnect", { userId: user.id }); diff --git a/src/app/insights/ecg/page.tsx b/src/app/insights/ecg/page.tsx index 10cc0205a..7090cd437 100644 --- a/src/app/insights/ecg/page.tsx +++ b/src/app/insights/ecg/page.tsx @@ -41,9 +41,10 @@ export default function InsightsEcgPage() { const data = page?.data; // A read failure must be visible, not a heading over an empty body that - // reads the same as "still loading" (§6). The section teaser below unmounts - // silently on error, which is right for the overview but wrong for the page - // the user opened to see this list. + // reads the same as "still loading" (§6). The section shows a one-line + // `QueryErrorRow`, the right weight for a supplementary block on the + // overview; on the page the person opened to see this list the failure is + // the whole content, so it gets the full `QueryErrorCard` instead. if (isError) { return ( ( return { data, cap: readListCap(meta) }; } -/** POST variant of {@link apiGetList}, for a list read with a body. */ -export async function apiPostList( - path: string, - body?: unknown, - init?: RequestInit, -): Promise<{ data: T; cap: ListCap | null }> { - const { data, meta } = await apiFetchEnvelope( - path, - withDefaultIdempotencyKey(withJsonBody("POST", body, init)), - ); - return { data, cap: readListCap(meta) }; -} - /** GET — resolves with the unwrapped `data` payload. */ export function apiGet( path: string, diff --git a/src/lib/api/list-cap.ts b/src/lib/api/list-cap.ts index 98cd2436c..45155609d 100644 --- a/src/lib/api/list-cap.ts +++ b/src/lib/api/list-cap.ts @@ -18,7 +18,7 @@ * reads `meta` decodes it unchanged. `truncated` is exact: true only when a * row past the limit exists. The OpenAPI side is `listEnvelope` in * `src/lib/openapi/routes/shared.ts`. A client reads it through `apiGetList` - * / `apiPostList` (`src/lib/api/api-fetch.ts`) and shows it with + * (`src/lib/api/api-fetch.ts`) and shows it with * `ListCapNotice`; `src/__tests__/list-cap-reader-guard.test.ts` binds every * capped route to such a reader. */ diff --git a/src/lib/auth/__tests__/session.test.ts b/src/lib/auth/__tests__/session.test.ts index 3646fbda9..65d80fd7f 100644 --- a/src/lib/auth/__tests__/session.test.ts +++ b/src/lib/auth/__tests__/session.test.ts @@ -312,6 +312,42 @@ describe("a session lookup that fails is not a missing session", () => { ); }); + it("raises for the pool's own connect timeout, which carries no code", async () => { + // pg-pool rejects with exactly this message and nothing else; the driver + // adapter hands it through unchanged. + vi.mocked(prisma.session.findUnique).mockRejectedValueOnce( + new Error("timeout exceeded when trying to connect") as never, + ); + await expect(validateSessionFromCookieValue(SECRET)).rejects.toBeInstanceOf( + SessionStoreUnavailableError, + ); + }); + + it.each([ + ["53300", "sorry, too many clients already"], + ["55P03", "canceling statement due to lock timeout"], + ["53200", "out of memory"], + ])("raises for SQLSTATE %s, which is about the store", async (state, msg) => { + vi.mocked(prisma.session.findUnique).mockRejectedValueOnce( + Object.assign(new Error(msg), { + code: "P2010", + meta: { driverAdapterError: { cause: { originalCode: state } } }, + }) as never, + ); + await expect(validateSessionFromCookieValue(SECRET)).rejects.toBeInstanceOf( + SessionStoreUnavailableError, + ); + }); + + it("answers null for a Prisma validation refusal", async () => { + vi.mocked(prisma.session.findUnique).mockRejectedValueOnce( + Object.assign(new Error("Invalid value provided"), { + name: "PrismaClientValidationError", + }) as never, + ); + await expect(validateSessionFromCookieValue(SECRET)).resolves.toBe(null); + }); + it("answers null, not 503, when the query refused the value itself", async () => { // Postgres 22021: a byte sequence it will not accept in text. Nothing is // wrong with the store; a 503 here would let any anonymous caller produce diff --git a/src/lib/auth/session.ts b/src/lib/auth/session.ts index d779fa9c5..72dc0e0af 100644 --- a/src/lib/auth/session.ts +++ b/src/lib/auth/session.ts @@ -6,7 +6,7 @@ import { hashToken } from "@/lib/auth/hmac"; import { ensureDbCompatibility } from "@/lib/db-compat"; import { getEvent } from "@/lib/logging/context"; import { shouldEmitSecureCookie } from "@/lib/auth/secure-cookie"; -import { isDatabaseUnavailable, isP2025 } from "@/lib/prisma-errors"; +import { isInputRefusal, isP2025 } from "@/lib/prisma-errors"; import { SessionStoreUnavailableError } from "@/lib/api-errors"; import { locales, type Locale } from "@/lib/i18n/config"; import { LOCALE_COOKIE, setLocaleCookie } from "@/lib/i18n/locale-cookie"; @@ -116,11 +116,13 @@ async function findSessionByCookie(cookieValue: string) { // signed its person out. It is a 503 now, with a code a client can retry // on (`auth.session.unavailable`). // - // Only for a store that could not be reached. Any other query error says - // something about the value asked for, not about the store, and is - // answered like a value that matches nothing: a 503 there would let any - // anonymous caller produce server errors at will. - if (isDatabaseUnavailable(error)) { + // The shape check above already refused every crafted value before the + // query, so a failure here is about the store, not the caller: the pool + // had no connection free, the server ran out of connections or memory, a + // lock wait ran out. All of those are a 503. Only a refusal that names + // the value itself (a data exception, a Prisma validation error) is + // answered like a value that matches nothing. + if (!isInputRefusal(error)) { throw new SessionStoreUnavailableError(error); } getEvent()?.addWarning( diff --git a/src/lib/day/__tests__/notable-history-cache.test.ts b/src/lib/day/__tests__/notable-history-cache.test.ts index 256ded2e3..7fb0e9843 100644 --- a/src/lib/day/__tests__/notable-history-cache.test.ts +++ b/src/lib/day/__tests__/notable-history-cache.test.ts @@ -132,6 +132,42 @@ describe("notable history cache", () => { }); }); +describe("a write while the history is being read", () => { + it("is not hidden by the read that started before it", async () => { + // The first read is held until a write lands and invalidates the + // account's analytics entries, as `invalidateUserMeasurements` does. + const real = readLocalDailyStats.getMockImplementation()!; + let release!: () => void; + const held = new Promise((r) => (release = r)); + readLocalDailyStats.mockImplementationOnce(async (args) => { + const before = await real(args); + await held; + return before; + }); + const stale = load(SPIKE); + await new Promise((r) => setTimeout(r, 0)); + // The write lowers the spike, then evicts. + const original = real; + readLocalDailyStats.mockImplementation(async (args) => { + const stats = (await original(args)) as Map>; + stats.get("BLOOD_PRESSURE_SYS")?.set(SPIKE, 120); + return stats; + }); + caches.analytics.deleteByPrefix("user-1|"); + release(); + // The read already under way answers with what it saw. + expect((await stale).map((n) => n.kind)).toEqual(["extremeHigh"]); + // The next one reads again and sees the write. + expect(await load(SPIKE)).toEqual([]); + }); + + it("joins a read of the same block already under way", async () => { + await Promise.all([load(SPIKE), load(SPIKE)]); + // One read of the missing span, shared by both. + expect(readLocalDailyStats).toHaveBeenCalledTimes(1); + }); +}); + describe("notable history cache size", () => { const cached: number[] = []; beforeEach(() => { diff --git a/src/lib/day/notable.ts b/src/lib/day/notable.ts index 32ab02261..2fb50b3af 100644 --- a/src/lib/day/notable.ts +++ b/src/lib/day/notable.ts @@ -274,26 +274,43 @@ async function readExtremeHistory(args: { }); } - const found = new Map(); - const missing = blocks.filter((block) => { - const hit = cache.get(block.key); - if (hit) found.set(block.key, hit); - return !hit; - }); - if (missing.length > 0) { - const first = missing[0]!; - const last = missing[missing.length - 1]!; - const stats = await read( - startOfLocalDayKey(first.start, args.tz), - last.end, + // Every block goes through `wrap`, so a build is coalesced with a + // concurrent read of the same block and, above all, is fenced: a write + // that invalidates the account's analytics entries while the read is in + // flight stops the result from being cached, instead of the pre-write + // days being served for the block's whole TTL. The blocks missing right + // now still share one read of their contiguous span. + const missing = blocks.filter((block) => cache.get(block.key) === null); + let span: Promise | null = null; + const readSpan = (): Promise => { + span ??= read( + startOfLocalDayKey(missing[0]!.start, args.tz), + missing[missing.length - 1]!.end, ); - for (const block of blocks) { - if (block.start < first.start || block.end > last.end) continue; - const cell = sliceDays(stats, block.start, block.end); - cache.set(block.key, cell, HISTORY_TTL_MS); - found.set(block.key, cell); - } - } + return span; + }; + const inSpan = (block: { start: string; end: string }) => + missing.length > 0 && + block.start >= missing[0]!.start && + block.end <= missing[missing.length - 1]!.end; + const cells = await Promise.all( + blocks.map(async (block) => { + const { value } = await cache.wrap( + block.key, + async () => + sliceDays( + inSpan(block) + ? await readSpan() + : await read(startOfLocalDayKey(block.start, args.tz), block.end), + block.start, + block.end, + ), + HISTORY_TTL_MS, + ); + return [block.key, value] as const; + }), + ); + const found = new Map(cells); const out: DailyStats = new Map(); for (const block of blocks) { diff --git a/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts b/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts index ecaba31c6..4054883b7 100644 --- a/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts +++ b/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts @@ -152,6 +152,22 @@ describe("a resumable walk", () => { expect(budget.stopReason).toBe("rate_limited"); }); + it("gives way to a waiting heavy job after its first page, keeping its place", async () => { + safeFetchMock.mockImplementation(async (url: string) => + page(Number(new URL(url).searchParams.get("pageToken") ?? 0), true), + ); + const budget = googleHealthSliceBudget(60_000, 500, Date.now, () => true); + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk({ resumable: true }), + budget, + ); + // Never before a page is on disk: a slice that ended empty would fail. + expect(safeFetchMock).toHaveBeenCalledTimes(1); + expect(outcome.truncated).toBe(true); + expect(budget.stopReason).toBe("lane_wanted"); + }); + it("still fails on a rate limit outside a slice", async () => { safeFetchMock.mockResolvedValue( page(0, true, 429, { "retry-after": "3600" }), @@ -178,6 +194,20 @@ describe("a walk that is not resumable", () => { expect(outcome.truncated).toBe(false); }); + it("does not give way part-way, since it would start over", async () => { + safeFetchMock.mockImplementation(async (url: string) => { + const n = Number(new URL(url).searchParams.get("pageToken") ?? 0); + return page(n, n < 3); + }); + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk(), + googleHealthSliceBudget(60_000, 500, Date.now, () => true), + ); + expect(safeFetchMock).toHaveBeenCalledTimes(4); + expect(outcome.truncated).toBe(false); + }); + it("still stops on the job's own budget", async () => { safeFetchMock.mockImplementation(async () => page(0, true)); let asked = 0; diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index d12006fc1..a0e3a7aa4 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -155,7 +155,7 @@ interface GoogleHealthWalkLimits { /** Why a backfill slice stopped walking. */ export type GoogleHealthSliceStopReason = - "time_budget" | "page_budget" | "rate_limited" | "job_budget"; + "time_budget" | "page_budget" | "rate_limited" | "job_budget" | "lane_wanted"; /** * One backfill slice's share of the work: a span of wall-clock time and a @@ -184,6 +184,14 @@ export interface GoogleHealthSliceBudget { skippedPages: number; stopReason: GoogleHealthSliceStopReason | null; now: () => number; + /** + * Whether another heavy job waits for the lane this slice holds + * (`laneWanted` in `jobs/heavy-lane.ts`). Asked where the slice's own + * limits are, so giving way keeps every walk's place, and only once the + * slice has read a page, so the slice it ends has something on disk and is + * continued rather than failed. + */ + yieldWanted: () => boolean; } /** A fresh slice budget of `ms` milliseconds and `maxPages` pages. */ @@ -191,6 +199,7 @@ export function googleHealthSliceBudget( ms: number, maxPages: number, now: () => number = Date.now, + yieldWanted: () => boolean = () => false, ): GoogleHealthSliceBudget { return { ms, @@ -200,6 +209,7 @@ export function googleHealthSliceBudget( skippedPages: 0, stopReason: null, now, + yieldWanted, }; } @@ -212,6 +222,8 @@ function sliceExhausted(slice: GoogleHealthSliceBudget): boolean { if (slice.now() >= slice.deadline) slice.stopReason = "time_budget"; else if (slice.pages - slice.skippedPages >= slice.maxPages) { slice.stopReason = "page_budget"; + } else if (slice.pages > 0 && slice.yieldWanted()) { + slice.stopReason = "lane_wanted"; } return slice.stopReason !== null; } diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index 3fd42cdc0..8a8d33ba1 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -133,6 +133,10 @@ import { runGoogleHealthBackfillForUser, } from "../google-health-backfill"; import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission"; +import { + LANE_DEFER_SECONDS, + LANE_YIELD_DELAY_SECONDS, +} from "@/lib/jobs/heavy-lane"; import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys"; import { GOOGLE_HEALTH_DATA_TYPES, @@ -472,6 +476,35 @@ describe("runGoogleHealthBackfillForUser — slices", () => { }); }); + it("queues the next slice past the waiting job's return after giving way", async () => { + resumeState.advances = ["fetchHeartRate"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true }); + const client = await import("@/lib/google-health/client"); + const spy = vi + .spyOn(client, "googleHealthSliceBudget") + .mockImplementation(() => ({ + ms: 300_000, + deadline: null, + maxPages: 500, + pages: 3, + skippedPages: 0, + stopReason: "lane_wanted", + now: Date.now, + yieldWanted: () => true, + })); + try { + await expect(runGoogleHealthBackfillForUser("u1")).resolves.toMatchObject( + { continued: true }, + ); + } finally { + spy.mockRestore(); + } + expect(bossSend.mock.calls[0]![2]).toMatchObject({ + startAfter: LANE_YIELD_DELAY_SECONDS, + }); + expect(LANE_YIELD_DELAY_SECONDS).toBeGreaterThan(LANE_DEFER_SECONDS); + }); + it("waits longer before the next slice when this one ended on a rate limit", async () => { resumeState.advances = ["fetchHeartRate"]; syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true }); @@ -488,6 +521,7 @@ describe("runGoogleHealthBackfillForUser — slices", () => { skippedPages: 0, stopReason: "rate_limited", now: Date.now, + yieldWanted: () => false, })); try { await runGoogleHealthBackfillForUser("u1"); diff --git a/src/lib/jobs/__tests__/heavy-lane.test.ts b/src/lib/jobs/__tests__/heavy-lane.test.ts index 0b97297ed..6b1bed63b 100644 --- a/src/lib/jobs/__tests__/heavy-lane.test.ts +++ b/src/lib/jobs/__tests__/heavy-lane.test.ts @@ -8,8 +8,10 @@ import type { Job } from "pg-boss"; vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); vi.mock("@/lib/db", () => ({ prisma: {} })); +vi.mock("@/lib/logging/signal", () => ({ emitSignal: vi.fn() })); -import { heavyLane } from "../heavy-lane"; +import { emitSignal } from "@/lib/logging/signal"; +import { heavyLane, LANE_WAIT_WARN_MS, laneWanted } from "../heavy-lane"; import { jobDone } from "../job-outcome"; type TestJob = Job<{ force?: boolean }>; @@ -36,7 +38,7 @@ function deferrals() { const back: Array<{ queue: string; ids: string[] }> = []; const defer = vi.fn(async (queue: string, jobs: readonly TestJob[]) => { back.push({ queue, ids: jobs.map((j) => j.id) }); - return jobs.length; + return { deferred: jobs.length, oldestCreatedOn: null }; }); return { back, defer }; } @@ -160,3 +162,107 @@ describe("heavyLane", () => { ).toThrow(/not in HEAVY_QUEUES/); }); }); + +describe("laneWanted", () => { + it("tells the holder, and only the holder, that another queue is waiting", async () => { + const seen: Record = { holder: [], beside: [] }; + let release!: () => void; + const released = new Promise((r) => (release = r)); + const { defer } = deferrals(); + expect(laneWanted()).toBe(false); + const held = heavyLane( + "integration-backfill-admission", + async () => { + seen.holder.push(laneWanted()); + await released; + seen.holder.push(laneWanted()); + return jobDone({ jobs: 1 }); + }, + undefined, + defer, + )([job("slice")]); + await tick(); + // A job of the same queue going back is not a reason to give way. + await heavyLane( + "integration-backfill-admission", + vi.fn(), + undefined, + defer, + )([job("next-slice")]); + expect(laneWanted()).toBe(false); + await heavyLane( + "insights-general-status", + vi.fn(), + undefined, + defer, + )([job("nightly")]); + // A light job of a heavy queue running beside the lane is never asked, even now. + await heavyLane( + "insight-pregenerate", + async () => { + seen.beside.push(laneWanted()); + return jobDone({ jobs: 1 }); + }, + () => false, + defer, + )([job("warm")]); + release(); + await held; + expect(seen.holder).toEqual([false, true]); + expect(seen.beside).toEqual([false]); + // The next holder starts with a clean slate. + await heavyLane( + "data-backup", + async () => { + seen.holder.push(laneWanted()); + return jobDone({ jobs: 1 }); + }, + undefined, + defer, + )([job("weekly")]); + expect(seen.holder).toEqual([false, true, false]); + }); +}); + +describe("a long wait for the lane", () => { + it("is reported at warn once per holder when a job has waited past the limit", async () => { + vi.mocked(emitSignal).mockClear(); + const now = Date.parse("2026-10-20T06:00:00Z"); + let release!: () => void; + const released = new Promise((r) => (release = r)); + const queuedAt = (ms: number) => async () => ({ + deferred: 1, + oldestCreatedOn: new Date(now - ms), + }); + const held = heavyLane("integration-backfill-admission", async () => { + await released; + return jobDone({ jobs: 1 }); + })([job("import")]); + await tick(); + const wait = (ms: number) => + heavyLane( + "insights-general-status", + vi.fn(), + undefined, + queuedAt(ms), + () => now, + )([job("nightly")]); + + await wait(LANE_WAIT_WARN_MS - 60_000); + expect(emitSignal).not.toHaveBeenCalled(); + await wait(LANE_WAIT_WARN_MS + 60_000); + await wait(LANE_WAIT_WARN_MS + 120_000); + expect(emitSignal).toHaveBeenCalledTimes(1); + expect(emitSignal).toHaveBeenCalledWith({ + action: "job.heavy_lane.waiting_long", + level: "warn", + meta: { + queue: "insights-general-status", + holder: "integration-backfill-admission", + waited_minutes: 121, + }, + }); + release(); + await held; + }); +}); diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index 82b30ed07..d070872de 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -38,9 +38,11 @@ import { import { GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock"; import { googleHealthSliceBudget, + type GoogleHealthSliceStopReason, runWithGoogleHealthStop, } from "@/lib/google-health/client"; import { reportJobProgress } from "@/lib/jobs/job-observer"; +import { LANE_YIELD_DELAY_SECONDS, laneWanted } from "@/lib/jobs/heavy-lane"; import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys"; export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill"; @@ -122,6 +124,21 @@ export function googleHealthBackfillRetryKey( return `${identity}|retry-${new Date(dueAt).toISOString()}`; } +/** + * The pause before the next slice, by why this one stopped: longer after a + * rate limit, and past the waiting job's return after giving way to it, so + * that job takes the lane before this import does again. + */ +export function nextSliceDelaySeconds( + stopReason: GoogleHealthSliceStopReason | null, +): number { + if (stopReason === "rate_limited") { + return GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS; + } + if (stopReason === "lane_wanted") return LANE_YIELD_DELAY_SECONDS; + return INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS; +} + /** Test seam: a slice's budget, when not the production one. */ export interface GoogleHealthBackfillSliceOptions { sliceMs?: number; @@ -240,6 +257,8 @@ export async function runGoogleHealthBackfillForUser( const budget = googleHealthSliceBudget( sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS, sliceOptions.slicePages ?? GOOGLE_HEALTH_BACKFILL_SLICE_PAGES, + Date.now, + laneWanted, ); const startedAt = Date.now(); const { @@ -387,9 +406,7 @@ export async function runGoogleHealthBackfillForUser( const progressed = finished > 0 || advanced.size > 0 || newlyParked > 0; if (progressed) { const sent = await sendNextSlice( - budget.stopReason === "rate_limited" - ? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS - : INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + nextSliceDelaySeconds(budget.stopReason), ); if (sent) { annotate({ diff --git a/src/lib/jobs/heavy-lane.ts b/src/lib/jobs/heavy-lane.ts index 0f924880d..1437e99b8 100644 --- a/src/lib/jobs/heavy-lane.ts +++ b/src/lib/jobs/heavy-lane.ts @@ -23,12 +23,24 @@ * longer than its own expiry was failed, stopped counting as in use, and had * its staged file swept from under it. A deferred job keeps its id, its * retries and its staged upload, and its expiry starts when it runs. + * + * A deferred job also leaves word that the lane is wanted (`laneWanted`). A + * holder that runs in slices asks it at the boundaries where it can already + * stop and keep its place, ends its slice there, and queues the rest a little + * later than the job that waited (`LANE_YIELD_DELAY_SECONDS`), so a long + * import makes room for the nightly runs and for an upload instead of holding + * them back for hours. A holder that cannot stop part-way runs to its end; a + * job that has stood in line for `LANE_WAIT_WARN_MS` is reported at `warn` + * (`job.heavy_lane.waiting_long`), once per holder, so a lane that is held + * for hours shows in the alerts instead of only in `info` lines. */ +import { AsyncLocalStorage } from "node:async_hooks"; import type { Job } from "pg-boss"; import { prisma } from "@/lib/db"; import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { annotate } from "@/lib/logging/context"; +import { emitSignal } from "@/lib/logging/signal"; export interface HeavyQueue { /** What makes the job heavy. */ @@ -73,21 +85,39 @@ export const HEAVY_QUEUES: Readonly> = { /** How far ahead a deferred job is put back in the queue. */ export const LANE_DEFER_SECONDS = 60; +/** + * How far ahead a holder that gave way queues the rest of its work: past the + * deferred job's own return, so the job it gave way to takes the lane first. + */ +export const LANE_YIELD_DELAY_SECONDS = 2 * LANE_DEFER_SECONDS; + +/** How long a job may stand in line before the wait is reported at `warn`. */ +export const LANE_WAIT_WARN_MS = 2 * 60 * 60 * 1000; + +/** What one trip back to the queue did. */ +export interface LaneDeferral { + /** Rows that went back. */ + deferred: number; + /** When the oldest of them was first queued, if any went back. */ + oldestCreatedOn: Date | null; +} + /** * Put `jobs` of `queue` back in the queue to run `seconds` from now. The same * reset pg-boss's own `restore()` makes, with `start_after` moved in the same * statement: a restore followed by a separate update leaves a moment in which * another fetch could claim the row, and the completion of this handler would * then settle that newer claim, since a restored job's next claim keeps its - * attempt number. Answers how many rows went back. + * attempt number. Answers how many rows went back, and since when the oldest + * of them has been waiting. */ export async function deferLaneJobs( queue: string, jobs: readonly Job[], seconds: number = LANE_DEFER_SECONDS, -): Promise { +): Promise { const ids = jobs.map((job) => job.id); - return prisma.$executeRaw` + const rows = await prisma.$queryRaw>` UPDATE pgboss.job SET state = 'created', started_on = NULL, @@ -97,42 +127,102 @@ export async function deferLaneJobs( WHERE name = ${queue} AND id = ANY(${ids}::uuid[]) AND state = 'active' + RETURNING created_on `; + const oldest = rows.reduce( + (min, row) => (min === null || row.created_on < min ? row.created_on : min), + null, + ); + return { deferred: rows.length, oldestCreatedOn: oldest }; } -let busy = false; +/** The job holding the lane, and what has happened while it held it. */ +interface LaneHold { + queue: string; + /** The first other queue that found the lane taken during this hold. */ + wantedBy: string | null; + /** Queues whose long wait has been reported during this hold. */ + warned: Set; +} + +let hold: LaneHold | null = null; +const holdScope = new AsyncLocalStorage(); + +/** + * Whether the caller holds the lane and a job of another queue has been sent + * back while it did. False outside the lane, so a job that runs beside the + * lane (a light job of a heavy queue, any other queue) never gives way. + */ +export function laneWanted(): boolean { + const current = holdScope.getStore(); + return current !== undefined && current === hold && current.wantedBy !== null; +} + +function reportLongWait( + current: LaneHold, + queue: string, + deferral: LaneDeferral, + now: number, +): void { + if (!deferral.oldestCreatedOn || current.warned.has(queue)) return; + const waitedMs = now - deferral.oldestCreatedOn.getTime(); + if (waitedMs < LANE_WAIT_WARN_MS) return; + current.warned.add(queue); + emitSignal({ + action: "job.heavy_lane.waiting_long", + level: "warn", + meta: { + queue, + holder: current.queue, + waited_minutes: Math.floor(waitedMs / 60_000), + }, + }); +} /** * Wrap `handler` so the jobs of `queue` run in the lane. `inLane` picks the * jobs that are heavy where a queue carries light ones too (a single * account's warm beside the nightly walk); a batch with none of them runs at - * once. `defer` is the way back to the queue; tests replace it. + * once. `defer` is the way back to the queue; tests replace it, and `now`. */ export function heavyLane>( queue: string, handler: (jobs: J[]) => Promise, inLane: (job: J) => boolean = () => true, - defer: (queue: string, jobs: readonly J[]) => Promise = deferLaneJobs, + defer: ( + queue: string, + jobs: readonly J[], + ) => Promise = deferLaneJobs, + now: () => number = Date.now, ): (jobs: J[]) => Promise { if (!HEAVY_QUEUES[queue]) { throw new Error(`heavyLane: ${queue} is not in HEAVY_QUEUES`); } return async (jobs) => { if (!jobs.some(inLane)) return handler(jobs); - if (busy) { - const deferred = await defer(queue, jobs); + const current = hold; + if (current) { + const deferral = await defer(queue, jobs); + if (queue !== current.queue) current.wantedBy ??= queue; annotate({ action: { name: "job.heavy_lane.deferred" }, - meta: { queue, jobs: jobs.length, deferred }, + meta: { + queue, + holder: current.queue, + jobs: jobs.length, + deferred: deferral.deferred, + }, }); - return jobDone({ lane_deferred: deferred }); + reportLongWait(current, queue, deferral, now()); + return jobDone({ lane_deferred: deferral.deferred }); } - busy = true; + const entered: LaneHold = { queue, wantedBy: null, warned: new Set() }; + hold = entered; annotate({ action: { name: "job.heavy_lane.entered" }, meta: { queue } }); try { - return await handler(jobs); + return await holdScope.run(entered, () => handler(jobs)); } finally { - busy = false; + hold = null; } }; } diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index e4922f110..a5546c376 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -42,6 +42,7 @@ import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-int import { HEALTH_CONNECT_IMPORT_SEND_OPTIONS } from "@/lib/jobs/health-connect-import-worker"; import { MEASUREMENT_MAINTENANCE_SEND_OPTIONS } from "@/lib/jobs/measurement-maintenance"; import { PULSE_DAY_MEAN_SEND_OPTIONS } from "@/lib/jobs/pulse-day-mean"; +import { ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS } from "@/lib/rollups/measurement-rollups"; /** pg-boss's default job expiry, in seconds. */ export const DEFAULT_JOB_EXPIRE_SECONDS = 15 * 60; @@ -190,9 +191,14 @@ export const QUEUE_RUNTIME: Readonly> = { // ── Rollups ──────────────────────────────────────────────────────── "rollup-recompute": short("One bucket."), - "rollup-full-backfill": short( - "One account's rollup tiers: four aggregate statements, each held to the 60-second statement timeout.", - ), + "rollup-full-backfill": { + runtime: "long", + expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, + expiryVia: ["ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS"], + stop: "binding", + exclusive: "lockedPass", + why: "One account's five-year fold: four aggregates under the statement timeout and four persist transactions of up to 120 s each, about thirteen minutes at worst under the defaults and unbounded when the operator lifts the statement timeout.", + }, "mood-rollup-recompute": short("One bucket."), "mood-rollup-full-backfill": short( "One account's mood entries, a few a day.", diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index e779f0f48..27534267c 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -109,6 +109,7 @@ import { BOOT_BACKFILL_STAGGER_SECONDS, } from "./shared"; import { heavyLane } from "@/lib/jobs/heavy-lane"; +import { lockedPass } from "@/lib/jobs/long-pass"; import { createAndSchedule, createAndWork, @@ -332,22 +333,37 @@ export async function registerRollupQueues( boss, ROLLUP_FULL_BACKFILL_QUEUE, { localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY }, - heavyLane(ROLLUP_FULL_BACKFILL_QUEUE, async (jobs) => { - let rowsUpsertedTotal = 0; - for (const job of jobs) { - const { userId } = job.data; - const { rowsUpserted, durationMs } = await recomputeUserRollups(userId); - rowsUpsertedTotal += rowsUpserted; - workerLog( - "info", - `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, - ); - } - return jobDone({ - users: jobs.length, - rows_upserted: rowsUpsertedTotal, - }); - }), + // The lock sits outside the lane: a retry pg-boss starts after the expiry + // while the first fold still runs finds the account's lock and settles, + // instead of waiting in the lane and folding the account again. + lockedPass( + ROLLUP_FULL_BACKFILL_QUEUE, + (job) => job.data.userId, + heavyLane(ROLLUP_FULL_BACKFILL_QUEUE, async (jobs) => { + const shouldStop = jobBudget(jobs); + let rowsUpsertedTotal = 0; + for (const job of jobs) { + const { userId } = job.data; + const { rowsUpserted, durationMs, stopped } = + await recomputeUserRollups(userId, { shouldStop }); + if (stopped) { + // Each granularity committed on its own; the retry folds again. + throw new Error( + `rollup-full-backfill for user ${userId} stopped on its job budget; the retry folds the rest`, + ); + } + rowsUpsertedTotal += rowsUpserted; + workerLog( + "info", + `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, + ); + } + return jobDone({ + users: jobs.length, + rows_upserted: rowsUpsertedTotal, + }); + }), + ), ); // v1.5.6 — legacy step consolidation worker. The boot enqueue helper diff --git a/src/lib/medications/intake-import-admission.ts b/src/lib/medications/intake-import-admission.ts index abd0d4197..591140da9 100644 --- a/src/lib/medications/intake-import-admission.ts +++ b/src/lib/medications/intake-import-admission.ts @@ -171,6 +171,10 @@ export async function admitIntakeImportJob({ completedAt, }, }); + // An audit row, unlike the upload refusals of the export imports, which + // are signals: this one is shown to the record's owner in the activity + // feed (`record-activity/activity-verb.ts`) as an import that was + // admitted and never ran, and it carries the delegate who started it. await auditLog("medication.intake.import.kickoff.denied", { userId: recordUserId, actorUserId, diff --git a/src/lib/openapi/routes/account-sharing.ts b/src/lib/openapi/routes/account-sharing.ts index d0bc27ec7..a42bc4a88 100644 --- a/src/lib/openapi/routes/account-sharing.ts +++ b/src/lib/openapi/routes/account-sharing.ts @@ -354,6 +354,11 @@ const recordActivityResponse = z .describe( "How far back this feed can see, in days. Resolved from the instance's audit retention setting (365 by default, operator-configurable), NOT a constant — an instance running a 90-day window sends 90. The query is bounded to this window, so it is the real limit of the list and not only a caption. Show it beside the rows.", ), + truncated: z.boolean().meta({ + deprecated: true, + description: + "Deprecated: read `meta.truncated`, which carries the same value. Whether the list stopped at its row ceiling; kept in `data` for clients built against v1.43.6 and earlier.", + }), }) .meta({ id: "RecordActivity", @@ -870,7 +875,7 @@ export const accountSharingPaths: NonNullable = { tags: ["Account sharing"], summary: "Who opened this record, and when", description: - "Every audit row filed under this account that somebody else performed, newest first — one row per delegate per day with a count, rather than one per request. `retentionDays` rides the response because audit history is purged on a schedule the operator configures: a view that showed the rows without the window would imply a completeness it does not have, and clients must state the number the server sends rather than assume the 365-day default. Whether the list stopped at its row ceiling is `meta.truncated` (it was a `truncated` field in `data` before v1.43.7). Not delegable: a grant is not an introduction to the other people in the household.", + "Every audit row filed under this account that somebody else performed, newest first — one row per delegate per day with a count, rather than one per request. `retentionDays` rides the response because audit history is purged on a schedule the operator configures: a view that showed the rows without the window would imply a completeness it does not have, and clients must state the number the server sends rather than assume the 365-day default. Whether the list stopped at its row ceiling is `meta.truncated`; the `truncated` field in `data` carries the same value and is deprecated. Not delegable: a grant is not an introduction to the other people in the household.", responses: { ...stdResponses, "200": { diff --git a/src/lib/prisma-errors.ts b/src/lib/prisma-errors.ts index d5612ea8e..96e07e2e4 100644 --- a/src/lib/prisma-errors.ts +++ b/src/lib/prisma-errors.ts @@ -41,23 +41,46 @@ export function isP2025(err: unknown): boolean { } /** - * Whether a query failed because the database could not be reached, as opposed - * to refusing what it was asked. The first is worth a retry and is a 503; the - * second is an answer about the input. - * - * Reads the same signals the restore classifier does (`connection` and - * `timeout`: a connection SQLSTATE, a server shutting down, a statement or - * idle timeout, Prisma's P1001 / P1002 / P1017 / P2028, a terminated - * connection), plus the two that only a pooled request meets: P2024, no - * connection free in the pool within its timeout, and P1008, an operation that - * timed out. + * Prisma's own codes for a query it refused because of the value it was given: + * a value too long for its column (P2000), an invalid value or type for a + * field (P2006, P2007, P2023), and a query that failed validation (P2009, + * P2012, P2019, P2020). */ -export function isDatabaseUnavailable(err: unknown): boolean { - if (hasPrismaCode(err, "P2024") || hasPrismaCode(err, "P1008")) return true; - const { cause } = classifyRestoreFailure(err); - if (cause === "connection" || cause === "timeout") return true; - const message = err instanceof Error ? err.message : ""; - return /ECONNREFUSED|ECONNRESET|ETIMEDOUT|Can't reach database server|timed out/i.test( - message, - ); +const PRISMA_INPUT_CODES = new Set([ + "P2000", + "P2006", + "P2007", + "P2009", + "P2012", + "P2019", + "P2020", + "P2023", +]); + +/** + * Whether a query failed because of the value it was asked about, rather than + * because of the store it asked. Deliberately a short allowlist: Postgres' + * data-exception class (SQLSTATE 22, an encoding or a cast it will not accept) + * and Prisma's validation refusals. Everything else, including failures with + * no recognisable code at all, is not an answer about the input. + * + * The allowlist runs this way round because the other direction cannot be + * closed: the pool's own "timeout exceeded when trying to connect" carries no + * code, nor do half the ways a connection ends, and a resource SQLSTATE + * (53300 too many connections, 53200 out of memory, 55P03 a lock timeout) says + * nothing about the value either. A denylist of "the database is down" signals + * folded each of those into "this value matches nothing". + */ +export function isInputRefusal(err: unknown): boolean { + if ( + typeof err === "object" && + err !== null && + (err as { name?: unknown }).name === "PrismaClientValidationError" + ) { + return true; + } + const { code } = classifyRestoreFailure(err); + if (code === null) return false; + if (PRISMA_INPUT_CODES.has(code)) return true; + return /^22[0-9A-Z]{3}$/.test(code); } diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index 850e55fa3..3426575fa 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -496,6 +496,19 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { * instant, a different oldest bucket was corrupted on each run and nothing * repaired it. The window must be snapped out to whole buckets first. */ + it("stops between granularities once the job's budget is spent", async () => { + queryRawUnsafe.mockResolvedValue([]); + let asked = 0; + const result = await recomputeUserRollups("user-1", { + // The budget runs out after the first granularity was folded. + shouldStop: () => ++asked > 1, + }); + expect(result.stopped).toBe(true); + // The first granularity's aggregate, then nothing more. + expect(queryRawUnsafe).toHaveBeenCalledTimes(1); + queryRawUnsafe.mockReset(); + }); + describe("bucket alignment of the recompute window", () => { it("snaps a mid-day `from` down to the start of its DAY bucket", async () => { queryRawUnsafe.mockResolvedValueOnce(syntheticRows(1)); diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 9dd95cc2c..cf33a2c6d 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -126,6 +126,17 @@ export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill"; /** Boot-fold worker concurrency. Serial so we don't crowd the pool. */ export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1; +/** + * The full fold's expiry: an hour. One fold is four granularities, each an + * aggregate held to the connection's statement timeout (60 s by default) and + * a persist transaction of up to 120 s, so the worst case under the defaults + * is close to thirteen minutes, too near pg-boss's fifteen-minute default to + * rely on, and an operator who raises `DATABASE_STATEMENT_TIMEOUT_MS` (or + * sets it to 0) lifts that bound altogether. Past the expiry pg-boss retries + * the job, and the retry folded the same account a second time. + */ +export const ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS = 60 * 60; + /** Payload the boot-time backfill helper enqueues. */ export interface RollupFullBackfillPayload { userId: string; @@ -203,8 +214,14 @@ export async function recomputeUserRollups( * statement group, so it ends with it. */ statementTimeoutMs?: number; + /** + * Asked before each granularity: a fold that runs under a job's budget + * (`jobBudget`) stops between granularities, each of which commits on + * its own, and reports `stopped`. + */ + shouldStop?: () => boolean; } = {}, -): Promise<{ rowsUpserted: number; durationMs: number }> { +): Promise<{ rowsUpserted: number; durationMs: number; stopped?: true }> { const startedAt = Date.now(); const granularities = opts.granularities ?? ALL_GRANULARITIES; const fiveYearsAgo = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); @@ -214,6 +231,13 @@ export async function recomputeUserRollups( let rowsUpserted = 0; for (const granularity of granularities) { + if (opts.shouldStop?.()) { + return { + rowsUpserted, + durationMs: Date.now() - startedAt, + stopped: true, + }; + } // Snap the window out to whole buckets before aggregating. The // aggregate filters rows by `measured_at` but GROUPS by // `date_trunc(, measured_at)`, and `persistRollupRows` @@ -1219,6 +1243,7 @@ export async function enqueueBootTimeRollupBackfill( enqueuedAt: new Date().toISOString(), }; const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, { + expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, retryLimit: 3, retryDelay: 60, retryBackoff: true, diff --git a/src/lib/strava/credentials.ts b/src/lib/strava/credentials.ts index ff14695d6..768911c8e 100644 --- a/src/lib/strava/credentials.ts +++ b/src/lib/strava/credentials.ts @@ -70,6 +70,22 @@ export async function clearStravaClientCredentials( }); } +/** + * The `User` columns a removed Strava connection leaves behind: no tokens, no + * athlete, and no backfill state. The backfill markers belong to the account + * that was connected. Left standing, a later connection inherited them: a + * completed marker kept its history from ever being read, and a resume point + * made its first run start below the old account's oldest activity, so + * everything newer was never asked for. + */ +export const STRAVA_DISCONNECTED = { + stravaAccessTokenEncrypted: null, + stravaRefreshTokenEncrypted: null, + stravaAthleteId: null, + stravaBackfillCompletedAt: null, + stravaBackfillResumeBefore: null, +} as const; + export interface StravaConnection { accessToken: string; refreshToken: string; diff --git a/tests/integration/heavy-lane-defer.test.ts b/tests/integration/heavy-lane-defer.test.ts index bbd0f0485..dfbb613dd 100644 --- a/tests/integration/heavy-lane-defer.test.ts +++ b/tests/integration/heavy-lane-defer.test.ts @@ -59,7 +59,13 @@ describe("deferLaneJobs", () => { expect(fetched?.id).toBe(id); expect((await row(id!))?.state).toBe("active"); - expect(await deferLaneJobs(WAITER_QUEUE, [fetched!], 30)).toBe(1); + const deferral = await deferLaneJobs(WAITER_QUEUE, [fetched!], 30); + expect(deferral.deferred).toBe(1); + // When it was first queued, so a long wait can be told from a short one. + expect(deferral.oldestCreatedOn).toBeInstanceOf(Date); + expect(Date.now() - deferral.oldestCreatedOn!.getTime()).toBeLessThan( + 60_000, + ); expect(await row(id!)).toMatchObject({ state: "created", retry_count: 0, @@ -77,7 +83,10 @@ describe("deferLaneJobs", () => { it("leaves a job alone that is no longer active", async () => { const id = await boss.send(WAITER_QUEUE, { probe: 2 }); const job = { id, name: WAITER_QUEUE, data: {} } as unknown as Job; - expect(await deferLaneJobs(WAITER_QUEUE, [job], 30)).toBe(0); + expect(await deferLaneJobs(WAITER_QUEUE, [job], 30)).toEqual({ + deferred: 0, + oldestCreatedOn: null, + }); expect((await row(id!))?.future).toBe(false); await boss.deleteJob(WAITER_QUEUE, id!); }); diff --git a/tests/integration/measurement-hot-update.test.ts b/tests/integration/measurement-hot-update.test.ts index 731809db6..7b5144c1c 100644 --- a/tests/integration/measurement-hot-update.test.ts +++ b/tests/integration/measurement-hot-update.test.ts @@ -8,11 +8,10 @@ * here against Postgres: the no-op update keeps `updated_at` and * `sync_version` and is counted as HOT in the transaction's own statistics, * a real change still moves both, the nightly check reads the table's - * statistics and leaves the vacuuming to autovacuum, and the migration gives - * up on a lock it cannot get instead of queueing every writer behind it. + * statistics and leaves the vacuuming to autovacuum. The migration's lock + * wait is proven through the real `migrate deploy` in + * `migrate-deploy-lock-retry.test.ts`. */ -import { readFileSync } from "node:fs"; -import { join } from "node:path"; import { Client } from "pg"; import { beforeEach, describe, expect, it } from "vitest"; @@ -121,46 +120,3 @@ describe("the nightly check", () => { } }); }); - -describe("the migration's lock wait", () => { - it("fails within its lock timeout while another session holds the table", async () => { - // The migration's own text under fresh names, run the way `migrate - // deploy` runs a file: one multi-statement query, one transaction. - const sql = readFileSync( - join( - process.cwd(), - "prisma/migrations/0412_measurement_hot_updates/migration.sql", - ), - "utf8", - ) - .replaceAll("measurements_keep_stamps_on_noop", "lock_probe_fn") - .replaceAll("measurements_noop_update_keeps_stamps", "lock_probe_trg"); - const holder = new Client({ connectionString: process.env.DATABASE_URL }); - const migrator = new Client({ connectionString: process.env.DATABASE_URL }); - await holder.connect(); - await migrator.connect(); - try { - // What an anti-wraparound autovacuum holds, and does not give up. - await holder.query("BEGIN"); - await holder.query( - "LOCK TABLE measurements IN SHARE UPDATE EXCLUSIVE MODE", - ); - const startedAt = Date.now(); - const error = await migrator.query(sql).then( - () => null, - (err: { code?: string }) => err, - ); - expect(error?.code).toBe("55P03"); - expect(Date.now() - startedAt).toBeLessThan(30_000); - // Rolled back whole: nothing of the probe is left behind. - const { rows } = await migrator.query( - "SELECT to_regproc('lock_probe_fn') AS fn", - ); - expect(rows[0].fn).toBeNull(); - } finally { - await holder.query("ROLLBACK").catch(() => {}); - await holder.end(); - await migrator.end(); - } - }, 60_000); -}); diff --git a/tests/integration/migrate-deploy-lock-retry.test.ts b/tests/integration/migrate-deploy-lock-retry.test.ts new file mode 100644 index 000000000..b9678feaf --- /dev/null +++ b/tests/integration/migrate-deploy-lock-retry.test.ts @@ -0,0 +1,259 @@ +/** + * The boot's `migrate deploy` (`scripts/migrate-deploy.mjs`) against the real + * Prisma CLI and Postgres: migration 0412's lock wait, and the retry after it. + * + * Runs 0412's own text (its ten-second timeout cut to one) in a schema of its + * own, on a minimal `measurements` table, while another session holds the + * lock an anti-wraparound autovacuum would hold. Prisma runs a migration file + * statement by statement, so this is the only place the timeout and the + * all-or-nothing `DO` block are proven the way they will run on a host; a + * multi-statement `pg` query is one implicit transaction and would prove both + * even for a file that cannot deliver them. + */ +import { spawn } from "node:child_process"; +import { + cpSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Client } from "pg"; +import { afterAll, beforeEach, describe, expect, it } from "vitest"; + +const SCHEMA = "lock_retry_probe"; +const NAME = "0412_measurement_hot_updates"; +const ROOT = process.cwd(); + +const dirs: string[] = []; +afterAll(async () => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true }); + const db = await connect(); + await db.query(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`); + await db.end(); +}); + +function baseUrl(): string { + const url = process.env.DATABASE_URL; + if (!url) throw new Error("DATABASE_URL is not set for the integration run"); + return url; +} + +function probeUrl(): string { + const url = new URL(baseUrl()); + url.searchParams.set("schema", SCHEMA); + return url.toString(); +} + +async function connect(): Promise { + const url = new URL(baseUrl()); + url.searchParams.delete("schema"); + const client = new Client({ connectionString: url.toString() }); + await client.connect(); + return client; +} + +/** + * A migrations directory whose first migration creates the probe table + * (Prisma will not migrate a schema it did not build) and is applied here. + */ +async function migrationsDir(): Promise { + const dir = mkdtempSync(join(tmpdir(), "hl-lock-retry-")); + dirs.push(dir); + writeFileSync( + join(dir, "schema.prisma"), + 'datasource db {\n provider = "postgresql"\n}\n', + ); + mkdirSync(join(dir, "migrations", "0001_probe_table"), { recursive: true }); + cpSync( + join(ROOT, "prisma", "migrations", "migration_lock.toml"), + join(dir, "migrations", "migration_lock.toml"), + ); + writeFileSync( + join(dir, "migrations", "0001_probe_table", "migration.sql"), + `CREATE TABLE "measurements" ( + "id" integer PRIMARY KEY, "value" integer, + "updated_at" timestamp(3), "sync_version" integer);`, + ); + const setup = await migrateDeploy(dir); + expect(setup.code, setup.stderr).toBe(0); + return dir; +} + +/** Add 0412's own text, its timeout cut to one second, optionally unmarked. */ +function add0412(dir: string, { marked }: { marked: boolean }): void { + const original = readFileSync( + join(ROOT, "prisma", "migrations", NAME, "migration.sql"), + "utf8", + ); + let sql = original.replace("'lock_timeout', '10s'", "'lock_timeout', '1s'"); + expect( + sql, + "0412 still sets its timeout the way this test shortens it", + ).not.toBe(original); + if (!marked) sql = sql.replace(/^-- @retry-on-lock-timeout\n/m, ""); + mkdirSync(join(dir, "migrations", NAME), { recursive: true }); + writeFileSync(join(dir, "migrations", NAME, "migration.sql"), sql); +} + +function migrateDeploy( + dir: string, +): Promise<{ code: number; stdout: string; stderr: string }> { + return new Promise((resolve) => { + const child = spawn( + process.execPath, + [ + join(ROOT, "scripts", "migrate-deploy.mjs"), + "--schema", + join(dir, "schema.prisma"), + ], + { + cwd: ROOT, + env: { + ...process.env, + DATABASE_URL: probeUrl(), + HEALTHLOG_PRISMA_CLI: join( + ROOT, + "node_modules", + "prisma", + "build", + "index.js", + ), + HEALTHLOG_MIGRATIONS_DIR: join(dir, "migrations"), + HEALTHLOG_MIGRATE_LOCK_RETRY_DELAYS: "2,2,2", + }, + }, + ); + let stdout = ""; + let stderr = ""; + child.stdout.on("data", (chunk) => (stdout += chunk)); + child.stderr.on("data", (chunk) => (stderr += chunk)); + child.on("close", (code) => resolve({ code: code ?? 1, stdout, stderr })); + }); +} + +/** Hold what an anti-wraparound autovacuum holds on the probe table. */ +async function holdLock(): Promise { + const holder = await connect(); + await holder.query("BEGIN"); + await holder.query( + `LOCK TABLE ${SCHEMA}.measurements IN SHARE UPDATE EXCLUSIVE MODE`, + ); + return holder; +} + +async function state(db: Client) { + const rows = await db.query( + `SELECT migration_name, finished_at IS NOT NULL AS finished, + rolled_back_at IS NOT NULL AS rolled_back, logs + FROM ${SCHEMA}._prisma_migrations ORDER BY started_at`, + ); + const trigger = await db.query( + `SELECT 1 FROM pg_trigger t JOIN pg_class c ON c.oid = t.tgrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = $1 AND t.tgname = 'measurements_noop_update_keeps_stamps'`, + [SCHEMA], + ); + const options = await db.query( + `SELECT c.reloptions FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = $1 AND c.relname = 'measurements'`, + [SCHEMA], + ); + return { + migrations: rows.rows, + trigger: trigger.rowCount === 1, + reloptions: (options.rows[0]?.reloptions ?? []) as string[], + }; +} + +beforeEach(async () => { + const db = await connect(); + await db.query(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`); + await db.end(); +}); + +describe("migration 0412 under a held lock", () => { + it("gives up within its timeout, changes nothing, and the boot retries it", async () => { + const dir = await migrationsDir(); + add0412(dir, { marked: true }); + const holder = await holdLock(); + // The first attempt meets the lock; the holder lets go during the backoff. + const release = setTimeout(() => { + void holder.query("ROLLBACK"); + }, 2_500); + const startedAt = Date.now(); + const result = await migrateDeploy(dir); + clearTimeout(release); + await holder.query("ROLLBACK").catch(() => {}); + await holder.end(); + + expect(result.code, result.stderr).toBe(0); + expect(result.stderr).toContain("Database error code: 55P03"); + expect(result.stdout).toContain( + `migration ${NAME} timed out waiting for a lock (55P03)`, + ); + expect(Date.now() - startedAt).toBeLessThan(45_000); + + const db = await connect(); + const after = await state(db); + await db.end(); + expect( + after.migrations.map((m) => [ + m.migration_name, + m.finished, + m.rolled_back, + ]), + ).toEqual([ + ["0001_probe_table", true, false], + [NAME, false, true], + [NAME, true, false], + ]); + expect(after.trigger).toBe(true); + expect(after.reloptions).toContain("fillfactor=85"); + }, 90_000); + + it("leaves nothing half-applied, stays fatal unmarked, and a later boot recovers", async () => { + const dir = await migrationsDir(); + add0412(dir, { marked: false }); + const holder = await holdLock(); + let result; + try { + result = await migrateDeploy(dir); + } finally { + await holder.query("ROLLBACK").catch(() => {}); + await holder.end(); + } + expect(result.code).not.toBe(0); + expect(result.stderr).toContain("not marked safe to re-run"); + + const db = await connect(); + const failed = await state(db); + // The lock-taking statements rolled back as one: no trigger, no options. + expect(failed.trigger).toBe(false); + expect(failed.reloptions).toEqual([]); + expect(failed.migrations).toHaveLength(2); + expect(failed.migrations[1]).toMatchObject({ + migration_name: NAME, + finished: false, + rolled_back: false, + }); + expect(failed.migrations[1].logs).toContain("Database error code: 55P03"); + + // The next image ships the marked file; its boot meets P3009 and reads + // the cause from the stored row. + add0412(dir, { marked: true }); + const recovered = await migrateDeploy(dir); + expect(recovered.code, recovered.stderr).toBe(0); + expect(recovered.stderr).toContain("P3009"); + const after = await state(db); + await db.end(); + expect(after.trigger).toBe(true); + expect(after.migrations.at(-1)).toMatchObject({ + finished: true, + rolled_back: false, + }); + }, 90_000); +}); diff --git a/tests/integration/session-time-zone.test.ts b/tests/integration/session-time-zone.test.ts index 3743de7e8..4a58deab4 100644 --- a/tests/integration/session-time-zone.test.ts +++ b/tests/integration/session-time-zone.test.ts @@ -23,6 +23,7 @@ import { } from "vitest"; import { PrismaClient } from "@/generated/prisma/client"; +import { fixtureSessionSecret } from "@/__tests__/helpers/session-secret"; import { getPrismaClient, truncateAllTables } from "./setup"; const ZONE = "Europe/Berlin"; @@ -167,17 +168,19 @@ describe("on a database whose default zone is Europe/Berlin", () => { const { validateSessionFromCookieValue } = await import("@/lib/auth/session"); const { hashToken } = await import("@/lib/auth/hmac"); + const liveSecret = fixtureSessionSecret(); + const expiredSecret = fixtureSessionSecret(); const live = await getPrismaClient().session.create({ data: { userId: USER_ID, - tokenHash: hashToken("hls_live-secret"), + tokenHash: hashToken(liveSecret), expiresAt: new Date(Date.now() + 60 * 60_000), }, }); await getPrismaClient().session.create({ data: { userId: USER_ID, - tokenHash: hashToken("hls_expired-secret"), + tokenHash: hashToken(expiredSecret), expiresAt: new Date(Date.now() - 60_000), }, }); @@ -185,10 +188,10 @@ describe("on a database whose default zone is Europe/Berlin", () => { 60_000, ); await expect( - validateSessionFromCookieValue("hls_live-secret"), + validateSessionFromCookieValue(liveSecret), ).resolves.not.toBeNull(); await expect( - validateSessionFromCookieValue("hls_expired-secret"), + validateSessionFromCookieValue(expiredSecret), ).resolves.toBeNull(); }); diff --git a/tests/integration/sharing-audit-actor.test.ts b/tests/integration/sharing-audit-actor.test.ts index f29121eb2..57874c92f 100644 --- a/tests/integration/sharing-audit-actor.test.ts +++ b/tests/integration/sharing-audit-actor.test.ts @@ -462,6 +462,7 @@ describe("the activity feed states its own limits", () => { data: { entries: { action: string }[]; retentionDays: number; + truncated: boolean; }; meta: { truncated: boolean; limit: number }; }; @@ -511,7 +512,9 @@ describe("the activity feed states its own limits", () => { expect(small.data.entries).toHaveLength(3); // The cap rides the envelope, like every capped list's. expect(small.meta.truncated).toBe(false); - expect(small.data).not.toHaveProperty("truncated"); + // The field v1.43.6 shipped in `data` stays, deprecated, with the same + // value. + expect(small.data.truncated).toBe(false); // Over it: same shape, different claim. await getPrismaClient().auditLog.createMany({ @@ -525,5 +528,6 @@ describe("the activity feed states its own limits", () => { const large = await feed(); expect(large.data.entries).toHaveLength(100); expect(large.meta).toMatchObject({ truncated: true, limit: 100 }); + expect(large.data.truncated).toBe(true); }); });