From f2ef1f0db3badac11ea6febbafc97a5c581beb77 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 9 Sep 2026 19:48:46 -0400 Subject: [PATCH] recover from interrupted CREATE INDEX CONCURRENTLY in migrations --- db/migrate.go | 66 ++++++++++++++++++++++++++++++++- db/migrate_integration_test.go | 68 ++++++++++++++++++++++++++++++++++ db/migrate_test.go | 66 +++++++++++++++++++++++++++++++++ 3 files changed, 199 insertions(+), 1 deletion(-) create mode 100644 db/migrate_integration_test.go create mode 100644 db/migrate_test.go diff --git a/db/migrate.go b/db/migrate.go index 48b023b..eebad5f 100644 --- a/db/migrate.go +++ b/db/migrate.go @@ -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) } diff --git a/db/migrate_integration_test.go b/db/migrate_integration_test.go new file mode 100644 index 0000000..1239f5d --- /dev/null +++ b/db/migrate_integration_test.go @@ -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) + } +} diff --git a/db/migrate_test.go b/db/migrate_test.go new file mode 100644 index 0000000..f93f444 --- /dev/null +++ b/db/migrate_test.go @@ -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) + } + }) + } +}