From f5d6710670fcbeb99583df25ca9c92ffd4a53861 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 22 Oct 2026 07:24:59 +0200 Subject: [PATCH] perf(analytics): keep closed blocks of past readings cached across syncs Every sync marked all of an account's analytics cells stale, so the slice re-read the pre-fold remainder (every reading older than five years, folded per source and day, the slice's largest statement on a long history) and the day view re-read its past history blocks, although a sync of today's readings cannot change either. A cache key can now name the instant its value's readings end before (closedBlockKeyPart). invalidateUserMeasurements takes touchedFrom, the earliest measuredAt a write created, changed or removed, and leaves such blocks in place when it is not earlier; without it every block goes as before. The batch route passes it from the identities it wrote. The pre-fold remainder moves into its own module and is cached as a closed block; the day view's past history blocks carry their end, with a day of margin for a night across midnight. The batch route's two up-front reads move into batch-stored-rows.ts with the once-per-batch key rule they share. --- .../baselines/source-file-lines.json | 10 +- .../baselines/source-function-lines.json | 6 +- .../baselines/version-stamp-comments.json | 4 +- .../__tests__/duplicate-prefilter.test.ts | 17 +- src/app/api/measurements/batch/route.ts | 238 +++++------------- src/lib/analytics/pre-fold-remainder.ts | 124 +++++++++ src/lib/analytics/summaries-slice.ts | 74 +----- .../closed-block-invalidation.test.ts | 49 ++++ src/lib/cache/invalidate.ts | 42 +++- src/lib/cache/server-cache.ts | 22 +- src/lib/day/notable.ts | 9 +- src/lib/measurements/batch-stored-rows.ts | 177 +++++++++++++ 12 files changed, 493 insertions(+), 279 deletions(-) create mode 100644 src/lib/analytics/pre-fold-remainder.ts create mode 100644 src/lib/cache/__tests__/closed-block-invalidation.test.ts create mode 100644 src/lib/measurements/batch-stored-rows.ts diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 43b2d9c5c..eda0f96b5 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/src/__tests__/baselines/source-file-lines.json @@ -1,12 +1,12 @@ { "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 129, - "total": 156466, + "total": 156380, "entries": { "app/api/dashboard/summary/route.ts": 1032, "app/api/documents/inbound/route.ts": 811, "app/api/insights/generate/route.ts": 1209, - "app/api/measurements/batch/route.ts": 1576, + "app/api/measurements/batch/route.ts": 1569, "app/api/measurements/route.ts": 938, "app/api/measurements/series/route.ts": 879, "app/api/medications/[id]/route.ts": 1281, @@ -52,11 +52,11 @@ "lib/ai/coach/types.ts": 1011, "lib/ai/codex-client.ts": 1094, "lib/ai/prompts/insight-generator.ts": 1121, - "lib/ai/provider.ts": 1644, + "lib/ai/provider.ts": 1632, "lib/ai/safety/outbound-screen.ts": 1463, "lib/analytics/score/reader.ts": 998, "lib/analytics/sleep-night.ts": 1345, - "lib/analytics/summaries-slice.ts": 1314, + "lib/analytics/summaries-slice.ts": 1248, "lib/api-handler.ts": 1978, "lib/auth/session.ts": 811, "lib/cache/server-cache.ts": 851, @@ -110,7 +110,7 @@ "lib/jobs/reminder/register-status.ts": 878, "lib/managed-profiles/handover.ts": 986, "lib/mcp/record-search.ts": 1467, - "lib/mcp/rich-reads.ts": 1468, + "lib/mcp/rich-reads.ts": 1467, "lib/mcp/tools.ts": 2654, "lib/measurements/dense-intraday-retention.ts": 1057, "lib/measurements/import-apple-health-export.ts": 1729, diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index 02291aeae..092ca4d38 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": 415, - "total": 154643, + "total": 154564, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -38,7 +38,7 @@ "app/api/mcp/oauth/token/route.ts :: POST > withBackgroundEvent(arg)": 248, "app/api/measurement-reminders/[id]/route.ts :: apiHandler(arg) #2": 240, "app/api/measurements/[id]/route.ts :: apiHandler(arg) #2": 203, - "app/api/measurements/batch/route.ts :: postBatch": 1277, + "app/api/measurements/batch/route.ts :: postBatch": 1262, "app/api/measurements/route.ts :: apiHandler(arg)": 286, "app/api/measurements/route.ts :: postMeasurement": 377, "app/api/measurements/series/route.ts :: apiHandler(arg)": 670, @@ -299,7 +299,7 @@ "lib/analytics/correlations-fast-path.ts :: computeCorrelationHypothesesFastPath": 256, "lib/analytics/intraday-pulse-io.ts :: loadIntradayPulse": 206, "lib/analytics/score/reader.ts :: computeUserHealthScore": 324, - "lib/analytics/summaries-slice.ts :: computeFromRollups": 441, + "lib/analytics/summaries-slice.ts :: computeFromRollups": 377, "lib/api-handler.ts :: apiHandler": 207, "lib/api-handler.ts :: apiHandler > wrapped": 201, "lib/auth/profile-update.ts :: applyProfileUpdate": 353, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index d54528c42..4ba5626b7 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": 2962, - "total": 10592, + "total": 10591, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -1885,7 +1885,7 @@ "lib/analytics/sleep-night.ts": 3, "lib/analytics/source-priority.ts": 2, "lib/analytics/source-rank-sql.ts": 3, - "lib/analytics/summaries-slice.ts": 37, + "lib/analytics/summaries-slice.ts": 36, "lib/analytics/trend-delta.ts": 1, "lib/analytics/trends.ts": 6, "lib/analytics/value-bands.ts": 1, diff --git a/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts b/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts index 7ce59b9f5..af4a4a395 100644 --- a/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts +++ b/src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts @@ -576,15 +576,21 @@ describe("a batch of duplicates only (v1.42)", () => { }); describe("cache invalidation by origin (v1.42)", () => { + // Every call also says how far back the batch reached (`touchedFrom`), so + // closed blocks of older readings stay cached. it.each([ ["foreground", { evict: true }], ["manual", { evict: true }], - ["background", undefined], - ["push", undefined], - [undefined, undefined], + ["background", {}], + ["push", {}], + [undefined, {}], ])("syncTrigger %s → %o", async (trigger, expected) => { - await post([weight(`uuid-${trigger}`, new Date(NOW - DAY))], trigger); - expect(invalidateUserMeasurements).toHaveBeenCalledWith(USER, expected); + const at = new Date(NOW - DAY); + await post([weight(`uuid-${trigger}`, at)], trigger); + expect(invalidateUserMeasurements).toHaveBeenCalledWith(USER, { + ...expected, + touchedFrom: at, + }); }); it("evicts for a background batch that wrote a hand-entered reading", async () => { @@ -599,6 +605,7 @@ describe("cache invalidation by origin (v1.42)", () => { ); expect(invalidateUserMeasurements).toHaveBeenCalledWith(USER, { evict: true, + touchedFrom: new Date(NOW - DAY), }); }); }); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index c6e430dee..8f8c06549 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -66,6 +66,14 @@ import { oppositeMergeSource, type MergeCandidate, } from "@/lib/measurements/cross-source-merge"; +import { + externalKeyOf, + onceInBatch, + readStoredSampleIdentities, + readStoredStatsRows, + sampleIsStoredDuplicate, + statsWriteUnchanged, +} from "@/lib/measurements/batch-stored-rows"; import { lockMeasurementUser, reconcileExternalMeasurement, @@ -775,181 +783,52 @@ async function postBatch(request: NextRequest): Promise { // before this one is reached, and then the reconciler has to decide. // `stats:` rows are never prefiltered (overwrite semantics), and neither // is a value the reconciler's own plausibility gate would refuse. - const externalKeyCounts = new Map(); - const naturalKeyCounts = new Map(); - const naturalKeyOf = ( - type: unknown, - source: unknown, - measuredAt: Date, - sleepStage: unknown, - ) => `${type}::${source}::${measuredAt.getTime()}::${sleepStage ?? "null"}`; - for (const p of prepared) { - const externalKey = `${p.row.type}::${p.row.source}::${p.row.externalId}`; - externalKeyCounts.set( - externalKey, - (externalKeyCounts.get(externalKey) ?? 0) + 1, - ); - const naturalKey = naturalKeyOf( - p.row.type, - p.row.source, - p.row.measuredAt as Date, - p.row.sleepStage, - ); - naturalKeyCounts.set( - naturalKey, - (naturalKeyCounts.get(naturalKey) ?? 0) + 1, - ); - } - const sampleExternalIds = [ - ...new Set( - prepared - .map((p) => p.row.externalId as string) - .filter((externalId) => !isStatsExternalId(externalId)), - ), - ]; - const existingByExternalKey = new Map< - string, - { measuredAt: Date; sleepStage: unknown } - >(); - if (sampleExternalIds.length > 0) { - const existing = await prisma.measurement.findMany({ - where: { userId: user.id, externalId: { in: sampleExternalIds } }, - select: { - type: true, - source: true, - externalId: true, - measuredAt: true, - sleepStage: true, - }, - }); - for (const row of existing) { - existingByExternalKey.set( - `${row.type}::${row.source}::${row.externalId}`, - { measuredAt: row.measuredAt, sleepStage: row.sleepStage }, - ); - } - } - function isExactStoredDuplicate(p: Prepared): boolean { - const externalKey = `${p.row.type}::${p.row.source}::${p.row.externalId}`; - const stored = existingByExternalKey.get(externalKey); - if (!stored) return false; - const measuredAt = p.row.measuredAt as Date; - const naturalKey = naturalKeyOf( - p.row.type, - p.row.source, - measuredAt, - p.row.sleepStage, - ); - return ( - stored.measuredAt.getTime() === measuredAt.getTime() && - (stored.sleepStage ?? null) === (p.row.sleepStage ?? null) && - externalKeyCounts.get(externalKey) === 1 && - naturalKeyCounts.get(naturalKey) === 1 && - classifyRangeRejection( - p.row.type as MeasurementType, - p.row.value as number, - ) === null - ); - } + const onceInThisBatch = onceInBatch(prepared.map((p) => p.row)); + const existingByExternalKey = await readStoredSampleIdentities( + prisma, + user.id, + [ + ...new Set( + prepared + .map((p) => p.row.externalId as string) + .filter((externalId) => !isStatsExternalId(externalId)), + ), + ], + ); + const isExactStoredDuplicate = (p: Prepared): boolean => + sampleIsStoredDuplicate( + existingByExternalKey.get( + externalKeyOf(p.row.type, p.row.source, p.row.externalId), + ), + p.row, + ) && + onceInThisBatch(p.row) && + classifyRangeRejection( + p.row.type as MeasurementType, + p.row.value as number, + ) === null; - // An unchanged `stats:` re-post. The phone re-sends the same daily - // totals on every sync, and each used to go through the reconciler - // (about ten round trips in its own transaction) only to write the same - // values back, and then counted as `updated`, which started the whole - // tail: the DAY refold, rollup jobs, PR detection and a cache stale for - // the account. Such a row is still reported `updated`, exactly as - // before, but nothing is written and it starts nothing. - // - // "Unchanged" is the reconciler's own update with nothing to change: a - // live row holds the entry's external key AND its natural identity - // (both indexes are unique, so that row is the only candidate), already - // carries the HealthKit-statistics provenance, and every column the - // update would write already has the value it would write. Like the - // sample prefilter above, it applies only where neither key occurs a - // second time in the batch, and never to a value the plausibility gate - // would refuse. A concurrent writer is no different from one that ran a - // moment later: its write lands after this no-op either way. - const statsExternalIds = [ + // An unchanged `stats:` re-post is reported `updated` without a write + // (see `batch-stored-rows.ts`), under the same once-per-batch key rule. + const storedStats = await readStoredStatsRows(prisma, user.id, [ ...new Set( prepared .map((p) => p.row.externalId as string) .filter((externalId) => isStatsExternalId(externalId)), ), - ]; - const storedStatsByExternalKey = new Map< - string, - Record & { measuredAt: Date } - >(); - if (statsExternalIds.length > 0) { - const stored = await prisma.measurement.findMany({ - where: { - userId: user.id, - externalId: { in: statsExternalIds }, - deletedAt: null, - }, - select: { - type: true, - source: true, - externalId: true, - measuredAt: true, - sleepStage: true, - value: true, - valueMin: true, - valueMax: true, - unit: true, - externalSourceVersion: true, - glucoseContext: true, - rhythmClassification: true, - deviceType: true, - aggregationProvenance: true, - }, - }); - for (const row of stored) { - storedStatsByExternalKey.set( - `${row.type}::${row.source}::${row.externalId}`, - row, - ); - } - } - function isUnchangedStatsRow(p: Prepared): boolean { - const externalKey = `${p.row.type}::${p.row.source}::${p.row.externalId}`; - const stored = storedStatsByExternalKey.get(externalKey); - if (!stored) return false; - const measuredAt = p.row.measuredAt as Date; - const naturalKey = naturalKeyOf( - p.row.type, - p.row.source, - measuredAt, - p.row.sleepStage, + ]); + const isUnchangedStatsRow = (p: Prepared): boolean => + onceInThisBatch(p.row) && + classifyRangeRejection( + p.row.type as MeasurementType, + p.row.value as number, + ) === null && + statsWriteUnchanged( + storedStats.get( + externalKeyOf(p.row.type, p.row.source, p.row.externalId), + ), + desiredFor(p), ); - if ( - !(stored.measuredAt instanceof Date) || - stored.measuredAt.getTime() !== measuredAt.getTime() || - externalKeyCounts.get(externalKey) !== 1 || - naturalKeyCounts.get(naturalKey) !== 1 || - stored.aggregationProvenance !== "HEALTHKIT_STATISTICS" || - classifyRangeRejection( - p.row.type as MeasurementType, - p.row.value as number, - ) !== null - ) { - return false; - } - const desired = desiredFor(p); - const same = (column: keyof typeof desired) => - (stored[column] ?? null) === (desired[column] ?? null); - return ( - same("value") && - same("valueMin") && - same("valueMax") && - same("unit") && - same("sleepStage") && - same("externalSourceVersion") && - same("glucoseContext") && - same("rhythmClassification") && - same("deviceType") - ); - } // v1.42 — the `folded_window` guard (`folded-window.ts`). A raw Apple // Health sample of a dense or mean type, old enough to have been folded, @@ -1205,16 +1084,7 @@ async function postBatch(request: NextRequest): Promise { continue; } - const naturalKey = naturalKeyOf( - p.row.type, - p.row.source, - p.row.measuredAt as Date, - p.row.sleepStage, - ); - if ( - externalKeyCounts.get(key) === 1 && - naturalKeyCounts.get(naturalKey) === 1 - ) { + if (onceInThisBatch(p.row)) { pendingInserts.push(p); pushMergeCandidate(p); continue; @@ -1610,7 +1480,17 @@ async function postBatch(request: NextRequest): Promise { syncTrigger === "foreground" || syncTrigger === "manual" || wroteManualRow; - invalidateUserMeasurements(user.id, evict ? { evict: true } : undefined); + // Every row this batch created, changed or retired is in + // `writtenIdentities`, so its earliest `measuredAt` bounds what changed: + // closed blocks of older readings (the pre-fold remainder, the day + // view's past history blocks) stay cached across the sync. + const touchedFrom = new Date( + Math.min(...writtenIdentities.map((w) => w.measuredAt.getTime())), + ); + invalidateUserMeasurements(user.id, { + ...(evict ? { evict: true } : {}), + ...(Number.isFinite(touchedFrom.getTime()) ? { touchedFrom } : {}), + }); // v1.37.19 — shared post-mutation tail for every distinct // (type, day) the batch touched. Updates participate as well as diff --git a/src/lib/analytics/pre-fold-remainder.ts b/src/lib/analytics/pre-fold-remainder.ts new file mode 100644 index 000000000..ec48d8d3e --- /dev/null +++ b/src/lib/analytics/pre-fold-remainder.ts @@ -0,0 +1,124 @@ +/** + * The pre-fold remainder of the dashboard slice: per type, the all-time + * figures of every reading older than the rollup fold window. + * + * The slice splices it into its per-type figures, so "all-time" means all time + * on the rollup path too (empty for any account younger than the + * window). It keeps one source per (type, day) the way the rollup half + * and the live path do: without it, a vital recorded by two devices + * counted both of them on every pre-fold day. Each source is folded per + * day first (a hash aggregate, a few rows per day however dense the + * stream), each day then keeps its ladder-canonical source, so the + * sort never sees more than one row per source per day. + * An hourly-mean type folds each source's day as the mean of its + * hours' means (`wsum` / `wdays`, see `day-mean.ts`) and the days into + * the remainder once each, the statistic the DAY rollup stores. + * + * It is a closed block: it aggregates only readings measured before + * `foldSplit`, the start of a UTC day five years back, so a sync of newer + * readings cannot change it. On a long history it was the slice's largest + * statement (every reading older than five years, folded per source and day), + * and the slice is rebuilt after every sync. It is cached under a closed-block + * key (`closedBlockKeyPart`): a measurement write that reaches back before + * `foldSplit`, and every write that does not say how far back it reached, + * drops it; the split moves on at the next UTC day, which names a new key. + * The TTL bounds an entry that a write in another process (a split worker) + * cannot reach. + */ +import { createHash } from "node:crypto"; + +import { Prisma } from "@/generated/prisma/client"; +import { closedBlockKeyPart } from "@/lib/cache/invalidate"; +import { caches, type ServerCache } from "@/lib/cache/server-cache"; +import { prisma } from "@/lib/db"; +import { dayWeightedRowsSql, foldMeanSql } from "@/lib/measurements/day-mean"; + +export type PreFoldRemainderRow = { + type: string; + count: number; + days: number; + min: number; + max: number; + mean: number; +}; + +/** A closed block's lifetime when no write reaches it. */ +const PRE_FOLD_REMAINDER_TTL_MS = 15 * 60_000; + +/** The remainder as a live read, uncached. */ +export function readPreFoldRemainderLive( + userId: string, + foldSplit: Date, + rankPerSource: string, +): Promise { + return prisma.$queryRaw` + WITH src AS ( + SELECT * + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."deleted_at" IS NULL + AND m."measured_at" < ${foldSplit} + ), + per_source AS ( + SELECT + m."type", + m."source", + date_trunc('day', m."measured_at") AS day, + COUNT(*)::int AS cnt, + SUM(m."value")::double precision AS total, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value, + SUM(m."value" * m.day_weight)::double precision AS wsum, + SUM(m.day_weight)::double precision AS wdays + FROM ${dayWeightedRowsSql("src", null, { bySource: true })} m + GROUP BY m."type", m."source", 3 + ), + canon AS ( + SELECT DISTINCT ON (p."type", p.day) + p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.wdays + FROM per_source p + ORDER BY p."type", p.day, (${Prisma.raw(rankPerSource)}), p."source"::text + ) + SELECT + c."type"::text AS type, + SUM(c.cnt)::int AS count, + COUNT(*)::int AS days, + MIN(c.min_value)::double precision AS min, + MAX(c.max_value)::double precision AS max, + ${Prisma.raw( + foldMeanSql({ + typeColumn: 'c."type"', + total: "c.total", + count: "c.cnt", + weightedSum: "c.wsum", + weightSum: "c.wdays", + }), + )}::double precision AS mean + FROM canon c + GROUP BY c."type" +`; +} + +/** + * The remainder through its closed-block cache cell. `rankPerSource` is the + * ladder CASE over `p."type"` / `p."source"`, built from `priorityJson`, + * which names the cell. + */ +export async function readPreFoldRemainder( + userId: string, + foldSplit: Date, + priorityJson: unknown, + rankPerSource: string, +): Promise { + const ladder = createHash("sha256") + .update(JSON.stringify(priorityJson ?? null)) + .digest("base64url") + .slice(0, 16); + const cache = caches.analytics as ServerCache; + const { value } = await cache.wrap( + `${userId}|pre-fold-remainder|${closedBlockKeyPart(foldSplit)}|${ladder}`, + () => readPreFoldRemainderLive(userId, foldSplit, rankPerSource), + PRE_FOLD_REMAINDER_TTL_MS, + ); + return value; +} diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 26b0b15ad..10ea07b03 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -57,7 +57,6 @@ import pLimit from "p-limit"; import type { MeasurementType } from "@/generated/prisma/client"; -import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; import { @@ -69,8 +68,6 @@ import { SESSION_DAY_FRAME, dayValue, dayWeightedRows, - dayWeightedRowsSql, - foldMeanSql, isHourlyMeanTypeSql, windowMeanSql, } from "@/lib/measurements/day-mean"; @@ -78,6 +75,7 @@ import { wireSummary, type DataSummary } from "@/lib/analytics/trends"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { annotate } from "@/lib/logging/context"; import { caches, type ServerCache } from "@/lib/cache/server-cache"; +import { readPreFoldRemainder } from "@/lib/analytics/pre-fold-remainder"; import { ensureUserRollupsFresh, readRollupBuckets, @@ -704,73 +702,9 @@ async function computeFromRollups(userId: string): Promise { sumYy: true, }, }), - // v1.37.19 — all-time remainder OLDER than the fold window, - // spliced into the per-type figures below so "all-time" means all time - // on the rollup path too (empty for any account younger than the - // window). It keeps one source per (type, day) the way the rollup half - // and the live path do: without it, a vital recorded by two devices - // counted both of them on every pre-fold day. Each source is folded per - // day first (a hash aggregate, a few rows per day however dense the - // stream), each day then keeps its ladder-canonical source, so the - // sort never sees more than one row per source per day. - // An hourly-mean type folds each source's day as the mean of its - // hours' means (`wsum` / `wdays`, see `day-mean.ts`) and the days into - // the remainder once each, the statistic the DAY rollup stores. - prisma.$queryRaw< - Array<{ - type: string; - count: number; - days: number; - min: number; - max: number; - mean: number; - }> - >` - WITH src AS ( - SELECT * - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."deleted_at" IS NULL - AND m."measured_at" < ${foldSplit} - ), - per_source AS ( - SELECT - m."type", - m."source", - date_trunc('day', m."measured_at") AS day, - COUNT(*)::int AS cnt, - SUM(m."value")::double precision AS total, - MIN(m."value")::double precision AS min_value, - MAX(m."value")::double precision AS max_value, - SUM(m."value" * m.day_weight)::double precision AS wsum, - SUM(m.day_weight)::double precision AS wdays - FROM ${dayWeightedRowsSql("src", null, { bySource: true })} m - GROUP BY m."type", m."source", 3 - ), - canon AS ( - SELECT DISTINCT ON (p."type", p.day) - p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.wdays - FROM per_source p - ORDER BY p."type", p.day, (${Prisma.raw(rankPerSource)}), p."source"::text - ) - SELECT - c."type"::text AS type, - SUM(c.cnt)::int AS count, - COUNT(*)::int AS days, - MIN(c.min_value)::double precision AS min, - MAX(c.max_value)::double precision AS max, - ${Prisma.raw( - foldMeanSql({ - typeColumn: 'c."type"', - total: "c.total", - count: "c.cnt", - weightedSum: "c.wsum", - weightSum: "c.wdays", - }), - )}::double precision AS mean - FROM canon c - GROUP BY c."type" - `, + // The all-time remainder older than the fold window (a closed block, + // see `pre-fold-remainder.ts`). + readPreFoldRemainder(userId, foldSplit, priorityJson, rankPerSource), ]); const preFoldByType = new Map< diff --git a/src/lib/cache/__tests__/closed-block-invalidation.test.ts b/src/lib/cache/__tests__/closed-block-invalidation.test.ts new file mode 100644 index 000000000..4aec873b1 --- /dev/null +++ b/src/lib/cache/__tests__/closed-block-invalidation.test.ts @@ -0,0 +1,49 @@ +import { beforeEach, describe, expect, it } from "vitest"; + +import { + closedBlockKeyPart, + closedBlockUntouched, + invalidateUserMeasurements, +} from "../invalidate"; +import { caches, type ServerCache } from "../server-cache"; + +const analytics = caches.analytics as ServerCache<{ v: number }>; + +const USER = "u-closed"; +const SPLIT = Date.UTC(2021, 9, 22); +const closedKey = `${USER}|pre-fold-remainder|${closedBlockKeyPart(SPLIT)}|x`; +const openKey = `${USER}|dashboard-snapshot|en`; + +describe("closed analytics blocks", () => { + beforeEach(() => { + analytics.deleteByPrefix(`${USER}|`); + analytics.set(closedKey, { v: 1 }); + analytics.set(openKey, { v: 2 }); + }); + + it("reads the instant a closed block ends before", () => { + expect(closedBlockUntouched(closedKey, new Date(SPLIT))).toBe(true); + expect(closedBlockUntouched(closedKey, new Date(SPLIT - 1))).toBe(false); + expect(closedBlockUntouched(openKey, new Date(SPLIT))).toBe(false); + }); + + it("survives a write of newer readings, evicting or marking stale", () => { + for (const evict of [false, true]) { + invalidateUserMeasurements(USER, { + evict, + touchedFrom: new Date(SPLIT + 86_400_000), + }); + expect(analytics.get(closedKey)).toEqual({ v: 1 }); + expect(analytics.get(openKey)).toBeNull(); + analytics.set(openKey, { v: 2 }); + } + }); + + it("goes with a write that reaches before it, or does not say", () => { + invalidateUserMeasurements(USER, { touchedFrom: new Date(SPLIT - 1) }); + expect(analytics.get(closedKey)).toBeNull(); + analytics.set(closedKey, { v: 1 }); + invalidateUserMeasurements(USER); + expect(analytics.get(closedKey)).toBeNull(); + }); +}); diff --git a/src/lib/cache/invalidate.ts b/src/lib/cache/invalidate.ts index 9ecf5ad9d..826026763 100644 --- a/src/lib/cache/invalidate.ts +++ b/src/lib/cache/invalidate.ts @@ -48,6 +48,30 @@ export function invalidateUserDashboardSnapshot(userId: string): void { caches.analytics.deleteByPrefix(dashboardSnapshotCacheKey(userId)); } +/** + * The key part that marks an analytics entry as a closed block: a value that + * depends only on readings measured before `before`. Such a block is the + * same however many syncs land after it, as long as none of them touches a + * reading that old, so a measurement write that says how far back it + * reached (`touchedFrom`) leaves it in place. Every other invalidation, and + * every write that does not say, drops it as before. + */ +export function closedBlockKeyPart(before: Date | number): string { + const ms = typeof before === "number" ? before : before.getTime(); + return `before:${ms}`; +} + +const CLOSED_BLOCK_RE = /\|before:(\d+)(?:\||$)/; + +/** + * Whether `key` is a closed block that a write reaching back to + * `touchedFrom` cannot have changed. Exported for its test. + */ +export function closedBlockUntouched(key: string, touchedFrom: Date): boolean { + const match = CLOSED_BLOCK_RE.exec(key); + return match !== null && Number(match[1]) <= touchedFrom.getTime(); +} + /** * Invalidate every cache that may reflect a user's measurement set. * @@ -74,16 +98,28 @@ export function invalidateUserDashboardSnapshot(userId: string): void { */ export function invalidateUserMeasurements( userId: string, - opts?: { evict?: boolean }, + opts?: { + evict?: boolean; + /** + * The earliest `measuredAt` the write created, changed or removed. With + * it, closed blocks of older readings (`closedBlockKeyPart`) stay. + * Leave it out unless the caller knows every row it touched. + */ + touchedFrom?: Date; + }, ): void { // A coverage probe already running read the set before this write; the // next reader must not join it. forgetCoverageProbe(userId); + const touchedFrom = opts?.touchedFrom; + const spare = touchedFrom + ? (key: string) => closedBlockUntouched(key, touchedFrom) + : undefined; // The `${userId}|` prefix covers the slim / thick analytics cells, the // iOS summary cell, AND the v1.7.0 dashboard snapshot // (`${userId}|dashboard-snapshot`) in one pass. if (opts?.evict) { - caches.analytics.deleteByPrefix(`${userId}|`); + caches.analytics.deleteByPrefix(`${userId}|`, spare); } else { // v1.12.7 — mark stale rather than hard-evict. Measurement writes are // the highest-frequency dirty signal (every iOS Apple-Health sync @@ -94,7 +130,7 @@ export function invalidateUserMeasurements( // summary cells read via plain `cached` are unaffected: a marked-stale // entry has `expiresAt === now`, so their next read is a clean miss and // rebuilds fresh — identical to the old evict for those keys. - caches.analytics.markStaleByPrefix(`${userId}|`); + caches.analytics.markStaleByPrefix(`${userId}|`, spare); } caches.achievements.deleteByPrefix(userId); caches.workouts.deleteByPrefix(`${userId}|`); diff --git a/src/lib/cache/server-cache.ts b/src/lib/cache/server-cache.ts index 22bafff84..48a2dd842 100644 --- a/src/lib/cache/server-cache.ts +++ b/src/lib/cache/server-cache.ts @@ -88,15 +88,15 @@ export interface ServerCacheStats { * fence — invalidation flips `invalidated` and detaches the slot from * `pending`, so (a) the builder's `.then` skips the `set()` and (b) the * next read starts a fresh build instead of joining the stale one. - * Callers already awaiting the detached promise still receive its value - * (equivalent to having read a moment before the write); it just never - * enters the cache. + * Callers already awaiting the detached promise still get its value (as + * if read just before the write); it never enters the cache. */ interface PendingBuild { promise: Promise; invalidated: boolean; } +type KeyFilter = (key: string) => boolean; // closed blocks, `invalidate.ts` export class ServerCache { private readonly map = new Map< string, @@ -124,9 +124,9 @@ export class ServerCache { } /** Fence + detach every in-flight builder under `prefix`. */ - private detachPendingByPrefix(prefix: string): void { + private detachPendingByPrefix(prefix: string, spare?: KeyFilter): void { for (const [key, slot] of this.pending) { - if (key.startsWith(prefix)) { + if (key.startsWith(prefix) && !spare?.(key)) { slot.invalidated = true; this.pending.delete(key); } @@ -283,17 +283,17 @@ export class ServerCache { * so the next read is a clean miss — identical to the old evict * behaviour). */ - markStaleByPrefix(prefix: string): number { + markStaleByPrefix(prefix: string, spare?: KeyFilter): number { // v1.16.9 — fence in-flight builders too: a refresh that started // before the write would otherwise complete after this mark and // re-cache the pre-write body as FRESH for the full TTL. Detaching // keeps the SWR contract — the marked entry still serves stale, and // the next read kicks off a fresh (post-write) recompute. - this.detachPendingByPrefix(prefix); + this.detachPendingByPrefix(prefix, spare); let marked = 0; const now = Date.now(); for (const [key, entry] of this.map) { - if (key.startsWith(prefix)) { + if (key.startsWith(prefix) && !spare?.(key)) { entry.expiresAt = now; marked += 1; } @@ -320,12 +320,12 @@ export class ServerCache { * entire user-bucket from a cache without having to enumerate every * possible cache-key variant the user might have warmed. */ - deleteByPrefix(prefix: string): number { + deleteByPrefix(prefix: string, spare?: KeyFilter): number { // v1.16.9 — fence in-flight builders under the prefix (see `delete`). - this.detachPendingByPrefix(prefix); + this.detachPendingByPrefix(prefix, spare); let removed = 0; for (const key of this.map.keys()) { - if (key.startsWith(prefix)) { + if (key.startsWith(prefix) && !spare?.(key)) { const entry = this.map.get(key); this.map.delete(key); this.totalWeight -= entry?.weight ?? 0; diff --git a/src/lib/day/notable.ts b/src/lib/day/notable.ts index 2fb50b3af..5b3704fd4 100644 --- a/src/lib/day/notable.ts +++ b/src/lib/day/notable.ts @@ -31,6 +31,7 @@ import type { MeasurementType } from "@/generated/prisma/enums"; import type { DayNotable } from "@/lib/day/contract"; import { readLocalDailyStats, type DailyStats } from "@/lib/day/daily-stats"; +import { closedBlockKeyPart } from "@/lib/cache/invalidate"; import { caches, type ServerCache } from "@/lib/cache/server-cache"; import { prisma } from "@/lib/db"; import { startOfLocalDayKey } from "@/lib/tz/local-day"; @@ -267,10 +268,16 @@ async function readExtremeHistory(args: { ) { const last = shiftDateKey(start, HISTORY_BLOCK_DAYS - 1); const open = last >= today; + // A closed block names the instant its readings end before, with a day + // of margin for a night that runs past midnight, so a sync that only + // touched newer readings leaves it cached (`closedBlockKeyPart`). + const closed = closedBlockKeyPart( + startOfLocalDayKey(shiftDateKey(last, 2), args.tz), + ); blocks.push({ start, end: open ? today : last, - key: `${prefix}|${start}|${open ? today : "closed"}`, + key: `${prefix}|${start}|${open ? today : closed}`, }); } diff --git a/src/lib/measurements/batch-stored-rows.ts b/src/lib/measurements/batch-stored-rows.ts new file mode 100644 index 000000000..9ab4702fe --- /dev/null +++ b/src/lib/measurements/batch-stored-rows.ts @@ -0,0 +1,177 @@ +/** + * The two reads `POST /api/measurements/batch` makes up front about rows the + * batch may already hold, one statement each for the whole batch. + * + * Samples: a re-sent sample used to cost one interactive reconciler + * transaction (about ten round trips) only for the reconciler to answer + * `duplicate`. One read over the batch's sample ids answers it for every row + * at once (the route's exact-duplicate prefilter). + * + * `stats:` rows: the phone re-sends the same daily totals on every sync, and + * each went through the reconciler only to write the same values back, then + * counted as `updated`, which started the whole tail: the DAY refold, rollup + * jobs, PR detection and a cache stale for the account. Such a row is still + * reported `updated`, exactly as before, but nothing is written and it starts + * nothing. "Unchanged" is the reconciler's own update with nothing to change: + * a live row holds the entry's external key AND its natural identity (both + * indexes are unique, so that row is the only candidate), already carries the + * HealthKit-statistics provenance, and every column the update would write + * already has the value it would write. The route applies it only where + * neither key occurs a second time in the batch, and never to a value the + * plausibility gate would refuse. A concurrent writer is no different from + * one that ran a moment later: its write lands after this no-op either way. + */ +import type { PrismaClient } from "@/generated/prisma/client"; + +/** `type::source::externalId`, the external identity of a row. */ +export function externalKeyOf( + type: unknown, + source: unknown, + externalId: unknown, +): string { + return `${type}::${source}::${externalId}`; +} + +/** Stored rows (live or tombstoned) at the batch's sample ids. */ +export async function readStoredSampleIdentities( + db: Pick, + userId: string, + externalIds: readonly string[], +): Promise> { + const out = new Map(); + if (externalIds.length === 0) return out; + const rows = await db.measurement.findMany({ + where: { userId, externalId: { in: [...externalIds] } }, + select: { + type: true, + source: true, + externalId: true, + measuredAt: true, + sleepStage: true, + }, + }); + for (const row of rows) { + out.set(externalKeyOf(row.type, row.source, row.externalId), { + measuredAt: row.measuredAt, + sleepStage: row.sleepStage, + }); + } + return out; +} + +/** The columns the reconciler's update writes, besides the identity. */ +const STATS_COLUMNS = [ + "value", + "valueMin", + "valueMax", + "unit", + "sleepStage", + "externalSourceVersion", + "glucoseContext", + "rhythmClassification", + "deviceType", +] as const; + +export type StoredStatsRow = Record< + (typeof STATS_COLUMNS)[number] | "aggregationProvenance", + unknown +> & { measuredAt: Date }; + +/** Live stored rows at the batch's `stats:` ids, with every written column. */ +export async function readStoredStatsRows( + db: Pick, + userId: string, + externalIds: readonly string[], +): Promise> { + const out = new Map(); + if (externalIds.length === 0) return out; + const rows = await db.measurement.findMany({ + where: { userId, externalId: { in: [...externalIds] }, deletedAt: null }, + select: { + type: true, + source: true, + externalId: true, + measuredAt: true, + aggregationProvenance: true, + ...Object.fromEntries(STATS_COLUMNS.map((c) => [c, true])), + }, + }); + for (const row of rows as unknown as Array< + StoredStatsRow & { type: string; source: string; externalId: string } + >) { + out.set(externalKeyOf(row.type, row.source, row.externalId), row); + } + return out; +} + +/** + * Whether the update a `stats:` entry would make to `stored` writes nothing + * new: same instant, the statistics provenance already, every column equal. + */ +export function statsWriteUnchanged( + stored: StoredStatsRow | undefined, + desired: Record<(typeof STATS_COLUMNS)[number], unknown> & { + measuredAt: Date; + }, +): boolean { + if (!stored || !(stored.measuredAt instanceof Date)) return false; + if (stored.measuredAt.getTime() !== desired.measuredAt.getTime()) { + return false; + } + if (stored.aggregationProvenance !== "HEALTHKIT_STATISTICS") return false; + return STATS_COLUMNS.every( + (column) => (stored[column] ?? null) === (desired[column] ?? null), + ); +} + +/** The identity columns of one batch entry. */ +export interface BatchKeyRow { + type?: unknown; + source?: unknown; + externalId?: unknown; + measuredAt?: unknown; + sleepStage?: unknown; +} + +/** `type::source::epochMs::sleepStage`, the natural identity of a row. */ +export function naturalKeyOf(row: BatchKeyRow): string { + return `${row.type}::${row.source}::${(row.measuredAt as Date).getTime()}::${row.sleepStage ?? "null"}`; +} + +/** + * Whether an entry's external key and natural identity each occur once in + * its batch. Only such an entry may be decided ahead of the reconciler: an + * earlier entry sharing either key could move or retire the stored row + * before this one is reached. + */ +export function onceInBatch( + rows: readonly BatchKeyRow[], +): (row: BatchKeyRow) => boolean { + const external = new Map(); + const natural = new Map(); + for (const row of rows) { + const e = externalKeyOf(row.type, row.source, row.externalId); + external.set(e, (external.get(e) ?? 0) + 1); + const n = naturalKeyOf(row); + natural.set(n, (natural.get(n) ?? 0) + 1); + } + return (row) => + external.get(externalKeyOf(row.type, row.source, row.externalId)) === 1 && + natural.get(naturalKeyOf(row)) === 1; +} + +/** + * Whether a sample entry is an exact duplicate of `stored`, the row at its + * external key: the same instant and sleep stage, which makes that row its + * natural match too, the reconciler's `duplicate`. + */ +export function sampleIsStoredDuplicate( + stored: { measuredAt: Date; sleepStage: unknown } | undefined, + row: BatchKeyRow, +): boolean { + return ( + stored !== undefined && + stored.measuredAt.getTime() === (row.measuredAt as Date).getTime() && + (stored.sleepStage ?? null) === (row.sleepStage ?? null) + ); +}