Skip to content

fix(postgres,pgx): roll back aborted explicit transaction on migration error #1446

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cipherprofessor wants to merge 1 commit into
base: master
Choose a base branch
from
+218 −0
Open

fix(postgres,pgx): roll back aborted explicit transaction on migration error #1446

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions database/pgx/pgx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,6 +390,29 @@ func (p *Postgres) runStatement(statement []byte) error {
return nil
}
if _, err := p.conn.ExecContext(ctx, query); err != nil {
// An explicit BEGIN/COMMIT in the migration script (unlike the
// implicit transaction Postgres's simple query protocol wraps
// around it by default) is not automatically rolled back on
// error, leaving the connection in an aborted-transaction state
// that fails every subsequent statement -- including the
// advisory-unlock call issued after a failed run. Issuing a
// ROLLBACK here is a safe no-op when no explicit transaction was
// open (Postgres reports only a WARNING, not an error). See #581.
//
// Give the rollback its own fresh timeout window derived from the
// same configured StatementTimeout, rather than reusing ctx
// (which may already be expired -- e.g. if that's what caused
// the statement to fail) or leaving it unbounded.
rollbackCtx := context.Background()
if p.config.StatementTimeout != 0 {
var cancel context.CancelFunc
rollbackCtx, cancel = context.WithTimeout(rollbackCtx, p.config.StatementTimeout)
defer cancel()
}
var rollbackErr error
if _, rbErr := p.conn.ExecContext(rollbackCtx, "ROLLBACK"); rbErr != nil {
rollbackErr = rbErr
}

if pgErr, ok := err.(*pgconn.PgError); ok {
var line uint
Expand All @@ -403,8 +426,16 @@ func (p *Postgres) runStatement(statement []byte) error {
if pgErr.Detail != "" {
message = fmt.Sprintf("%s, %s", message, pgErr.Detail)
}
if rollbackErr != nil {
message = fmt.Sprintf("%s (additionally, rolling back the aborted transaction failed: %s)", message, rollbackErr)
return database.Error{OrigErr: errors.Join(err, rollbackErr), Err: message, Query: statement, Line: line}
}
return database.Error{OrigErr: err, Err: message, Query: statement, Line: line}
}
if rollbackErr != nil {
message := fmt.Sprintf("migration failed (additionally, rolling back the aborted transaction failed: %s)", rollbackErr)
return database.Error{OrigErr: errors.Join(err, rollbackErr), Err: message, Query: statement}
}
return database.Error{OrigErr: err, Err: "migration failed", Query: statement}
}
return nil
Expand Down
41 changes: 41 additions & 0 deletions database/pgx/pgx_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,47 @@ func TestMigrateLockTable(t *testing.T) {
})
}

// TestExplicitTransactionRolledBackOnError verifies #581: an explicit
// BEGIN/COMMIT migration block that errors before COMMIT must not leave the
// connection in Postgres's aborted-transaction state. Unlike an implicit
// (no BEGIN/COMMIT) migration, which Postgres's simple query protocol rolls
// back automatically on error, an explicit BEGIN requires an explicit
// ROLLBACK -- without it, every subsequent statement on that connection
// (including the advisory-unlock call migrate issues after a failed run)
// fails with "current transaction is aborted".
func TestExplicitTransactionRolledBackOnError(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ip, port, err := c.FirstPort()
if err != nil {
t.Fatal(err)
}

addr := pgConnectionString(ip, port)
p := &Postgres{}
d, err := p.Open(addr)
if err != nil {
t.Fatal(err)
}
defer func() {
if err := d.Close(); err != nil {
t.Error(err)
}
}()

if err := d.Run(strings.NewReader("BEGIN;\nSELECT 1/0;\nCOMMIT;")); err == nil {
t.Fatal("expected an error from the division by zero, got nil")
}

var one int
if err := d.(*Postgres).conn.QueryRowContext(context.Background(), "SELECT 1").Scan(&one); err != nil {
t.Fatalf("connection left in an unusable state after the failed explicit transaction: %v", err)
}
if one != 1 {
t.Fatalf("expected 1, got %d", one)
}
})
}

