pgx + go-transaction-manager

Wichtig. Das Repository bezieht eine Verbindung über DefaultTrOrDB, Transaktionsgrenzen setzt der Service über trManager.Do. Werte in Abfragen – nur als Argumente $1, $2. Erwartete Duplikate – ON CONFLICT, nicht abgefangenes 23505. Bei 40001 / 40P01 / 55P03 wiederhole die gesamte Transaktion, die Schleife um das äußere Do. Was tun bei Wettläufen um dieselben Zeilen – Skill postgres-concurrency.

Überprüfung der Behauptungen – CHECKS.md.

Versionen

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 – nicht niedriger als v5.9.2. Darin wurde eine Sicherheitslücke geschlossen: SQL-Injection via Dollar-quoted Literale im Simple Protocol. In v5.9.1 wurde die Beschädigung von Batch-Ergebnissen bei gecachten Prepared Statements behoben. Die neueste Patch-Version ermittle über go list -m -u github.com/jackc/pgx/v5.
trm und Treiber pgxv5 – nicht niedriger als v2.0.4, besser v2.0.5. In 2.0.4 wurde aus dem Treiber die Hintergrund-Goroutine awaitDone entfernt: Kontext-Abbruch führt nicht mehr zu einem Rollback aus einer anderen Goroutine (pgx.Tx ist nicht threadsicher, bei laufender Abfrage führte dies zu einer Panik). Rollback erfolgt jetzt nach Rückkehr aus der Transaktionsfunktion. In 2.0.5 wurde ein Context-Leak in manager.Init bei Fehlern behoben. Das Modul wird mit Tags der Form trm/v2.0.5 versioniert.

Pool

cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
	return nil, fmt.Errorf("parse dsn: %w", err)
}
cfg.MaxConns = 25 // wie viel die Datenbank verkraftet, nicht wie viele Kerne der Pod hat
cfg.MinConns = 2
cfg.ConnConfig.ConnectTimeout = 5 * time.Second
pool, err := pgxpool.NewWithConfig(context.Background(), cfg)

MaxConns explizit setzen. Standard ist max(4, runtime.NumCPU()), also die Anzahl der CPU-Kerne der Maschine, aber die Poolgröße sollte von der Datenbank abhängen. Die Summe der MaxConns aller Dienste und Replikate sollte sich in max_connections (standardmäßig 100) einfügen: Jede Verbindung ist ein separater PostgreSQL-Prozess.
Weitere Standardwerte von pgxpool: MinConns 0, MaxConnLifetime 1 Std., MaxConnIdleTime 30 Min., HealthCheckPeriod 1 Min. Verkürzt du MaxConnIdleTime – halte MinConns >= 2, sonst zahlt die Abfrage nach einer Leerlaufzeit eine neue Verbindung.
MaxConnLifetime heilt keine hängenden Abfragen. Dagegen hilft statement_timeout und context.WithTimeout auf die Abfrage: Ein erschöpfter Pool gibt DeadlineExceeded zurück, statt hängen zu bleiben.
Im Service – nur pgxpool. pgx.Conn ist nicht threadsicher: Aus mehreren Goroutinen erhältst du conn busy und Daten-Races. pgx.Conn – für LISTEN/NOTIFY auf einer dedizierten Verbindung und einmalige Skripte.
NewWithConfig verbindet sich nicht mit der Datenbank: Der Pool ist lazy. Die Nichtverfügbarkeit der Datenbank beim Start ist erst beim ersten Ping sichtbar. Ohne Timeout für den Verbindungsaufbau wartet pgx auf den TCP-Timeout des Betriebssystems (unter Linux etwa zwei Minuten). Ping mit Wiederholung, Klassifikator für Netzwerkfehler, Hooks PrepareConn und AfterRelease – references/pool-startup.md.

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

Das Repository weiß nicht, ob es sich in einer Transaktion befindet: DefaultTrOrDB gibt die Transaktion aus dem Kontext zurück, falls vorhanden, sonst den 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
}

