Update morph dependency (#19308)

* Update morph dependency to use a newer version

* remove timeout check for migrations statements

* store/sqlstore: reset timeout for mysql while creating db for migrations

* update morph to v0.2.1

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/fsnotify/fsnotify v1.5.1
 	github.com/getsentry/sentry-go v0.11.0
 	github.com/go-asn1-ber/asn1-ber v1.5.3 // indirect
-	github.com/go-morph/morph v0.0.0-20211202070306-dbdc0736c17e
+	github.com/go-morph/morph v0.2.1
 	github.com/go-redis/redis/v8 v8.11.4 // indirect
 	github.com/go-resty/resty/v2 v2.7.0 // indirect
 	github.com/go-sql-driver/mysql v1.6.0

go.sum

@@ -529,6 +529,10 @@ github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTg
 github.com/go-martini/martini v0.0.0-20170121215854-22fa46961aab/go.mod h1:/P9AEU963A2AYjv4d1V5eVL1CQbEJq6aCNHDDjibzu8=
 github.com/go-morph/morph v0.0.0-20211202070306-dbdc0736c17e h1:iS/jL0Xij+6O5d/VXrt+VOZfATHCTDIjBVVthVpHFY0=
 github.com/go-morph/morph v0.0.0-20211202070306-dbdc0736c17e/go.mod h1:XQh5WcM351wOV3z3zEWRM8RaJ65E2p4P7WWbmFAi8x4=
+github.com/go-morph/morph v0.0.0-20220110203813-7e65a95885ea h1:l2xnsVP6YW9DZet6HfUhMKtEOPyGg2QsRyDPLECnW1o=
+github.com/go-morph/morph v0.0.0-20220110203813-7e65a95885ea/go.mod h1:XQh5WcM351wOV3z3zEWRM8RaJ65E2p4P7WWbmFAi8x4=
+github.com/go-morph/morph v0.2.1 h1:Wo+TUt4+jCwKJh57mpuRBSMioYjA2TwjQD027hhOgro=
+github.com/go-morph/morph v0.2.1/go.mod h1:XQh5WcM351wOV3z3zEWRM8RaJ65E2p4P7WWbmFAi8x4=
 github.com/go-openapi/jsonpointer v0.19.2/go.mod h1:3akKfEdA7DF1sugOqz1dVQHBcuDBPKZGEoHC/NkiQRg=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonreference v0.19.2/go.mod h1:jMjeRr2HHw6nAVajTXJ4eiUwohSTlpa0o73RUL1owJc=

i18n/en.json

@@ -8287,10 +8287,6 @@
     "id": "model.config.is_valid.sql_max_conn.app_error",
     "translation": "Invalid maximum open connection for SQL settings. Must be a positive number."
   },
-  {
-    "id": "model.config.is_valid.sql_migrations_statement_timout.app_error",
-    "translation": "Invalid timeout value for migrations statements. Must be a positive number."
-  },
   {
     "id": "model.config.is_valid.sql_query_timeout.app_error",
     "translation": "Invalid query timeout for SQL settings. Must be a positive number."

model/config.go

@@ -3292,10 +3292,6 @@ func (s *SqlSettings) isValid() *AppError {
 		return NewAppError("Config.IsValid", "model.config.is_valid.sql_max_conn.app_error", nil, "", http.StatusBadRequest)
 	}
 
-	if *s.MigrationsStatementTimeoutSeconds <= 0 {
-		return NewAppError("Config.IsValid", "model.config.is_valid.sql_migrations_statement_timout.app_error", nil, "", http.StatusBadRequest)
-	}
-
 	return nil
 }
 

store/sqlstore/store.go

