diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 3c3d007ff..3bc085fd0 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -190,7 +190,9 @@ export const QUEUE_RUNTIME: Readonly> = { "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, diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 0a9eae21b..a5277687f 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -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 { - 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) => [ diff --git a/tests/integration/rollup-keyed-refold-session-tz.test.ts b/tests/integration/rollup-keyed-refold-session-tz.test.ts new file mode 100644 index 000000000..30e88bb3c --- /dev/null +++ b/tests/integration/rollup-keyed-refold-session-tz.test.ts @@ -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(); + 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>` + 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); + }); +});