pgx + go-transaction-manager

Que fait pgx + go-transaction-manager ?

Essentiel. Le repository obtient une connexion via DefaultTrOrDB, les limites de la transaction sont définies par le service via trManager.Do. Les valeurs dans les requêtes sont uniquement passées en arguments $1, $2. Un doublon attendu se gère avec ON CONFLICT, pas en attrapant 23505. Sur 40001 / 40P01 / 55P03, répète la transaction entière, la boucle se fait autour du Do externe. Pour les problèmes de concurrence sur les mêmes lignes, voir le skill postgres-concurrency.

Vérification des assertions dans CHECKS.md.

Versions

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 — minimum v5.9.2. Elle corrige une vulnérabilité : injection SQL via des littéraux dollar-quoted avec le protocole simple. En v5.9.1, on a corrigé la corruption des résultats de batch avec des requêtes préparées en cache. Vérifie la dernière version de patch avec go list -m -u github.com/jackc/pgx/v5.
• trm et le driver pgxv5 — minimum v2.0.4, de préférence v2.0.5. En 2.0.4, la goroutine d'arrière-plan awaitDone a été retirée du driver : l'annulation du contexte ne provoque plus de rollback depuis une autre goroutine (pgx.Tx n'est pas thread-safe, ce qui pouvait causer une panique sur une requête en cours d'exécution). Le rollback est maintenant effectué après le retour de la fonction transactionnelle. En 2.0.5, une fuite de contexte dans manager.Init sur les erreurs a été corrigée. Le module est versionné avec des tags de la forme trm/v2.0.5.

Pool

cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
	return nil, fmt.Errorf("parse dsn: %w", err)
}
cfg.MaxConns = 25 // en fonction de ce que la base peut supporter, pas du nombre de cœurs du pod
cfg.MinConns = 2
cfg.ConnConfig.ConnectTimeout = 5 * time.Second
pool, err := pgxpool.NewWithConfig(context.Background(), cfg)

• Définis MaxConns explicitement. La valeur par défaut est max(4, runtime.NumCPU()), soit le nombre de cœurs de la machine, alors que la taille du pool doit dépendre de la base. La somme des MaxConns de tous les services et réplicas doit tenir dans max_connections (par défaut 100) : chaque connexion est un processus PostgreSQL séparé.
• Autres valeurs par défaut de pgxpool : MinConns 0, MaxConnLifetime 1 h, MaxConnIdleTime 30 min, HealthCheckPeriod 1 min. Si tu réduis MaxConnIdleTime, garde MinConns ≥ 2, sinon après une période d'inactivité, la prochaine requête paiera le coût d'une nouvelle connexion.
• MaxConnLifetime ne résout pas les requêtes bloquées. Pour cela, utilise statement_timeout et context.WithTimeout sur la requête : un pool épuisé retournera DeadlineExceeded, sans bloquer.
• Dans le service, utilise uniquement pgxpool. pgx.Conn n'est pas thread-safe : depuis plusieurs goroutines, tu obtiendras conn busy et des race conditions. pgx.Conn est réservé pour LISTEN/NOTIFY sur une connexion dédiée et des scripts ponctuels.
• NewWithConfig ne se connecte pas à la base : le pool est paresseux. L'indisponibilité de la base au démarrage n'est visible que lors du premier Ping. Sans timeout de connexion, pgx attend le timeout TCP du système d'exploitation (sur Linux, environ deux minutes). Ping avec répétition, classifieur d'erreurs réseau, hooks PrepareConn et AfterRelease dans references/pool-startup.md.

Gestionnaire de transactions et repository

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

Le repository ne sait pas s'il est en transaction : DefaultTrOrDB retourne la transaction depuis le contexte si elle existe, sinon le 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
}

Structures et lecture de lignes

• Le Scan basique ne fait pas de mapping par tags. La norme est le tag db:"..." avec pgx.RowToStructByName[T] ; une liste avec pgx.CollectRows, une seule ligne avec pgx.CollectOneRow (sur un résultat vide, retourne pgx.ErrNoRows). rows.Close() et rows.Err() sont gérés automatiquement.
• Si tu lis manuellement avec rows.Next() (avec sortie anticipée), mets defer rows.Close() immédiatement après Query. Des rows non fermés maintiennent la connexion : avec MaxConns=3, après trois fuites, la quatrième requête bloque jusqu'au timeout du contexte.
• Colonne nullable : pgtype.Text, pgtype.Timestamptz, etc. Valid == false correspond à NULL, aussi bien en lecture qu'en écriture. Les slices de relations (Orders []Order) : sans tag, deuxième requête et assemblage manuel.

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

Paramètres de requête — uniquement $1, $2

Toute valeur — numéro de téléphone, nom, chaîne de recherche — doit être passée comme argument $N. Ne construis pas de SQL avec + ou fmt.Sprintf, même si la chaîne semble sûre.

• Exec sans arguments envoie toujours par simple protocol (dans conn.go : « Always use simple protocol when there are no arguments »), et le simple protocol exécute plusieurs commandes à la suite : '; DELETE FROM ...; -- passera. Exec avec des arguments rejettera une deuxième commande (cannot insert multiple commands into a prepared statement). Query sans arguments ne bascule pas en simple protocol, mais tu ne dois pas non plus concaténer de chaînes.
• Le protocole étendu ne protège pas contre l'injection : UNION SELECT est une seule requête, et elle passera via Query. C'est ainsi que des données personnelles fuient.
• Les noms de colonnes, ORDER BY et la direction du tri ne peuvent pas être passés comme arguments. Utilise une liste blanche : map de la valeur de l'API vers un fragment SQL prêt ; pour une valeur inconnue, retourne une erreur.

