diff --git a/src/lib/measurements/__tests__/rollups.test.ts b/src/lib/measurements/__tests__/rollups.test.ts index 104376630..a761b92ef 100644 --- a/src/lib/measurements/__tests__/rollups.test.ts +++ b/src/lib/measurements/__tests__/rollups.test.ts @@ -83,6 +83,7 @@ void queryRaw; import { ROLLUP_FULL_BACKFILL_QUEUE, ROLLUP_RECOMPUTE_QUEUE, + _resetEnsureUserRollupsFreshInFlightForTests, collapseToTypeDayKeys, enqueueBootTimeRollupBackfill, enqueueRollupRecompute, @@ -101,6 +102,10 @@ beforeEach(() => { findFirstMeasurement.mockReset(); bossSend.mockReset(); getGlobalBossMock.mockReset(); + // v1.4.38 — clear the per-userId in-flight map so a previous test's + // resolved promise does not short-circuit the next test's recompute + // assertion. + _resetEnsureUserRollupsFreshInFlightForTests(); // Default: $transaction takes an array of pre-built promises (the // populator passes `slice.map(prisma.measurementRollup.upsert(...))`) // and returns them awaited. @@ -291,6 +296,76 @@ describe("ensureUserRollupsFresh", () => { expect(result.recomputed).toBe(false); }); + // v1.4.38 — concurrent callers for the same userId share one + // in-flight promise so a cold fan-out can never queue N parallel + // recompute runs against the same 90-day window. + it("dedups concurrent callers for the same userId onto one in-flight promise", async () => { + const rollupAt = new Date("2026-05-10T10:00:00.000Z"); + const measurementAt = new Date("2026-05-10T12:00:00.000Z"); + // Both round-trips inside the helper run twice (the helper issues + // a parallel `Promise.all` for the rollup-most-recent + measurement + // -most-recent). With dedup, those should fire ONCE across two + // concurrent callers. + findFirst.mockResolvedValue({ computedAt: rollupAt }); + findFirstMeasurement.mockResolvedValue({ + updatedAt: measurementAt, + measuredAt: measurementAt, + }); + queryRawUnsafe.mockResolvedValue([]); + + const [a, b, c] = await Promise.all([ + ensureUserRollupsFresh("user-1"), + ensureUserRollupsFresh("user-1"), + ensureUserRollupsFresh("user-1"), + ]); + expect(a.recomputed).toBe(true); + expect(b.recomputed).toBe(true); + expect(c.recomputed).toBe(true); + // The Postgres-side recompute fires exactly once across the three + // concurrent callers — proof the dedup map collapsed the fan-out. + expect(queryRawUnsafe).toHaveBeenCalledTimes(1); + // The findFirst probe fires exactly once per call type (rollup, + // measurement) across all three concurrent callers. + expect(findFirst).toHaveBeenCalledTimes(1); + expect(findFirstMeasurement).toHaveBeenCalledTimes(1); + }); + + it("clears the in-flight slot after resolution so the next call runs fresh", async () => { + const rollupAt = new Date("2026-05-10T10:00:00.000Z"); + const measurementAt = new Date("2026-05-10T12:00:00.000Z"); + findFirst.mockResolvedValue({ computedAt: rollupAt }); + findFirstMeasurement.mockResolvedValue({ + updatedAt: measurementAt, + measuredAt: measurementAt, + }); + queryRawUnsafe.mockResolvedValue([]); + + await ensureUserRollupsFresh("user-1"); + await ensureUserRollupsFresh("user-1"); + // Two serial calls => two recompute runs (no dedup short-circuit + // because the first call already resolved and dropped its slot). + expect(queryRawUnsafe).toHaveBeenCalledTimes(2); + }); + + it("clears the in-flight slot on rejection so the next call retries", async () => { + findFirst.mockRejectedValueOnce(new Error("pool exhausted")); + findFirstMeasurement.mockRejectedValueOnce(new Error("pool exhausted")); + const first = await ensureUserRollupsFresh("user-1"); + expect(first.recomputed).toBe(false); + + // Second call: probe succeeds, stale watermark triggers a recompute. + const rollupAt = new Date("2026-05-10T10:00:00.000Z"); + const measurementAt = new Date("2026-05-10T12:00:00.000Z"); + findFirst.mockResolvedValueOnce({ computedAt: rollupAt }); + findFirstMeasurement.mockResolvedValueOnce({ + updatedAt: measurementAt, + measuredAt: measurementAt, + }); + queryRawUnsafe.mockResolvedValueOnce([]); + const second = await ensureUserRollupsFresh("user-1"); + expect(second.recomputed).toBe(true); + }); + it("annotates the failure when the inner recompute throws (H3)", async () => { // Stale watermark — rollup older than the newest measurement so // the inner recompute branch fires. diff --git a/src/lib/measurements/rollups.ts b/src/lib/measurements/rollups.ts index 66f73aa3a..faee36a83 100644 --- a/src/lib/measurements/rollups.ts +++ b/src/lib/measurements/rollups.ts @@ -555,6 +555,26 @@ function bucketSpan( } } +/** + * v1.4.38 — per-userId in-flight dedup. v1.4.37.1 fire-and-forgets the + * `ensureUserRollupsFresh` call from the read path so a stale 90-day + * recompute can no longer stall the response, but if a request DOES + * end up awaiting a synchronous recompute (worker code path, future + * caller without the fire-and-forget) and the same user has a cold + * fan-out hitting analytics + comprehensive + summaries within the + * same tick, the legacy shape would queue N parallel + * `recomputeUserRollups` invocations against the same 90-day window. + * The map below collapses those into one shared promise — every + * caller after the first re-uses the in-flight resolution and only + * one Postgres aggregate runs. Mirrors the `buildCoachSnapshot` dedup + * pattern that lives one layer up. Cleared in `finally` so a slow + * promise that eventually rejects does not poison the next request. + */ +const ensureUserRollupsFreshInFlight = new Map< + string, + Promise<{ recomputed: boolean }> +>(); + /** * Cheap read-side warm-up. Called from the analytics + comprehensive * read paths so the rollup table stays current for downstream @@ -572,9 +592,39 @@ function bucketSpan( * here. * - Best-effort: thrown errors are swallowed so the read path * never fails because of a populator hiccup. + * - v1.4.38: concurrent callers for the same `userId` share one + * in-flight promise so a cold fan-out can never queue N parallel + * `recomputeUserRollups` runs against the same 90-day window. */ export async function ensureUserRollupsFresh( userId: string, +): Promise<{ recomputed: boolean }> { + // v1.4.38 — share the in-flight promise across concurrent callers + // for the same userId. The map is keyed on `userId` because the + // helper has no other parameters; the next caller hits the + // resolved promise as soon as it lands. We delete the slot in + // `finally` so a slow rejected promise does not stay cached and + // poison the next request. + const existing = ensureUserRollupsFreshInFlight.get(userId); + if (existing) return existing; + const inflight = ensureUserRollupsFreshImpl(userId).finally(() => { + ensureUserRollupsFreshInFlight.delete(userId); + }); + ensureUserRollupsFreshInFlight.set(userId, inflight); + return inflight; +} + +/** + * Test-only handle on the in-flight map. Mirrors the + * `_resetTrustViolationWarningForTests` pattern in api-response.ts — + * production code never calls this. + */ +export function _resetEnsureUserRollupsFreshInFlightForTests(): void { + ensureUserRollupsFreshInFlight.clear(); +} + +async function ensureUserRollupsFreshImpl( + userId: string, ): Promise<{ recomputed: boolean }> { try { // Single round-trip — the most-recent DAY rollup for the user