pgx + go-transaction-manager

O que pgx + go-transaction-manager faz?

Principal. O repositório obtém a conexão através de DefaultTrOrDB, os limites da transação são definidos pelo serviço através de trManager.Do. Valores em consultas — apenas como argumentos $1, $2. Duplicata esperada — ON CONFLICT, e não um 23505 capturado. Em 40001 / 40P01 / 55P03 repita a transação inteira, o loop em volta do Do externo. O que fazer com condições de corrida pelas mesmas linhas — skill postgres-concurrency.

Verificação de afirmações — CHECKS.md.

Versões

go get github.com/jackc/pgx/v5@latest
go get github.com/avito-tech/go-transaction-manager/trm/v2@latest
go get github.com/avito-tech/go-transaction-manager/drivers/pgxv5/v2@latest

• pgx — não inferior a v5.9.2. Nele foi fechada uma vulnerabilidade: injeção de SQL através de literais dollar-quoted no protocolo simples. Na v5.9.1 corrigiram a corrupção de resultados de batch com consultas preparadas em cache. Veja a última versão de patch através de go list -m -u github.com/jackc/pgx/v5.
• trm e driver pgxv5 — não inferiores a v2.0.4, de preferência v2.0.5. Na 2.0.4 foi removida do driver a goroutine em segundo plano awaitDone: o cancelamento de contexto não faz mais rollback de outra goroutine (pgx.Tx não é thread-safe, em uma consulta em execução isso causava pânico). Rollback agora é emitido após o retorno da função transacional. Na 2.0.5 foi fechado um vazamento de contexto em manager.Init em erros. O módulo é versionado com tags do tipo trm/v2.0.5.

Pool

cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
	return nil, fmt.Errorf("parse dsn: %w", err)
}
cfg.MaxConns = 25 // quanto o banco aguenta, não quantos núcleos o pod tem
cfg.MinConns = 2
cfg.ConnConfig.ConnectTimeout = 5 * time.Second
pool, err := pgxpool.NewWithConfig(context.Background(), cfg)

• Defina MaxConns explicitamente. O padrão é max(4, runtime.NumCPU()), ou seja, o número de núcleos da máquina, mas o tamanho do pool deve depender do banco. A soma dos MaxConns de todos os serviços e réplicas deve caber em max_connections (padrão 100): cada conexão é um processo PostgreSQL separado.
• Outros padrões do pgxpool: MinConns 0, MaxConnLifetime 1 h, MaxConnIdleTime 30 min, HealthCheckPeriod 1 min. Se encurtar MaxConnIdleTime — mantenha MinConns ≥ 2, caso contrário, após inatividade, a consulta pagará por uma nova conexão.
• MaxConnLifetime não trata consultas penduradas. Para isso — statement_timeout e context.WithTimeout na consulta: um pool esgotado retornará DeadlineExceeded, não ficará pendurado.
• No serviço — apenas pgxpool. pgx.Conn não é thread-safe: de várias goroutines você obterá conn busy e condição de corrida de dados. pgx.Conn é para LISTEN/NOTIFY em uma conexão dedicada e scripts descartáveis.
• NewWithConfig não se conecta ao banco: o pool é preguiçoso. A indisponibilidade do banco na inicialização só é visível no primeiro Ping. Sem timeout de conexão, pgx aguarda o timeout TCP do SO (no Linux cerca de dois minutos). Ping com repetição, classificador de erros de rede, hooks PrepareConn e AfterRelease — references/pool-startup.md.

Gerenciador de transações e repositório

import (
	trmpgxv5 "github.com/avito-tech/go-transaction-manager/drivers/pgxv5/v2"
	"github.com/avito-tech/go-transaction-manager/trm/v2/manager"
)

trManager := manager.Must(trmpgxv5.NewDefaultFactory(pool))

O repositório não sabe se está em uma transação: DefaultTrOrDB retorna a transação do contexto, se houver, caso contrário, o pool.

type UserRepository struct {
	pool   *pgxpool.Pool
	getter *trmpgxv5.CtxGetter
}

func NewUserRepository(pool *pgxpool.Pool) *UserRepository {
	return &UserRepository{pool: pool, getter: trmpgxv5.DefaultCtxGetter}
}

func (r *UserRepository) GetByID(ctx context.Context, id int64) (User, error) {
	rows, err := r.getter.DefaultTrOrDB(ctx, r.pool).Query(ctx,
		`SELECT id, name, email, bio FROM users WHERE id = $1`, id)
	if err != nil {
		return User{}, fmt.Errorf("select user %d: %w", id, err)
	}
	u, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[User])
	if errors.Is(err, pgx.ErrNoRows) {
		return User{}, fmt.Errorf("user %d: %w", id, ErrNotFound)
	}
	return u, err
}

