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

@@ -69,7 +69,8 @@ func RunMigrations(ctx context.Context, db *bun.DB) (retErr error) {
 		return fmt.Errorf("initializing migrator: %w", err)
 	}
 	if err := migrator.Lock(ctx); err != nil {
-		return fmt.Errorf("locking migrator: %w", err)
+		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)
@@ -93,6 +94,18 @@ func RunMigrations(ctx context.Context, db *bun.DB) (retErr 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,6 +14,7 @@ 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"
@@ -42,6 +43,40 @@ func (h *migrationLockBarrier) AfterQuery(ctx context.Context, event *bun.QueryE
 	})
 }
 
+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",

internal/db/migrations/schema_integrity_test.go

@@ -95,10 +95,16 @@ func testLegacyMigrationUpgradePreservesDataAndIsIdempotent(t *testing.T, db *bu
 	}
 	markAppliedMigration(t, ctx, migrator, "2026040501", 1)
 	seedLegacyMigrationData(t, db)
+	walletBefore := readLegacyWalletRow(t, db)
 	if _, err := migrator.Migrate(ctx); err != nil {
 		t.Fatalf("upgrade legacy schema: %v", err)
 	}
 
+	// 2026062201 rebuilds wallet_operations on SQLite by copying rows.
+	if walletAfter := readLegacyWalletRow(t, db); walletAfter != walletBefore {
+		t.Fatalf("wallet row changed across the upgrade:\n before=%s\n after =%s", walletBefore, walletAfter)
+	}
+
 	var generation int
 	var current bool
 	if err := db.NewRaw("SELECT generation, is_current FROM storage_data_sets WHERE id = 1").Scan(ctx, &generation, &current); err != nil {
@@ -177,6 +183,12 @@ func seedLegacyMigrationData(t *testing.T, db *bun.DB) {
 		`INSERT INTO storage_data_sets
 			(id, bucket_id, provider_id, copy_index, status)
 		 VALUES (1, 1, '101', 0, 'ready')`,
+		`INSERT INTO wallet_operations
+			(id, type, client_request_id, amount, status, tx_hash, last_error,
+			 lease_until, started_at, submitted_at, completed_at, created_at, updated_at)
+		 VALUES (1, 'fund', 'legacy-fund', '1234', 'confirmed', '0xfeed', 'boom',
+			 '2026-01-01 01:00:00', '2026-01-02 02:00:00', '2026-01-03 03:00:00',
+			 '2026-01-04 04:00:00', '2026-01-05 05:00:00', '2026-01-06 06:00:00')`,
 	}
 	for _, statement := range statements {
 		if _, err := db.ExecContext(ctx, statement); err != nil {
@@ -185,6 +197,41 @@ func seedLegacyMigrationData(t *testing.T, db *bun.DB) {
 	}
 }
 
+func readLegacyWalletRow(t *testing.T, db *bun.DB) string {
+	t.Helper()
+	var (
+		opType, requestID, amount, status string
+		txHash, lastError                 *string
+		lease, started, submitted, done   *time.Time
+		created, updated                  time.Time
+	)
+	if err := db.NewRaw(`SELECT type, client_request_id, amount, status, tx_hash, last_error,
+		lease_until, started_at, submitted_at, completed_at, created_at, updated_at
+		FROM wallet_operations WHERE id = 1`).
+		Scan(context.Background(), &opType, &requestID, &amount, &status, &txHash, &lastError,
+			&lease, &started, &submitted, &done, &created, &updated); err != nil {
+		t.Fatalf("read wallet row: %v", err)
+	}
+	return fmt.Sprintf("%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s",
+		opType, requestID, amount, status, derefString(txHash), derefString(lastError),
+		formatTime(lease), formatTime(started), formatTime(submitted), formatTime(done),
+		formatTime(&created), formatTime(&updated))
+}
+
+func derefString(v *string) string {
+	if v == nil {
+		return "<nil>"
+	}
+	return *v
+}
+
+func formatTime(v *time.Time) string {
+	if v == nil {
+		return "<nil>"
+	}
+	return v.UTC().Format(time.RFC3339Nano)
+}
+
 func countDomainTables(ctx context.Context, db *bun.DB) (int, error) {
 	query := `SELECT COUNT(*) FROM sqlite_schema
 		WHERE type = 'table' AND name NOT LIKE 'sqlite_%'