diff --git a/tx.go b/tx.go index 9e7fa9d4..4183f403 100644 --- a/tx.go +++ b/tx.go @@ -13,8 +13,21 @@ const ( ReadUncommitted = "read uncommitted" ) +const ( + TxStatusInProgress = 0 + TxStatusCommitFailure = -1 + TxStatusRollbackFailure = -2 + TxStatusCommitSuccess = 1 + TxStatusRollbackSuccess = 2 +) + var ErrTxClosed = errors.New("tx is closed") +// ErrTxCommitRollback occurs when an error has occurred in a transaction and +// Commit() is called. PostgreSQL accepts COMMIT on aborted transactions, but +// it is treated as ROLLBACK. +var ErrTxCommitRollback = errors.New("commit unexpectedly resulted in rollback") + // Begin starts a transaction with the default isolation level for the current // connection. To use a specific isolation level see BeginIso. func (c *Conn) Begin() (*Tx, error) { @@ -56,18 +69,30 @@ func (c *Conn) begin(isoLevel string) (*Tx, error) { type Tx struct { conn *Conn afterClose func(*Tx) - closed bool + err error + status int8 } // Commit commits the transaction func (tx *Tx) Commit() error { - if tx.closed { + if tx.status != TxStatusInProgress { return ErrTxClosed } - _, err := tx.conn.Exec("commit") - tx.close() - return err + commandTag, err := tx.conn.Exec("commit") + if err == nil && commandTag == "COMMIT" { + tx.status = TxStatusCommitSuccess + } else if err == nil && commandTag == "ROLLBACK" { + tx.status = TxStatusCommitFailure + tx.err = ErrTxCommitRollback + } else { + tx.err = err + } + + if tx.afterClose != nil { + tx.afterClose(tx) + } + return tx.err } // Rollback rolls back the transaction. Rollback will return ErrTxClosed if the @@ -75,25 +100,26 @@ func (tx *Tx) Commit() error { // defer tx.Rollback() is safe even if tx.Commit() will be called first in a // non-error condition. func (tx *Tx) Rollback() error { - if tx.closed { + if tx.status != TxStatusInProgress { return ErrTxClosed } - _, err := tx.conn.Exec("rollback") - tx.close() - return err -} + _, tx.err = tx.conn.Exec("rollback") + if tx.err == nil { + tx.status = TxStatusRollbackSuccess + } else { + tx.status = TxStatusRollbackFailure + } -func (tx *Tx) close() { if tx.afterClose != nil { tx.afterClose(tx) } - tx.closed = true + return tx.err } // Exec delegates to the underlying *Conn func (tx *Tx) Exec(sql string, arguments ...interface{}) (commandTag CommandTag, err error) { - if tx.closed { + if tx.status != TxStatusInProgress { return CommandTag(""), ErrTxClosed } @@ -102,7 +128,7 @@ func (tx *Tx) Exec(sql string, arguments ...interface{}) (commandTag CommandTag, // Query delegates to the underlying *Conn func (tx *Tx) Query(sql string, args ...interface{}) (*Rows, error) { - if tx.closed { + if tx.status != TxStatusInProgress { // Because checking for errors can be deferred to the *Rows, build one with the error err := ErrTxClosed return &Rows{closed: true, err: err}, err @@ -122,6 +148,17 @@ func (tx *Tx) Conn() *Conn { return tx.conn } +// Status returns the status of the transaction from the set of +// pgx.TxStatus* constants. +func (tx *Tx) Status() int8 { + return tx.status +} + +// Err returns the final error state, if any, of calling Commit or Rollback. +func (tx *Tx) Err() error { + return tx.err +} + // AfterClose adds f to a LILO queue of functions that will be called when // the transaction is closed (either Commit or Rollback). func (tx *Tx) AfterClose(f func(*Tx)) { diff --git a/tx_test.go b/tx_test.go index bc8faf5a..c0019970 100644 --- a/tx_test.go +++ b/tx_test.go @@ -48,6 +48,52 @@ func TestTransactionSuccessfulCommit(t *testing.T) { } } +func TestTxCommitWhenTxBroken(t *testing.T) { + t.Parallel() + + conn := mustConnect(t, *defaultConnConfig) + defer closeConn(t, conn) + + createSql := ` + create temporary table foo( + id integer, + unique (id) initially deferred + ); + ` + + if _, err := conn.Exec(createSql); err != nil { + t.Fatalf("Failed to create table: %v", err) + } + + tx, err := conn.Begin() + if err != nil { + t.Fatalf("conn.Begin failed: %v", err) + } + + if _, err := tx.Exec("insert into foo(id) values (1)"); err != nil { + t.Fatalf("tx.Exec failed: %v", err) + } + + // Purposely break transaction + if _, err := tx.Exec("syntax error"); err == nil { + t.Fatal("Unexpected success") + } + + err = tx.Commit() + if err != pgx.ErrTxCommitRollback { + t.Fatalf("Expected error %v, got %v", pgx.ErrTxCommitRollback, err) + } + + var n int64 + err = conn.QueryRow("select count(*) from foo").Scan(&n) + if err != nil { + t.Fatalf("QueryRow Scan failed: %v", err) + } + if n != 0 { + t.Fatalf("Did not receive correct number of rows: %v", n) + } +} + func TestTransactionSuccessfulRollback(t *testing.T) { t.Parallel() @@ -150,3 +196,56 @@ func TestTxAfterClose(t *testing.T) { t.Errorf("AfterClose callbacks called out of order: %v, %v", t1, t2) } } + +func TestTxStatus(t *testing.T) { + t.Parallel() + + conn := mustConnect(t, *defaultConnConfig) + defer closeConn(t, conn) + + tx, err := conn.Begin() + if err != nil { + t.Fatal(err) + } + + if status := tx.Status(); status != pgx.TxStatusInProgress { + t.Fatalf("Expected status to be %v, but it was %v", pgx.TxStatusInProgress, status) + } + + if err := tx.Rollback(); err != nil { + t.Fatal(err) + } + + if status := tx.Status(); status != pgx.TxStatusRollbackSuccess { + t.Fatalf("Expected status to be %v, but it was %v", pgx.TxStatusRollbackSuccess, status) + } +} + +func TestTxErr(t *testing.T) { + t.Parallel() + + conn := mustConnect(t, *defaultConnConfig) + defer closeConn(t, conn) + + tx, err := conn.Begin() + if err != nil { + t.Fatal(err) + } + + // Purposely break transaction + if _, err := tx.Exec("syntax error"); err == nil { + t.Fatal("Unexpected success") + } + + if err := tx.Commit(); err != pgx.ErrTxCommitRollback { + t.Fatalf("Expected error %v, got %v", pgx.ErrTxCommitRollback, err) + } + + if status := tx.Status(); status != pgx.TxStatusCommitFailure { + t.Fatalf("Expected status to be %v, but it was %v", pgx.TxStatusRollbackSuccess, status) + } + + if err := tx.Err(); err != pgx.ErrTxCommitRollback { + t.Fatalf("Expected error %v, got %v", pgx.ErrTxCommitRollback, err) + } +}