diff --git a/src/lib/export/restore-lock.ts b/src/lib/export/restore-lock.ts index 991ea6f63..fa9919bcf 100644 --- a/src/lib/export/restore-lock.ts +++ b/src/lib/export/restore-lock.ts @@ -79,6 +79,22 @@ export async function holdAccountAgainstRestore( if (!row?.held) throw new AccountRestoreInProgressError(userId); } +/** + * Transaction options for a day transaction that takes + * {@link holdAccountFoldLock}. The wait for the lock runs inside the + * transaction, so it counts against the interactive timeout, and Prisma's + * default of five seconds would abort a day whenever the other pass holds the + * lock a little longer (P2028). The other side holds it for one day's + * transaction, itself bounded by this timeout, and every statement (the lock + * wait included) by the connection's 60-second `statement_timeout`; two + * minutes covers a full wait and the day's own work. `maxWait` bounds the + * wait for a pool connection, as elsewhere in the repo. + */ +export const FOLD_TRANSACTION_OPTIONS = { + maxWait: 10_000, + timeout: 120_000, +} as const; + /** * Serialise the passes that rewrite an account's `stats:` means from their * samples: the daily-mean consolidation, the dense hourly fold and the @@ -89,6 +105,10 @@ export async function holdAccountAgainstRestore( * the same row at once. It waits rather than refusing: the other side holds * it for one day's transaction. * + * The compaction-tombstone purge takes it too, for each account it deletes + * from, so it never removes leftovers between a re-fold's reads + * (`fold-constituents.ts`). + * * The two-key form keeps it apart from the restore lock's single-key space. */ export async function holdAccountFoldLock( diff --git a/src/lib/jobs/compaction-tombstone-purge.ts b/src/lib/jobs/compaction-tombstone-purge.ts index 0b2c87690..cc3700f9b 100644 --- a/src/lib/jobs/compaction-tombstone-purge.ts +++ b/src/lib/jobs/compaction-tombstone-purge.ts @@ -41,12 +41,18 @@ * part of the day (`measurement-fold-repair.ts`). The purge therefore leaves * an account alone until the repair has been through it (a * `MeasurementFoldRepair` row), and the boot sends no run while no account - * has one. The repair queues a run as it finishes an account. + * has one. The repair queues a run as it finishes an account. It deletes + * class-A tombstones whatever their age, so from then on the leftovers inside + * the horizon of `fold-constituents.ts` are no longer all there: the folds' + * re-fold of a stored window reads the marker and leaves such an account's + * means as stored, and each account's delete takes the fold lock so it never + * lands between a re-fold's reads. */ import { createHash } from "node:crypto"; import type { Job } from "pg-boss"; import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import { holdAccountFoldLock } from "@/lib/export/restore-lock"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { jobBudget } from "@/lib/jobs/job-budget"; import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; @@ -243,7 +249,12 @@ export async function purgeCompactionTombstones( } return ready.splice(0, take); }, - deleteIds: async (tx, ids) => { + deleteIds: async (tx, ids, userId) => { + // A re-fold reads an account's live samples and its leftovers under the + // fold lock, so it never sees the leftovers half gone; it reads the + // repair marker under the same lock and stays off them once the repair + // has run (`fold-constituents.ts`). + await holdAccountFoldLock(tx, userId); // Re-checked at delete time: a row a restore wrote back, or one a // person resurrected since the scan, is not a tombstone any more. const result = await tx.measurement.deleteMany({ diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 8f863f1b6..0263d7850 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -143,8 +143,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ // with the mean over every sample, and those it corrected. Counts only. "means_checked", "means_corrected", - // Windows before the horizon it left as stored, and live samples it took - // into a checked mean. + // Windows it left as stored because a tombstone in them may be a person's + // deletion, windows before the horizon it left as stored, and live samples + // it took into a checked mean. + "means_left_ambiguous", "means_samples_absorbed", "means_skipped_beyond_horizon", "mean_rows_soft_deleted", diff --git a/src/lib/jobs/measurement-fold-repair.ts b/src/lib/jobs/measurement-fold-repair.ts index 86b7f625e..1efc75a9c 100644 --- a/src/lib/jobs/measurement-fold-repair.ts +++ b/src/lib/jobs/measurement-fold-repair.ts @@ -26,20 +26,25 @@ * Per account, per folded type and source, day by day in `measuredAt` order, * over the local days that ended before the fold boundary (`foldBoundary`; a * day the fold has not finished is the regular fold's to complete): - * - the scan finds the days with fold leftovers (`fold-constituents.ts`; a - * person's deletions are left out); a day without any has nothing to - * repair and is passed over; + * - the scan finds the days with a tombstone that is certainly a fold's + * (`isCertainFoldLeftover`); a day without one has nothing to repair and + * is passed over; * - inside the day's transaction, under the fold lock, the day's live raw - * samples and its fold leftovers are read again; - * - for each hour (dense) or the day (mean) with a live `stats:` row and at - * least one fold leftover, the mean over every sample is compared with the - * stored value, and only a different value is written. A window without - * leftovers is left alone: its samples in hand are not all of it; + * samples and its tombstones are read again and classified + * (`classifyFoldTombstones`); + * - for each hour (dense) or the day (mean) with a live `stats:` row, the + * mean over every sample is compared with the stored value, and only a + * different value is written, but only for a window the record shows was + * taken in two parts (`foldWindowVerdict`). A window one run took whole + * already holds the mean of all of it; a window with a tombstone that may + * be a person's deletion, or without any leftover, is left as it is, + * because its samples in hand are not provably all of it; * - the live samples of every window it checked are soft-deleted as fold * leftovers (`absorbIntoFold`): they are now part of the stored mean, and * a second run then computes the same mean from the same samples; * - for an Apple Health heart-rate day, the derived resting row is - * recomputed from the whole day the same way, when one exists. + * recomputed from the whole day the same way, when one exists and the day + * holds no ambiguous tombstone. * * Concurrency. The day transaction takes the restore lock and then the fold * lock (`restore-lock.ts`), which the daily-mean consolidation and the dense @@ -60,8 +65,9 @@ * purge run as it finishes. * * Reported per run: one outcome per account (`reportJobRun`) and, per type, - * how many windows were checked, corrected and left beyond the horizon, and - * how many live samples were taken into a checked mean. Counts only, never a + * how many windows were checked, corrected, left beyond the horizon and left + * because a tombstone in them may be a person's deletion, and how many live + * samples were taken into a checked mean. Counts only, never a * value. * * The queue name MUST be registered in `allQueues` in @@ -79,6 +85,7 @@ import type { import { prisma as defaultPrisma } from "@/lib/db"; import { AccountRestoreInProgressError, + FOLD_TRANSACTION_OPTIONS, holdAccountAgainstRestore, holdAccountFoldLock, } from "@/lib/export/restore-lock"; @@ -119,8 +126,9 @@ import { import { absorbIntoFold, foldConstituentHorizon, - isFoldLeftover, - loadFoldLeftovers, + foldWindowVerdict, + isCertainFoldLeftover, + loadFoldTombstones, loadLiveSamples, meanDiffers, meanOf, @@ -187,7 +195,10 @@ export interface MeasurementFoldRepairPayload { /** Per-type counts. Never a value. */ export interface FoldRepairTypeCounts { - /** Hours or days with a live `stats:` row and fold leftovers. */ + /** + * Hours or days with a live `stats:` row that the record shows were taken + * in two parts, and so were recomputed. + */ windowsChecked: number; /** Of those, the ones whose stored mean was corrected. */ windowsCorrected: number; @@ -198,6 +209,11 @@ export interface FoldRepairTypeCounts { * they are (see the header). */ windowsSkippedBeyondHorizon: number; + /** + * Hours or days left as stored because a tombstone in them may be a + * person's deletion (see `fold-constituents.ts`). + */ + windowsLeftAmbiguous: number; /** Live samples taken into a checked mean and kept as fold leftovers. */ samplesAbsorbed: number; } @@ -209,6 +225,7 @@ export function emptyFoldRepairCounts(): FoldRepairTypeCounts { windowsCorrected: 0, restingCorrected: 0, windowsSkippedBeyondHorizon: 0, + windowsLeftAmbiguous: 0, samplesAbsorbed: 0, }; } @@ -341,7 +358,7 @@ export async function repairFoldedMeans( return result; } const leftovers = dayRows.filter((row) => - isFoldLeftover(row, step.foldedAfterMs), + isCertainFoldLeftover(row, step.foldedAfterMs), ); if (leftovers.length === 0) continue; const window = localDayWindow(dateKey, tz); @@ -428,6 +445,7 @@ async function repairDay( let corrected = 0; let resting = 0; let absorbed = 0; + let ambiguousWindows = 0; await prisma.$transaction(async (tx) => { // First, before any reading is touched: see `restore-lock.ts`. await holdAccountAgainstRestore(tx, userId); @@ -441,32 +459,43 @@ async function repairDay( to: window.dayEnd, }; const live = await loadLiveSamples(tx, span); - const leftovers = await loadFoldLeftovers(tx, { + const { leftovers, ambiguous } = await loadFoldTombstones(tx, { ...span, foldedAfterMs: step.foldedAfterMs, }); if (leftovers.length === 0) return; - // Window id → its samples. Only a window with a fold leftover is - // checked: one without has no record of what its stored mean was taken - // from beyond the rows in hand. + // Window id → its samples, and what the record says about how it was + // folded (`foldWindowVerdict`). const byWindow = new Map< string, - { values: number[]; liveIds: string[]; leftovers: number } + { + values: number[]; + liveIds: string[]; + leftovers: { deletedAt: Date }[]; + ambiguous: number; + } >(); - const add = (row: FoldConstituent, isLive: boolean) => { + const entryFor = (row: { measuredAt: Date }) => { const id = windowId(step, hk, dateKey, row, tz); let entry = byWindow.get(id); if (!entry) { - entry = { values: [], liveIds: [], leftovers: 0 }; + entry = { values: [], liveIds: [], leftovers: [], ambiguous: 0 }; byWindow.set(id, entry); } - entry.values.push(row.value); - if (isLive) entry.liveIds.push(row.id); - else entry.leftovers += 1; + return entry; }; - for (const row of live) add(row, true); - for (const row of leftovers) add(row, false); + for (const row of live) { + const entry = entryFor(row); + entry.values.push(row.value); + entry.liveIds.push(row.id); + } + for (const row of leftovers) { + const entry = entryFor(row); + entry.values.push(row.value); + entry.leftovers.push(row); + } + for (const row of ambiguous) entryFor(row).ambiguous += 1; const stored = await tx.measurement.findMany({ where: { @@ -475,7 +504,7 @@ async function repairDay( source: step.source, externalId: { in: [...byWindow.entries()] - .filter(([, entry]) => entry.leftovers > 0) + .filter(([, entry]) => entry.leftovers.length > 0) .map(([id]) => id), }, deletedAt: null, @@ -486,6 +515,13 @@ async function repairDay( for (const row of stored) { const entry = row.externalId ? byWindow.get(row.externalId) : undefined; if (!entry) continue; + const verdict = foldWindowVerdict({ + leftovers: entry.leftovers, + ambiguous: entry.ambiguous, + live: entry.liveIds.length, + }); + if (verdict === "ambiguous") ambiguousWindows += 1; + if (verdict !== "recompute") continue; checked += 1; takeIds.push(...entry.liveIds); if (entry.liveIds.length > 0 && step.grain === "day") { @@ -502,7 +538,9 @@ async function repairDay( } absorbed = await absorbIntoFold(tx, takeIds); - if (restingExternalId) { + // The resting figure is taken from the whole day, so only a day with no + // tombstone that may be a person's deletion has all of it in hand. + if (restingExternalId && ambiguous.length === 0) { const restingRow = await tx.measurement.findFirst({ where: { userId, @@ -515,7 +553,7 @@ async function repairDay( }); const derived = restingRow ? deriveDailyRestingFromPulse( - [...live, ...leftovers].map((row) => ({ + [...live, ...leftovers].map((row: FoldConstituent) => ({ id: row.id, type: step.type, value: row.value, @@ -532,10 +570,11 @@ async function repairDay( resting += 1; } } - }); + }, FOLD_TRANSACTION_OPTIONS); counts.windowsChecked += checked; counts.windowsCorrected += corrected; counts.restingCorrected += resting; + counts.windowsLeftAmbiguous += ambiguousWindows; counts.samplesAbsorbed += absorbed; // A daily mean is the only live row of its day once its samples are taken @@ -583,6 +622,7 @@ export async function handleMeasurementFoldRepair( corrected: 0, resting: 0, skipped: 0, + ambiguous: 0, absorbed: 0, completed: 0, }; @@ -617,11 +657,13 @@ export async function handleMeasurementFoldRepair( sum.windowsCorrected += counts.windowsCorrected; sum.restingCorrected += counts.restingCorrected; sum.windowsSkippedBeyondHorizon += counts.windowsSkippedBeyondHorizon; + sum.windowsLeftAmbiguous += counts.windowsLeftAmbiguous; sum.samplesAbsorbed += counts.samplesAbsorbed; totals.checked += counts.windowsChecked; totals.corrected += counts.windowsCorrected; totals.resting += counts.restingCorrected; totals.skipped += counts.windowsSkippedBeyondHorizon; + totals.ambiguous += counts.windowsLeftAmbiguous; totals.absorbed += counts.samplesAbsorbed; } @@ -683,6 +725,7 @@ export async function handleMeasurementFoldRepair( evt.addMeta("fold_repair_means_corrected", totals.corrected); evt.addMeta("fold_repair_resting_corrected", totals.resting); evt.addMeta("fold_repair_skipped_beyond_horizon", totals.skipped); + evt.addMeta("fold_repair_left_ambiguous", totals.ambiguous); evt.addMeta("fold_repair_samples_absorbed", totals.absorbed); return jobDone({ @@ -690,6 +733,7 @@ export async function handleMeasurementFoldRepair( means_corrected: totals.corrected, resting_corrected: totals.resting, means_skipped_beyond_horizon: totals.skipped, + means_left_ambiguous: totals.ambiguous, means_samples_absorbed: totals.absorbed, accounts_completed: totals.completed, duration_ms: Date.now() - startedAt, diff --git a/src/lib/jobs/measurement-tombstone-cleanup.ts b/src/lib/jobs/measurement-tombstone-cleanup.ts index bd84ad636..e52072316 100644 --- a/src/lib/jobs/measurement-tombstone-cleanup.ts +++ b/src/lib/jobs/measurement-tombstone-cleanup.ts @@ -53,8 +53,16 @@ export interface TombstonePurgeOptions { prisma: PrismaClient; /** Up to `take` expired tombstones, none of them owned by `skipUserIds`. */ findRows: (take: number, skipUserIds: string[]) => Promise; - /** Remove exactly those ids inside `tx`; returns the rows removed. */ - deleteIds: (tx: Prisma.TransactionClient, ids: string[]) => Promise; + /** + * Remove exactly those ids of `userId` inside `tx`; returns the rows + * removed. Runs after the restore lock, so a caller that needs a further + * per-account lock takes it here. + */ + deleteIds: ( + tx: Prisma.TransactionClient, + ids: string[], + userId: string, + ) => Promise; batchSize?: number; maxBatches?: number; /** @@ -124,7 +132,7 @@ export async function purgeTombstonesByAccount({ deleted += await prisma.$transaction( async (tx) => { await holdAccountAgainstRestore(tx, userId); - return deleteIds(tx, ids); + return deleteIds(tx, ids, userId); }, // A full batch of one account is one statement over thousands of // rows; the interactive default of five seconds is too tight for diff --git a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts index 8dfb3c19c..4dc1e74d7 100644 --- a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts +++ b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts @@ -115,6 +115,8 @@ describe("consolidateDailyMean — drain flow (mocked Prisma)", () => { findFirst: txFindFirst, findMany: txFindMany, }, + // The fold repair has not been through the account. + measurementFoldRepair: { findUnique: vi.fn().mockResolvedValue(null) }, }; return { mock: { @@ -308,6 +310,8 @@ describe("consolidateDailyMean — canonical-slot collision (second unique index findFirst: txFindFirst, findMany: txFindMany, }, + // The fold repair has not been through the account. + measurementFoldRepair: { findUnique: vi.fn().mockResolvedValue(null) }, }; return { mock: { diff --git a/src/lib/measurements/consolidate-daily-mean.ts b/src/lib/measurements/consolidate-daily-mean.ts index e7febac45..2dd559713 100644 --- a/src/lib/measurements/consolidate-daily-mean.ts +++ b/src/lib/measurements/consolidate-daily-mean.ts @@ -32,7 +32,8 @@ * that already has a live daily row gets the mean over every sample of * the day, its fold leftovers included (`fold-constituents.ts`), but * only inside the tombstone horizon, where those are provably all of - * them; any other such day keeps its stored mean and its live rows + * them, and only until the one-time fold repair has been through the + * account; any other such day keeps its stored mean and its live rows * (`refoldStoredDay`). * Because mean types are NOT in `CUMULATIVE_HK_TYPES`, the daily * read path averages over the single stats row (count = 1) and @@ -52,6 +53,7 @@ * `step-consolidation` boot-time converging-backfill pattern. */ import { + FOLD_TRANSACTION_OPTIONS, holdAccountAgainstRestore, holdAccountFoldLock, } from "@/lib/export/restore-lock"; @@ -78,7 +80,9 @@ import { localDayWindow } from "@/lib/tz/local-day"; import { absorbIntoFold, foldConstituentHorizon, - loadFoldLeftovers, + foldWindowVerdict, + hasFoldRepairRun, + loadFoldTombstones, loadLiveSamples, meanDiffers, meanOf, @@ -235,11 +239,15 @@ async function nextFreeInstant( * guard existed. * * The stored mean is replaced only by the mean over every sample of the day, - * and only when the samples present are provably all of them: the day starts - * at or after the horizon of `fold-constituents.ts` (inside it no fold - * tombstone has been purged) and the fold left tombstones for it. The live - * samples it takes are then soft-deleted as fold leftovers, so the one-time - * repair and any later pass compute the same mean from the same samples. + * and only when the samples present are provably all of them: the fold + * repair has not been through the account yet (after it, the + * compaction-tombstone purge may have removed leftovers inside the horizon), + * the day starts at or after the horizon of `fold-constituents.ts` (inside it + * the retention cleanup has purged no fold tombstone), the fold left + * tombstones for it, and none of the day's tombstones may be a person's + * deletion. The live samples it takes are then soft-deleted as fold + * leftovers, so the one-time repair and any later pass compute the same mean + * from the same samples. * * Otherwise the stored mean was taken from samples that are no longer all in * the table, and the live rows may be a re-upload of some of them or a @@ -274,11 +282,21 @@ async function refoldStoredDay( from: dayStart, to: dayEnd, }; - // Read under the fold lock: the repair may have taken them meanwhile. + // Read under the fold lock, as the purge takes it too: the repair may have + // taken the samples meanwhile, and finished the account. + if (await hasFoldRepairRun(tx, userId)) return { kind: "left-as-stored" }; const live = await loadLiveSamples(tx, window); if (live.length === 0) return { kind: "written", sourceRowsRemoved: 0 }; - const leftovers = await loadFoldLeftovers(tx, { ...window, foldedAfterMs }); - if (leftovers.length === 0) return { kind: "left-as-stored" }; + const { leftovers, ambiguous } = await loadFoldTombstones(tx, { + ...window, + foldedAfterMs, + }); + const verdict = foldWindowVerdict({ + leftovers, + ambiguous: ambiguous.length, + live: live.length, + }); + if (verdict !== "recompute") return { kind: "left-as-stored" }; const mean = meanOf([...live, ...leftovers].map((row) => row.value)); if (meanDiffers(stored.value, mean)) { await tx.measurement.update({ @@ -505,7 +523,7 @@ export async function consolidateDailyMean( where: { id: { in: sourceRowIds }, deletedAt: null }, }); outcome = { kind: "written", sourceRowsRemoved: del.count }; - }); + }, FOLD_TRANSACTION_OPTIONS); if (outcome.kind === "left-as-stored") return outcome; // T3 — recompute the affected (user, type, day) rollup buckets after diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index e23fb9433..fd7a1babc 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -53,6 +53,7 @@ import { holdAccountAgainstRestore, holdAccountFoldLock, + FOLD_TRANSACTION_OPTIONS, } from "@/lib/export/restore-lock"; import type { MeasurementSource, @@ -80,10 +81,13 @@ import { localDayWindow } from "@/lib/tz/local-day"; import { absorbIntoFold, foldConstituentHorizon, - loadFoldLeftovers, + foldWindowVerdict, + hasFoldRepairRun, + loadFoldTombstones, loadLiveSamples, meanOf, type FoldConstituent, + type FoldLeftover, } from "./fold-constituents"; /** @@ -709,12 +713,14 @@ export async function runDenseIntradayRetention( // time. await holdAccountFoldLock(tx, userId); // An hour that already has a live hourly row is never re-minted - // from the samples in hand. Inside the horizon, with fold - // leftovers for the hour, it gets the mean over every sample of + // from the samples in hand. Inside the horizon, before the fold + // repair has been through the account, with fold leftovers for + // the hour and no tombstone in it that may be a person's deletion + // (`fold-constituents.ts`), it gets the mean over every sample of // the hour (live + leftovers) and the live samples are kept as // leftovers (`absorbIntoFold`), so the repair and any later pass // compute the same mean. Otherwise the stored mean was taken from - // samples that are no longer all in the table, and the hour is + // samples that are not provably all in the table, and the hour is // left exactly as it is, its live samples included: a mean of // part of the hour must not replace it (see `refoldStoredDay` in // `consolidate-daily-mean.ts`). @@ -737,9 +743,16 @@ export async function runDenseIntradayRetention( ); // Read under the fold lock: the repair may have taken some of the // samples meanwhile. - let leftovers: FoldConstituent[] = []; + let leftovers: FoldLeftover[] = []; + let ambiguous: FoldConstituent[] = []; let liveIds = new Set(sourceRowIds); - if (liveHourly.size > 0 && withinHorizon) { + // After the repair the purge may have removed leftovers inside + // the horizon; read under the fold lock, which the purge takes. + const refoldable = + liveHourly.size > 0 && + withinHorizon && + !(await hasFoldRepairRun(tx, userId)); + if (refoldable) { const window = { userId, type, @@ -750,18 +763,25 @@ export async function runDenseIntradayRetention( liveIds = new Set( (await loadLiveSamples(tx, window)).map((row) => row.id), ); - leftovers = await loadFoldLeftovers(tx, { + const tombstones = await loadFoldTombstones(tx, { ...window, foldedAfterMs, }); + leftovers = tombstones.leftovers; + ambiguous = tombstones.ambiguous; } - const leftoversByHour = new Map(); + const leftoversByHour = new Map(); for (const row of leftovers) { const hour = hourOfDayForUserTz(row.measuredAt, tz); const slot = leftoversByHour.get(hour) ?? []; slot.push(row); leftoversByHour.set(hour, slot); } + const ambiguousByHour = new Map(); + for (const row of ambiguous) { + const hour = hourOfDayForUserTz(row.measuredAt, tz); + ambiguousByHour.set(hour, (ambiguousByHour.get(hour) ?? 0) + 1); + } const slots: HourlySlot[] = []; const deleteIds: string[] = []; @@ -770,13 +790,22 @@ export async function runDenseIntradayRetention( const slotId = hourlyStatsExternalId(hkIdentifier, dateKey, hour); let value = meanBucketValue(hourRows); if (liveHourly.has(slotId)) { - const earlier = leftoversByHour.get(hour) ?? []; - if (!withinHorizon || earlier.length === 0) { + if (!refoldable) { hoursLeftAsStored += 1; continue; } + const earlier = leftoversByHour.get(hour) ?? []; const live = hourRows.filter((row) => liveIds.has(row.id)); if (live.length === 0) continue; + const verdict = foldWindowVerdict({ + leftovers: earlier, + ambiguous: ambiguousByHour.get(hour) ?? 0, + live: live.length, + }); + if (verdict !== "recompute") { + hoursLeftAsStored += 1; + continue; + } value = meanOf([ ...live.map((row) => row.value), ...earlier.map((row) => row.value), @@ -852,12 +881,14 @@ export async function runDenseIntradayRetention( // 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 // folded before, the live rows plus the leftovers, but only when - // every hour of it could be recomputed. + // every hour of it could be recomputed and no tombstone of the + // day may be a person's deletion. if (!dayAlreadyFolded && liveHourly.size === 0) { wholeDay = dayRows; } else if ( - withinHorizon && + refoldable && leftovers.length > 0 && + ambiguous.length === 0 && hoursLeftAsStored === 0 ) { wholeDay = [ @@ -865,7 +896,7 @@ export async function runDenseIntradayRetention( ...leftovers, ]; } - }); + }, FOLD_TRANSACTION_OPTIONS); return { removed, retiredDaily, diff --git a/src/lib/measurements/fold-constituents.ts b/src/lib/measurements/fold-constituents.ts index 3af332551..53627c4ba 100644 --- a/src/lib/measurements/fold-constituents.ts +++ b/src/lib/measurements/fold-constituents.ts @@ -9,20 +9,48 @@ * part only. Recomputing such a mean correctly needs every sample of the * window: the ones still live and the ones a fold soft-deleted. * - * Which tombstones a fold left, as opposed to a person's deletions: + * Which tombstones a fold left, as opposed to a person's deletions. Nothing + * on a row records who deleted it: there is no origin column, the folds and + * the person's paths both set `updatedAt` with `deletedAt`, and the audit + * rows of the bulk and by-external-ids deletes carry counts, not ids. What + * the rows do carry: * - every path that deletes on a person's behalf (`DELETE * /api/measurements/:id`, the bulk delete, the by-external-ids delete) - * raises `syncVersion`; the folds never did, and a raw HealthKit or Health - * Connect sample is never updated in place, so a fold's tombstone carries - * `syncVersion = 1`; + * raises `syncVersion`, so a tombstone with `syncVersion = 1` was never + * deleted by a person. The converse does not hold: a raw sample can be + * raised in place before a fold takes it (the collision merge in + * `reconcile-external-measurement.ts`, a changed value in the Health + * Connect import), and the fold then leaves it with `syncVersion > 1`; + * - a fold soft-deleted a day's samples in one statement, so its + * tombstones of one run share one `deletedAt` to the millisecond. A + * person's deletion lands on that instant only by coincidence; * - a fold only took samples already past its threshold, so the sample was - * older than the threshold when it was deleted. + * older than the threshold when it was deleted. A tombstone younger than + * that was never a fold's. + * So {@link classifyFoldTombstones} reads a tombstone of the right age as a + * fold leftover when it carries `syncVersion = 1`, or when it shares its + * deletion instant with such a tombstone of the same day (a raised sample the + * same run took). Anything else of that age is ambiguous: a person's + * deletion, or a raised sample whose run left no `syncVersion = 1` row to + * recognise it by. A window holding an ambiguous tombstone is never + * recomputed, because whichever it is, the samples in hand are not provably + * the samples the stored mean should be taken from. + * + * And a window whose leftovers all come from one run, with no live sample, + * is never recomputed either ({@link foldWindowVerdict}): one run took the + * whole window and its stored mean is that run's mean of all of it. Only a + * window the record shows was taken in two parts, by fold runs at least + * {@link FOLD_RUN_GAP_MS} apart, or by a run and the live samples that + * reached it afterwards, can hold a mean of part of it. + * * A collision retirement (`retired:` id) and a `stats:` row are never samples. * * This is deliberately wider than the class-A rule in `folded-window.ts`: a * day folded in two runs left the tombstones of its first run in a window * that was not yet complete, so they are class B (kept for the 75-day - * retention) and still constituents of the day's mean. + * retention) and still constituents of the day's mean. The age test is the + * same one `isFoldedWindowCandidate` applies at the deletion instant, so every + * class-A tombstone has the age of a fold leftover. * * The horizon. Tombstones are purged 75 days after they were written, so for * an older window the samples present may be a fragment: what a re-upload @@ -31,12 +59,21 @@ * far off a stored mean that was computed from the whole window. Every fold * tombstone of a day was written after `dayStart + foldedAfter` (the fold * only took samples older than its threshold), so when that instant is still - * inside the retention, none of them has been purged yet, and the live rows - * plus the fold leftovers are the whole window. {@link foldConstituentHorizon} - * is the earliest day start for which that holds; a pass recomputes a stored - * mean only for a day starting at or after it, and leaves an older one as it - * is. For the daily means that is the last 75 days; for the dense hourly - * means, which are folded 90 days late, the 75 days before the raw window. + * inside the retention, the retention cleanup has purged none of them. + * {@link foldConstituentHorizon} is the earliest day start for which that + * holds; a pass recomputes a stored mean only for a day starting at or after + * it, and leaves an older one as it is. For the daily means that is the last + * 75 days; for the dense hourly means, which are folded 90 days late, the 75 + * days before the raw window. + * + * The horizon holds only until the compaction-tombstone purge has run for the + * account: the purge deletes class-A tombstones whatever their age, so inside + * the horizon only class-B leftovers may remain. The purge waits for the + * account's `MeasurementFoldRepair` row, and from then on nothing recomputes + * a stored mean from leftovers: the folds' re-fold of a stored window checks + * {@link hasFoldRepairRun} and leaves the window as it is. The purge takes + * the fold lock for each account it deletes from, so a re-fold that read no + * marker under that lock reads its samples before the purge can touch them. * * A pass that takes a window's live samples into a stored mean it recomputed * soft-deletes them ({@link absorbIntoFold}) rather than deleting them, so they @@ -58,6 +95,14 @@ const DAY_MS = 86_400_000; */ export const FOLD_HORIZON_MARGIN_MS = DAY_MS; +/** + * How far apart two deletion instants must lie to count as two fold runs. One + * run takes a day in one transaction, seconds at most; two runs of the + * nightly folds are a day apart, and a boot-time run is still hours from the + * nightly one. + */ +export const FOLD_RUN_GAP_MS = 3_600_000; + /** * The earliest day start whose fold leftovers are certainly all still in the * table, for a fold that takes samples `foldedAfterMs` after they were @@ -78,29 +123,120 @@ export interface FoldConstituent { measuredAt: Date; } -/** The fields a row needs for {@link isFoldLeftover}. */ -export interface FoldLeftoverCandidate { +/** A tombstone of the right age, read as a fold leftover. */ +export interface FoldLeftover extends FoldConstituent { + deletedAt: Date; +} + +/** The fields a row needs for {@link classifyFoldTombstones}. */ +export interface FoldTombstoneCandidate extends FoldConstituent { externalId: string | null; - measuredAt: Date; deletedAt: Date | null; syncVersion: number; } -/** Whether a soft-deleted row is a sample a fold removed. Pure. */ -export function isFoldLeftover( - row: FoldLeftoverCandidate, +/** + * Whether a soft-deleted raw sample was old enough, when it was deleted, for + * a fold to have taken it. Pure. + */ +export function hasFoldAge( + row: Pick, foldedAfterMs: number, ): boolean { if (row.deletedAt === null) return false; - if (row.syncVersion !== 1) return false; if (row.externalId === null) return false; if (row.externalId.startsWith("stats:")) return false; if (row.externalId.startsWith("retired:")) return false; return row.measuredAt.getTime() < row.deletedAt.getTime() - foldedAfterMs; } -/** Prisma filter for the rows {@link isFoldLeftover} then decides on. */ -export function foldLeftoverWhere(input: { +/** + * Whether a tombstone is certainly a fold's: of fold age, and never touched by + * a person's delete (`syncVersion = 1`). A day without one has no fold + * leftover at all. Pure. + */ +export function isCertainFoldLeftover( + row: Pick< + FoldTombstoneCandidate, + "externalId" | "measuredAt" | "deletedAt" | "syncVersion" + >, + foldedAfterMs: number, +): boolean { + return row.syncVersion === 1 && hasFoldAge(row, foldedAfterMs); +} + +/** The tombstones of one span, as {@link classifyFoldTombstones} reads them. */ +export interface FoldTombstones { + leftovers: FoldLeftover[]; + /** Tombstones of fold age that may be a person's deletion. */ + ambiguous: FoldConstituent[]; +} + +/** + * Split the tombstones of one local day (one account, type and source) into + * fold leftovers and ambiguous rows. See the header. Pure. + */ +export function classifyFoldTombstones( + rows: readonly FoldTombstoneCandidate[], + foldedAfterMs: number, +): FoldTombstones { + const runInstants = new Set(); + for (const row of rows) { + if (isCertainFoldLeftover(row, foldedAfterMs) && row.deletedAt) { + runInstants.add(row.deletedAt.getTime()); + } + } + const out: FoldTombstones = { leftovers: [], ambiguous: [] }; + for (const row of rows) { + if (!row.deletedAt || !hasFoldAge(row, foldedAfterMs)) continue; + const constituent = { + id: row.id, + value: row.value, + measuredAt: row.measuredAt, + }; + if (row.syncVersion === 1 || runInstants.has(row.deletedAt.getTime())) { + out.leftovers.push({ ...constituent, deletedAt: row.deletedAt }); + } else { + out.ambiguous.push(constituent); + } + } + return out; +} + +/** What a pass may do with one window's stored mean. */ +export type FoldWindowVerdict = + /** Taken in two parts: recompute it from every sample. */ + | "recompute" + /** One run took all of it: the stored mean is already that mean. */ + | "single-run" + /** A tombstone may be a person's deletion: leave it as stored. */ + | "ambiguous" + /** No fold leftover: nothing records what the mean was taken from. */ + | "no-leftovers"; + +/** + * Whether one window's stored mean may be recomputed from its live samples + * and its leftovers. Pure; see the header. + */ +export function foldWindowVerdict(input: { + leftovers: readonly Pick[]; + ambiguous: number; + live: number; +}): FoldWindowVerdict { + if (input.ambiguous > 0) return "ambiguous"; + if (input.leftovers.length === 0) return "no-leftovers"; + if (input.live > 0) return "recompute"; + const instants = input.leftovers + .map((row) => row.deletedAt.getTime()) + .sort((a, b) => a - b); + for (let i = 1; i < instants.length; i++) { + if (instants[i] - instants[i - 1] >= FOLD_RUN_GAP_MS) return "recompute"; + } + return "single-run"; +} + +/** Prisma filter for the tombstones {@link classifyFoldTombstones} reads. */ +export function foldTombstoneWhere(input: { userId: string; type: MeasurementType; source: MeasurementSource; @@ -110,7 +246,6 @@ export function foldLeftoverWhere(input: { type: input.type, source: input.source, deletedAt: { not: null }, - syncVersion: 1, externalId: { not: null }, NOT: [ { externalId: { startsWith: "stats:" } }, @@ -120,10 +255,11 @@ export function foldLeftoverWhere(input: { } /** - * The fold leftovers of one `[from, to)` span of one account, type and source. - * The span is one local day, so the result is at most a day of samples. + * The fold leftovers and ambiguous tombstones of one `[from, to)` span of one + * account, type and source. The span is one local day, so the result is at + * most a day of samples, and the run instants are read across the whole day. */ -export async function loadFoldLeftovers( +export async function loadFoldTombstones( client: Pick, input: { userId: string; @@ -133,10 +269,10 @@ export async function loadFoldLeftovers( to: Date; foldedAfterMs: number; }, -): Promise { +): Promise { const rows = await client.measurement.findMany({ where: { - ...foldLeftoverWhere(input), + ...foldTombstoneWhere(input), measuredAt: { gte: input.from, lt: input.to }, }, select: { @@ -148,13 +284,23 @@ export async function loadFoldLeftovers( syncVersion: true, }, }); - return rows - .filter((row) => isFoldLeftover(row, input.foldedAfterMs)) - .map((row) => ({ - id: row.id, - value: row.value, - measuredAt: row.measuredAt, - })); + return classifyFoldTombstones(rows, input.foldedAfterMs); +} + +/** + * Whether the one-time fold repair has been through an account. From then on + * the compaction-tombstone purge may have removed leftovers inside the + * horizon, so no pass recomputes a stored mean from them (see the header). + */ +export async function hasFoldRepairRun( + client: Pick, + userId: string, +): Promise { + const row = await client.measurementFoldRepair.findUnique({ + where: { userId }, + select: { userId: true }, + }); + return row !== null; } /** @@ -191,10 +337,9 @@ export async function loadLiveSamples( /** * Soft-delete live samples a pass has just taken into a stored mean it - * recomputed, leaving them as fold leftovers (see the header). Raw HealthKit - * and Health Connect samples are never updated in place, so they keep - * `syncVersion = 1`, which {@link isFoldLeftover} reads. Returns how many rows - * it took. + * recomputed, leaving them as fold leftovers (see the header): all of one + * call share one deletion instant, which is how a raised sample among them is + * still read as a leftover. Returns how many rows it took. */ export async function absorbIntoFold( tx: Pick, diff --git a/tests/integration/fold-complete-windows.test.ts b/tests/integration/fold-complete-windows.test.ts index f092ddf39..24ce9a024 100644 --- a/tests/integration/fold-complete-windows.test.ts +++ b/tests/integration/fold-complete-windows.test.ts @@ -476,8 +476,16 @@ async function seedPartialFolds(userId: string) { tomb("ws2", "WALKING_SPEED", 1, "2026-09-18T10:00:00.000Z", run1), tomb("ws3", "WALKING_SPEED", 2, "2026-09-18T20:00:00.000Z", run2), tomb("ws4", "WALKING_SPEED", 2, "2026-09-18T22:00:00.000Z", run2), - // A person deleted this one: not part of the mean. - tomb("ws5", "WALKING_SPEED", 9, "2026-09-18T23:00:00.000Z", run2, 2), + // A person deleted this one the morning after: not part of the mean. + // Younger than the fold threshold when deleted, so never a fold's. + tomb( + "ws5", + "WALKING_SPEED", + 9, + "2026-09-18T23:00:00.000Z", + new Date("2026-09-19T08:00:00.000Z"), + 2, + ), tomb("ok1", "WALKING_SPEED", 3, "2026-09-17T08:00:00.000Z", run1), tomb("ok2", "WALKING_SPEED", 3, "2026-09-17T20:00:00.000Z", run1), tomb( @@ -529,11 +537,13 @@ describe("fold repair", () => { const first = await repairFoldedMeans(prisma, "repair"); expect(first.status).toBe("completed"); + // The right day was folded in one run and is not even recomputed. expect(first.byType.WALKING_SPEED).toEqual({ - windowsChecked: 2, + windowsChecked: 1, windowsCorrected: 1, restingCorrected: 0, windowsSkippedBeyondHorizon: 0, + windowsLeftAmbiguous: 0, samplesAbsorbed: 0, }); expect(first.byType.HEART_RATE_VARIABILITY).toEqual({ @@ -541,6 +551,7 @@ describe("fold repair", () => { windowsCorrected: 1, restingCorrected: 0, windowsSkippedBeyondHorizon: 0, + windowsLeftAmbiguous: 0, samplesAbsorbed: 0, }); expect( diff --git a/tests/integration/fold-horizon.test.ts b/tests/integration/fold-horizon.test.ts index 32eb92a34..f0f6d72b2 100644 --- a/tests/integration/fold-horizon.test.ts +++ b/tests/integration/fold-horizon.test.ts @@ -234,6 +234,7 @@ describe("fold repair horizon", () => { windowsCorrected: 1, restingCorrected: 0, windowsSkippedBeyondHorizon: 1, + windowsLeftAmbiguous: 0, samplesAbsorbed: 1, }); expect( @@ -271,6 +272,7 @@ describe("fold repair horizon", () => { windowsCorrected: 0, restingCorrected: 0, windowsSkippedBeyondHorizon: 1, + windowsLeftAmbiguous: 0, samplesAbsorbed: 0, }); expect( diff --git a/tests/integration/fold-leftover-provenance.test.ts b/tests/integration/fold-leftover-provenance.test.ts new file mode 100644 index 000000000..34052920f --- /dev/null +++ b/tests/integration/fold-leftover-provenance.test.ts @@ -0,0 +1,437 @@ +/** + * v1.42 — which tombstones a stored mean may be recomputed from, and when a + * pass may recompute at all. Against Postgres. + * + * - A window one fold run took whole is never recomputed, even when one of + * its samples had been raised in place before the fold took it. + * - In a window folded in two runs, a raised sample a run took counts as a + * leftover through the run's shared deletion instant, and a tombstone of + * fold age that may be a person's deletion leaves the window as stored. + * - Once the repair has been through an account, the folds' re-fold of a + * stored day or hour leaves it as stored, and the compaction purge waits + * for the fold lock before it deletes. + * - The repair, the daily-mean consolidation and the dense fold wait for + * the fold lock longer than Prisma's default transaction timeout. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const WS_HK = "HKQuantityTypeIdentifierWalkingSpeed"; +const HRV_HK = "HKQuantityTypeIdentifierHeartRateVariabilitySDNN"; + +/** The instant every test reads the horizon against. */ +const NOW = "2026-10-08T12:00:00.000Z"; +/** Inside the daily horizon. */ +const DAY = "2026-09-18"; +/** The nights an older release folded `DAY` on, in two runs. */ +const RUN1 = "2026-09-20T03:00:00.000Z"; +const RUN2 = "2026-09-21T03:00:00.000Z"; +/** Inside the hourly horizon, before the dense raw window. */ +const HOUR_DAY = "2026-06-20"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date(NOW)); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +async function account(id: string, repaired = false) { + const prisma = getPrismaClient(); + await prisma.user.create({ + data: { id, username: id, email: `${id}@example.test`, timezone: "UTC" }, + }); + if (repaired) { + await prisma.measurementFoldRepair.create({ + data: { userId: id, completedAt: new Date("2026-10-01T00:00:00.000Z") }, + }); + } +} + +async function liveValue(userId: string, externalId: string) { + const row = await getPrismaClient().measurement.findFirst({ + where: { userId, externalId, deletedAt: null }, + select: { value: true }, + }); + return row?.value; +} + +async function liveSampleCount(userId: string) { + return getPrismaClient().measurement.count({ + where: { + userId, + deletedAt: null, + NOT: { externalId: { startsWith: "stats:" } }, + }, + }); +} + +type Row = { + id: string; + value: number; + at: string; + deletedAt?: string; + syncVersion?: number; +}; + +/** A stored daily walking-speed mean for `DAY` and its samples. */ +async function seedDay(userId: string, stored: number, samples: Row[]) { + await getPrismaClient().measurement.createMany({ + data: [ + { + id: `${userId}-mean`, + userId, + type: "WALKING_SPEED", + value: stored, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date(`${DAY}T12:00:00.000Z`), + externalId: `stats:${WS_HK}:${DAY}`, + }, + ...samples.map((row) => ({ + id: `${userId}-${row.id}`, + userId, + type: "WALKING_SPEED" as const, + value: row.value, + unit: "m/s", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(`${DAY}T${row.at}:00.000Z`), + externalId: `uuid-${userId}-${row.id}`, + deletedAt: row.deletedAt ? new Date(row.deletedAt) : null, + syncVersion: row.syncVersion ?? 1, + })), + ], + }); +} + +/** A stored HRV mean for hour 10 of `HOUR_DAY` and its samples. */ +async function seedHour(userId: string, stored: number, samples: Row[]) { + await getPrismaClient().measurement.createMany({ + data: [ + { + id: `${userId}-hour`, + userId, + type: "HEART_RATE_VARIABILITY", + value: stored, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date(`${HOUR_DAY}T10:30:00.000Z`), + externalId: `stats:${HRV_HK}:${HOUR_DAY}T10`, + }, + ...samples.map((row) => ({ + id: `${userId}-${row.id}`, + userId, + type: "HEART_RATE_VARIABILITY" as const, + value: row.value, + unit: "ms", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(`${HOUR_DAY}T${row.at}:00.000Z`), + externalId: `uuid-${userId}-${row.id}`, + deletedAt: row.deletedAt ? new Date(row.deletedAt) : null, + syncVersion: row.syncVersion ?? 1, + })), + ], + }); +} + +/** The day an old release split: stored is the evening's mean, 2. */ +const SPLIT_DAY: Row[] = [ + { id: "t1", value: 1, at: "08:00", deletedAt: RUN1 }, + { id: "t2", value: 1, at: "10:00", deletedAt: RUN1 }, + { id: "t3", value: 2, at: "20:00", deletedAt: RUN2 }, + { id: "t4", value: 2, at: "22:00", deletedAt: RUN2 }, +]; + +/** + * Hold the fold lock of `userIds` in a transaction of its own until the + * returned release is called. + */ +async function holdFoldLocks(userIds: string[]) { + const { holdAccountFoldLock } = await import("@/lib/export/restore-lock"); + let release!: () => void; + const released = new Promise((resolve) => { + release = resolve; + }); + let acquired!: () => void; + const held = new Promise((resolve) => { + acquired = resolve; + }); + const holder = getPrismaClient().$transaction( + async (tx) => { + for (const userId of userIds) await holdAccountFoldLock(tx, userId); + acquired(); + await released; + }, + { timeout: 60_000 }, + ); + await held; + return async () => { + release(); + await holder; + }; +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe("fold leftovers: which tombstones count", () => { + it("leaves a day one run took whole alone, though a raised sample is among its tombstones", async () => { + const prisma = getPrismaClient(); + await account("one-run"); + // One run took all four; the 3 had been raised in place before. Stored is + // their mean, right: (1 + 1 + 2 + 3) / 4. + await seedDay("one-run", 1.75, [ + { id: "a", value: 1, at: "08:00", deletedAt: RUN1 }, + { id: "b", value: 1, at: "10:00", deletedAt: RUN1 }, + { id: "c", value: 2, at: "20:00", deletedAt: RUN1 }, + { id: "d", value: 3, at: "22:00", deletedAt: RUN1, syncVersion: 2 }, + ]); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + + const result = await repairFoldedMeans(prisma, "one-run"); + expect(result.byType.WALKING_SPEED?.windowsChecked ?? 0).toBe(0); + expect(result.byType.WALKING_SPEED?.windowsCorrected ?? 0).toBe(0); + expect(await liveValue("one-run", `stats:${WS_HK}:${DAY}`)).toBe(1.75); + }); + + it("counts a raised sample a run took through the run's deletion instant", async () => { + const prisma = getPrismaClient(); + await account("raised"); + // Run 2 took a raised 5 with the evening: the whole day is + // (1 + 1 + 2 + 2 + 5) / 5 = 2.2. Stored is run 2's mean, 3. + await seedDay("raised", 3, [ + ...SPLIT_DAY, + { id: "t5", value: 5, at: "23:00", deletedAt: RUN2, syncVersion: 3 }, + ]); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + + const result = await repairFoldedMeans(prisma, "raised"); + expect(result.byType.WALKING_SPEED).toMatchObject({ + windowsChecked: 1, + windowsCorrected: 1, + windowsLeftAmbiguous: 0, + }); + expect(await liveValue("raised", `stats:${WS_HK}:${DAY}`)).toBeCloseTo( + 2.2, + 9, + ); + }); + + it("leaves a split day as stored when a tombstone of fold age may be a person's deletion", async () => { + const prisma = getPrismaClient(); + await account("ambiguous"); + // A raised sample deleted between the runs, at an instant no fold run + // used: a person's deletion, or a raised sample a run took alone. Which + // one decides the mean, so the day stays as it is. + await seedDay("ambiguous", 2, [ + ...SPLIT_DAY, + { + id: "p", + value: 9, + at: "23:00", + deletedAt: "2026-09-20T18:00:00.000Z", + syncVersion: 2, + }, + ]); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + + const result = await repairFoldedMeans(prisma, "ambiguous"); + expect(result.byType.WALKING_SPEED).toMatchObject({ + windowsChecked: 0, + windowsCorrected: 0, + windowsLeftAmbiguous: 1, + }); + expect(await liveValue("ambiguous", `stats:${WS_HK}:${DAY}`)).toBe(2); + }); + + it("leaves an hour one run took whole alone, though a raised sample is among its tombstones", async () => { + const prisma = getPrismaClient(); + await account("one-run-hour"); + const run = "2026-09-19T10:30:00.000Z"; + await seedHour("one-run-hour", 55, [ + { id: "a", value: 40, at: "10:05", deletedAt: run }, + { id: "b", value: 40, at: "10:10", deletedAt: run }, + { id: "c", value: 80, at: "10:50", deletedAt: run }, + { id: "d", value: 60, at: "10:55", deletedAt: run, syncVersion: 2 }, + ]); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + + await repairFoldedMeans(prisma, "one-run-hour"); + expect( + await liveValue("one-run-hour", `stats:${HRV_HK}:${HOUR_DAY}T10`), + ).toBe(55); + }); +}); + +describe("after the repair", () => { + it("the daily-mean consolidation leaves a stored day and its late sample as they are", async () => { + const prisma = getPrismaClient(); + await account("refold-day", true); + await seedDay("refold-day", 2, [ + ...SPLIT_DAY, + { id: "late", value: 2, at: "23:30" }, + ]); + const { consolidateDailyMean } = + await import("@/lib/measurements/consolidate-daily-mean"); + + const summary = await consolidateDailyMean(prisma, { + userId: "refold-day", + cutoffHours: 36, + log: () => {}, + }); + expect(summary.totals.daysLeftAsStored).toBe(1); + expect(await liveValue("refold-day", `stats:${WS_HK}:${DAY}`)).toBe(2); + expect(await liveSampleCount("refold-day")).toBe(1); + }); + + it("the dense fold leaves a stored hour and its late sample as they are", async () => { + const prisma = getPrismaClient(); + await account("refold-hour", true); + await seedHour("refold-hour", 80, [ + { id: "a", value: 40, at: "10:05", deletedAt: "2026-09-18T10:30:00Z" }, + { id: "b", value: 40, at: "10:10", deletedAt: "2026-09-18T10:30:00Z" }, + { id: "c", value: 80, at: "10:50", deletedAt: "2026-09-19T10:30:00Z" }, + { id: "d", value: 80, at: "10:55", deletedAt: "2026-09-19T10:30:00Z" }, + { id: "late", value: 100, at: "10:58" }, + ]); + const { runDenseIntradayRetention } = + await import("@/lib/measurements/dense-intraday-retention"); + + const summary = await runDenseIntradayRetention(prisma, { + userId: "refold-hour", + log: () => {}, + }); + expect(summary.totals.hoursLeftAsStored).toBe(1); + expect( + await liveValue("refold-hour", `stats:${HRV_HK}:${HOUR_DAY}T10`), + ).toBe(80); + expect(await liveSampleCount("refold-hour")).toBe(1); + }); + + it("the compaction purge waits for the fold lock before it deletes", async () => { + const prisma = getPrismaClient(); + await account("purge-lock", true); + // A class-A tombstone: its day was complete when the fold deleted it, and + // the day's mean is live. + await prisma.measurement.createMany({ + data: [ + { + userId: "purge-lock", + type: "WALKING_SPEED", + value: 1, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-09-10T12:00:00.000Z"), + externalId: `stats:${WS_HK}:2026-09-10`, + }, + { + id: "purge-lock-a", + userId: "purge-lock", + type: "WALKING_SPEED", + value: 1, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-09-10T08:00:00.000Z"), + externalId: "uuid-purge-lock-a", + deletedAt: new Date("2026-09-13T03:00:00.000Z"), + }, + ], + }); + const { purgeCompactionTombstones } = + await import("@/lib/jobs/compaction-tombstone-purge"); + + const release = await holdFoldLocks(["purge-lock"]); + const purge = purgeCompactionTombstones(prisma, { pauseMs: 0 }); + await sleep(1_000); + const whileHeld = await prisma.measurement.count({ + where: { id: "purge-lock-a" }, + }); + await release(); + const outcome = await purge; + expect(whileHeld).toBe(1); + expect(outcome.deleted).toBe(1); + expect( + await prisma.measurement.count({ where: { id: "purge-lock-a" } }), + ).toBe(0); + }); +}); + +describe("the fold lock wait", () => { + it("outlasts Prisma's default transaction timeout in the repair and both folds", async () => { + const prisma = getPrismaClient(); + await account("wait-repair"); + await seedDay("wait-repair", 2, SPLIT_DAY); + await account("wait-mean"); + await prisma.measurement.createMany({ + data: [1, 2].map((value, i) => ({ + userId: "wait-mean", + type: "WALKING_SPEED" as const, + value, + unit: "m/s", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(`2026-10-01T0${i + 8}:00:00.000Z`), + externalId: `uuid-wait-mean-${i}`, + })), + }); + await account("wait-dense"); + await prisma.measurement.createMany({ + data: [40, 80].map((value, i) => ({ + userId: "wait-dense", + type: "HEART_RATE_VARIABILITY" as const, + value, + unit: "ms", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(`${HOUR_DAY}T10:${i + 10}:00.000Z`), + externalId: `uuid-wait-dense-${i}`, + })), + }); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + const { consolidateDailyMean } = + await import("@/lib/measurements/consolidate-daily-mean"); + const { runDenseIntradayRetention } = + await import("@/lib/measurements/dense-intraday-retention"); + + const release = await holdFoldLocks([ + "wait-repair", + "wait-mean", + "wait-dense", + ]); + const work = Promise.allSettled([ + repairFoldedMeans(prisma, "wait-repair"), + consolidateDailyMean(prisma, { + userId: "wait-mean", + cutoffHours: 36, + log: () => {}, + }), + runDenseIntradayRetention(prisma, { + userId: "wait-dense", + log: () => {}, + }), + ]); + // Past the five seconds an interactive transaction gets by default. + await sleep(6_500); + await release(); + const [repair, mean, dense] = await work; + + expect(repair.status).toBe("fulfilled"); + expect(mean.status).toBe("fulfilled"); + expect(dense.status).toBe("fulfilled"); + expect(await liveValue("wait-repair", `stats:${WS_HK}:${DAY}`)).toBe(1.5); + expect(await liveValue("wait-mean", `stats:${WS_HK}:2026-10-01`)).toBe(1.5); + expect( + await liveValue("wait-dense", `stats:${HRV_HK}:${HOUR_DAY}T10`), + ).toBe(60); + }); +});