Strukturen und Zeilen lesen

Einfaches Scan über Tags mapped nicht. Standard – Tag db:"..." und pgx.RowToStructByName[T]; Liste – pgx.CollectRows, eine Zeile – pgx.CollectOneRow (bei leerem Ergebnis – pgx.ErrNoRows). rows.Close() und rows.Err() erledigen sie selbst.
Liest du manuell über rows.Next() (mit frühem Ausstieg) – setze direkt nach Query defer rows.Close(). Nicht geschlossene Rows halten eine Verbindung: Bei MaxConns=3 hängt nach drei Leaks die vierte Abfrage bis zum Context-Timeout.
Nullable-Spalte – pgtype.Text, pgtype.Timestamptz usw.: Valid == false bedeutet NULL sowohl beim Lesen als auch beim Schreiben. Slice von Beziehungen (Orders []Order) – ohne Tag, zweite Abfrage und manuelle Zusammenstellung.

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

Abfrageparameter – nur `$1, $2`

Jeden Wert – Telefonnummer, Name, Suchzeichenfolge – als Argument $N übergeben. SQL nicht über + oder fmt.Sprintf zusammenbauen, selbst wenn der String harmlos aussieht.

Exec ohne Argumente sendet pgx immer über das Simple Protocol (in conn.go: „Always use simple protocol when there are no arguments“), und das Simple Protocol führt mehrere Befehle nacheinander aus: '; DELETE FROM ...; -- wird durchkommen. Exec mit Argumenten weist den zweiten Befehl ab (cannot insert multiple commands into a prepared statement). Query ohne Argumente wechselt nicht zum Simple Protocol, aber Strings darf man auch dort nicht zusammenkleben.
Extended Protocol schützt nicht vor Injection: UNION SELECT ist eine einzelne Abfrage und geht durch Query. So gelangen persönliche Daten nach außen.
Spaltennamen, ORDER BY und Sortierrichtung kann man nicht als Argument übergeben. Nimm sie aus einer Whitelist: map vom Wert aus der API zum fertigen SQL-Fragment, bei unbekanntem Wert – Fehler.

Transaktion über 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)
	}) // Fehler – rollback, nil – commit
}

Verschachteltes Do wird der offenen Transaktion beigetreten, erstellt aber keinen Savepoint. Standard-Propagation ist PropagationRequired: Ein Fehler im verschachtelten Aufruf rollt alles vollständig zurück. Teilweises Zurücksetzen fordere explizit an – PropagationNested, der Treiber erstellt 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
	}
	// Ein Fehler rollt bis zum Savepoint zurück, die äußere Transaktion wird fortgesetzt.
	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
})

Teilweises Zurücksetzen ist nicht nötig – mach es nicht kompliziert: Ein verschachteltes Do reicht, es isoliert einfach nichts.

Wiederholung der Transaktion

Wenn eine Wiederholung nötig ist (repeatable read und serializable, Deadlock, lock_timeout, NOWAIT) – Skill postgres-concurrency. Hier – wie man das in Go macht.

Wiederhole die gesamte Transaktion, inklusive Lesen und Logik, die entscheidet, was geschrieben wird. Eine Wiederholung der letzten Anweisung ist nutzlos: Die Transaktion ist bereits abgebrochen.
Die Schleife – um das äußere Do. Ein verschachteltes Do tritt einer offenen Transaktion bei und eine Wiederholung darin startet keine neue.
40001 kann auch beim COMMIT kommen: Der Manager gibt ihn zurück, eingewickelt („transaction: commit; ...“), errors.As findet ihn. Das ist ein weiterer Grund, die gesamte Do in der Schleife zu halten.
Versuche begrenzen, dazwischen – eine kleine, wachsende Pause.

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) // alles Lesen und alle Entscheidungen – darin
		})
		if !isRetryableTxErr(err) {
			return err
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(i+1) * 20 * time.Millisecond):
		}
	}
	return err
}

Fehler

