github.com/jackc/pgx/v5 v5.10.0: Execute SQL statements within pgx transactions using pgx.Tx.Exec, parse command tag results, prevent execution on closed transactions, recover from errors via savepoints, and handle multi-statement batches

# Notes

## Search Results
`search_known_solution` returned `NO_SAFE_MATCH` for `pgx.Tx.Exec` on `github.com/jackc/pgx/v5@v5.10.0`. Existing samples in the library cover collector cardinality (`CollectOneRow` vs `CollectExactlyOneRow`), DSN parsing modes, connection pools, and structured server error inspections. We author a dedicated contract proving transaction execution semantics, command tag parsing, closed transaction lifecycle boundaries, savepoint error recovery, and simple protocol multi-statement handling via `pgx.Tx.Exec`.

## Pinned Release
- `github.com/jackc/pgx/v5` `v5.10.0`

## Observed Behaviour

1. **Command Tag Parsing vs `sql.Result`**:
   - `pgx.Tx.Exec` returns `(pgconn.CommandTag, error)` instead of standard Go `database/sql.Result`.
   - `pgconn.CommandTag` exposes typed inspection methods (`tag.Insert()`, `tag.Update()`, `tag.Delete()`) and parses row counts directly via `tag.RowsAffected()` (e.g. `"INSERT 0 3"` yields `RowsAffected() == 3` and `Insert() == true`).
   - DDL statements (such as `CREATE TABLE`) yield `tag.RowsAffected() == 0` with `tag.String() == "CREATE TABLE"`.

2. **Client-Side Closed Transaction Guard**:
   - Once a transaction is committed (`tx.Commit`) or rolled back (`tx.Rollback`), calling `tx.Exec` returns `pgx.ErrTxClosed` immediately on the client side without issuing any wire protocol commands to the backend.
   - When a nested subtransaction created via `tx.Begin` is rolled back (`subTx.Rollback`), calling `subTx.Exec` returns `pgx.ErrTxClosed`, while the parent transaction `tx` remains active and capable of executing further commands.

3. **Savepoint Error Recovery**:
   - In PostgreSQL, when a statement fails inside a transaction block, the transaction is marked as aborted (transaction status `'E'`), and subsequent queries fail with `SQLSTATE 25P02` (`current transaction is aborted, commands ignored until end of transaction block`).
   - When executing a failing query inside a subtransaction (`subTx.Exec`), rolling back the subtransaction (`subTx.Rollback`) issues `rollback to savepoint sp_1`, recovering backend transaction status to `'T'` (in transaction) and allowing the parent transaction to commit or execute further statements successfully.

4. **Parameter Binding and SQL NULL Encoding**:
   - Passing `nil` (untyped or typed `nil` pointer) or `pgtype.Text{Valid: false}` to `tx.Exec` encodes the argument as PostgreSQL `NULL` (length `-1` in the PostgreSQL Bind protocol message).
   - Positional parameters (`$1, $2, ...`) are mapped and type-encoded without client-side string concatenation.

5. **Simple Protocol Multi-Statement Execution**:
   - Under simple protocol mode (`QueryExecModeSimpleProtocol`), `tx.Exec` executes multiple semicolon-separated SQL statements in a single roundtrip.
   - `tx.Exec` returns the `pgconn.CommandTag` of the last executed statement in the batch.

6. **Context Cancellation Handling**:
   - Passing a pre-canceled context (`context.WithCancel`) to `tx.Exec` returns `context.Canceled` immediately.

## How the Wrong Expectation Fails

- **Closed Transaction Execution**: Loudly at runtime with `pgx.ErrTxClosed` error. If code expects `tx.Exec` to lazily reopen or silently ignore post-commit writes, execution fails immediately.
- **Transaction Abort State**: Loudly at runtime with PostgreSQL error `25P02` (`current transaction is aborted, commands ignored until end of transaction block`). Without rolling back to a savepoint, any subsequent `tx.Exec` or `tx.Commit` fails.
- **Multi-Statement Execution under Extended Protocol**: Loudly at runtime with PostgreSQL syntax error `42601` (`cannot insert multiple commands into a prepared statement`) unless simple protocol mode is configured.

