mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 20:42:43 +00:00
perf(consolidation): fold a day at a time and batch the hourly writes
The consolidation drains (cumulative, daily mean, dense intra-day retention, legacy steps) gathered a type's whole history into one array before folding the first day. On an account of 1.25 million rows the heart-rate array alone held about 230 MB, and the dense retention peaked at 460 MB of heap against a 524 MB limit; in a 1 GB container running the web server and the worker together the process died of heap exhaustion within a minute and the job was later marked "job timed out" (#1031). - Pages are now handed on one at a time and each local day is folded as soon as its last row has been read, so a pass holds one page and one day. - The keyset cursor gains a redundant `measuredAt >= cursor` bound. Without it Postgres could not use the cursor as an index condition and re-read every earlier row on each page (47 ms and 248 000 filtered rows for one page halfway through the heart-rate history; 5 ms now). - The dense fold resolves a day's hourly rows with one lookup and one insert instead of two lookups and a write per hour. On the seeded account the pass went from about 290 000 statements to 54 000. - `shouldStop` lets a caller end a pass between days; the result says `stoppedEarly`. Days commit one by one and a folded day leaves the scan, so the next run continues at the first day not reached.
This commit is contained in:
Marc-André Bombeck committed
1 parent
32c3248b57
commit
f4b51cf134
13 files changed
+992
-281
No files matched your search
@@ -0,0 +1,331 @@
|
||||
/**
|
||||
* `adoptOrMintHourlyRows` resolves a whole folded day in one lookup and one
|
||||
* insert. It replaced a per-slot loop (two lookups and a write per hour, about
|
||||
* 75 statements a day), which was most of the 380 000 statements the dense
|
||||
* retention spent on a 1.25-million-row account (issue #1031).
|
||||
*
|
||||
* Two things are pinned here. The statement shape: one read, at most one
|
||||
* insert, no per-slot lookups. And the semantics: against an in-memory table
|
||||
* that enforces both unique identities, the batch leaves exactly the rows the
|
||||
* per-slot loop leaves and returns the same ids, over randomised days that
|
||||
* include rows already sitting on an anchor, rows already carrying a target
|
||||
* externalId, tombstones, and a row holding one hour's externalId while
|
||||
* parked on another hour's anchor.
|
||||
*/
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import type { MeasurementType, Prisma } from "@/generated/prisma/client";
|
||||
|
||||
import {
|
||||
adoptOrMintHourlyRows,
|
||||
type HourlySlot,
|
||||
} from "../dense-intraday-retention";
|
||||
|
||||
interface Row {
|
||||
id: string;
|
||||
externalId: string | null;
|
||||
measuredAt: Date;
|
||||
value: number;
|
||||
unit: string;
|
||||
deletedAt: Date | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* A measurement table for one (user, type, APPLE_HEALTH), with the two unique
|
||||
* identities the real one has: externalId and measuredAt (sleepStage is NULL
|
||||
* for these types). A write that would duplicate either throws, as P2002 does.
|
||||
*/
|
||||
function fakeTable(seed: Row[]) {
|
||||
const rows = seed.map((row) => ({ ...row }));
|
||||
let minted = 0;
|
||||
const calls = { findMany: 0, findFirst: 0, createMany: 0, create: 0 };
|
||||
|
||||
const assertUnique = () => {
|
||||
const ext = rows
|
||||
.filter((r) => r.externalId !== null)
|
||||
.map((r) => r.externalId);
|
||||
const at = rows.map((r) => r.measuredAt.getTime());
|
||||
if (new Set(ext).size !== ext.length) throw new Error("P2002 externalId");
|
||||
if (new Set(at).size !== at.length) throw new Error("P2002 measuredAt");
|
||||
};
|
||||
const mint = (data: {
|
||||
externalId: string;
|
||||
measuredAt: Date;
|
||||
value: number;
|
||||
unit: string;
|
||||
}) => {
|
||||
minted += 1;
|
||||
const row: Row = {
|
||||
id: `z-minted-${String(minted).padStart(3, "0")}`,
|
||||
externalId: data.externalId,
|
||||
measuredAt: data.measuredAt,
|
||||
value: data.value,
|
||||
unit: data.unit,
|
||||
deletedAt: null,
|
||||
};
|
||||
rows.push(row);
|
||||
assertUnique();
|
||||
return row;
|
||||
};
|
||||
const byId = () => [...rows].sort((a, b) => (a.id < b.id ? -1 : 1));
|
||||
|
||||
const tx = {
|
||||
measurement: {
|
||||
findMany: vi.fn(
|
||||
async (args: {
|
||||
where: {
|
||||
OR: [
|
||||
{ externalId: { in: string[] } },
|
||||
{ measuredAt: { in: Date[] } },
|
||||
];
|
||||
};
|
||||
}) => {
|
||||
calls.findMany += 1;
|
||||
const ext = new Set(args.where.OR[0].externalId.in);
|
||||
const at = new Set(
|
||||
args.where.OR[1].measuredAt.in.map((d) => d.getTime()),
|
||||
);
|
||||
return byId()
|
||||
.filter(
|
||||
(r) =>
|
||||
(r.externalId !== null && ext.has(r.externalId)) ||
|
||||
at.has(r.measuredAt.getTime()),
|
||||
)
|
||||
.map((r) => ({
|
||||
id: r.id,
|
||||
externalId: r.externalId,
|
||||
measuredAt: r.measuredAt,
|
||||
}));
|
||||
},
|
||||
),
|
||||
findFirst: vi.fn(
|
||||
async (args: { where: { externalId?: string; measuredAt?: Date } }) => {
|
||||
calls.findFirst += 1;
|
||||
const hit = byId().find((r) =>
|
||||
args.where.externalId !== undefined
|
||||
? r.externalId === args.where.externalId
|
||||
: r.measuredAt.getTime() === args.where.measuredAt!.getTime(),
|
||||
);
|
||||
return hit ? { id: hit.id } : null;
|
||||
},
|
||||
),
|
||||
update: vi.fn(
|
||||
async (args: {
|
||||
where: { id: string };
|
||||
data: Partial<Row> & { measuredAt?: Date };
|
||||
}) => {
|
||||
const row = rows.find((r) => r.id === args.where.id)!;
|
||||
Object.assign(row, args.data);
|
||||
assertUnique();
|
||||
return row;
|
||||
},
|
||||
),
|
||||
create: vi.fn(
|
||||
async (args: {
|
||||
data: {
|
||||
externalId: string;
|
||||
measuredAt: Date;
|
||||
value: number;
|
||||
unit: string;
|
||||
};
|
||||
}) => {
|
||||
calls.create += 1;
|
||||
return { id: mint(args.data).id };
|
||||
},
|
||||
),
|
||||
createManyAndReturn: vi.fn(
|
||||
async (args: {
|
||||
data: Array<{
|
||||
externalId: string;
|
||||
measuredAt: Date;
|
||||
value: number;
|
||||
unit: string;
|
||||
}>;
|
||||
}) => {
|
||||
calls.createMany += 1;
|
||||
return args.data.map((d) => {
|
||||
const row = mint(d);
|
||||
return { id: row.id, externalId: row.externalId };
|
||||
});
|
||||
},
|
||||
),
|
||||
},
|
||||
};
|
||||
return {
|
||||
tx: tx as unknown as Prisma.TransactionClient,
|
||||
rows,
|
||||
calls,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* The per-slot loop the batch replaced, kept verbatim in behaviour as the
|
||||
* reference: externalId first, then the anchor occupant, lowest id on a tie;
|
||||
* re-anchor only onto a free slot; otherwise mint.
|
||||
*/
|
||||
async function perSlotReference(
|
||||
tx: Prisma.TransactionClient,
|
||||
unit: string,
|
||||
slots: readonly HourlySlot[],
|
||||
): Promise<string[]> {
|
||||
const ids: string[] = [];
|
||||
for (const slot of slots) {
|
||||
const eidRow = await tx.measurement.findFirst({
|
||||
where: { externalId: slot.externalId },
|
||||
} as Prisma.MeasurementFindFirstArgs);
|
||||
const slotRow = await tx.measurement.findFirst({
|
||||
where: { measuredAt: slot.anchor },
|
||||
} as Prisma.MeasurementFindFirstArgs);
|
||||
const target = eidRow ?? slotRow;
|
||||
if (target) {
|
||||
const free = slotRow === null || slotRow.id === target.id;
|
||||
await tx.measurement.update({
|
||||
where: { id: target.id },
|
||||
data: {
|
||||
value: slot.value,
|
||||
unit,
|
||||
externalId: slot.externalId,
|
||||
deletedAt: null,
|
||||
...(free ? { measuredAt: slot.anchor } : {}),
|
||||
},
|
||||
});
|
||||
ids.push(target.id);
|
||||
} else {
|
||||
const created = await tx.measurement.create({
|
||||
data: {
|
||||
value: slot.value,
|
||||
unit,
|
||||
measuredAt: slot.anchor,
|
||||
externalId: slot.externalId,
|
||||
},
|
||||
} as Prisma.MeasurementCreateArgs);
|
||||
ids.push(created.id);
|
||||
}
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
|
||||
const HOUR = 3_600_000;
|
||||
const DAY0 = Date.UTC(2026, 4, 1, 0, 30);
|
||||
const anchor = (h: number) => new Date(DAY0 + h * HOUR);
|
||||
const ext = (h: number) => `stats:HK:2026-05-01T${String(h).padStart(2, "0")}`;
|
||||
|
||||
/** A deterministic pseudo-random generator, so a failure reproduces. */
|
||||
function rng(seed: number) {
|
||||
let x = seed;
|
||||
return () => {
|
||||
x = (x * 1103515245 + 12345) % 2 ** 31;
|
||||
return x / 2 ** 31;
|
||||
};
|
||||
}
|
||||
|
||||
function scenario(seed: number): { seed: Row[]; slots: HourlySlot[] } {
|
||||
const rand = rng(seed);
|
||||
const hours = [...Array(24).keys()].filter(() => rand() < 0.7);
|
||||
const slots = hours.map((h) => ({
|
||||
externalId: ext(h),
|
||||
anchor: anchor(h),
|
||||
value: Math.round(rand() * 100),
|
||||
}));
|
||||
const rows: Row[] = [];
|
||||
const takenAt = new Set<number>();
|
||||
const takenExt = new Set<string>();
|
||||
let n = 0;
|
||||
const add = (externalId: string | null, at: Date) => {
|
||||
if (takenAt.has(at.getTime())) return;
|
||||
if (externalId !== null && takenExt.has(externalId)) return;
|
||||
takenAt.add(at.getTime());
|
||||
if (externalId !== null) takenExt.add(externalId);
|
||||
n += 1;
|
||||
rows.push({
|
||||
id: `r-${String(Math.floor(rand() * 900) + 100)}-${n}`,
|
||||
externalId,
|
||||
measuredAt: at,
|
||||
value: 1,
|
||||
unit: "bpm",
|
||||
deletedAt: rand() < 0.3 ? new Date(DAY0) : null,
|
||||
});
|
||||
};
|
||||
for (const h of hours) {
|
||||
const roll = rand();
|
||||
if (roll < 0.15)
|
||||
add(ext(h), anchor(h)); // a prior fold's own row
|
||||
else if (roll < 0.3)
|
||||
add(`hk-sample-${h}`, anchor(h)); // a sample on the anchor
|
||||
else if (roll < 0.4)
|
||||
add(ext(h), new Date(DAY0 + h * HOUR + 60_000)); // stats row off-anchor
|
||||
else if (roll < 0.5) {
|
||||
// This hour's externalId parked on ANOTHER hour's anchor.
|
||||
const other = hours[Math.floor(rand() * hours.length)]!;
|
||||
add(ext(h), anchor(other));
|
||||
}
|
||||
}
|
||||
return { seed: rows, slots };
|
||||
}
|
||||
|
||||
function snapshot(rows: Row[]) {
|
||||
return [...rows]
|
||||
.map((r) => ({
|
||||
id: r.id.startsWith("z-minted") ? "minted" : r.id,
|
||||
externalId: r.externalId,
|
||||
at: r.measuredAt.toISOString(),
|
||||
value: r.value,
|
||||
deleted: r.deletedAt !== null,
|
||||
}))
|
||||
.sort((a, b) =>
|
||||
`${a.externalId}|${a.at}` < `${b.externalId}|${b.at}` ? -1 : 1,
|
||||
);
|
||||
}
|
||||
|
||||
describe("adoptOrMintHourlyRows", () => {
|
||||
it("resolves a fresh day with one lookup and one insert, no per-slot reads", async () => {
|
||||
const { tx, calls, rows } = fakeTable([]);
|
||||
const slots = [...Array(24).keys()].map((h) => ({
|
||||
externalId: ext(h),
|
||||
anchor: anchor(h),
|
||||
value: h,
|
||||
}));
|
||||
|
||||
const ids = await adoptOrMintHourlyRows(tx, {
|
||||
userId: "u",
|
||||
type: "PULSE" as MeasurementType,
|
||||
unit: "bpm",
|
||||
slots,
|
||||
});
|
||||
|
||||
expect(calls).toEqual({
|
||||
findMany: 1,
|
||||
findFirst: 0,
|
||||
createMany: 1,
|
||||
create: 0,
|
||||
});
|
||||
expect(rows).toHaveLength(24);
|
||||
expect(ids).toEqual(
|
||||
slots.map((s) => rows.find((r) => r.externalId === s.externalId)!.id),
|
||||
);
|
||||
});
|
||||
|
||||
it("leaves the same rows and returns the same ids as the per-slot loop", async () => {
|
||||
for (let seed = 1; seed <= 300; seed++) {
|
||||
const { seed: rows, slots } = scenario(seed);
|
||||
|
||||
const batch = fakeTable(rows);
|
||||
const batchIds = await adoptOrMintHourlyRows(batch.tx, {
|
||||
userId: "u",
|
||||
type: "PULSE" as MeasurementType,
|
||||
unit: "bpm",
|
||||
slots,
|
||||
});
|
||||
|
||||
const reference = fakeTable(rows);
|
||||
const referenceIds = await perSlotReference(reference.tx, "bpm", slots);
|
||||
|
||||
const norm = (ids: string[]) =>
|
||||
ids.map((id) => (id.startsWith("z-minted") ? "minted" : id));
|
||||
expect(norm(batchIds), `seed ${seed}`).toEqual(norm(referenceIds));
|
||||
expect(snapshot(batch.rows), `seed ${seed}`).toEqual(
|
||||
snapshot(reference.rows),
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -1,31 +1,50 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { bucketRowsByDay, scanSourceRowsPaged } from "../consolidation-base";
|
||||
import {
|
||||
bucketRowsByDay,
|
||||
iterateDayBuckets,
|
||||
iterateSourcePages,
|
||||
runConsolidation,
|
||||
type ConsolidationParams,
|
||||
} from "../consolidation-base";
|
||||
import type { PerSampleRow } from "../consolidation-tz";
|
||||
import type { Prisma, PrismaClient } from "@/generated/prisma/client";
|
||||
import type {
|
||||
MeasurementType,
|
||||
Prisma,
|
||||
PrismaClient,
|
||||
} from "@/generated/prisma/client";
|
||||
|
||||
// `scanSourceRowsPaged` keyset-paginates the per-(user, type) source scan so a
|
||||
// multi-year tenant no longer materialises its whole history in one Prisma
|
||||
// result set. These tests pin the two properties that matter: the accumulated
|
||||
// array equals a single unbounded scan (same rows, same order, no gaps /
|
||||
// duplicates), and the cursor advances on `(measuredAt, id)` after every full
|
||||
// page.
|
||||
// The consolidation drains walk a user's whole history of a type. Issue #1031:
|
||||
// on an account with 1.25 million rows the walk gathered every page into one
|
||||
// array before folding a single day (≈230 MB of heap for the heart-rate type
|
||||
// alone) and paged with a cursor Postgres could not use as an index condition,
|
||||
// so every page re-read the rows before it. These tests pin the three
|
||||
// properties the fix rests on: pages are handed on one at a time and a day is
|
||||
// folded before the scan reads past it, the cursor carries an index-usable
|
||||
// bound, and a pass stops cleanly when asked to.
|
||||
|
||||
/** Build a deterministic per-sample row at a 1-minute-per-index offset. */
|
||||
function row(index: number): PerSampleRow {
|
||||
/** A deterministic per-sample row at `index` hours past 2026-01-01T00:00Z. */
|
||||
function row(index: number, minuteOffset = 0): PerSampleRow {
|
||||
return {
|
||||
id: `m-${String(index).padStart(4, "0")}`,
|
||||
type: "ACTIVITY_STEPS",
|
||||
value: index,
|
||||
measuredAt: new Date(Date.UTC(2026, 0, 1, 0, index, 0)),
|
||||
measuredAt: new Date(Date.UTC(2026, 0, 1, index, minuteOffset, 0)),
|
||||
externalId: `hk-uuid-${index}`,
|
||||
};
|
||||
}
|
||||
|
||||
interface CursorWhere {
|
||||
AND?: [
|
||||
Prisma.MeasurementWhereInput,
|
||||
{ measuredAt?: { gte?: Date } },
|
||||
{ OR?: Array<{ measuredAt?: Date | { gt?: Date }; id?: { gt?: string } }> },
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* A findMany mock that serves `allRows` in keyset order, honouring `take`
|
||||
* (page size) and the `(measuredAt, id)` cursor clause the helper AND-combines
|
||||
* onto the base where. This mirrors how Postgres would resolve the keyset.
|
||||
* A findMany mock that serves `allRows` in keyset order, honouring `take` and
|
||||
* the `(measuredAt, id)` cursor the pager AND-combines onto the base where.
|
||||
*/
|
||||
function buildPagingFindMany(allRows: PerSampleRow[]) {
|
||||
const ordered = [...allRows].sort((a, b) => {
|
||||
@@ -34,43 +53,29 @@ function buildPagingFindMany(allRows: PerSampleRow[]) {
|
||||
});
|
||||
|
||||
return vi.fn(
|
||||
async (args: {
|
||||
where: Prisma.MeasurementWhereInput;
|
||||
take?: number;
|
||||
orderBy?: unknown;
|
||||
}) => {
|
||||
async (args: { where: CursorWhere; take?: number; orderBy?: unknown }) => {
|
||||
const take = args.take ?? ordered.length;
|
||||
|
||||
// Extract the keyset cursor the helper threads through `where.AND[1].OR`.
|
||||
let afterTime: number | null = null;
|
||||
let afterId: string | null = null;
|
||||
const and = (args.where as { AND?: unknown[] }).AND;
|
||||
if (Array.isArray(and)) {
|
||||
const or = (and[1] as { OR?: Array<Record<string, unknown>> }).OR;
|
||||
const gt = or?.[0]?.measuredAt as { gt?: Date } | undefined;
|
||||
const tie = or?.[1] as
|
||||
{ measuredAt?: Date; id?: { gt?: string } } | undefined;
|
||||
afterTime = gt?.gt ? gt.gt.getTime() : null;
|
||||
afterId = tie?.id?.gt ?? null;
|
||||
}
|
||||
|
||||
const remaining =
|
||||
afterTime === null
|
||||
? ordered
|
||||
: ordered.filter((r) => {
|
||||
const t = r.measuredAt.getTime();
|
||||
return t > afterTime! || (t === afterTime! && r.id > afterId!);
|
||||
});
|
||||
|
||||
return remaining.slice(0, take);
|
||||
const and = args.where.AND;
|
||||
if (!and) return ordered.slice(0, take);
|
||||
const gte = and[1].measuredAt?.gte?.getTime();
|
||||
const or = and[2].OR ?? [];
|
||||
const gt = (
|
||||
or[0]?.measuredAt as { gt?: Date } | undefined
|
||||
)?.gt?.getTime();
|
||||
const tieId = or[1]?.id?.gt;
|
||||
return ordered
|
||||
.filter((r) => {
|
||||
const t = r.measuredAt.getTime();
|
||||
if (gte !== undefined && t < gte) return false;
|
||||
return t > gt! || (t === gt! && r.id > tieId!);
|
||||
})
|
||||
.slice(0, take);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
function mockClient(findMany: ReturnType<typeof vi.fn>): PrismaClient {
|
||||
return {
|
||||
measurement: { findMany },
|
||||
} as unknown as PrismaClient;
|
||||
return { measurement: { findMany } } as unknown as PrismaClient;
|
||||
}
|
||||
|
||||
const BASE_WHERE: Prisma.MeasurementWhereInput = {
|
||||
@@ -85,21 +90,25 @@ const SELECT: Prisma.MeasurementSelect = {
|
||||
externalId: true,
|
||||
};
|
||||
|
||||
describe("scanSourceRowsPaged", () => {
|
||||
it("accumulates every row in keyset order across multiple pages with no gaps or duplicates", async () => {
|
||||
async function collectPages(
|
||||
pages: AsyncIterable<PerSampleRow[]>,
|
||||
): Promise<PerSampleRow[][]> {
|
||||
const out: PerSampleRow[][] = [];
|
||||
for await (const page of pages) out.push(page);
|
||||
return out;
|
||||
}
|
||||
|
||||
describe("iterateSourcePages", () => {
|
||||
it("yields every row once, in keyset order, one bounded page at a time", async () => {
|
||||
const all = Array.from({ length: 23 }, (_, i) => row(i));
|
||||
const findMany = buildPagingFindMany(all);
|
||||
|
||||
const result = await scanSourceRowsPaged(
|
||||
mockClient(findMany),
|
||||
BASE_WHERE,
|
||||
SELECT,
|
||||
10,
|
||||
const pages = await collectPages(
|
||||
iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 10),
|
||||
);
|
||||
|
||||
// 23 rows at page size 10 → pages of 10, 10, 3 (the short page ends it).
|
||||
expect(result.map((r) => r.id)).toEqual(all.map((r) => r.id));
|
||||
expect(new Set(result.map((r) => r.id)).size).toBe(all.length);
|
||||
expect(pages.map((p) => p.length)).toEqual([10, 10, 3]);
|
||||
expect(pages.flat().map((r) => r.id)).toEqual(all.map((r) => r.id));
|
||||
expect(findMany).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
@@ -107,88 +116,191 @@ describe("scanSourceRowsPaged", () => {
|
||||
const all = Array.from({ length: 20 }, (_, i) => row(i));
|
||||
const findMany = buildPagingFindMany(all);
|
||||
|
||||
const result = await scanSourceRowsPaged(
|
||||
mockClient(findMany),
|
||||
BASE_WHERE,
|
||||
SELECT,
|
||||
10,
|
||||
const pages = await collectPages(
|
||||
iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 10),
|
||||
);
|
||||
|
||||
// 20 rows at page size 10 → full pages 10, 10, then an empty probe page.
|
||||
expect(result).toHaveLength(20);
|
||||
expect(pages.flat()).toHaveLength(20);
|
||||
expect(findMany).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("advances the cursor on (measuredAt, id) after each full page", async () => {
|
||||
it("yields nothing when the first page is empty", async () => {
|
||||
const findMany = buildPagingFindMany([]);
|
||||
const pages = await collectPages(
|
||||
iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 10),
|
||||
);
|
||||
expect(pages).toEqual([]);
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("bounds every later page with an index-usable measuredAt >= cursor conjunct", async () => {
|
||||
// `measuredAt > c OR (measuredAt = c AND id > i)` alone is not an index
|
||||
// condition: Postgres walked the index from the user's first row on
|
||||
// every page. The redundant `>=` bound is what the planner can use.
|
||||
const all = Array.from({ length: 12 }, (_, i) => row(i));
|
||||
const findMany = buildPagingFindMany(all);
|
||||
|
||||
await scanSourceRowsPaged(mockClient(findMany), BASE_WHERE, SELECT, 5);
|
||||
await collectPages(
|
||||
iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 5),
|
||||
);
|
||||
|
||||
// First page: no cursor (base where only).
|
||||
const firstWhere = findMany.mock.calls[0]?.[0]?.where;
|
||||
expect((firstWhere as { AND?: unknown }).AND).toBeUndefined();
|
||||
expect(firstWhere).toEqual(BASE_WHERE);
|
||||
|
||||
// Second page: cursor resumes after the 5th row (id m-0004).
|
||||
const secondWhere = findMany.mock.calls[1]?.[0]?.where as {
|
||||
AND: unknown[];
|
||||
};
|
||||
expect(Array.isArray(secondWhere.AND)).toBe(true);
|
||||
expect(secondWhere.AND[0]).toEqual(BASE_WHERE);
|
||||
const cursorClause = secondWhere.AND[1] as {
|
||||
OR: Array<{ measuredAt?: { gt?: Date }; id?: { gt?: string } }>;
|
||||
};
|
||||
expect(cursorClause.OR[0].measuredAt?.gt).toEqual(row(4).measuredAt);
|
||||
expect(cursorClause.OR[1].id?.gt).toBe(row(4).id);
|
||||
|
||||
// Every page is requested in stable keyset order.
|
||||
expect(findMany.mock.calls[0]?.[0]?.where).toEqual(BASE_WHERE);
|
||||
const second = findMany.mock.calls[1]?.[0]?.where as CursorWhere;
|
||||
expect(second.AND?.[0]).toEqual(BASE_WHERE);
|
||||
expect(second.AND?.[1]).toEqual({
|
||||
measuredAt: { gte: row(4).measuredAt },
|
||||
});
|
||||
expect(second.AND?.[2]).toEqual({
|
||||
OR: [
|
||||
{ measuredAt: { gt: row(4).measuredAt } },
|
||||
{ measuredAt: row(4).measuredAt, id: { gt: row(4).id } },
|
||||
],
|
||||
});
|
||||
for (const call of findMany.mock.calls) {
|
||||
expect(call[0]?.orderBy).toEqual([{ measuredAt: "asc" }, { id: "asc" }]);
|
||||
}
|
||||
});
|
||||
|
||||
it("returns the empty array when the first page is empty", async () => {
|
||||
const findMany = buildPagingFindMany([]);
|
||||
const result = await scanSourceRowsPaged(
|
||||
mockClient(findMany),
|
||||
BASE_WHERE,
|
||||
SELECT,
|
||||
10,
|
||||
);
|
||||
expect(result).toEqual([]);
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("produces output that buckets identically to a single unpaged scan", async () => {
|
||||
// Rows spanning three calendar days; the paged accumulation must bucket
|
||||
// byte-identically to the same rows fed in one array.
|
||||
const all = [
|
||||
{ ...row(1), measuredAt: new Date("2026-05-16T08:00:00.000Z") },
|
||||
{ ...row(2), measuredAt: new Date("2026-05-16T14:00:00.000Z") },
|
||||
{ ...row(3), measuredAt: new Date("2026-05-17T06:00:00.000Z") },
|
||||
{ ...row(4), measuredAt: new Date("2026-05-17T20:00:00.000Z") },
|
||||
{ ...row(5), measuredAt: new Date("2026-05-18T09:00:00.000Z") },
|
||||
];
|
||||
it("resumes past rows that share the boundary instant without gaps or duplicates", async () => {
|
||||
// Five rows on the same instant straddle the page boundary.
|
||||
const all = [0, 1, 2, 3, 4].map((i) => ({
|
||||
...row(i),
|
||||
measuredAt: new Date("2026-05-16T08:00:00.000Z"),
|
||||
}));
|
||||
const findMany = buildPagingFindMany(all);
|
||||
|
||||
const paged = await scanSourceRowsPaged(
|
||||
mockClient(findMany),
|
||||
BASE_WHERE,
|
||||
SELECT,
|
||||
2,
|
||||
const pages = await collectPages(
|
||||
iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 2),
|
||||
);
|
||||
expect(pages.flat().map((r) => r.id)).toEqual(all.map((r) => r.id));
|
||||
});
|
||||
});
|
||||
|
||||
const baselineByDay = bucketRowsByDay(all, "Europe/Berlin", "stats:");
|
||||
const pagedByDay = bucketRowsByDay(paged, "Europe/Berlin", "stats:");
|
||||
describe("iterateDayBuckets", () => {
|
||||
const days = [
|
||||
{ ...row(1), measuredAt: new Date("2026-05-16T08:00:00.000Z") },
|
||||
{ ...row(2), measuredAt: new Date("2026-05-16T14:00:00.000Z") },
|
||||
{
|
||||
...row(3),
|
||||
externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16",
|
||||
measuredAt: new Date("2026-05-16T15:00:00.000Z"),
|
||||
},
|
||||
{ ...row(4), measuredAt: new Date("2026-05-17T06:00:00.000Z") },
|
||||
{ ...row(5), measuredAt: new Date("2026-05-17T20:00:00.000Z") },
|
||||
{ ...row(6), measuredAt: new Date("2026-05-18T09:00:00.000Z") },
|
||||
];
|
||||
|
||||
expect([...pagedByDay.keys()].sort()).toEqual(
|
||||
[...baselineByDay.keys()].sort(),
|
||||
);
|
||||
for (const [dateKey, rows] of baselineByDay) {
|
||||
expect(pagedByDay.get(dateKey)?.map((r) => r.id)).toEqual(
|
||||
it("groups exactly as the whole-array bucketing does, stats rows skipped", async () => {
|
||||
const findMany = buildPagingFindMany(days);
|
||||
let scanned = 0;
|
||||
const streamed = new Map<string, string[]>();
|
||||
for await (const [key, rows] of iterateDayBuckets(
|
||||
iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 2),
|
||||
"Europe/Berlin",
|
||||
"stats:",
|
||||
() => {
|
||||
scanned += 1;
|
||||
},
|
||||
)) {
|
||||
streamed.set(
|
||||
key,
|
||||
rows.map((r) => r.id),
|
||||
);
|
||||
}
|
||||
|
||||
const baseline = bucketRowsByDay(days, "Europe/Berlin", "stats:");
|
||||
expect([...streamed.keys()]).toEqual([...baseline.keys()]);
|
||||
for (const [key, rows] of baseline) {
|
||||
expect(streamed.get(key)).toEqual(rows.map((r) => r.id));
|
||||
}
|
||||
expect(scanned).toBe(days.length);
|
||||
});
|
||||
|
||||
it("hands a day on before the scan reads the page after the one that closed it", async () => {
|
||||
const findMany = buildPagingFindMany(days);
|
||||
const pagesReadWhenYielded: number[] = [];
|
||||
for await (const _day of iterateDayBuckets(
|
||||
iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 2),
|
||||
"Europe/Berlin",
|
||||
"stats:",
|
||||
)) {
|
||||
pagesReadWhenYielded.push(findMany.mock.calls.length);
|
||||
}
|
||||
// Pages: [r1 r2] [r3 r4] [r5 r6] [] — 05-16 closes on page 2 (r4),
|
||||
// 05-17 on page 3 (r6), 05-18 at the end of the scan.
|
||||
expect(pagesReadWhenYielded).toEqual([2, 3, 4]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("runConsolidation — streamed walk", () => {
|
||||
/** One row per hour for `dayCount` UTC days. */
|
||||
function hourlyRows(dayCount: number): PerSampleRow[] {
|
||||
return Array.from({ length: dayCount * 24 }, (_, i) => row(i, 30));
|
||||
}
|
||||
|
||||
function params(
|
||||
findMany: ReturnType<typeof vi.fn>,
|
||||
writeDay: ConsolidationParams<MeasurementType>["writeDay"],
|
||||
shouldStop?: () => boolean,
|
||||
): ConsolidationParams<MeasurementType> {
|
||||
const client = {
|
||||
user: {
|
||||
findMany: vi
|
||||
.fn()
|
||||
.mockResolvedValue([{ id: "user-1", timezone: "UTC" }]),
|
||||
},
|
||||
measurement: { findMany },
|
||||
} as unknown as PrismaClient;
|
||||
return {
|
||||
prismaClient: client,
|
||||
options: { shouldStop },
|
||||
types: ["ACTIVITY_STEPS"] as MeasurementType[],
|
||||
hkIdentifierForType: () => "HKQuantityTypeIdentifierStepCount",
|
||||
dailyStatsExternalId: (hk: string, day: string) => `stats:${hk}:${day}`,
|
||||
statsPrefix: "stats:",
|
||||
reduce: (rows: readonly PerSampleRow[]) => rows.length,
|
||||
buildScanWhere: () => BASE_WHERE,
|
||||
writeDay,
|
||||
recordBucket: () => {},
|
||||
};
|
||||
}
|
||||
|
||||
it("folds the first day before the scan has read the whole history", async () => {
|
||||
// 500 days × 24 rows = 12 000 rows → default pages of 5000, 5000, 2000.
|
||||
const all = hourlyRows(500);
|
||||
const findMany = buildPagingFindMany(all);
|
||||
const pagesReadAtFirstWrite: number[] = [];
|
||||
const writeDay = vi.fn(async () => {
|
||||
if (pagesReadAtFirstWrite.length === 0) {
|
||||
pagesReadAtFirstWrite.push(findMany.mock.calls.length);
|
||||
}
|
||||
return { kind: "written" as const, sourceRowsRemoved: 0 };
|
||||
});
|
||||
|
||||
const result = await runConsolidation(params(findMany, writeDay));
|
||||
|
||||
expect(writeDay).toHaveBeenCalledTimes(500);
|
||||
expect(findMany).toHaveBeenCalledTimes(3);
|
||||
// The old walk read all three pages before the first write.
|
||||
expect(pagesReadAtFirstWrite).toEqual([1]);
|
||||
expect(result.stoppedEarly).toBe(false);
|
||||
});
|
||||
|
||||
it("stops before the next day once shouldStop answers true, and reads no further", async () => {
|
||||
const all = hourlyRows(500);
|
||||
const findMany = buildPagingFindMany(all);
|
||||
const writeDay = vi.fn(async () => ({
|
||||
kind: "written" as const,
|
||||
sourceRowsRemoved: 0,
|
||||
}));
|
||||
let budget = 3;
|
||||
const shouldStop = () => budget-- <= 0;
|
||||
|
||||
const result = await runConsolidation(
|
||||
params(findMany, writeDay, shouldStop),
|
||||
);
|
||||
|
||||
expect(writeDay).toHaveBeenCalledTimes(3);
|
||||
expect(result.stoppedEarly).toBe(true);
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -30,6 +30,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
|
||||
|
||||
import { runDenseIntradayHourlyRebuild } from "../dense-intraday-hourly-rebuild";
|
||||
import type { PrismaClient } from "@/generated/prisma/client";
|
||||
import { candidateLookup, createManyVia } from "./hourly-mint-mock";
|
||||
|
||||
const HRV_DAILY_ID =
|
||||
"stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01";
|
||||
@@ -79,6 +80,8 @@ function buildPrismaMock(opts: {
|
||||
create: txCreate,
|
||||
update: txUpdate,
|
||||
findFirst: txFindFirst,
|
||||
findMany: candidateLookup(),
|
||||
createManyAndReturn: createManyVia(txCreate),
|
||||
},
|
||||
};
|
||||
|
||||
@@ -444,6 +447,10 @@ describe("runDenseIntradayHourlyRebuild — rebuild flow", () => {
|
||||
create: vi.fn().mockResolvedValue({ id: "minted" }),
|
||||
update: vi.fn().mockResolvedValue({}),
|
||||
findFirst: vi.fn().mockResolvedValue(null),
|
||||
findMany: candidateLookup(),
|
||||
createManyAndReturn: createManyVia(
|
||||
vi.fn().mockResolvedValue({ id: "minted" }),
|
||||
),
|
||||
},
|
||||
});
|
||||
},
|
||||
|
||||
@@ -33,6 +33,7 @@ import {
|
||||
} from "../dense-intraday-retention";
|
||||
import type { PerSampleRow } from "../drain-per-sample-cumulative";
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
import { candidateLookup, createManyVia } from "./hourly-mint-mock";
|
||||
|
||||
function pulseRow(id: string, value: number, iso: string): PerSampleRow {
|
||||
return {
|
||||
@@ -68,6 +69,8 @@ function buildPrismaMock(opts: {
|
||||
update: txUpdate,
|
||||
findFirst: txFindFirst,
|
||||
updateMany: txUpdateMany,
|
||||
findMany: candidateLookup(),
|
||||
createManyAndReturn: createManyVia(txCreate),
|
||||
},
|
||||
};
|
||||
|
||||
@@ -325,6 +328,10 @@ describe("PULSE fold — per-day failure boundary", () => {
|
||||
update: vi.fn(),
|
||||
findFirst: vi.fn().mockResolvedValue(null),
|
||||
updateMany: vi.fn().mockResolvedValue({ count: 5 }),
|
||||
findMany: candidateLookup(),
|
||||
createManyAndReturn: createManyVia(
|
||||
vi.fn().mockResolvedValue({ id: "minted" }),
|
||||
),
|
||||
},
|
||||
});
|
||||
},
|
||||
|
||||
@@ -32,6 +32,7 @@ import {
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
} from "../dense-intraday-retention";
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
import { candidateLookup, createManyVia } from "./hourly-mint-mock";
|
||||
|
||||
/** A day well outside the retention window, so the fold definitely runs. */
|
||||
const FOLD_DAY = new Date(
|
||||
@@ -76,7 +77,15 @@ function buildPrismaMock(existingResting: Array<{ source: string }>) {
|
||||
);
|
||||
|
||||
const tx = {
|
||||
measurement: { create, update, findFirst: txFindFirst, updateMany, upsert },
|
||||
measurement: {
|
||||
create,
|
||||
update,
|
||||
findFirst: txFindFirst,
|
||||
updateMany,
|
||||
upsert,
|
||||
findMany: candidateLookup(),
|
||||
createManyAndReturn: createManyVia(create),
|
||||
},
|
||||
};
|
||||
|
||||
return {
|
||||
|
||||
@@ -25,6 +25,7 @@ import {
|
||||
} from "../dense-intraday-retention";
|
||||
import type { PerSampleRow } from "../drain-per-sample-cumulative";
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
import { candidateLookup, createManyVia } from "./hourly-mint-mock";
|
||||
|
||||
function row(
|
||||
id: string,
|
||||
@@ -56,7 +57,16 @@ function buildPrismaMock(
|
||||
async (args: { where: { type: string } }) =>
|
||||
rowsByType[args.where.type] ?? [],
|
||||
);
|
||||
const tx = { measurement: { create, update, findFirst, updateMany } };
|
||||
const tx = {
|
||||
measurement: {
|
||||
create,
|
||||
update,
|
||||
findFirst,
|
||||
updateMany,
|
||||
findMany: candidateLookup(existingCanonicalId),
|
||||
createManyAndReturn: createManyVia(create),
|
||||
},
|
||||
};
|
||||
return {
|
||||
mock: {
|
||||
user: {
|
||||
|
||||
@@ -29,6 +29,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
|
||||
import { runDenseIntradayRetention } from "../dense-intraday-retention";
|
||||
import type { PerSampleRow } from "../drain-per-sample-cumulative";
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
import { candidateLookup, createManyVia } from "./hourly-mint-mock";
|
||||
|
||||
function spo2Row(id: string, value: number, iso: string): PerSampleRow {
|
||||
return {
|
||||
@@ -60,6 +61,8 @@ function buildPrismaMock(opts: { spo2Rows: PerSampleRow[] }) {
|
||||
update: txUpdate,
|
||||
findFirst: txFindFirst,
|
||||
updateMany: txUpdateMany,
|
||||
findMany: candidateLookup(),
|
||||
createManyAndReturn: createManyVia(txCreate),
|
||||
},
|
||||
};
|
||||
|
||||
@@ -254,6 +257,10 @@ describe("SpO2 fold — per-day failure boundary", () => {
|
||||
update: vi.fn(),
|
||||
findFirst: vi.fn().mockResolvedValue(null),
|
||||
updateMany: vi.fn().mockResolvedValue({ count: 2 }),
|
||||
findMany: candidateLookup(),
|
||||
createManyAndReturn: createManyVia(
|
||||
vi.fn().mockResolvedValue({ id: "minted" }),
|
||||
),
|
||||
},
|
||||
});
|
||||
},
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
/**
|
||||
* Transaction-client stubs for `adoptOrMintHourlyRows`, shared by the mocked
|
||||
* dense-tier fold tests.
|
||||
*
|
||||
* The hourly mint reads the day's candidate rows with one `findMany` and
|
||||
* writes its mints with one `createManyAndReturn`. The tests were written
|
||||
* against the per-slot loop and assert on individual `create` calls, so the
|
||||
* batch insert here is routed through the test's own `create` spy, one call
|
||||
* per minted row: the assertions keep reading what was minted, in slot order.
|
||||
*/
|
||||
import { vi } from "vitest";
|
||||
|
||||
type CreateSpy = (args: {
|
||||
data: { externalId: string };
|
||||
}) => Promise<{ id: string }>;
|
||||
|
||||
/** `createManyAndReturn` that forwards every row to `create`. */
|
||||
export function createManyVia(create: CreateSpy) {
|
||||
return vi.fn(async (args: { data: Array<{ externalId: string }> }) => {
|
||||
const out: Array<{ id: string; externalId: string }> = [];
|
||||
for (const data of args.data) {
|
||||
const row = await create({ data });
|
||||
out.push({ id: row.id, externalId: data.externalId });
|
||||
}
|
||||
return out;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* The day's candidate lookup. With `existingId`, one row already occupies the
|
||||
* first slot's anchor instant (the adopt-in-place case); otherwise nothing is
|
||||
* in the way.
|
||||
*/
|
||||
export function candidateLookup(existingId: string | null = null) {
|
||||
return vi.fn(
|
||||
async (args: {
|
||||
where: { OR: [unknown, { measuredAt: { in: Date[] } }] };
|
||||
}) =>
|
||||
existingId === null
|
||||
? []
|
||||
: [
|
||||
{
|
||||
id: existingId,
|
||||
externalId: null,
|
||||
measuredAt: args.where.OR[1].measuredAt.in[0]!,
|
||||
},
|
||||
],
|
||||
);
|
||||
}
|
||||
@@ -93,6 +93,12 @@ export interface MeanConsolidationBucket {
|
||||
|
||||
export interface MeanConsolidationSummary {
|
||||
dryRun: boolean;
|
||||
/**
|
||||
* `shouldStop` ended the walk before every day was reached. Nothing is
|
||||
* lost: folded days committed one by one, and the next run starts at the
|
||||
* first day this one did not reach.
|
||||
*/
|
||||
stoppedEarly: boolean;
|
||||
buckets: MeanConsolidationBucket[];
|
||||
totals: {
|
||||
usersScanned: number;
|
||||
@@ -123,6 +129,11 @@ export interface MeanConsolidationOptions {
|
||||
* `undefined` to drain everything.
|
||||
*/
|
||||
cutoffHours?: number;
|
||||
/**
|
||||
* Ends the pass cleanly before the next day when it returns `true`; the
|
||||
* summary then reports `stoppedEarly`. See `ConsolidationOptions`.
|
||||
*/
|
||||
shouldStop?: () => boolean;
|
||||
}
|
||||
|
||||
/** Arithmetic mean of a per-day bucket. Exposed for unit testing. */
|
||||
@@ -205,6 +216,7 @@ export async function consolidateDailyMean(
|
||||
|
||||
const summary: MeanConsolidationSummary = {
|
||||
dryRun: options.dryRun ?? false,
|
||||
stoppedEarly: false,
|
||||
buckets: [],
|
||||
totals: {
|
||||
usersScanned: 0,
|
||||
@@ -215,7 +227,7 @@ export async function consolidateDailyMean(
|
||||
},
|
||||
};
|
||||
|
||||
const { usersScanned, daysFailed } = await runConsolidation<MeasurementType>({
|
||||
const walk = await runConsolidation<MeasurementType>({
|
||||
prismaClient,
|
||||
options,
|
||||
types: HIGH_FREQUENCY_MEAN_TYPES,
|
||||
@@ -420,8 +432,9 @@ export async function consolidateDailyMean(
|
||||
},
|
||||
});
|
||||
|
||||
summary.totals.usersScanned = usersScanned;
|
||||
summary.totals.daysFailed = daysFailed;
|
||||
summary.totals.usersScanned = walk.usersScanned;
|
||||
summary.stoppedEarly = walk.stoppedEarly;
|
||||
summary.totals.daysFailed = walk.daysFailed;
|
||||
|
||||
log(
|
||||
`[mean-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFailed=${summary.totals.daysFailed}${options.dryRun ? " (dry-run)" : ""}`,
|
||||
|
||||
@@ -32,6 +32,8 @@ import type {
|
||||
Prisma,
|
||||
} from "@/generated/prisma/client";
|
||||
|
||||
import { reportJobProgress } from "@/lib/jobs/job-observer";
|
||||
|
||||
import {
|
||||
CONSOLIDATION_GRACE_CUTOFF_HOURS,
|
||||
canonicalDailyTimestamp,
|
||||
@@ -57,6 +59,18 @@ export interface ConsolidationOptions {
|
||||
dryRun?: boolean;
|
||||
/** Logger sink — defaults to `console.log`. */
|
||||
log?: (line: string) => void;
|
||||
/**
|
||||
* Asked before every day bucket. Returning `true` ends the pass cleanly
|
||||
* after the day in flight has committed, and the result reports
|
||||
* `stoppedEarly`. The queue handlers wire it to the job's time budget and
|
||||
* to pg-boss's abort signal, so a pass that cannot finish inside its expiry
|
||||
* stops on its own instead of being declared dead while it keeps writing.
|
||||
*
|
||||
* Stopping loses nothing: every day commits on its own, and a folded day's
|
||||
* source rows leave the scan predicate, so the next run starts at the first
|
||||
* day this one did not reach.
|
||||
*/
|
||||
shouldStop?: () => boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -229,7 +243,7 @@ export interface ConsolidationParams<TType extends MeasurementType> {
|
||||
dryRun: boolean;
|
||||
}) => void;
|
||||
/**
|
||||
* Fired once per (user, type) immediately after the rows are bucketed,
|
||||
* Fired once per (user, type) after the type's rows have been walked,
|
||||
* with the raw scanned-row count and per-day bucket count. Only invoked
|
||||
* for types that yielded at least one source row. Lets a single-type
|
||||
* drain (legacy steps) reproduce its scan-time log line. Optional.
|
||||
@@ -275,35 +289,38 @@ const DEFAULT_SCAN_SELECT: Prisma.MeasurementSelect = {
|
||||
};
|
||||
|
||||
/**
|
||||
* Page size for the keyset-paginated source-row scan. Heavy multi-year
|
||||
* tenants previously materialised the entire per-(user, type) history in a
|
||||
* single unbounded `findMany` result set — the dominant Prisma intermediate
|
||||
* cost and the root of the consolidation boot-storm on large accounts.
|
||||
* Scanning in bounded keyset pages caps the per-query result set while still
|
||||
* accumulating the same full row set.
|
||||
* Page size for the keyset-paginated source-row scan.
|
||||
*/
|
||||
const CONSOLIDATION_SCAN_PAGE_SIZE = 5000;
|
||||
|
||||
/**
|
||||
* Keyset-paginate the per-(user, type) source-row scan on `(measuredAt, id)`
|
||||
* ascending, accumulating every page into one array. Output is identical to a
|
||||
* single unbounded `findMany` — same `where`, same `select`, same ascending
|
||||
* order — but the Prisma client never holds more than one page of rows at a
|
||||
* time, so a multi-year tenant no longer materialises its whole history in one
|
||||
* result set at worker boot.
|
||||
* ascending, one page at a time.
|
||||
*
|
||||
* `(measuredAt, id)` is a stable keyset: `measuredAt` is the primary order and
|
||||
* `id` breaks ties, so each page resumes exactly after the last row with no
|
||||
* gaps or duplicates. The cursor clause is AND-combined with the caller's
|
||||
* `baseWhere` so the scope / soft-delete / grace filters always still apply.
|
||||
* Two properties matter, and both used to be missing.
|
||||
*
|
||||
* The pages are yielded, not accumulated. The previous helper gathered every
|
||||
* page into one array before a single day was folded, so the pass held a
|
||||
* type's whole history at once: on an account with 766 000 heart-rate samples
|
||||
* that is about 230 MB of live heap for the array alone, most of a 524 MB
|
||||
* limit, and nothing was written until the whole scan had finished.
|
||||
*
|
||||
* The cursor is index-usable. `measuredAt > c OR (measuredAt = c AND id > i)`
|
||||
* cannot be an index condition, so Postgres walked the index from the user's
|
||||
* first row on every page and filtered its way to the cursor, which makes the
|
||||
* whole scan quadratic (measured: 47 ms and 248 000 rows filtered for one
|
||||
* page halfway through). The redundant `measuredAt >= c` conjunct is the
|
||||
* index condition; the OR only breaks ties at the boundary instant.
|
||||
*
|
||||
* The caller's `baseWhere` is AND-combined, so the scope / soft-delete /
|
||||
* grace filters always still apply.
|
||||
*/
|
||||
export async function scanSourceRowsPaged(
|
||||
export async function* iterateSourcePages(
|
||||
prismaClient: PrismaClient,
|
||||
baseWhere: Prisma.MeasurementWhereInput,
|
||||
scanSelect: Prisma.MeasurementSelect,
|
||||
pageSize: number = CONSOLIDATION_SCAN_PAGE_SIZE,
|
||||
): Promise<PerSampleRow[]> {
|
||||
const accumulated: PerSampleRow[] = [];
|
||||
): AsyncGenerator<PerSampleRow[], void, void> {
|
||||
let cursor: { measuredAt: Date; id: string } | null = null;
|
||||
|
||||
for (;;) {
|
||||
@@ -311,6 +328,7 @@ export async function scanSourceRowsPaged(
|
||||
? {
|
||||
AND: [
|
||||
baseWhere,
|
||||
{ measuredAt: { gte: cursor.measuredAt } },
|
||||
{
|
||||
OR: [
|
||||
{ measuredAt: { gt: cursor.measuredAt } },
|
||||
@@ -328,18 +346,54 @@ export async function scanSourceRowsPaged(
|
||||
take: pageSize,
|
||||
})) as PerSampleRow[];
|
||||
|
||||
if (page.length === 0) break;
|
||||
|
||||
for (const row of page) accumulated.push(row);
|
||||
|
||||
// A short page is the last page — no further rows can satisfy the keyset.
|
||||
if (page.length < pageSize) break;
|
||||
if (page.length === 0) return;
|
||||
|
||||
const last = page[page.length - 1]!;
|
||||
const isLastPage = page.length < pageSize;
|
||||
yield page;
|
||||
|
||||
// A short page is the last page — no further rows can satisfy the keyset.
|
||||
if (isLastPage) return;
|
||||
cursor = { measuredAt: last.measuredAt, id: last.id };
|
||||
}
|
||||
}
|
||||
|
||||
return accumulated;
|
||||
/**
|
||||
* Group a stream of ascending pages into complete per-day buckets, in the
|
||||
* user's timezone, without holding more than one day at a time.
|
||||
*
|
||||
* The rows arrive in `measuredAt` order and a local calendar day is one
|
||||
* contiguous span of instants, so a day is complete the moment a row from a
|
||||
* later day shows up. Rows already in the daily-stats shape are skipped, the
|
||||
* same as `bucketRowsByDay`. `onRow` sees every scanned row, stats rows
|
||||
* included, for the scan-count log line.
|
||||
*/
|
||||
export async function* iterateDayBuckets(
|
||||
pages: AsyncIterable<readonly PerSampleRow[]>,
|
||||
tz: string,
|
||||
statsPrefix: string,
|
||||
onRow?: () => void,
|
||||
): AsyncGenerator<[dateKey: string, rows: PerSampleRow[]], void, void> {
|
||||
let currentKey: string | null = null;
|
||||
let current: PerSampleRow[] = [];
|
||||
for await (const page of pages) {
|
||||
for (const row of page) {
|
||||
onRow?.();
|
||||
if (row.externalId !== null && row.externalId.startsWith(statsPrefix)) {
|
||||
continue;
|
||||
}
|
||||
const key = dayKeyForUserTz(row.measuredAt, tz);
|
||||
if (key !== currentKey) {
|
||||
if (currentKey !== null && current.length > 0) {
|
||||
yield [currentKey, current];
|
||||
}
|
||||
currentKey = key;
|
||||
current = [];
|
||||
}
|
||||
current.push(row);
|
||||
}
|
||||
}
|
||||
if (currentKey !== null && current.length > 0) yield [currentKey, current];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -348,7 +402,8 @@ export async function scanSourceRowsPaged(
|
||||
* and delegates the per-day mint + delete to `writeDay`. Returns the
|
||||
* number of users scanned (so the caller can seed its summary totals)
|
||||
* plus the number of day buckets absorbed by `onBucketError` (0 when the
|
||||
* boundary is not supplied).
|
||||
* boundary is not supplied), and whether `options.shouldStop` ended the
|
||||
* walk before every day was reached.
|
||||
*
|
||||
* Idempotency, the grace-window cutoff, and the "skip already-collapsed
|
||||
* rows" predicate are all owned here; the divergent reducer / delete /
|
||||
@@ -356,16 +411,24 @@ export async function scanSourceRowsPaged(
|
||||
*/
|
||||
export async function runConsolidation<TType extends MeasurementType>(
|
||||
params: ConsolidationParams<TType>,
|
||||
): Promise<{ usersScanned: number; dryRun: boolean; daysFailed: number }> {
|
||||
): Promise<{
|
||||
usersScanned: number;
|
||||
dryRun: boolean;
|
||||
daysFailed: number;
|
||||
stoppedEarly: boolean;
|
||||
}> {
|
||||
const { prismaClient, options } = params;
|
||||
const dryRun = options.dryRun ?? false;
|
||||
const cutoffAt = resolveCutoffInstant(options.cutoffHours);
|
||||
const scanSelect = params.scanSelect ?? DEFAULT_SCAN_SELECT;
|
||||
const shouldStop = options.shouldStop ?? (() => false);
|
||||
let daysFailed = 0;
|
||||
let daysWalked = 0;
|
||||
let stoppedEarly = false;
|
||||
|
||||
const users = await loadConsolidationUsers(prismaClient, options.userId);
|
||||
|
||||
for (const user of users) {
|
||||
walk: for (const user of users) {
|
||||
const tz = resolveUserTimezone(user.timezone);
|
||||
params.onUserStart?.({ userId: user.id, tz, dryRun });
|
||||
|
||||
@@ -373,36 +436,45 @@ export async function runConsolidation<TType extends MeasurementType>(
|
||||
const hkIdentifier = params.hkIdentifierForType(type);
|
||||
if (!hkIdentifier) continue;
|
||||
|
||||
// Keyset-paginate the scan instead of loading the whole per-(user, type)
|
||||
// history in one result set; the accumulated array is byte-identical to a
|
||||
// single unbounded `findMany`, but a multi-year tenant no longer spikes
|
||||
// worker memory at boot. See `scanSourceRowsPaged`.
|
||||
const sourceRows = await scanSourceRowsPaged(
|
||||
prismaClient,
|
||||
params.buildScanWhere({
|
||||
userId: user.id,
|
||||
type,
|
||||
cutoffAt,
|
||||
statsPrefix: params.statsPrefix,
|
||||
}),
|
||||
scanSelect,
|
||||
// Streamed: each day is folded as soon as its last row has been read,
|
||||
// so the pass holds one page and one day, never the type's history.
|
||||
// See `iterateSourcePages` and `iterateDayBuckets`.
|
||||
let rowCount = 0;
|
||||
let dayCount = 0;
|
||||
const days = iterateDayBuckets(
|
||||
iterateSourcePages(
|
||||
prismaClient,
|
||||
params.buildScanWhere({
|
||||
userId: user.id,
|
||||
type,
|
||||
cutoffAt,
|
||||
statsPrefix: params.statsPrefix,
|
||||
}),
|
||||
scanSelect,
|
||||
),
|
||||
tz,
|
||||
params.statsPrefix,
|
||||
() => {
|
||||
rowCount += 1;
|
||||
},
|
||||
);
|
||||
|
||||
if (sourceRows.length === 0) continue;
|
||||
|
||||
const byDay = bucketRowsByDay(sourceRows, tz, params.statsPrefix);
|
||||
|
||||
params.onScan?.({
|
||||
userId: user.id,
|
||||
type,
|
||||
tz,
|
||||
rowCount: sourceRows.length,
|
||||
dayCount: byDay.size,
|
||||
dryRun,
|
||||
});
|
||||
|
||||
for (const [dateKey, dayRows] of byDay) {
|
||||
if (dayRows.length === 0) continue;
|
||||
for await (const [dateKey, dayRows] of days) {
|
||||
if (shouldStop()) {
|
||||
stoppedEarly = true;
|
||||
// Leaving the loop closes both generators, so no further page is
|
||||
// requested.
|
||||
break walk;
|
||||
}
|
||||
dayCount += 1;
|
||||
daysWalked += 1;
|
||||
// How far the pass has got, for the job's progress and expiry lines.
|
||||
reportJobProgress({
|
||||
consolidation_user: user.id,
|
||||
consolidation_type: type,
|
||||
consolidation_day: dateKey,
|
||||
consolidation_days_walked: daysWalked,
|
||||
});
|
||||
|
||||
try {
|
||||
const reducedValue = params.reduce(dayRows);
|
||||
@@ -464,10 +536,21 @@ export async function runConsolidation<TType extends MeasurementType>(
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (rowCount > 0) {
|
||||
params.onScan?.({
|
||||
userId: user.id,
|
||||
type,
|
||||
tz,
|
||||
rowCount,
|
||||
dayCount,
|
||||
dryRun,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
params.onUserComplete?.({ userId: user.id, tz, dryRun });
|
||||
}
|
||||
|
||||
return { usersScanned: users.length, dryRun, daysFailed };
|
||||
return { usersScanned: users.length, dryRun, daysFailed, stoppedEarly };
|
||||
}
|
||||
@@ -19,7 +19,7 @@
|
||||
* day's local window (the fold's soft-deleted inputs).
|
||||
*
|
||||
* Per qualifying day: compute per-local-hour means from the tombstoned raw
|
||||
* rows (same `meanBucketValue` reducer + `adoptOrMintHourlyRow`
|
||||
* rows (same `meanBucketValue` reducer + `adoptOrMintHourlyRows`
|
||||
* adopt-in-place slot logic as the live fold, so a re-run converges), then
|
||||
* tombstone the daily row IN THE SAME TRANSACTION as the hourly mint — at
|
||||
* no instant are the daily row and its hourly rows both live, so an
|
||||
@@ -73,7 +73,7 @@ import { meanBucketValue } from "./consolidate-daily-mean";
|
||||
import {
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
adoptOrMintHourlyRow,
|
||||
adoptOrMintHourlyRows,
|
||||
bucketRowsByLocalHour,
|
||||
hourlyStatsExternalId,
|
||||
} from "./dense-intraday-retention";
|
||||
@@ -318,11 +318,11 @@ async function rebuildDay(
|
||||
const rebuildOnce = async (): Promise<{ dailyRetired: boolean }> => {
|
||||
let dailyRetired = false;
|
||||
await prismaClient.$transaction(async (tx) => {
|
||||
const canonicalRowIds: string[] = [];
|
||||
for (const [hour, hourRows] of byHour) {
|
||||
const rowId = await adoptOrMintHourlyRow(tx, {
|
||||
userId: input.userId,
|
||||
type: input.type,
|
||||
const canonicalRowIds = await adoptOrMintHourlyRows(tx, {
|
||||
userId: input.userId,
|
||||
type: input.type,
|
||||
unit,
|
||||
slots: byHour.map(([hour, hourRows]) => ({
|
||||
externalId: hourlyStatsExternalId(
|
||||
input.hkIdentifier,
|
||||
input.dateKey,
|
||||
@@ -330,10 +330,8 @@ async function rebuildDay(
|
||||
),
|
||||
anchor: canonicalHourlyTimestamp(input.dateKey, hour, input.tz),
|
||||
value: meanBucketValue(hourRows),
|
||||
unit,
|
||||
});
|
||||
canonicalRowIds.push(rowId);
|
||||
}
|
||||
})),
|
||||
});
|
||||
|
||||
// Retire the daily row in the SAME transaction — at no instant are
|
||||
// the daily row and the hourly rows both live (an AVG-over-live-rows
|
||||
@@ -359,7 +357,7 @@ async function rebuildDay(
|
||||
if (!isUniqueConstraintViolation(err)) throw err;
|
||||
// A concurrent writer (the nightly fold racing this rebuild) won a slot
|
||||
// mid-transaction. Retry once: the deterministic lookup inside
|
||||
// `adoptOrMintHourlyRow` now resolves the winning row and adopts it.
|
||||
// `adoptOrMintHourlyRows` now resolves the winning row and adopts it.
|
||||
outcome = await rebuildOnce();
|
||||
}
|
||||
|
||||
|
||||
@@ -191,10 +191,20 @@ export function bucketRowsByLocalHour(
|
||||
return byHour;
|
||||
}
|
||||
|
||||
/** One hourly slot of a folded day, as `adoptOrMintHourlyRows` receives it. */
|
||||
export interface HourlySlot {
|
||||
/** Target hourly `stats:` externalId (index-A identity). */
|
||||
externalId: string;
|
||||
/** Local HH:30 anchor instant (index-B identity). */
|
||||
anchor: Date;
|
||||
value: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adopt-or-mint one hourly `stats:` row inside the caller's transaction.
|
||||
* Adopt-or-mint a day's hourly `stats:` rows inside the caller's transaction.
|
||||
* Returns the canonical row id for every slot, in slot order.
|
||||
*
|
||||
* TWO unique indexes can collide on the mint:
|
||||
* TWO unique indexes can collide on a mint:
|
||||
* A) `(userId, type, source, externalId)` — a row already carrying the
|
||||
* target hourly `stats:` externalId (a prior fold / rebuild pass).
|
||||
* B) `(userId, type, measuredAt, source, sleepStage)` (NULLS NOT
|
||||
@@ -202,98 +212,149 @@ export function bucketRowsByLocalHour(
|
||||
* anchor instant, which may carry a DIFFERENT externalId (a
|
||||
* per-sample row that happened at HH:30, or a sibling stats row).
|
||||
*
|
||||
* Determinism (VECTOR 1): resolve by index A FIRST — the row already
|
||||
* holding the target externalId IS the canonical hourly row by
|
||||
* construction. Only when no such row exists fall back to the index-B row
|
||||
* physically occupying the anchor. A stable `orderBy: id` pins the pick.
|
||||
* Adopting the wrong sibling and stamping the target externalId onto it
|
||||
* would otherwise collide with the real `stats:` row on index A.
|
||||
* Determinism: per slot, resolve by index A FIRST — the row already holding
|
||||
* the target externalId IS the canonical hourly row by construction. Only
|
||||
* when no such row exists fall back to the index-B row physically occupying
|
||||
* the anchor. The lowest id wins a tie. Adopting the wrong sibling and
|
||||
* stamping the target externalId onto it would otherwise collide with the
|
||||
* real `stats:` row on index A.
|
||||
*
|
||||
* Concurrency (VECTOR 2): the resolve→create is a check-then-act not
|
||||
* serialised by the unique index under READ COMMITTED. If a concurrent
|
||||
* writer wins the slot between the lookup and the INSERT, the `create`
|
||||
* throws P2002 and Postgres aborts the whole transaction. Callers
|
||||
* therefore catch P2002 OUTSIDE their transaction (a caught error inside
|
||||
* would leave the tx in the aborted `25P02` state) and retry the fold
|
||||
* once: on the retry the deterministic lookup finds the now-existing row
|
||||
* and ADOPTS it, so the create path never fires twice. Built
|
||||
* field-by-field (no spread) per the no-mass-assignment convention.
|
||||
* Why a batch. This used to run per slot — two lookups and a write for each
|
||||
* of up to 24 hours, about 75 statements for one day — and on a multi-year
|
||||
* heart-rate history that loop was most of the pass's 380 000 statements.
|
||||
* Now one lookup reads every row that could matter for the whole day, the
|
||||
* slots are resolved against that snapshot in order, and the mints go out as
|
||||
* one insert. The snapshot is kept up to date as each slot adopts or mints,
|
||||
* so a later slot sees exactly what the one-by-one loop would have read back
|
||||
* from the database: an adopted row that moves to its anchor frees the
|
||||
* instant it left and occupies the one it moved to, and a minted row occupies
|
||||
* its anchor and externalId. Updates run in slot order, as before, and no
|
||||
* mint can collide with them because a slot mints only when nothing in the
|
||||
* snapshot holds its externalId or its anchor.
|
||||
*
|
||||
* Concurrency: the resolve→write is a check-then-act not serialised by the
|
||||
* unique index under READ COMMITTED. If a concurrent writer wins a slot
|
||||
* between the lookup and the INSERT, the insert throws P2002 and Postgres
|
||||
* aborts the whole transaction. Callers therefore catch P2002 OUTSIDE their
|
||||
* transaction (a caught error inside would leave the tx in the aborted
|
||||
* `25P02` state) and retry the fold once: on the retry the lookup finds the
|
||||
* now-existing row and ADOPTS it. Built field-by-field (no spread) per the
|
||||
* no-mass-assignment convention.
|
||||
*/
|
||||
export async function adoptOrMintHourlyRow(
|
||||
export async function adoptOrMintHourlyRows(
|
||||
tx: Prisma.TransactionClient,
|
||||
input: {
|
||||
userId: string;
|
||||
type: MeasurementType;
|
||||
/** Target hourly `stats:` externalId (index-A identity). */
|
||||
externalId: string;
|
||||
/** Local HH:30 anchor instant (index-B identity). */
|
||||
anchor: Date;
|
||||
value: number;
|
||||
unit: string;
|
||||
slots: readonly HourlySlot[];
|
||||
},
|
||||
): Promise<string> {
|
||||
const eidRow = await tx.measurement.findFirst({
|
||||
): Promise<string[]> {
|
||||
if (input.slots.length === 0) return [];
|
||||
|
||||
// Every row that either index could put in the way, lowest id first — the
|
||||
// order the per-slot `findFirst … orderBy id` used to resolve ties by.
|
||||
// `sleepStage` is NULL for these continuous types, matched via the
|
||||
// NULLS-NOT-DISTINCT index.
|
||||
const existing = await tx.measurement.findMany({
|
||||
where: {
|
||||
userId: input.userId,
|
||||
type: input.type,
|
||||
source: "APPLE_HEALTH",
|
||||
externalId: input.externalId,
|
||||
OR: [
|
||||
{ externalId: { in: input.slots.map((slot) => slot.externalId) } },
|
||||
{
|
||||
measuredAt: { in: input.slots.map((slot) => slot.anchor) },
|
||||
sleepStage: null,
|
||||
},
|
||||
],
|
||||
},
|
||||
select: { id: true },
|
||||
orderBy: { id: "asc" },
|
||||
});
|
||||
// Index-B occupant (`sleepStage` is NULL for these continuous types,
|
||||
// matched via the NULLS-NOT-DISTINCT index). At most one row can exist.
|
||||
const slotRow = await tx.measurement.findFirst({
|
||||
where: {
|
||||
userId: input.userId,
|
||||
type: input.type,
|
||||
source: "APPLE_HEALTH",
|
||||
measuredAt: input.anchor,
|
||||
sleepStage: null,
|
||||
},
|
||||
select: { id: true },
|
||||
select: { id: true, externalId: true, measuredAt: true },
|
||||
orderBy: { id: "asc" },
|
||||
});
|
||||
|
||||
const adoptTarget = eidRow ?? slotRow;
|
||||
if (adoptTarget) {
|
||||
// Pin the adopted row to the anchor, but only when the anchor slot is
|
||||
// free or already this row — a DIFFERENT row occupying the slot would
|
||||
// otherwise collide on index B. In that rare case the row keeps its
|
||||
// existing instant; the `stats:` externalId is the identity,
|
||||
// measuredAt is secondary.
|
||||
const slotIsFreeForTarget =
|
||||
slotRow === null || slotRow.id === adoptTarget.id;
|
||||
// Adopt: refresh value, stamp the hourly externalId, optionally
|
||||
// re-anchor, and un-tombstone. This coexists with any pre-existing row
|
||||
// on the slot instead of colliding with it.
|
||||
await tx.measurement.update({
|
||||
where: { id: adoptTarget.id },
|
||||
data: {
|
||||
value: input.value,
|
||||
unit: input.unit,
|
||||
externalId: input.externalId,
|
||||
deletedAt: null,
|
||||
...(slotIsFreeForTarget ? { measuredAt: input.anchor } : {}),
|
||||
},
|
||||
});
|
||||
return adoptTarget.id;
|
||||
// The snapshot the slots resolve against. `id: null` marks a row this
|
||||
// call is about to mint.
|
||||
const state: Array<{
|
||||
id: string | null;
|
||||
externalId: string | null;
|
||||
at: number;
|
||||
}> = existing.map((row) => ({
|
||||
id: row.id,
|
||||
externalId: row.externalId,
|
||||
at: row.measuredAt.getTime(),
|
||||
}));
|
||||
|
||||
const resolved: Array<string | null> = [];
|
||||
const mints: HourlySlot[] = [];
|
||||
|
||||
for (const slot of input.slots) {
|
||||
const anchorAt = slot.anchor.getTime();
|
||||
const eidRow = state.find((row) => row.externalId === slot.externalId);
|
||||
const slotRow = state.find((row) => row.at === anchorAt);
|
||||
const adoptTarget = eidRow ?? slotRow;
|
||||
|
||||
if (adoptTarget && adoptTarget.id !== null) {
|
||||
// Pin the adopted row to the anchor, but only when the anchor slot is
|
||||
// free or already this row — a DIFFERENT row occupying the slot would
|
||||
// otherwise collide on index B. In that rare case the row keeps its
|
||||
// existing instant; the `stats:` externalId is the identity,
|
||||
// measuredAt is secondary.
|
||||
const slotIsFreeForTarget =
|
||||
slotRow === undefined || slotRow === adoptTarget;
|
||||
// Adopt: refresh value, stamp the hourly externalId, optionally
|
||||
// re-anchor, and un-tombstone. This coexists with any pre-existing row
|
||||
// on the slot instead of colliding with it.
|
||||
await tx.measurement.update({
|
||||
where: { id: adoptTarget.id },
|
||||
data: {
|
||||
value: slot.value,
|
||||
unit: input.unit,
|
||||
externalId: slot.externalId,
|
||||
deletedAt: null,
|
||||
...(slotIsFreeForTarget ? { measuredAt: slot.anchor } : {}),
|
||||
},
|
||||
});
|
||||
adoptTarget.externalId = slot.externalId;
|
||||
if (slotIsFreeForTarget) adoptTarget.at = anchorAt;
|
||||
resolved.push(adoptTarget.id);
|
||||
} else {
|
||||
// Nothing holds this slot's identity or instant — including rows this
|
||||
// call already decided to mint, which carry a distinct anchor and
|
||||
// externalId by construction (one slot per local hour).
|
||||
state.push({ id: null, externalId: slot.externalId, at: anchorAt });
|
||||
mints.push(slot);
|
||||
resolved.push(null);
|
||||
}
|
||||
}
|
||||
|
||||
const created = await tx.measurement.create({
|
||||
data: {
|
||||
userId: input.userId,
|
||||
type: input.type,
|
||||
value: input.value,
|
||||
unit: input.unit,
|
||||
source: "APPLE_HEALTH",
|
||||
measuredAt: input.anchor,
|
||||
externalId: input.externalId,
|
||||
},
|
||||
select: { id: true },
|
||||
});
|
||||
return created.id;
|
||||
if (mints.length > 0) {
|
||||
const created = await tx.measurement.createManyAndReturn({
|
||||
data: mints.map((slot) => ({
|
||||
userId: input.userId,
|
||||
type: input.type,
|
||||
value: slot.value,
|
||||
unit: input.unit,
|
||||
source: "APPLE_HEALTH" as const,
|
||||
measuredAt: slot.anchor,
|
||||
externalId: slot.externalId,
|
||||
})),
|
||||
select: { id: true, externalId: true },
|
||||
});
|
||||
const idByExternalId = new Map(
|
||||
created.map((row) => [row.externalId, row.id] as const),
|
||||
);
|
||||
input.slots.forEach((slot, index) => {
|
||||
if (resolved[index] !== null) return;
|
||||
const id = idByExternalId.get(slot.externalId);
|
||||
if (id === undefined) {
|
||||
throw new Error(`hourly mint returned no row for ${slot.externalId}`);
|
||||
}
|
||||
resolved[index] = id;
|
||||
});
|
||||
}
|
||||
|
||||
return resolved as string[];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -347,6 +408,12 @@ export function deriveDailyRestingFromPulse(
|
||||
|
||||
export interface DenseIntradayRetentionSummary {
|
||||
dryRun: boolean;
|
||||
/**
|
||||
* `shouldStop` ended the walk before every day was reached. Nothing is
|
||||
* lost: folded days committed one by one, and the next run starts at the
|
||||
* first day this one did not reach.
|
||||
*/
|
||||
stoppedEarly: boolean;
|
||||
totals: {
|
||||
usersScanned: number;
|
||||
daysConsolidated: number;
|
||||
@@ -385,6 +452,11 @@ export interface DenseIntradayRetentionOptions {
|
||||
* full-collapse, never by the scheduled pass.
|
||||
*/
|
||||
retentionDays?: number;
|
||||
/**
|
||||
* Ends the pass cleanly before the next day when it returns `true`; the
|
||||
* summary then reports `stoppedEarly`. See `ConsolidationOptions`.
|
||||
*/
|
||||
shouldStop?: () => boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -412,6 +484,7 @@ export async function runDenseIntradayRetention(
|
||||
|
||||
const summary: DenseIntradayRetentionSummary = {
|
||||
dryRun: options.dryRun ?? false,
|
||||
stoppedEarly: false,
|
||||
totals: {
|
||||
usersScanned: 0,
|
||||
daysConsolidated: 0,
|
||||
@@ -462,7 +535,7 @@ export async function runDenseIntradayRetention(
|
||||
// for the dry-run hourly fan-out estimate (writeDay never runs on dry-run).
|
||||
let currentTz = "Europe/Berlin";
|
||||
|
||||
const { usersScanned } = await runConsolidation<MeasurementType>({
|
||||
const walk = await runConsolidation<MeasurementType>({
|
||||
prismaClient,
|
||||
options: { ...options, cutoffHours },
|
||||
types: DENSE_INTRADAY_RETENTION_TYPES,
|
||||
@@ -543,18 +616,16 @@ export async function runDenseIntradayRetention(
|
||||
let removed = 0;
|
||||
let retiredDaily = false;
|
||||
await pc.$transaction(async (tx) => {
|
||||
const canonicalRowIds: string[] = [];
|
||||
for (const [hour, hourRows] of byHour) {
|
||||
const rowId = await adoptOrMintHourlyRow(tx, {
|
||||
userId,
|
||||
type,
|
||||
const canonicalRowIds = await adoptOrMintHourlyRows(tx, {
|
||||
userId,
|
||||
type,
|
||||
unit,
|
||||
slots: byHour.map(([hour, hourRows]) => ({
|
||||
externalId: hourlyStatsExternalId(hkIdentifier, dateKey, hour),
|
||||
anchor: canonicalHourlyTimestamp(dateKey, hour, tz),
|
||||
value: meanBucketValue(hourRows),
|
||||
unit,
|
||||
});
|
||||
canonicalRowIds.push(rowId);
|
||||
}
|
||||
})),
|
||||
});
|
||||
|
||||
// Retire a live pre-hourly DAILY `stats:` row for this day (the
|
||||
// late-sync path: raw samples arriving for a day folded to the
|
||||
@@ -712,7 +783,8 @@ export async function runDenseIntradayRetention(
|
||||
},
|
||||
});
|
||||
|
||||
summary.totals.usersScanned = usersScanned;
|
||||
summary.totals.usersScanned = walk.usersScanned;
|
||||
summary.stoppedEarly = walk.stoppedEarly;
|
||||
|
||||
log(
|
||||
`[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`,
|
||||
|
||||
@@ -124,6 +124,12 @@ export interface DrainBucket {
|
||||
export interface DrainSummary {
|
||||
/** Did the run actually write to the DB (false for `dryRun`). */
|
||||
dryRun: boolean;
|
||||
/**
|
||||
* `shouldStop` ended the walk before every day was reached. Nothing is
|
||||
* lost: folded days committed one by one, and the next run starts at the
|
||||
* first day this one did not reach.
|
||||
*/
|
||||
stoppedEarly: boolean;
|
||||
/** Per-user-day buckets the drain rewrote (or would rewrite). */
|
||||
buckets: DrainBucket[];
|
||||
/** Aggregate counts across the run. */
|
||||
@@ -167,6 +173,11 @@ export interface DrainOptions {
|
||||
* the operator points at) for explicit one-shot use.
|
||||
*/
|
||||
cutoffHours?: number;
|
||||
/**
|
||||
* Ends the pass cleanly before the next day when it returns `true`; the
|
||||
* summary then reports `stoppedEarly`. See `ConsolidationOptions`.
|
||||
*/
|
||||
shouldStop?: () => boolean;
|
||||
}
|
||||
|
||||
export interface BucketedRows {
|
||||
@@ -210,6 +221,7 @@ export async function drainPerSampleCumulative(
|
||||
|
||||
const summary: DrainSummary = {
|
||||
dryRun: options.dryRun ?? false,
|
||||
stoppedEarly: false,
|
||||
buckets: [],
|
||||
totals: {
|
||||
usersScanned: 0,
|
||||
@@ -229,7 +241,7 @@ export async function drainPerSampleCumulative(
|
||||
let beforePerSampleDeleted = 0;
|
||||
let beforeDailyUpserted = 0;
|
||||
|
||||
const { usersScanned } = await runConsolidation<MeasurementType>({
|
||||
const walk = await runConsolidation<MeasurementType>({
|
||||
prismaClient,
|
||||
options,
|
||||
types: CUMULATIVE_HK_TYPES,
|
||||
@@ -521,7 +533,8 @@ export async function drainPerSampleCumulative(
|
||||
},
|
||||
});
|
||||
|
||||
summary.totals.usersScanned = usersScanned;
|
||||
summary.totals.usersScanned = walk.usersScanned;
|
||||
summary.stoppedEarly = walk.stoppedEarly;
|
||||
|
||||
// v1.34.0 — bound the profile sidecar. One indexed range delete per run,
|
||||
// across every account, on the same nightly tick that writes the profiles.
|
||||
|
||||
Reference in new issue
Block a user