fix(measurements): keep unchanged updates on their page and vacuum by threshold

`updated_at` sits in the sync-feed index, so every update that moved it
wrote a new row version on another page and an entry in every index.
Provider re-syncs rewrite many unchanged rows, which grew the table to
several times its live size and sent each unchanged row down the sync
feed again. Migration 0412 adds a row trigger that keeps `updated_at` and
`sync_version` when an update changes nothing else, so such an update is
heap-only and the feed stays silent; it sets `fillfactor = 85` and a 5 %
autovacuum threshold. The maintenance queue gains a nightly check that
vacuums when dead rows pass a tenth of the live ones and warns when the
heap is more than three times what its live rows need, which only a
`VACUUM FULL` in a window gives back. The admin card keeps showing the
operator's runs only.
This commit is contained in:
Marc-André Bombeck committed 2026-10-20 05:06:25 +02:00
1 parent 483c84b42e
commit 4a8bf1963b
11 files changed
+472 -38

No files matched your search

@@ -0,0 +1,91 @@
/**
* The nightly measurements check: what it decides from the table's
* statistics. It vacuums only when dead rows pile up and never rebuilds; the
* operator's run keeps the steps it names.
*/
import { describe, expect, it, vi } from "vitest";
vi.mock("@/lib/db", () => ({
prisma: {},
sessionOptionsDisabled: () => false,
}));
import {
DEAD_ROW_VACUUM_FLOOR,
resolveSteps,
tableHealth,
vacuumDue,
type TableHealth,
} from "@/lib/jobs/measurement-maintenance";
const health = (over: Partial<TableHealth> = {}): TableHealth => ({
liveRows: 1_000_000,
deadRows: 0,
heapBytes: 0,
bloatFactor: null,
...over,
});
describe("vacuumDue", () => {
it("vacuums once dead rows pass a tenth of the live ones", () => {
expect(vacuumDue(health({ deadRows: 100_001 }))).toBe(true);
expect(vacuumDue(health({ deadRows: 99_999 }))).toBe(false);
});
it("leaves a small table alone below the floor", () => {
expect(
vacuumDue(health({ liveRows: 100, deadRows: DEAD_ROW_VACUUM_FLOOR - 1 })),
).toBe(false);
});
});
describe("resolveSteps", () => {
it("gives the nightly tick a vacuum when one is due and never a rebuild", () => {
expect(resolveSteps({}, health({ deadRows: 500_000 }))).toEqual({
vacuum: true,
reindex: false,
});
expect(resolveSteps(undefined, health())).toEqual({
vacuum: false,
reindex: false,
});
});
it("keeps the operator's steps whatever the statistics say", () => {
expect(resolveSteps({ vacuum: false, reindex: true }, health())).toEqual({
vacuum: false,
reindex: true,
});
expect(resolveSteps({ vacuum: true, reindex: false }, health())).toEqual({
vacuum: true,
reindex: false,
});
});
});
describe("tableHealth", () => {
it("measures the heap against what the live rows need", async () => {
const db = {
query: vi.fn().mockResolvedValue({
// 1,000 rows of 72 bytes plus 28 of overhead at fill factor 0.85
// need about 117,647 bytes; the heap holds 1,176,470.
rows: [{ live: "1000", dead: "50", heap: "1176470", width: "72" }],
}),
};
expect(await tableHealth(db as never)).toEqual({
liveRows: 1000,
deadRows: 50,
heapBytes: 1176470,
bloatFactor: 10,
});
});
it("has no factor before the table has statistics", async () => {
const db = {
query: vi.fn().mockResolvedValue({
rows: [{ live: "0", dead: "0", heap: "8192", width: null }],
}),
};
expect((await tableHealth(db as never)).bloatFactor).toBeNull();
});
});
+3
View File
@@ -57,6 +57,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
"assessments_warmed",
"auto_resolved",
"backed",
// The nightly measurements check: heap size over what the live rows need.
"bloat_factor",
"buckets_collapsed",
"buckets_recomputed",
"budget_blocked",
@@ -78,6 +80,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
"days_skipped",
"days_skipped_no_tombstones",
"days_stored",
"dead_rows",
// Tombstone purge: accounts left for the next run because a restore of
// the account was running.
"deferred_accounts",
@@ -1,8 +1,8 @@
/**
* What the admin card shows about the measurement table maintenance: the
* newest run and where it stands, whether the compaction-tombstone purge is
* still at work (the run refuses to start while it is), and the table's size
* right now.
* newest run the operator started and where it stands, whether the
* compaction-tombstone purge is still at work (the run refuses to start while
* it is), and the table's size right now.
*
* Read straight from `pgboss.job`, the same source `job-failures.ts` reads,
* and fail-soft the same way: a web-only deployment has no `pgboss` schema,
@@ -90,6 +90,9 @@ export async function readMeasurementMaintenanceStatus(): Promise<MeasurementMai
output->'did'->>'outcome' AS outcome
FROM pgboss.job
WHERE name = ${MEASUREMENT_MAINTENANCE_QUEUE}
-- The operator's runs name their steps; the nightly check's tick
-- names none and reports through its wide event instead.
AND (data ? 'vacuum' OR data ? 'reindex')
ORDER BY created_on DESC
LIMIT 1
`,
+144 -24
View File
@@ -1,12 +1,22 @@
/**
* Operator-triggered maintenance of the `measurements` table (v1.42).
* Maintenance of the `measurements` table (v1.42).
*
* `VACUUM (ANALYZE)` and a `REINDEX INDEX CONCURRENTLY` per index, largest
* first, after the compaction-tombstone purge has shrunk the table. Neither
* statement can run inside a transaction, so this is a job rather than a
* Prisma migration. Started only from `POST /api/admin/maintenance/measurements`
* (cookie-only admin), never on a schedule: the operator picks a quiet window.
* Runbook: `docs/ops/measurement-maintenance.md`.
* Prisma migration. Runbook: `docs/ops/measurement-maintenance.md`.
*
* Two ways in:
* - The operator, from `POST /api/admin/maintenance/measurements`
* (cookie-only admin), with the steps named; the rebuild only ever runs
* this way, in a window the operator picks.
* - The nightly check (`MEASUREMENT_MAINTENANCE_CRON`, a tick without
* steps). It reads the table's dead and live rows and vacuums only when
* the dead ones pass `vacuumDue`. A plain VACUUM takes no lock that stops
* reads or writes. It also measures how far the heap has grown past its
* live rows (`TableHealth.bloatFactor`); only a rewrite (`VACUUM FULL`,
* which locks the table) gives that space back, so past
* `TABLE_BLOAT_WARN_FACTOR` the run warns instead of acting.
*
* Why the indexes: after the purge the heap is small, but a B-tree does not
* give back the pages its deleted entries held. VACUUM marks the heap space
@@ -59,12 +69,51 @@ export const MEASUREMENT_MAINTENANCE_SEND_OPTIONS = {
singletonKey: MEASUREMENT_MAINTENANCE_SINGLETON_KEY,
} as const;
/** Which steps the operator asked for. */
/** Which steps the operator asked for. The nightly tick names none. */
export interface MeasurementMaintenancePayload {
vacuum?: boolean;
reindex?: boolean;
}
/** The steps a run takes, once resolved. */
export interface MaintenanceSteps {
vacuum: boolean;
reindex: boolean;
}
/** 03:40 Europe/Berlin, inside the nightly maintenance band. */
export const MEASUREMENT_MAINTENANCE_CRON = "40 3 * * *";
/** Dead rows, as a share of live ones, past which the nightly check vacuums. */
export const DEAD_ROW_VACUUM_SHARE = 0.1;
/** Fewer dead rows than this are never worth a vacuum of their own. */
export const DEAD_ROW_VACUUM_FLOOR = 10_000;
/**
* Heap size against what the live rows need, past which the run warns that
* only a rewrite gives the space back.
*/
export const TABLE_BLOAT_WARN_FACTOR = 3;
/** Bytes a heap row costs beyond its data: header, item pointer, alignment. */
const ROW_OVERHEAD_BYTES = 28;
/** The fill factor migration 0412 gives the table. */
const FILL_FACTOR = 0.85;
/** What the statistics say about the table's heap. */
export interface TableHealth {
liveRows: number;
deadRows: number;
heapBytes: number;
/**
* Heap bytes over what the live rows need at the planner's average row
* width and the table's fill factor; null before the table has statistics.
*/
bloatFactor: number | null;
}
/**
* Session settings of the maintenance connection.
*
@@ -113,7 +162,12 @@ export interface MaintenanceStep {
}
export interface MeasurementMaintenanceSummary {
outcome: "completed" | "refused_purge_running" | "already_running" | "failed";
outcome:
| "completed"
| "nothing_due"
| "refused_purge_running"
| "already_running"
| "failed";
steps: MaintenanceStep[];
tableBytesBefore: number;
tableBytesAfter: number;
@@ -143,6 +197,62 @@ async function relationSizes(
};
}
/** The table's heap as the statistics collector and the planner see it. */
export async function tableHealth(
db: MaintenanceConnection,
): Promise<TableHealth> {
const { rows } = await db.query<{
live: string;
dead: string;
heap: string;
width: string | null;
}>(
`SELECT s.n_live_tup::text AS live,
s.n_dead_tup::text AS dead,
pg_relation_size('measurements'::regclass)::text AS heap,
(SELECT sum(avg_width)::text FROM pg_stats
WHERE schemaname = s.schemaname AND tablename = 'measurements') AS width
FROM pg_stat_user_tables s
WHERE s.relid = 'measurements'::regclass`,
);
const liveRows = Number(rows[0]?.live ?? 0);
const heapBytes = Number(rows[0]?.heap ?? 0);
const width = Number(rows[0]?.width ?? 0);
const needed = (liveRows * (width + ROW_OVERHEAD_BYTES)) / FILL_FACTOR;
return {
liveRows,
deadRows: Number(rows[0]?.dead ?? 0),
heapBytes,
bloatFactor:
liveRows > 0 && width > 0
? Math.round((heapBytes / needed) * 10) / 10
: null,
};
}
/** Whether the dead rows are worth a vacuum of their own tonight. */
export function vacuumDue(health: TableHealth): boolean {
return (
health.deadRows >= DEAD_ROW_VACUUM_FLOOR &&
health.deadRows > health.liveRows * DEAD_ROW_VACUUM_SHARE
);
}
/**
* The steps a delivery takes: the operator's as asked, the nightly tick's a
* vacuum when one is due and never a rebuild.
*/
export function resolveSteps(
payload: MeasurementMaintenancePayload | undefined,
health: TableHealth,
): MaintenanceSteps {
const named =
typeof payload?.vacuum === "boolean" ||
typeof payload?.reindex === "boolean";
if (!named) return { vacuum: vacuumDue(health), reindex: false };
return { vacuum: payload?.vacuum ?? true, reindex: payload?.reindex ?? true };
}
async function measurementIndexes(
db: MaintenanceConnection,
): Promise<IndexRow[]> {
@@ -223,7 +333,7 @@ async function dropInvalidCopies(
*/
export async function runMeasurementMaintenance(
db: MaintenanceConnection,
payload: MeasurementMaintenancePayload,
payload: MaintenanceSteps,
options: { shouldStop?: () => boolean } = {},
): Promise<MeasurementMaintenanceSummary> {
const before = await relationSizes(db);
@@ -240,6 +350,8 @@ export async function runMeasurementMaintenance(
indexBytesAfter: after.indexes,
});
if (!payload.vacuum && !payload.reindex) return summary("nothing_due");
// The queue table exists wherever a worker has started; a database without
// one has no purge to wait for.
const { rows: queueTable } = await db.query<{ present: boolean }>(
@@ -327,10 +439,6 @@ export async function handleMeasurementMaintenance(
jobs: Job<MeasurementMaintenancePayload>[],
): Promise<JobOutcome> {
const job = jobs[0];
const payload: MeasurementMaintenancePayload = {
vacuum: job?.data?.vacuum ?? true,
reindex: job?.data?.reindex ?? true,
};
return withBackgroundEvent("job.measurement_maintenance", async (evt) => {
const startedAt = Date.now();
const guarded = await withJobLock(
@@ -350,7 +458,10 @@ export async function handleMeasurementMaintenance(
await client.query(statement);
}
}
return await runMeasurementMaintenance(client, payload);
const health = await tableHealth(client);
const steps = resolveSteps(job?.data, health);
const summary = await runMeasurementMaintenance(client, steps);
return { summary, health, steps };
} finally {
await client
.end()
@@ -363,20 +474,26 @@ export async function handleMeasurementMaintenance(
}
},
);
const result: MeasurementMaintenanceSummary = guarded.ran
? guarded.result
: {
outcome: "already_running",
steps: [],
tableBytesBefore: 0,
tableBytesAfter: 0,
indexBytesBefore: 0,
indexBytesAfter: 0,
};
if (!guarded.ran) {
evt.addMeta("maintenance_outcome", "already_running");
return jobDone({ outcome: "already_running" });
}
const { summary: result, health, steps } = guarded.result;
evt.addMeta("maintenance_outcome", result.outcome);
evt.addMeta("maintenance_vacuum", payload.vacuum);
evt.addMeta("maintenance_reindex", payload.reindex);
evt.addMeta("maintenance_vacuum", steps.vacuum);
evt.addMeta("maintenance_reindex", steps.reindex);
evt.addMeta("maintenance_live_rows", health.liveRows);
evt.addMeta("maintenance_dead_rows", health.deadRows);
evt.addMeta("maintenance_heap_bytes", health.heapBytes);
if (health.bloatFactor !== null) {
evt.addMeta("maintenance_bloat_factor", health.bloatFactor);
if (health.bloatFactor > TABLE_BLOAT_WARN_FACTOR) {
evt.addWarning(
`measurements heap is ${health.bloatFactor}x what its live rows need; only VACUUM FULL in a maintenance window gives the space back`,
);
}
}
evt.addMeta("maintenance_table_bytes_before", result.tableBytesBefore);
evt.addMeta("maintenance_table_bytes_after", result.tableBytesAfter);
evt.addMeta("maintenance_index_bytes_before", result.indexBytesBefore);
@@ -412,6 +529,9 @@ export async function handleMeasurementMaintenance(
processed: result.steps.filter((s) => s.ok).length,
failed: result.steps.filter((s) => !s.ok).length,
duration_ms: Date.now() - startedAt,
dead_rows: health.deadRows,
// Zero before the table has statistics to measure against.
bloat_factor: health.bloatFactor ?? 0,
};
if (result.outcome === "failed") {
const firstError = result.steps.find((s) => !s.ok);
+2 -2
View File
@@ -391,7 +391,7 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
expiryVia: ["MEASUREMENT_MAINTENANCE_SEND_OPTIONS"],
stop: {
oneShot:
"VACUUM and REINDEX CONCURRENTLY cannot be resumed mid-statement; the send sets retryLimit 0 and the operator re-triggers.",
"VACUUM and REINDEX CONCURRENTLY cannot be resumed mid-statement; the send and the nightly tick set retryLimit 0, and the next night or the operator runs it again.",
},
exclusive: {
marker: "retryLimit: 0",
@@ -400,7 +400,7 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
fn: "MEASUREMENT_MAINTENANCE_SEND_OPTIONS",
},
},
why: "Rebuilds every index of the largest table, one at a time.",
why: "The operator's run rebuilds every index of the largest table, one at a time; the nightly check is one VACUUM at most.",
},
"pulse-day-mean": {
runtime: "long",
+15 -5
View File
@@ -80,7 +80,9 @@ import {
type MeasurementFoldRepairPayload,
} from "@/lib/jobs/measurement-fold-repair";
import {
MEASUREMENT_MAINTENANCE_CRON,
MEASUREMENT_MAINTENANCE_QUEUE,
MEASUREMENT_MAINTENANCE_SEND_OPTIONS,
handleMeasurementMaintenance,
type MeasurementMaintenancePayload,
} from "@/lib/jobs/measurement-maintenance";
@@ -507,9 +509,9 @@ const allQueues = [
// their window; the purge above waits for it. Without this entry the boot
// enqueue silently no-ops and the purge never runs.
MEASUREMENT_FOLD_REPAIR_QUEUE,
// v1.42 — operator-triggered VACUUM / REINDEX of `measurements`. No
// cron: the admin route enqueues it. Without this entry the trigger
// silently no-ops.
// v1.42 — VACUUM / REINDEX of `measurements`: the admin route enqueues the
// operator's run, and a nightly check vacuums when dead rows pile up.
// Without this entry both silently no-op.
MEASUREMENT_MAINTENANCE_QUEUE,
// v1.43 — the pulse day-mean dry run and rebuild, admin-triggered only.
PULSE_DAY_MEAN_QUEUE,
@@ -634,6 +636,14 @@ const allQueues = [
// `retryLimit: 0` in its own options for the other reason: it pages each
// failed account itself, and a retry would page the same accounts again.
const schedules: ScheduleEntry[] = [
// v1.43.7 — nightly 03:40 check of `measurements`: a plain VACUUM when the
// dead rows pass the threshold, never a rebuild. Same singleton key as the
// operator's run, so the two collapse.
[
MEASUREMENT_MAINTENANCE_QUEUE,
MEASUREMENT_MAINTENANCE_CRON,
{ ...MEASUREMENT_MAINTENANCE_SEND_OPTIONS },
],
// v1.25 (W-ENV) — daily 02:10 Europe/Berlin discovery tick (empty payload)
// that fans out one per-user environment fetch per opted-in account.
[ENVIRONMENT_FETCH_QUEUE, ENVIRONMENT_FETCH_CRON],
@@ -849,11 +859,11 @@ const queuePolicies: QueuePolicyTable = {
reason:
"Fixed singleton key, admin-triggered. Two concurrent corpus rotations must never overlap; the route already reports a suppressed send back as alreadyQueued.",
},
// v1.42 — operator-triggered VACUUM / REINDEX of `measurements`.
// v1.42 — VACUUM / REINDEX of `measurements`, operator or nightly.
[MEASUREMENT_MAINTENANCE_QUEUE]: {
policy: "exclusive",
reason:
"Fixed singleton key, admin-triggered. Two maintenance passes would rebuild the same indexes side by side; the route reports a suppressed send back as enqueued: false.",
"Fixed singleton key on the admin send and the nightly tick alike. Two maintenance passes would vacuum or rebuild side by side; the route reports a suppressed send back as enqueued: false.",
},
// v1.43 — the pulse day-mean dry run and rebuild.
[PULSE_DAY_MEAN_QUEUE]: {
@@ -312,6 +312,7 @@ export const SCHEDULED_AND_SENT_QUEUES: ReadonlySet<string> = new Set([
"geolite2-fetch",
"insight-pregenerate",
"intake-slot-dedup",
"measurement-maintenance",
"med-notes-encryption-backfill",
"nightscout-sync",
"note-encryption-backfill",