diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 31b681f6f..4f04a28da 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/src/__tests__/baselines/source-file-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 84, - "total": 94686, + "total": 94692, "entries": { "app/medications/page-client.tsx": 839, "components/charts/mood-chart.tsx": 1166, @@ -71,7 +71,7 @@ "lib/managed-profiles/handover.ts": 986, "lib/mcp/record-search.ts": 1467, "lib/mcp/rich-reads.ts": 1467, - "lib/measurements/dense-intraday-retention.ts": 1010, + "lib/measurements/dense-intraday-retention.ts": 1016, "lib/measurements/import-apple-health-export.ts": 1722, "lib/medications/scheduling/cadence.ts": 976, "lib/medications/scheduling/recurrence.ts": 1100, diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index e43f8a5c3..05a8b99d1 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 346, - "total": 112215, + "total": 112219, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 286, @@ -319,9 +319,9 @@ "lib/labs/biomarker-merge.ts :: runMerge": 254, "lib/measurements/consolidate-daily-mean.ts :: consolidateDailyMean": 278, "lib/measurements/consolidate-legacy-steps.ts :: consolidateLegacySteps": 263, - "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention": 537, - "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource": 447, - "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay": 373, + "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention": 539, + "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource": 448, + "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay": 374, "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay > foldOnce": 212, "lib/measurements/drain-per-sample-cumulative.ts :: drainPerSampleCumulative": 344, "lib/measurements/drain-per-sample-cumulative.ts :: drainPerSampleCumulative > writeDay": 205, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index c1d8207df..2e5febe0e 100644 --- a/src/__tests__/baselines/version-stamp-comments.json +++ b/src/__tests__/baselines/version-stamp-comments.json @@ -1,7 +1,7 @@ { "$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.", "count": 2896, - "total": 9229, + "total": 9228, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -2458,7 +2458,7 @@ "lib/measurements/cumulative-day-sum.ts": 6, "lib/measurements/daily-series-read.ts": 2, "lib/measurements/dense-intraday-hourly-rebuild.ts": 1, - "lib/measurements/dense-intraday-retention.ts": 4, + "lib/measurements/dense-intraday-retention.ts": 3, "lib/measurements/display-transform.ts": 6, "lib/measurements/drain-per-sample-cumulative.ts": 8, "lib/measurements/import-apple-health-export.ts": 5, diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 66e358b74..3d3d11f24 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -110,6 +110,8 @@ const WRITE_SITES: Record = { "value-preserving hourly fold; refolds the touched spans", "lib/measurements/dense-intraday-retention.ts": "value-preserving retention fold; refolds the touched days", + "lib/measurements/dense-fold-removal.ts": + "removes the samples the retention fold has just taken into its hourly means, inside the fold's transaction; the fold refolds the touched days", "lib/jobs/sleep-timeline-backfill.ts": "one-shot re-stamp of stored sleep rows onto their timeline; refolds the touched days", "lib/jobs/step-consolidation-repair.ts": diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index 6b679fbfe..16891f3a5 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/src/lib/jobs/dense-intraday-retention.ts @@ -123,6 +123,9 @@ export async function runDenseIntradayRetentionForUser( // days for proxy users, preserving the resting signal post-fold. derived_resting_rows_upserted: summary.totals.derivedRestingRowsUpserted, + // Backlog days of a server-pulled source that expired the account's + // delta cursors rather than leave millions of tombstones. + sync_resets_stamped: summary.totals.syncResetsStamped, stopped_early: summary.stoppedEarly, }, }, diff --git a/src/lib/measurements/dense-fold-removal.ts b/src/lib/measurements/dense-fold-removal.ts new file mode 100644 index 000000000..93eae7254 --- /dev/null +++ b/src/lib/measurements/dense-fold-removal.ts @@ -0,0 +1,89 @@ +/** + * How the dense fold (`dense-intraday-retention.ts`) removes the raw samples + * of a day it folded, and how the change feed learns of it. + * + * Apple Health and Health Connect samples are deleted outright (v1.42), like + * the cumulative drain's rows, where they used to be tombstoned for the + * 75-day retention. They come from the person's own health store and no + * client reads them from `/api/sync/changes`, so the tombstone told nobody + * anything; it only held the re-upload of a folded sample off, and the + * `folded_window` ingest guard (`folded-window.ts`) does that now from the + * live hourly row. The tombstones made up most of the table and its indexes. + * + * A server-pulled source (`SERVER_PULLED_FOLD_SOURCES`: Google Health, Oura) + * is different. The feed served its samples, and a client mirroring the feed + * keeps them until the feed says they are gone. So: + * - the nightly increment, a day at most {@link FOLD_FEED_TOMBSTONE_DAYS} + * past the retention boundary, is tombstoned. The feed reports every + * sample, and the 75-day tombstone cleanup purges them later. The + * classifier of compaction tombstones leaves these sources out, so the + * feed, the backlog purge and the restore treat them like any deletion. + * - an older day is backlog: an account's first fold, an old range a + * backfill brought in, nights the fold did not run. Tombstoning that + * would put millions of rows back into the table (#1297), so its samples + * are deleted outright and the account's delta cursors expire instead + * (`stampSyncReset`). A client that pulled them re-initialises rather + * than keep them beside the hourly means. + * The tombstones share one deletion instant and keep their `syncVersion`, so + * the re-fold reads them as fold leftovers (`fold-constituents.ts`), exactly + * like the Apple Health tombstones of the releases before v1.42. + */ +import type { MeasurementSource, Prisma } from "@/generated/prisma/client"; +import { stampSyncReset } from "@/lib/sync/reset"; + +import { SERVER_PULLED_FOLD_SOURCES } from "./dense-intraday-sources"; +import { absorbIntoFold } from "./fold-constituents"; + +/** Days past the retention boundary that still count as the nightly fold. */ +const FOLD_FEED_TOMBSTONE_DAYS = 7; + +const DAY_MS = 86_400_000; + +export type FoldRemoval = "delete" | "tombstone" | "delete-and-reset"; + +/** How a fold removes the samples of one local day. Pure. */ +export function foldRemovalFor( + source: MeasurementSource, + dayStart: Date, + retentionDays: number, + now: Date = new Date(), +): FoldRemoval { + if (!SERVER_PULLED_FOLD_SOURCES.has(source)) return "delete"; + const nightly = + now.getTime() - (retentionDays + FOLD_FEED_TOMBSTONE_DAYS) * DAY_MS; + return dayStart.getTime() >= nightly ? "tombstone" : "delete-and-reset"; +} + +/** + * Remove the folded samples `ids`, never one of `keep` (the hourly rows the + * fold just adopted), inside the fold's transaction. Returns how many went + * and whether the account's delta cursors were expired. The caller stamps + * the reset once more after the commit, so a cursor issued while the + * transaction ran expires too; the stamp in here survives a crash after it. + */ +export async function removeFoldedSamples( + tx: Prisma.TransactionClient, + input: { + userId: string; + removal: FoldRemoval; + ids: readonly string[]; + keep: readonly string[]; + }, +): Promise<{ removed: number; resetFeed: boolean }> { + if (input.removal === "tombstone") { + const removed = await absorbIntoFold( + tx, + input.ids.filter((id) => !input.keep.includes(id)), + ); + return { removed, resetFeed: false }; + } + const del = await tx.measurement.deleteMany({ + where: { + id: { in: [...input.ids], notIn: [...input.keep] }, + deletedAt: null, + }, + }); + const resetFeed = input.removal === "delete-and-reset" && del.count > 0; + if (resetFeed) await stampSyncReset(tx, input.userId); + return { removed: del.count, resetFeed }; +} diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index e75cd7546..4e4234201 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -86,6 +86,8 @@ import { DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE, type DenseIntradayFoldSource, } from "./dense-intraday-sources"; +import { foldRemovalFor, removeFoldedSamples } from "./dense-fold-removal"; +import { stampSyncReset } from "@/lib/sync/reset"; import { absorbIntoFold, foldConstituentHorizon, @@ -433,6 +435,12 @@ export interface DenseIntradayRetentionSummary { * resting signal the raw-PULSE proxy can no longer compute after fold. */ derivedRestingRowsUpserted: number; + /** + * Folded days of a server-pulled source whose samples were deleted + * outright, each of which expired the account's delta cursors + * (`dense-fold-removal.ts`). + */ + syncResetsStamped: number; /** * Hours that already had a live hourly row whose samples could not be * shown to be complete (`fold-constituents.ts`): the stored mean and the @@ -458,7 +466,9 @@ export interface DenseIntradayRetentionOptions extends ConsolidationOptions { * Run the dense intra-day retention drain. Folds per-sample dense-tier * rows older than the retention window into hourly-mean `stats:` rows * (user-local hours, anchored at local HH:30) and deletes the raw rows - * (soft-deleted before v1.42); rows inside the window stay raw. Idempotent — + * (soft-deleted before v1.42; a server-pulled source's nightly increment is + * still tombstoned, see `dense-fold-removal.ts`); rows inside the window + * stay raw. Idempotent — * re-invocation after a successful pass converges to zero work because the * folded rows are gone from the live scan, and the minted hourly * rows are excluded by the `NOT startsWith('stats:')` predicate. @@ -487,6 +497,7 @@ export async function runDenseIntradayRetention( hourlyRowsUpserted: 0, dailyRowsRetired: 0, derivedRestingRowsUpserted: 0, + syncResetsStamped: 0, hoursLeftAsStored: 0, }, }; @@ -647,8 +658,10 @@ export async function runDenseIntradayRetention( hoursLeftAsStored: number; /** Every raw sample of the day, when that is provably all of it. */ wholeDay: FoldConstituent[] | null; + resetFeed: boolean; }> => { let removed = 0; + let resetFeed = false; let retiredDaily = false; let slotsWritten = 0; let hoursLeftAsStored = 0; @@ -806,24 +819,20 @@ export async function runDenseIntradayRetention( // anchor is the row just adopted as that hour's mean, so removing // it would erase the fold. // - // v1.42 — deleted outright, like the cumulative drain's rows, - // where they used to be tombstoned for the 75-day retention. No - // client ever pulled these rows, so the tombstone told nobody - // anything; it only held the re-upload of a folded sample off, - // and the `folded_window` ingest guard (`folded-window.ts`) does - // that now from the live hourly row. The tombstones made up most - // of the table and its indexes. - const del = await tx.measurement.deleteMany({ - where: { - id: { in: deleteIds, notIn: canonicalRowIds }, - deletedAt: null, - }, + // How, and whether the change feed hears of it, depends on the + // source and the day's age: `dense-fold-removal.ts`. + const dropped = await removeFoldedSamples(tx, { + userId, + removal: foldRemovalFor(source, dayStart, retentionDays), + ids: deleteIds, + keep: canonicalRowIds, }); + resetFeed = dropped.resetFeed; const absorbed = await absorbIntoFold( tx, absorbIds.filter((id) => !canonicalRowIds.includes(id)), ); - removed = del.count + absorbed; + removed = dropped.removed + absorbed; // The day's raw samples, for the derived resting figure: the rows // in hand for a day folded here for the first time; for a day @@ -844,12 +853,14 @@ export async function runDenseIntradayRetention( ]; } }, FOLD_TRANSACTION_OPTIONS); + if (resetFeed) await stampSyncReset(pc, userId); return { removed, retiredDaily, slotsWritten, hoursLeftAsStored, wholeDay, + resetFeed, }; }; @@ -868,6 +879,7 @@ export async function runDenseIntradayRetention( summary.totals.hourlyRowsUpserted += foldResult.slotsWritten; summary.totals.hoursLeftAsStored += foldResult.hoursLeftAsStored; if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1; + if (foldResult.resetFeed) summary.totals.syncResetsStamped += 1; // iOS#34 — preserve the resting signal for PROXY users. For a user // with zero native `RESTING_HEART_RATE` rows, the read-path resolver @@ -997,7 +1009,7 @@ export async function runDenseIntradayRetention( } log( - `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted} hoursLeftAsStored=${summary.totals.hoursLeftAsStored}${options.dryRun ? " (dry-run)" : ""}`, + `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted} syncResetsStamped=${summary.totals.syncResetsStamped} hoursLeftAsStored=${summary.totals.hoursLeftAsStored}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; diff --git a/src/lib/measurements/dense-intraday-sources.ts b/src/lib/measurements/dense-intraday-sources.ts index 3c1891e54..08c01a69c 100644 --- a/src/lib/measurements/dense-intraday-sources.ts +++ b/src/lib/measurements/dense-intraday-sources.ts @@ -84,3 +84,18 @@ export type DenseIntradayFoldSource = export const DENSE_INTRADAY_FOLD_SOURCES = Object.keys( DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE, ) as readonly DenseIntradayFoldSource[]; + +/** + * The fold sources whose rows the server pulled from a provider itself. A + * client can hold one of these rows only because `/api/sync/changes` served + * it, so when the fold removes them the feed has to say so: the nightly fold + * leaves tombstones for them, and a fold of an older backlog expires the + * account's delta cursors instead (`dense-intraday-retention.ts`). + * + * Apple Health and Health Connect samples come from the person's own health + * store, uploaded by the device or imported from its export, and the fold has + * deleted them outright since v1.42. No client reads the feed for them: the + * iOS app reads its own HealthKit store and has no feed consumer. + */ +export const SERVER_PULLED_FOLD_SOURCES: ReadonlySet = + new Set(["GOOGLE_HEALTH", "OURA"]); diff --git a/src/lib/measurements/folded-window.ts b/src/lib/measurements/folded-window.ts index eb84dfb2c..9990632c2 100644 --- a/src/lib/measurements/folded-window.ts +++ b/src/lib/measurements/folded-window.ts @@ -79,6 +79,7 @@ import { import { DENSE_INTRADAY_FOLD_SOURCES, DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE, + SERVER_PULLED_FOLD_SOURCES, DENSE_INTRADAY_RETENTION_TYPES, } from "@/lib/measurements/dense-intraday-sources"; @@ -475,12 +476,17 @@ export interface TombstoneCandidateRow extends FoldedWindowCandidateRow { * enough: an older release folded a day in two runs, and a person who * deleted a sample of such a half-folded day deleted something the fold had * not taken. That deletion is class B. + * + * A server-pulled source (`SERVER_PULLED_FOLD_SOURCES`) has no class A. The + * fold tombstones its samples so the change feed reports them to a client + * that pulled them, which is exactly what this class is kept out of. */ export function compactionTombstoneCoverId( row: TombstoneCandidateRow, tz: string, ): string | null { if (row.deletedAt === null) return null; + if (SERVER_PULLED_FOLD_SOURCES.has(row.source)) return null; if (!isFoldedWindowCandidate(row, row.deletedAt)) return null; if (!isWindowFoldedAt(row, row.deletedAt, tz)) return null; return coveringStatsExternalId(row, tz); diff --git a/tests/integration/provider-dense-fold.test.ts b/tests/integration/provider-dense-fold.test.ts index d36c21a52..d25d0a9f6 100644 --- a/tests/integration/provider-dense-fold.test.ts +++ b/tests/integration/provider-dense-fold.test.ts @@ -11,11 +11,16 @@ * hour whose anchor a sample fell on (Google Health's natural-key rescue * used to re-key that row back into the sample). * - Samples inside the window still write as before. + * - The change feed served these samples, so their removal reaches it: the + * nightly increment leaves tombstones the feed reports, and a backlog + * day, deleted outright, expires the account's delta cursors. An Apple + * Health fold expires nothing. */ import { beforeEach, describe, expect, it, vi } from "vitest"; import { upsertGoogleHealthMeasurements } from "@/lib/google-health/sync-core"; import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; +import { pullSyncChanges } from "@/lib/sync/changes/pull"; import { TouchedDays, writeOuraHeartRatePage, @@ -182,4 +187,138 @@ describe("dense fold of provider heart rate (#1297)", () => { recent.length, ); }); + + describe("and the change feed", () => { + /** Drain the feed from the start; the last page's cursor. */ + async function caughtUpCursor(userId: string): Promise { + let token: string | undefined; + for (;;) { + const page = await pullSyncChanges({ + userId, + token, + limit: 500, + serverNow: new Date(), + }); + expect(page.cursorExpired).toBe(false); + token = page.cursor!; + if (!page.hasMore) return token; + } + } + + async function pullFrom(userId: string, token: string) { + return pullSyncChanges({ + userId, + token, + limit: 500, + serverNow: new Date(), + }); + } + + async function syncResetAt(userId: string) { + const user = await prisma.user.findUniqueOrThrow({ + where: { id: userId }, + select: { syncResetAt: true }, + }); + return user.syncResetAt; + } + + it("reports the nightly increment as tombstones and keeps the cursor", async () => { + const user = await makeUser({ timezone: "UTC" }); + // Three days past the window: the day the nightly fold reaches next. + const day = samples(dayStart(93)).map((s) => ({ + bpm: s.bpm, + timestamp: s.at.toISOString(), + source: "awake", + })); + await writeOuraHeartRatePage(user.id, day, new TouchedDays(), { + emitArrivals: false, + }); + const raw = await prisma.measurement.findMany({ + where: { userId: user.id, source: "OURA" }, + select: { id: true }, + }); + expect(raw).toHaveLength(day.length); + const cursor = await caughtUpCursor(user.id); + + const summary = await runDenseIntradayRetention(prisma, { + userId: user.id, + log: () => {}, + }); + expect(summary.totals.syncResetsStamped).toBe(0); + expect(await syncResetAt(user.id)).toBeNull(); + expect( + (await liveRows(user.id, "OURA")).every((r) => + r.externalId?.startsWith("stats:"), + ), + ).toBe(true); + + const page = await pullFrom(user.id, cursor); + expect(page.cursorExpired).toBe(false); + const { tombstones } = page.changes.measurements; + // The derived resting figure rides along under COMPUTED. + const upserts = page.changes.measurements.upserts.filter( + (u) => u.source === "OURA", + ); + // The samples on 10:30 and 11:30 became the two hourly means; every + // other sample arrives as a tombstone, so the client's mirror ends up + // holding exactly the hourly means. + expect(upserts).toHaveLength(2); + expect(upserts.every((u) => u.externalId?.startsWith("stats:"))).toBe( + true, + ); + expect(tombstones).toHaveLength(day.length - 2); + expect( + [...tombstones.map((t) => t.id), ...upserts.map((u) => u.id)].sort(), + ).toEqual(raw.map((r) => r.id).sort()); + }); + + it("expires the cursor when a backlog day is deleted outright", async () => { + const user = await makeUser({ timezone: "UTC" }); + await upsertGoogleHealthMeasurements( + user.id, + samples(dayStart(120)).map(gh), + ); + const cursor = await caughtUpCursor(user.id); + + const summary = await runDenseIntradayRetention(prisma, { + userId: user.id, + log: () => {}, + }); + expect(summary.totals.syncResetsStamped).toBe(1); + expect(await syncResetAt(user.id)).not.toBeNull(); + // No tombstone is left behind for the backlog. + expect( + await prisma.measurement.count({ + where: { userId: user.id, deletedAt: { not: null } }, + }), + ).toBe(0); + expect((await pullFrom(user.id, cursor)).cursorExpired).toBe(true); + // A client that re-initialises catches up from a fresh cursor. + await caughtUpCursor(user.id); + }); + + it("expires nothing for an Apple Health fold", async () => { + const user = await makeUser({ timezone: "UTC" }); + await prisma.measurement.createMany({ + data: samples(dayStart(120)).map((s) => ({ + userId: user.id, + type: "PULSE" as const, + value: s.bpm, + unit: "bpm", + source: "APPLE_HEALTH" as const, + measuredAt: s.at, + externalId: `uuid-${s.at.getTime()}`, + })), + }); + const cursor = await caughtUpCursor(user.id); + const summary = await runDenseIntradayRetention(prisma, { + userId: user.id, + log: () => {}, + }); + expect(summary.totals.daysConsolidated).toBe(1); + expect(summary.totals.syncResetsStamped).toBe(0); + expect(await syncResetAt(user.id)).toBeNull(); + expect((await pullFrom(user.id, cursor)).cursorExpired).toBe(false); + }); + }); });