diff --git a/go.mod b/go.mod index 305ac4f93a..c700e4bc3c 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 3dea1e0591..3e9f15f959 100644 --- a/go.sum +++ b/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= diff --git a/i18n/en.json b/i18n/en.json index 2a57301985..b1d3fa787d 100644 --- a/i18n/en.json +++ b/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." diff --git a/model/config.go b/model/config.go index 7701ff038c..54ed41c68f 100644 --- a/model/config.go +++ b/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 } diff --git a/store/sqlstore/store.go b/store/sqlstore/store.go index 4ad2deef77..a48c329121 100644 --- a/store/sqlstore/store.go +++ b/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 } diff --git a/vendor/github.com/go-morph/morph/README.md b/vendor/github.com/go-morph/morph/README.md index e5b0c4a106..709d7f7322 100644 --- a/vendor/github.com/go-morph/morph/README.md +++ b/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 } diff --git a/vendor/github.com/go-morph/morph/drivers/driver.go b/vendor/github.com/go-morph/morph/drivers/driver.go index 0d8a481154..d162031de6 100644 --- a/vendor/github.com/go-morph/morph/drivers/driver.go +++ b/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 diff --git a/vendor/github.com/go-morph/morph/drivers/lock.go b/vendor/github.com/go-morph/morph/drivers/lock.go new file mode 100644 index 0000000000..9a80480fc9 --- /dev/null +++ b/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 +} diff --git a/vendor/github.com/go-morph/morph/drivers/mysql/lock.go b/vendor/github.com/go-morph/morph/drivers/mysql/lock.go new file mode 100644 index 0000000000..4e55b8cfdc --- /dev/null +++ b/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() {} diff --git a/vendor/github.com/go-morph/morph/drivers/mysql/mysql.go b/vendor/github.com/go-morph/morph/drivers/mysql/mysql.go index 6cb1b1f365..d1d7e1deb3 100644 --- a/vendor/github.com/go-morph/morph/drivers/mysql/mysql.go +++ b/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 diff --git a/vendor/github.com/go-morph/morph/drivers/postgres/lock.go b/vendor/github.com/go-morph/morph/drivers/postgres/lock.go new file mode 100644 index 0000000000..6676a6f519 --- /dev/null +++ b/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() {} diff --git a/vendor/github.com/go-morph/morph/drivers/postgres/postgres.go b/vendor/github.com/go-morph/morph/drivers/postgres/postgres.go index 33e0c64e89..5aab6847d7 100644 --- a/vendor/github.com/go-morph/morph/drivers/postgres/postgres.go +++ b/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 diff --git a/vendor/github.com/go-morph/morph/drivers/utils.go b/vendor/github.com/go-morph/morph/drivers/utils.go index f09c094e08..166d1afca6 100644 --- a/vendor/github.com/go-morph/morph/drivers/utils.go +++ b/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()) +} diff --git a/vendor/github.com/go-morph/morph/morph.go b/vendor/github.com/go-morph/morph/morph.go index c066ac6ff6..a8837a11ee 100644 --- a/vendor/github.com/go-morph/morph/morph.go +++ b/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 { diff --git a/vendor/github.com/go-morph/morph/sources/file/file.go b/vendor/github.com/go-morph/morph/sources/file/file.go index 603c656765..2785bcd942 100644 --- a/vendor/github.com/go-morph/morph/sources/file/file.go +++ b/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 } diff --git a/vendor/modules.txt b/vendor/modules.txt index 61d710e1b8..f28999d747 100644 --- a/vendor/modules.txt +++ b/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