diff --git a/src/lib/cache/invalidate.ts b/src/lib/cache/invalidate.ts index e770551f3..40157bb1b 100644 --- a/src/lib/cache/invalidate.ts +++ b/src/lib/cache/invalidate.ts @@ -18,6 +18,8 @@ * the write endpoint's `apiSuccess()` return. No `await` overhead. */ +import { forgetCoverageProbe } from "@/lib/rollups/coverage-inflight"; + import { caches } from "./server-cache"; /** @@ -68,6 +70,9 @@ export function invalidateUserMeasurements( userId: string, opts?: { evict?: boolean }, ): void { + // A coverage probe already running read the set before this write; the + // next reader must not join it. + forgetCoverageProbe(userId); // 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. diff --git a/src/lib/rollups/__tests__/coverage-inflight.test.ts b/src/lib/rollups/__tests__/coverage-inflight.test.ts new file mode 100644 index 000000000..55154e91f --- /dev/null +++ b/src/lib/rollups/__tests__/coverage-inflight.test.ts @@ -0,0 +1,83 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { queryRaw } = vi.hoisted(() => ({ queryRaw: vi.fn() })); +vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: queryRaw } })); + +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { probeRollupCoverage } from "../measurement-coverage"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (err: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +beforeEach(() => { + queryRaw.mockReset(); +}); + +describe("probeRollupCoverage single-flight", () => { + it("shares one query between concurrent probes of an account", async () => { + const rows = deferred>(); + queryRaw.mockReturnValueOnce(rows.promise); + + const a = probeRollupCoverage("u1"); + const b = probeRollupCoverage("u1"); + rows.resolve([ + { type: "PULSE", has_buckets: true }, + { type: "WEIGHT", has_buckets: false }, + ]); + const [ma, mb] = await Promise.all([a, b]); + + expect(queryRaw).toHaveBeenCalledTimes(1); + expect(Object.fromEntries(ma)).toEqual({ PULSE: true, WEIGHT: false }); + expect(mb).toEqual(ma); + // Each caller owns its map. + expect(mb).not.toBe(ma); + ma.set("PULSE", false); + expect(mb.get("PULSE")).toBe(true); + }); + + it("keeps nothing once the query settles, and never joins across accounts", async () => { + queryRaw.mockResolvedValue([{ type: "PULSE", has_buckets: true }]); + await probeRollupCoverage("u1"); + await probeRollupCoverage("u1"); + await Promise.all([probeRollupCoverage("u1"), probeRollupCoverage("u2")]); + expect(queryRaw).toHaveBeenCalledTimes(4); + }); + + it("starts afresh after a measurement write", async () => { + const first = deferred>(); + queryRaw + .mockReturnValueOnce(first.promise) + .mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }]); + + const before = probeRollupCoverage("u1"); + invalidateUserMeasurements("u1"); + const after = probeRollupCoverage("u1"); + first.resolve([]); + + expect((await before).size).toBe(0); + expect(Object.fromEntries(await after)).toEqual({ WEIGHT: false }); + expect(queryRaw).toHaveBeenCalledTimes(2); + }); + + it("passes a failure to every joined caller and does not keep it", async () => { + const failing = deferred>(); + queryRaw + .mockReturnValueOnce(failing.promise) + .mockResolvedValueOnce([{ type: "PULSE", has_buckets: true }]); + + const a = probeRollupCoverage("u1"); + const b = probeRollupCoverage("u1"); + failing.reject(new Error("db down")); + await expect(a).rejects.toThrow("db down"); + await expect(b).rejects.toThrow("db down"); + + expect((await probeRollupCoverage("u1")).get("PULSE")).toBe(true); + }); +}); diff --git a/src/lib/rollups/coverage-inflight.ts b/src/lib/rollups/coverage-inflight.ts new file mode 100644 index 000000000..f62666469 --- /dev/null +++ b/src/lib/rollups/coverage-inflight.ts @@ -0,0 +1,48 @@ +/** + * In-flight registry for `probeRollupCoverage`, kept in its own module so the + * cache invalidators can reach it without importing the rollup read path. + * + * A single Coach snapshot, derived-metric batch or Insights page fans out into + * a dozen readers that each probe the same account's coverage at the same + * moment; one MCP `get_metrics` call was seen holding 59 of these probes open + * on the database at once. Concurrent probes for one account now share the + * query that is already running. Nothing is kept once it settles, and a + * measurement write drops the entry, so a probe that starts after a write + * always reads afresh. + */ + +// Parked on `globalThis` like the Prisma client: a production build loads this +// module once per server bundle, and a write handled in one bundle must reach +// a probe started in another. +const registry = globalThis as unknown as { + __healthlogCoverageInFlight?: Map>>; +}; +const inFlight: Map< + string, + Promise> +> = (registry.__healthlogCoverageInFlight ??= new Map()); + +/** Join the probe already running for `userId`, or start `run` as the one. */ +export function joinCoverageProbe( + userId: string, + run: () => Promise>, +): Promise> { + let shared = inFlight.get(userId); + if (!shared) { + const started = run(); + shared = started; + inFlight.set(userId, started); + const release = () => { + if (inFlight.get(userId) === started) inFlight.delete(userId); + }; + started.then(release, release); + } + // Each caller gets its own map: a reader that adjusts its copy must not + // change what the others see. + return shared.then((coverage) => new Map(coverage)); +} + +/** Called on a measurement write: the next probe for the account starts afresh. */ +export function forgetCoverageProbe(userId: string): void { + inFlight.delete(userId); +} diff --git a/src/lib/rollups/measurement-coverage.ts b/src/lib/rollups/measurement-coverage.ts index 330e0df33..9a28ff997 100644 --- a/src/lib/rollups/measurement-coverage.ts +++ b/src/lib/rollups/measurement-coverage.ts @@ -20,6 +20,7 @@ * read regardless of how many types the user has logged. */ import { prisma } from "@/lib/db"; +import { joinCoverageProbe } from "@/lib/rollups/coverage-inflight"; import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups"; /** @@ -50,10 +51,17 @@ export type RollupCoverageMap = Map; * DAY buckets on each call; on a tenant with years of Apple Health data * the planner chose a sequential scan of `measurement_rollups` for the * join. It returns the same map; only the reads changed. + * + * Concurrent probes for one account share the query already in flight (see + * `coverage-inflight.ts`); each caller still receives its own map. */ -export async function probeRollupCoverage( +export function probeRollupCoverage( userId: string, ): Promise { + return joinCoverageProbe(userId, () => readRollupCoverage(userId)); +} + +async function readRollupCoverage(userId: string): Promise { const rows = await prisma.$queryRaw< Array<{ type: string; has_buckets: boolean }> >`