21007012926

Committed 14 Jan 2026 07:25PM UTC coverage: 54.586% (+0.2%) from 54.432%

Build # 21007012926

Build Type

Pull #1352

github

Committed by

cstoneham

Commit Message

more changes

Pull Request Pull Request #1352: use try advisory lock

Run Details

33 of 39 new or added lines in 1 file covered. (84.62%)

69 existing lines in 1 file now uncovered.

4410 of 8079 relevant lines covered (54.59%)

71.08 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.19

/database/postgres/postgres.go

//go:build go1.9

package postgres

import (
        "context"
        "database/sql"
        "errors"
        "fmt"
        "io"
        nurl "net/url"
        "regexp"
        "strconv"
        "strings"
        "sync/atomic"
        "time"

        "github.com/cenkalti/backoff/v4"
        "github.com/golang-migrate/migrate/v4"
        "github.com/golang-migrate/migrate/v4/database"
        "github.com/golang-migrate/migrate/v4/database/multistmt"
        "github.com/lib/pq"
)

func init() {
        db := Postgres{}
        database.Register("postgres", &db)
        database.Register("postgresql", &db)
}

var (
        multiStmtDelimiter = []byte(";")

        DefaultMigrationsTable       = "schema_migrations"
        DefaultMultiStatementMaxSize = 10 * 1 << 20 // 10 MB

        DefaultLockInitialRetryInterval = 100 * time.Millisecond
        DefaultLockMaxRetryInterval     = 1000 * time.Millisecond
)

var (
        ErrNilConfig      = fmt.Errorf("no config")
        ErrNoDatabaseName = fmt.Errorf("no database name")
        ErrNoSchema       = fmt.Errorf("no schema")
        ErrDatabaseDirty  = fmt.Errorf("database is dirty")
)

type Config struct {
        MigrationsTable       string
        MigrationsTableQuoted bool
        MultiStatementEnabled bool
        DatabaseName          string
        SchemaName            string
        migrationsSchemaName  string
        migrationsTableName   string
        StatementTimeout      time.Duration
        MultiStatementMaxSize int
        Locking               LockConfig
}

type LockConfig struct {
        // InitialRetryInterval the initial (minimum) retry interval used for exponential backoff
        // to try acquire a lock
        InitialRetryInterval time.Duration

        // MaxRetryInterval the maximum retry interval. Once the exponential backoff reaches this limit,
        // the retry interval remains the same
        MaxRetryInterval time.Duration
}

type Postgres struct {
        // Locking and unlocking need to use the same connection
        conn     *sql.Conn
        db       *sql.DB
        isLocked atomic.Bool

        // Open and WithInstance need to guarantee that config is never nil
        config *Config
}

func WithConnection(ctx context.Context, conn *sql.Conn, config *Config) (*Postgres, error) {
        if config == nil {
                return nil, ErrNilConfig
        }

        if err := conn.PingContext(ctx); err != nil {
                return nil, err
        }

        if config.DatabaseName == "" {
                query := `SELECT CURRENT_DATABASE()`
                var databaseName string
                if err := conn.QueryRowContext(ctx, query).Scan(&databaseName); err != nil {
                        return nil, &database.Error{OrigErr: err, Query: []byte(query)}
                }

                if len(databaseName) == 0 {
                        return nil, ErrNoDatabaseName
                }

                config.DatabaseName = databaseName
        }

        if config.SchemaName == "" {
                query := `SELECT CURRENT_SCHEMA()`
                var schemaName sql.NullString
                if err := conn.QueryRowContext(ctx, query).Scan(&schemaName); err != nil {
                        return nil, &database.Error{OrigErr: err, Query: []byte(query)}
                }

                if !schemaName.Valid {
                        return nil, ErrNoSchema
                }

                config.SchemaName = schemaName.String
        }

        if len(config.MigrationsTable) == 0 {
                config.MigrationsTable = DefaultMigrationsTable
        }

        config.migrationsSchemaName = config.SchemaName
        config.migrationsTableName = config.MigrationsTable
        if config.MigrationsTableQuoted {
                re := regexp.MustCompile(`"(.*?)"`)
                result := re.FindAllStringSubmatch(config.MigrationsTable, -1)
                config.migrationsTableName = result[len(result)-1][1]
                if len(result) == 2 {
                        config.migrationsSchemaName = result[0][1]
                } else if len(result) > 2 {
                        return nil, fmt.Errorf("\"%s\" MigrationsTable contains too many dot characters", config.MigrationsTable)
                }
        }

        px := &Postgres{
                conn:   conn,
                config: config,
        }

        if err := px.ensureVersionTable(); err != nil {
                return nil, err
        }

        return px, nil
}

func WithInstance(instance *sql.DB, config *Config) (database.Driver, error) {
        ctx := context.Background()

        if err := instance.Ping(); err != nil {
                return nil, err
        }

        conn, err := instance.Conn(ctx)
        if err != nil {
                return nil, err
        }

        px, err := WithConnection(ctx, conn, config)
        if err != nil {
                return nil, err
        }
        px.db = instance
        return px, nil
}