Transaction 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)
	}) // erreur → rollback, nil → commit
}

Un Do imbriqué se joint à la transaction ouverte, il ne crée pas de savepoint. La propagation par défaut est PropagationRequired : une erreur dans l'appel imbriqué annule tout. Pour une annulation partielle, demande explicitement PropagationNested ; le driver créera un 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
	}
	// L'erreur annulera jusqu'au savepoint, la transaction externe continue.
	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 une annulation partielle n'est pas nécessaire, ne complique pas : un Do imbriqué suffit, il n'isole simplement rien.

Répétition de transaction

Quand la répétition est nécessaire (repeatable read et serializable, deadlock, lock_timeout, NOWAIT), voir le skill postgres-concurrency. Ici, comment le faire en Go.

• Répète la transaction entière, y compris la lecture et la logique qui décide quoi écrire. Répéter la dernière instruction est inutile : la transaction est déjà avortée.
• La boucle doit entourer le Do externe. Un Do imbriqué se joint à la transaction ouverte, et une répétition à l'intérieur ne démarrera pas une nouvelle transaction.
• 40001 peut aussi arriver sur COMMIT : le gestionnaire le retournera encapsulé ("transaction: commit; ..."), errors.As le trouvera. C'est une raison supplémentaire de garder le Do entier dans la boucle.
• Limite le nombre de tentatives, avec une petite pause croissante entre elles.

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) // toutes les lectures et toutes les décisions sont à l'intérieur
		})
		if !isRetryableTxErr(err) {
			return err
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(i+1) * 20 * time.Millisecond):
		}
	}
	return err
}

Erreurs

Analyse les erreurs du driver dans le repository, et remonte des erreurs métier vers l'extérieur : le usecase ne doit pas connaître 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)
}

Un doublon attendu se gère avec ON CONFLICT, pas en attrapant 23505. Toute erreur dans Do, y compris 23505, interrompt la transaction : chaque commande suivante recevra 25P02 (« current transaction is aborted »). Si le doublon est un résultat normal (renvoi, double 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) { // non créé : email déjà pris
	return 0, fmt.Errorf("create user %q: %w", email, ErrConflict)
}

• Spécifie toujours la cible du conflit (colonnes). Sans elle, DO NOTHING étouffe toute violation de contrainte d'unicité, et on ne peut plus savoir laquelle a été déclenchée.
• Zéro lignes dans RETURNING signifie que l'enregistrement n'a pas été créé : la ligne existe déjà ou vient d'être validée par une transaction parallèle. Ce n'est pas un succès.
• La violation d'une autre contrainte d'unicité arrivera toujours comme 23505 — distingue-les avec pgErr.ConstraintName.
• Garantis l'unicité avec un index ou une contrainte, pas avec un SELECT avant INSERT : entre les deux, une transaction parallèle aura le temps de prendre la valeur.

Codes fréquents : 23505 unique_violation, 23503 foreign_key_violation, 23502 not_null_violation, 23P01 exclusion_violation, 40001 serialization_failure, 40P01 deadlock_detected, 55P03 lock_not_available. Les trois derniers ont un sens à être répétés, et seulement la transaction entière.

Batch et CopyFrom

• pgx.Batch — plusieurs requêtes différentes en un seul aller-retour. Après SendBatch, lis le résultat de chaque requête et ferme BatchResults.
• CopyFrom — chargement massif via le protocole COPY : l'analyse n'a lieu qu'une fois, pas pour chaque ligne. Pour 100 000 lignes, il est environ dix fois plus rapide qu'un batch d'INSERT.
• CopyFrom ne gère pas ON CONFLICT et ne retourne pas d'id. Si tu as besoin d'upsert ou de RETURNING, charge d'abord avec CopyFrom dans une table temporaire, puis fais 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
	}),
)

Aide-mémoire

Tâche	Outil
Démarrage avec base indisponible	ConnConfig.ConnectTimeout + Ping avec répétition (references/pool-startup.md)
Liste / un seul enregistrement	CollectRows / CollectOneRow + RowToStructByName
rows.Next() manuel	defer rows.Close() immédiatement après Query
Requête dans le repository	getter.DefaultTrOrDB(ctx, pool)
Valeurs issues de l'entrée utilisateur	uniquement $1, $2 ; colonnes et ORDER BY avec liste blanche
Transaction	trManager.Do(ctx, fn)
Annulation partielle	DoWithSettings + trm.PropagationNested
Niveau d'isolation	trmpgxv5.MustSettings(settings.Must(), trmpgxv5.WithTxOptions(...))
40001 / 40P01 / 55P03	répétition de la transaction entière autour du Do externe
Ligne absente / doublon	errors.Is(err, pgx.ErrNoRows) / PgError.Code == "23505"
Doublon comme résultat normal	INSERT ... ON CONFLICT (cols) DO NOTHING RETURNING id
Chargement massif	CopyFrom ; différentes requêtes en une fois — pgx.Batch
Nullable	pgtype.Text / Timestamptz / ...
Concurrence sur les lignes, verrous, files d'attente	skill postgres-concurrency