{"case":{"believed":"tx.Exec returns standard sql.Result types, allows execution after transaction commit or rollback, and leaves transactions permanently broken upon query failure without savepoint recovery","caseId":"case:sha256:f1ec7a90c154f4230de96581e02329c4d5e6bf67f73de091fb14e4e28ffd25e5","contract":["pgx.Tx.Exec returns a structured pgconn.CommandTag with RowsAffected and command predicates rather than sql.Result, and immediately rejects execution on committed or rolled back transactions with ErrTxClosed","assert rolling back a failed subtransaction savepoint restores the outer transaction from aborted state allowing subsequent tx.Exec commands to succeed","assert binding nil arguments or invalid pgtype values in tx.Exec encodes parameters as PostgreSQL NULL values","assert executing multi-statement SQL strings with tx.Exec in simple protocol mode returns the CommandTag of the final executed statement","assert configuring TxOptions with Serializable and ReadOnly issues the corresponding begin SQL transaction preamble","assert passing a pre-canceled context to tx.Exec immediately returns context.Canceled without executing on the connection"],"goal":"Execute SQL statements within pgx transactions using pgx.Tx.Exec, parse command tag results, prevent execution on closed transactions, recover from errors via savepoints, and handle multi-statement batches","kind":"HOW","packages":["pkg:golang/github.com/jackc/pgx/v5@v5.10.0"],"schemaVersion":1,"symbols":["pgx.Tx.Exec"]},"contractCommand":["go","test","./..."],"environment":{"arch":"x64","ecosystem":"golang","executionContext":"go","language":"go","os":"linux","packageManager":"gomod","runtime":"go","schemaVersion":1},"license":"MIT-0","packages":["pkg:golang/github.com/jackc/pgx/v5@v5.10.0"],"schemaVersion":1,"symbols":["pgx.Tx.Exec"],"verifierAdapter":"golang@1"}

module sample

go 1.26.5

require github.com/jackc/pgx/v5 v5.10.0

require (
	github.com/jackc/pgpassfile v1.0.0 // indirect
	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
	golang.org/x/text v0.29.0 // indirect
)

github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

package sample

import (
	"context"
	"errors"
	"fmt"
	"net"
	"strings"
	"sync/atomic"
	"testing"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgproto3"
	"github.com/jackc/pgx/v5/pgtype"
)

