From c08e049a05cc44448988b5aacbd87661cd8bf1c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 16:04:04 +0200 Subject: [PATCH] perf(rollups): share an in-flight coverage probe between concurrent readers A Coach snapshot, a derived-metric batch or the Insights page fans out into a dozen readers that each probe the same account's rollup coverage at once. In production one MCP get_metrics call held 59 of these probes open on the database together, each taking about 320 ms under that contention, and the same burst repeated every 15 minutes. Concurrent probes for one account now join the query already running. Nothing is kept after it settles, each caller gets its own map, and a measurement write drops the in-flight entry so a probe that starts after a write always reads afresh. The registry lives on globalThis so every server bundle shares it. --- src/lib/cache/invalidate.ts | 5 ++ .../__tests__/coverage-inflight.test.ts | 83 +++++++++++++++++++ src/lib/rollups/coverage-inflight.ts | 48 +++++++++++ src/lib/rollups/measurement-coverage.ts | 10 ++- 4 files changed, 145 insertions(+), 1 deletion(-) create mode 100644 src/lib/rollups/__tests__/coverage-inflight.test.ts create mode 100644 src/lib/rollups/coverage-inflight.ts 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 }> >`