@@ -1444,18 +1444,27 @@ func (ss *SqlStore) migrate(direction migrationDirection) error {
 	var driver drivers.Driver
 	switch ss.DriverName() {
 	case model.DatabaseDriverMysql:
-		dataSource, err2 := ss.appendMultipleStatementsFlag(*ss.settings.DataSource)
-		if err2 != nil {
-			return err2
+		dataSource, rErr := resetReadTimeout(*ss.settings.DataSource)
+		if rErr != nil {
+			mlog.Fatal("Failed to reset read timeout from datasource.", mlog.Err(rErr), mlog.String("src", *ss.settings.DataSource))
+			return rErr
+		}
+		dataSource, err = ss.appendMultipleStatementsFlag(dataSource)
+		if err != nil {
+			return err
 		}
 		db := setupConnection("master", dataSource, ss.settings)
 		driver, err = ms.WithInstance(db, &ms.Config{
-			StatementTimeoutInSecs: *ss.settings.MigrationsStatementTimeoutSeconds,
+			Config: drivers.Config{
+				StatementTimeoutInSecs: *ss.settings.MigrationsStatementTimeoutSeconds,
+			},
 		})
 		defer db.Close()
 	case model.DatabaseDriverPostgres:
 		driver, err = ps.WithInstance(ss.GetMasterX().DB.DB, &ps.Config{
-			StatementTimeoutInSecs: *ss.settings.MigrationsStatementTimeoutSeconds,
+			Config: drivers.Config{
+				StatementTimeoutInSecs: *ss.settings.MigrationsStatementTimeoutSeconds,
+			},
 		})
 	default:
 		err = fmt.Errorf("unsupported database type %s for migration", ss.DriverName())
@@ -1464,7 +1473,7 @@ func (ss *SqlStore) migrate(direction migrationDirection) error {
 		return err
 	}
 
-	engine, err := morph.New(driver, src)
+	engine, err := morph.New(context.Background(), driver, src, morph.WithLock("mm-lock-key"))
 	if err != nil {
 		return err
 	}

vendor/github.com/go-morph/morph/README.md

@@ -16,6 +16,8 @@ It can be used as a library or a CLI tool.
 
 ```Go
 import (
+    "context"
+
     "github.com/go-morph/morph"
     "github.com/go-morph/morph/drivers/mysql"
     bindata "github.com/go-morph/morph/sources/go_bindata"
@@ -37,7 +39,7 @@ if err != nil {
     return err
 }
 
-engine, err := morph.New(driver, src)
+engine, err := morph.New(context.Background(), driver, src)
 if err != nil {
     return err
 }

vendor/github.com/go-morph/morph/drivers/driver.go

@@ -4,6 +4,14 @@ import (
 	"github.com/go-morph/morph/models"
 )
 
+type Config struct {
+	MigrationsTable string
+	// StatementTimeoutInSecs is used to set a timeout for each migration file.
+	// Set below zero to disable timeout. Zero value will result in default value, which is 60 seconds.
+	StatementTimeoutInSecs int
+	MigrationMaxSize       int
+}
+
 type Driver interface {
 	Ping() error
 	// Close closes the underlying db connection. If the driver is created via Open() function

vendor/github.com/go-morph/morph/drivers/lock.go

@@ -0,0 +1,78 @@
+package drivers
+
+import (
+	"context"
+	"errors"
+	"math/rand"
+	"sync"
+	"time"
+)
+
+const (
+	// MutexTableName is the name being used for the mutex table
+	MutexTableName = "db_lock"
+
+	// minWaitInterval is the minimum amount of time to wait between locking attempts
+	minWaitInterval = 1 * time.Second
+
+	// maxWaitInterval is the maximum amount of time to wait between locking attempts
+	maxWaitInterval = 5 * time.Minute
+
+	// pollWaitInterval is the usual time to wait between unsuccessful locking attempts
+	pollWaitInterval = 1 * time.Second
+
+	// jitterWaitInterval is the amount of jitter to add when waiting to avoid thundering herds
+	jitterWaitInterval = minWaitInterval / 2
+
+	// TTL is the interval after which a locked mutex will expire unless refreshed
+	TTL = time.Second * 15
+
+	// RefreshInterval is the interval on which the mutex will be refreshed when locked
+	RefreshInterval = TTL / 2
+)
+
+// MakeLockKey returns the prefixed key used to namespace mutex keys.
+func MakeLockKey(key string) (string, error) {
+	if key == "" {
+		return "", errors.New("must specify valid mutex key")
+	}
+
+	return key, nil
+}
+
+// NextWaitInterval determines how long to wait until the next lock retry.
+func NextWaitInterval(lastWaitInterval time.Duration, err error) time.Duration {
+	nextWaitInterval := lastWaitInterval
+
+	if nextWaitInterval <= 0 {
+		nextWaitInterval = minWaitInterval
+	}
+
+	if err != nil {
+		nextWaitInterval *= 2
+		if nextWaitInterval > maxWaitInterval {
+			nextWaitInterval = maxWaitInterval
+		}
+	} else {
+		nextWaitInterval = pollWaitInterval
+	}
+
+	// Add some jitter to avoid unnecessary collision between competing other instances.
+	nextWaitInterval += time.Duration(rand.Int63n(int64(jitterWaitInterval)) - int64(jitterWaitInterval)/2)
+
+	return nextWaitInterval
+}
+
+type Locker interface {
+	sync.Locker
+	// LockWithContext locks m unless the context is canceled. If the mutex is already locked by any other
+	// instance, including the current one, the calling goroutine blocks until the mutex can be locked,
+	// or the context is canceled.
+	//
+	// The mutex is locked only if a nil error is returned.
+	LockWithContext(ctx context.Context) error
+}
+
+type Lockable interface {
+	DriverName() string
+}

vendor/github.com/go-morph/morph/drivers/mysql/lock.go

@@ -0,0 +1,268 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/go-morph/morph/drivers"
+)
+
+// Mutex is similar to sync.Mutex, except usable by morph to lock the db.
+//
+// Pick a unique name for each mutex your plugin requires.
+//
+// A Mutex must not be copied after first use.
+type Mutex struct {
+	noCopy
+	key string
+
+	// lock guards the variables used to manage the refresh task, and is not itself related to
+	// the db lock.
+	lock        sync.Mutex
+	stopRefresh chan bool
+	refreshDone chan bool
+	conn        *sql.Conn
+}
+
+// NewMutex creates a mutex with the given key name.
+//
+// returns error if key is empty.
+func NewMutex(key string, driver drivers.Driver) (*Mutex, error) {
+	key, err := drivers.MakeLockKey(key)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), drivers.TTL)
+	defer cancel()
+
+	ms, ok := driver.(*mysql)
+	if !ok {
+		return nil, errors.New("incorrect implementation of the driver")
+	}
+
+	conn, err := ms.db.Conn(context.Background())
+	if err != nil {
+		return nil, err
+	}
+
+	createTableIfNotExistsQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (Id varchar(64) NOT NULL, ExpireAt bigint(20) NOT NULL, PRIMARY KEY (Id)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", drivers.MutexTableName)
+	if _, err = conn.ExecContext(ctx, createTableIfNotExistsQuery); err != nil {
+		return nil, err
+	}
+
+	return &Mutex{
+		key:  key,
+		conn: conn,
+	}, nil
+}
+
+// lock makes a single attempt to lock the mutex, returning true only if successful.
+func (m *Mutex) tryLock(ctx context.Context) (bool, error) {
+	now := time.Now()
+	tx, err := m.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return false, err
+	}
+
+	query := fmt.Sprintf("INSERT INTO %s (Id, ExpireAt) VALUES (?, ?)", drivers.MutexTableName)
+	if _, err := tx.Exec(query, m.key, now.Add(drivers.TTL).Unix()); err != nil {
+		err2 := m.releaseLock(tx, now)
+		if err2 == nil { // lock has been released due to expiration
+			return true, nil
+		}
+
+		return false, fmt.Errorf("failed to lock mutex: %w", err)
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return false, txErr
+		}
+
+		return false, err
+	}
+
+	return true, nil
+}
+
+func (m *Mutex) releaseLock(tx *sql.Tx, t time.Time) error {
+	e, err := m.getExpireAt(tx)
+	if err != nil {
+		return err
+	}
+
+	if t.Unix() < e {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback: %w", txErr)
+		}
+
+		return errors.New("could not release the lock")
+	}
+
+	query := fmt.Sprintf("UPDATE %s SET ExpireAt = ? WHERE Id = ?", drivers.MutexTableName)
+	if err = executeTx(tx, query, t.Add(drivers.TTL).Unix(), m.key); err != nil {
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback transaction: %w", txErr)
+		}
+
+		return fmt.Errorf("unable to set new expireat for mutex: %w", err)
+	}
+
+	return nil
+}
+
+func (m *Mutex) getExpireAt(tx *sql.Tx) (int64, error) {
+	var expireAt int64
+	query := fmt.Sprintf("SELECT ExpireAt FROM %s WHERE Id = ?", drivers.MutexTableName)
+	err := tx.QueryRow(query, m.key).Scan(&expireAt)
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return -1, fmt.Errorf("could not rollback: %w", txErr)
+		}
+
+		return -1, fmt.Errorf("failed to fetch mutex from db: %w", err)
+	}
+
+	return expireAt, nil
+}
+
+// refreshLock rewrites the lock key value with a new expiry, returning nil only if successful.
+func (m *Mutex) refreshLock(ctx context.Context) error {
+	tx, err := m.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+
+	e, err := m.getExpireAt(tx)
+	if err != nil {
+		return err
+	}
+
+	tmp := time.Unix(e, 0)
+	query := fmt.Sprintf("UPDATE %s SET ExpireAt = ? WHERE Id = ?", drivers.MutexTableName)
+	if err = executeTx(tx, query, tmp.Add(drivers.TTL).Unix(), m.key); err != nil {
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback: %w", txErr)
+		}
+
+		return fmt.Errorf("unable to refresh expireat for mutex: %w", err)
+	}
+
+	return nil
+}
+
+// Lock locks m. If the mutex is already locked by any other morph instance, including the current one,
+// the calling goroutine blocks until the mutex can be locked.
+func (m *Mutex) Lock() {
+	_ = m.LockWithContext(context.Background())
+}
+
+// LockWithContext locks m unless the context is canceled. If the mutex is already locked by any other
+// instance, including the current one, the calling goroutine blocks until the mutex can be locked,
+// or the context is canceled.
+//
+// The mutex is locked only if a nil error is returned.
+func (m *Mutex) LockWithContext(ctx context.Context) error {
+	var waitInterval time.Duration
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(waitInterval):
+		}
+
+		ok, err := m.tryLock(ctx)
+		if err != nil || !ok {
+			waitInterval = drivers.NextWaitInterval(waitInterval, err)
+			continue
+		}
+
+		break
+	}
+
+	stop := make(chan bool)
+	done := make(chan bool)
+	go func() {
+		defer close(done)
+		t := time.NewTicker(drivers.RefreshInterval)
+		for {
+			select {
+			case <-t.C:
+				err := m.refreshLock(ctx)
+				if err != nil {
+					return
+				}
+			case <-stop:
+				return
+			}
+		}
+	}()
+
+	m.lock.Lock()
+	m.stopRefresh = stop
+	m.refreshDone = done
+	m.lock.Unlock()
+
+	return nil
+}
+
+// Unlock unlocks m. It is a run-time error if m is not locked on entry to Unlock.
+//
+// Just like sync.Mutex, a locked Lock is not associated with a particular goroutine or a process.
+func (m *Mutex) Unlock() {
+	m.lock.Lock()
+	if m.stopRefresh == nil {
+		m.lock.Unlock()
+		panic("mutex has not been acquired")
+	}
+
+	close(m.stopRefresh)
+	m.stopRefresh = nil
+	<-m.refreshDone
+	m.lock.Unlock()
+
+	defer m.conn.Close()
+
+	// If an error occurs deleting, the mutex will still expire, allowing later retry.
+	query := fmt.Sprintf("DELETE FROM %s WHERE Id = ?", drivers.MutexTableName)
+	_, _ = m.conn.ExecContext(context.Background(), query, m.key)
+}
+
+func executeTx(tx *sql.Tx, query string, args ...interface{}) error {
+	if _, err := tx.Exec(query, args...); err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback transaction: %w", txErr)
+		}
+
+		return err
+	}
+
+	return nil
+}
+
+// noCopy may be embedded into structs which must not be copied
+// after the first use.
+//
+// See https://golang.org/issues/8005#issuecomment-190753527
+// for details.
+type noCopy struct{}
+
+// Lock is a no-op used by -copylocks checker from `go vet`.
+func (*noCopy) Lock() {}

vendor/github.com/go-morph/morph/drivers/mysql/mysql.go

@@ -6,7 +6,6 @@ import (
 	"database/sql"
 	"fmt"
 	"strconv"
-	"time"
 
 	"github.com/pkg/errors"
 
@@ -18,9 +17,11 @@ import (
 const driverName = "mysql"
 const defaultMigrationMaxSize = 10 * 1 << 20 // 10 MB
 var defaultConfig = &Config{
-	MigrationsTable:        "db_migrations",
-	StatementTimeoutInSecs: 60,
-	MigrationMaxSize:       defaultMigrationMaxSize,
+	Config: drivers.Config{
+		MigrationsTable:        "db_migrations",
+		StatementTimeoutInSecs: 60,
+		MigrationMaxSize:       defaultMigrationMaxSize,
+	},
 }
 
 // add here any custom driver configuration
@@ -31,11 +32,9 @@ var configParams = []string{
 }
 
 type Config struct {
-	MigrationsTable        string
-	StatementTimeoutInSecs int
-	MigrationMaxSize       int
-	databaseName           string
-	closeDBonClose         bool
+	drivers.Config
+	databaseName   string
+	closeDBonClose bool
 }
 
 type mysql struct {
@@ -99,12 +98,16 @@ func Open(connURL string) (drivers.Driver, error) {
 }
 
 func (driver *mysql) Ping() error {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(driver.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(driver.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	return driver.conn.PingContext(ctx)
 }
 
+func (mysql) DriverName() string {
+	return driverName
+}
+
 func (driver *mysql) Close() error {
 	if driver.conn != nil {
 		if err := driver.conn.Close(); err != nil {
@@ -144,7 +147,7 @@ func (driver *mysql) Lock() error {
 	// This will wait until the lock can be acquired or until the statement timeout has reached.
 	query := fmt.Sprintf("SELECT GET_LOCK(?, %d)", driver.config.StatementTimeoutInSecs)
 	var success bool
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(driver.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(driver.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	if err := driver.conn.QueryRowContext(ctx, query, aid).Scan(&success); err != nil {
@@ -177,7 +180,7 @@ func (driver *mysql) Unlock() error {
 	}
 
 	query := `SELECT RELEASE_LOCK(?)`
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(driver.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(driver.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	if _, err := driver.conn.ExecContext(ctx, query, aid); err != nil {
@@ -194,7 +197,7 @@ func (driver *mysql) Unlock() error {
 }
 
 func (driver *mysql) createSchemaTableIfNotExists() (err error) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(driver.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(driver.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	createTableIfNotExistsQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (Version bigint(20) NOT NULL, Name varchar(64) NOT NULL, PRIMARY KEY (Version)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", driver.config.MigrationsTable)
@@ -232,7 +235,7 @@ func (driver *mysql) Apply(migration *models.Migration, saveVersion bool) (err e
 		}
 	}
 	defer migration.Close()
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(driver.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(driver.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	if _, err := driver.conn.ExecContext(ctx, query); err != nil {
@@ -245,7 +248,7 @@ func (driver *mysql) Apply(migration *models.Migration, saveVersion bool) (err e
 		}
 	}
 
-	updateVersionContext, cancel := context.WithTimeout(context.Background(), time.Duration(driver.config.StatementTimeoutInSecs)*time.Second)
+	updateVersionContext, cancel := drivers.GetContext(driver.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	if !saveVersion {
@@ -291,7 +294,7 @@ func (driver *mysql) AppliedMigrations() (migrations []*models.Migration, err er
 	}
 
 	query := fmt.Sprintf("SELECT version, name FROM %s", driver.config.MigrationsTable)
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(driver.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(driver.config.StatementTimeoutInSecs)
 	defer cancel()
 	var appliedMigrations []*models.Migration
 	var version uint32
@@ -307,6 +310,7 @@ func (driver *mysql) AppliedMigrations() (migrations []*models.Migration, err er
 			Query:   []byte(query),
 		}
 	}
+	defer rows.Close()
 
 	for rows.Next() {
 		if err := rows.Scan(&version, &name); err != nil {
@@ -331,7 +335,7 @@ func (driver *mysql) AppliedMigrations() (migrations []*models.Migration, err er
 func currentDatabaseNameFromDB(conn *sql.Conn, config *Config) (string, error) {
 	query := "SELECT DATABASE()"
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(config.StatementTimeoutInSecs)
 	defer cancel()
 
 	var databaseName string

vendor/github.com/go-morph/morph/drivers/postgres/lock.go

@@ -0,0 +1,268 @@
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/go-morph/morph/drivers"
+)
+
+// Mutex is similar to sync.Mutex, except usable by morph to lock the db.
+//
+// Pick a unique name for each mutex your plugin requires.
+//
+// A Mutex must not be copied after first use.
+type Mutex struct {
+	noCopy
+	key string
+
+	// lock guards the variables used to manage the refresh task, and is not itself related to
+	// the db lock.
+	lock        sync.Mutex
+	stopRefresh chan bool
+	refreshDone chan bool
+	conn        *sql.Conn
+}
+
+// NewMutex creates a mutex with the given key name.
+//
+// returns error if key is empty.
+func NewMutex(key string, driver drivers.Driver) (*Mutex, error) {
+	key, err := drivers.MakeLockKey(key)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), drivers.TTL)
+	defer cancel()
+
+	ps, ok := driver.(*postgres)
+	if !ok {
+		return nil, errors.New("incorrect implementation of the driver")
+	}
+
+	conn, err := ps.db.Conn(context.Background())
+	if err != nil {
+		return nil, err
+	}
+
+	createTableIfNotExistsQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id varchar(64) PRIMARY KEY, expireat bigint);", drivers.MutexTableName)
+	if _, err = conn.ExecContext(ctx, createTableIfNotExistsQuery); err != nil {
+		return nil, err
+	}
+
+	return &Mutex{
+		key:  key,
+		conn: conn,
+	}, nil
+}
+
+// lock makes a single attempt to lock the mutex, returning true only if successful.
+func (m *Mutex) tryLock(ctx context.Context) (bool, error) {
+	now := time.Now()
+	tx, err := m.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return false, err
+	}
+
+	query := fmt.Sprintf("INSERT INTO %s (id, expireat) VALUES ($1, $2)", drivers.MutexTableName)
+	if _, err := tx.Exec(query, m.key, now.Add(drivers.TTL).Unix()); err != nil {
+		err2 := m.releaseLock(tx, now)
+		if err2 == nil { // lock has been released due to expiration
+			return true, nil
+		}
+
+		return false, fmt.Errorf("failed to lock mutex: %w", err)
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return false, txErr
+		}
+
+		return false, err
+	}
+
+	return true, nil
+}
+
+func (m *Mutex) releaseLock(tx *sql.Tx, t time.Time) error {
+	e, err := m.getExpireAt(tx)
+	if err != nil {
+		return err
+	}
+
+	if t.Unix() < e {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback: %w", txErr)
+		}
+
+		return errors.New("could not release the lock")
+	}
+
+	query := fmt.Sprintf("UPDATE %s SET expireat = $1 WHERE id = $2", drivers.MutexTableName)
+	if err = executeTx(tx, query, t.Add(drivers.TTL).Unix(), m.key); err != nil {
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback transaction: %w", txErr)
+		}
+
+		return fmt.Errorf("unable to set new expireat for mutex: %w", err)
+	}
+
+	return nil
+}
+
+func (m *Mutex) getExpireAt(tx *sql.Tx) (int64, error) {
+	var expireAt int64
+	query := fmt.Sprintf("SELECT expireat FROM %s WHERE id = $1", drivers.MutexTableName)
+	err := tx.QueryRow(query, m.key).Scan(&expireAt)
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return -1, fmt.Errorf("could not rollback: %w", txErr)
+		}
+
+		return -1, fmt.Errorf("failed to fetch mutex from db: %w", err)
+	}
+
+	return expireAt, nil
+}
+
+// refreshLock rewrites the lock key value with a new expiry, returning nil only if successful.
+func (m *Mutex) refreshLock(ctx context.Context) error {
+	tx, err := m.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+
+	e, err := m.getExpireAt(tx)
+	if err != nil {
+		return err
+	}
+
+	tmp := time.Unix(e, 0)
+	query := fmt.Sprintf("UPDATE %s SET expireat = $1 WHERE id = $2", drivers.MutexTableName)
+	if err = executeTx(tx, query, tmp.Add(drivers.TTL).Unix(), m.key); err != nil {
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback: %w", txErr)
+		}
+
+		return fmt.Errorf("unable to refresh expireat for mutex: %w", err)
+	}
+
+	return nil
+}
+
+// Lock locks m. If the mutex is already locked by any other morph instance, including the current one,
+// the calling goroutine blocks until the mutex can be locked.
+func (m *Mutex) Lock() {
+	_ = m.LockWithContext(context.Background())
+}
+
+// LockWithContext locks m unless the context is canceled. If the mutex is already locked by any other
+// instance, including the current one, the calling goroutine blocks until the mutex can be locked,
+// or the context is canceled.
+//
+// The mutex is locked only if a nil error is returned.
+func (m *Mutex) LockWithContext(ctx context.Context) error {
+	var waitInterval time.Duration
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(waitInterval):
+		}
+
+		ok, err := m.tryLock(ctx)
+		if err != nil || !ok {
+			waitInterval = drivers.NextWaitInterval(waitInterval, err)
+			continue
+		}
+
+		break
+	}
+
+	stop := make(chan bool)
+	done := make(chan bool)
+	go func() {
+		defer close(done)
+		t := time.NewTicker(drivers.RefreshInterval)
+		for {
+			select {
+			case <-t.C:
+				err := m.refreshLock(ctx)
+				if err != nil {
+					return
+				}
+			case <-stop:
+				return
+			}
+		}
+	}()
+
+	m.lock.Lock()
+	m.stopRefresh = stop
+	m.refreshDone = done
+	m.lock.Unlock()
+
+	return nil
+}
+
+// Unlock unlocks m. It is a run-time error if m is not locked on entry to Unlock.
+//
+// Just like sync.Mutex, a locked Lock is not associated with a particular goroutine or a process.
+func (m *Mutex) Unlock() {
+	m.lock.Lock()
+	if m.stopRefresh == nil {
+		m.lock.Unlock()
+		panic("mutex has not been acquired")
+	}
+
+	close(m.stopRefresh)
+	m.stopRefresh = nil
+	<-m.refreshDone
+	m.lock.Unlock()
+
+	defer m.conn.Close()
+
+	// If an error occurs deleting, the mutex will still expire, allowing later retry.
+	query := fmt.Sprintf("DELETE FROM %s WHERE id = $1", drivers.MutexTableName)
+	_, _ = m.conn.ExecContext(context.Background(), query, m.key)
+}
+
+func executeTx(tx *sql.Tx, query string, args ...interface{}) error {
+	if _, err := tx.Exec(query, args...); err != nil {
+		if txErr := tx.Rollback(); txErr != nil {
+			return fmt.Errorf("could not rollback transaction: %w", txErr)
+		}
+
+		return err
+	}
+
+	return nil
+}
+
+// noCopy may be embedded into structs which must not be copied
+// after the first use.
+//
+// See https://golang.org/issues/8005#issuecomment-190753527
+// for details.
+type noCopy struct{}
+
+// Lock is a no-op used by -copylocks checker from `go vet`.
+func (*noCopy) Lock() {}

vendor/github.com/go-morph/morph/drivers/postgres/postgres.go

@@ -5,7 +5,6 @@ import (
 	"database/sql"
 	"fmt"
 	"strconv"
-	"time"
 
 	"github.com/pkg/errors"
 
@@ -17,9 +16,11 @@ import (
 var (
 	driverName    = "postgres"
 	defaultConfig = &Config{
-		MigrationsTable:        "db_migrations",
-		StatementTimeoutInSecs: 60,
-		MigrationMaxSize:       defaultMigrationMaxSize,
+		Config: drivers.Config{
+			MigrationsTable:        "db_migrations",
+			StatementTimeoutInSecs: 60,
+			MigrationMaxSize:       defaultMigrationMaxSize,
+		},
 	}
 	defaultMigrationMaxSize = 10 * 1 << 20 // 10 MB
 	configParams            = []string{
@@ -30,12 +31,10 @@ var (
 )
 
 type Config struct {
-	MigrationsTable        string
-	StatementTimeoutInSecs int
-	MigrationMaxSize       int
-	databaseName           string
-	schemaName             string
-	closeDBonClose         bool
+	drivers.Config
+	databaseName   string
+	schemaName     string
+	closeDBonClose bool
 }
 
 type postgres struct {
@@ -113,7 +112,7 @@ func Open(connURL string) (drivers.Driver, error) {
 func currentSchema(conn *sql.Conn, config *Config) (string, error) {
 	query := "SELECT CURRENT_SCHEMA()"
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(config.StatementTimeoutInSecs)
 	defer cancel()
 
 	var schemaName string
@@ -169,14 +168,14 @@ func mergeConfigs(config, defaultConfig *Config) *Config {
 }
 
 func (pg *postgres) Ping() error {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(pg.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	return pg.conn.PingContext(ctx)
 }
 
 func (pg *postgres) createSchemaTableIfNotExists() (err error) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(pg.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	createTableIfNotExistsQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (version bigint not null primary key, name varchar not null)", pg.config.MigrationsTable)
@@ -193,6 +192,10 @@ func (pg *postgres) createSchemaTableIfNotExists() (err error) {
 	return nil
 }
 
+func (postgres) DriverName() string {
+	return driverName
+}
+
 func (pg *postgres) Close() error {
 	if pg.conn != nil {
 		if err := pg.conn.Close(); err != nil {
@@ -231,7 +234,7 @@ func (pg *postgres) Lock() error {
 
 	// This will wait until the lock can be acquired or until the statement timeout has reached.
 	query := "SELECT pg_advisory_lock($1)"
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(pg.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	if _, err := pg.conn.ExecContext(ctx, query, aid); err != nil {
@@ -254,7 +257,7 @@ func (pg *postgres) Unlock() error {
 	}
 
 	query := "SELECT pg_advisory_unlock($1)"
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(pg.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	if _, err := pg.conn.ExecContext(ctx, query, aid); err != nil {
@@ -292,7 +295,7 @@ func (pg *postgres) Apply(migration *models.Migration, saveVersion bool) (err er
 	}
 	defer migration.Close()
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(pg.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
 	defer cancel()
 
 	transaction, err := pg.conn.BeginTx(ctx, nil)
@@ -353,7 +356,7 @@ func (pg *postgres) AppliedMigrations() (migrations []*models.Migration, err err
 	}
 
 	query := fmt.Sprintf("SELECT version, name FROM %s", pg.config.MigrationsTable)
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(pg.config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
 	defer cancel()
 	var appliedMigrations []*models.Migration
 	var version uint32
@@ -369,6 +372,7 @@ func (pg *postgres) AppliedMigrations() (migrations []*models.Migration, err err
 			Query:   []byte(query),
 		}
 	}
+	defer rows.Close()
 
 	for rows.Next() {
 		if err := rows.Scan(&version, &name); err != nil {
@@ -424,7 +428,7 @@ func executeQuery(transaction *sql.Tx, query string) error {
 func currentDatabaseNameFromDB(conn *sql.Conn, config *Config) (string, error) {
 	query := "SELECT CURRENT_DATABASE()"
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(config.StatementTimeoutInSecs)*time.Second)
+	ctx, cancel := drivers.GetContext(config.StatementTimeoutInSecs)
 	defer cancel()
 
 	var databaseName string

vendor/github.com/go-morph/morph/drivers/utils.go

@@ -1,10 +1,12 @@
 package drivers
 
 import (
+	"context"
 	"fmt"
 	"hash/crc32"
 	"regexp"
 	"strings"
+	"time"
 )
 
 func ExtractCustomParams(conn string, params []string) (map[string]string, error) {
@@ -21,7 +23,7 @@ func ExtractCustomParams(conn string, params []string) (map[string]string, error
 }
 
 func RemoveParamsFromURL(conn string, params []string) (string, error) {
-	prefixCorrection := regexp.MustCompile("\\?&+")
+	prefixCorrection := regexp.MustCompile(`\?&+`)
 	repeatedAmber := regexp.MustCompile("&+")
 
 	for _, param := range params {
@@ -47,3 +49,10 @@ func GenerateAdvisoryLockID(databaseName, schemaName string) (string, error) {
 	sum = sum * uint32(advisoryLockIDSalt)
 	return fmt.Sprint(sum), nil
 }
+
+func GetContext(timeoutInSeconds int) (context.Context, context.CancelFunc) {
+	if t := timeoutInSeconds; t > 0 {
+		return context.WithTimeout(context.Background(), time.Second*time.Duration(t))
+	}
+	return context.WithCancel(context.Background())
+}

vendor/github.com/go-morph/morph/morph.go

@@ -1,6 +1,7 @@
 package morph
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"log"
@@ -14,16 +15,13 @@ import (
 	"github.com/go-morph/morph/drivers"
 	"github.com/go-morph/morph/sources"
 
-	_ "github.com/go-morph/morph/drivers/mysql"
-	_ "github.com/go-morph/morph/drivers/postgres"
+	ms "github.com/go-morph/morph/drivers/mysql"
+	ps "github.com/go-morph/morph/drivers/postgres"
 
 	_ "github.com/go-morph/morph/sources/file"
 	_ "github.com/go-morph/morph/sources/go_bindata"
 )
 
-// DefaultLockTimeout sets the max time a database driver has to acquire a lock.
-var DefaultLockTimeout = 15 * time.Second
-
 var migrationProgressStart = "==  %s: migrating  ================================================="
 var migrationProgressFinished = "==  %s: migrated (%s)  ========================================"
 
@@ -33,18 +31,19 @@ type Morph struct {
 	config *Config
 	driver drivers.Driver
 	source sources.Source
+	mutex  drivers.Locker
 }
 
 type Config struct {
 	Logger      Logger
 	LockTimeout time.Duration
+	LockKey     string
 }
 
 type EngineOption func(*Morph)
 
 var defaultConfig = &Config{
-	LockTimeout: DefaultLockTimeout,
-	Logger:      log.New(os.Stderr, "", log.LstdFlags), // add default logger
+	Logger: log.New(os.Stderr, "", log.LstdFlags), // add default logger
 }
 
 func WithLogger(logger *log.Logger) EngineOption {
@@ -71,8 +70,17 @@ func SetSatementTimeoutInSeconds(n int) EngineOption {
 	}
 }
 
-// New creates a new instance of the migrations engine from an existing db instance and a migrations source
-func New(driver drivers.Driver, source sources.Source, options ...EngineOption) (*Morph, error) {
+// WithLock creates a lock table in the database so that the migrations are
+// guaranteed to be executed from a single instance. The key is used for naming
+// the mutex.
+func WithLock(key string) EngineOption {
+	return func(m *Morph) {
+		m.config.LockKey = key
+	}
+}
+
+// New creates a new instance of the migrations engine from an existing db instance and a migrations source.
+func New(ctx context.Context, driver drivers.Driver, source sources.Source, options ...EngineOption) (*Morph, error) {
 	engine := &Morph{
 		config: defaultConfig,
 		source: source,
@@ -87,11 +95,32 @@ func New(driver drivers.Driver, source sources.Source, options ...EngineOption)
 		return nil, err
 	}
 
+	if impl, ok := driver.(drivers.Lockable); ok && engine.config.LockKey != "" {
+		var mx drivers.Locker
+		var err error
+		switch impl.DriverName() {
+		case "mysql":
+			mx, err = ms.NewMutex(engine.config.LockKey, driver)
+		case "postgres":
+			mx, err = ps.NewMutex(engine.config.LockKey, driver)
+		}
+		if err != nil {
+			return nil, err
+		}
+
+		engine.mutex = mx
+		_ = mx.LockWithContext(ctx)
+	}
+
 	return engine, nil
 }
 
 // Close closes the underlying database connection of the engine.
 func (m *Morph) Close() error {
+	if m.mutex != nil {
+		m.mutex.Unlock()
+	}
+
 	return m.driver.Close()
 }
 
@@ -101,7 +130,7 @@ func (m *Morph) ApplyAll() error {
 	return err
 }
 
-// Applies limited number of migrations
+// Applies limited number of migrations upwards.
 func (m *Morph) Apply(limit int) (int, error) {
 	appliedMigrations, err := m.driver.AppliedMigrations()
 	if err != nil {
@@ -113,9 +142,14 @@ func (m *Morph) Apply(limit int) (int, error) {
 		return -1, err
 	}
 
-	migrations, rollbacks, err := findUpScripts(sortMigrations(pendingMigrations))
-	if err != nil {
-		return -1, err
+	migrations := make([]*models.Migration, 0)
+	sortedMigrations := sortMigrations(pendingMigrations)
+
+	for _, migration := range sortedMigrations {
+		if migration.Direction != models.Up {
+			continue
+		}
+		migrations = append(migrations, migration)
 	}
 
 	steps := limit
@@ -133,18 +167,6 @@ func (m *Morph) Apply(limit int) (int, error) {
 		migrationName := migrations[i].Name
 		m.config.Logger.Println(InfoLoggerLight.Sprint(formatProgress(fmt.Sprintf(migrationProgressStart, migrationName))))
 		if err := m.driver.Apply(migrations[i], true); err != nil {
-			rollback, ok := rollbacks[migrationName]
-			if ok {
-				m.config.Logger.Println(ErrorLoggerLight.Sprint(formatProgress(fmt.Sprintf("failed to apply %s, rolling back.", migrationName))))
-				m.config.Logger.Println(InfoLoggerLight.Sprint(formatProgress(fmt.Sprintf("trying to apply %s (%s)", rollback.Name, rollback.Direction))))
-
-				if err2 := m.driver.Apply(rollback, false); err2 != nil {
-					return applied, fmt.Errorf("could not rollback the migration %s: %w", migrationName, err)
-				}
-				m.config.Logger.Println(InfoLoggerLight.Sprint(formatProgress(fmt.Sprintf("rollback completed for %s. Aborting gracefully.", migrationName))))
-				return applied, err
-			}
-
 			return applied, err
 		}
 
@@ -166,6 +188,9 @@ func (m *Morph) ApplyDown(limit int) (int, error) {
 
 	sortedMigrations := reverseSortMigrations(appliedMigrations)
 	downMigrations, err := findDownScripts(sortedMigrations, m.source.Migrations())
+	if err != nil {
+		return -1, err
+	}
 
 	steps := limit
 	if len(sortedMigrations) < steps {
@@ -230,27 +255,6 @@ func computePendingMigrations(appliedMigrations []*models.Migration, sourceMigra
 	return pendingMigrations, nil
 }
 
-func findUpScripts(migrations []*models.Migration) ([]*models.Migration, map[string]*models.Migration, error) {
-	rollbackMigrations := make(map[string]*models.Migration)
-	toBeAppliedMigrations := make([]*models.Migration, 0)
-	for _, migration := range migrations {
-		if migration.Direction != models.Up {
-			rollbackMigrations[migration.Name] = migration
-			continue
-		}
-		toBeAppliedMigrations = append(toBeAppliedMigrations, migration)
-	}
-
-	for _, migration := range toBeAppliedMigrations {
-		_, ok := rollbackMigrations[migration.Name]
-		if !ok {
-			return nil, nil, fmt.Errorf("the rollback migration file for %s is missing", migration.RawName)
-		}
-	}
-
-	return toBeAppliedMigrations, rollbackMigrations, nil
-}
-
 func findDownScripts(appliedMigrations []*models.Migration, sourceMigrations []*models.Migration) (map[string]*models.Migration, error) {
 	tmp := make(map[string]*models.Migration)
 	for _, m := range sourceMigrations {

vendor/github.com/go-morph/morph/sources/file/file.go

@@ -70,7 +70,7 @@ func (f *File) readMigrations() error {
 	}
 
 	migrations := []*models.Migration{}
-	walkerr := filepath.Walk(f.path, func(path string, info os.FileInfo, err error) error {
+	walkerr := filepath.Walk(f.path, func(path string, info os.FileInfo, _ error) error {
 		if info.IsDir() {
 			return nil
 		}

vendor/modules.txt

@@ -214,7 +214,7 @@ github.com/gigawattio/window
 # github.com/go-asn1-ber/asn1-ber v1.5.3
 ## explicit
 github.com/go-asn1-ber/asn1-ber
-# github.com/go-morph/morph v0.0.0-20211202070306-dbdc0736c17e
+# github.com/go-morph/morph v0.2.1
 ## explicit
 github.com/go-morph/morph
 github.com/go-morph/morph/drivers