// mockBackend creates an in-memory PostgreSQL server mock using net.Pipe
func mockBackend(t *testing.T, handler func(msg pgproto3.FrontendMessage, b *pgproto3.Backend, txStatus *byte) bool) (*pgx.Conn, context.Context) {
	t.Helper()
	cfg, err := pgx.ParseConfig("postgres://127.0.0.1/testdb?sslmode=disable")
	if err != nil {
		t.Fatalf("ParseConfig failed: %v", err)
	}

	cfg.DialFunc = func(ctx context.Context, _, _ string) (net.Conn, error) {
		c, s := net.Pipe()
		go func(conn net.Conn) {
			defer conn.Close()
			b := pgproto3.NewBackend(conn, conn)
			if msg, _ := b.ReceiveStartupMessage(); msg != nil {
				if _, ok := msg.(*pgproto3.SSLRequest); ok {
					_, _ = conn.Write([]byte{'N'})
					_, _ = b.ReceiveStartupMessage()
				}
			}
			b.Send(&pgproto3.AuthenticationOk{})
			b.Send(&pgproto3.ParameterStatus{Name: "standard_conforming_strings", Value: "on"})
			b.Send(&pgproto3.ParameterStatus{Name: "server_version", Value: "16.0"})
			b.Send(&pgproto3.ParameterStatus{Name: "client_encoding", Value: "UTF8"})
			b.Send(&pgproto3.ReadyForQuery{TxStatus: 'I'})
			_ = b.Flush()

			txStatus := byte('I')
			paramCount := 0
			for {
				msg, err := b.Receive()
				if err != nil || msg == nil {
					return
				}
				if handler != nil && handler(msg, b, &txStatus) {
					continue
				}
				switch m := msg.(type) {
				case *pgproto3.Terminate:
					return
				case *pgproto3.Query:
					queryStr := strings.TrimSpace(strings.ToLower(m.String))
					switch {
					case strings.HasPrefix(queryStr, "begin"):
						txStatus = 'T'
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("BEGIN")})
					case strings.HasPrefix(queryStr, "commit"):
						txStatus = 'I'
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("COMMIT")})
					case strings.HasPrefix(queryStr, "rollback to savepoint"):
						txStatus = 'T'
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("ROLLBACK")})
					case strings.HasPrefix(queryStr, "rollback"):
						txStatus = 'I'
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("ROLLBACK")})
					case strings.HasPrefix(queryStr, "savepoint"):
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("SAVEPOINT")})
					case strings.HasPrefix(queryStr, "release savepoint"):
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("RELEASE")})
					default:
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("OK")})
					}
					b.Send(&pgproto3.ReadyForQuery{TxStatus: txStatus})
					_ = b.Flush()
				case *pgproto3.Parse:
					paramCount = len(m.ParameterOIDs)
					if paramCount == 0 && strings.Contains(m.Query, "$") {
						// Estimate parameter count from max $N in query
						for i := 1; i <= 20; i++ {
							if strings.Contains(m.Query, fmt.Sprintf("$%d", i)) {
								paramCount = i
							}
						}
					}
					b.Send(&pgproto3.ParseComplete{})
					_ = b.Flush()
				case *pgproto3.Describe:
					if m.ObjectType == 'S' {
						oids := make([]uint32, paramCount)
						b.Send(&pgproto3.ParameterDescription{ParameterOIDs: oids})
					}
					b.Send(&pgproto3.NoData{})
					_ = b.Flush()
				case *pgproto3.Close:
					b.Send(&pgproto3.CloseComplete{})
					_ = b.Flush()
				case *pgproto3.Bind:
					b.Send(&pgproto3.BindComplete{})
					_ = b.Flush()
				case *pgproto3.Execute:
					b.Send(&pgproto3.CommandComplete{CommandTag: []byte("UPDATE 1")})
					_ = b.Flush()
				case *pgproto3.Sync:
					b.Send(&pgproto3.ReadyForQuery{TxStatus: txStatus})
					_ = b.Flush()
				}
			}
		}(s)
		return c, nil
	}

	ctx := context.Background()
	conn, err := pgx.ConnectConfig(ctx, cfg)
	if err != nil {
		t.Fatalf("ConnectConfig failed: %v", err)
	}
	t.Cleanup(func() { _ = conn.Close(ctx) })
	return conn, ctx
}

func TestTxExec_CommandTagsAndRowsAffected(t *testing.T) {
	var receivedExecutes int32
	conn, ctx := mockBackend(t, func(msg pgproto3.FrontendMessage, b *pgproto3.Backend, txStatus *byte) bool {
		switch m := msg.(type) {
		case *pgproto3.Query:
			queryStr := strings.TrimSpace(strings.ToLower(m.String))
			switch {
			case strings.HasPrefix(queryStr, "begin"):
				*txStatus = 'T'
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("BEGIN")})
			case strings.HasPrefix(queryStr, "create table"):
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("CREATE TABLE")})
			case strings.HasPrefix(queryStr, "commit"):
				*txStatus = 'I'
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("COMMIT")})
			default:
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("OK")})
			}
			b.Send(&pgproto3.ReadyForQuery{TxStatus: *txStatus})
			_ = b.Flush()
			return true
		case *pgproto3.Execute:
			idx := atomic.AddInt32(&receivedExecutes, 1)
			switch idx {
			case 1:
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("INSERT 0 3")})
			case 2:
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("UPDATE 42")})
			case 3:
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("DELETE 7")})
			case 4:
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("CREATE TABLE")})
			default:
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("OK")})
			}
			_ = b.Flush()
			return true
		}
		return false
	})

	tx, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("tx.Begin failed: %v", err)
	}

	// 1. INSERT tag parsing
	tagInsert, err := tx.Exec(ctx, "INSERT INTO items (name) VALUES ($1), ($2), ($3)", "a", "b", "c")
	if err != nil {
		t.Fatalf("tx.Exec INSERT failed: %v", err)
	}
	if !tagInsert.Insert() || tagInsert.RowsAffected() != 3 || tagInsert.String() != "INSERT 0 3" {
		t.Fatalf("unexpected INSERT tag: %+v", tagInsert)
	}

	// 2. UPDATE tag parsing
	tagUpdate, err := tx.Exec(ctx, "UPDATE items SET active = true WHERE category = $1", "books")
	if err != nil {
		t.Fatalf("tx.Exec UPDATE failed: %v", err)
	}
	if !tagUpdate.Update() || tagUpdate.RowsAffected() != 42 || tagUpdate.String() != "UPDATE 42" {
		t.Fatalf("unexpected UPDATE tag: %+v", tagUpdate)
	}

	// 3. DELETE tag parsing
	tagDelete, err := tx.Exec(ctx, "DELETE FROM items WHERE expired = $1", true)
	if err != nil {
		t.Fatalf("tx.Exec DELETE failed: %v", err)
	}
	if !tagDelete.Delete() || tagDelete.RowsAffected() != 7 || tagDelete.String() != "DELETE 7" {
		t.Fatalf("unexpected DELETE tag: %+v", tagDelete)
	}

	// 4. DDL tag parsing (RowsAffected == 0)
	tagDDL, err := tx.Exec(ctx, "CREATE TABLE temp_logs (id int)")
	if err != nil {
		t.Fatalf("tx.Exec CREATE TABLE failed: %v", err)
	}
	if tagDDL.RowsAffected() != 0 || tagDDL.String() != "CREATE TABLE" {
		t.Fatalf("unexpected CREATE TABLE tag: %+v", tagDDL)
	}

	if err := tx.Commit(ctx); err != nil {
		t.Fatalf("tx.Commit failed: %v", err)
	}
}

