chore(rollups): per-userId in-flight dedup for ensureUserRollupsFresh

v1.4.37.1 fire-and-forgets the helper from the analytics read path
so a stale recompute can no longer stall the response, but if a
request DOES await the helper synchronously (worker code path,
future caller without the fire-and-forget) a cold fan-out across
analytics + comprehensive + summaries within the same tick would
queue N parallel recomputeUserRollups runs against the same
90-day window. A module-level Map<userId, Promise> collapses the
fan-out onto one shared promise. Mirrors the buildCoachSnapshot
dedup pattern that lives one layer up. The slot is cleared in
.finally so a slow rejected promise does not poison the next call.
This commit is contained in:
Marc-André Bombeck committed 2026-05-17 17:21:47 +02:00
1 parent a8c16dc3b5
commit 4e284235fc
2 files changed
+125

No files matched your search

@@ -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.
+50
View File
@@ -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