func (p *Postgres) Open(url string) (database.Driver, error) {
        purl, err := nurl.Parse(url)
        if err != nil {
                return nil, err
        }

        db, err := sql.Open("postgres", migrate.FilterCustomQuery(purl).String())
        if err != nil {
                return nil, err
        }

        migrationsTable := purl.Query().Get("x-migrations-table")
        migrationsTableQuoted := false
        if s := purl.Query().Get("x-migrations-table-quoted"); len(s) > 0 {
                migrationsTableQuoted, err = strconv.ParseBool(s)
                if err != nil {
                        return nil, fmt.Errorf("unable to parse option x-migrations-table-quoted: %w", err)
                }
        }
        if (len(migrationsTable) > 0) && (migrationsTableQuoted) && ((migrationsTable[0] != '"') || (migrationsTable[len(migrationsTable)-1] != '"')) {
                return nil, fmt.Errorf("x-migrations-table must be quoted (for instance '\"migrate\".\"schema_migrations\"') when x-migrations-table-quoted is enabled, current value is: %s", migrationsTable)
        }

        statementTimeoutString := purl.Query().Get("x-statement-timeout")
        statementTimeout := 0
        if statementTimeoutString != "" {
                statementTimeout, err = strconv.Atoi(statementTimeoutString)
                if err != nil {
                        return nil, err
                }
        }

        multiStatementMaxSize := DefaultMultiStatementMaxSize
        if s := purl.Query().Get("x-multi-statement-max-size"); len(s) > 0 {
                multiStatementMaxSize, err = strconv.Atoi(s)
                if err != nil {
                        return nil, err
                }
                if multiStatementMaxSize <= 0 {
                        multiStatementMaxSize = DefaultMultiStatementMaxSize
                }
        }

        multiStatementEnabled := false
        if s := purl.Query().Get("x-multi-statement"); len(s) > 0 {
                multiStatementEnabled, err = strconv.ParseBool(s)
                if err != nil {
                        return nil, fmt.Errorf("unable to parse option x-multi-statement: %w", err)
                }
        }

        lockConfig := LockConfig{
                InitialRetryInterval: DefaultLockInitialRetryInterval,
                MaxRetryInterval:     DefaultLockMaxRetryInterval,
        }
        if s := purl.Query().Get("x-lock-retry-max-interval"); len(s) > 0 {
                maxRetryIntervalMillis, err := strconv.Atoi(s)
                if err != nil {
                        return nil, fmt.Errorf("unable to parse option x-lock-retry-max-interval: %w", err)
                }
                maxRetryInterval := time.Duration(maxRetryIntervalMillis) * time.Millisecond
                if maxRetryInterval > DefaultLockInitialRetryInterval {
                        lockConfig.MaxRetryInterval = maxRetryInterval
                }
        }

        px, err := WithInstance(db, &Config{
                DatabaseName:          purl.Path,
                MigrationsTable:       migrationsTable,
                MigrationsTableQuoted: migrationsTableQuoted,
                StatementTimeout:      time.Duration(statementTimeout) * time.Millisecond,
                MultiStatementEnabled: multiStatementEnabled,
                MultiStatementMaxSize: multiStatementMaxSize,
                Locking:               lockConfig,
        })

        if err != nil {
                return nil, err
        }

        return px, nil
}

func (p *Postgres) Close() error {
        connErr := p.conn.Close()
        var dbErr error
        if p.db != nil {
                dbErr = p.db.Close()
        }

        if connErr != nil || dbErr != nil {
                return fmt.Errorf("conn: %v, db: %v", connErr, dbErr)
        }
        return nil
}

// Lock tries to acquire an advisory lock and retries indefinitely with an exponential backoff strategy
// https://www.postgresql.org/docs/9.6/static/explicit-locking.html#ADVISORY-LOCKS
func (p *Postgres) Lock() error {
        return database.CasRestoreOnErr(&p.isLocked, false, true, database.ErrLocked, func() error {
                backOff := p.config.Locking.nonStopBackoff()
                err := backoff.Retry(func() error {
                        ok, err := p.tryLock()
                        if err != nil {
                                return fmt.Errorf("p.tryLock: %w", err)
                        }

                        if ok {
                                return nil
                        }

                        return fmt.Errorf("could not acquire lock") // causes retry
                }, backOff)

                return err
        })
}

func (p *Postgres) tryLock() (bool, error) {
        aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)
        if err != nil {
                return false, err
        }

        // https://www.postgresql.org/docs/current/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS
        // should always return true or false
        query := `SELECT pg_try_advisory_lock($1)`
        var ok bool
        if err := p.conn.QueryRowContext(context.Background(), query, aid).Scan(&ok); err != nil {
                return false, &database.Error{OrigErr: err, Err: "pg_try_advisory_lock failed", Query: []byte(query)}
        }

        return ok, nil
}

func (p *Postgres) Unlock() error {
        return database.CasRestoreOnErr(&p.isLocked, true, false, database.ErrNotLocked, func() error {
                aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)
                if err != nil {
                        return err
                }

                query := `SELECT pg_advisory_unlock($1)`
                if _, err := p.conn.ExecContext(context.Background(), query, aid); err != nil {
                        return &database.Error{OrigErr: err, Query: []byte(query)}
                }
                return nil
        })
}