Estruturas e leitura de linhas

• O Scan comum por tags não mapeia. Padrão — tag db:"..." e pgx.RowToStructByName[T]; lista — pgx.CollectRows, uma linha — pgx.CollectOneRow (em conjunto vazio — pgx.ErrNoRows). rows.Close() e rows.Err() eles fazem por conta própria.
• Se ler manualmente com rows.Next() (com saída antecipada) — logo após Query coloque defer rows.Close(). Linhas não fechadas seguram a conexão: com MaxConns=3, após três vazamentos, a quarta consulta fica pendurada até o timeout do contexto.
• Coluna anulável — pgtype.Text, pgtype.Timestamptz etc.: Valid == false é NULL tanto na leitura quanto na escrita. Slices de relacionamentos (Orders []Order) — sem tag, segunda consulta e montagem manual.

type User struct {
	ID    int64       `db:"id"`
	Name  string      `db:"name"`
	Email string      `db:"email"`
	Bio   pgtype.Text `db:"bio"` // anulável
}

Parâmetros de consulta — apenas $1, $2

Qualquer valor — telefone, nome, string de busca — passe como argumento $N. Não monte SQL com + ou fmt.Sprintf, mesmo que a string pareça segura.

• Exec sem argumentos pgx sempre envia pelo protocolo simples (em conn.go: «Always use simple protocol when there are no arguments»), e o protocolo simples executa vários comandos em sequência: '; DELETE FROM ...; -- passará. Exec com argumentos rejeitará o segundo comando (cannot insert multiple commands into a prepared statement). Query sem argumentos não muda para protocolo simples, mas colar strings também não é permitido.
• O protocolo estendido não salva de injeção: UNION SELECT é uma única consulta, e através de Query ela passa. Assim vazam dados pessoais.
• Nomes de colunas, ORDER BY e direção de ordenação não podem ser passados como argumento. Pegue-os de uma lista branca: map do valor da API para um fragmento SQL pronto; para desconhecido — erro.

Transação via Do

func (s *UserService) CreateWithProfile(ctx context.Context, name, email, bio string) error {
	return s.trManager.Do(ctx, func(ctx context.Context) error {
		userID, err := s.users.Create(ctx, name, email)
		if err != nil {
			return err
		}
		return s.profiles.Create(ctx, userID, bio)
	}) // erro — rollback, nil — commit
}

Do aninhado se junta à transação aberta, não cria savepoint. A propagação padrão é PropagationRequired: erro na chamada aninhada desfaz tudo completamente. Rollback parcial peça explicitamente — PropagationNested, o driver fará SAVEPOINT:

nested := settings.Must(settings.WithPropagation(trm.PropagationNested))

return s.trManager.Do(ctx, func(ctx context.Context) error {
	if err := s.repo.Save(ctx, u); err != nil {
		return err
	}
	// O erro será desfeito até o savepoint, a transação externa continuará.
	if err := s.trManager.DoWithSettings(ctx, nested, func(ctx context.Context) error {
		return s.repo.SaveOptionalPart(ctx, p)
	}); err != nil {
		s.log.Warn("parte opcional ignorada", "err", err)
	}
	return nil
})

Rollback parcial não é necessário — não complique: um Do aninhado é suficiente, apenas ele não isola nada.

Repetição de transação

Quando a repetição é necessária (repeatable read e serializable, deadlock, lock_timeout, NOWAIT), — skill postgres-concurrency. Aqui — como fazer isso em Go.

• Repita a transação inteira, junto com a leitura e a lógica que decide o que escrever. Repetir a última instrução é inútil: a transação já foi abortada.
• O loop em volta do Do externo. O Do aninhado se junta à transação aberta, e a repetição dentro dele não iniciará uma nova.
• 40001 pode chegar também no COMMIT: o gerenciador o retornará encapsulado («transaction: commit; ...»), errors.As o encontrará. Esta é mais uma razão para manter todo o Do no loop.
• Limite as tentativas, entre elas — uma pequena pausa crescente.

var serializable = trmpgxv5.MustSettings(
	settings.Must(),
	trmpgxv5.WithTxOptions(pgx.TxOptions{IsoLevel: pgx.Serializable}),
)

func isRetryableTxErr(err error) bool {
	var pgErr *pgconn.PgError
	if !errors.As(err, &pgErr) {
		return false
	}
	switch pgErr.Code {
	case "40001", "40P01", "55P03": // serialization_failure, deadlock_detected, lock_not_available
		return true
	}
	return false
}