func TestTxExec_ClosedTransactionRejection(t *testing.T) {
	conn, ctx := mockBackend(t, nil)

	// 1. tx.Exec after Commit returns pgx.ErrTxClosed without network roundtrip
	tx, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("conn.Begin failed: %v", err)
	}
	if err := tx.Commit(ctx); err != nil {
		t.Fatalf("tx.Commit failed: %v", err)
	}

	_, err = tx.Exec(ctx, "INSERT INTO items (name) VALUES ($1)", "test")
	if !errors.Is(err, pgx.ErrTxClosed) {
		t.Fatalf("expected ErrTxClosed after Commit, got: %v", err)
	}

	// 2. tx.Exec after Rollback returns pgx.ErrTxClosed
	tx2, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("conn.Begin failed: %v", err)
	}
	if err := tx2.Rollback(ctx); err != nil {
		t.Fatalf("tx2.Rollback failed: %v", err)
	}

	_, err = tx2.Exec(ctx, "UPDATE items SET name = $1", "test2")
	if !errors.Is(err, pgx.ErrTxClosed) {
		t.Fatalf("expected ErrTxClosed after Rollback, got: %v", err)
	}

	// 3. Subtransaction rollback closes subTx but parent tx remains open and runnable
	tx3, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("conn.Begin failed: %v", err)
	}
	subTx, err := tx3.Begin(ctx)
	if err != nil {
		t.Fatalf("tx3.Begin subTx failed: %v", err)
	}
	if err := subTx.Rollback(ctx); err != nil {
		t.Fatalf("subTx.Rollback failed: %v", err)
	}

	// subTx.Exec is closed
	_, err = subTx.Exec(ctx, "INSERT INTO items (name) VALUES ($1)", "sub")
	if !errors.Is(err, pgx.ErrTxClosed) {
		t.Fatalf("expected ErrTxClosed on rolled back subTx.Exec, got: %v", err)
	}

	// parent tx3.Exec remains active and succeeds
	tag, err := tx3.Exec(ctx, "INSERT INTO items (name) VALUES ($1)", "parent")
	if err != nil {
		t.Fatalf("expected parent tx3.Exec to succeed after subTx rollback, got: %v", err)
	}
	if tag.RowsAffected() != 1 {
		t.Fatalf("unexpected tag rows affected: %d", tag.RowsAffected())
	}
	if err := tx3.Commit(ctx); err != nil {
		t.Fatalf("tx3.Commit failed: %v", err)
	}
}