func TestMultipleStatements(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ip, port, err := c.FirstPort()
Expand Down
31 changes: 31 additions & 0 deletions database/pgx/v5/pgx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,6 +281,29 @@ func (p *Postgres) runStatement(statement []byte) error {
return nil
}
if _, err := p.conn.ExecContext(ctx, query); err != nil {
// An explicit BEGIN/COMMIT in the migration script (unlike the
// implicit transaction Postgres's simple query protocol wraps
// around it by default) is not automatically rolled back on
// error, leaving the connection in an aborted-transaction state
// that fails every subsequent statement -- including the
// advisory-unlock call issued after a failed run. Issuing a
// ROLLBACK here is a safe no-op when no explicit transaction was
// open (Postgres reports only a WARNING, not an error). See #581.
//
// Give the rollback its own fresh timeout window derived from the
// same configured StatementTimeout, rather than reusing ctx
// (which may already be expired -- e.g. if that's what caused
// the statement to fail) or leaving it unbounded.
rollbackCtx := context.Background()
if p.config.StatementTimeout != 0 {
var cancel context.CancelFunc
rollbackCtx, cancel = context.WithTimeout(rollbackCtx, p.config.StatementTimeout)
defer cancel()
}
var rollbackErr error
if _, rbErr := p.conn.ExecContext(rollbackCtx, "ROLLBACK"); rbErr != nil {
rollbackErr = rbErr
}

if pgErr, ok := err.(*pgconn.PgError); ok {
var line uint
Expand All @@ -294,8 +317,16 @@ func (p *Postgres) runStatement(statement []byte) error {
if pgErr.Detail != "" {
message = fmt.Sprintf("%s, %s", message, pgErr.Detail)
}
if rollbackErr != nil {
message = fmt.Sprintf("%s (additionally, rolling back the aborted transaction failed: %s)", message, rollbackErr)
return database.Error{OrigErr: errors.Join(err, rollbackErr), Err: message, Query: statement, Line: line}
}
return database.Error{OrigErr: err, Err: message, Query: statement, Line: line}
}
if rollbackErr != nil {
message := fmt.Sprintf("migration failed (additionally, rolling back the aborted transaction failed: %s)", rollbackErr)
return database.Error{OrigErr: errors.Join(err, rollbackErr), Err: message, Query: statement}
}
return database.Error{OrigErr: err, Err: "migration failed", Query: statement}
}
return nil
Expand Down
41 changes: 41 additions & 0 deletions database/pgx/v5/pgx_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,47 @@ func TestMigrate(t *testing.T) {
})
}

// TestExplicitTransactionRolledBackOnError verifies #581: an explicit
// BEGIN/COMMIT migration block that errors before COMMIT must not leave the
// connection in Postgres's aborted-transaction state. Unlike an implicit
// (no BEGIN/COMMIT) migration, which Postgres's simple query protocol rolls
// back automatically on error, an explicit BEGIN requires an explicit
// ROLLBACK -- without it, every subsequent statement on that connection
// (including the advisory-unlock call migrate issues after a failed run)
// fails with "current transaction is aborted".
func TestExplicitTransactionRolledBackOnError(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ip, port, err := c.FirstPort()
if err != nil {
t.Fatal(err)
}

addr := pgConnectionString(ip, port)
p := &Postgres{}
d, err := p.Open(addr)
if err != nil {
t.Fatal(err)
}
defer func() {
if err := d.Close(); err != nil {
t.Error(err)
}
}()

if err := d.Run(strings.NewReader("BEGIN;\nSELECT 1/0;\nCOMMIT;")); err == nil {
t.Fatal("expected an error from the division by zero, got nil")
}

var one int
if err := d.(*Postgres).conn.QueryRowContext(context.Background(), "SELECT 1").Scan(&one); err != nil {
t.Fatalf("connection left in an unusable state after the failed explicit transaction: %v", err)
}
if one != 1 {
t.Fatalf("expected 1, got %d", one)
}
})
}

