mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
Merge branch 'fix/v1437-r3' into release/v1.43.7
This commit is contained in:
commit
71fa8ad7b3
49 files changed
+1753
-220
No files matched your search
+6
-3
@@ -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
|
||||
|
||||
@@ -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 ────────────────────────────────────────
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(<pid>)`.
|
||||
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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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$;
|
||||
@@ -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);
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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: <name>` and `Database error code: <state>`
|
||||
* 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();
|
||||
}
|
||||
@@ -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")}`;
|
||||
}
|
||||
@@ -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<string, string> = {};
|
||||
const notLong = (table: Readonly<Record<string, QueueRuntime>>) =>
|
||||
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
|
||||
|
||||
@@ -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}(?:[?"\`])`,
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -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([]);
|
||||
});
|
||||
});
|
||||
@@ -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<string, string> = {
|
||||
"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([]);
|
||||
});
|
||||
|
||||
@@ -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`).
|
||||
|
||||
@@ -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", {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<typeof import("@/lib/strava/credentials")>()
|
||||
).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,
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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: <T extends (...args: unknown[]) => 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<typeof import("@/lib/strava/credentials")>()
|
||||
).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<typeof vi.fn>;
|
||||
|
||||
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,
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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 });
|
||||
|
||||
@@ -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 (
|
||||
<SubPageShell
|
||||
|
||||
@@ -231,19 +231,6 @@ export async function apiGetList<T = unknown>(
|
||||
return { data, cap: readListCap(meta) };
|
||||
}
|
||||
|
||||
/** POST variant of {@link apiGetList}, for a list read with a body. */
|
||||
export async function apiPostList<T = unknown>(
|
||||
path: string,
|
||||
body?: unknown,
|
||||
init?: RequestInit,
|
||||
): Promise<{ data: T; cap: ListCap | null }> {
|
||||
const { data, meta } = await apiFetchEnvelope<T>(
|
||||
path,
|
||||
withDefaultIdempotencyKey(withJsonBody("POST", body, init)),
|
||||
);
|
||||
return { data, cap: readListCap(meta) };
|
||||
}
|
||||
|
||||
/** GET — resolves with the unwrapped `data` payload. */
|
||||
export function apiGet<T = unknown>(
|
||||
path: string,
|
||||
|
||||
@@ -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.
|
||||
*/
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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<void>((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<string, Map<string, number>>;
|
||||
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(() => {
|
||||
|
||||
+36
-19
@@ -274,26 +274,43 @@ async function readExtremeHistory(args: {
|
||||
});
|
||||
}
|
||||
|
||||
const found = new Map<string, DailyStats>();
|
||||
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<DailyStats> | null = null;
|
||||
const readSpan = (): Promise<DailyStats> => {
|
||||
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<string, DailyStats>(cells);
|
||||
|
||||
const out: DailyStats = new Map();
|
||||
for (const block of blocks) {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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<string, boolean[]> = { holder: [], beside: [] };
|
||||
let release!: () => void;
|
||||
const released = new Promise<void>((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<void>((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;
|
||||
});
|
||||
});
|
||||
@@ -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({
|
||||
|
||||
+103
-13
@@ -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<Record<string, HeavyQueue>> = {
|
||||
/** 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<unknown>[],
|
||||
seconds: number = LANE_DEFER_SECONDS,
|
||||
): Promise<number> {
|
||||
): Promise<LaneDeferral> {
|
||||
const ids = jobs.map((job) => job.id);
|
||||
return prisma.$executeRaw`
|
||||
const rows = await prisma.$queryRaw<Array<{ created_on: Date }>>`
|
||||
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<Date | null>(
|
||||
(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<string>;
|
||||
}
|
||||
|
||||
let hold: LaneHold | null = null;
|
||||
const holdScope = new AsyncLocalStorage<LaneHold>();
|
||||
|
||||
/**
|
||||
* 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<J extends Job<unknown>>(
|
||||
queue: string,
|
||||
handler: (jobs: J[]) => Promise<JobOutcome>,
|
||||
inLane: (job: J) => boolean = () => true,
|
||||
defer: (queue: string, jobs: readonly J[]) => Promise<number> = deferLaneJobs,
|
||||
defer: (
|
||||
queue: string,
|
||||
jobs: readonly J[],
|
||||
) => Promise<LaneDeferral> = deferLaneJobs,
|
||||
now: () => number = Date.now,
|
||||
): (jobs: J[]) => Promise<JobOutcome> {
|
||||
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;
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -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<Record<string, QueueRuntime>> = {
|
||||
|
||||
// ── 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.",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<ZodOpenApiObject["paths"]> = {
|
||||
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": {
|
||||
|
||||
+41
-18
@@ -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);
|
||||
}
|
||||
@@ -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));
|
||||
|
||||
@@ -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(<granularity>, 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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<object>;
|
||||
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!);
|
||||
});
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
@@ -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<Client> {
|
||||
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<string> {
|
||||
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<Client> {
|
||||
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);
|
||||
});
|
||||
@@ -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();
|
||||
});
|
||||
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user