Source: tx.go in package github.com/jackc/pgx/v4

package pgx

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"strconv"

	"github.com/jackc/pgconn"
)

type TxIsoLevel string

Transaction isolation levels

const (
	Serializable    = TxIsoLevel("serializable")
	RepeatableRead  = TxIsoLevel("repeatable read")
	ReadCommitted   = TxIsoLevel("read committed")
	ReadUncommitted = TxIsoLevel("read uncommitted")
)

type TxAccessMode string

Transaction access modes

const (
	ReadWrite = TxAccessMode("read write")
	ReadOnly  = TxAccessMode("read only")
)

type TxDeferrableMode string

Transaction deferrable modes

const (
	Deferrable    = TxDeferrableMode("deferrable")
	NotDeferrable = TxDeferrableMode("not deferrable")
)

type TxOptions struct {
	IsoLevel       TxIsoLevel
	AccessMode     TxAccessMode
	DeferrableMode TxDeferrableMode
}

func (txOptions TxOptions) beginSQL() string {
	buf := &bytes.Buffer{}
	buf.WriteString("begin")
	if txOptions.IsoLevel != "" {
		fmt.Fprintf(buf, " isolation level %s", txOptions.IsoLevel)
	}
	if txOptions.AccessMode != "" {
		fmt.Fprintf(buf, " %s", txOptions.AccessMode)
	}
	if txOptions.DeferrableMode != "" {
		fmt.Fprintf(buf, " %s", txOptions.DeferrableMode)
	}

	return buf.String()
}

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. Unlike database/sql, the context only affects the begin command. i.e. there is no auto-rollback on context cancellation.

func (c *Conn) Begin(ctx context.Context) (Tx, error) {
	return c.BeginTx(ctx, TxOptions{})
}

BeginTx starts a transaction with txOptions determining the transaction mode. Unlike database/sql, the context only affects the begin command. i.e. there is no auto-rollback on context cancellation.

func (c *Conn) BeginTx(ctx context.Context, txOptions TxOptions) (Tx, error) {
	_, err := c.Exec(ctx, txOptions.beginSQL())

begin should never fail unless there is an underlying connection issue or a context timeout. In either case, the connection is possibly broken.

		c.die(errors.New("failed to begin transaction"))
		return nil, err
	}

	return &dbTx{conn: c}, nil
}

BeginFunc starts a transaction and calls f. If f does not return an error the transaction is committed. If f returns an error the transaction is rolled back. The context will be used when executing the transaction control statements (BEGIN, ROLLBACK, and COMMIT) but does not otherwise affect the execution of f.

func (c *Conn) BeginFunc(ctx context.Context, f func(Tx) error) (err error) {
	return c.BeginTxFunc(ctx, TxOptions{}, f)
}

BeginTxFunc starts a transaction with txOptions determining the transaction mode and calls f. If f does not return an error the transaction is committed. If f returns an error the transaction is rolled back. The context will be used when executing the transaction control statements (BEGIN, ROLLBACK, and COMMIT) but does not otherwise affect the execution of f.

func (c *Conn) BeginTxFunc(ctx context.Context, txOptions TxOptions, f func(Tx) error) (err error) {
	var tx Tx
	tx, err = c.BeginTx(ctx, txOptions)
	if err != nil {
		return err
	}
	defer func() {
		rollbackErr := tx.Rollback(ctx)
		if !(rollbackErr == nil || errors.Is(rollbackErr, ErrTxClosed)) {
			err = rollbackErr
		}
	}()

	fErr := f(tx)
	if fErr != nil {
		_ = tx.Rollback(ctx) // ignore rollback error as there is already an error to return
		return fErr
	}

	return tx.Commit(ctx)
}

Tx represents a database transaction. Tx is an interface instead of a struct to enable connection pools to be implemented without relying on internal pgx state, to support pseudo-nested transactions with savepoints, and to allow tests to mock transactions. However, adding a method to an interface is technically a breaking change. If new methods are added to Conn it may be desirable to add them to Tx as well. Because of this the Tx interface is partially excluded from semantic version requirements. Methods will not be removed or changed, but new methods may be added.

Begin starts a pseudo nested transaction.

	Begin(ctx context.Context) (Tx, error)

BeginFunc starts a pseudo nested transaction and executes f. If f does not return an err the pseudo nested transaction will be committed. If it does then it will be rolled back.

	BeginFunc(ctx context.Context, f func(Tx) error) (err error)

Commit commits the transaction if this is a real transaction or releases the savepoint if this is a pseudo nested transaction. Commit will return ErrTxClosed if the Tx is already closed, but is otherwise safe to call multiple times. If the commit fails with a rollback status (e.g. the transaction was already in a broken state) then ErrTxCommitRollback will be returned.

	Commit(ctx context.Context) error

