diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index d8a7a2747..d8792c49d 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -567,6 +567,124 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { }); }); +/** + * A recompute REPLACES every bucket in the window it was asked for. + * + * Scoping the delete to the [min, max] range of the rows the aggregate + * RETURNED made the repair self-limiting in exactly the case that needs it: a + * bucket whose source rows all disappeared is not in the returned range, so it + * survived with its stale aggregate while `probeRollupCoverage` still reported + * `hasBuckets = true` and the read-swap served it instead of falling back to + * live SQL. The write hook masked this for DAY (it deletes its own bucket by + * hand when the day empties) but nothing covered the WEEK / MONTH / YEAR + * worker, which recomputes one bucket span at a time. + */ +describe("recomputeUserRollups — prunes the whole recomputed window", () => { + function rowsInBucket(bucketStart: Date, type = "WEIGHT") { + return [ + { + type, + source: "MANUAL", + bucket_start: bucketStart, + count: BigInt(1), + mean: 80, + min_value: 80, + max_value: 80, + sum_value: 80, + sd: 0, + slope: 0, + r2: 0, + }, + ]; + } + + it("deletes the stale bucket when the window's last reading is gone", async () => { + // The exact damage: the user deleted their last WEIGHT reading of the + // week, so the WEEK aggregate for that span now returns nothing. + queryRawUnsafe.mockResolvedValueOnce([]); + + await recomputeUserRollups("user-1", { + granularities: ["WEEK"], + types: ["WEIGHT"], + from: new Date(Date.UTC(2024, 2, 4)), + to: new Date(Date.UTC(2024, 2, 11)), + }); + + expect(deleteMany).toHaveBeenCalledTimes(1); + const where = deleteMany.mock.calls[0][0].where; + expect(where).toMatchObject({ + userId: "user-1", + granularity: "WEEK", + type: { in: ["WEIGHT"] }, + }); + expect(where.bucketStart.gte.toISOString()).toBe( + "2024-03-04T00:00:00.000Z", + ); + expect(where.bucketStart.lt.toISOString()).toBe("2024-03-11T00:00:00.000Z"); + // Nothing to write — the window is authoritatively empty. + expect(txExecuteRawUnsafe).not.toHaveBeenCalled(); + }); + + it("covers the requested window, not just the range the aggregate returned", async () => { + // Readings survive only in the middle of the window; the buckets at both + // edges lost theirs. The delete must still reach the edges. + queryRawUnsafe.mockResolvedValueOnce( + rowsInBucket(new Date(Date.UTC(2024, 2, 10))), + ); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from: new Date(Date.UTC(2024, 2, 1)), + to: new Date(Date.UTC(2024, 2, 20)), + }); + + const where = deleteMany.mock.calls[0][0].where; + expect(where.bucketStart.gte.toISOString()).toBe( + "2024-03-01T00:00:00.000Z", + ); + expect(where.bucketStart.lt.toISOString()).toBe("2024-03-20T00:00:00.000Z"); + }); + + it("prunes every type when the recompute was not type-scoped", async () => { + // A type that lost all its readings drops out of the returned rows + // entirely, so a `type: { in: }` filter could never + // reach its buckets. + queryRawUnsafe.mockResolvedValueOnce( + rowsInBucket(new Date(Date.UTC(2024, 2, 10))), + ); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from: new Date(Date.UTC(2024, 2, 1)), + to: new Date(Date.UTC(2024, 2, 20)), + }); + + const where = deleteMany.mock.calls[0][0].where; + expect(where.type).toBeUndefined(); + }); + + it("still writes the fresh rows it did compute", async () => { + queryRawUnsafe.mockResolvedValueOnce( + rowsInBucket(new Date(Date.UTC(2024, 2, 10))), + ); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + const result = await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from: new Date(Date.UTC(2024, 2, 1)), + to: new Date(Date.UTC(2024, 2, 20)), + }); + + // Delete and upsert share the one interactive transaction, so a reader + // never observes the pruned window without its replacement rows. + expect(transaction).toHaveBeenCalledTimes(1); + expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1); + expect(result.rowsUpserted).toBe(1); + }); +}); + describe("enqueueRollupRecompute", () => { it("is a silent no-op when no boss is attached", async () => { getGlobalBossMock.mockReturnValue(null); diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 519750ed2..c3e5d6552 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -26,6 +26,13 @@ * side (`analytics`, `comprehensive`, `summaries-slice`) decides * whether to fall through to the live aggregator + persist-on-miss * via its own per-tier freshness probe. + * + * Repair contract: a recompute REPLACES every bucket in the window it was + * asked for. That includes buckets the live aggregate returns nothing for — + * without it, a bucket outlives the rows it was folded from. See the + * `deleteWhere` note in `persistRollupRows` for the three shapes that hole + * took. Nothing here reads both tiers; the rollup replaces the legacy read and + * falls back to live SQL only on a coverage miss. */ import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; @@ -217,9 +224,15 @@ export async function recomputeUserRollups( from: alignedFrom, to: alignedTo, }); - if (rows.length === 0) continue; - - rowsUpserted += await persistRollupRows(userId, granularity, rows); + // No `continue` on an empty aggregate: an empty result is a STATEMENT — + // this window holds no readings any more — and the prune below is what + // acts on it. Skipping the write here is what let a bucket outlive its + // source rows. + rowsUpserted += await persistRollupRows(userId, granularity, rows, { + from: alignedFrom, + to: alignedTo, + types: opts.types, + }); } return { rowsUpserted, durationMs: Date.now() - startedAt }; @@ -263,7 +276,9 @@ export async function recomputeBucketsForMeasurement( }, }); } else { - await persistRollupRows(userId, "DAY", rows); + // No prune scope: this call owns exactly one (type, day) bucket and the + // branch above already deletes it by hand when the day emptied out. + await persistRollupRows(userId, "DAY", rows, null); } // WEEK / MONTH / YEAR — enqueue. Each granularity covers a range @@ -530,40 +545,102 @@ async function runRollupAggregate(input: { } /** - * UPSERT the aggregate rows into `measurement_rollups`. Returns the - * number of rows touched. Uses a chunked raw insert with - * `ON CONFLICT DO UPDATE` so the round-trip count is bounded - * regardless of fan-out width AND a concurrent writer on the same - * partition can never raise a unique violation. + * The window a recompute authoritatively re-derived. Every rollup bucket inside + * it is replaced by the freshly computed rows — including buckets the aggregate + * returned NOTHING for, which is the only way a bucket whose source rows all + * disappeared can be removed. `types` narrows the prune to the types the + * aggregate was scoped to; `undefined` means it covered every type. */ -async function persistRollupRows( +interface PruneScope { + from: Date; + to: Date; + types?: MeasurementType[]; +} + +/** + * Delete scope for a caller that wrote a single known bucket rather than a + * re-derived window: the [min, max] bucket range of the rows themselves, per + * type. Correct there precisely because the caller already knows which bucket + * it touched and deletes it by hand when the aggregate comes back empty. + */ +function buildReturnedRangeDeleteWhere( userId: string, granularity: RollupGranularity, rows: RollupRow[], -): Promise { - if (rows.length === 0) return 0; - - // v1.11.1 — rows are now minted per source. Delete-then-upsert the affected - // (type, bucket) partitions across ALL sources before writing, so a source - // that disappeared from a bucket (the user disconnected a provider, or - // deleted the last reading from one device) does not strand a stale - // per-source row that a plain upsert would never revisit. The delete is - // scoped to the [min, max] bucket range of the rows being written, per type - // — an indexed range delete, cheap on the single-day write hook and bounded - // on the full backfill. +) { const types = [...new Set(rows.map((r) => r.type as MeasurementType))]; - let minBucket = rows[0].bucket_start; - let maxBucket = rows[0].bucket_start; + let minBucket = rows[0]!.bucket_start; + let maxBucket = rows[0]!.bucket_start; for (const r of rows) { if (r.bucket_start < minBucket) minBucket = r.bucket_start; if (r.bucket_start > maxBucket) maxBucket = r.bucket_start; } - const deleteWhere = { + return { userId, granularity, type: { in: types }, bucketStart: { gte: minBucket, lte: maxBucket }, }; +} + +/** + * UPSERT the aggregate rows into `measurement_rollups`. Returns the + * number of rows touched. Uses a chunked raw insert with + * `ON CONFLICT DO UPDATE` so the round-trip count is bounded + * regardless of fan-out width AND a concurrent writer on the same + * partition can never raise a unique violation. + * + * `prune` is the recomputed window when the caller re-derived a whole span + * (`recomputeUserRollups`), or `null` when it wrote a single known bucket and + * handles its own empty case (`recomputeBucketsForMeasurement`). See the + * `deleteWhere` note below for why the distinction matters. + */ +async function persistRollupRows( + userId: string, + granularity: RollupGranularity, + rows: RollupRow[], + prune: PruneScope | null, +): Promise { + // v1.11.1 — rows are now minted per source. Delete-then-upsert the affected + // (type, bucket) partitions across ALL sources before writing, so a source + // that disappeared from a bucket (the user disconnected a provider, or + // deleted the last reading from one device) does not strand a stale + // per-source row that a plain upsert would never revisit. + // + // The delete is scoped to the RECOMPUTED WINDOW when the caller supplies one, + // not to the [min, max] range of the rows coming back. Scoping it to the + // returned rows made the repair self-limiting in exactly the case that needs + // it: a recompute can only ever overwrite a bucket the live aggregate still + // produces, so a bucket whose source rows all disappeared was never in the + // delete range and survived with its stale aggregate. Three shapes of the + // same hole — an empty aggregate skipped the write entirely, a returned range + // narrower than the requested window left the edges untouched, and a type + // that lost every reading dropped out of `type: { in: … }` altogether. The + // WEEK / MONTH / YEAR worker hit all three: deleting the last reading of a + // type in a week dropped the DAY bucket (the write hook deletes that one by + // hand) while the coarser buckets kept the stale aggregate forever, and the + // coverage probe still reported `hasBuckets = true` so the read-swap served + // them instead of falling back to live SQL. + const pruneWhere = prune + ? { + userId, + granularity, + ...(prune.types ? { type: { in: prune.types } } : {}), + bucketStart: { gte: prune.from, lt: prune.to }, + } + : null; + + if (rows.length === 0) { + // Nothing to write. With a prune scope there is still work to do: the + // window is authoritatively empty, so every bucket in it must go. No + // transaction needed — there is no insert leg that could leave a gap. + if (!pruneWhere) return 0; + await prisma.measurementRollup.deleteMany({ where: pruneWhere }); + return 0; + } + + const deleteWhere = + pruneWhere ?? buildReturnedRangeDeleteWhere(userId, granularity, rows); // v1.28.33 (issue #486) — the insert leg is a raw // `INSERT … ON CONFLICT (pk) DO UPDATE` instead of `createMany`. The