func (p *Postgres) Run(migration io.Reader) error {
        if p.config.MultiStatementEnabled {
                var err error
                if e := multistmt.Parse(migration, multiStmtDelimiter, p.config.MultiStatementMaxSize, func(m []byte) bool {
                        if err = p.runStatement(m); err != nil {
                                return false
                        }
                        return true
                }); e != nil {
                        return e
                }
                return err
        }
        migr, err := io.ReadAll(migration)
        if err != nil {
                return err
        }
        return p.runStatement(migr)
}

func (p *Postgres) runStatement(statement []byte) error {
        ctx := context.Background()
        if p.config.StatementTimeout != 0 {
                var cancel context.CancelFunc
                ctx, cancel = context.WithTimeout(ctx, p.config.StatementTimeout)
                defer cancel()
        }
        query := string(statement)
        if strings.TrimSpace(query) == "" {
                return nil
        }
        if _, err := p.conn.ExecContext(ctx, query); err != nil {
                if pgErr, ok := err.(*pq.Error); ok {
                        var line uint
                        var col uint
                        var lineColOK bool
                        if pgErr.Position != "" {
                                if pos, err := strconv.ParseUint(pgErr.Position, 10, 64); err == nil {
                                        line, col, lineColOK = computeLineFromPos(query, int(pos))
                                }
                        }
                        message := fmt.Sprintf("migration failed: %s", pgErr.Message)
                        if lineColOK {
                                message = fmt.Sprintf("%s (column %d)", message, col)
                        }
                        if pgErr.Detail != "" {
                                message = fmt.Sprintf("%s, %s", message, pgErr.Detail)
                        }
                        return database.Error{OrigErr: err, Err: message, Query: statement, Line: line}
                }
                return database.Error{OrigErr: err, Err: "migration failed", Query: statement}
        }
        return nil
}

func computeLineFromPos(s string, pos int) (line uint, col uint, ok bool) {
        // replace crlf with lf
        s = strings.ReplaceAll(s, "\r\n", "\n")
        // pg docs: pos uses index 1 for the first character, and positions are measured in characters not bytes
        runes := []rune(s)
        if pos > len(runes) {
                return 0, 0, false
        }
        sel := runes[:pos]
        line = uint(runesCount(sel, newLine) + 1)
        col = uint(pos - 1 - runesLastIndex(sel, newLine))
        return line, col, true
}

const newLine = '\n'

func runesCount(input []rune, target rune) int {
        var count int
        for _, r := range input {
                if r == target {
                        count++
                }
        }
        return count
}

func runesLastIndex(input []rune, target rune) int {
        for i := len(input) - 1; i >= 0; i-- {
                if input[i] == target {
                        return i
                }
        }
        return -1
}

func (p *Postgres) SetVersion(version int, dirty bool) error {
        tx, err := p.conn.BeginTx(context.Background(), &sql.TxOptions{})
        if err != nil {
                return &database.Error{OrigErr: err, Err: "transaction start failed"}
        }

        query := `TRUNCATE ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName)
        if _, err := tx.Exec(query); err != nil {
                if errRollback := tx.Rollback(); errRollback != nil {
                        err = errors.Join(err, errRollback)
                }
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        // Also re-write the schema version for nil dirty versions to prevent
        // empty schema version for failed down migration on the first migration
        // See: https://github.com/golang-migrate/migrate/issues/330
        if version >= 0 || (version == database.NilVersion && dirty) {
                query = `INSERT INTO ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version, dirty) VALUES ($1, $2)`
                if _, err := tx.Exec(query, version, dirty); err != nil {
                        if errRollback := tx.Rollback(); errRollback != nil {
                                err = errors.Join(err, errRollback)
                        }
                        return &database.Error{OrigErr: err, Query: []byte(query)}
                }
        }

        if err := tx.Commit(); err != nil {
                return &database.Error{OrigErr: err, Err: "transaction commit failed"}
        }

        return nil
}

func (p *Postgres) Version() (version int, dirty bool, err error) {
        query := `SELECT version, dirty FROM ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` LIMIT 1`
        err = p.conn.QueryRowContext(context.Background(), query).Scan(&version, &dirty)
        switch {
        case err == sql.ErrNoRows:
                return database.NilVersion, false, nil

        case err != nil:
                if e, ok := err.(*pq.Error); ok {
                        if e.Code.Name() == "undefined_table" {
                                return database.NilVersion, false, nil
                        }
                }
                return 0, false, &database.Error{OrigErr: err, Query: []byte(query)}

        default:
                return version, dirty, nil
        }
}

