mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-10 05:12:14 +00:00
Merge branch 'fix/v1462-review' into release/v1.46.2
This commit is contained in:
Marc-André Bombeck committed
commit
d1ea803260
27 files changed
+1062
-157
No files matched your search
@@ -268,8 +268,10 @@ turns the limit off).
|
||||
`canceling statement due to statement timeout`.
|
||||
- The admin status page lists `rollup-full-backfill` under failing
|
||||
background jobs, and every admin gets a push once a system job has
|
||||
failed three times in a row, a reminder at most once a day while it keeps
|
||||
failing, and a notice when it runs again.
|
||||
failed three jobs in a row (retries of one job count once; a job that runs
|
||||
per account counts per account), a reminder at most once a day while it
|
||||
keeps failing, and a notice when the failing job runs again. A one-off job
|
||||
that failed and does not run again is announced once.
|
||||
|
||||
### What the app does about it
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
|
||||
"count": 84,
|
||||
"total": 94686,
|
||||
"total": 94692,
|
||||
"entries": {
|
||||
"app/medications/page-client.tsx": 839,
|
||||
"components/charts/mood-chart.tsx": 1166,
|
||||
@@ -71,7 +71,7 @@
|
||||
"lib/managed-profiles/handover.ts": 986,
|
||||
"lib/mcp/record-search.ts": 1467,
|
||||
"lib/mcp/rich-reads.ts": 1467,
|
||||
"lib/measurements/dense-intraday-retention.ts": 1010,
|
||||
"lib/measurements/dense-intraday-retention.ts": 1016,
|
||||
"lib/measurements/import-apple-health-export.ts": 1722,
|
||||
"lib/medications/scheduling/cadence.ts": 976,
|
||||
"lib/medications/scheduling/recurrence.ts": 1100,
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
|
||||
"count": 346,
|
||||
"total": 112215,
|
||||
"total": 112219,
|
||||
"entries": {
|
||||
"app/achievements/page.tsx :: AchievementsPage": 206,
|
||||
"app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 286,
|
||||
@@ -319,9 +319,9 @@
|
||||
"lib/labs/biomarker-merge.ts :: runMerge": 254,
|
||||
"lib/measurements/consolidate-daily-mean.ts :: consolidateDailyMean": 278,
|
||||
"lib/measurements/consolidate-legacy-steps.ts :: consolidateLegacySteps": 263,
|
||||
"lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention": 537,
|
||||
"lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource": 447,
|
||||
"lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay": 373,
|
||||
"lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention": 539,
|
||||
"lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource": 448,
|
||||
"lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay": 374,
|
||||
"lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay > foldOnce": 212,
|
||||
"lib/measurements/drain-per-sample-cumulative.ts :: drainPerSampleCumulative": 344,
|
||||
"lib/measurements/drain-per-sample-cumulative.ts :: drainPerSampleCumulative > writeDay": 205,
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.",
|
||||
"count": 2896,
|
||||
"total": 9229,
|
||||
"total": 9228,
|
||||
"entries": {
|
||||
"__tests__/account-payload-consumer-guard.test.ts": 2,
|
||||
"__tests__/acting-account-boundary-guard.test.ts": 7,
|
||||
@@ -2458,7 +2458,7 @@
|
||||
"lib/measurements/cumulative-day-sum.ts": 6,
|
||||
"lib/measurements/daily-series-read.ts": 2,
|
||||
"lib/measurements/dense-intraday-hourly-rebuild.ts": 1,
|
||||
"lib/measurements/dense-intraday-retention.ts": 4,
|
||||
"lib/measurements/dense-intraday-retention.ts": 3,
|
||||
"lib/measurements/display-transform.ts": 6,
|
||||
"lib/measurements/drain-per-sample-cumulative.ts": 8,
|
||||
"lib/measurements/import-apple-health-export.ts": 5,
|
||||
|
||||
@@ -110,6 +110,8 @@ const WRITE_SITES: Record<string, string> = {
|
||||
"value-preserving hourly fold; refolds the touched spans",
|
||||
"lib/measurements/dense-intraday-retention.ts":
|
||||
"value-preserving retention fold; refolds the touched days",
|
||||
"lib/measurements/dense-fold-removal.ts":
|
||||
"removes the samples the retention fold has just taken into its hourly means, inside the fold's transaction; the fold refolds the touched days",
|
||||
"lib/jobs/sleep-timeline-backfill.ts":
|
||||
"one-shot re-stamp of stored sleep rows onto their timeline; refolds the touched days",
|
||||
"lib/jobs/step-consolidation-repair.ts":
|
||||
|
||||
@@ -16,8 +16,9 @@
|
||||
*
|
||||
* - `userId` — single user mode. Synchronously calls
|
||||
* `foldUserRollupHistory(userId)` (the full fold, a calendar year per
|
||||
* statement) and returns the row count when the fold completes. The call is awaited inside the request, so
|
||||
* the operator sees the duration in the response.
|
||||
* statement, each statement under the boot job's budget) and returns the
|
||||
* row count when the fold completes. The call is awaited inside the
|
||||
* request, so the operator sees the duration in the response.
|
||||
* - `userId` omitted — calls `enqueueBootTimeRollupBackfill()` to
|
||||
* re-run the worker-boot discovery (`{ enqueued, skipped }` shape).
|
||||
*
|
||||
@@ -38,6 +39,7 @@ import { annotate } from "@/lib/logging/context";
|
||||
import {
|
||||
enqueueBootTimeRollupBackfill,
|
||||
foldUserRollupHistory,
|
||||
ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
} from "@/lib/rollups/full-backfill";
|
||||
|
||||
interface RecomputeBody {
|
||||
@@ -70,7 +72,12 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
|
||||
let result: Awaited<ReturnType<typeof foldUserRollupHistory>>;
|
||||
try {
|
||||
result = await foldUserRollupHistory(userId);
|
||||
// The per-statement budget the boot job folds with: this mode is for
|
||||
// the accounts the boot job could not finish, and under the database's
|
||||
// ordinary limit their yearly statements time out here too.
|
||||
result = await foldUserRollupHistory(userId, {
|
||||
statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
});
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
return apiError(message, 500);
|
||||
|
||||
@@ -27,7 +27,7 @@ import { recentProofMethods } from "@/lib/auth/second-factor";
|
||||
export const dynamic = "force-dynamic";
|
||||
|
||||
export const GET = apiHandler(async () => {
|
||||
const { user } = await requireAuth();
|
||||
const { user, authMethod } = await requireAuth();
|
||||
|
||||
const [recoveryCodesRemaining, webauthn, proofMethods] = await Promise.all([
|
||||
countRemainingRecoveryCodes(user.id),
|
||||
@@ -50,7 +50,12 @@ export const GET = apiHandler(async () => {
|
||||
passkeyNudgeDismissed: Boolean(user.passkeyUpgradeNudgeDismissed),
|
||||
// The ways this account can confirm it is you, the list the recent-proof
|
||||
// refusal names. The passkey card offers exactly these before adding one,
|
||||
// so an account without a password is not handed a password field.
|
||||
proofMethods,
|
||||
// so an account without a password is not handed a password field. The
|
||||
// provider round trip stamps a browser session and nothing else, so a
|
||||
// token is never offered it.
|
||||
proofMethods:
|
||||
authMethod === "cookie"
|
||||
? proofMethods
|
||||
: proofMethods.filter((method) => method !== "oidc"),
|
||||
});
|
||||
});
|
||||
@@ -47,6 +47,7 @@ import {
|
||||
reproofReturnUrl,
|
||||
} from "@/lib/auth/oidc-reproof";
|
||||
import { readStoredOidcState } from "@/lib/auth/oidc-callback/state";
|
||||
import { RECENT_PROOF_MAX_AGE_MS } from "@/lib/auth/existing-factor-proof";
|
||||
|
||||
const ISSUER = "https://idp.example.test";
|
||||
const OIDC_ENV = {
|
||||
@@ -195,9 +196,10 @@ describe("recordOidcReproof", () => {
|
||||
identity: { sub: "sub-1", authTime: Math.floor(now / 1000) - 5 },
|
||||
});
|
||||
expect(outcome).toEqual({ ok: true });
|
||||
// Stamped at the authentication, not at the callback.
|
||||
expect(db.updateSession).toHaveBeenCalledWith({
|
||||
where: { id: "s1" },
|
||||
data: { reproofAt: new Date(now) },
|
||||
data: { reproofAt: new Date((Math.floor(now / 1000) - 5) * 1000) },
|
||||
});
|
||||
expect(refundRateLimit).toHaveBeenCalledWith("auth:step-up:u1");
|
||||
expect(auditLog).toHaveBeenCalledWith(
|
||||
@@ -206,6 +208,34 @@ describe("recordOidcReproof", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("runs the window from the authentication, so a token near its end buys no extra minutes", async () => {
|
||||
db.findUser.mockResolvedValue(accountRow());
|
||||
const authTime = Math.floor(now / 1000) - 4 * 60;
|
||||
await recordOidcReproof({ ...args, identity: { sub: "sub-1", authTime } });
|
||||
const [call] = db.updateSession.mock.calls[0] as [
|
||||
{ data: { reproofAt: Date } },
|
||||
];
|
||||
const stamped = call.data.reproofAt;
|
||||
expect(stamped.getTime()).toBe(authTime * 1000);
|
||||
// Two minutes after the callback the proof is six minutes old: past the
|
||||
// window every recent-proof gate applies.
|
||||
expect(now + 2 * 60_000 - stamped.getTime()).toBeGreaterThan(
|
||||
RECENT_PROOF_MAX_AGE_MS,
|
||||
);
|
||||
});
|
||||
|
||||
it("never stamps a proof in the future for a provider clock a little ahead", async () => {
|
||||
db.findUser.mockResolvedValue(accountRow());
|
||||
await recordOidcReproof({
|
||||
...args,
|
||||
identity: { sub: "sub-1", authTime: Math.floor(now / 1000) + 30 },
|
||||
});
|
||||
expect(db.updateSession).toHaveBeenCalledWith({
|
||||
where: { id: "s1" },
|
||||
data: { reproofAt: new Date(now) },
|
||||
});
|
||||
});
|
||||
|
||||
it("refuses a stale auth_time without stamping", async () => {
|
||||
db.findUser.mockResolvedValue(accountRow());
|
||||
const outcome = await recordOidcReproof({
|
||||
|
||||
@@ -140,9 +140,18 @@ export async function recordOidcReproof(args: {
|
||||
return { ok: false, reason: refusal };
|
||||
}
|
||||
|
||||
// The proof is as old as the authentication, not the callback: the gate's
|
||||
// window then runs from the moment the person signed in at the provider,
|
||||
// so a token near the end of its window buys no extra minutes. A provider
|
||||
// clock a little ahead is clamped to now, never stamped into the future.
|
||||
// (A token without `auth_time` was refused above.)
|
||||
const authenticatedAtMs =
|
||||
identity.authTime === null
|
||||
? nowMs
|
||||
: Math.min(identity.authTime * 1000, nowMs);
|
||||
await prisma.session.update({
|
||||
where: { id: sessionId },
|
||||
data: { reproofAt: new Date(nowMs) },
|
||||
data: { reproofAt: new Date(authenticatedAtMs) },
|
||||
});
|
||||
await refundReproof(userId);
|
||||
await auditLog("auth.reproof.succeeded", {
|
||||
|
||||
@@ -13,68 +13,140 @@ import {
|
||||
} from "@/lib/jobs/system-job-alert";
|
||||
|
||||
const at = new Date("2026-10-29T10:00:00.000Z");
|
||||
const HOUR_MS = 3_600_000;
|
||||
|
||||
function run(overrides: Partial<TerminalRun> = {}): TerminalRun {
|
||||
/** The `ago`-th newest job (0 = newest), an hour apart. */
|
||||
function run(ago = 0, overrides: Partial<TerminalRun> = {}): TerminalRun {
|
||||
return {
|
||||
key: "",
|
||||
state: "failed",
|
||||
retryCount: 0,
|
||||
allFailed: false,
|
||||
completedOn: at,
|
||||
completedOn: new Date(at.getTime() - ago * HOUR_MS),
|
||||
error: "boom",
|
||||
expireSeconds: null,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
/** Jobs newest first, one hour apart, from a list of outcomes. */
|
||||
function history(key: string, outcomes: Array<"fail" | "ok">, from = 0) {
|
||||
return outcomes.map((o, i) =>
|
||||
run(from + i, {
|
||||
key,
|
||||
state: o === "fail" ? "failed" : "completed",
|
||||
error: o === "fail" ? `boom ${key}` : null,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
describe("judgeQueueRuns", () => {
|
||||
it("is inactive with no run left in the ledger", () => {
|
||||
expect(judgeQueueRuns([]).state).toBe("inactive");
|
||||
});
|
||||
|
||||
it("counts every attempt of a run, so one job that used up its retries is a streak", () => {
|
||||
const verdict = judgeQueueRuns([run({ retryCount: 3 })]);
|
||||
expect(verdict).toMatchObject({ state: "failing", failures: 4 });
|
||||
expect(verdict.lastError).toBe("boom");
|
||||
});
|
||||
|
||||
it("calls the streak at the threshold, not before", () => {
|
||||
const below = Array.from({ length: SYSTEM_JOB_FAILURE_STREAK - 1 }, () =>
|
||||
run(),
|
||||
);
|
||||
expect(judgeQueueRuns(below).state).toBe("pending");
|
||||
expect(judgeQueueRuns([...below, run()]).state).toBe("failing");
|
||||
});
|
||||
|
||||
it("ends the streak at the newest success", () => {
|
||||
const verdict = judgeQueueRuns([
|
||||
run({ error: "newest" }),
|
||||
run({ state: "completed", error: null }),
|
||||
run(),
|
||||
run(),
|
||||
run(),
|
||||
]);
|
||||
expect(verdict).toMatchObject({
|
||||
state: "ok",
|
||||
failures: 1,
|
||||
lastError: "newest",
|
||||
it("counts jobs, not attempts: one job that used up its retries is one failure", () => {
|
||||
expect(judgeQueueRuns([run(0)])).toMatchObject({
|
||||
state: "pending",
|
||||
failures: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("counts a completed run that says every candidate failed", () => {
|
||||
const allFailed = run({ state: "completed", allFailed: true });
|
||||
expect(judgeQueueRuns([allFailed, allFailed, allFailed]).state).toBe(
|
||||
"failing",
|
||||
it("calls the streak at the threshold, not before", () => {
|
||||
const below = Array.from(
|
||||
{ length: SYSTEM_JOB_FAILURE_STREAK - 1 },
|
||||
(_, i) => run(i),
|
||||
);
|
||||
expect(judgeQueueRuns(below).state).toBe("pending");
|
||||
const verdict = judgeQueueRuns([...below, run(below.length)]);
|
||||
expect(verdict).toMatchObject({
|
||||
state: "failing",
|
||||
failures: SYSTEM_JOB_FAILURE_STREAK,
|
||||
lastError: "boom",
|
||||
lastFailureAt: at,
|
||||
});
|
||||
});
|
||||
|
||||
it("does not report a recovery while the newest job failed", () => {
|
||||
expect(
|
||||
judgeQueueRuns(history("", ["fail", "ok", "fail", "fail", "fail"])),
|
||||
).toMatchObject({ state: "pending", failures: 0 });
|
||||
expect(
|
||||
judgeQueueRuns(history("", ["ok", "fail", "fail", "fail"])).state,
|
||||
).toBe("ok");
|
||||
});
|
||||
|
||||
it("has nothing to report for a queue that never failed", () => {
|
||||
expect(judgeQueueRuns(history("", ["ok", "ok"])).state).toBe("inactive");
|
||||
});
|
||||
|
||||
it("counts a completed job that says every candidate failed", () => {
|
||||
const allFailed = (ago: number) =>
|
||||
run(ago, { state: "completed", allFailed: true });
|
||||
expect(
|
||||
judgeQueueRuns([allFailed(0), allFailed(1), allFailed(2)]).state,
|
||||
).toBe("failing");
|
||||
});
|
||||
|
||||
it("explains a bare timeout the way the admin card does", () => {
|
||||
const verdict = judgeQueueRuns([
|
||||
run({ error: "job timed out", expireSeconds: 3600, retryCount: 2 }),
|
||||
]);
|
||||
const verdict = judgeQueueRuns(
|
||||
[0, 1, 2].map((ago) =>
|
||||
run(ago, { error: "job timed out", expireSeconds: 3600 }),
|
||||
),
|
||||
);
|
||||
expect(verdict.lastError).toMatch(
|
||||
/^job timed out: did not finish within its 1 h limit/,
|
||||
);
|
||||
});
|
||||
|
||||
describe("per account", () => {
|
||||
it("is failing while one account fails every hour and the others succeed", () => {
|
||||
// Interleaved: the queue's newest job is always another account's
|
||||
// success, which used to read as a recovery every other hour.
|
||||
const verdict = judgeQueueRuns([
|
||||
...history("a", ["fail", "fail", "fail"], 1),
|
||||
...history("b", ["ok", "ok", "ok", "ok"], 0),
|
||||
...history("c", ["ok", "ok", "ok"], 0),
|
||||
]);
|
||||
expect(verdict).toMatchObject({
|
||||
state: "failing",
|
||||
failures: 3,
|
||||
lastError: "boom a",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not build a streak from three accounts failing once each", () => {
|
||||
const verdict = judgeQueueRuns([
|
||||
...history("a", ["fail"], 0),
|
||||
...history("b", ["fail"], 1),
|
||||
...history("c", ["fail"], 2),
|
||||
]);
|
||||
expect(verdict.state).toBe("pending");
|
||||
});
|
||||
|
||||
it("recovers only when the failing account itself succeeds", () => {
|
||||
const others = history("b", ["ok", "ok"], 0);
|
||||
expect(
|
||||
judgeQueueRuns([
|
||||
...history("a", ["fail", "fail", "fail"], 0),
|
||||
...others,
|
||||
]).state,
|
||||
).toBe("failing");
|
||||
expect(
|
||||
judgeQueueRuns([
|
||||
...history("a", ["ok", "fail", "fail", "fail"], 0),
|
||||
...others,
|
||||
]).state,
|
||||
).toBe("ok");
|
||||
});
|
||||
|
||||
it("reports the account that failed most recently", () => {
|
||||
const verdict = judgeQueueRuns([
|
||||
...history("old", ["fail", "fail", "fail", "fail"], 5),
|
||||
...history("new", ["fail", "fail", "fail"], 0),
|
||||
]);
|
||||
expect(verdict).toMatchObject({ failures: 3, lastError: "boom new" });
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("watchedQueues", () => {
|
||||
|
||||
@@ -123,6 +123,9 @@ export async function runDenseIntradayRetentionForUser(
|
||||
// days for proxy users, preserving the resting signal post-fold.
|
||||
derived_resting_rows_upserted:
|
||||
summary.totals.derivedRestingRowsUpserted,
|
||||
// Backlog days of a server-pulled source that expired the account's
|
||||
// delta cursors rather than leave millions of tombstones.
|
||||
sync_resets_stamped: summary.totals.syncResetsStamped,
|
||||
stopped_early: summary.stoppedEarly,
|
||||
},
|
||||
},
|
||||
|
||||
@@ -317,6 +317,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
|
||||
"total",
|
||||
"trusted_devices_deleted",
|
||||
"unchanged",
|
||||
// The lab date repair: staged facts that would not decrypt, skipped.
|
||||
"undecryptable",
|
||||
"updated",
|
||||
"users",
|
||||
"users_clean_zero",
|
||||
|
||||
@@ -44,7 +44,7 @@ import { annotate } from "@/lib/logging/context";
|
||||
import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
import { dateOnlyAtNoonUtc } from "@/lib/tz/date-only";
|
||||
import { userDayKey } from "@/lib/tz/format";
|
||||
import { resolveUserTimezone } from "@/lib/tz/resolver";
|
||||
import { DEFAULT_TIMEZONE, resolveUserTimezone } from "@/lib/tz/resolver";
|
||||
import type { ObservationFactData } from "@/lib/validations/inbound-documents";
|
||||
|
||||
import {
|
||||
@@ -76,11 +76,25 @@ interface CandidateRow {
|
||||
function candidateQuery(limit: number | null): Prisma.Sql {
|
||||
// The day comparison runs here as well as in the repair, so a value that
|
||||
// really was taken on its confirmation day drops out of the boot check and
|
||||
// the pass converges to an empty set. A zone Postgres does not know is read
|
||||
// as UTC rather than failing the query; the repair re-checks every row in
|
||||
// the zone the app resolves.
|
||||
// the pass converges to an empty set. Both must read the same zone: a
|
||||
// stored zone Postgres does not know falls back the way
|
||||
// `resolveUserTimezone` falls back, to the instance default and then to
|
||||
// `DEFAULT_TIMEZONE`. Reading it as UTC instead kept a value confirmed
|
||||
// near midnight in the candidate set that the repair, in the app's zone,
|
||||
// left alone, so every boot enqueued the job again. The repair still
|
||||
// re-checks every row in the zone the app resolves.
|
||||
return Prisma.sql`
|
||||
WITH "known_zones" AS (SELECT "name" FROM pg_timezone_names)
|
||||
WITH "known_zones" AS (SELECT "name" FROM pg_timezone_names),
|
||||
"instance_zone" AS (
|
||||
SELECT COALESCE(
|
||||
(
|
||||
SELECT s."default_user_timezone" FROM "app_settings" s
|
||||
WHERE s."id" = 'singleton'
|
||||
AND s."default_user_timezone" IN (SELECT "name" FROM "known_zones")
|
||||
),
|
||||
${DEFAULT_TIMEZONE}
|
||||
) AS "name"
|
||||
)
|
||||
SELECT r."id",
|
||||
r."user_id" AS "userId",
|
||||
r."taken_at" AS "takenAt",
|
||||
@@ -106,7 +120,8 @@ function candidateQuery(limit: number | null): Prisma.Sql {
|
||||
<> to_char(
|
||||
(r."taken_at" AT TIME ZONE 'UTC') AT TIME ZONE (
|
||||
CASE WHEN u."timezone" IN (SELECT "name" FROM "known_zones")
|
||||
THEN u."timezone" ELSE 'UTC' END
|
||||
THEN u."timezone"
|
||||
ELSE (SELECT "name" FROM "instance_zone") END
|
||||
),
|
||||
'YYYY-MM-DD'
|
||||
)
|
||||
@@ -128,6 +143,13 @@ export interface LabDocumentDateRepairResult {
|
||||
updated: number;
|
||||
/** Rows left alone: an own date, or already on the document's day. */
|
||||
unchanged: number;
|
||||
/**
|
||||
* Rows whose staged fact would not decrypt (a key retired too early, a
|
||||
* damaged row). Skipped: without the fact there is no telling whether the
|
||||
* value carries a date of its own, and one such row must not stop the
|
||||
* repair of every other account.
|
||||
*/
|
||||
undecryptable: number;
|
||||
}
|
||||
|
||||
/** The repair itself, across every account. */
|
||||
@@ -138,9 +160,17 @@ export async function repairUndatedDocumentLabValues(
|
||||
const touchedUsers = new Set<string>();
|
||||
let updated = 0;
|
||||
let unchanged = 0;
|
||||
// Reported as `undecryptable` in the outcome facts.
|
||||
let decryptFailed = 0;
|
||||
|
||||
for (const row of rows) {
|
||||
const data = decryptFactData(row.dataEncrypted) as ObservationFactData;
|
||||
let data: ObservationFactData;
|
||||
try {
|
||||
data = decryptFactData(row.dataEncrypted) as ObservationFactData;
|
||||
} catch {
|
||||
decryptFailed += 1;
|
||||
continue;
|
||||
}
|
||||
// A value with a date of its own is never touched, whatever its row says.
|
||||
if (statedDay(data.effectiveDate ?? null) !== null) {
|
||||
unchanged += 1;
|
||||
@@ -175,7 +205,12 @@ export async function repairUndatedDocumentLabValues(
|
||||
}
|
||||
|
||||
for (const userId of touchedUsers) invalidateUserHealthScore(userId);
|
||||
return { scanned: rows.length, updated, unchanged };
|
||||
return {
|
||||
scanned: rows.length,
|
||||
updated,
|
||||
unchanged,
|
||||
undecryptable: decryptFailed,
|
||||
};
|
||||
}
|
||||
|
||||
/** The queue handler: one run over every account. */
|
||||
@@ -193,7 +228,7 @@ export async function handleLabDocumentDateRepair(
|
||||
}
|
||||
workerLog(
|
||||
"info",
|
||||
`[lab-document-date-repair] scanned=${result.scanned} updated=${result.updated} unchanged=${result.unchanged}`,
|
||||
`[lab-document-date-repair] scanned=${result.scanned} updated=${result.updated} unchanged=${result.unchanged} undecryptable=${result.undecryptable}`,
|
||||
);
|
||||
annotate({
|
||||
action: { name: "labs.documentDate.repair" },
|
||||
@@ -203,6 +238,7 @@ export async function handleLabDocumentDateRepair(
|
||||
candidates_scanned: result.scanned,
|
||||
updated: result.updated,
|
||||
unchanged: result.unchanged,
|
||||
undecryptable: result.undecryptable,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
/**
|
||||
* Push to every admin when a background job of the instance keeps failing:
|
||||
* once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` runs in a row, a
|
||||
* once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` jobs in a row, a
|
||||
* reminder at most once a day while it keeps failing, and once when it runs
|
||||
* successfully again. Nothing else.
|
||||
*
|
||||
@@ -12,11 +12,28 @@
|
||||
* account-facing status of its own (the `system` audience in
|
||||
* `subsystem-surface.ts`).
|
||||
*
|
||||
* The streak is read from pg-boss's own ledger, newest terminal run first: a
|
||||
* run that failed (or completed and said every candidate failed,
|
||||
* `did.all_failed`) adds its attempts, retries included, and the first run
|
||||
* that succeeded ends the streak. A queue with no terminal run left in the
|
||||
* ledger is not judged at all.
|
||||
* The streak is read from pg-boss's own ledger, newest terminal job first,
|
||||
* and it counts jobs, not attempts: a job that used up its retries is one
|
||||
* failure. A job that failed, or completed and said every candidate failed
|
||||
* (`did.all_failed`), adds one; the first job that succeeded ends it.
|
||||
*
|
||||
* Many of these queues run one job per account (a rollup recompute, a
|
||||
* backfill, a score), so the streak is kept per job key: the account the job
|
||||
* carries (`data.userId`), or the whole queue for a job that carries none.
|
||||
* Not the `singletonKey`: it often names a bucket, a window or a single
|
||||
* workout, which makes nearly every job its own key, and a queue failing on
|
||||
* every job would then never build a streak. A queue is failing while any
|
||||
* key has failed `SYSTEM_JOB_FAILURE_STREAK` jobs in a row, and it has
|
||||
* recovered only when no key is failing and a key that had failed ran
|
||||
* successfully since; another account's success says nothing about the one
|
||||
* that fails. One account failing every hour while the rest succeed is
|
||||
* therefore one problem notice and one recovery, not a notice each hour.
|
||||
*
|
||||
* A reminder needs a failure newer than the notice before it. A queue that
|
||||
* keeps failing is reminded once a day; a one-shot job (a boot backfill) that
|
||||
* failed and does not run again is announced once and then left to the admin
|
||||
* page, instead of a reminder every day until pg-boss deletes its row. A
|
||||
* queue with nothing left in the ledger to judge is forgotten silently.
|
||||
*
|
||||
* The memory is one row per announced queue (`SystemJobAlertNotice`), claimed
|
||||
* with a compare-and-set before the notice is sent, exactly as the readiness
|
||||
@@ -41,21 +58,24 @@ import {
|
||||
type StoredNotice,
|
||||
} from "@/lib/recovery/readiness-notify";
|
||||
|
||||
/** Failed runs in a row, retries included, before the admins are told. */
|
||||
/** Failed jobs in a row, of one job key, before the admins are told. */
|
||||
export const SYSTEM_JOB_FAILURE_STREAK = 3;
|
||||
|
||||
/** Newest terminal runs read per queue. */
|
||||
const LOOKBACK_ROWS = 50;
|
||||
/** Newest terminal jobs read per queue, across every key. */
|
||||
const LOOKBACK_ROWS = 5000;
|
||||
|
||||
/** Newest terminal jobs judged per job key. */
|
||||
const LOOKBACK_ROWS_PER_KEY = 20;
|
||||
|
||||
/** Where the notice takes the admin: the page with the failing-jobs card. */
|
||||
const ADMIN_STATUS_PATH = "/admin/system-status";
|
||||
|
||||
/** One terminal run as the ledger holds it, newest first. */
|
||||
/** One terminal job as the ledger holds it. */
|
||||
export interface TerminalRun {
|
||||
/** The account the job ran for (`data.userId`), or "" for the queue. */
|
||||
key: string;
|
||||
state: "completed" | "failed";
|
||||
/** Retries pg-boss ran before this run settled. */
|
||||
retryCount: number;
|
||||
/** The run completed but reported that every candidate failed. */
|
||||
/** The job completed but reported that every candidate failed. */
|
||||
allFailed: boolean;
|
||||
completedOn: Date;
|
||||
error: string | null;
|
||||
@@ -65,39 +85,66 @@ export interface TerminalRun {
|
||||
/** What the ledger says about one queue. */
|
||||
export interface QueueVerdict {
|
||||
/**
|
||||
* `failing`: the streak reached the threshold. `ok`: the newest run, or a
|
||||
* run after the last failures, succeeded. `pending`: fewer failures than
|
||||
* the threshold and no success to judge by. `inactive`: no run left.
|
||||
* `failing`: some key reached the streak. `ok`: no key is failing, and a
|
||||
* key that had failed has since run successfully. `pending`: a failure
|
||||
* short of the streak, and no recovery to report. `inactive`: nothing to
|
||||
* report, no job left or none that failed.
|
||||
*/
|
||||
state: "failing" | "ok" | "pending" | "inactive";
|
||||
/** Failed attempts in a row, newest first. */
|
||||
/** Failed jobs in a row of the key that failed most recently. */
|
||||
failures: number;
|
||||
/** The newest failure's message, redacted and truncated. */
|
||||
/** That key's newest failure, redacted and truncated. */
|
||||
lastError: string;
|
||||
/** When the newest failure of a failing key settled, if any key is. */
|
||||
lastFailureAt: Date | null;
|
||||
}
|
||||
|
||||
/** Pure: judge one queue's newest terminal runs. */
|
||||
function failed(run: TerminalRun): boolean {
|
||||
return run.state === "failed" || run.allFailed;
|
||||
}
|
||||
|
||||
/** Pure: judge one queue's terminal jobs, in any order. */
|
||||
export function judgeQueueRuns(runs: readonly TerminalRun[]): QueueVerdict {
|
||||
if (runs.length === 0)
|
||||
return { state: "inactive", failures: 0, lastError: "" };
|
||||
let failures = 0;
|
||||
let lastError: string | null = null;
|
||||
let succeeded = false;
|
||||
const byKey = new Map<string, TerminalRun[]>();
|
||||
for (const run of runs) {
|
||||
if (run.state === "completed" && !run.allFailed) {
|
||||
succeeded = true;
|
||||
break;
|
||||
}
|
||||
failures += run.retryCount + 1;
|
||||
lastError ??= presentError(run.error, run.expireSeconds);
|
||||
const list = byKey.get(run.key) ?? [];
|
||||
list.push(run);
|
||||
byKey.set(run.key, list);
|
||||
}
|
||||
const state =
|
||||
failures >= SYSTEM_JOB_FAILURE_STREAK
|
||||
? "failing"
|
||||
: succeeded
|
||||
? "ok"
|
||||
: "pending";
|
||||
return { state, failures, lastError: lastError ?? "" };
|
||||
let failing: { streak: number; newest: TerminalRun } | null = null;
|
||||
let recovered = false;
|
||||
let anyFailure = false;
|
||||
for (const list of byKey.values()) {
|
||||
list.sort((a, b) => b.completedOn.getTime() - a.completedOn.getTime());
|
||||
let streak = 0;
|
||||
while (streak < list.length && failed(list[streak])) streak += 1;
|
||||
if (list.some(failed)) anyFailure = true;
|
||||
if (streak === 0) {
|
||||
// Newest job succeeded: a recovery, when this key had failed before.
|
||||
if (list.some(failed)) recovered = true;
|
||||
continue;
|
||||
}
|
||||
if (
|
||||
streak >= SYSTEM_JOB_FAILURE_STREAK &&
|
||||
(failing === null ||
|
||||
list[0].completedOn.getTime() > failing.newest.completedOn.getTime())
|
||||
) {
|
||||
failing = { streak, newest: list[0] };
|
||||
}
|
||||
}
|
||||
if (failing) {
|
||||
return {
|
||||
state: "failing",
|
||||
failures: failing.streak,
|
||||
lastError: presentError(
|
||||
failing.newest.error,
|
||||
failing.newest.expireSeconds,
|
||||
),
|
||||
lastFailureAt: failing.newest.completedOn,
|
||||
};
|
||||
}
|
||||
const state = recovered ? "ok" : anyFailure ? "pending" : "inactive";
|
||||
return { state, failures: 0, lastError: "", lastFailureAt: null };
|
||||
}
|
||||
|
||||
/** The queues this alert watches: every one whose failure has no other surface. */
|
||||
@@ -108,9 +155,11 @@ export function watchedQueues(): string[] {
|
||||
}
|
||||
|
||||
/**
|
||||
* The newest terminal runs of each queue, newest first, one statement over
|
||||
* pg-boss's `(name, completed_on)` index. `null` when the queue schema is not
|
||||
* there (a web-only deployment): no queue to ask is not a healthy queue.
|
||||
* The newest terminal jobs of each queue, keyed by account, one statement
|
||||
* over pg-boss's `(name, completed_on)` index: the newest `LOOKBACK_ROWS` of
|
||||
* the queue, and of those the newest `LOOKBACK_ROWS_PER_KEY` of each key.
|
||||
* `null` when the queue schema is not there (a web-only deployment): no
|
||||
* queue to ask is not a healthy queue.
|
||||
*/
|
||||
export async function readTerminalRuns(
|
||||
queues: readonly string[],
|
||||
@@ -119,40 +168,48 @@ export async function readTerminalRuns(
|
||||
const rows = await prisma.$queryRaw<
|
||||
Array<{
|
||||
name: string;
|
||||
job_key: string;
|
||||
state: string;
|
||||
retry_count: number;
|
||||
all_failed: string | null;
|
||||
completed_on: Date;
|
||||
error: string | null;
|
||||
expire_seconds: number | null;
|
||||
}>
|
||||
>`
|
||||
SELECT q.name, j.state, j.retry_count, j.all_failed, j.completed_on,
|
||||
j.error, j.expire_seconds
|
||||
SELECT q.name, k.job_key, k.state, k.all_failed, k.completed_on,
|
||||
k.error, k.expire_seconds
|
||||
FROM unnest(${queues as string[]}::text[]) AS q(name)
|
||||
CROSS JOIN LATERAL (
|
||||
SELECT state, retry_count, completed_on, expire_seconds,
|
||||
output->'did'->>'all_failed' AS all_failed,
|
||||
COALESCE(
|
||||
output->>'message',
|
||||
output->'value'->>'message',
|
||||
output::text
|
||||
) AS error
|
||||
FROM pgboss.job
|
||||
WHERE name = q.name
|
||||
AND completed_on IS NOT NULL
|
||||
AND state IN ('completed', 'failed')
|
||||
ORDER BY completed_on DESC
|
||||
LIMIT ${LOOKBACK_ROWS}
|
||||
) j
|
||||
ORDER BY q.name, j.completed_on DESC
|
||||
SELECT j.*,
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY j.job_key ORDER BY j.completed_on DESC
|
||||
) AS key_rank
|
||||
FROM (
|
||||
SELECT state, completed_on, expire_seconds,
|
||||
COALESCE(data->>'userId', '') AS job_key,
|
||||
output->'did'->>'all_failed' AS all_failed,
|
||||
COALESCE(
|
||||
output->>'message',
|
||||
output->'value'->>'message',
|
||||
output::text
|
||||
) AS error
|
||||
FROM pgboss.job
|
||||
WHERE name = q.name
|
||||
AND completed_on IS NOT NULL
|
||||
AND state IN ('completed', 'failed')
|
||||
ORDER BY completed_on DESC
|
||||
LIMIT ${LOOKBACK_ROWS}
|
||||
) j
|
||||
) k
|
||||
WHERE k.key_rank <= ${LOOKBACK_ROWS_PER_KEY}
|
||||
ORDER BY q.name, k.completed_on DESC
|
||||
`;
|
||||
const byQueue = new Map<string, TerminalRun[]>();
|
||||
for (const row of rows) {
|
||||
const runs = byQueue.get(row.name) ?? [];
|
||||
runs.push({
|
||||
key: row.job_key,
|
||||
state: row.state === "failed" ? "failed" : "completed",
|
||||
retryCount: Number(row.retry_count ?? 0),
|
||||
allFailed: row.all_failed === "true",
|
||||
completedOn: row.completed_on,
|
||||
error: row.error,
|
||||
@@ -222,6 +279,16 @@ export async function notifySystemJobFailures(
|
||||
// With no admin yet there is nobody to tell, so nothing is claimed: the
|
||||
// first admin hears about a queue that is still failing.
|
||||
if (!decision.next || admins.length === 0) continue;
|
||||
// Nothing failed since the last notice: a job that does not run again is
|
||||
// not news a second time.
|
||||
if (
|
||||
decision.send === "reminder" &&
|
||||
stored?.notifiedAt &&
|
||||
verdict.lastFailureAt &&
|
||||
verdict.lastFailureAt.getTime() <= stored.notifiedAt.getTime()
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Claim: only the evaluation that moves the row from what it read sends.
|
||||
const claimed = row
|
||||
|
||||
@@ -14,6 +14,14 @@
|
||||
* Deleting a systolic or a diastolic takes the whole reading with it.
|
||||
* Deleting a pulse takes only the pulse: a pulse is also a value of its own,
|
||||
* and removing one is no reason to remove a blood pressure.
|
||||
*
|
||||
* An aggregate row is never part of a reading. The dense fold stores an
|
||||
* hour of heart rate as one `stats:` row at the local HH:30 of that hour
|
||||
* (`canonicalHourlyTimestamp`), under the same source as the samples it
|
||||
* folded, and the client's hourly heart-rate bucket is a `stats:` row as
|
||||
* well. A cuff reading taken at half past, from the same source, shares its
|
||||
* instant with that row; it is still an hour of somebody's heart rate, not
|
||||
* the cuff's pulse, and a delete must not take it.
|
||||
*/
|
||||
import type { MeasurementType, Prisma } from "@/generated/prisma/client";
|
||||
|
||||
@@ -30,6 +38,9 @@ const BP_READING_TYPES: MeasurementType[] = [
|
||||
"PULSE",
|
||||
];
|
||||
|
||||
/** The externalId prefix every aggregate (folded or bucketed) row carries. */
|
||||
const AGGREGATE_EXTERNAL_ID_PREFIX = "stats:";
|
||||
|
||||
export interface ReadingAnchor {
|
||||
id: string;
|
||||
type: MeasurementType | string;
|
||||
@@ -70,6 +81,20 @@ export async function bloodPressureSiblings(
|
||||
userId,
|
||||
id: { notIn: rows.map((r) => r.id) },
|
||||
type: { in: BP_READING_TYPES },
|
||||
// Spelled with the NULL arm: a bare `NOT startsWith` is NULL, not
|
||||
// true, for the NULL externalId of a manual row and would drop it.
|
||||
AND: [
|
||||
{
|
||||
OR: [
|
||||
{ externalId: null },
|
||||
{
|
||||
NOT: {
|
||||
externalId: { startsWith: AGGREGATE_EXTERNAL_ID_PREFIX },
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
OR: anchors.map((a) => ({
|
||||
source: a.source as Prisma.MeasurementWhereInput["source"],
|
||||
measuredAt: a.measuredAt,
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
/**
|
||||
* How the dense fold (`dense-intraday-retention.ts`) removes the raw samples
|
||||
* of a day it folded, and how the change feed learns of it.
|
||||
*
|
||||
* Apple Health and Health Connect samples are deleted outright (v1.42), like
|
||||
* the cumulative drain's rows, where they used to be tombstoned for the
|
||||
* 75-day retention. They come from the person's own health store and no
|
||||
* client reads them from `/api/sync/changes`, so the tombstone told nobody
|
||||
* anything; it only held the re-upload of a folded sample off, and the
|
||||
* `folded_window` ingest guard (`folded-window.ts`) does that now from the
|
||||
* live hourly row. The tombstones made up most of the table and its indexes.
|
||||
*
|
||||
* A server-pulled source (`SERVER_PULLED_FOLD_SOURCES`: Google Health, Oura)
|
||||
* is different. The feed served its samples, and a client mirroring the feed
|
||||
* keeps them until the feed says they are gone. So:
|
||||
* - the nightly increment, a day at most {@link FOLD_FEED_TOMBSTONE_DAYS}
|
||||
* past the retention boundary, is tombstoned. The feed reports every
|
||||
* sample, and the 75-day tombstone cleanup purges them later. The
|
||||
* classifier of compaction tombstones leaves these sources out, so the
|
||||
* feed, the backlog purge and the restore treat them like any deletion.
|
||||
* - an older day is backlog: an account's first fold, an old range a
|
||||
* backfill brought in, nights the fold did not run. Tombstoning that
|
||||
* would put millions of rows back into the table (#1297), so its samples
|
||||
* are deleted outright and the account's delta cursors expire instead
|
||||
* (`stampSyncReset`). A client that pulled them re-initialises rather
|
||||
* than keep them beside the hourly means.
|
||||
* The tombstones share one deletion instant and keep their `syncVersion`, so
|
||||
* the re-fold reads them as fold leftovers (`fold-constituents.ts`), exactly
|
||||
* like the Apple Health tombstones of the releases before v1.42.
|
||||
*/
|
||||
import type { MeasurementSource, Prisma } from "@/generated/prisma/client";
|
||||
import { stampSyncReset } from "@/lib/sync/reset";
|
||||
|
||||
import { SERVER_PULLED_FOLD_SOURCES } from "./dense-intraday-sources";
|
||||
import { absorbIntoFold } from "./fold-constituents";
|
||||
|
||||
/** Days past the retention boundary that still count as the nightly fold. */
|
||||
const FOLD_FEED_TOMBSTONE_DAYS = 7;
|
||||
|
||||
const DAY_MS = 86_400_000;
|
||||
|
||||
export type FoldRemoval = "delete" | "tombstone" | "delete-and-reset";
|
||||
|
||||
/** How a fold removes the samples of one local day. Pure. */
|
||||
export function foldRemovalFor(
|
||||
source: MeasurementSource,
|
||||
dayStart: Date,
|
||||
retentionDays: number,
|
||||
now: Date = new Date(),
|
||||
): FoldRemoval {
|
||||
if (!SERVER_PULLED_FOLD_SOURCES.has(source)) return "delete";
|
||||
const nightly =
|
||||
now.getTime() - (retentionDays + FOLD_FEED_TOMBSTONE_DAYS) * DAY_MS;
|
||||
return dayStart.getTime() >= nightly ? "tombstone" : "delete-and-reset";
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the folded samples `ids`, never one of `keep` (the hourly rows the
|
||||
* fold just adopted), inside the fold's transaction. Returns how many went
|
||||
* and whether the account's delta cursors were expired. The caller stamps
|
||||
* the reset once more after the commit, so a cursor issued while the
|
||||
* transaction ran expires too; the stamp in here survives a crash after it.
|
||||
*/
|
||||
export async function removeFoldedSamples(
|
||||
tx: Prisma.TransactionClient,
|
||||
input: {
|
||||
userId: string;
|
||||
removal: FoldRemoval;
|
||||
ids: readonly string[];
|
||||
keep: readonly string[];
|
||||
},
|
||||
): Promise<{ removed: number; resetFeed: boolean }> {
|
||||
if (input.removal === "tombstone") {
|
||||
const removed = await absorbIntoFold(
|
||||
tx,
|
||||
input.ids.filter((id) => !input.keep.includes(id)),
|
||||
);
|
||||
return { removed, resetFeed: false };
|
||||
}
|
||||
const del = await tx.measurement.deleteMany({
|
||||
where: {
|
||||
id: { in: [...input.ids], notIn: [...input.keep] },
|
||||
deletedAt: null,
|
||||
},
|
||||
});
|
||||
const resetFeed = input.removal === "delete-and-reset" && del.count > 0;
|
||||
if (resetFeed) await stampSyncReset(tx, input.userId);
|
||||
return { removed: del.count, resetFeed };
|
||||
}
|
||||
@@ -86,6 +86,8 @@ import {
|
||||
DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
|
||||
type DenseIntradayFoldSource,
|
||||
} from "./dense-intraday-sources";
|
||||
import { foldRemovalFor, removeFoldedSamples } from "./dense-fold-removal";
|
||||
import { stampSyncReset } from "@/lib/sync/reset";
|
||||
import {
|
||||
absorbIntoFold,
|
||||
foldConstituentHorizon,
|
||||
@@ -433,6 +435,12 @@ export interface DenseIntradayRetentionSummary {
|
||||
* resting signal the raw-PULSE proxy can no longer compute after fold.
|
||||
*/
|
||||
derivedRestingRowsUpserted: number;
|
||||
/**
|
||||
* Folded days of a server-pulled source whose samples were deleted
|
||||
* outright, each of which expired the account's delta cursors
|
||||
* (`dense-fold-removal.ts`).
|
||||
*/
|
||||
syncResetsStamped: number;
|
||||
/**
|
||||
* Hours that already had a live hourly row whose samples could not be
|
||||
* shown to be complete (`fold-constituents.ts`): the stored mean and the
|
||||
@@ -458,7 +466,9 @@ export interface DenseIntradayRetentionOptions extends ConsolidationOptions {
|
||||
* Run the dense intra-day retention drain. Folds per-sample dense-tier
|
||||
* rows older than the retention window into hourly-mean `stats:` rows
|
||||
* (user-local hours, anchored at local HH:30) and deletes the raw rows
|
||||
* (soft-deleted before v1.42); rows inside the window stay raw. Idempotent —
|
||||
* (soft-deleted before v1.42; a server-pulled source's nightly increment is
|
||||
* still tombstoned, see `dense-fold-removal.ts`); rows inside the window
|
||||
* stay raw. Idempotent —
|
||||
* re-invocation after a successful pass converges to zero work because the
|
||||
* folded rows are gone from the live scan, and the minted hourly
|
||||
* rows are excluded by the `NOT startsWith('stats:')` predicate.
|
||||
@@ -487,6 +497,7 @@ export async function runDenseIntradayRetention(
|
||||
hourlyRowsUpserted: 0,
|
||||
dailyRowsRetired: 0,
|
||||
derivedRestingRowsUpserted: 0,
|
||||
syncResetsStamped: 0,
|
||||
hoursLeftAsStored: 0,
|
||||
},
|
||||
};
|
||||
@@ -647,8 +658,10 @@ export async function runDenseIntradayRetention(
|
||||
hoursLeftAsStored: number;
|
||||
/** Every raw sample of the day, when that is provably all of it. */
|
||||
wholeDay: FoldConstituent[] | null;
|
||||
resetFeed: boolean;
|
||||
}> => {
|
||||
let removed = 0;
|
||||
let resetFeed = false;
|
||||
let retiredDaily = false;
|
||||
let slotsWritten = 0;
|
||||
let hoursLeftAsStored = 0;
|
||||
@@ -806,24 +819,20 @@ export async function runDenseIntradayRetention(
|
||||
// anchor is the row just adopted as that hour's mean, so removing
|
||||
// it would erase the fold.
|
||||
//
|
||||
// v1.42 — deleted outright, like the cumulative drain's rows,
|
||||
// where they used to be tombstoned for the 75-day retention. No
|
||||
// client ever pulled these rows, so the tombstone told nobody
|
||||
// anything; it only held the re-upload of a folded sample off,
|
||||
// and the `folded_window` ingest guard (`folded-window.ts`) does
|
||||
// that now from the live hourly row. The tombstones made up most
|
||||
// of the table and its indexes.
|
||||
const del = await tx.measurement.deleteMany({
|
||||
where: {
|
||||
id: { in: deleteIds, notIn: canonicalRowIds },
|
||||
deletedAt: null,
|
||||
},
|
||||
// How, and whether the change feed hears of it, depends on the
|
||||
// source and the day's age: `dense-fold-removal.ts`.
|
||||
const dropped = await removeFoldedSamples(tx, {
|
||||
userId,
|
||||
removal: foldRemovalFor(source, dayStart, retentionDays),
|
||||
ids: deleteIds,
|
||||
keep: canonicalRowIds,
|
||||
});
|
||||
resetFeed = dropped.resetFeed;
|
||||
const absorbed = await absorbIntoFold(
|
||||
tx,
|
||||
absorbIds.filter((id) => !canonicalRowIds.includes(id)),
|
||||
);
|
||||
removed = del.count + absorbed;
|
||||
removed = dropped.removed + absorbed;
|
||||
|
||||
// The day's raw samples, for the derived resting figure: the rows
|
||||
// in hand for a day folded here for the first time; for a day
|
||||
@@ -844,12 +853,14 @@ export async function runDenseIntradayRetention(
|
||||
];
|
||||
}
|
||||
}, FOLD_TRANSACTION_OPTIONS);
|
||||
if (resetFeed) await stampSyncReset(pc, userId);
|
||||
return {
|
||||
removed,
|
||||
retiredDaily,
|
||||
slotsWritten,
|
||||
hoursLeftAsStored,
|
||||
wholeDay,
|
||||
resetFeed,
|
||||
};
|
||||
};
|
||||
|
||||
@@ -868,6 +879,7 @@ export async function runDenseIntradayRetention(
|
||||
summary.totals.hourlyRowsUpserted += foldResult.slotsWritten;
|
||||
summary.totals.hoursLeftAsStored += foldResult.hoursLeftAsStored;
|
||||
if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1;
|
||||
if (foldResult.resetFeed) summary.totals.syncResetsStamped += 1;
|
||||
|
||||
// iOS#34 — preserve the resting signal for PROXY users. For a user
|
||||
// with zero native `RESTING_HEART_RATE` rows, the read-path resolver
|
||||
@@ -997,7 +1009,7 @@ export async function runDenseIntradayRetention(
|
||||
}
|
||||
|
||||
log(
|
||||
`[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted} hoursLeftAsStored=${summary.totals.hoursLeftAsStored}${options.dryRun ? " (dry-run)" : ""}`,
|
||||
`[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted} syncResetsStamped=${summary.totals.syncResetsStamped} hoursLeftAsStored=${summary.totals.hoursLeftAsStored}${options.dryRun ? " (dry-run)" : ""}`,
|
||||
);
|
||||
|
||||
return summary;
|
||||
|
||||
@@ -84,3 +84,18 @@ export type DenseIntradayFoldSource =
|
||||
export const DENSE_INTRADAY_FOLD_SOURCES = Object.keys(
|
||||
DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
|
||||
) as readonly DenseIntradayFoldSource[];
|
||||
|
||||
/**
|
||||
* The fold sources whose rows the server pulled from a provider itself. A
|
||||
* client can hold one of these rows only because `/api/sync/changes` served
|
||||
* it, so when the fold removes them the feed has to say so: the nightly fold
|
||||
* leaves tombstones for them, and a fold of an older backlog expires the
|
||||
* account's delta cursors instead (`dense-intraday-retention.ts`).
|
||||
*
|
||||
* Apple Health and Health Connect samples come from the person's own health
|
||||
* store, uploaded by the device or imported from its export, and the fold has
|
||||
* deleted them outright since v1.42. No client reads the feed for them: the
|
||||
* iOS app reads its own HealthKit store and has no feed consumer.
|
||||
*/
|
||||
export const SERVER_PULLED_FOLD_SOURCES: ReadonlySet<MeasurementSource> =
|
||||
new Set<DenseIntradayFoldSource>(["GOOGLE_HEALTH", "OURA"]);
|
||||
@@ -79,6 +79,7 @@ import {
|
||||
import {
|
||||
DENSE_INTRADAY_FOLD_SOURCES,
|
||||
DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
|
||||
SERVER_PULLED_FOLD_SOURCES,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
} from "@/lib/measurements/dense-intraday-sources";
|
||||
|
||||
@@ -475,12 +476,17 @@ export interface TombstoneCandidateRow extends FoldedWindowCandidateRow {
|
||||
* enough: an older release folded a day in two runs, and a person who
|
||||
* deleted a sample of such a half-folded day deleted something the fold had
|
||||
* not taken. That deletion is class B.
|
||||
*
|
||||
* A server-pulled source (`SERVER_PULLED_FOLD_SOURCES`) has no class A. The
|
||||
* fold tombstones its samples so the change feed reports them to a client
|
||||
* that pulled them, which is exactly what this class is kept out of.
|
||||
*/
|
||||
export function compactionTombstoneCoverId(
|
||||
row: TombstoneCandidateRow,
|
||||
tz: string,
|
||||
): string | null {
|
||||
if (row.deletedAt === null) return null;
|
||||
if (SERVER_PULLED_FOLD_SOURCES.has(row.source)) return null;
|
||||
if (!isFoldedWindowCandidate(row, row.deletedAt)) return null;
|
||||
if (!isWindowFoldedAt(row, row.deletedAt, tz)) return null;
|
||||
return coveringStatsExternalId(row, tz);
|
||||
|
||||
@@ -24,6 +24,7 @@
|
||||
import { activeAccountSql } from "@/lib/account-deletion/active-account";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import {
|
||||
applyStatementBudget,
|
||||
recomputeUserRollups,
|
||||
@@ -199,13 +200,24 @@ export async function sendRollupFullBackfillContinuation(
|
||||
* leaves such an account alone (`ConsolidationOptions.deferUser`): it deletes
|
||||
* the raw samples a fold still has to read, and a year the full backfill
|
||||
* reached after the dense fold would take its day figures from hourly means.
|
||||
* False when the queue schema cannot be read, as a web-only deployment has
|
||||
* no full backfill either.
|
||||
*
|
||||
* False when there is no queue table (a web-only deployment has no full
|
||||
* backfill either). Any other failure to read the queue counts as pending:
|
||||
* the dense fold's deletion cannot be undone, and deferring an account costs
|
||||
* one night. The failure is annotated on the caller's event.
|
||||
*
|
||||
* A full fold that failed for good (retries spent) is not pending, and the
|
||||
* dense fold goes ahead on that account; the system job alert reports the
|
||||
* failed backfill, and the next boot's discovery enqueues it again.
|
||||
*/
|
||||
export async function isRollupFullBackfillPending(
|
||||
userId: string,
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const [queue] = await prisma.$queryRaw<Array<{ present: boolean }>>`
|
||||
SELECT to_regclass('pgboss.job') IS NOT NULL AS present
|
||||
`;
|
||||
if (!queue?.present) return false;
|
||||
const rows = await prisma.$queryRaw<Array<{ n: number }>>`
|
||||
SELECT count(*)::int AS n
|
||||
FROM pgboss.job
|
||||
@@ -214,8 +226,15 @@ export async function isRollupFullBackfillPending(
|
||||
AND data->>'userId' = ${userId}
|
||||
`;
|
||||
return (rows[0]?.n ?? 0) > 0;
|
||||
} catch {
|
||||
return false;
|
||||
} catch (err) {
|
||||
annotate({
|
||||
meta: {
|
||||
rollup_backfill_pending_check_failed: true,
|
||||
rollup_backfill_pending_check_error:
|
||||
err instanceof Error ? err.message.slice(0, 200) : String(err),
|
||||
},
|
||||
});
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
/**
|
||||
* `POST /api/admin/rollups/recompute` for one account, against Postgres.
|
||||
*
|
||||
* The single-account mode exists for the accounts the boot backfill could not
|
||||
* finish, which are the large ones. It folds the account in the request, and
|
||||
* it must give every statement the same budget the boot job gives it
|
||||
* (`ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS`): under the database's ordinary
|
||||
* statement limit the yearly fold of such an account times out, which is the
|
||||
* very failure the operator is trying to get past. The fold itself runs for
|
||||
* real; the wrapper only records what the route handed it.
|
||||
*/
|
||||
import { NextRequest } from "next/server";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS } from "@/lib/rollups/full-backfill";
|
||||
|
||||
import { cookieJar } from "./mock-next-headers";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
const folds = vi.hoisted(() => ({
|
||||
calls: [] as Array<{ userId: string; statementTimeoutMs?: number }>,
|
||||
}));
|
||||
|
||||
vi.mock("next/headers", async () => {
|
||||
const { cookieJar, headerJar } = await import("./mock-next-headers");
|
||||
return {
|
||||
headers: vi.fn(async () => ({
|
||||
get: (name: string) => headerJar.get(name.toLowerCase()) ?? null,
|
||||
})),
|
||||
cookies: vi.fn(async () => ({
|
||||
get: (name: string) => {
|
||||
const value = cookieJar.get(name);
|
||||
return value ? { name, value } : undefined;
|
||||
},
|
||||
set: (name: string, value: string) => {
|
||||
cookieJar.set(name, value);
|
||||
},
|
||||
delete: (name: string) => {
|
||||
cookieJar.delete(name);
|
||||
},
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/rollups/full-backfill", async (importOriginal) => {
|
||||
const actual =
|
||||
await importOriginal<typeof import("@/lib/rollups/full-backfill")>();
|
||||
return {
|
||||
...actual,
|
||||
foldUserRollupHistory: async (
|
||||
...args: Parameters<typeof actual.foldUserRollupHistory>
|
||||
) => {
|
||||
folds.calls.push({
|
||||
userId: args[0],
|
||||
statementTimeoutMs: args[1]?.statementTimeoutMs,
|
||||
});
|
||||
return actual.foldUserRollupHistory(...args);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const ADMIN = "user-admin-recompute";
|
||||
const TARGET = "user-recompute-target";
|
||||
|
||||
beforeEach(async () => {
|
||||
const prisma = getPrismaClient();
|
||||
await truncateAllTables(prisma);
|
||||
cookieJar.clear();
|
||||
folds.calls.length = 0;
|
||||
await prisma.user.createMany({
|
||||
data: [
|
||||
{
|
||||
id: ADMIN,
|
||||
username: "admin-recompute",
|
||||
email: "admin-recompute@example.test",
|
||||
role: "ADMIN",
|
||||
},
|
||||
{
|
||||
id: TARGET,
|
||||
username: "recompute-target",
|
||||
email: "recompute-target@example.test",
|
||||
},
|
||||
],
|
||||
});
|
||||
const session = await prisma.session.create({
|
||||
data: { userId: ADMIN, expiresAt: new Date(Date.now() + 3_600_000) },
|
||||
});
|
||||
cookieJar.set("healthlog_session", session.id);
|
||||
await prisma.measurement.create({
|
||||
data: {
|
||||
userId: TARGET,
|
||||
type: "WEIGHT",
|
||||
value: 80,
|
||||
unit: "kg",
|
||||
source: "MANUAL",
|
||||
measuredAt: new Date(Date.now() - 30 * 86_400_000),
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
describe("POST /api/admin/rollups/recompute for one account", () => {
|
||||
it("folds it with the boot job's statement budget", async () => {
|
||||
const { POST } = await import("@/app/api/admin/rollups/recompute/route");
|
||||
const res = await POST(
|
||||
new NextRequest("http://localhost/api/admin/rollups/recompute", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ userId: TARGET }),
|
||||
}),
|
||||
);
|
||||
expect(res.status).toBe(200);
|
||||
const body = (await res.json()) as {
|
||||
data: { mode: string; rowsUpserted: number };
|
||||
};
|
||||
expect(body.data.mode).toBe("single");
|
||||
expect(body.data.rowsUpserted).toBeGreaterThan(0);
|
||||
|
||||
expect(folds.calls).toEqual([
|
||||
{
|
||||
userId: TARGET,
|
||||
statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
},
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -10,11 +10,16 @@
|
||||
* its own syncVersion (so the sync feed carries each tombstone), and the
|
||||
* undo, which names only the row the person tapped, brings the whole
|
||||
* reading back. A reading of another source at the same instant, a reading
|
||||
* a minute later, and a pulse deleted on its own are left alone.
|
||||
* a minute later, and a pulse deleted on its own are left alone. So is the
|
||||
* folded hour of heart rate the dense fold anchors at the local HH:30: a
|
||||
* reading taken at half past shares its instant and its source, and is
|
||||
* still not the same thing.
|
||||
*/
|
||||
import { NextRequest } from "next/server";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { canonicalHourlyTimestamp } from "@/lib/measurements/consolidation-tz";
|
||||
|
||||
import { cookieJar, headerJar } from "./mock-next-headers";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
@@ -195,4 +200,56 @@ describe("deleting a blood-pressure reading", () => {
|
||||
// The pulse deleted at another moment stays deleted.
|
||||
expect(await deletedIds()).toEqual(["next-pulse"]);
|
||||
});
|
||||
|
||||
it("BP at :30 beside a folded hour: leaves the hour of heart rate alone", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
// 09:30 local (Europe/Berlin, the default zone) on a day the fold took:
|
||||
// the anchor of that hour's mean.
|
||||
const halfPast = canonicalHourlyTimestamp("2026-10-20", 9, "Europe/Berlin");
|
||||
await prisma.measurement.createMany({
|
||||
data: [
|
||||
{
|
||||
id: "ah-sys",
|
||||
userId: USER,
|
||||
type: "BLOOD_PRESSURE_SYS",
|
||||
value: 134,
|
||||
unit: "mmHg",
|
||||
source: "APPLE_HEALTH",
|
||||
measuredAt: halfPast,
|
||||
externalId: "uuid-ah-sys",
|
||||
},
|
||||
{
|
||||
id: "ah-dia",
|
||||
userId: USER,
|
||||
type: "BLOOD_PRESSURE_DIA",
|
||||
value: 86,
|
||||
unit: "mmHg",
|
||||
source: "APPLE_HEALTH",
|
||||
measuredAt: halfPast,
|
||||
externalId: "uuid-ah-dia",
|
||||
},
|
||||
{
|
||||
id: "folded-hour",
|
||||
userId: USER,
|
||||
type: "PULSE",
|
||||
value: 71,
|
||||
unit: "bpm",
|
||||
source: "APPLE_HEALTH",
|
||||
measuredAt: halfPast,
|
||||
externalId: "stats:HKQuantityTypeIdentifierHeartRate:2026-10-20T09",
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
await deleteOne("ah-sys");
|
||||
expect(await deletedIds()).toEqual(["ah-dia", "ah-sys"]);
|
||||
expect((await rows())["folded-hour"]!.syncVersion).toBe(1);
|
||||
|
||||
const { restored } = await post("restore", ["ah-sys"]);
|
||||
expect(restored).toBe(2);
|
||||
|
||||
const { deleted } = await post("bulk-delete", ["ah-dia"]);
|
||||
expect(deleted).toBe(2);
|
||||
expect(await deletedIds()).toEqual(["ah-dia", "ah-sys"]);
|
||||
});
|
||||
});
|
||||
@@ -8,12 +8,15 @@
|
||||
* at noon UTC, and leaves alone a value with a date of its own, a value that
|
||||
* really was taken on its confirmation day in the owner's zone, one the
|
||||
* person re-dated since, and one whose document has no date. A second run
|
||||
* changes nothing.
|
||||
* changes nothing. A fact that does not decrypt is skipped and counted
|
||||
* rather than stopping the run, and an account zone Postgres does not know
|
||||
* is read the way the app reads it, so the candidate set still converges.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it } from "vitest";
|
||||
|
||||
import { encryptFactData, encryptFactProvenance } from "@/lib/documents/store";
|
||||
import { repairUndatedDocumentLabValues } from "@/lib/jobs/lab-document-date-repair";
|
||||
import { invalidateUserTimezone } from "@/lib/tz/resolver";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
const USER = "user-lab-document-date-repair";
|
||||
@@ -22,6 +25,7 @@ const noon = (day: string) => new Date(`${day}T12:00:00.000Z`);
|
||||
beforeEach(async () => {
|
||||
const prisma = getPrismaClient();
|
||||
await truncateAllTables(prisma);
|
||||
invalidateUserTimezone(USER);
|
||||
await prisma.user.create({
|
||||
data: {
|
||||
id: USER,
|
||||
@@ -229,4 +233,58 @@ describe("lab document date repair", () => {
|
||||
expect(result.updated).toBe(0);
|
||||
expect((await takenAt())["linked-twice"]).toBe(confirmed.toISOString());
|
||||
});
|
||||
|
||||
it("skips and counts a fact that does not decrypt, and repairs the rest", async () => {
|
||||
const confirmed = new Date("2026-10-20T08:00:00.000Z");
|
||||
for (const id of ["damaged", "misdated"]) {
|
||||
await confirmedValue({
|
||||
id,
|
||||
reportDate: "2021-03-01",
|
||||
documentDate: null,
|
||||
effectiveDate: null,
|
||||
createdAt: confirmed,
|
||||
takenAt: confirmed,
|
||||
});
|
||||
}
|
||||
const prisma = getPrismaClient();
|
||||
await prisma.extractedFact.updateMany({
|
||||
where: { committedRecordId: "damaged" },
|
||||
data: { dataEncrypted: Buffer.from("not a ciphertext at all") },
|
||||
});
|
||||
|
||||
const result = await repairUndatedDocumentLabValues(prisma);
|
||||
|
||||
expect(result).toMatchObject({ updated: 1, undecryptable: 1 });
|
||||
expect(await takenAt()).toEqual({
|
||||
damaged: confirmed.toISOString(),
|
||||
misdated: "2021-03-01T12:00:00.000Z",
|
||||
});
|
||||
});
|
||||
|
||||
it("reads a zone Postgres does not know the way the app does, so the candidates converge", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
await prisma.user.update({
|
||||
where: { id: USER },
|
||||
data: { timezone: "Mars/Olympus_Mons" },
|
||||
});
|
||||
// 00:30 on the report day in the instance default (Europe/Berlin), the
|
||||
// evening before in UTC: on its document's day for the app.
|
||||
const justAfterMidnight = new Date("2026-10-19T22:30:00.000Z");
|
||||
await confirmedValue({
|
||||
id: "taken-on-confirm-day",
|
||||
reportDate: "2026-10-20",
|
||||
documentDate: null,
|
||||
effectiveDate: null,
|
||||
createdAt: justAfterMidnight,
|
||||
takenAt: justAfterMidnight,
|
||||
});
|
||||
|
||||
const result = await repairUndatedDocumentLabValues(prisma);
|
||||
|
||||
// Not even a candidate: the boot check finds nothing to enqueue.
|
||||
expect(result).toMatchObject({ scanned: 0, updated: 0 });
|
||||
expect((await takenAt())["taken-on-confirm-day"]).toBe(
|
||||
justAfterMidnight.toISOString(),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -59,6 +59,7 @@ vi.mock("@/lib/auth/oidc", async (importOriginal) => {
|
||||
});
|
||||
|
||||
const oidc = await import("@/lib/auth/oidc");
|
||||
const { hashToken } = await import("@/lib/auth/hmac");
|
||||
|
||||
const USER_ID = "user-oidc-reproof";
|
||||
const SUB = "idp-sub-reproof";
|
||||
@@ -313,4 +314,32 @@ describe("confirming at the identity provider", () => {
|
||||
expect(back.searchParams.get("reproof")).toBe("failed");
|
||||
expect(oidc.discoverOidcMetadata).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("is listed on the status read for the browser, never for a token", async () => {
|
||||
const { GET } = await import("@/app/api/auth/me/mfa/route");
|
||||
const status = async () => {
|
||||
const res = await (GET as unknown as RouteFn)(
|
||||
new NextRequest("http://localhost:3000/api/auth/me/mfa"),
|
||||
);
|
||||
expect(res.status).toBe(200);
|
||||
return ((await res.json()) as { data: { proofMethods: string[] } }).data
|
||||
.proofMethods;
|
||||
};
|
||||
|
||||
await startSession(HOUR);
|
||||
expect(await status()).toEqual(["oidc"]);
|
||||
|
||||
cookieJar.clear();
|
||||
const raw = `hlk_mfastatus${"0".repeat(64 - "mfastatus".length)}`;
|
||||
await getPrismaClient().apiToken.create({
|
||||
data: {
|
||||
userId: USER_ID,
|
||||
name: "device",
|
||||
tokenHash: hashToken(raw),
|
||||
permissions: ["*"],
|
||||
},
|
||||
});
|
||||
headerJar.set("authorization", `Bearer ${raw}`);
|
||||
expect(await status()).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -11,11 +11,16 @@
|
||||
* hour whose anchor a sample fell on (Google Health's natural-key rescue
|
||||
* used to re-key that row back into the sample).
|
||||
* - Samples inside the window still write as before.
|
||||
* - The change feed served these samples, so their removal reaches it: the
|
||||
* nightly increment leaves tombstones the feed reports, and a backlog
|
||||
* day, deleted outright, expires the account's delta cursors. An Apple
|
||||
* Health fold expires nothing.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { upsertGoogleHealthMeasurements } from "@/lib/google-health/sync-core";
|
||||
import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention";
|
||||
import { pullSyncChanges } from "@/lib/sync/changes/pull";
|
||||
import {
|
||||
TouchedDays,
|
||||
writeOuraHeartRatePage,
|
||||
@@ -182,4 +187,138 @@ describe("dense fold of provider heart rate (#1297)", () => {
|
||||
recent.length,
|
||||
);
|
||||
});
|
||||
|
||||
describe("and the change feed", () => {
|
||||
/** Drain the feed from the start; the last page's cursor. */
|
||||
async function caughtUpCursor(userId: string): Promise<string> {
|
||||
let token: string | undefined;
|
||||
for (;;) {
|
||||
const page = await pullSyncChanges({
|
||||
userId,
|
||||
token,
|
||||
limit: 500,
|
||||
serverNow: new Date(),
|
||||
});
|
||||
expect(page.cursorExpired).toBe(false);
|
||||
token = page.cursor!;
|
||||
if (!page.hasMore) return token;
|
||||
}
|
||||
}
|
||||
|
||||
async function pullFrom(userId: string, token: string) {
|
||||
return pullSyncChanges({
|
||||
userId,
|
||||
token,
|
||||
limit: 500,
|
||||
serverNow: new Date(),
|
||||
});
|
||||
}
|
||||
|
||||
async function syncResetAt(userId: string) {
|
||||
const user = await prisma.user.findUniqueOrThrow({
|
||||
where: { id: userId },
|
||||
select: { syncResetAt: true },
|
||||
});
|
||||
return user.syncResetAt;
|
||||
}
|
||||
|
||||
it("reports the nightly increment as tombstones and keeps the cursor", async () => {
|
||||
const user = await makeUser({ timezone: "UTC" });
|
||||
// Three days past the window: the day the nightly fold reaches next.
|
||||
const day = samples(dayStart(93)).map((s) => ({
|
||||
bpm: s.bpm,
|
||||
timestamp: s.at.toISOString(),
|
||||
source: "awake",
|
||||
}));
|
||||
await writeOuraHeartRatePage(user.id, day, new TouchedDays(), {
|
||||
emitArrivals: false,
|
||||
});
|
||||
const raw = await prisma.measurement.findMany({
|
||||
where: { userId: user.id, source: "OURA" },
|
||||
select: { id: true },
|
||||
});
|
||||
expect(raw).toHaveLength(day.length);
|
||||
const cursor = await caughtUpCursor(user.id);
|
||||
|
||||
const summary = await runDenseIntradayRetention(prisma, {
|
||||
userId: user.id,
|
||||
log: () => {},
|
||||
});
|
||||
expect(summary.totals.syncResetsStamped).toBe(0);
|
||||
expect(await syncResetAt(user.id)).toBeNull();
|
||||
expect(
|
||||
(await liveRows(user.id, "OURA")).every((r) =>
|
||||
r.externalId?.startsWith("stats:"),
|
||||
),
|
||||
).toBe(true);
|
||||
|
||||
const page = await pullFrom(user.id, cursor);
|
||||
expect(page.cursorExpired).toBe(false);
|
||||
const { tombstones } = page.changes.measurements;
|
||||
// The derived resting figure rides along under COMPUTED.
|
||||
const upserts = page.changes.measurements.upserts.filter(
|
||||
(u) => u.source === "OURA",
|
||||
);
|
||||
// The samples on 10:30 and 11:30 became the two hourly means; every
|
||||
// other sample arrives as a tombstone, so the client's mirror ends up
|
||||
// holding exactly the hourly means.
|
||||
expect(upserts).toHaveLength(2);
|
||||
expect(upserts.every((u) => u.externalId?.startsWith("stats:"))).toBe(
|
||||
true,
|
||||
);
|
||||
expect(tombstones).toHaveLength(day.length - 2);
|
||||
expect(
|
||||
[...tombstones.map((t) => t.id), ...upserts.map((u) => u.id)].sort(),
|
||||
).toEqual(raw.map((r) => r.id).sort());
|
||||
});
|
||||
|
||||
it("expires the cursor when a backlog day is deleted outright", async () => {
|
||||
const user = await makeUser({ timezone: "UTC" });
|
||||
await upsertGoogleHealthMeasurements(
|
||||
user.id,
|
||||
samples(dayStart(120)).map(gh),
|
||||
);
|
||||
const cursor = await caughtUpCursor(user.id);
|
||||
|
||||
const summary = await runDenseIntradayRetention(prisma, {
|
||||
userId: user.id,
|
||||
log: () => {},
|
||||
});
|
||||
expect(summary.totals.syncResetsStamped).toBe(1);
|
||||
expect(await syncResetAt(user.id)).not.toBeNull();
|
||||
// No tombstone is left behind for the backlog.
|
||||
expect(
|
||||
await prisma.measurement.count({
|
||||
where: { userId: user.id, deletedAt: { not: null } },
|
||||
}),
|
||||
).toBe(0);
|
||||
expect((await pullFrom(user.id, cursor)).cursorExpired).toBe(true);
|
||||
// A client that re-initialises catches up from a fresh cursor.
|
||||
await caughtUpCursor(user.id);
|
||||
});
|
||||
|
||||
it("expires nothing for an Apple Health fold", async () => {
|
||||
const user = await makeUser({ timezone: "UTC" });
|
||||
await prisma.measurement.createMany({
|
||||
data: samples(dayStart(120)).map((s) => ({
|
||||
userId: user.id,
|
||||
type: "PULSE" as const,
|
||||
value: s.bpm,
|
||||
unit: "bpm",
|
||||
source: "APPLE_HEALTH" as const,
|
||||
measuredAt: s.at,
|
||||
externalId: `uuid-${s.at.getTime()}`,
|
||||
})),
|
||||
});
|
||||
const cursor = await caughtUpCursor(user.id);
|
||||
const summary = await runDenseIntradayRetention(prisma, {
|
||||
userId: user.id,
|
||||
log: () => {},
|
||||
});
|
||||
expect(summary.totals.daysConsolidated).toBe(1);
|
||||
expect(summary.totals.syncResetsStamped).toBe(0);
|
||||
expect(await syncResetAt(user.id)).toBeNull();
|
||||
expect((await pullFrom(user.id, cursor)).cursorExpired).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -344,4 +344,25 @@ describe("isRollupFullBackfillPending", () => {
|
||||
await boss.complete(ROLLUP_FULL_BACKFILL_QUEUE, job.id);
|
||||
expect(await isRollupFullBackfillPending("user-a")).toBe(false);
|
||||
});
|
||||
|
||||
it("counts a queue it cannot read as pending, so the dense fold waits", async () => {
|
||||
// The table is there but the read fails.
|
||||
await prisma.$executeRawUnsafe(
|
||||
`ALTER TABLE pgboss.job RENAME COLUMN data TO data_unreadable`,
|
||||
);
|
||||
try {
|
||||
expect(await isRollupFullBackfillPending("user-a")).toBe(true);
|
||||
} finally {
|
||||
await prisma.$executeRawUnsafe(
|
||||
`ALTER TABLE pgboss.job RENAME COLUMN data_unreadable TO data`,
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("isRollupFullBackfillPending without a queue", () => {
|
||||
it("is false where there is no queue table at all", async () => {
|
||||
await prisma.$executeRawUnsafe(`DROP SCHEMA IF EXISTS pgboss CASCADE`);
|
||||
expect(await isRollupFullBackfillPending("user-a")).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -4,7 +4,10 @@
|
||||
* A background job of the instance that keeps failing pushes every admin
|
||||
* once, a reminder at most once a day while it keeps failing, and once when it
|
||||
* runs successfully again. Fewer failures than the streak, and queues that
|
||||
* have an account-facing surface of their own, push nothing.
|
||||
* have an account-facing surface of their own, push nothing. The streak
|
||||
* counts jobs, not retries, and is kept per account: one account failing
|
||||
* every hour beside accounts that succeed is one problem and one recovery,
|
||||
* and a job that does not run again is announced once, not every day.
|
||||
*
|
||||
* pg-boss provisions its own schema here and the jobs are failed and completed
|
||||
* through its own API, as in `job-failures.test.ts`, so the reader is proven
|
||||
@@ -54,19 +57,32 @@ const HOUR_MS = 3_600_000;
|
||||
|
||||
let boss: PgBoss;
|
||||
|
||||
async function runOnce(queue: string, outcome: "fail" | "complete") {
|
||||
const id = await boss.send(queue, { userId: "u" }, { retryLimit: 0 });
|
||||
async function runOnce(
|
||||
queue: string,
|
||||
outcome: "fail" | "complete",
|
||||
userId = "u",
|
||||
retryLimit = 0,
|
||||
) {
|
||||
const id = await boss.send(
|
||||
queue,
|
||||
{ userId },
|
||||
{ retryLimit, retryDelay: 0, retryBackoff: false },
|
||||
);
|
||||
if (!id) throw new Error("queue rejected the job");
|
||||
const [job] = await boss.fetch(queue);
|
||||
if (!job) throw new Error("nothing to fetch");
|
||||
if (outcome === "fail") {
|
||||
await boss.fail(
|
||||
queue,
|
||||
job.id,
|
||||
new Error("canceling statement due to statement timeout"),
|
||||
);
|
||||
} else {
|
||||
await boss.complete(queue, job.id, { ok: true });
|
||||
// Every attempt, retries included, until pg-boss settles the job.
|
||||
for (let attempt = 0; attempt <= retryLimit; attempt++) {
|
||||
const [job] = await boss.fetch(queue);
|
||||
if (!job) throw new Error("nothing to fetch");
|
||||
if (outcome === "fail") {
|
||||
await boss.fail(
|
||||
queue,
|
||||
job.id,
|
||||
new Error(`canceling statement due to statement timeout (${userId})`),
|
||||
);
|
||||
} else {
|
||||
await boss.complete(queue, job.id, { ok: true });
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -127,7 +143,8 @@ describe("notifySystemJobFailures", () => {
|
||||
expect(again.sent).toEqual([]);
|
||||
expect(dispatched.calls).toEqual([]);
|
||||
|
||||
// A day later: one reminder.
|
||||
// It fails once more; a day after the notice: one reminder.
|
||||
await runOnce(SYSTEM_QUEUE, "fail");
|
||||
const reminder = await notifySystemJobFailures(
|
||||
new Date(now.getTime() + 25 * HOUR_MS),
|
||||
);
|
||||
@@ -175,4 +192,58 @@ describe("notifySystemJobFailures", () => {
|
||||
expect(outcome.failing).toBe(0);
|
||||
expect(outcome.sent).toEqual([]);
|
||||
});
|
||||
|
||||
it("counts a job that used up its retries as one failure", async () => {
|
||||
await makeUser({ role: "ADMIN" });
|
||||
await runOnce(SYSTEM_QUEUE, "fail", "u", 2);
|
||||
const outcome = await notifySystemJobFailures();
|
||||
expect(outcome.failing).toBe(0);
|
||||
expect(outcome.sent).toEqual([]);
|
||||
});
|
||||
|
||||
it("tells once about one account failing every hour beside accounts that succeed", async () => {
|
||||
await makeUser({ role: "ADMIN" });
|
||||
const start = Date.now();
|
||||
const sent: string[] = [];
|
||||
// Ten hours: account `a` fails every hour, the others succeed after it,
|
||||
// so the queue's newest job is always a success.
|
||||
for (let hour = 0; hour < 10; hour++) {
|
||||
await runOnce(SYSTEM_QUEUE, "fail", "a");
|
||||
await runOnce(SYSTEM_QUEUE, "complete", "b");
|
||||
await runOnce(SYSTEM_QUEUE, "complete", "c");
|
||||
const outcome = await notifySystemJobFailures(
|
||||
new Date(start + hour * HOUR_MS),
|
||||
);
|
||||
sent.push(...outcome.sent.map((n) => n.kind));
|
||||
}
|
||||
expect(sent).toEqual(["problem"]);
|
||||
expect(String(dispatched.calls[0].params?.error)).toContain("(a)");
|
||||
|
||||
// `a` succeeds: one recovery.
|
||||
await runOnce(SYSTEM_QUEUE, "complete", "a");
|
||||
const recovered = await notifySystemJobFailures(
|
||||
new Date(start + 10 * HOUR_MS),
|
||||
);
|
||||
expect(recovered.sent).toEqual([
|
||||
{ queue: SYSTEM_QUEUE, kind: "recovered" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not remind about a job that has not run again since the notice", async () => {
|
||||
await makeUser({ role: "ADMIN" });
|
||||
for (let i = 0; i < SYSTEM_JOB_FAILURE_STREAK; i++) {
|
||||
await runOnce(SYSTEM_QUEUE, "fail");
|
||||
}
|
||||
const now = new Date(Date.now() + 1_000);
|
||||
expect((await notifySystemJobFailures(now)).sent).toEqual([
|
||||
{ queue: SYSTEM_QUEUE, kind: "problem" },
|
||||
]);
|
||||
for (const days of [1, 2, 5]) {
|
||||
const later = await notifySystemJobFailures(
|
||||
new Date(now.getTime() + days * 24 * HOUR_MS),
|
||||
);
|
||||
expect(later.failing).toBe(1);
|
||||
expect(later.sent).toEqual([]);
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user