func TestTxExec_SavepointErrorRecovery(t *testing.T) {
	var currentQuery string
	conn, ctx := mockBackend(t, func(msg pgproto3.FrontendMessage, b *pgproto3.Backend, txStatus *byte) bool {
		switch m := msg.(type) {
		case *pgproto3.Parse:
			currentQuery = m.Query
			return false
		case *pgproto3.Query:
			queryStr := strings.TrimSpace(strings.ToLower(m.String))
			switch {
			case strings.HasPrefix(queryStr, "begin"):
				*txStatus = 'T'
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("BEGIN")})
			case strings.HasPrefix(queryStr, "savepoint"):
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("SAVEPOINT")})
			case strings.HasPrefix(queryStr, "rollback to savepoint"):
				*txStatus = 'T' // recovers transaction from error state 'E' back to valid tx 'T'
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("ROLLBACK")})
			case strings.HasPrefix(queryStr, "commit"):
				*txStatus = 'I'
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("COMMIT")})
			default:
				b.Send(&pgproto3.CommandComplete{CommandTag: []byte("OK")})
			}
			b.Send(&pgproto3.ReadyForQuery{TxStatus: *txStatus})
			_ = b.Flush()
			return true
		case *pgproto3.Execute:
			if *txStatus == 'E' {
				b.Send(&pgproto3.ErrorResponse{
					Severity: "ERROR",
					Code:     "25P02",
					Message:  "current transaction is aborted, commands ignored until end of transaction block",
				})
				_ = b.Flush()
				return true
			}
			if strings.Contains(strings.ToLower(currentQuery), "insert into users") {
				// Simulate unique constraint violation for the failing insert query
				*txStatus = 'E'
				b.Send(&pgproto3.ErrorResponse{
					Severity: "ERROR",
					Code:     "23505",
					Message:  "duplicate key value violates unique constraint \"users_email_key\"",
				})
				_ = b.Flush()
				return true
			}
			b.Send(&pgproto3.CommandComplete{CommandTag: []byte("OK")})
			_ = b.Flush()
			return true
		case *pgproto3.Sync:
			b.Send(&pgproto3.ReadyForQuery{TxStatus: *txStatus})
			_ = b.Flush()
			return true
		}
		return false
	})

	tx, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("tx.Begin failed: %v", err)
	}

	// Start subtransaction (savepoint)
	subTx, err := tx.Begin(ctx)
	if err != nil {
		t.Fatalf("subTx.Begin failed: %v", err)
	}

	// Executing failing query inside subtransaction
	_, err = subTx.Exec(ctx, "INSERT INTO users (email) VALUES ($1)", "duplicate@example.com")
	if err == nil {
		t.Fatalf("expected error from duplicate INSERT, got nil")
	}
	var pgErr *pgconn.PgError
	if !errors.As(err, &pgErr) || pgErr.Code != "23505" {
		t.Fatalf("expected PgError SQLState 23505, got: %v", err)
	}

	// Rolling back subtransaction restores outer transaction to valid state 'T'
	if err := subTx.Rollback(ctx); err != nil {
		t.Fatalf("subTx.Rollback failed: %v", err)
	}

	// Commit outer transaction succeeds because savepoint rollback recovered the transaction
	if err := tx.Commit(ctx); err != nil {
		t.Fatalf("tx.Commit failed: %v", err)
	}
}

