pgx + go-transaction-manager

重要ポイント. リポジトリは DefaultTrOrDB を使って接続を取得し、トランザクションの境界はサービス側で trManager.Do を使って定義します。クエリへの値の受け渡しは必ず $1, $2 のようなプレースホルダを使います。重複が想定される場合は ON CONFLICT を使い、エラーコード 23505 をキャッチするのではありません。40001 / 40P01 / 55P03 のエラーが発生した場合は、トランザクション全体をリトライします。ループは外側の Do を囲むように記述します。同一行への競合アクセスについては、postgres-concurrency スキルを参照してください。

各種主張の確認は CHECKS.md を参照。

バージョン

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 は v5.9.2 以上必須。 このバージョンで、シンプルプロトコル利用時のドルクォートリテラルによるSQLインジェクションの脆弱性が修正されています。v5.9.1では、キャッシュされたプリペアドステートメント利用時のバッチ結果の破損が修正されました。最新のパッチバージョンは go list -m -u github.com/jackc/pgx/v5 で確認してください。
trm および pgxv5 ドライバは v2.0.4 以上、できれば v2.0.5 以上必須。 v2.0.4では、ドライバからバックグラウンドゴルーチン awaitDone が削除されました。これにより、コンテキストのキャンセルが別のゴルーチンからロールバックを実行することがなくなりました（pgx.Tx はスレッドセーフではないため、実行中のクエリがあるとパニックを引き起こしていました）。ロールバックはトランザクション関数からのリターン後に実行されるようになりました。v2.0.5では、manager.Init のエラー時におけるコンテキストリークが修正されました。モジュールは trm/v2.0.5 のようなタグでバージョン管理されます。

プール

cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
	return nil, fmt.Errorf("parse dsn: %w", err)
}
cfg.MaxConns = 25 // データベースが処理可能な数に設定。Podのコア数ではない。
cfg.MinConns = 2
cfg.ConnConfig.ConnectTimeout = 5 * time.Second
pool, err := pgxpool.NewWithConfig(context.Background(), cfg)

MaxConns は明示的に設定すること。 デフォルト値は max(4, runtime.NumCPU())、つまりマシンのコア数ですが、プールサイズはデータベースの負荷容量に基づくべきです。全サービスとレプリカの MaxConns の合計が、データベースの max_connections（デフォルトは100）に収まるように設定します。各接続はPostgreSQLの個別のプロセスに対応します。
pgxpoolのその他のデフォルト値: MinConns は0、MaxConnLifetime は1時間、MaxConnIdleTime は30分、HealthCheckPeriod は1分です。MaxConnIdleTime を短くする場合は、MinConns を2以上に設定してください。そうしないと、アイドル期間後に最初のクエリで新しい接続を確立するための待ち時間が発生します。
MaxConnLifetime はハングしたクエリを解決しません。 ハングしたクエリへの対策としては、statement_timeout とクエリ単位の context.WithTimeout を使用します。これにより、プールが枯渇した場合、ハングする代わりに DeadlineExceeded エラーが返ります。
サービス内では pgxpool のみを使用すること。 pgx.Conn はスレッドセーフではありません。複数のゴルーチンからアクセスすると conn busy エラーやデータ競合が発生します。pgx.Conn は、専用接続が必要な LISTEN/NOTIFY や、使い捨てスクリプトでのみ使用してください。
NewWithConfig はデータベースに接続しません。 プールはレイジーに初期化されます。起動時のデータベース障害は、最初の Ping が行われるまで検出されません。接続タイムアウトを設定しない場合、pgxはOSのTCPタイムアウト（Linuxでは約2分）を待ちます。リトライ付きPing、ネットワークエラーの分類、PrepareConn や AfterRelease フックについては、references/pool-startup.md を参照してください。

トランザクションマネージャとリポジトリ

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

