fix(db): harden migration integrity (#12)

* fix(db): harden migration integrity

* fix(db): serialize migration execution

* fix(db): recover from a migration lock left by a killed run

Author: strahe

cmd/synaps3/main.go

@@ -159,6 +159,12 @@ func migrateCommand() *cli.Command {
 	return &cli.Command{
 		Name:  "migrate",
 		Usage: "run database migrations and exit",
+		Flags: []cli.Flag{
+			&cli.BoolFlag{
+				Name:  "force-unlock",
+				Usage: "release a migration lock left by a killed run, then exit; only use it when no migration is in progress",
+			},
+		},
 		Action: func(ctx context.Context, cmd *cli.Command) error {
 			if cmd.Args().Len() > 0 {
 				return fmt.Errorf("unexpected argument %q, migrate takes no positional arguments", cmd.Args().First())
@@ -167,6 +173,9 @@ func migrateCommand() *cli.Command {
 			if err != nil {
 				return err
 			}
+			if cmd.Bool("force-unlock") {
+				return runForceUnlock(ctx, src)
+			}
 			return runMigrate(ctx, src)
 		},
 	}
@@ -241,6 +250,20 @@ func runMigrate(ctx context.Context, src config.Source) error {
 	return nil
 }
 
+func runForceUnlock(ctx context.Context, src config.Source) error {
+	_, database, err := loadConfigAndDB(ctx, src)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = database.Close() }()
+
+	if err := db.ForceUnlockMigrations(ctx, database); err != nil {
+		return err
+	}
+	slog.Info("migration lock released")
+	return nil
+}
+
 func runServe(ctx context.Context, src config.Source) error {
 	cfg, database, err := loadConfigAndDB(ctx, src)
 	if err != nil {

internal/db/db.go

@@ -3,25 +3,28 @@ package db
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net/url"
 	"os"
 	"path/filepath"
 	"runtime"
 	"strings"
+	"time"
 
 	"github.com/strahe/synaps3/internal/config"
 	"github.com/strahe/synaps3/internal/db/migrations"
 	"github.com/uptrace/bun"
 	"github.com/uptrace/bun/dialect/pgdialect"
 	"github.com/uptrace/bun/dialect/sqlitedialect"
-	"github.com/uptrace/bun/migrate"
 
 	_ "github.com/jackc/pgx/v5/stdlib"
 	_ "modernc.org/sqlite"
 )
 
+const migrationUnlockTimeout = 5 * time.Second
+
 // New creates a Bun database connection based on the provided configuration.
 func New(cfg config.DatabaseConfig) (*bun.DB, error) {
 	var (
@@ -59,12 +62,23 @@ func New(cfg config.DatabaseConfig) (*bun.DB, error) {
 }
 
 // RunMigrations initialises the Bun migrator and applies all pending migrations.
-func RunMigrations(ctx context.Context, db *bun.DB) error {
-	migrator := migrate.NewMigrator(db, migrations.Migrations)
+func RunMigrations(ctx context.Context, db *bun.DB) (retErr error) {
+	migrator := migrations.NewMigrator(db)
 
 	if err := migrator.Init(ctx); err != nil {
 		return fmt.Errorf("initializing migrator: %w", err)
 	}
+	if err := migrator.Lock(ctx); err != nil {
+		return fmt.Errorf("locking migrator: %w; if no migration is running, "+
+			"clear a lock left by a killed run with `synaps3 migrate --force-unlock`", err)
+	}
+	defer func() {
+		unlockCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), migrationUnlockTimeout)
+		defer cancel()
+		if err := migrator.Unlock(unlockCtx); err != nil {
+			retErr = errors.Join(retErr, fmt.Errorf("unlocking migrator: %w", err))
+		}
+	}()
 
 	group, err := migrator.Migrate(ctx)
 	if err != nil {
@@ -80,6 +94,18 @@ func RunMigrations(ctx context.Context, db *bun.DB) error {
 	return nil
 }
 
+// ForceUnlockMigrations releases a migration lock left by a killed run.
+func ForceUnlockMigrations(ctx context.Context, db *bun.DB) error {
+	migrator := migrations.NewMigrator(db)
+	if err := migrator.Init(ctx); err != nil {
+		return fmt.Errorf("initializing migrator: %w", err)
+	}
+	if err := migrator.Unlock(ctx); err != nil {
+		return fmt.Errorf("releasing migration lock: %w", err)
+	}
+	return nil
+}
+
 // Ping verifies the database connection is alive.
 func Ping(ctx context.Context, db *bun.DB) error {
 	return db.PingContext(ctx)

internal/db/db_test.go

@@ -14,11 +14,113 @@ import (
 	"time"
 
 	"github.com/strahe/synaps3/internal/config"
+	"github.com/strahe/synaps3/internal/db/migrations"
 	"github.com/strahe/synaps3/internal/db/repository"
 	"github.com/strahe/synaps3/internal/model"
 	"github.com/uptrace/bun"
 )
 
+type migrationLockBarrier struct {
+	locked  chan struct{}
+	release chan struct{}
+	once    sync.Once
+}
+
+func (h *migrationLockBarrier) BeforeQuery(ctx context.Context, _ *bun.QueryEvent) context.Context {
+	return ctx
+}
+
+func (h *migrationLockBarrier) AfterQuery(ctx context.Context, event *bun.QueryEvent) {
+	if event.Err != nil || event.Operation() != "INSERT" || !strings.Contains(event.Query, "bun_migration_locks") {
+		return
+	}
+	h.once.Do(func() {
+		close(h.locked)
+		select {
+		case <-h.release:
+		case <-ctx.Done():
+		}
+	})
+}
+
+func TestForceUnlockMigrationsClearsALockLeftByAKilledRun(t *testing.T) {
+	cfg := config.DatabaseConfig{
+		Driver:       "sqlite",
+		DSN:          "file:" + filepath.Join(t.TempDir(), "force-unlock.db"),
+		MaxOpenConns: 2,
+		MaxIdleConns: 2,
+	}
+	db, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	ctx := context.Background()
+
+	if err := RunMigrations(ctx, db); err != nil {
+		t.Fatalf("RunMigrations() error = %v", err)
+	}
+	// A killed run never reaches its deferred unlock.
+	if err := migrations.NewMigrator(db).Lock(ctx); err != nil {
+		t.Fatalf("acquiring the leaked lock: %v", err)
+	}
+	err = RunMigrations(ctx, db)
+	if err == nil || !strings.Contains(err.Error(), "--force-unlock") {
+		t.Fatalf("RunMigrations() error = %v, want the stale-lock remediation", err)
+	}
+
+	if err := ForceUnlockMigrations(ctx, db); err != nil {
+		t.Fatalf("ForceUnlockMigrations() error = %v", err)
+	}
+	if err := RunMigrations(ctx, db); err != nil {
+		t.Fatalf("RunMigrations() after force unlock = %v, want success", err)
+	}
+}
+
+func TestRunMigrationsSerializesConcurrentRunnersAndUnlocksAfterCancellation(t *testing.T) {
+	cfg := config.DatabaseConfig{
+		Driver:       "sqlite",
+		DSN:          "file:" + filepath.Join(t.TempDir(), "migration-lock.db") + "?_pragma=journal_mode(WAL)",
+		MaxOpenConns: 2,
+		MaxIdleConns: 2,
+	}
+	db, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+
+	barrier := &migrationLockBarrier{locked: make(chan struct{}), release: make(chan struct{})}
+	db.AddQueryHook(barrier)
+	ctx, cancel := context.WithCancel(context.Background())
+	firstResult := make(chan error, 1)
+	go func() {
+		firstResult <- RunMigrations(ctx, db)
+	}()
+
+	select {
+	case <-barrier.locked:
+	case <-time.After(5 * time.Second):
+		cancel()
+		close(barrier.release)
+		t.Fatal("first migration runner did not acquire the lock")
+	}
+	if err := RunMigrations(context.Background(), db); err == nil || !strings.Contains(err.Error(), "already locked") {
+		cancel()
+		close(barrier.release)
+		t.Fatalf("concurrent RunMigrations() error = %v, want migration lock conflict", err)
+	}
+
+	cancel()
+	close(barrier.release)
+	if err := <-firstResult; err == nil {
+		t.Fatal("cancelled RunMigrations() succeeded")
+	}
+	if err := RunMigrations(context.Background(), db); err != nil {
+		t.Fatalf("RunMigrations() after cancelled owner = %v, want released lock", err)
+	}
+}
+
 func TestNew_SQLiteConcurrentClaimsDoNotBusy(t *testing.T) {
 	t.Parallel()