fix(db): serialize concurrent migrations

internal/database/database.go

@@ -3,6 +3,7 @@ package database
 import (
 	"context"
 	"database/sql"
+	"database/sql/driver"
 	"embed"
 	"fmt"
 	"log/slog"
@@ -15,6 +16,8 @@ import (
 //go:embed migrations/*.sql
 var migrationsFS embed.FS
 
+const migrationLockID int64 = 0x706179737461626c
+
 // 1)Connect: establishes a connection to the PostgreSQL database using the provided URL
 func Connect(databaseURL string) (*sql.DB, error) {
 	db, err := sql.Open("postgres", databaseURL)
@@ -29,7 +32,19 @@ func Connect(databaseURL string) (*sql.DB, error) {
 
 // 2)Migrate: ensures the schema_migrations table exists, reads migration files from the embedded filesystem, and used for executing all sql files
 func Migrate(db *sql.DB) error {
-	_, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (
+	ctx := context.Background()
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return fmt.Errorf("connect for migrations: %w", err)
+	}
+	defer conn.Close() //nolint:errcheck
+	// Close the session to release its lock and any failed transaction.
+	defer conn.Raw(func(any) error { return driver.ErrBadConn }) //nolint:errcheck
+	if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_lock($1)", migrationLockID); err != nil {
+		return fmt.Errorf("lock migrations: %w", err)
+	}
+
+	_, err = conn.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
 		version text PRIMARY KEY,
 		applied_at timestamptz NOT NULL DEFAULT now()
 	)`)
@@ -54,7 +69,7 @@ func Migrate(db *sql.DB) error {
 		version := strings.TrimSuffix(name, ".up.sql")
 
 		var exists bool
-		err := db.QueryRow("SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)", version).Scan(&exists)
+		err := conn.QueryRowContext(ctx, "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)", version).Scan(&exists)
 		if err != nil {
 			return fmt.Errorf("check migration %s: %w", version, err)
 		}
@@ -67,11 +82,11 @@ func Migrate(db *sql.DB) error {
 			return fmt.Errorf("read migration %s: %w", name, err)
 		}
 
-		if _, err := db.Exec(string(content)); err != nil {
+		if _, err := conn.ExecContext(ctx, string(content)); err != nil {
 			return fmt.Errorf("apply migration %s: %w", name, err)
 		}
 
-		if _, err := db.Exec("INSERT INTO schema_migrations (version) VALUES ($1)", version); err != nil {
+		if _, err := conn.ExecContext(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version); err != nil {
 			return fmt.Errorf("record migration %s: %w", version, err)
 		}
 

internal/database/database_test.go

@@ -0,0 +1,123 @@
+package database
+
+import (
+	"database/sql"
+	"fmt"
+	"net/url"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/lib/pq"
+)
+
+func migrationTestDB(t *testing.T) *sql.DB {
+	t.Helper()
+	dsn := os.Getenv("DATABASE_URL")
+	if dsn == "" {
+		t.Skip("DATABASE_URL not set")
+	}
+	admin, err := sql.Open("postgres", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = admin.Close() })
+	schema := fmt.Sprintf("migration_test_%d", time.Now().UnixNano())
+	if _, err := admin.Exec("CREATE SCHEMA " + pq.QuoteIdentifier(schema)); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if _, err := admin.Exec("DROP SCHEMA " + pq.QuoteIdentifier(schema) + " CASCADE"); err != nil {
+			t.Error(err)
+		}
+	})
+	if strings.HasPrefix(dsn, "postgres://") || strings.HasPrefix(dsn, "postgresql://") {
+		u, err := url.Parse(dsn)
+		if err != nil {
+			t.Fatal("invalid test database URL")
+		}
+		q := u.Query()
+		q.Set("search_path", schema)
+		q.Set("statement_timeout", "10000")
+		u.RawQuery = q.Encode()
+		dsn = u.String()
+	} else {
+		dsn += " search_path=" + schema + " statement_timeout=10000"
+	}
+	db, err := sql.Open("postgres", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	return db
+}
+
+func TestMigrateConcurrent(t *testing.T) {
+	db := migrationTestDB(t)
+	const callers = 8
+	db.SetMaxOpenConns(callers)
+	var tables int
+	if err := db.QueryRow("SELECT count(*) FROM information_schema.tables WHERE table_schema=current_schema()").Scan(&tables); err != nil {
+		t.Fatal(err)
+	}
+	if tables != 0 {
+		t.Fatalf("schema contains %d tables before migration", tables)
+	}
+	start := make(chan struct{})
+	results := make(chan error, callers)
+	for range callers {
+		go func() {
+			<-start
+			results <- Migrate(db)
+		}()
+	}
+	close(start)
+	for range callers {
+		if err := <-results; err != nil {
+			t.Error(err)
+		}
+	}
+	entries, err := migrationsFS.ReadDir("migrations")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, entry := range entries {
+		if !strings.HasSuffix(entry.Name(), ".up.sql") {
+			continue
+		}
+		var count int
+		version := strings.TrimSuffix(entry.Name(), ".up.sql")
+		if err := db.QueryRow("SELECT count(*) FROM schema_migrations WHERE version=$1", version).Scan(&count); err != nil {
+			t.Fatal(err)
+		}
+		if count != 1 {
+			t.Errorf("migration %s recorded %d times", version, count)
+		}
+	}
+	var available bool
+	if err := db.QueryRow("SELECT pg_try_advisory_lock($1)", migrationLockID).Scan(&available); err != nil {
+		t.Fatal(err)
+	}
+	if !available {
+		t.Fatal("migration lock remains held")
+	}
+}
+
+func TestMigrateFailureReleasesSession(t *testing.T) {
+	db := migrationTestDB(t)
+	db.SetMaxOpenConns(1)
+	db.SetMaxIdleConns(0)
+	if _, err := db.Exec("CREATE TABLE holds (id bigint)"); err != nil {
+		t.Fatal(err)
+	}
+	if err := Migrate(db); err == nil {
+		t.Fatal("expected migration failure for an existing holds table")
+	}
+	if _, err := db.Exec("DROP TABLE holds"); err != nil {
+		t.Fatal(err)
+	}
+	if err := Migrate(db); err != nil {
+		t.Fatalf("migration after failure: %v", err)
+	}
+}