gotosocial/vendor/github.com/jackc/pgx/v4/tx.go

445 lines
14 KiB
Go
Raw Normal View History

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
}
var emptyTxOptions TxOptions
func (txOptions TxOptions) beginSQL() string {
if txOptions == emptyTxOptions {
return "begin"
}
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())
if err != nil {
// 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.
type Tx interface {
// 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
if err != nil {
// 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) {
if tx.closed {
// 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) {
if sp.closed {
// 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()
}