func TestTxExec_BindParametersAndNullEncoding(t *testing.T) {
	var capturedBinds []pgproto3.Bind
	conn, ctx := mockBackend(t, func(msg pgproto3.FrontendMessage, b *pgproto3.Backend, txStatus *byte) bool {
		switch m := msg.(type) {
		case *pgproto3.Bind:
			capturedBinds = append(capturedBinds, *m)
			b.Send(&pgproto3.BindComplete{})
			_ = b.Flush()
			return true
		}
		return false
	})

	tx, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("tx.Begin failed: %v", err)
	}

	// Pass various parameter types including untyped nil, string, int, and pgtype.Text NULL
	var nilPtr *string
	nullText := pgtype.Text{Valid: false}
	validText := pgtype.Text{String: "custom", Valid: true}

	_, err = tx.Exec(ctx,
		"UPDATE accounts SET balance = $1, note = $2, ref = $3, descr = $4, custom = $5 WHERE id = $6",
		1500, nil, nilPtr, nullText, validText, int64(99),
	)
	if err != nil {
		t.Fatalf("tx.Exec with parameters failed: %v", err)
	}

	if len(capturedBinds) != 1 {
		t.Fatalf("expected 1 captured Bind message, got %d", len(capturedBinds))
	}

	bind := capturedBinds[0]
	if len(bind.Parameters) != 6 {
		t.Fatalf("expected 6 bound parameters, got %d", len(bind.Parameters))
	}

	// $1 (1500) - binary/text encoded integer
	if len(bind.Parameters[0]) == 0 {
		t.Fatalf("expected non-empty param for balance")
	}

	// $2 (untyped nil) -> NULL (represented as nil []byte slice in pgproto3 Bind)
	if bind.Parameters[1] != nil {
		t.Fatalf("expected nil bytes for untyped nil parameter, got %v", bind.Parameters[1])
	}

	// $3 (typed nil pointer) -> NULL
	if bind.Parameters[2] != nil {
		t.Fatalf("expected nil bytes for typed nil pointer parameter, got %v", bind.Parameters[2])
	}

	// $4 (pgtype.Text{Valid: false}) -> NULL
	if bind.Parameters[3] != nil {
		t.Fatalf("expected nil bytes for pgtype.Text{Valid: false}, got %v", bind.Parameters[3])
	}

	// $5 (pgtype.Text{String: "custom", Valid: true}) -> "custom"
	if string(bind.Parameters[4]) != "custom" {
		t.Fatalf("expected 'custom' parameter bytes, got %q", string(bind.Parameters[4]))
	}

	if err := tx.Commit(ctx); err != nil {
		t.Fatalf("tx.Commit failed: %v", err)
	}
}

func TestTxExec_SimpleProtocolMultiStatement(t *testing.T) {
	// Configure connection for Simple Protocol
	cfg, err := pgx.ParseConfig("postgres://127.0.0.1/testdb?sslmode=disable&default_query_exec_mode=simple_protocol")
	if err != nil {
		t.Fatalf("ParseConfig failed: %v", err)
	}

	var executedQueries []string
	cfg.DialFunc = func(ctx context.Context, _, _ string) (net.Conn, error) {
		c, s := net.Pipe()
		go func(conn net.Conn) {
			defer conn.Close()
			b := pgproto3.NewBackend(conn, conn)
			if msg, _ := b.ReceiveStartupMessage(); msg != nil {
				if _, ok := msg.(*pgproto3.SSLRequest); ok {
					_, _ = conn.Write([]byte{'N'})
					_, _ = b.ReceiveStartupMessage()
				}
			}
			b.Send(&pgproto3.AuthenticationOk{})
			b.Send(&pgproto3.ParameterStatus{Name: "server_version", Value: "16.0"})
			b.Send(&pgproto3.ReadyForQuery{TxStatus: 'I'})
			_ = b.Flush()

			for {
				msg, err := b.Receive()
				if err != nil || msg == nil {
					return
				}
				switch m := msg.(type) {
				case *pgproto3.Terminate:
					return
				case *pgproto3.Query:
					executedQueries = append(executedQueries, m.String)
					queryStr := strings.TrimSpace(strings.ToLower(m.String))
					switch {
					case strings.HasPrefix(queryStr, "begin"):
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("BEGIN")})
						b.Send(&pgproto3.ReadyForQuery{TxStatus: 'T'})
					case strings.Contains(queryStr, ";"):
						// Multi-statement execution emits CommandComplete for each statement, ReadyForQuery once at end
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("INSERT 0 2")})
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("UPDATE 5")})
						b.Send(&pgproto3.ReadyForQuery{TxStatus: 'T'})
					case strings.HasPrefix(queryStr, "commit"):
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("COMMIT")})
						b.Send(&pgproto3.ReadyForQuery{TxStatus: 'I'})
					default:
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("OK")})
						b.Send(&pgproto3.ReadyForQuery{TxStatus: 'T'})
					}
					_ = b.Flush()
				}
			}
		}(s)
		return c, nil
	}

	ctx := context.Background()
	conn, err := pgx.ConnectConfig(ctx, cfg)
	if err != nil {
		t.Fatalf("ConnectConfig failed: %v", err)
	}
	defer conn.Close(ctx)

	tx, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("conn.Begin failed: %v", err)
	}

	// In simple protocol, multi-statement tx.Exec sends one Query string and returns the CommandTag of the LAST statement
	tag, err := tx.Exec(ctx, "INSERT INTO items VALUES (1), (2); UPDATE items SET status = 'ready' WHERE id <= 5;")
	if err != nil {
		t.Fatalf("tx.Exec multi-statement failed: %v", err)
	}

	if !tag.Update() || tag.RowsAffected() != 5 || tag.String() != "UPDATE 5" {
		t.Fatalf("expected last statement CommandTag 'UPDATE 5', got %+v", tag)
	}

	if err := tx.Commit(ctx); err != nil {
		t.Fatalf("tx.Commit failed: %v", err)
	}
}

