perf(mood): write a rollup fold in one statement per chunk

The full mood fold upserted every day bucket with a statement of its own,
so the read that refreshed the tier paid one statement per day of mood
history. The dashboard summary in the hot-route budget fixture did exactly
that: 57 to 59 single-row inserts out of its 85 to 87 statements, a count
that moved with the calendar the fixture is dated against, which is why it
crossed its budget of 84 on CI without a new query in the route. The fold
now writes each chunk of up to 500 buckets as one multi-row upsert, and the
route's budget goes to the measured 28.
This commit is contained in:
Marc-André Bombeck committed 2026-10-22 21:31:28 +02:00
1 parent bafda973ee
commit 70c1077121
3 files changed
+48 -52

No files matched your search

+22 -16
View File
@@ -387,33 +387,39 @@ describe("recomputeUserMoodRollups replace (delete-then-refold, v1.32.12)", () =
sd: 0.5,
};
queryRawUnsafe.mockResolvedValue([aggRow]);
upsert.mockResolvedValue({});
// The fold writes every bucket in one multi-row upsert; the values of
// that statement are the bucket key and the folded stats, in column
// order, with the run's computedAt last.
const writtenRows = () =>
executeRaw.mock.calls.map((call) =>
((call[1] as { values: unknown[] }).values ?? []).slice(0, -1),
);
await recomputeUserMoodRollups("user-1", {
granularities: ["DAY"],
replace: true,
});
const first = upsert.mock.calls.map((c) => c[0] as Record<string, unknown>);
upsert.mockClear();
const first = writtenRows();
executeRaw.mockClear();
await recomputeUserMoodRollups("user-1", {
granularities: ["DAY"],
replace: true,
});
const second = upsert.mock.calls.map(
(c) => c[0] as Record<string, unknown>,
);
const second = writtenRows();
expect(first).toHaveLength(1);
expect(second).toHaveLength(1);
// Same bucket key + same folded stats across both rebuilds (only the
// computedAt wall-clock differs, which we don't compare).
expect(first[0].where).toEqual(second[0].where);
expect((first[0].create as { count: number }).count).toBe(
(second[0].create as { count: number }).count,
);
expect((first[0].create as { mean: number }).mean).toBe(
(second[0].create as { mean: number }).mean,
);
expect(second).toEqual(first);
expect(first[0]).toEqual([
"user-1",
"DAY",
new Date("2026-05-10T00:00:00.000Z"),
3,
4.2,
3,
5,
0.5,
]);
expect(upsert).not.toHaveBeenCalled();
});
});
+25 -35
View File
@@ -51,10 +51,10 @@ import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready";
import { annotate } from "@/lib/logging/context";
import { trackBackgroundTask } from "@/lib/logging/background-tasks";
import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
import type {
import {
Prisma,
PrismaClient,
RollupGranularity,
type PrismaClient,
type RollupGranularity,
} from "@/generated/prisma/client";
import { dayKeyAsUtcMidnight } from "@/lib/tz/date-only";
@@ -668,42 +668,32 @@ async function persistMoodRollupRows(
rows: MoodRollupRow[],
): Promise<number> {
if (rows.length === 0) return 0;
// One multi-row upsert per chunk. An upsert per bucket cost a statement for
// every day of the account's history on the read that refreshed the tier
// (the dashboard summary among them), and the count moved with the
// calendar. A single statement is also atomic on its own, so it is safe
// inside a caller's interactive transaction or outside one.
const CHUNK = 500;
const computedAt = new Date();
let touched = 0;
for (let i = 0; i < rows.length; i += CHUNK) {
const slice = rows.slice(i, i + CHUNK);
// Run each upsert sequentially through the supplied client (which
// may be a transactional handle). Wrapping in `$transaction` here
// would crash when the caller already holds an interactive
// transaction, so we rely on the parent caller (or atomicity of
// the single upsert) for crash safety. The hot path is a single
// row per DAY recompute, so the round-trip cost is negligible.
for (const row of slice) {
const payload = {
count: Number(row.count),
mean: row.mean ?? 0,
minScore: row.min_score ?? 0,
maxScore: row.max_score ?? 0,
sd: row.sd,
computedAt: new Date(),
} satisfies Prisma.MoodEntryRollupUpdateInput;
await client.moodEntryRollup.upsert({
where: {
userId_granularity_bucketStart: {
userId,
granularity,
bucketStart: row.bucket_start,
},
},
create: {
userId,
granularity,
bucketStart: row.bucket_start,
...payload,
},
update: payload,
});
}
const values = slice.map(
(row) =>
Prisma.sql`(${userId}, ${granularity}::"measurement_rollup_granularity", ${row.bucket_start}, ${Number(row.count)}, ${row.mean ?? 0}, ${row.min_score ?? 0}, ${row.max_score ?? 0}, ${row.sd}, ${computedAt})`,
);
await client.$executeRaw`
INSERT INTO "mood_entry_rollups"
("user_id", "granularity", "bucket_start", "count", "mean", "min_score", "max_score", "sd", "computed_at")
VALUES ${Prisma.join(values)}
ON CONFLICT ("user_id", "granularity", "bucket_start") DO UPDATE
SET "count" = EXCLUDED."count",
"mean" = EXCLUDED."mean",
"min_score" = EXCLUDED."min_score",
"max_score" = EXCLUDED."max_score",
"sd" = EXCLUDED."sd",
"computed_at" = EXCLUDED."computed_at"
`;
touched += slice.length;
}
return touched;
@@ -67,7 +67,7 @@ const DAY_MS = 86_400_000;
*/
const BUDGET = {
authMe: 13,
dashboardSummary: 84,
dashboardSummary: 28,
dashboardSnapshot: 99,
series30: 6,
series365: 5,