From 47f20b773daf385917fc0607b4cfca525752d351 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:36:16 +0200 Subject: [PATCH 01/12] fix(auth): answer 503 for every session lookup failure that is not about the value The catch kept a denylist of "store unreachable" signals and folded everything else into a missing session. The pool's own connect timeout carries no code, and 53300, 53200 and 55P03 are about the store, so all four answered 401 and signed the client out. The shape check already refuses crafted values before the query; the catch now keeps only a short allowlist of input refusals (SQLSTATE class 22, Prisma validation) as null and raises 503 for the rest. --- src/lib/auth/__tests__/session.test.ts | 36 ++++++++++++++++ src/lib/auth/session.ts | 14 +++--- src/lib/prisma-errors.ts | 59 ++++++++++++++++++-------- 3 files changed, 85 insertions(+), 24 deletions(-) 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/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); } From 9c895030190f99a132e4c6ce2f09b3e529b92494 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:45:14 +0200 Subject: [PATCH 02/12] fix(migrations): bound 0412's lock wait where it takes effect and retry a lock timeout on boot migrate deploy runs a migration file statement by statement, so the top-level SET LOCAL lock_timeout in 0412 did nothing: the migration waited as long as the lock was held, with every write to measurements queued behind it, and a file that failed half-way stayed half-applied. 0412 now takes its locks inside one DO block with a statement-scoped timeout, which rolls back whole, after an idempotent CREATE OR REPLACE. The entrypoint runs migrate deploy through scripts/migrate-deploy.mjs. A 55P03 failure of a migration marked -- @retry-on-lock-timeout is marked rolled back and retried after 15/30/60/120/240 s, read from Prisma's report or, after P3009, from the stored row. Every other failure stays fatal. A guard holds migrations to the shape that makes the marker true, and 0411/0413 add their columns with IF NOT EXISTS so a manual re-run is safe too. --- Dockerfile | 9 +- docker-entrypoint.sh | 5 +- docs/ops/deploy.md | 49 +++ .../migration.sql | 2 +- .../migration.sql | 47 ++- .../0413_strava_backfill_resume/migration.sql | 2 +- scripts/__tests__/migrate-deploy.test.ts | 283 ++++++++++++++++++ scripts/migrate-deploy.mjs | 239 +++++++++++++++ .../migration-lock-timeout-guard.test.ts | 128 ++++++++ .../measurement-hot-update.test.ts | 50 +--- .../migrate-deploy-lock-retry.test.ts | 259 ++++++++++++++++ 11 files changed, 1004 insertions(+), 69 deletions(-) create mode 100644 scripts/__tests__/migrate-deploy.test.ts create mode 100644 scripts/migrate-deploy.mjs create mode 100644 src/__tests__/migration-lock-timeout-guard.test.ts create mode 100644 tests/integration/migrate-deploy-lock-retry.test.ts 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/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__/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/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); +}); From 19732b5aa187f34ac18a9793b7f8ccf1b7b061de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:52:54 +0200 Subject: [PATCH 03/12] fix(jobs): let a sliced import give way to a waiting heavy job and warn on long lane waits A job sent back by the heavy lane now leaves word that the lane is wanted. The holder, and only the holder, sees it through laneWanted(); the Google Health backfill asks it where its slice budget already stops resumable walks, after the first page, ends the slice there and queues the rest past the deferred job's return. Holders that cannot stop part-way run to their end as before. The way back to the queue now returns when the oldest deferred job was first queued; a job that has stood in line for two hours is reported once per holder as job.heavy_lane.waiting_long at warn, so a lane held for hours reaches the alerts. --- docs/integrations/google-health.md | 6 +- .../__tests__/walk-resume-and-retry.test.ts | 30 +++++ src/lib/google-health/client.ts | 14 ++- .../__tests__/google-health-backfill.test.ts | 34 +++++ src/lib/jobs/__tests__/heavy-lane.test.ts | 110 ++++++++++++++++- src/lib/jobs/google-health-backfill.ts | 23 +++- src/lib/jobs/heavy-lane.ts | 116 ++++++++++++++++-- tests/integration/heavy-lane-defer.test.ts | 13 +- 8 files changed, 323 insertions(+), 23 deletions(-) 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/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/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!); }); From c50007ef616547946548de5596043473c876c78e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:55:09 +0200 Subject: [PATCH 04/12] fix(day): read the notable history through the fenced cache path The comparison history blocks were read with cache.get and stored with cache.set, outside the single-flight slot that an invalidation fences. A read that spanned a write cached the days from before it for the block's whole TTL, and two concurrent reads of a block both queried. Each block now goes through cache.wrap; the blocks missing at the time still share one read of their span. --- .../__tests__/notable-history-cache.test.ts | 36 ++++++++++++ src/lib/day/notable.ts | 55 ++++++++++++------- 2 files changed, 72 insertions(+), 19 deletions(-) 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) { From 4f020041467a88a4c81501ab6409b4b8d4938c92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:55:42 +0200 Subject: [PATCH 05/12] test(guards): derive the rounding guard's directories from the connection keys The guard kept its own provider list and its own comment stripper next to the shared PROVIDER_DIRS and stripComments helpers, so a new integration could slip past it. It now uses both, plus the file importers, which are not a connection. --- src/__tests__/provider-rounding-guard.test.ts | 31 ++++++------------- 1 file changed, 9 insertions(+), 22 deletions(-) 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([]); }); From b08d5270457618ab264a482af6a025800d274b0b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:56:55 +0200 Subject: [PATCH 06/12] test(auth): mint session fixtures in the shape the lookup accepts The session time-zone test stored sessions under readable secrets that the cookie shape check now refuses before the query, so it tested the refusal instead of the expiry. A shared helper mints hls_ plus 32 hex bytes, and the two route tests that set a session cookie use it too. --- src/__tests__/helpers/session-secret.ts | 11 +++++++++++ .../api/auth/native/complete/__tests__/route.test.ts | 3 ++- .../api/auth/oidc/callback/__tests__/route.test.ts | 4 +++- tests/integration/session-time-zone.test.ts | 11 +++++++---- 4 files changed, 23 insertions(+), 6 deletions(-) create mode 100644 src/__tests__/helpers/session-secret.ts 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/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/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(); }); From 2be262d5748b9e16955f97489bee71e94f23d57d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:58:03 +0200 Subject: [PATCH 07/12] refactor(api): drop apiPostList, which no capped list reads Every capped route is read with GET, so the POST reader had no caller and failed the dead-code gate. The list-cap guard now binds a capped route to apiGetList alone, which is the reader that exists. --- src/__tests__/list-cap-reader-guard.test.ts | 6 +++--- src/lib/api/api-fetch.ts | 13 ------------- src/lib/api/list-cap.ts | 2 +- 3 files changed, 4 insertions(+), 17 deletions(-) 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/lib/api/api-fetch.ts b/src/lib/api/api-fetch.ts index ffabbc4a8..6f2648c1d 100644 --- a/src/lib/api/api-fetch.ts +++ b/src/lib/api/api-fetch.ts @@ -231,19 +231,6 @@ export async function apiGetList( 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. */ From 9a5c2b8ec1ce65f4105b6cb03c136f89f1f2f67b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 11:00:25 +0200 Subject: [PATCH 08/12] fix(api): keep data.truncated on the account activity feed, deprecated v1.43.6 shipped truncated as a required field of the activity response's data, and moving it to meta alone would break a client built against that in a patch release. data.truncated is back with the same value as meta.truncated and is marked deprecated in the spec; the web client keeps reading meta. --- docs/api/openapi.yaml | 11 +++++++++-- src/app/api/account/activity/route.ts | 4 ++++ src/lib/openapi/routes/account-sharing.ts | 7 ++++++- tests/integration/sharing-audit-actor.test.ts | 6 +++++- 4 files changed, 24 insertions(+), 4 deletions(-) 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/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/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/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); }); }); From 742705eadd67b25a51f0b678ea8616d8eb75f014 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 11:01:19 +0200 Subject: [PATCH 09/12] docs(audit): say why two import refusals stay audit rows The export imports log a refused upload as a signal. The admin import and the medication intake import kickoff keep audit rows on purpose: one is an administrator acting on another account, the other is shown to the record's owner in the activity feed. --- src/app/api/admin/import-apple-health-export/route.ts | 4 ++++ src/lib/medications/intake-import-admission.ts | 4 ++++ 2 files changed, 8 insertions(+) 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/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, From bd00845be384774928bfa32d055cd85b0e8fda4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 11:02:50 +0200 Subject: [PATCH 10/12] fix(strava): clear the backfill markers when the connection is removed Disconnecting Strava, or deleting its credentials, cleared the tokens and the athlete but kept stravaBackfillCompletedAt and stravaBackfillResumeBefore. A later connection inherited them: a completed marker kept its history from being read, and a resume point started its first walk below the previous account's oldest activity. Both paths now clear one shared set of columns. --- .../credentials/__tests__/route.test.ts | 11 +++- src/app/api/strava/credentials/route.ts | 7 +-- .../strava/disconnect/__tests__/route.test.ts | 63 +++++++++++++++++++ src/app/api/strava/disconnect/route.ts | 11 ++-- src/lib/strava/credentials.ts | 16 +++++ 5 files changed, 95 insertions(+), 13 deletions(-) create mode 100644 src/app/api/strava/disconnect/__tests__/route.test.ts 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/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; From 7e6ed676e0fdfbd574e2df56a457b8215bd9a309 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 11:03:18 +0200 Subject: [PATCH 11/12] docs(insights): describe the ECG section's failure row as it is The page comment still said the overview section unmounts silently on a read error; it shows a QueryErrorRow, which is what the UI standards ask of a supplementary block. The page keeps the full error card. --- src/app/insights/ecg/page.tsx | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) 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 ( Date: Tue, 20 Oct 2026 11:06:17 +0200 Subject: [PATCH 12/12] fix(jobs): give the full rollup fold an expiry, a budget and a lock The fold was declared short on the strength of its 60-second statements, but four aggregates and four persist transactions of up to 120 s come to about thirteen minutes at worst under the defaults, and nothing bounds them once an operator lifts the statement timeout. Past pg-boss's fifteen-minute default the retry folded the account a second time. The send now carries an hour's expiry, the fold stops between granularities on the job's budget, and the binding holds a per-account lockedPass outside the lane. The runtime guard now requires every heavy queue to be declared long. --- src/__tests__/job-queue-runtime-guard.test.ts | 22 +++++++++ src/lib/jobs/queue-runtime.ts | 12 +++-- src/lib/jobs/reminder/register-rollup.ts | 48 ++++++++++++------- .../__tests__/measurement-rollups.test.ts | 13 +++++ src/lib/rollups/measurement-rollups.ts | 27 ++++++++++- 5 files changed, 102 insertions(+), 20 deletions(-) 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/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/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,