リポジトリは自身がトランザクション内で実行されているかどうかを認識しません。DefaultTrOrDB は、コンテキストにトランザクションが存在すればそれを返し、そうでなければプールを返します。

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
}

構造体と行の読み取り

通常の Scan はタグに基づくマッピングを行いません。標準的な方法は、db:"..." タグと pgx.RowToStructByName[T] の組み合わせです。複数行の取得には pgx.CollectRows、単一行の取得には pgx.CollectOneRow を使用します（空の結果セットの場合、pgx.ErrNoRows が返ります）。CollectRows と CollectOneRow は、内部的に rows.Close() と rows.Err() を適切に処理します。
rows.Next() を手動で使う場合（早期 return がある場合など）は、Query の直後に defer rows.Close() を記述してください。クローズされていない rows は接続を保持し続けます。MaxConns=3 の場合、3つのリークが発生すると、4番目のクエリはコンテキストのタイムアウトまでハングします。
NULL 許容カラムには、pgtype.Text、pgtype.Timestamptz などを使用します。 Valid == false は、読み取り時も書き込み時も NULL として扱われます。関連するスライス（Orders []Order など）はタグなしで管理し、個別のクエリと手動でのアセンブルが必要です。

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

クエリパラメータは `$1, $2` のみ

電話番号、名前、検索文字列など、あらゆる値は $N プレースホルダを使って引数として渡してください。たとえ安全に見える文字列でも、+ や fmt.Sprintf でSQLを組み立てないでください。

引数なしの Exec は、pgxによって常に シンプルプロトコル で送信されます（conn.go に「引数がない場合は常にシンプルプロトコルを使用する」とあります）。シンプルプロトコルでは複数のコマンドを順次実行できるため、'; DELETE FROM ...; -- といった攻撃が成功する可能性があります。引数付きの Exec は、複数のコマンドを含む文字列を拒否します（cannot insert multiple commands into a prepared statement）。引数なしの Query はシンプルプロトコルに切り替わりませんが、文字列連結は避けるべきです。
拡張プロトコルもSQLインジェクションを完全には防ぎません。UNION SELECT は単一のクエリとして扱われるため、Query 経由で実行可能です。これにより個人情報が漏洩する可能性があります。
カラム名や ORDER BY の方向などを引数で渡すことはできません。これらは、APIから受け取った値を安全なSQLフラグメントにマッピングする map のようなホワイトリストを使用し、未知の値に対してはエラーを返すようにしてください。

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)
	}) // エラーならロールバック、nilならコミット
}

入れ子になった Do は、セーブポイントを作成する代わりに、既存のトランザクションに参加します。 デフォルトの伝播動作は PropagationRequired です。入れ子になった呼び出しでエラーが発生すると、トランザクション全体がロールバックされます。部分的なロールバックが必要な場合は、明示的に PropagationNested を指定します。ドライバが 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
	}
	// エラーが発生した場合、セーブポイントまでロールバックされ、外側のトランザクションは継続されます。
	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
})

部分的なロールバックが不要な場合は、複雑にしないでください。入れ子になった Do で十分ですが、それは何も分離しません。

トランザクションのリトライ

リトライが必要なケース（repeatable read や serializable、デッドロック、lock_timeout、NOWAIT）については、postgres-concurrency スキルを参照してください。ここでは、Goでこれを実装する方法を示します。

トランザクション全体 をリトライします。読み取りと、書き込む内容を決定するロジックも含めてリトライします。最後のステートメントのみをリトライしても意味がありません。トランザクションは既に中断されているからです。
ループは外側の Do を囲むように記述します。入れ子になった Do は既存のトランザクションに参加するため、その内部でリトライしても新しいトランザクションは開始されません。
40001 エラーは COMMIT 時にも発生する可能性があります。マネージャはこれをラップして返します（"transaction: commit; ..."）。errors.As でエラーを検出できます。これも、ループで Do 全体を囲む理由の一つです。
試行回数を制限し、試行間には短い増分の待機時間を設けます。

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) // 全ての読み取りと決定はこの内部で行う
		})
		if !isRetryableTxErr(err) {
			return err
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(i+1) * 20 * time.Millisecond):
		}
	}
	return err
}

エラー

ドライバのエラーはリポジトリ内で解析し、外側にはドメインエラーを返すようにします。ユースケースは 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)
}

重複が想定される場合は、エラーコード 23505 をキャッチするのではなく、ON CONFLICT を使用します。 Do 内部でのあらゆるエラー（23505 を含む）はトランザクションを中断します。その後、後続のコマンドはすべて 25P02（「current transaction is aborted」）エラーになります。重複が通常の結果である場合（再送信、ダブルクリックなど）は、次のようにします。

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) { // 作成されなかった: email は既に使用中
	return 0, fmt.Errorf("create user %q: %w", email, ErrConflict)
}

競合のターゲット (カラム) は常に指定してください。これがないと DO NOTHING はすべてのユニーク制約違反を抑制し、どの制約がトリガーされたかを特定できなくなります。
RETURNING で0行が返ってきた場合は、レコードが作成されなかったことを意味します。行が既に存在するか、並行トランザクションによってちょうどコミットされたところです。これは成功ではありません。
別のユニーク制約の違反は、依然として 23505 として返されます。区別するには pgErr.ConstraintName を使用します。
ユニーク性は、SELECT してから INSERT するのではなく、インデックスや制約で保証してください。SELECT と INSERT の間に、並行トランザクションが値を取得する可能性があります。

よくあるエラーコード：23505 unique_violation、23503 foreign_key_violation、23502 not_null_violation、23P01 exclusion_violation、40001 serialization_failure、40P01 deadlock_detected、55P03 lock_not_available。リトライが有効なのは最後の3つであり、かつトランザクション全体に対してのみです。

バッチと CopyFrom

pgx.Batch は、複数の 異なる クエリを1ラウンドトリップで実行します。SendBatch の後は、各クエリの結果を読み取り、BatchResults をクローズしてください。
CopyFrom は、COPY プロトコルを使用した一括ロードを行います。クエリの解析は1回で済み、行ごとに解析する必要はありません。10万行の場合、INSERT のバッチよりも約1桁高速です。
CopyFrom は ON CONFLICT をサポートしておらず、id を返しません。upsert や RETURNING が必要な場合は、CopyFrom で一時テーブルにロードし、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
	}),
)

チートシート

タスク	ツール
データベースが利用不可の場合の起動	`ConnConfig.ConnectTimeout` + リトライ付き Ping（`references/pool-startup.md`）
複数行 / 単一行の取得	`CollectRows` / `CollectOneRow` + `RowToStructByName`
手動 `rows.Next()`	`Query` 直後に `defer rows.Close()`
リポジトリでのクエリ実行	`getter.DefaultTrOrDB(ctx, pool)`
ユーザー入力値	`$1, $2` のみ；カラム名と `ORDER BY` はホワイトリスト
トランザクション	`trManager.Do(ctx, fn)`
部分的なロールバック	`DoWithSettings` + `trm.PropagationNested`
分離レベルの指定	`trmpgxv5.MustSettings(settings.Must(), trmpgxv5.WithTxOptions(...))`
`40001` / `40P01` / `55P03`	外側の `Do` を囲んでトランザクション全体をリトライ
行なし / 重複エラー	`errors.Is(err, pgx.ErrNoRows)` / `PgError.Code == "23505"`
重複が想定される場合	`INSERT ... ON CONFLICT (cols) DO NOTHING RETURNING id`
一括ロード	`CopyFrom`；複数の異なるクエリは `pgx.Batch`
NULL 許容カラム	`pgtype.Text` / `Timestamptz` / ...
行の競合、ロック、キュー	`postgres-concurrency` スキルを参照

scary-production/claude-plugins

claude-plugins とは？

お気に入りのAIに質問する

ドキュメント