mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-11 13:52:20 +00:00
fix(fold): recompute a stored mean only where the record shows two folds
The fold leftovers were told from a person's deletions by syncVersion = 1, but raw samples are raised in place before a fold takes them (the collision merge, a changed Health Connect value). The repair then recomputed windows without those samples and could overwrite a correct mean, also in windows one run had folded whole. A tombstone of fold age now counts as a leftover when it carries syncVersion = 1 or shares its deletion instant with such a tombstone of the same day; anything else of that age is ambiguous and leaves its window as stored. Only a window taken in two parts (fold runs at least an hour apart, or a run and later live samples) is recomputed. Once the repair has been through an account, the purge may remove class-A leftovers inside the horizon, so the folds' re-fold of a stored day or hour leaves it as stored, and the purge takes the fold lock per account. The fold-lock transactions get an explicit timeout and maxWait, since the lock wait counts against Prisma's five-second default.
This commit is contained in:
1 parent
c40b21de3d
commit
814d3e0179
12 files changed
+836
-103
No files matched your search
@@ -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(
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -143,8 +143,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = 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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<TombstoneRow[]>;
|
||||
/** Remove exactly those ids inside `tx`; returns the rows removed. */
|
||||
deleteIds: (tx: Prisma.TransactionClient, ids: string[]) => Promise<number>;
|
||||
/**
|
||||
* 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<number>;
|
||||
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
|
||||
|
||||
@@ -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: {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<number, FoldConstituent[]>();
|
||||
const leftoversByHour = new Map<number, FoldLeftover[]>();
|
||||
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<number, number>();
|
||||
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,
|
||||
|
||||
@@ -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<FoldTombstoneCandidate, "externalId" | "measuredAt" | "deletedAt">,
|
||||
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<number>();
|
||||
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<FoldLeftover, "deletedAt">[];
|
||||
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<Prisma.TransactionClient, "measurement">,
|
||||
input: {
|
||||
userId: string;
|
||||
@@ -133,10 +269,10 @@ export async function loadFoldLeftovers(
|
||||
to: Date;
|
||||
foldedAfterMs: number;
|
||||
},
|
||||
): Promise<FoldConstituent[]> {
|
||||
): Promise<FoldTombstones> {
|
||||
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<Prisma.TransactionClient, "measurementFoldRepair">,
|
||||
userId: string,
|
||||
): Promise<boolean> {
|
||||
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<Prisma.TransactionClient, "measurement">,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
let acquired!: () => void;
|
||||
const held = new Promise<void>((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);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user