diff --git a/database/pgx/pgx.go b/database/pgx/pgx.go index e97f15e9f..00b203336 100644 --- a/database/pgx/pgx.go +++ b/database/pgx/pgx.go @@ -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 @@ -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 diff --git a/database/pgx/pgx_test.go b/database/pgx/pgx_test.go index ca04068f1..ae143f7ac 100644 --- a/database/pgx/pgx_test.go +++ b/database/pgx/pgx_test.go @@ -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() diff --git a/database/pgx/v5/pgx.go b/database/pgx/v5/pgx.go index ef5f05674..eb5f1c76f 100644 --- a/database/pgx/v5/pgx.go +++ b/database/pgx/v5/pgx.go @@ -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 @@ -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 diff --git a/database/pgx/v5/pgx_test.go b/database/pgx/v5/pgx_test.go index 6a933f735..fb9e57385 100644 --- a/database/pgx/v5/pgx_test.go +++ b/database/pgx/v5/pgx_test.go @@ -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() diff --git a/database/postgres/postgres.go b/database/postgres/postgres.go index ed96fe63e..120a44fec 100644 --- a/database/postgres/postgres.go +++ b/database/postgres/postgres.go @@ -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 @@ -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 diff --git a/database/postgres/postgres_test.go b/database/postgres/postgres_test.go index 50329ca4a..ec49371e3 100644 --- a/database/postgres/postgres_test.go +++ b/database/postgres/postgres_test.go @@ -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 { @@ -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()