fix(migrations): rebuild an index a failed concurrent build left invalid

A CREATE INDEX CONCURRENTLY that fails half-way leaves an INVALID index,
and the migration re-run after migrate resolve --rolled-back finds it
through IF NOT EXISTS and does nothing: recorded applied, with an index
the planner never uses. 0477 has this shape.

The boot now drops an INVALID index that a not-yet-applied migration
builds concurrently before it runs migrate deploy, with a bounded lock
wait, so the re-run builds it again. An INVALID index whose migration is
recorded applied is named on stderr with the rebuild that mends it. The
deploy runbook says both.

docs/ops/deploy.md

@@ -109,6 +109,20 @@ that for you, and each boot retries). Anything else is a session you can
 judge: a stuck client may be ended with `SELECT pg_terminate_backend(<pid>)`.
 You never need to run `migrate resolve` by hand for this case.
 
+### A concurrent index build that failed
+
+A migration that builds an index `CONCURRENTLY` (0477 is one) and fails
+half-way, on a constraint violation, a cancel or a killed container, leaves an
+INVALID index behind and its attempt on record as failed. Once you have mended
+the cause and run `prisma migrate resolve --rolled-back <migration>`, the next
+boot drops that INVALID index before it migrates, logs
+`dropped the invalid index …`, and the re-run builds it again. Without that
+step the re-run would find the index by name and skip it.
+
+An INVALID index whose migration is already recorded applied is not touched at
+boot. The boot names it on stderr with `WARNING — index … is INVALID`; rebuild
+it with `REINDEX INDEX CONCURRENTLY "<index>";` when the database is quiet.
+
 ### v1.45.0 notes
 
 A single-container install (`HEALTHLOG_PROCESS_TYPE=all`, the shipped

scripts/__tests__/migrate-deploy.test.ts

@@ -22,7 +22,9 @@ import { join } from "node:path";
 import { afterEach, describe, expect, it } from "vitest";
 
 import {
+  concurrentIndexesIn,
   decideLockRetry,
+  invalidIndexPlan,
   lockTimeoutInReport,
   prismaEnv,
   retryDelays,
@@ -386,3 +388,48 @@ process.exit(0);
     expect(result.stderr).toContain("not marked safe to re-run");
   });
 });
+
+describe("the invalid indexes a failed concurrent build leaves", () => {
+  it("reads the index each concurrent migration builds", () => {
+    const sql = readFileSync(
+      join(
+        process.cwd(),
+        "prisma",
+        "migrations",
+        "0477_measurements_ingest_token_index",
+        "migration.sql",
+      ),
+      "utf8",
+    );
+    expect(concurrentIndexesIn(sql)).toEqual([
+      "measurements_ingest_token_id_id_idx",
+    ]);
+    expect(
+      concurrentIndexesIn(
+        '-- CREATE INDEX CONCURRENTLY IF NOT EXISTS "commented"\nCREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "u" ON "t" ("c");',
+      ),
+    ).toEqual(["u"]);
+    expect(concurrentIndexesIn('CREATE INDEX "plain" ON "t" ("c");')).toEqual(
+      [],
+    );
+  });
+
+  it("drops only an index whose building migration has not been applied", () => {
+    const builtBy = new Map([
+      ["pending_idx", "0477_pending"],
+      ["applied_idx", "0300_applied"],
+    ]);
+    const plan = invalidIndexPlan({
+      invalid: ["pending_idx", "applied_idx", "stray_idx"],
+      builtBy,
+      applied: new Set(["0300_applied"]),
+    });
+    expect(plan.drop).toEqual([
+      { index: "pending_idx", migration: "0477_pending" },
+    ]);
+    expect(plan.report).toEqual([
+      { index: "applied_idx", migration: "0300_applied" },
+      { index: "stray_idx", migration: null },
+    ]);
+  });
+});

scripts/migrate-deploy.mjs

@@ -46,6 +46,15 @@
  * Prisma passes to the server. `DATABASE_SESSION_OPTIONS_DISABLED` (a pooler
  * that refuses `options`) turns it off, as it does for the app.
  *
