mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-11 13:52:20 +00:00
fix(migrations): retire the measurements prefix index without waiting at boot
0416 ran DROP INDEX CONCURRENTLY at boot, which waits without bound for every transaction on the table; a pg_dump or a backfill outlasted the start period and the next boot stopped at P3009. The migration now tries a plain drop with a two-second lock wait and tolerates the timeout, and the nightly measurement maintenance drops what is left concurrently under a bounded wait, retrying the next night. A fresh or shadow database takes the lock at once, so the history still ends where the schema does. The guard refuses a concurrent drop in a migration and ties every retired index to a tolerant one.
This commit is contained in:
1 parent
a275a5bb65
commit
2b81b4f23e
5 files changed
+388
-19
No files matched your search
@@ -23,6 +23,8 @@ import { readdirSync, readFileSync, existsSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { RETIRED_MEASUREMENT_INDEXES } from "@/lib/jobs/measurement-maintenance";
|
||||
|
||||
import { RETRY_MARKER } from "../../scripts/migrate-deploy.mjs";
|
||||
|
||||
const MIGRATIONS_DIR = join(process.cwd(), "prisma", "migrations");
|
||||
@@ -114,13 +116,14 @@ describe("concurrent index statements", () => {
|
||||
expect(found, found.join("\n")).toEqual([]);
|
||||
});
|
||||
|
||||
it("sees the concurrent migrations there are", () => {
|
||||
const concurrent = migrations()
|
||||
.filter(({ sql }) =>
|
||||
/\bCONCURRENTLY\b/i.test(topLevelStatements(sql).join(";")),
|
||||
)
|
||||
.map(({ name }) => name);
|
||||
expect(concurrent).toContain("0416_drop_measurements_type_prefix_index");
|
||||
it("reads the statements there are", () => {
|
||||
// No migration holds a concurrent statement since 0416 stopped dropping
|
||||
// concurrently; the rule stands for the next one. A reader that stopped
|
||||
// splitting statements would see none and pass, so it must see many.
|
||||
const statements = migrations().flatMap(({ sql }) =>
|
||||
topLevelStatements(sql),
|
||||
);
|
||||
expect(statements.length).toBeGreaterThan(1000);
|
||||
});
|
||||
|
||||
it("refuses a concurrent statement beside another one", () => {
|
||||
@@ -139,6 +142,58 @@ describe("concurrent index statements", () => {
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* `DROP INDEX CONCURRENTLY` blocks no reader or writer, but it waits without
|
||||
* bound for every transaction on the table, and in a migration that wait is
|
||||
* the boot's: a pg_dump or a backfill outlasts the container's start period,
|
||||
* the half-applied migration is left on record, and the next boot stops at
|
||||
* P3009. An index is retired instead by a bounded `DROP INDEX` that tolerates
|
||||
* the timeout, finished by the nightly measurement maintenance (0416).
|
||||
*/
|
||||
function concurrentDropViolations(name: string, sql: string): string[] {
|
||||
return topLevelStatements(sql).some((statement) =>
|
||||
/^DROP INDEX CONCURRENTLY\b/i.test(statement),
|
||||
)
|
||||
? [`${name}: DROP INDEX CONCURRENTLY waits on the table at boot`]
|
||||
: [];
|
||||
}
|
||||
|
||||
describe("retired indexes", () => {
|
||||
it("are never dropped concurrently in the boot path", () => {
|
||||
const found = migrations().flatMap(({ name, sql }) =>
|
||||
concurrentDropViolations(name, sql),
|
||||
);
|
||||
expect(found, found.join("\n")).toEqual([]);
|
||||
});
|
||||
|
||||
it("are each dropped by a migration that bounds its wait and tolerates the timeout", () => {
|
||||
expect(RETIRED_MEASUREMENT_INDEXES.length).toBeGreaterThan(0);
|
||||
for (const index of RETIRED_MEASUREMENT_INDEXES) {
|
||||
const owner = migrations().find(({ sql }) =>
|
||||
new RegExp(`DROP INDEX IF EXISTS "${index}"`).test(sql),
|
||||
);
|
||||
expect(owner, index).toBeDefined();
|
||||
expect(owner?.sql).toMatch(/set_config\(\s*'lock_timeout'/);
|
||||
expect(owner?.sql).toMatch(/EXCEPTION WHEN lock_not_available THEN/);
|
||||
}
|
||||
});
|
||||
|
||||
it("counter-check: refuses a concurrent drop, not one in a comment", () => {
|
||||
expect(
|
||||
concurrentDropViolations(
|
||||
"a",
|
||||
'DROP INDEX CONCURRENTLY IF EXISTS "measurements_x_idx";',
|
||||
),
|
||||
).toHaveLength(1);
|
||||
expect(
|
||||
concurrentDropViolations(
|
||||
"b",
|
||||
'-- DROP INDEX CONCURRENTLY "x";\nDROP INDEX IF EXISTS "x";',
|
||||
),
|
||||
).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("migration lock waits", () => {
|
||||
it("bounds lock waits where they take effect and marks only repeatable files", () => {
|
||||
const found = migrations().flatMap(({ name, sql }) =>
|
||||
@@ -154,6 +209,7 @@ describe("migration lock waits", () => {
|
||||
expect(marked).toContain("0412_measurement_hot_updates");
|
||||
expect(marked).toContain("0414_drop_medication_compliance_rollups");
|
||||
expect(marked).toContain("0415_foreign_key_and_prefix_indexes");
|
||||
expect(marked).toContain("0416_drop_measurements_type_prefix_index");
|
||||
});
|
||||
|
||||
it("catches each shape it exists to refuse", () => {
|
||||
|
||||
@@ -46,6 +46,15 @@
|
||||
* before the first rebuild.
|
||||
* - Every step's size before and after, and its duration, are on the run's
|
||||
* wide event; the totals are job facts.
|
||||
*
|
||||
* Retired indexes. An index the schema no longer declares is dropped by the
|
||||
* migration that retires it only when the lock is there at once (a `DO`
|
||||
* block with a two-second `lock_timeout` that swallows the timeout), so the
|
||||
* boot never waits on a long reader of the table. Where the migration had to
|
||||
* leave it, every run of this job, the nightly tick included, drops it with
|
||||
* `DROP INDEX CONCURRENTLY IF EXISTS` under a bounded lock wait and tries
|
||||
* again the next night if the wait runs out. Runbook:
|
||||
* `docs/ops/measurement-maintenance.md`, "Retired indexes".
|
||||
*/
|
||||
import { Client } from "pg";
|
||||
import type { Job } from "pg-boss";
|
||||
@@ -153,6 +162,22 @@ export function maintenanceSessionStatements(): string[] {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Indexes on `measurements` the schema no longer declares, which a run drops
|
||||
* when the migration that retired them could not. Each one's migration names
|
||||
* it; `IF EXISTS` makes the drop a no-op once it is gone.
|
||||
*/
|
||||
export const RETIRED_MEASUREMENT_INDEXES: readonly string[] = [
|
||||
// 0416: a prefix of the unique key `(user_id, type, measured_at, source,
|
||||
// sleep_stage)`, which plans every query it served at the same cost.
|
||||
"measurements_user_id_type_measured_at_idx",
|
||||
];
|
||||
|
||||
/** How long a run waits for the locks that drop a retired index. */
|
||||
export const RETIRED_INDEX_LOCK_TIMEOUT = "60s";
|
||||
|
||||
const LOCK_TIMEOUT_VALUE = /^\d{1,6}(ms|s)$/;
|
||||
|
||||
/** The only shape an index name may have before it is spliced into SQL. */
|
||||
const INDEX_NAME = /^[a-z0-9_]{1,63}$/;
|
||||
|
||||
@@ -162,7 +187,7 @@ const CONCURRENT_COPY = /_ccnew\d*$|_ccold\d*$/;
|
||||
const PURGE_QUEUE = "compaction-tombstone-purge";
|
||||
|
||||
export interface MaintenanceStep {
|
||||
step: "vacuum" | "reindex" | "drop_invalid";
|
||||
step: "vacuum" | "reindex" | "drop_invalid" | "drop_retired";
|
||||
target: string;
|
||||
bytesBefore: number;
|
||||
bytesAfter: number;
|
||||
@@ -336,6 +361,62 @@ async function dropInvalidCopies(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop every retired index still present. `DROP INDEX CONCURRENTLY` takes no
|
||||
* lock that stops reads or writes, but it waits for every transaction that
|
||||
* may use the index, so the wait is bounded for the statement and the
|
||||
* session's own setting is put back after it. A timed-out drop may leave the
|
||||
* index marked invalid; the next run's `IF EXISTS` drop finishes it.
|
||||
*/
|
||||
export async function dropRetiredIndexes(
|
||||
db: MaintenanceConnection,
|
||||
steps: MaintenanceStep[],
|
||||
lockTimeout: string = RETIRED_INDEX_LOCK_TIMEOUT,
|
||||
): Promise<void> {
|
||||
if (!LOCK_TIMEOUT_VALUE.test(lockTimeout)) {
|
||||
throw new Error(`refusing lock timeout of unexpected shape`);
|
||||
}
|
||||
for (const name of RETIRED_MEASUREMENT_INDEXES) {
|
||||
const bytesBefore = await indexBytes(db, name);
|
||||
const { rows } = await db.query<{ present: boolean }>(
|
||||
`SELECT to_regclass($1) IS NOT NULL AS present`,
|
||||
[name],
|
||||
);
|
||||
if (!rows[0]?.present) continue;
|
||||
const { rows: setting } = await db.query<{ value: string }>(
|
||||
`SELECT current_setting('lock_timeout') AS value`,
|
||||
);
|
||||
const startedAt = Date.now();
|
||||
try {
|
||||
await db.query(`SET lock_timeout = '${lockTimeout}'`);
|
||||
await db.query(`DROP INDEX CONCURRENTLY IF EXISTS ${quoted(name)}`);
|
||||
steps.push({
|
||||
step: "drop_retired",
|
||||
target: name,
|
||||
bytesBefore,
|
||||
bytesAfter: 0,
|
||||
durationMs: Date.now() - startedAt,
|
||||
ok: true,
|
||||
});
|
||||
} catch (err) {
|
||||
logCaught("measurements.maintenance.drop_retired_failed", err);
|
||||
steps.push({
|
||||
step: "drop_retired",
|
||||
target: name,
|
||||
bytesBefore,
|
||||
bytesAfter: await indexBytes(db, name),
|
||||
durationMs: Date.now() - startedAt,
|
||||
ok: false,
|
||||
error: errorText(err),
|
||||
});
|
||||
} finally {
|
||||
await db.query(`SELECT set_config('lock_timeout', $1, false)`, [
|
||||
setting[0]?.value ?? "0",
|
||||
]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The pass itself, on a connection the caller opened with
|
||||
* {@link MAINTENANCE_SESSION_OPTIONS}. Exported for the integration test.
|
||||
@@ -343,7 +424,7 @@ async function dropInvalidCopies(
|
||||
export async function runMeasurementMaintenance(
|
||||
db: MaintenanceConnection,
|
||||
payload: MaintenanceSteps,
|
||||
options: { shouldStop?: () => boolean } = {},
|
||||
options: { shouldStop?: () => boolean; retiredLockTimeout?: string } = {},
|
||||
): Promise<MeasurementMaintenanceSummary> {
|
||||
const before = await relationSizes(db);
|
||||
const steps: MaintenanceStep[] = [];
|
||||
@@ -359,6 +440,10 @@ export async function runMeasurementMaintenance(
|
||||
indexBytesAfter: after.indexes,
|
||||
});
|
||||
|
||||
// Every run, the nightly tick included: a retired index costs every write
|
||||
// until it is gone, and the drop blocks no reader or writer.
|
||||
await dropRetiredIndexes(db, steps, options.retiredLockTimeout);
|
||||
|
||||
if (!payload.vacuum && !payload.reindex) return summary("nothing_due");
|
||||
|
||||
// The queue table exists wherever a worker has started; a database without
|
||||
@@ -406,7 +491,10 @@ export async function runMeasurementMaintenance(
|
||||
if (payload.reindex) {
|
||||
await dropInvalidCopies(db, steps);
|
||||
const targets = (await measurementIndexes(db)).filter(
|
||||
(index) => index.valid && !CONCURRENT_COPY.test(index.name),
|
||||
(index) =>
|
||||
index.valid &&
|
||||
!CONCURRENT_COPY.test(index.name) &&
|
||||
!RETIRED_MEASUREMENT_INDEXES.includes(index.name),
|
||||
);
|
||||
for (const index of targets) {
|
||||
if (options.shouldStop?.()) break;
|
||||
@@ -509,6 +597,15 @@ export async function handleMeasurementMaintenance(
|
||||
);
|
||||
}
|
||||
}
|
||||
const retiredLeft = result.steps.filter(
|
||||
(s) => s.step === "drop_retired" && !s.ok,
|
||||
);
|
||||
if (retiredLeft.length > 0) {
|
||||
evt.addMeta("maintenance_retired_index_pending", true);
|
||||
evt.addWarning(
|
||||
`retired index ${retiredLeft.map((s) => s.target).join(", ")} could not be dropped within ${RETIRED_INDEX_LOCK_TIMEOUT} (a long transaction on measurements); the next nightly run tries again`,
|
||||
);
|
||||
}
|
||||
evt.addMeta("maintenance_table_bytes_before", result.tableBytesBefore);
|
||||
evt.addMeta("maintenance_table_bytes_after", result.tableBytesAfter);
|
||||
evt.addMeta("maintenance_index_bytes_before", result.indexBytesBefore);
|
||||
|
||||
Reference in new issue
Block a user