fix(rollups): match the keyed refold's days in UTC whatever the session zone

The keyed refold's day filter and delete compared the naive UTC
measured_at and bucket_start with a timestamptz[] parameter, which goes
through the session time zone; in any zone but UTC neither matched and
the refold left its buckets stale. Both now bind timestamp[] literals.
Keys are refolded in (type, day) order in chunks of 500, and the
rollup-recompute queue describes the runs of buckets a job now covers.
This commit is contained in:
Marc-André Bombeck committed 2026-10-21 00:06:21 +02:00
1 parent 1be2c27b27
commit 769d0334a7
3 files changed
+108 -9

No files matched your search

+3 -1
View File
@@ -190,7 +190,9 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
"daily-briefing": short("A light cohort scan every fifteen minutes."),
// ── Rollups ────────────────────────────────────────────────────────
"rollup-recompute": short("One bucket."),
"rollup-recompute": short(
"One type at one granularity over one bucket or a run of adjacent buckets, inside the five-year fold window.",
),
"rollup-full-backfill": {
runtime: "long",
expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS,
+29 -8
View File
@@ -362,8 +362,19 @@ export async function recomputeBucketsForMeasurement(
);
}
/**
* UTC wall-clock literals for a `timestamp[]` parameter. `measured_at` and
* `bucket_start` are naive UTC timestamps, so the instants they are compared
* with must be too: a `timestamptz[]` parameter is converted through the
* session time zone, and in any zone but UTC neither the day filter nor the
* keyed delete matched a single day, so a refold left its buckets stale.
*/
function utcStamps(dates: readonly Date[]): string[] {
return dates.map((d) => d.toISOString().replace("Z", ""));
}
/** Keys one set-based DAY refold carries; bounds the statement's arrays. */
const KEYED_REFOLD_CHUNK = 2_000;
const KEYED_REFOLD_CHUNK = 500;
/**
* The set-based sibling of {@link recomputeBucketsForMeasurement}, for a write
@@ -381,7 +392,17 @@ export async function recomputeBucketsForKeys(
userId: string,
rows: Array<{ type: MeasurementType; measuredAt: Date }>,
): Promise<void> {
const keys = collapseToTypeDayKeys(rows);
// Sorted on (type, day), the order every upsert of the tier takes its row
// locks in (`persistRollupRows` sorts the same way), and chunked small, so
// a keyed refold racing the full fold or another refold holds a short,
// deterministic slice and waits rather than interleaving lock orders.
const keys = collapseToTypeDayKeys(rows).sort((a, b) =>
a.type !== b.type
? a.type < b.type
? -1
: 1
: a.measuredAt.getTime() - b.measuredAt.getTime(),
);
for (let i = 0; i < keys.length; i += KEYED_REFOLD_CHUNK) {
await refoldDayBuckets(userId, keys.slice(i, i + KEYED_REFOLD_CHUNK));
}
@@ -441,11 +462,11 @@ async function refoldDayBuckets(
AND granularity = 'DAY'::"measurement_rollup_granularity"
AND (type, bucket_start) IN (
SELECT k.t::"measurement_type", k.b
FROM unnest($2::text[], $3::timestamptz[]) AS k(t, b)
FROM unnest($2::text[], $3::timestamp[]) AS k(t, b)
)`,
userId,
keys.map((k) => k.type),
keys.map((k) => k.measuredAt.toISOString()),
utcStamps(keys.map((k) => k.measuredAt)),
);
for (let i = 0; i < ordered.length; i += 500) {
const { sql, params } = buildRollupUpsert(
@@ -642,9 +663,9 @@ async function runRollupAggregate(
if (input.types && input.types.length === 0) return [];
// Bound as the fourth parameter, never spliced.
const dayFilter = input.days
? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])`
? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])`
: "";
const dayParams = input.days ? [input.days.map((d) => d.toISOString())] : [];
const dayParams = input.days ? [utcStamps(input.days)] : [];
// Postgres rejects `date_trunc($1::text, m."measured_at")` in a
// GROUP BY because the planner sees the parameterised expression
@@ -857,7 +878,7 @@ async function applyHourlyMeanDay(
AND m."deleted_at" IS NULL
${
input.days
? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])`
? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])`
: ""
}
GROUP BY m."type", m."source", date_trunc('hour', m."measured_at")
@@ -869,7 +890,7 @@ async function applyHourlyMeanDay(
input.userId,
input.from,
input.to,
...(input.days ? [input.days.map((d) => d.toISOString())] : []),
...(input.days ? [utcStamps(input.days)] : []),
);
const byKey = new Map(
means.map((m) => [
@@ -0,0 +1,76 @@
/**
* The keyed DAY refold does not depend on the session time zone.
*
* `measured_at` and `bucket_start` are naive UTC timestamps. The refold's day
* filter and its delete compared them with a `timestamptz[]` parameter, which
* Postgres converts through the session time zone: outside UTC neither matched
* a single day, so the refold silently left every touched bucket stale. The
* app pins its sessions to UTC, so this runs the pool in another zone on
* purpose.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
vi.mock("@/lib/db-session", async (importOriginal) => {
const actual = await importOriginal<typeof import("@/lib/db-session")>();
return { ...actual, TIME_ZONE_OPTION: "-c TimeZone=Pacific/Auckland" };
});
vi.mock("@/lib/db-compat", () => ({
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
}));
import { getPrismaClient, truncateAllTables } from "./setup";
import { prisma } from "@/lib/db";
import {
recomputeBucketsForKeys,
recomputeBucketsForMeasurement,
} from "@/lib/rollups/measurement-rollups";
beforeEach(async () => {
await truncateAllTables(getPrismaClient());
});
describe("keyed DAY refold in a non-UTC session", () => {
it("folds a new reading into its day", async () => {
const [{ zone }] = await prisma.$queryRaw<Array<{ zone: string }>>`
SELECT current_setting('TimeZone') AS zone
`;
expect(zone).toBe("Pacific/Auckland");
const user = await getPrismaClient().user.create({
data: { username: "tz-refold", email: "tz-refold@example.test" },
});
const day = new Date("2026-05-10T00:00:00.000Z");
await getPrismaClient().measurement.createMany({
data: [7, 19].map((hour, i) => ({
userId: user.id,
type: "WEIGHT" as const,
value: 80 + i,
unit: "kg",
measuredAt: new Date(day.getTime() + hour * 3_600_000),
})),
});
await recomputeBucketsForMeasurement(user.id, "WEIGHT", day);
// A third reading lands; the refold has to fold it in.
await getPrismaClient().measurement.create({
data: {
userId: user.id,
type: "WEIGHT",
value: 82,
unit: "kg",
measuredAt: new Date(day.getTime() + 21 * 3_600_000),
},
});
await recomputeBucketsForKeys(user.id, [
{ type: "WEIGHT", measuredAt: new Date(day.getTime() + 7 * 3_600_000) },
]);
const rows = await getPrismaClient().measurementRollup.findMany({
where: { userId: user.id, granularity: "DAY" },
});
expect(rows).toHaveLength(1);
expect(rows[0]!.bucketStart.toISOString()).toBe(day.toISOString());
expect(rows[0]!.count).toBe(3);
});
});