Fehler des Treibers im Repository analysieren, nach außen Domänenfehler ausgeben: Der Use-Case sollte nichts von pgx wissen.

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

Erwartete Duplikate – ON CONFLICT, nicht abgefangenes 23505. Jeder Fehler innerhalb von Do, einschließlich 23505, bricht die Transaktion ab: Jeder weitere Befehl erhält 25P02 („current transaction is aborted“). Ist ein Duplikat ein normaler Ausgang (erneutes Senden, Doppelklick):

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) { // nicht erstellt: E-Mail bereits belegt
	return 0, fmt.Errorf("create user %q: %w", email, ErrConflict)
}

Das Ziel des Konflikts (Spalten) immer angeben. Ohne sie unterdrückt DO NOTHING die Verletzung jeder eindeutigen Einschränkung, und welche ausgelöst wurde, ist nicht mehr erkennbar.
Null Zeilen in RETURNING – Datensatz nicht erstellt: Die Zeile existiert bereits oder wurde gerade von einer parallelen Transaktion bestätigt. Das ist kein Erfolg.
Die Verletzung einer anderen eindeutigen Einschränkung kommt trotzdem als 23505 – unterscheide anhand von pgErr.ConstraintName.
Eindeutigkeit durch Index oder Einschränkung sicherstellen, nicht durch SELECT vor INSERT: Dazwischen könnte eine parallele Transaktion den Wert bereits belegt haben.

Häufige Codes: 23505 unique_violation, 23503 foreign_key_violation, 23502 not_null_violation, 23P01 exclusion_violation, 40001 serialization_failure, 40P01 deadlock_detected, 55P03 lock_not_available. Wiederholen sinnvoll bei den letzten drei, und nur die gesamte Transaktion.

Batch und CopyFrom

pgx.Batch – mehrere verschiedene Abfragen in einem Roundtrip. Nach SendBatch das Ergebnis jeder Abfrage lesen und BatchResults schließen.
CopyFrom – Massenladen über das COPY-Protokoll: Parsing einmal, nicht für jede Zeile. Bei 100.000 Zeilen ist es etwa eine Größenordnung schneller als ein Batch von INSERT.
CopyFrom kann kein ON CONFLICT und gibt keine IDs zurück. Brauchst du Upsert oder RETURNING – lade mit CopyFrom in eine temporäre Tabelle und mache 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
	}),
)

Spickzettel

Aufgabe	Werkzeug
Start bei nicht verfügbarer Datenbank	`ConnConfig.ConnectTimeout` + Ping mit Wiederholung (`references/pool-startup.md`)
Liste / einzelner Datensatz	`CollectRows` / `CollectOneRow` + `RowToStructByName`
Manuelles `rows.Next()`	`defer rows.Close()` direkt nach `Query`
Abfrage im Repository	`getter.DefaultTrOrDB(ctx, pool)`
Eingabewerte	nur `$1, $2`; Spalten und `ORDER BY` – Whitelist
Transaktion	`trManager.Do(ctx, fn)`
Teilweises Zurücksetzen	`DoWithSettings` + `trm.PropagationNested`
Isolationslevel	`trmpgxv5.MustSettings(settings.Must(), trmpgxv5.WithTxOptions(...))`
`40001` / `40P01` / `55P03`	Wiederholung der gesamten Transaktion um das äußere `Do`
Keine Zeile / Duplikat	`errors.Is(err, pgx.ErrNoRows)` / `PgError.Code == "23505"`
Duplikat – normaler Ausgang	`INSERT ... ON CONFLICT (cols) DO NOTHING RETURNING id`
Massenladen	`CopyFrom`; verschiedene Abfragen auf einmal – `pgx.Batch`
Nullable	`pgtype.Text` / `Timestamptz` / ...
Wettläufe um Zeilen, Sperren, Warteschlangen	Skill `postgres-concurrency`

pgx + go-transaction-manager (Deutsch)

Was ist pgx + go-transaction-manager (Deutsch)?

In Ihrer bevorzugten KI fragen

Dokumentation