Rollback rolls back the transaction if this is a real transaction or rolls back to the savepoint if this is a pseudo nested transaction. Rollback will return ErrTxClosed if the Tx is already closed, but is otherwise safe to call multiple times. Hence, a defer tx.Rollback() is safe even if tx.Commit() will be called first in a non-error condition. Any other failure of a real transaction will result in the connection being closed.

	Rollback(ctx context.Context) error

	CopyFrom(ctx context.Context, tableName Identifier, columnNames []string, rowSrc CopyFromSource) (int64, error)
	SendBatch(ctx context.Context, b *Batch) BatchResults
	LargeObjects() LargeObjects

	Prepare(ctx context.Context, name, sql string) (*pgconn.StatementDescription, error)

	Exec(ctx context.Context, sql string, arguments ...interface{}) (commandTag pgconn.CommandTag, err error)
	Query(ctx context.Context, sql string, args ...interface{}) (Rows, error)
	QueryRow(ctx context.Context, sql string, args ...interface{}) Row
	QueryFunc(ctx context.Context, sql string, args []interface{}, scans []interface{}, f func(QueryFuncRow) error) (pgconn.CommandTag, error)

Conn returns the underlying *Conn that on which this transaction is executing.

	Conn() *Conn
}

dbTx represents a database transaction. All dbTx methods return ErrTxClosed if Commit or Rollback has already been called on the dbTx.

type dbTx struct {
	conn         *Conn
	err          error
	savepointNum int64
	closed       bool
}

Begin starts a pseudo nested transaction implemented with a savepoint.

func (tx *dbTx) Begin(ctx context.Context) (Tx, error) {
	if tx.closed {
		return nil, ErrTxClosed
	}

	tx.savepointNum++
	_, err := tx.conn.Exec(ctx, "savepoint sp_"+strconv.FormatInt(tx.savepointNum, 10))
	if err != nil {
		return nil, err
	}

	return &dbSavepoint{tx: tx, savepointNum: tx.savepointNum}, nil
}

func (tx *dbTx) BeginFunc(ctx context.Context, f func(Tx) error) (err error) {
	if tx.closed {
		return ErrTxClosed
	}

	var savepoint Tx
	savepoint, err = tx.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() {
		rollbackErr := savepoint.Rollback(ctx)
		if !(rollbackErr == nil || errors.Is(rollbackErr, ErrTxClosed)) {
			err = rollbackErr
		}
	}()

	fErr := f(savepoint)
	if fErr != nil {
		_ = savepoint.Rollback(ctx) // ignore rollback error as there is already an error to return
		return fErr
	}

	return savepoint.Commit(ctx)
}

Commit commits the transaction.

func (tx *dbTx) Commit(ctx context.Context) error {
	if tx.closed {
		return ErrTxClosed
	}

	commandTag, err := tx.conn.Exec(ctx, "commit")
	tx.closed = true
	if err != nil {
		if tx.conn.PgConn().TxStatus() != 'I' {
			_ = tx.conn.Close(ctx) // already have error to return
		}
		return err
	}
	if string(commandTag) == "ROLLBACK" {
		return ErrTxCommitRollback
	}

	return nil
}

Rollback rolls back the transaction. Rollback will return ErrTxClosed if the Tx is already closed, but is otherwise safe to call multiple times. Hence, a defer tx.Rollback() is safe even if tx.Commit() will be called first in a non-error condition.

func (tx *dbTx) Rollback(ctx context.Context) error {
	if tx.closed {
		return ErrTxClosed
	}

	_, err := tx.conn.Exec(ctx, "rollback")
	tx.closed = true

A rollback failure leaves the connection in an undefined state

		tx.conn.die(fmt.Errorf("rollback failed: %w", err))
		return err
	}

	return nil
}

Exec delegates to the underlying *Conn

func (tx *dbTx) Exec(ctx context.Context, sql string, arguments ...interface{}) (commandTag pgconn.CommandTag, err error) {
	return tx.conn.Exec(ctx, sql, arguments...)
}

Prepare delegates to the underlying *Conn

func (tx *dbTx) Prepare(ctx context.Context, name, sql string) (*pgconn.StatementDescription, error) {
	if tx.closed {
		return nil, ErrTxClosed
	}

	return tx.conn.Prepare(ctx, name, sql)
}

Query delegates to the underlying *Conn

func (tx *dbTx) Query(ctx context.Context, sql string, args ...interface{}) (Rows, error) {

Because checking for errors can be deferred to the *Rows, build one with the error

		err := ErrTxClosed
		return &connRows{closed: true, err: err}, err
	}

	return tx.conn.Query(ctx, sql, args...)
}

QueryRow delegates to the underlying *Conn

func (tx *dbTx) QueryRow(ctx context.Context, sql string, args ...interface{}) Row {
	rows, _ := tx.Query(ctx, sql, args...)
	return (*connRow)(rows.(*connRows))
}

QueryFunc delegates to the underlying *Conn.

func (tx *dbTx) QueryFunc(ctx context.Context, sql string, args []interface{}, scans []interface{}, f func(QueryFuncRow) error) (pgconn.CommandTag, error) {
	if tx.closed {
		return nil, ErrTxClosed
	}

	return tx.conn.QueryFunc(ctx, sql, args, scans, f)
}

