perf(illness): parallelise the insights recovery-gap fan-out

The cross-episode insights route looped up to 24 episodes, each running a
sequential 10-vital scan with two serial reads per vital — ~500 serialised
round-trips on a year of episodes, starving the shared Prisma pool.

Pair the baseline + episode-window reads, fan the per-vital scan out under a
p-limit(4) cap, and run the per-episode correlations concurrently under the
same discipline. Series order and the gated outputs are unchanged.
This commit is contained in:
Marc-André Bombeck committed 2026-06-18 18:28:37 +02:00
1 parent cbfc4a36d2
commit 1a73afa7e9
2 files changed
+51 -25

No files matched your search

+22 -13
View File
@@ -11,6 +11,7 @@
* contributes no gap rather than a wrong one. Born-gated + owner-scoped.
*/
import { NextRequest } from "next/server";
import pLimit from "p-limit";
import { prisma } from "@/lib/db";
import { apiHandler, requireAuth } from "@/lib/api-handler";
@@ -29,6 +30,9 @@ import { resolveUserTimezone } from "@/lib/measurements/consolidation-base";
const MS_PER_DAY = 24 * 60 * 60 * 1000;
/** Cap the per-episode gap computations so a heavy history can't fan out. */
const MAX_GAP_COMPUTATIONS = 24;
/** Run the per-episode correlations under a bounded concurrency budget so
* the up-to-24 candidates don't serialise into ~500 round-trips. */
const EPISODE_GAP_CONCURRENCY = 4;
export const GET = apiHandler(async (request: NextRequest) => {
const { user } = await requireAuth();
@@ -69,19 +73,24 @@ export const GET = apiHandler(async (request: NextRequest) => {
const gapCandidates = rows
.filter((r) => r.resolvedAt !== null && r.lifecycle !== "CHRONIC_ONGOING")
.slice(0, MAX_GAP_COMPUTATIONS);
const gapById = new Map<string, number | null>();
for (const r of gapCandidates) {
const derived = await computeEpisodeCorrelation(
user.id,
{ id: r.id, onsetAt: r.onsetAt, resolvedAt: r.resolvedAt, lifecycle: r.lifecycle },
tz,
now,
);
gapById.set(
r.id,
derived.status === "ok" ? derived.value.recoveryGapDays : null,
);
}
const limit = pLimit(EPISODE_GAP_CONCURRENCY);
const gapEntries = await Promise.all(
gapCandidates.map((r) =>
limit(async (): Promise<[string, number | null]> => {
const derived = await computeEpisodeCorrelation(
user.id,
{ id: r.id, onsetAt: r.onsetAt, resolvedAt: r.resolvedAt, lifecycle: r.lifecycle },
tz,
now,
);
return [
r.id,
derived.status === "ok" ? derived.value.recoveryGapDays : null,
];
}),
),
);
const gapById = new Map<string, number | null>(gapEntries);
const episodes: RetrospectiveEpisode[] = rows.map((r) => ({
id: r.id,
+29 -12
View File
@@ -24,6 +24,7 @@
* Rollup DAY buckets are minted at the UTC midnight of the user-tz day, so
* keying their `bucketStart` through `dayKeyForUserTz` is exact.
*/
import pLimit from "p-limit";
import type { MeasurementType } from "@/generated/prisma/client";
import { prisma } from "@/lib/db";
import { readBestGranularityRollups } from "@/lib/rollups/measurement-read-wmy";
@@ -43,6 +44,13 @@ import {
const MS_PER_DAY = 24 * 60 * 60 * 1000;
/**
* Bound the per-vital fan-out so a single episode's scan can't pin the
* shared Prisma pool. Mirrors the W-POOL `p-limit(4)` discipline the
* analytics summaries slice applies.
*/
const VITAL_SCAN_CONCURRENCY = 4;
/** The episode shape the read layer needs (a subset of the Prisma row). */
export interface EpisodeForCorrelation {
id: string;
@@ -88,16 +96,30 @@ export async function computeEpisodeCorrelation(
// mitigation — the same pattern the derived route uses).
const coverage = await probeRollupCoverage(userId);
// Attempt every scan type under a bounded concurrency cap. The two
// windows per type (baseline + episode) are independent reads, so they
// run paired rather than serially (halving the per-type round-trips).
// The day-log fever read shares the same fan-out budget.
const limit = pLimit(VITAL_SCAN_CONCURRENCY);
const [perType, dayLogFever] = await Promise.all([
Promise.all(
ILLNESS_SCAN_TYPES.map((type) =>
limit(async () => {
const [baseline, episodeWin] = await Promise.all([
readWindowMeans(userId, type, baselineStart, baselineEnd, coverage, tz, now),
readWindowMeans(userId, type, preOnsetStart, end, coverage, tz, now),
]);
return { type, baseline, episodeWin };
}),
),
),
limit(() => readDayLogFever(episode.id, preOnsetStart, end, tz)),
]);
const series: VitalSeries[] = [];
let anyLive = false;
let anyDay = false;
// We attempt every scan type. An untracked vital returns empty points and
// is dropped below. Each read is a single windowed query, so attempting all
// types is bounded.
for (const type of ILLNESS_SCAN_TYPES) {
const baseline = await readWindowMeans(userId, type, baselineStart, baselineEnd, coverage, tz, now);
const episodeWin = await readWindowMeans(userId, type, preOnsetStart, end, coverage, tz, now);
for (const { type, baseline, episodeWin } of perType) {
if (baseline.points.length === 0 && episodeWin.points.length === 0) continue;
if (baseline.source === "live" || episodeWin.source === "live") anyLive = true;
if (baseline.source === "DAY" || episodeWin.source === "DAY") anyDay = true;
@@ -111,11 +133,6 @@ export async function computeEpisodeCorrelation(
});
}
// Union the canonical journaling-fever path: the illness day-log `feverC`
// series over the episode window. A user logging 39.2 °C for days with no
// HealthKit thermometer must still escalate. Per-day MAX.
const dayLogFever = await readDayLogFever(episode.id, preOnsetStart, end, tz);
const source: "DAY" | "live" | "none" = anyDay
? "DAY"
: anyLive