pgx + go-transaction-manager

¿Qué hace pgx + go-transaction-manager?

Importante. El repositorio obtiene la conexión mediante DefaultTrOrDB, los límites de la transacción los define el servicio a través de trManager.Do. Los valores en las consultas siempre se pasan como argumentos $1, $2. Si se espera un duplicado, usar ON CONFLICT, no capturar 23505. En caso de 40001 / 40P01 / 55P03, reintentar toda la transacción completa, el ciclo alrededor del Do externo. Para problemas de concurrencia sobre las mismas filas, consultar el skill postgres-concurrency.

La verificación de afirmaciones está en CHECKS.md.

Versiones

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 — no inferior a v5.9.2. Corrige una vulnerabilidad de inyección SQL mediante literales dollar-quoted con el protocolo simple. En v5.9.1 se corrigió la corrupción de resultados de batch con consultas preparadas en caché. Verifica la última versión de parche con go list -m -u github.com/jackc/pgx/v5.
• trm y el driver pgxv5 — no inferior a v2.0.4, preferible v2.0.5. En 2.0.4 se eliminó la gorutina de fondo awaitDone del driver: la cancelación del contexto ya no realiza rollback desde otra gorutina (pgx.Tx no es thread-safe, y en una consulta en ejecución causaba pánico). El rollback ahora se produce después de retornar de la función transaccional. En 2.0.5 se corrigió una fuga de contexto en manager.Init en caso de errores. El módulo se versiona con etiquetas como trm/v2.0.5.

Pool

cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
	return nil, fmt.Errorf("parse dsn: %w", err)
}
cfg.MaxConns = 25 // lo que la base pueda soportar, no los núcleos del pod
cfg.MinConns = 2
cfg.ConnConfig.ConnectTimeout = 5 * time.Second
pool, err := pgxpool.NewWithConfig(context.Background(), cfg)

• Define MaxConns explícitamente. El valor por defecto es max(4, runtime.NumCPU()), es decir, el número de núcleos de la máquina, pero el tamaño del pool debe depender de la base de datos. La suma de MaxConns de todos los servicios y réplicas debe estar dentro de max_connections (por defecto 100): cada conexión es un proceso PostgreSQL independiente.
• Otros valores por defecto de pgxpool: MinConns 0, MaxConnLifetime 1 h, MaxConnIdleTime 30 min, HealthCheckPeriod 1 min. Si acortas MaxConnIdleTime, mantén MinConns ≥ 2, de lo contrario, después de un periodo de inactividad la primera consulta pagará el costo de una nueva conexión.
• MaxConnLifetime no soluciona consultas colgadas. Para eso, usa statement_timeout y context.WithTimeout en la consulta: un pool agotado devolverá DeadlineExceeded, no se quedará colgado.
• En el servicio, usa solo pgxpool. pgx.Conn no es thread-safe: desde varias gorutinas obtendrás conn busy y condiciones de carrera. pgx.Conn es para LISTEN/NOTIFY en una conexión dedicada y scripts de un solo uso.
• NewWithConfig no se conecta a la base de datos: el pool es perezoso. La indisponibilidad de la base al inicio solo se detecta en el primer Ping. Sin timeout de conexión, pgx espera el timeout TCP del SO (en Linux, unos dos minutos). Ping con reintento, clasificador de errores de red, hooks PrepareConn y AfterRelease — en references/pool-startup.md.

Administrador de transacciones y repositorio

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))

El repositorio no sabe si está dentro de una transacción: DefaultTrOrDB devuelve la transacción del contexto si existe, de lo contrario, el 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
}

Estructuras y lectura de filas

• El Scan normal no mapea por etiquetas. El estándar es la etiqueta db:"..." y pgx.RowToStructByName[T]; para listas, pgx.CollectRows; para una sola fila, pgx.CollectOneRow (en conjunto vacío, pgx.ErrNoRows). rows.Close() y rows.Err() los manejan automáticamente.
• Si lees manualmente con rows.Next() (con salida temprana), coloca defer rows.Close() inmediatamente después de Query. Las filas no cerradas mantienen la conexión: con MaxConns=3, después de tres fugas la cuarta consulta se cuelga hasta el timeout del contexto.
• Columna nullable — pgtype.Text, pgtype.Timestamptz, etc.: Valid == false significa NULL tanto en lectura como en escritura. Slices de relaciones (Orders []Order) — sin etiqueta, segunda consulta y ensamblado manual.

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

Parámetros de consulta — solo $1, $2

Cualquier valor — teléfono, nombre, cadena de búsqueda — pásalo como argumento $N. No construyas SQL con + o fmt.Sprintf, incluso si la cadena parece segura.

• Exec sin argumentos pgx siempre lo envía por simple protocol (en conn.go: «Always use simple protocol when there are no arguments»), y simple protocol ejecuta varios comandos seguidos: '; DELETE FROM ...; -- pasará. Exec con argumentos rechazará el segundo comando (cannot insert multiple commands into a prepared statement). Query sin argumentos no cambia a simple protocol, pero tampoco se deben concatenar cadenas.
• Extended protocol no protege contra inyección: UNION SELECT es una sola consulta, y pasa por Query. Así se filtran datos personales.
• Los nombres de columna, ORDER BY y la dirección de orden no se pueden pasar como argumento. Tómalos de una lista blanca: un map desde el valor de la API al fragmento SQL preparado; si es desconocido, error.