func (p *Postgres) Drop() (err error) {
        // select all tables in current schema
        query := `SELECT table_name FROM information_schema.tables WHERE table_schema=(SELECT current_schema()) AND table_type='BASE TABLE'`
        tables, err := p.conn.QueryContext(context.Background(), query)
        if err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }
        defer func() {
                if errClose := tables.Close(); errClose != nil {
                        err = errors.Join(err, errClose)
                }
        }()

        // delete one table after another
        tableNames := make([]string, 0)
        for tables.Next() {
                var tableName string
                if err := tables.Scan(&tableName); err != nil {
                        return err
                }
                if len(tableName) > 0 {
                        tableNames = append(tableNames, tableName)
                }
        }
        if err := tables.Err(); err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        if len(tableNames) > 0 {
                // delete one by one ...
                for _, t := range tableNames {
                        query = `DROP TABLE IF EXISTS ` + pq.QuoteIdentifier(t) + ` CASCADE`
                        if _, err := p.conn.ExecContext(context.Background(), query); err != nil {
                                return &database.Error{OrigErr: err, Query: []byte(query)}
                        }
                }
        }

        return nil
}

// ensureVersionTable checks if versions table exists and, if not, creates it.
// Note that this function locks the database, which deviates from the usual
// convention of "caller locks" in the Postgres type.
func (p *Postgres) ensureVersionTable() (err error) {
        if err = p.Lock(); err != nil {
                return err
        }

        defer func() {
                if e := p.Unlock(); e != nil {
                        err = errors.Join(err, e)
                }
        }()

        // This block checks whether the `MigrationsTable` already exists. This is useful because it allows read only postgres
        // users to also check the current version of the schema. Previously, even if `MigrationsTable` existed, the
        // `CREATE TABLE IF NOT EXISTS...` query would fail because the user does not have the CREATE permission.
        // Taken from https://github.com/mattes/migrate/blob/master/database/postgres/postgres.go#L258
        query := `SELECT COUNT(1) FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2 LIMIT 1`
        row := p.conn.QueryRowContext(context.Background(), query, p.config.migrationsSchemaName, p.config.migrationsTableName)

        var count int
        err = row.Scan(&count)
        if err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        if count == 1 {
                return nil
        }

        query = `CREATE TABLE IF NOT EXISTS ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version bigint not null primary key, dirty boolean not null)`
        if _, err = p.conn.ExecContext(context.Background(), query); err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        return nil
}

func (l *LockConfig) nonStopBackoff() backoff.BackOff {
        b := backoff.NewExponentialBackOff()
        b.InitialInterval = l.InitialRetryInterval
        b.MaxInterval = l.MaxRetryInterval
        b.MaxElapsedTime = 0 // this backoff won't stop
        b.Reset()

        return b
}