func (s *BookingService) Book(ctx context.Context, req BookRequest) error {
	const attempts = 3
	var err error
	for i := range attempts {
		err = s.trManager.DoWithSettings(ctx, serializable, func(ctx context.Context) error {
			return s.book(ctx, req) // toda leitura e todas as decisões — dentro
		})
		if !isRetryableTxErr(err) {
			return err
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(i+1) * 20 * time.Millisecond):
		}
	}
	return err
}

Erros

Analise os erros do driver no repositório, para fora retorne erros de domínio: o caso de uso não deve saber sobre pgx.

var (
	ErrNotFound = errors.New("não encontrado")
	ErrConflict = errors.New("já existe")
)

var pgErr *pgconn.PgError
if errors.As(err, &pgErr) && pgErr.Code == "23505" { // unique_violation
	return 0, fmt.Errorf("criar usuário %q: %w", email, ErrConflict)
}

Duplicata esperada — ON CONFLICT, e não um 23505 capturado. Qualquer erro dentro de Do, incluindo 23505, interrompe a transação: em seguida, cada comando receberá 25P02 («current transaction is aborted»). Se a duplicata é um resultado normal (reenvio, clique duplo):

err := q.QueryRow(ctx,
	`INSERT INTO users (name, email) VALUES ($1, $2)
	 ON CONFLICT (email) DO NOTHING
	 RETURNING id`, name, email,
).Scan(&id)
if errors.Is(err, pgx.ErrNoRows) { // não criado: email já ocupado
	return 0, fmt.Errorf("criar usuário %q: %w", email, ErrConflict)
}

• O alvo do conflito (colunas) especifique sempre. Sem ele, DO NOTHING suprime a violação de qualquer restrição única, e qual delas foi acionada já não é possível saber.
• Zero linhas em RETURNING — registro não criado: a linha já existe ou acabou de ser confirmada por uma transação paralela. Isso não é sucesso.
• A violação de outra restrição única ainda virá como 23505 — distinga por pgErr.ConstraintName.
• Mantenha a unicidade com índice ou restrição, e não com SELECT antes de INSERT: entre eles, uma transação paralela pode ocupar o valor.

Códigos frequentes: 23505 unique_violation, 23503 foreign_key_violation, 23502 not_null_violation, 23P01 exclusion_violation, 40001 serialization_failure, 40P01 deadlock_detected, 55P03 lock_not_available. Repetir faz sentido apenas para os três últimos, e somente a transação inteira.

Batch e CopyFrom

• pgx.Batch — várias diferentes consultas em uma única viagem de ida e volta. Após SendBatch leia o resultado de cada consulta e feche BatchResults.
• CopyFrom — carregamento em massa pelo protocolo COPY: análise uma vez, não para cada linha. Em 100 mil linhas, ele é cerca de uma ordem de grandeza mais rápido que um batch de INSERTs.
• CopyFrom não suporta ON CONFLICT e não retorna id. Precisa de upsert ou RETURNING — carregue CopyFrom em uma tabela temporária e faça INSERT ... SELECT ... ON CONFLICT.

n, err := r.getter.DefaultTrOrDB(ctx, r.pool).CopyFrom(ctx,
	pgx.Identifier{"orders"},
	[]string{"user_id", "total"},
	pgx.CopyFromSlice(len(orders), func(i int) ([]any, error) {
		return []any{orders[i].UserID, orders[i].Total}, nil
	}),
)

Folha de dicas

Tarefa	Ferramenta
Inicializar com banco indisponível	ConnConfig.ConnectTimeout + Ping com repetição (references/pool-startup.md)
Lista / um registro	CollectRows / CollectOneRow + RowToStructByName
rows.Next() manual	defer rows.Close() imediatamente após Query
Consulta no repositório	getter.DefaultTrOrDB(ctx, pool)
Valores da entrada	apenas $1, $2; colunas e ORDER BY — lista branca
Transação	trManager.Do(ctx, fn)
Rollback parcial	DoWithSettings + trm.PropagationNested
Nível de isolamento	trmpgxv5.MustSettings(settings.Must(), trmpgxv5.WithTxOptions(...))
40001 / 40P01 / 55P03	repetição de toda a transação em volta do Do externo
Sem linha / duplicata	errors.Is(err, pgx.ErrNoRows) / PgError.Code == "23505"
Duplicata — resultado normal	INSERT ... ON CONFLICT (cols) DO NOTHING RETURNING id
Carga em massa	CopyFrom; consultas diferentes de uma vez — pgx.Batch
Anulável	pgtype.Text / Timestamptz / ...
Condições de corrida por linhas, bloqueios, filas	skill postgres-concurrency