recover from interrupted CREATE INDEX CONCURRENTLY in migrations

2026-09-16 17:02:38 +00:00 · 2026-09-09 19:48:46 -04:00
parent 5c5de43e83
commit f2ef1f0db3
3 changed files with 199 additions and 1 deletions
@@ -6,17 +6,81 @@ package db
 import (
 	"context"
 	"embed"
+	"errors"
 	"fmt"
 	"io/fs"
+	"regexp"
 	"sort"
 	"strings"

+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 )

 //go:embed migrations/*.sql
 var migrationFiles embed.FS

+// execQuerier is the slice of pgxpool.Pool / pgx.Conn migrations need; it lets
+// integration tests drive applyMigration over a plain connection.
+type execQuerier interface {
+	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
+	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
+}
+
+var (
+	lineCommentRe     = regexp.MustCompile(`(?m)--[^\n]*`)
+	concurrentIndexRe = regexp.MustCompile(`(?is)\bCREATE\s+(?:UNIQUE\s+)?INDEX\s+CONCURRENTLY\s+(?:IF\s+NOT\s+EXISTS\s+)?("?[\w.]+"?)`)
+)
+
+// concurrentIndexName returns the index a CREATE INDEX CONCURRENTLY migration builds.
+func concurrentIndexName(sql string) (string, bool) {
+	m := concurrentIndexRe.FindStringSubmatch(lineCommentRe.ReplaceAllString(sql, ""))
+	if m == nil {
+		return "", false
+	}
+	return strings.Trim(m[1], `"`), true
+}
+
+// isDuplicateRelation reports SQLSTATE 42P07 (relation already exists).
+func isDuplicateRelation(err error) bool {
+	var pgErr *pgconn.PgError
+	return errors.As(err, &pgErr) && pgErr.Code == "42P07"
+}
+
+// applyMigration runs one file outside a transaction. An interrupted CREATE INDEX
+// CONCURRENTLY leaves an invalid index that trips 42P07 on retry; drop it and rebuild once.
+func applyMigration(ctx context.Context, db execQuerier, sql string) error {
+	_, err := db.Exec(ctx, sql)
+	if err == nil || !isDuplicateRelation(err) {
+		return err
+	}
+	name, ok := concurrentIndexName(sql)
+	if !ok {
+		return err
+	}
+	ident := pgx.Identifier(strings.Split(name, ".")).Sanitize()
+	var valid bool
+	if scanErr := db.QueryRow(ctx,
+		"SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass($1)", ident,
+	).Scan(&valid); scanErr != nil {
+		if errors.Is(scanErr, pgx.ErrNoRows) {
+			return err
+		}
+		return fmt.Errorf("%w (checking index %s: %v)", err, name, scanErr)
+	}
+	if valid {
+		fmt.Printf("index %s already built, recording migration\n", name)
+		return nil
+	}
+	if _, dropErr := db.Exec(ctx, "DROP INDEX CONCURRENTLY IF EXISTS "+ident); dropErr != nil {
+		return fmt.Errorf("dropping invalid index %s: %w", name, dropErr)
+	}
+	fmt.Printf("dropped invalid index %s, rebuilding\n", name)
+	_, err = db.Exec(ctx, sql)
+	return err
+}
+
 func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error {
 	_, err := pool.Exec(ctx, `
 		CREATE TABLE IF NOT EXISTS schema_migrations (
@@ -90,7 +154,7 @@ func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error {
 			return fmt.Errorf("failed to read migration %s: %w", entry.Name(), err)
 		}

-		if _, err := pool.Exec(ctx, string(sql)); err != nil {
+		if err := applyMigration(ctx, pool, string(sql)); err != nil {
 			return fmt.Errorf("failed to apply migration %s: %w", entry.Name(), err)
 		}

@@ -0,0 +1,68 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package db
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// CONCURRENTLY is illegal inside a transaction, so this uses a plain connection
+// and a real table rather than the usual tx + TEMP table pattern.
+func TestApplyMigrationRecoversInvalidIndexPostgres(t *testing.T) {
+	dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
+	if dsn == "" {
+		t.Skip("set BEACON_TEST_POSTGRES_DSN for the PostgreSQL regression test")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	conn, err := pgx.Connect(ctx, dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close(context.Background())
+
+	table := fmt.Sprintf("migrate_recovery_test_%d", time.Now().UnixNano())
+	index := table + "_idx"
+	if _, err := conn.Exec(ctx, "CREATE TABLE "+table+" (v INT)"); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _, _ = conn.Exec(context.Background(), "DROP TABLE IF EXISTS "+table+" CASCADE") }()
+
+	build := "CREATE INDEX CONCURRENTLY " + index + " ON " + table + " (v)"
+	if err := applyMigration(ctx, conn, build); err != nil {
+		t.Fatalf("initial build: %v", err)
+	}
+
+	// A valid, already-built index (process died before recording) is a success.
+	if err := applyMigration(ctx, conn, build); err != nil {
+		t.Fatalf("valid existing index must be accepted: %v", err)
+	}
+
+	// Simulate an interrupted build by flipping the catalog flag.
+	if _, err := conn.Exec(ctx, "UPDATE pg_index SET indisvalid = false WHERE indexrelid = to_regclass($1)", index); err != nil {
+		t.Skipf("cannot mark index invalid (needs catalog write privilege): %v", err)
+	}
+	if err := applyMigration(ctx, conn, build); err != nil {
+		t.Fatalf("invalid index must be dropped and rebuilt: %v", err)
+	}
+	var valid bool
+	if err := conn.QueryRow(ctx, "SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass($1)", index).Scan(&valid); err != nil {
+		t.Fatal(err)
+	}
+	if !valid {
+		t.Fatal("index still invalid after recovery")
+	}
+
+	// Recovery is scoped to CONCURRENTLY migrations; a plain duplicate still errors.
+	plain := "CREATE INDEX " + index + " ON " + table + " (v)"
+	if err := applyMigration(ctx, conn, plain); !isDuplicateRelation(err) {
+		t.Fatalf("plain duplicate index must surface 42P07, got %v", err)
+	}
+}
@@ -0,0 +1,66 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package db
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+func TestConcurrentIndexName(t *testing.T) {
+	cases := []struct {
+		name string
+		sql  string
+		want string
+		ok   bool
+	}{
+		{"bare 027 style with comments", `-- RunMigrations executes each file outside an explicit transaction. Keep this
+-- as one statement so other subscribers can keep writing during the build.
+-- If interrupted, check pg_index.indisvalid and remove this named index before
+-- retrying; do not silently accept an invalid index with IF NOT EXISTS.
+CREATE INDEX CONCURRENTLY idx_known_routes_iata_last_seen
+ON known_routes (iata, last_seen DESC);`, "idx_known_routes_iata_last_seen", true},
+		{"if not exists", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_a ON t (c);", "idx_a", true},
+		{"unique", "CREATE UNIQUE INDEX CONCURRENTLY idx_u ON t (c);", "idx_u", true},
+		{"quoted", `CREATE INDEX CONCURRENTLY "Idx_Q" ON t (c);`, "Idx_Q", true},
+		{"schema qualified", "CREATE INDEX CONCURRENTLY public.idx_s ON t (c);", "public.idx_s", true},
+		{"lowercase multiline", "create index\n  concurrently\n  idx_l\n  on t (c);", "idx_l", true},
+		{"plain create index", "CREATE INDEX idx_p ON t (c);", "", false},
+		{"only in comment", "-- CREATE INDEX CONCURRENTLY idx_c ON t (c);\nSELECT 1;", "", false},
+		{"empty", "", "", false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, ok := concurrentIndexName(tc.sql)
+			if ok != tc.ok || got != tc.want {
+				t.Fatalf("got (%q, %v), want (%q, %v)", got, ok, tc.want, tc.ok)
+			}
+		})
+	}
+}
+
+func TestIsDuplicateRelation(t *testing.T) {
+	dup := &pgconn.PgError{Code: "42P07"}
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"42P07", dup, true},
+		{"wrapped 42P07", fmt.Errorf("apply: %w", dup), true},
+		{"other sqlstate", &pgconn.PgError{Code: "42501"}, false},
+		{"plain error", errors.New("boom"), false},
+		{"nil", nil, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := isDuplicateRelation(tc.err); got != tc.want {
+				t.Fatalf("got %v, want %v", got, tc.want)
+			}
+		})
+	}
+}