feat(rollups): v1.4.35.1 — auto-converge rollups on worker boot

Removes the operator step that was implicit in the v1.4.35 foundation.
Self-hosters no longer need to run the backfill script after upgrade.

New `rollup-full-backfill` queue + handler folds one user per job at
serial concurrency. `enqueueBootTimeRollupBackfill` runs once at
`startReminderWorker` boot, discovers accounts with measurements but
zero rollup coverage, and enqueues one full-fold job per uncovered
user with a singleton key so fast restarts don't pile up duplicates.

Best-effort: discovery errors surface through the helper's return
value and are logged via workerLog; the worker boot never fails
because the backfill missed.

Tests: 4280 → 4285 unit + 228 → 230 integration. tsc + lint clean.
This commit is contained in:
Marc-André Bombeck committed 2026-05-17 11:28:21 +02:00
1 parent ba9dd3fd2f
commit 01ae3b542a
6 files changed
+392 -2

No files matched your search

+53
View File
@@ -53,9 +53,13 @@ import {
type AppleHealthImportPayload,
} from "@/lib/jobs/apple-health-import-worker";
import {
ROLLUP_FULL_BACKFILL_QUEUE,
ROLLUP_FULL_BACKFILL_CONCURRENCY,
ROLLUP_RECOMPUTE_QUEUE,
ROLLUP_RECOMPUTE_CONCURRENCY,
enqueueBootTimeRollupBackfill,
recomputeUserRollups,
type RollupFullBackfillPayload,
type RollupRecomputePayload,
} from "@/lib/measurements/rollups";
import { expireStaleInUseItems } from "@/lib/medications/inventory/service";
@@ -1466,6 +1470,7 @@ export async function startReminderWorker() {
MEDICATION_INVENTORY_EXPIRE_QUEUE,
APPLE_HEALTH_IMPORT_QUEUE,
ROLLUP_RECOMPUTE_QUEUE,
ROLLUP_FULL_BACKFILL_QUEUE,
];
for (const q of allQueues) {
@@ -1666,6 +1671,54 @@ export async function startReminderWorker() {
},
);
// v1.4.35.1 — boot-time full-fold worker. The boot enqueue helper
// below sends one job per uncovered user; this handler runs the
// full `recomputeUserRollups` against the default 5-year window
// across every granularity. Serial concurrency so the populator
// never crowds the dashboard request pool.
await boss.work<RollupFullBackfillPayload>(
ROLLUP_FULL_BACKFILL_QUEUE,
{ localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY },
async (jobs) => {
for (const job of jobs) {
const { userId } = job.data;
const { rowsUpserted, durationMs } = await recomputeUserRollups(userId);
workerLog(
"info",
`[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
);
}
},
);
// v1.4.35.1 — fire-and-forget boot-time backfill discovery. Finds
// every user with measurements but no rollup coverage and enqueues
// a full-fold per account. Idempotent across reboots: the discovery
// query only matches accounts with zero rollup rows, so once a
// fold completes the user drops off the list. Errors are returned
// through the helper's result value — the worker boot never fails
// because of a backfill miss.
try {
const { enqueued, skipped, error } = await enqueueBootTimeRollupBackfill();
if (error) {
workerLog(
"error",
`[rollup-full-backfill] boot discovery failed: ${error}`,
);
} else if (enqueued > 0 || skipped > 0) {
workerLog(
"info",
`[rollup-full-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
);
}
} catch (err) {
workerLog(
"error",
"[rollup-full-backfill] boot discovery threw an unexpected error",
err,
);
}
return boss;
}
@@ -72,8 +72,10 @@ void mocks.findMany;
void queryRaw;
import {
ROLLUP_FULL_BACKFILL_QUEUE,
ROLLUP_RECOMPUTE_QUEUE,
collapseToTypeDayKeys,
enqueueBootTimeRollupBackfill,
enqueueRollupRecompute,
ensureUserRollupsFresh,
recomputeBucketsForMeasurement,
@@ -280,3 +282,70 @@ describe("ensureUserRollupsFresh", () => {
expect(result.recomputed).toBe(false);
});
});
describe("enqueueBootTimeRollupBackfill", () => {
it("is a silent no-op when no boss is attached", async () => {
getGlobalBossMock.mockReturnValue(null);
const result = await enqueueBootTimeRollupBackfill();
expect(result).toEqual({ enqueued: 0, skipped: 0, error: null });
expect(queryRaw).not.toHaveBeenCalled();
expect(bossSend).not.toHaveBeenCalled();
});
it("enqueues one full-fold job per user who has measurements but no rollups", async () => {
getGlobalBossMock.mockReturnValue({ send: bossSend });
queryRaw.mockResolvedValueOnce([
{ id: "user-a" },
{ id: "user-b" },
{ id: "user-c" },
]);
bossSend
.mockResolvedValueOnce("job-a")
.mockResolvedValueOnce("job-b")
.mockResolvedValueOnce("job-c");
const result = await enqueueBootTimeRollupBackfill();
expect(result).toEqual({ enqueued: 3, skipped: 0, error: null });
expect(bossSend).toHaveBeenCalledTimes(3);
// The queue name is the boot-fold queue, not the per-bucket queue.
for (const call of bossSend.mock.calls) {
expect(call[0]).toBe(ROLLUP_FULL_BACKFILL_QUEUE);
}
// Singleton key per user — coalesces across rapid reboots.
expect(bossSend.mock.calls[0][2].singletonKey).toBe("boot-backfill|user-a");
expect(bossSend.mock.calls[1][2].singletonKey).toBe("boot-backfill|user-b");
expect(bossSend.mock.calls[2][2].singletonKey).toBe("boot-backfill|user-c");
});
it("counts a `boss.send` returning null as 'skipped' (singleton coalesce)", async () => {
getGlobalBossMock.mockReturnValue({ send: bossSend });
queryRaw.mockResolvedValueOnce([{ id: "user-a" }, { id: "user-b" }]);
bossSend.mockResolvedValueOnce(null).mockResolvedValueOnce("job-b");
const result = await enqueueBootTimeRollupBackfill();
expect(result).toEqual({ enqueued: 1, skipped: 1, error: null });
});
it("returns the error message when the discovery query throws", async () => {
getGlobalBossMock.mockReturnValue({ send: bossSend });
queryRaw.mockRejectedValueOnce(new Error("pool exhausted"));
const result = await enqueueBootTimeRollupBackfill();
expect(result.enqueued).toBe(0);
expect(result.error).toBe("pool exhausted");
expect(bossSend).not.toHaveBeenCalled();
});
it("returns { enqueued: 0 } when no users need backfill", async () => {
getGlobalBossMock.mockReturnValue({ send: bossSend });
queryRaw.mockResolvedValueOnce([]);
const result = await enqueueBootTimeRollupBackfill();
expect(result).toEqual({ enqueued: 0, skipped: 0, error: null });
expect(bossSend).not.toHaveBeenCalled();
});
});
+101
View File
@@ -68,6 +68,28 @@ export interface RollupRecomputePayload {
enqueuedAt: string;
}
/**
* v1.4.35.1 — boot-time full-fold queue. The write-path hooks keep
* the DAY bucket current per measurement, and `ensureUserRollupsFresh`
* warm-on-read covers the trailing 90-day DAY window on cold mount,
* but neither path folds the deep WEEK / MONTH / YEAR history for an
* account that has measurements pre-dating the v1.4.35 deploy. This
* queue exists so the worker boot can enqueue one full
* `recomputeUserRollups(userId)` per uncovered user — no operator
* action needed on self-hosted instances.
*/
export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill";
/** Boot-fold worker concurrency. Serial so we don't crowd the pool. */
export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1;
/** Payload the boot-time backfill helper enqueues. */
export interface RollupFullBackfillPayload {
userId: string;
/** Wall-clock kick-off for debugging only. */
enqueuedAt: string;
}
/** Postgres `date_trunc` unit per granularity. */
const DATE_TRUNC_UNIT: Record<RollupGranularity, string> = {
DAY: "day",
@@ -600,6 +622,85 @@ export async function ensureUserRollupsFresh(
}
}
/**
* v1.4.35.1 — boot-time enqueue helper. Finds users with measurements
* but zero rollup coverage and enqueues one full-fold job per account
* onto `ROLLUP_FULL_BACKFILL_QUEUE`. Designed to be called once at
* worker boot so every self-hosted instance auto-converges on the
* persistent rollup tier without operator intervention.
*
* Idempotent across reboots:
* - the discovery query only matches accounts with no rollup rows at
* all, so once the fold completes the user drops off the list and
* subsequent boots skip them
* - pg-boss `singletonKey` coalesces duplicate sends within the
* queue, so a fast restart while a backfill is queued doesn't
* double up
*
* Best-effort: errors are swallowed and reported through the return
* value so the worker boot never fails because of a backfill miss.
*/
export async function enqueueBootTimeRollupBackfill(): Promise<{
enqueued: number;
skipped: number;
error: string | null;
}> {
const boss = getGlobalBoss();
if (!boss) {
return { enqueued: 0, skipped: 0, error: null };
}
try {
// Users with at least one measurement but no rollup row yet. The
// `LIMIT 1` inside each `EXISTS` keeps the planner from materialising
// the full child relation just to answer existence — important on
// power-user accounts with 100k+ measurements.
const users = await prisma.$queryRaw<Array<{ id: string }>>`
SELECT u."id"
FROM users u
WHERE EXISTS (
SELECT 1 FROM measurements m WHERE m."user_id" = u."id" LIMIT 1
)
AND NOT EXISTS (
SELECT 1 FROM measurement_rollups r WHERE r."user_id" = u."id" LIMIT 1
)
`;
if (users.length === 0) {
return { enqueued: 0, skipped: 0, error: null };
}
let enqueued = 0;
let skipped = 0;
for (const { id } of users) {
const payload: RollupFullBackfillPayload = {
userId: id,
enqueuedAt: new Date().toISOString(),
};
const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, {
retryLimit: 3,
retryDelay: 60,
retryBackoff: true,
// Coalesce: if a backfill for this user is already queued and
// we restart, pg-boss returns null instead of duplicating.
singletonKey: `boot-backfill|${id}`,
});
if (jobId) {
enqueued += 1;
} else {
skipped += 1;
}
}
return { enqueued, skipped, error: null };
} catch (err) {
return {
enqueued: 0,
skipped: 0,
error: err instanceof Error ? err.message : String(err),
};
}
}
/**
* Collapse a list of `(type, day)` keys to the set of distinct
* `(type, dayStart)` pairs. Used by the batch-ingest hooks so a 500-