+ * One more thing is done before every `migrate deploy`: a concurrent index
+ * build that failed leaves an INVALID index behind, and the re-run of its
+ * migration (`CREATE INDEX CONCURRENTLY IF NOT EXISTS`) would find that index
+ * and do nothing, recorded applied with an index the planner never uses. So
+ * an INVALID index that a not-yet-applied migration builds concurrently is
+ * dropped first, and the run builds it again. An INVALID index whose
+ * migration is recorded applied is not touched at boot; it is named on
+ * stderr with the rebuild that mends it.
+ *
  * Environment:
  *   HEALTHLOG_PRISMA_CLI      Prisma's CLI entry (default: the image's copy)
  *   HEALTHLOG_MIGRATIONS_DIR  where the migration files are (default:
@@ -55,7 +64,7 @@
  * Arguments are passed through to every Prisma call (`--schema …`).
  */
 import { spawn } from "node:child_process";
-import { readFileSync } from "node:fs";
+import { readdirSync, readFileSync } from "node:fs";
 import { join } from "node:path";
 import { pathToFileURL } from "node:url";
 
@@ -133,6 +142,138 @@ export function decideLockRetry({ output, failedRows, readMigration }) {
   return { retry: true, migration, code };
 }
 
+/** `CREATE [UNIQUE] INDEX CONCURRENTLY IF NOT EXISTS "<name>"`, comments aside. */
+const CONCURRENT_INDEX =
+  /^\s*CREATE\s+(?:UNIQUE\s+)?INDEX\s+CONCURRENTLY\s+IF\s+NOT\s+EXISTS\s+"([^"]+)"/gim;
+
+/** The indexes a migration file builds concurrently. */
+export function concurrentIndexesIn(sql) {
+  if (typeof sql !== "string") return [];
+  const code = sql.replace(/--[^\n]*/g, "");
+  return [...code.matchAll(CONCURRENT_INDEX)].map((m) => m[1]);
+}
+
+/**
+ * What to do about the INVALID indexes in the schema.
+ *
+ * @param {object} input
+ * @param {string[]} input.invalid      names of the INVALID indexes
+ * @param {Map<string, string>} input.builtBy
+ *   index name → the migration that builds it concurrently
+ * @param {Set<string>} input.applied   migrations recorded applied
+ * @returns {{ drop: Array<{ index: string, migration: string }>, report: Array<{ index: string, migration: string | null }> }}
+ */
+export function invalidIndexPlan({ invalid, builtBy, applied }) {
+  const drop = [];
+  const report = [];
+  for (const index of invalid) {
+    const migration = builtBy.get(index) ?? null;
+    if (migration !== null && !applied.has(migration)) {
+      drop.push({ index, migration });
+    } else {
+      report.push({ index, migration });
+    }
+  }
+  return { drop, report };
+}
+
+/** Every concurrently built index in the migrations directory, by name. */
+function concurrentIndexMap(dir) {
+  const builtBy = new Map();
+  let names = [];
+  try {
+    names = readdirSync(dir, { withFileTypes: true })
+      .filter((entry) => entry.isDirectory())
+      .map((entry) => entry.name);
+  } catch {
+    return builtBy;
+  }
+  const read = migrationReader(dir);
+  for (const name of names) {
+    for (const index of concurrentIndexesIn(read(name))) {
+      builtBy.set(index, name);
+    }
+  }
+  return builtBy;
+}
+
+/**
+ * Drop the INVALID indexes a migration still to run will build, and name the
+ * others. Best effort: a database that cannot be read here is left to
+ * `migrate deploy`, which reports its own errors.
+ */
+async function clearInvalidConcurrentIndexes(connectionString, migrationsDir) {
+  if (!connectionString) return;
+  const builtBy = concurrentIndexMap(migrationsDir);
+  let client;
+  try {
+    const { default: pg } = await import("pg");
+    const url = new URL(connectionString);
+    const schema = url.searchParams.get("schema") || "public";
+    url.searchParams.delete("schema");
+    client = new pg.Client({
+      connectionString: url.toString(),
+      connectionTimeoutMillis: 5_000,
+    });
+    await client.connect();
+    const invalid = await client.query(
+      `SELECT c.relname AS name FROM pg_index i
+         JOIN pg_class c ON c.oid = i.indexrelid
+         JOIN pg_namespace n ON n.oid = c.relnamespace
+        WHERE NOT i.indisvalid AND n.nspname = $1`,
+      [schema],
+    );
+    if (invalid.rows.length === 0) return;
+    const table = `${client.escapeIdentifier(schema)}._prisma_migrations`;
+    const exists = await client.query("SELECT to_regclass($1) AS t", [
+      `${schema}._prisma_migrations`,
+    ]);
+    const applied = new Set(
+      exists.rows[0]?.t
+        ? (
+            await client.query(
+              `SELECT migration_name FROM ${table}
+                WHERE finished_at IS NOT NULL AND rolled_back_at IS NULL`,
+            )
+          ).rows.map((row) => row.migration_name)
+        : [],
+    );
+    const plan = invalidIndexPlan({
+      invalid: invalid.rows.map((row) => row.name),
+      builtBy,
+      applied,
+    });
+    // Dropping an index takes its table's lock for a moment; never wait on
+    // it for long at boot.
+    await client.query("SET lock_timeout = '5s'");
+    for (const { index, migration } of plan.drop) {
+      await client.query(
+        `DROP INDEX IF EXISTS ${client.escapeIdentifier(schema)}.${client.escapeIdentifier(index)}`,
+      );
+      console.log(
+        `HealthLog: dropped the invalid index ${index} a failed concurrent ` +
+          `build left behind; ${migration} builds it again.`,
+      );
+    }
+    for (const { index, migration } of plan.report) {
+      console.error(
+        `HealthLog: WARNING — index ${index} is INVALID` +
+          (migration ? ` (built by ${migration}, recorded applied)` : "") +
+          `; the planner does not use it. Rebuild it with ` +
+          `REINDEX INDEX CONCURRENTLY "${index}".`,
+      );
+    }
+  } catch (error) {
+    console.error(
+      `HealthLog: could not check for invalid indexes before migrating: ${
+        error instanceof Error ? error.message : String(error)
+      }`,
+    );
+  } finally {
+    await client?.end().catch(() => {});
+  }
+}
+
 /** The backoff schedule, from the environment or the default. */
 export function retryDelays(raw) {
   if (!raw) return DEFAULT_RETRY_DELAYS_S;
@@ -258,6 +399,10 @@ async function main({
 } = {}) {
   const readMigration = migrationReader(migrationsDir);
   for (let attempt = 0; ; attempt++) {
+    await clearInvalidConcurrentIndexes(
+      process.env.DATABASE_URL,
+      migrationsDir,
+    );
     const deploy = await runPrisma(cli, ["migrate", "deploy", ...extraArgs]);
     if (deploy.code === 0) return 0;
 

tests/integration/migrate-deploy-invalid-index.test.ts

@@ -0,0 +1,173 @@
+/**
+ * A concurrent index build that failed leaves an INVALID index behind, and
+ * `CREATE INDEX CONCURRENTLY IF NOT EXISTS` finds that index and does
+ * nothing: the migration re-run after `migrate resolve --rolled-back` is
+ * recorded applied while the index it exists for is never used, and nothing
+ * says so (0477 has this shape).
+ *
+ * The boot (`scripts/migrate-deploy.mjs`) drops such an index first, when the
+ * migration that builds it is not applied, so the re-run builds it again. The
+ * probe runs the real script and the real CLI against a schema of its own.
+ */
+import { spawn } from "node:child_process";
+import { cpSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { Client } from "pg";
+import { afterAll, beforeEach, describe, expect, it } from "vitest";
+
+const SCHEMA = "invalid_index_probe";
+const NAME = "0002_probe_value_index";
+const INDEX = "probe_value_key";
+const ROOT = process.cwd();
+
+const dirs: string[] = [];
+afterAll(async () => {
+  for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true });
+  const db = await connect();
+  await db.query(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`);
+  await db.end();
+});
+
+function baseUrl(): string {
+  const url = process.env.DATABASE_URL;
+  if (!url) throw new Error("DATABASE_URL is not set for the integration run");
+  return url;
+}
+
+function probeUrl(): string {
+  const url = new URL(baseUrl());
+  url.searchParams.set("schema", SCHEMA);
+  return url.toString();
+}
+
+async function connect(): Promise<Client> {
+  const url = new URL(baseUrl());
+  url.searchParams.delete("schema");
+  const client = new Client({ connectionString: url.toString() });
+  await client.connect();
+  return client;
+}
+
+function write(dir: string, name: string, sql: string) {
+  mkdirSync(join(dir, "migrations", name), { recursive: true });
+  writeFileSync(join(dir, "migrations", name, "migration.sql"), sql);
+}
+
+function prisma(dir: string, args: string[]) {
+  return run(join(ROOT, "node_modules", "prisma", "build", "index.js"), [
+    ...args,
+    "--schema",
+    join(dir, "schema.prisma"),
+  ]);
+}
+
+function migrateDeploy(dir: string) {
+  return run(join(ROOT, "scripts", "migrate-deploy.mjs"), [
+    "--schema",
+    join(dir, "schema.prisma"),
+  ]);
+}
+
+function run(
+  script: string,
+  args: string[],
+): Promise<{ code: number; stdout: string; stderr: string }> {
+  return new Promise((resolve) => {
+    const child = spawn(process.execPath, [script, ...args], {
+      cwd: ROOT,
+      env: {
+        ...process.env,
+        DATABASE_URL: probeUrl(),
+        HEALTHLOG_PRISMA_CLI: join(
+          ROOT,
+          "node_modules",
+          "prisma",
+          "build",
+          "index.js",
+        ),
+        HEALTHLOG_MIGRATIONS_DIR: currentDir,
+        HEALTHLOG_MIGRATE_LOCK_RETRY_DELAYS: "1",
+      },
+    });
+    let stdout = "";
+    let stderr = "";
+    child.stdout.on("data", (chunk) => (stdout += chunk));
+    child.stderr.on("data", (chunk) => (stderr += chunk));
+    child.on("close", (code) => resolve({ code: code ?? 1, stdout, stderr }));
+  });
+}
+
+let currentDir = "";
+
+async function indexState(db: Client) {
+  const { rows } = await db.query<{ valid: boolean }>(
+    `SELECT i.indisvalid AS valid FROM pg_index i
+       JOIN pg_class c ON c.oid = i.indexrelid
+       JOIN pg_namespace n ON n.oid = c.relnamespace
+      WHERE n.nspname = $1 AND c.relname = $2`,
+    [SCHEMA, INDEX],
+  );
+  return rows[0]?.valid ?? null;
+}
+
+beforeEach(async () => {
+  const db = await connect();
+  await db.query(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`);
+  await db.end();
+});
+
+describe("a concurrent index build that failed", () => {
+  it("is built again when its migration is re-run, not skipped as present", async () => {
+    const dir = mkdtempSync(join(tmpdir(), "hl-invalid-index-"));
+    dirs.push(dir);
+    currentDir = join(dir, "migrations");
+    writeFileSync(
+      join(dir, "schema.prisma"),
+      'datasource db {\n  provider = "postgresql"\n}\n',
+    );
+    mkdirSync(join(dir, "migrations"), { recursive: true });
+    cpSync(
+      join(ROOT, "prisma", "migrations", "migration_lock.toml"),
+      join(dir, "migrations", "migration_lock.toml"),
+    );
+    // Two rows that break the unique build, the way a constraint violation,
+    // a cancel or a killed container leaves a concurrent build half done.
+    write(
+      dir,
+      "0001_probe_table",
+      `CREATE TABLE "probe" ("id" integer PRIMARY KEY, "value" integer);
+       INSERT INTO "probe" VALUES (1, 7), (2, 7);`,
+    );
+    write(
+      dir,
+      NAME,
+      `CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "${INDEX}" ON "probe" ("value");`,
+    );
+
+    const failed = await migrateDeploy(dir);
+    expect(failed.code).not.toBe(0);
+    const db = await connect();
+    try {
+      expect(await indexState(db)).toBe(false);
+
+      // The operator mends the data and marks the attempt rolled back, as
+      // docs/ops/deploy.md says.
+      await db.query(`DELETE FROM ${SCHEMA}.probe WHERE id = 2`);
+      const resolved = await prisma(dir, [
+        "migrate",
+        "resolve",
+        "--rolled-back",
+        NAME,
+      ]);
+      expect(resolved.code, resolved.stderr).toBe(0);
+
+      const again = await migrateDeploy(dir);
+      expect(again.code, again.stderr).toBe(0);
+      expect(again.stdout).toContain(`dropped the invalid index ${INDEX}`);
+      expect(await indexState(db)).toBe(true);
+    } finally {
+      await db.end();
+    }
+  });
+});