func TestMultipleStatements(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ip, port, err := c.FirstPort()
Expand Down
32 changes: 32 additions & 0 deletions database/postgres/postgres.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,6 +294,30 @@ func (p *Postgres) runStatement(statement []byte) error {
return nil
}
if _, err := p.conn.ExecContext(ctx, query); err != nil {
// An explicit BEGIN/COMMIT in the migration script (unlike the
// implicit transaction Postgres's simple query protocol wraps
// around it by default) is not automatically rolled back on
// error, leaving the connection in an aborted-transaction state
// that fails every subsequent statement -- including the
// advisory-unlock call issued after a failed run. Issuing a
// ROLLBACK here is a safe no-op when no explicit transaction was
// open (Postgres reports only a WARNING, not an error). See #581.
//
// Give the rollback its own fresh timeout window derived from the
// same configured StatementTimeout, rather than reusing ctx
// (which may already be expired -- e.g. if that's what caused
// the statement to fail) or leaving it unbounded.
rollbackCtx := context.Background()
if p.config.StatementTimeout != 0 {
var cancel context.CancelFunc
rollbackCtx, cancel = context.WithTimeout(rollbackCtx, p.config.StatementTimeout)
defer cancel()
}
var rollbackErr error
if _, rbErr := p.conn.ExecContext(rollbackCtx, "ROLLBACK"); rbErr != nil {
rollbackErr = rbErr
}

if pgErr, ok := err.(*pq.Error); ok {
var line uint
var col uint
Expand All @@ -310,8 +334,16 @@ func (p *Postgres) runStatement(statement []byte) error {
if pgErr.Detail != "" {
message = fmt.Sprintf("%s, %s", message, pgErr.Detail)
}
if rollbackErr != nil {
message = fmt.Sprintf("%s (additionally, rolling back the aborted transaction failed: %s)", message, rollbackErr)
return database.Error{OrigErr: errors.Join(err, rollbackErr), Err: message, Query: statement, Line: line}
}
return database.Error{OrigErr: err, Err: message, Query: statement, Line: line}
}
if rollbackErr != nil {
message := fmt.Sprintf("migration failed (additionally, rolling back the aborted transaction failed: %s)", rollbackErr)
return database.Error{OrigErr: errors.Join(err, rollbackErr), Err: message, Query: statement}
}
return database.Error{OrigErr: err, Err: "migration failed", Query: statement}
}
return nil
Expand Down
42 changes: 42 additions & 0 deletions database/postgres/postgres_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ func Test(t *testing.T) {
t.Run("testPostgresLock", testPostgresLock)
t.Run("testWithInstanceConcurrent", testWithInstanceConcurrent)
t.Run("testWithConnection", testWithConnection)
t.Run("testExplicitTransactionRolledBackOnError", testExplicitTransactionRolledBackOnError)

t.Cleanup(func() {
for _, spec := range specs {
Expand Down Expand Up @@ -158,6 +159,47 @@ func testMigrate(t *testing.T) {
})
}

// testExplicitTransactionRolledBackOnError verifies #581: an explicit
// BEGIN/COMMIT migration block that errors before COMMIT must not leave the
// connection in Postgres's aborted-transaction state. Unlike an implicit
// (no BEGIN/COMMIT) migration, which Postgres's simple query protocol rolls
// back automatically on error, an explicit BEGIN requires an explicit
// ROLLBACK -- without it, every subsequent statement on that connection
// (including the advisory-unlock call migrate issues after a failed run)
// fails with "current transaction is aborted".
func testExplicitTransactionRolledBackOnError(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ip, port, err := c.FirstPort()
if err != nil {
t.Fatal(err)
}

addr := pgConnectionString(ip, port)
p := &Postgres{}
d, err := p.Open(addr)
if err != nil {
t.Fatal(err)
}
defer func() {
if err := d.Close(); err != nil {
t.Error(err)
}
}()

if err := d.Run(strings.NewReader("BEGIN;\nSELECT 1/0;\nCOMMIT;")); err == nil {
t.Fatal("expected an error from the division by zero, got nil")
}

var one int
if err := d.(*Postgres).conn.QueryRowContext(context.Background(), "SELECT 1").Scan(&one); err != nil {
t.Fatalf("connection left in an unusable state after the failed explicit transaction: %v", err)
}
if one != 1 {
t.Fatalf("expected 1, got %d", one)
}
})
}

func testMultipleStatements(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ip, port, err := c.FirstPort()
Expand Down
Loading