Transacción mediante 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)
	}) // error — rollback, nil — commit
}

Un Do anidado se une a la transacción abierta, no crea un savepoint. La propagación por defecto es PropagationRequired: un error en la llamada anidada revierte todo por completo. Para un retroceso parcial, solicítalo explícitamente con PropagationNested; el driver hará 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
	}
	// El error revertirá hasta el savepoint, la transacción 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("optional part skipped", "err", err)
	}
	return nil
})

Si no necesitas retroceso parcial, no compliques: un Do anidado es suficiente, pero no aísla nada.

Reintento de transacción

Para cuándo se necesita reintento (repeatable read y serializable, deadlock, lock_timeout, NOWAIT) — ver skill postgres-concurrency. Aquí se explica cómo hacerlo en Go.

• Reintenta toda la transacción, incluyendo la lectura y la lógica que decide qué escribir. Reintentar solo la última operación es inútil: la transacción ya se ha interrumpido.
• El ciclo debe rodear al Do externo. Un Do anidado se une a la transacción abierta, y el reintento dentro de él no iniciará una nueva.
• 40001 puede llegar incluso en COMMIT: el manager lo devolverá envuelto («transaction: commit; ...»), errors.As lo encontrará. Esta es otra razón para mantener todo el Do dentro del ciclo.
• Limita los intentos, entre ellos una pequeña pausa creciente.

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 la lectura y decisiones, 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
}

Errores

Analiza los errores del driver en el repositorio; hacia afuera, devuelve errores de dominio: el caso de uso no debe conocer pgx.

var (
	ErrNotFound = errors.New("not found")
	ErrConflict = errors.New("already exists")
)

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

Si se espera un duplicado, usar ON CONFLICT, no capturar 23505. Cualquier error dentro de Do, incluido 23505, interrumpe la transacción: cada comando posterior obtendrá 25P02 («current transaction is aborted»). Si el duplicado es un resultado esperado (reenvío, doble clic):

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) { // no creado: email ya ocupado
	return 0, fmt.Errorf("create user %q: %w", email, ErrConflict)
}

• Especifica siempre el objetivo del conflicto (columnas). Sin él, DO NOTHING silencia cualquier violación de restricción única, y no se podrá saber cuál funcionó.
• Cero filas en RETURNING — registro no creado: la fila ya existe o acaba de ser confirmada por una transacción paralela. Esto no es éxito.
• La violación de otra restricción única seguirá llegando como 23505 — distínguela por pgErr.ConstraintName.
• Mantén la unicidad con un índice o restricción, no con SELECT antes de INSERT: entre ellos, una transacción paralela puede ocupar el valor.

Códigos frecuentes: 23505 unique_violation, 23503 foreign_key_violation, 23502 not_null_violation, 23P01 exclusion_violation, 40001 serialization_failure, 40P01 deadlock_detected, 55P03 lock_not_available. Tiene sentido reintentar solo los tres últimos, y solo toda la transacción.

Batch y CopyFrom

• pgx.Batch — varias diferentes consultas en un solo roundtrip. Después de SendBatch, lee el resultado de cada consulta y cierra BatchResults.
• CopyFrom — carga masiva mediante el protocolo COPY: el análisis se realiza una vez, no por cada fila. Con 100 mil filas, es aproximadamente un orden de magnitud más rápido que un batch de INSERT.
• CopyFrom no admite ON CONFLICT ni devuelve id. Si necesitas upsert o RETURNING, carga con CopyFrom en una tabla temporal y haz 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
	}),
)

Chuleta

Tarea	Herramienta
Inicio con base no disponible	ConnConfig.ConnectTimeout + Ping con reintento (references/pool-startup.md)
Lista / un registro	CollectRows / CollectOneRow + RowToStructByName
rows.Next() manual	defer rows.Close() inmediatamente después de Query
Consulta en el repositorio	getter.DefaultTrOrDB(ctx, pool)
Valores de entrada	solo $1, $2; columnas y ORDER BY — lista blanca
Transacción	trManager.Do(ctx, fn)
Retroceso parcial	DoWithSettings + trm.PropagationNested
Nivel de aislamiento	trmpgxv5.MustSettings(settings.Must(), trmpgxv5.WithTxOptions(...))
40001 / 40P01 / 55P03	reintento de toda la transacción alrededor del Do externo
Sin fila / duplicado	errors.Is(err, pgx.ErrNoRows) / PgError.Code == "23505"
Duplicado esperado	INSERT ... ON CONFLICT (cols) DO NOTHING RETURNING id
Carga masiva	CopyFrom; consultas diferentes en lote — pgx.Batch
Nullable	pgtype.Text / Timestamptz / ...
Condiciones de carrera, bloqueos, colas	skill postgres-concurrency