CopyFrom delegates to the underlying *Conn

func (tx *dbTx) CopyFrom(ctx context.Context, tableName Identifier, columnNames []string, rowSrc CopyFromSource) (int64, error) {
	if tx.closed {
		return 0, ErrTxClosed
	}

	return tx.conn.CopyFrom(ctx, tableName, columnNames, rowSrc)
}

SendBatch delegates to the underlying *Conn

func (tx *dbTx) SendBatch(ctx context.Context, b *Batch) BatchResults {
	if tx.closed {
		return &batchResults{err: ErrTxClosed}
	}

	return tx.conn.SendBatch(ctx, b)
}

LargeObjects returns a LargeObjects instance for the transaction.

func (tx *dbTx) LargeObjects() LargeObjects {
	return LargeObjects{tx: tx}
}

func (tx *dbTx) Conn() *Conn {
	return tx.conn
}

dbSavepoint represents a nested transaction implemented by a savepoint.

type dbSavepoint struct {
	tx           Tx
	savepointNum int64
	closed       bool
}

Begin starts a pseudo nested transaction implemented with a savepoint.

func (sp *dbSavepoint) Begin(ctx context.Context) (Tx, error) {
	if sp.closed {
		return nil, ErrTxClosed
	}

	return sp.tx.Begin(ctx)
}

func (sp *dbSavepoint) BeginFunc(ctx context.Context, f func(Tx) error) (err error) {
	if sp.closed {
		return ErrTxClosed
	}

	return sp.tx.BeginFunc(ctx, f)
}

Commit releases the savepoint essentially committing the pseudo nested transaction.

func (sp *dbSavepoint) Commit(ctx context.Context) error {
	if sp.closed {
		return ErrTxClosed
	}

	_, err := sp.Exec(ctx, "release savepoint sp_"+strconv.FormatInt(sp.savepointNum, 10))
	sp.closed = true
	return err
}

Rollback rolls back to the savepoint essentially rolling back the pseudo nested transaction. Rollback will return ErrTxClosed if the dbSavepoint is already closed, but is otherwise safe to call multiple times. Hence, a defer sp.Rollback() is safe even if sp.Commit() will be called first in a non-error condition.

func (sp *dbSavepoint) Rollback(ctx context.Context) error {
	if sp.closed {
		return ErrTxClosed
	}

	_, err := sp.Exec(ctx, "rollback to savepoint sp_"+strconv.FormatInt(sp.savepointNum, 10))
	sp.closed = true
	return err
}

Exec delegates to the underlying Tx

func (sp *dbSavepoint) Exec(ctx context.Context, sql string, arguments ...interface{}) (commandTag pgconn.CommandTag, err error) {
	if sp.closed {
		return nil, ErrTxClosed
	}

	return sp.tx.Exec(ctx, sql, arguments...)
}

Prepare delegates to the underlying Tx

func (sp *dbSavepoint) Prepare(ctx context.Context, name, sql string) (*pgconn.StatementDescription, error) {
	if sp.closed {
		return nil, ErrTxClosed
	}

	return sp.tx.Prepare(ctx, name, sql)
}

Query delegates to the underlying Tx

func (sp *dbSavepoint) Query(ctx context.Context, sql string, args ...interface{}) (Rows, error) {

Because checking for errors can be deferred to the *Rows, build one with the error

		err := ErrTxClosed
		return &connRows{closed: true, err: err}, err
	}

	return sp.tx.Query(ctx, sql, args...)
}

QueryRow delegates to the underlying Tx

func (sp *dbSavepoint) QueryRow(ctx context.Context, sql string, args ...interface{}) Row {
	rows, _ := sp.Query(ctx, sql, args...)
	return (*connRow)(rows.(*connRows))
}

QueryFunc delegates to the underlying Tx.

func (sp *dbSavepoint) QueryFunc(ctx context.Context, sql string, args []interface{}, scans []interface{}, f func(QueryFuncRow) error) (pgconn.CommandTag, error) {
	if sp.closed {
		return nil, ErrTxClosed
	}

	return sp.tx.QueryFunc(ctx, sql, args, scans, f)
}

CopyFrom delegates to the underlying *Conn

func (sp *dbSavepoint) CopyFrom(ctx context.Context, tableName Identifier, columnNames []string, rowSrc CopyFromSource) (int64, error) {
	if sp.closed {
		return 0, ErrTxClosed
	}

	return sp.tx.CopyFrom(ctx, tableName, columnNames, rowSrc)
}

SendBatch delegates to the underlying *Conn

func (sp *dbSavepoint) SendBatch(ctx context.Context, b *Batch) BatchResults {
	if sp.closed {
		return &batchResults{err: ErrTxClosed}
	}

	return sp.tx.SendBatch(ctx, b)
}

func (sp *dbSavepoint) LargeObjects() LargeObjects {
	return LargeObjects{tx: sp}
}

func (sp *dbSavepoint) Conn() *Conn {
	return sp.tx.Conn()