func TestTxExec_TxOptionsBeginStatement(t *testing.T) {
	var beginQuery string
	cfg, err := pgx.ParseConfig("postgres://127.0.0.1/testdb?sslmode=disable")
	if err != nil {
		t.Fatalf("ParseConfig failed: %v", err)
	}

	cfg.DialFunc = func(ctx context.Context, _, _ string) (net.Conn, error) {
		c, s := net.Pipe()
		go func(conn net.Conn) {
			defer conn.Close()
			b := pgproto3.NewBackend(conn, conn)
			if msg, _ := b.ReceiveStartupMessage(); msg != nil {
				if _, ok := msg.(*pgproto3.SSLRequest); ok {
					_, _ = conn.Write([]byte{'N'})
					_, _ = b.ReceiveStartupMessage()
				}
			}
			b.Send(&pgproto3.AuthenticationOk{})
			b.Send(&pgproto3.ParameterStatus{Name: "server_version", Value: "16.0"})
			b.Send(&pgproto3.ReadyForQuery{TxStatus: 'I'})
			_ = b.Flush()

			for {
				msg, err := b.Receive()
				if err != nil || msg == nil {
					return
				}
				switch m := msg.(type) {
				case *pgproto3.Query:
					if strings.HasPrefix(strings.ToLower(m.String), "begin") {
						beginQuery = m.String
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("BEGIN")})
						b.Send(&pgproto3.ReadyForQuery{TxStatus: 'T'})
						_ = b.Flush()
					} else if strings.HasPrefix(strings.ToLower(m.String), "rollback") {
						b.Send(&pgproto3.CommandComplete{CommandTag: []byte("ROLLBACK")})
						b.Send(&pgproto3.ReadyForQuery{TxStatus: 'I'})
						_ = b.Flush()
					}
				}
			}
		}(s)
		return c, nil
	}

	ctx := context.Background()
	conn, err := pgx.ConnectConfig(ctx, cfg)
	if err != nil {
		t.Fatalf("ConnectConfig failed: %v", err)
	}
	defer conn.Close(ctx)

	tx, err := conn.BeginTx(ctx, pgx.TxOptions{
		IsoLevel:       pgx.Serializable,
		AccessMode:     pgx.ReadOnly,
		DeferrableMode: pgx.Deferrable,
	})
	if err != nil {
		t.Fatalf("BeginTx failed: %v", err)
	}
	defer tx.Rollback(ctx)

	expectedBegin := "begin isolation level serializable read only deferrable"
	if beginQuery != expectedBegin {
		t.Fatalf("expected begin query %q, got %q", expectedBegin, beginQuery)
	}
}

func TestTxExec_ContextCancellation(t *testing.T) {
	conn, ctx := mockBackend(t, nil)

	tx, err := conn.Begin(ctx)
	if err != nil {
		t.Fatalf("conn.Begin failed: %v", err)
	}
	defer tx.Rollback(ctx)

	// A pre-canceled context returns context.Canceled immediately
	canceledCtx, cancel := context.WithCancel(ctx)
	cancel()

	_, err = tx.Exec(canceledCtx, "INSERT INTO items (name) VALUES ($1)", "canceled")
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("expected context.Canceled, got: %v", err)
	}
}