1	//go:build go1.9
2
3	package postgres
4
5	import (
6	"context"
7	"database/sql"
8	"errors"
9	"fmt"
10	"io"
11	nurl "net/url"
12	"regexp"
13	"strconv"
14	"strings"
15	"sync/atomic"
16	"time"
17
18	"github.com/cenkalti/backoff/v4"
19	"github.com/golang-migrate/migrate/v4"
20	"github.com/golang-migrate/migrate/v4/database"
21	"github.com/golang-migrate/migrate/v4/database/multistmt"
22	"github.com/lib/pq"
23	)
24
25	func init() {	2✔
26	db := Postgres{}	2✔
27	database.Register("postgres", &db)	2✔
28	database.Register("postgresql", &db)	2✔
29	}	2✔
30
31	var (
32	multiStmtDelimiter = []byte(";")
33
34	DefaultMigrationsTable = "schema_migrations"
35	DefaultMultiStatementMaxSize = 10 * 1 << 20 // 10 MB
36
37	DefaultLockInitialRetryInterval = 100 * time.Millisecond
38	DefaultLockMaxRetryInterval = 1000 * time.Millisecond
39	)
40
41	var (
42	ErrNilConfig = fmt.Errorf("no config")
43	ErrNoDatabaseName = fmt.Errorf("no database name")
44	ErrNoSchema = fmt.Errorf("no schema")
45	ErrDatabaseDirty = fmt.Errorf("database is dirty")
46	)
47
48	type Config struct {
49	MigrationsTable string
50	MigrationsTableQuoted bool
51	MultiStatementEnabled bool
52	DatabaseName string
53	SchemaName string
54	migrationsSchemaName string
55	migrationsTableName string
56	StatementTimeout time.Duration
57	MultiStatementMaxSize int
58	Locking LockConfig
59	}
60
61	type LockConfig struct {
62	// InitialRetryInterval the initial (minimum) retry interval used for exponential backoff
63	// to try acquire a lock
64	InitialRetryInterval time.Duration
65
66	// MaxRetryInterval the maximum retry interval. Once the exponential backoff reaches this limit,
67	// the retry interval remains the same
68	MaxRetryInterval time.Duration
69	}
70
71	type Postgres struct {
72	// Locking and unlocking need to use the same connection
73	conn *sql.Conn
74	db *sql.DB
75	isLocked atomic.Bool
76
77	// Open and WithInstance need to guarantee that config is never nil
78	config *Config
79	}
80
81	func WithConnection(ctx context.Context, conn sql.Conn, config Config) (*Postgres, error) {	560✔
82	if config == nil {	560✔
UNCOV 83	return nil, ErrNilConfig	×
UNCOV 84	}	×
85
86	if err := conn.PingContext(ctx); err != nil {	560✔
87	return nil, err	×
88	}	×
89
90	if config.DatabaseName == "" {	870✔
91	query := `SELECT CURRENT_DATABASE()`	310✔
92	var databaseName string	310✔
93	if err := conn.QueryRowContext(ctx, query).Scan(&databaseName); err != nil {	310✔
UNCOV 94	return nil, &database.Error{OrigErr: err, Query: []byte(query)}	×
UNCOV 95	}	×
96
97	if len(databaseName) == 0 {	310✔
98	return nil, ErrNoDatabaseName	×
99	}	×
100
101	config.DatabaseName = databaseName	310✔
102	}
103
104	if config.SchemaName == "" {	1,120✔
105	query := `SELECT CURRENT_SCHEMA()`	560✔
106	var schemaName sql.NullString	560✔
107	if err := conn.QueryRowContext(ctx, query).Scan(&schemaName); err != nil {	560✔
UNCOV 108	return nil, &database.Error{OrigErr: err, Query: []byte(query)}	×
UNCOV 109	}	×
110
111	if !schemaName.Valid {	560✔
112	return nil, ErrNoSchema	×
113	}	×
114
115	config.SchemaName = schemaName.String	560✔
116	}
117
118	if len(config.MigrationsTable) == 0 {	1,080✔
119	config.MigrationsTable = DefaultMigrationsTable	520✔
120	}	520✔
121
122	config.migrationsSchemaName = config.SchemaName	560✔
123	config.migrationsTableName = config.MigrationsTable	560✔
124	if config.MigrationsTableQuoted {	590✔
125	re := regexp.MustCompile(`"(.*?)"`)	30✔
126	result := re.FindAllStringSubmatch(config.MigrationsTable, -1)	30✔
127	config.migrationsTableName = result[len(result)-1][1]	30✔
128	if len(result) == 2 {	50✔
129	config.migrationsSchemaName = result[0][1]	20✔
130	} else if len(result) > 2 {	40✔
131	return nil, fmt.Errorf("\"%s\" MigrationsTable contains too many dot characters", config.MigrationsTable)	10✔
132	}	10✔
133	}
134
135	px := &Postgres{	550✔
136	conn: conn,	550✔
137	config: config,	550✔
138	}	550✔
139		550✔
140	if err := px.ensureVersionTable(); err != nil {	570✔
141	return nil, err	20✔
142	}	20✔
143
144	return px, nil	530✔
145	}
146
147	func WithInstance(instance sql.DB, config Config) (database.Driver, error) {	550✔
148	ctx := context.Background()	550✔
149		550✔
150	if err := instance.Ping(); err != nil {	550✔
UNCOV 151	return nil, err	×
UNCOV 152	}	×
153
154	conn, err := instance.Conn(ctx)	550✔
155	if err != nil {	550✔
156	return nil, err	×
UNCOV 157	}	×
158
159	px, err := WithConnection(ctx, conn, config)	550✔
160	if err != nil {	580✔
161	return nil, err	30✔
162	}	30✔
163	px.db = instance	520✔
164	return px, nil	520✔
165	}
166
167	func (p *Postgres) Open(url string) (database.Driver, error) {	260✔
168	purl, err := nurl.Parse(url)	260✔
169	if err != nil {	260✔
UNCOV 170	return nil, err	×
UNCOV 171	}	×
172
173	db, err := sql.Open("postgres", migrate.FilterCustomQuery(purl).String())	260✔
174	if err != nil {	260✔
175	return nil, err	×
UNCOV 176	}	×
177
178	migrationsTable := purl.Query().Get("x-migrations-table")	260✔
179	migrationsTableQuoted := false	260✔
180	if s := purl.Query().Get("x-migrations-table-quoted"); len(s) > 0 {	300✔
181	migrationsTableQuoted, err = strconv.ParseBool(s)	40✔
182	if err != nil {	40✔
UNCOV 183	return nil, fmt.Errorf("unable to parse option x-migrations-table-quoted: %w", err)	×
UNCOV 184	}	×
185	}
186	if (len(migrationsTable) > 0) && (migrationsTableQuoted) && ((migrationsTable[0] != '"') \|\| (migrationsTable[len(migrationsTable)-1] != '"')) {	270✔
187	return nil, fmt.Errorf("x-migrations-table must be quoted (for instance '\"migrate\".\"schema_migrations\"') when x-migrations-table-quoted is enabled, current value is: %s", migrationsTable)	10✔
188	}	10✔
189
190	statementTimeoutString := purl.Query().Get("x-statement-timeout")	250✔
191	statementTimeout := 0	250✔
192	if statementTimeoutString != "" {	250✔
UNCOV 193	statementTimeout, err = strconv.Atoi(statementTimeoutString)	×
UNCOV 194	if err != nil {	×
UNCOV 195	return nil, err	×
UNCOV 196	}	×
197	}
198
199	multiStatementMaxSize := DefaultMultiStatementMaxSize	250✔
200	if s := purl.Query().Get("x-multi-statement-max-size"); len(s) > 0 {	250✔
UNCOV 201	multiStatementMaxSize, err = strconv.Atoi(s)	×
UNCOV 202	if err != nil {	×
UNCOV 203	return nil, err	×
UNCOV 204	}	×
205	if multiStatementMaxSize <= 0 {	×
206	multiStatementMaxSize = DefaultMultiStatementMaxSize	×
207	}	×
208	}
209
210	multiStatementEnabled := false	250✔
211	if s := purl.Query().Get("x-multi-statement"); len(s) > 0 {	260✔
212	multiStatementEnabled, err = strconv.ParseBool(s)	10✔
213	if err != nil {	10✔
UNCOV 214	return nil, fmt.Errorf("unable to parse option x-multi-statement: %w", err)	×
UNCOV 215	}	×
216	}
217
218	lockConfig := LockConfig{	250✔
219	InitialRetryInterval: DefaultLockInitialRetryInterval,	250✔
220	MaxRetryInterval: DefaultLockMaxRetryInterval,	250✔
221	}	250✔
222	if s := purl.Query().Get("x-lock-retry-max-interval"); len(s) > 0 {	280✔
223	maxRetryIntervalMillis, err := strconv.Atoi(s)	30✔
224	if err != nil {	30✔
NEW 225	return nil, fmt.Errorf("unable to parse option x-lock-retry-max-interval: %w", err)	×
NEW 226	}	×
227	maxRetryInterval := time.Duration(maxRetryIntervalMillis) * time.Millisecond	30✔
228	if maxRetryInterval > DefaultLockInitialRetryInterval {	60✔
229	lockConfig.MaxRetryInterval = maxRetryInterval	30✔
230	}	30✔
231	}
232
233	px, err := WithInstance(db, &Config{	250✔
234	DatabaseName: purl.Path,	250✔
235	MigrationsTable: migrationsTable,	250✔
236	MigrationsTableQuoted: migrationsTableQuoted,	250✔
237	StatementTimeout: time.Duration(statementTimeout) * time.Millisecond,	250✔
238	MultiStatementEnabled: multiStatementEnabled,	250✔
239	MultiStatementMaxSize: multiStatementMaxSize,	250✔
240	Locking: lockConfig,	250✔
241	})	250✔
242		250✔
243	if err != nil {	280✔
244	return nil, err	30✔
245	}	30✔
246
247	return px, nil	220✔
248	}
249
250	func (p *Postgres) Close() error {	200✔
251	connErr := p.conn.Close()	200✔
252	var dbErr error	200✔
253	if p.db != nil {	390✔
254	dbErr = p.db.Close()	190✔
255	}	190✔
256
257	if connErr != nil \|\| dbErr != nil {	200✔
UNCOV 258	return fmt.Errorf("conn: %v, db: %v", connErr, dbErr)	×
UNCOV 259	}	×
260	return nil	200✔
261	}
262
263	// Lock tries to acquire an advisory lock and retries indefinitely with an exponential backoff strategy
264	// https://www.postgresql.org/docs/9.6/static/explicit-locking.html#ADVISORY-LOCKS
265	func (p *Postgres) Lock() error {	760✔
266	return database.CasRestoreOnErr(&p.isLocked, false, true, database.ErrLocked, func() error {	1,490✔
267	backOff := p.config.Locking.nonStopBackoff()	730✔
268	err := backoff.Retry(func() error {	14,064✔
269	ok, err := p.tryLock()	13,334✔
270	if err != nil {	13,334✔
UNCOV 271	return fmt.Errorf("p.tryLock: %w", err)	×
NEW 272	}	×
273
274	if ok {	14,064✔
275	return nil	730✔
276	}	730✔
277
278	return fmt.Errorf("could not acquire lock") // causes retry	12,604✔
279	}, backOff)
280
281	return err	730✔
282	})
283	}
284
285	func (p *Postgres) tryLock() (bool, error) {	13,334✔
286	aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)	13,334✔
287	if err != nil {	13,334✔
UNCOV 288	return false, err	×
UNCOV 289	}	×
290
291	// https://www.postgresql.org/docs/current/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS
292	// should always return true or false
293	query := `SELECT pg_try_advisory_lock($1)`	13,334✔
294	var ok bool	13,334✔
295	if err := p.conn.QueryRowContext(context.Background(), query, aid).Scan(&ok); err != nil {	13,334✔
NEW 296	return false, &database.Error{OrigErr: err, Err: "pg_try_advisory_lock failed", Query: []byte(query)}	×
NEW 297	}	×
298
299	return ok, nil	13,334✔
300	}
301
302	func (p *Postgres) Unlock() error {	730✔
303	return database.CasRestoreOnErr(&p.isLocked, true, false, database.ErrNotLocked, func() error {	1,460✔
304	aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)	730✔
305	if err != nil {	730✔
NEW 306	return err	×
UNCOV 307	}	×
308
309	query := `SELECT pg_advisory_unlock($1)`	730✔
310	if _, err := p.conn.ExecContext(context.Background(), query, aid); err != nil {	730✔
311	return &database.Error{OrigErr: err, Query: []byte(query)}	×
312	}	×
313	return nil	730✔
314	})
315	}
316
317	func (p *Postgres) Run(migration io.Reader) error {	570✔
318	if p.config.MultiStatementEnabled {	580✔
319	var err error	10✔
320	if e := multistmt.Parse(migration, multiStmtDelimiter, p.config.MultiStatementMaxSize, func(m []byte) bool {	30✔
321	if err = p.runStatement(m); err != nil {	20✔
UNCOV 322	return false	×
UNCOV 323	}	×
324	return true	20✔
UNCOV 325	}); e != nil {	×
UNCOV 326	return e	×
327	}	×
328	return err	10✔
329	}
330	migr, err := io.ReadAll(migration)	560✔
331	if err != nil {	560✔
332	return err	×
UNCOV 333	}	×
334	return p.runStatement(migr)	560✔
335	}
336
337	func (p *Postgres) runStatement(statement []byte) error {	580✔
338	ctx := context.Background()	580✔
339	if p.config.StatementTimeout != 0 {	580✔
UNCOV 340	var cancel context.CancelFunc	×
UNCOV 341	ctx, cancel = context.WithTimeout(ctx, p.config.StatementTimeout)	×
UNCOV 342	defer cancel()	×
UNCOV 343	}	×
344	query := string(statement)	580✔
345	if strings.TrimSpace(query) == "" {	580✔
346	return nil	×
347	}	×
348	if _, err := p.conn.ExecContext(ctx, query); err != nil {	590✔
349	if pgErr, ok := err.(*pq.Error); ok {	20✔
350	var line uint	10✔
351	var col uint	10✔
352	var lineColOK bool	10✔
353	if pgErr.Position != "" {	20✔
354	if pos, err := strconv.ParseUint(pgErr.Position, 10, 64); err == nil {	20✔
355	line, col, lineColOK = computeLineFromPos(query, int(pos))	10✔
356	}	10✔
357	}
358	message := fmt.Sprintf("migration failed: %s", pgErr.Message)	10✔
359	if lineColOK {	20✔
360	message = fmt.Sprintf("%s (column %d)", message, col)	10✔
361	}	10✔
362	if pgErr.Detail != "" {	10✔
UNCOV 363	message = fmt.Sprintf("%s, %s", message, pgErr.Detail)	×
UNCOV 364	}	×
365	return database.Error{OrigErr: err, Err: message, Query: statement, Line: line}	10✔
366	}
UNCOV 367	return database.Error{OrigErr: err, Err: "migration failed", Query: statement}	×
368	}
369	return nil	570✔
370	}
371
372	func computeLineFromPos(s string, pos int) (line uint, col uint, ok bool) {	74✔
373	// replace crlf with lf	74✔
374	s = strings.ReplaceAll(s, "\r\n", "\n")	74✔
375	// pg docs: pos uses index 1 for the first character, and positions are measured in characters not bytes	74✔
376	runes := []rune(s)	74✔
377	if pos > len(runes) {	82✔
378	return 0, 0, false	8✔
379	}	8✔
380	sel := runes[:pos]	66✔
381	line = uint(runesCount(sel, newLine) + 1)	66✔
382	col = uint(pos - 1 - runesLastIndex(sel, newLine))	66✔
383	return line, col, true	66✔
384	}
385
386	const newLine = '\n'
387
388	func runesCount(input []rune, target rune) int {	66✔
389	var count int	66✔
390	for _, r := range input {	1,404✔
391	if r == target {	1,442✔
392	count++	104✔
393	}	104✔
394	}
395	return count	66✔
396	}
397
398	func runesLastIndex(input []rune, target rune) int {	66✔
399	for i := len(input) - 1; i >= 0; i-- {	780✔
400	if input[i] == target {	770✔
401	return i	56✔
402	}	56✔
403	}
404	return -1	10✔
405	}
406
407	func (p *Postgres) SetVersion(version int, dirty bool) error {	920✔
408	tx, err := p.conn.BeginTx(context.Background(), &sql.TxOptions{})	920✔
409	if err != nil {	920✔
UNCOV 410	return &database.Error{OrigErr: err, Err: "transaction start failed"}	×
UNCOV 411	}	×
412
413	query := `TRUNCATE ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName)	920✔
414	if _, err := tx.Exec(query); err != nil {	920✔
415	if errRollback := tx.Rollback(); errRollback != nil {	×
416	err = errors.Join(err, errRollback)	×
UNCOV 417	}	×
UNCOV 418	return &database.Error{OrigErr: err, Query: []byte(query)}	×
419	}
420
421	// Also re-write the schema version for nil dirty versions to prevent
422	// empty schema version for failed down migration on the first migration
423	// See: https://github.com/golang-migrate/migrate/issues/330
424	if version >= 0 \|\| (version == database.NilVersion && dirty) {	1,810✔
425	query = `INSERT INTO ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version, dirty) VALUES ($1, $2)`	890✔
426	if _, err := tx.Exec(query, version, dirty); err != nil {	890✔
UNCOV 427	if errRollback := tx.Rollback(); errRollback != nil {	×
UNCOV 428	err = errors.Join(err, errRollback)	×
UNCOV 429	}	×
UNCOV 430	return &database.Error{OrigErr: err, Query: []byte(query)}	×
431	}
432	}
433
434	if err := tx.Commit(); err != nil {	920✔
435	return &database.Error{OrigErr: err, Err: "transaction commit failed"}	×
UNCOV 436	}	×
437
438	return nil	920✔
439	}
440
441	func (p *Postgres) Version() (version int, dirty bool, err error) {	290✔
442	query := `SELECT version, dirty FROM ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` LIMIT 1`	290✔
443	err = p.conn.QueryRowContext(context.Background(), query).Scan(&version, &dirty)	290✔
444	switch {	290✔
445	case err == sql.ErrNoRows:	120✔
446	return database.NilVersion, false, nil	120✔
447
UNCOV 448	case err != nil:	×
UNCOV 449	if e, ok := err.(*pq.Error); ok {	×
UNCOV 450	if e.Code.Name() == "undefined_table" {	×
UNCOV 451	return database.NilVersion, false, nil	×
UNCOV 452	}	×
453	}
454	return 0, false, &database.Error{OrigErr: err, Query: []byte(query)}	×
455
456	default:	170✔
457	return version, dirty, nil	170✔
458	}
459	}
460
461	func (p *Postgres) Drop() (err error) {	70✔
462	// select all tables in current schema	70✔
463	query := `SELECT table_name FROM information_schema.tables WHERE table_schema=(SELECT current_schema()) AND table_type='BASE TABLE'`	70✔
464	tables, err := p.conn.QueryContext(context.Background(), query)	70✔
465	if err != nil {	70✔
UNCOV 466	return &database.Error{OrigErr: err, Query: []byte(query)}	×
UNCOV 467	}	×
468	defer func() {	140✔
469	if errClose := tables.Close(); errClose != nil {	70✔
UNCOV 470	err = errors.Join(err, errClose)	×
471	}	×
472	}()
473
474	// delete one table after another
475	tableNames := make([]string, 0)	70✔
476	for tables.Next() {	260✔
477	var tableName string	190✔
478	if err := tables.Scan(&tableName); err != nil {	190✔
UNCOV 479	return err	×
UNCOV 480	}	×
481	if len(tableName) > 0 {	380✔
482	tableNames = append(tableNames, tableName)	190✔
483	}	190✔
484	}
485	if err := tables.Err(); err != nil {	70✔
UNCOV 486	return &database.Error{OrigErr: err, Query: []byte(query)}	×
UNCOV 487	}	×
488
489	if len(tableNames) > 0 {	140✔
490	// delete one by one ...	70✔
491	for _, t := range tableNames {	260✔
492	query = `DROP TABLE IF EXISTS ` + pq.QuoteIdentifier(t) + ` CASCADE`	190✔
493	if _, err := p.conn.ExecContext(context.Background(), query); err != nil {	190✔
UNCOV 494	return &database.Error{OrigErr: err, Query: []byte(query)}	×
UNCOV 495	}	×
496	}
497	}
498
499	return nil	70✔
500	}
501
502	// ensureVersionTable checks if versions table exists and, if not, creates it.
503	// Note that this function locks the database, which deviates from the usual
504	// convention of "caller locks" in the Postgres type.
505	func (p *Postgres) ensureVersionTable() (err error) {	550✔
506	if err = p.Lock(); err != nil {	550✔
UNCOV 507	return err	×
UNCOV 508	}	×
509
510	defer func() {	1,100✔
511	if e := p.Unlock(); e != nil {	550✔
512	err = errors.Join(err, e)	×
513	}	×
514	}()
515
516	// This block checks whether the `MigrationsTable` already exists. This is useful because it allows read only postgres
517	// users to also check the current version of the schema. Previously, even if `MigrationsTable` existed, the
518	// `CREATE TABLE IF NOT EXISTS...` query would fail because the user does not have the CREATE permission.
519	// Taken from https://github.com/mattes/migrate/blob/master/database/postgres/postgres.go#L258
520	query := `SELECT COUNT(1) FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2 LIMIT 1`	550✔
521	row := p.conn.QueryRowContext(context.Background(), query, p.config.migrationsSchemaName, p.config.migrationsTableName)	550✔
522		550✔
523	var count int	550✔
524	err = row.Scan(&count)	550✔
525	if err != nil {	550✔
UNCOV 526	return &database.Error{OrigErr: err, Query: []byte(query)}	×
UNCOV 527	}	×
528
529	if count == 1 {	850✔
530	return nil	300✔
531	}	300✔
532
533	query = `CREATE TABLE IF NOT EXISTS ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version bigint not null primary key, dirty boolean not null)`	250✔
534	if _, err = p.conn.ExecContext(context.Background(), query); err != nil {	270✔
535	return &database.Error{OrigErr: err, Query: []byte(query)}	20✔
536	}	20✔
537
538	return nil	230✔
539	}
540
541	func (l *LockConfig) nonStopBackoff() backoff.BackOff {	730✔
542	b := backoff.NewExponentialBackOff()	730✔
543	b.InitialInterval = l.InitialRetryInterval	730✔
544	b.MaxInterval = l.MaxRetryInterval	730✔
545	b.MaxElapsedTime = 0 // this backoff won't stop	730✔
546	b.Reset()	730✔
547		730✔
548	return b	730✔
549	}	730✔

golang-migrate / migrate / 21007012926

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous