pgx + go-transaction-manager

pgx + go-transaction-manager은(는) 무엇을 하나요?

핵심. 리포지토리는 DefaultTrOrDB를 통해 연결을 가져오고, 트랜잭션 경계는 서비스가 trManager.Do로 설정합니다. 쿼리 값은 반드시 $1, $2 인자로만 전달합니다. 예상되는 중복은 23505 오류를 잡는 대신 ON CONFLICT로 처리합니다. 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 이상. simple protocol 사용 시 달러-쿼트(dollar-quoted) 리터럴을 통한 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 // 데이터베이스가 감당할 수 있는 최대치, 파드의 CPU 코어 수가 아님
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은 멈춘(hanging) 요청을 해결하지 못합니다. 멈춤 방지에는 statement_timeout과 요청별 context.WithTimeout을 사용하세요. 풀이 고갈되면 DeadlineExceeded를 반환하고 멈추지 않습니다.
• 서비스에서는 pgxpool만 사용하세요. pgx.Conn은 스레드 안전하지 않습니다. 여러 고루틴에서 사용하면 conn busy 오류와 데이터 경쟁이 발생합니다. pgx.Conn은 전용 연결이 필요한 LISTEN/NOTIFY나 일회성 스크립트에만 사용하세요.
• NewWithConfig는 데이터베이스에 연결하지 않습니다: 풀은 지연(lazy) 방식입니다. 시작 시 데이터베이스에 접근할 수 없는 것은 첫 번째 Ping에서만 확인됩니다. 연결 타임아웃이 없으면 pgx는 OS의 TCP 타임아웃(리눅스 기준 약 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 반환). rows.Close()와 rows.Err()는 함수 내부에서 자동으로 처리됩니다.
• rows.Next()를 수동으로 읽는 경우(조기 종료 시) Query 직후에 defer rows.Close()를 설정하세요. 닫히지 않은 rows는 연결을 점유합니다. MaxConns=3일 때 세 번 누수되면 네 번째 요청은 컨텍스트 타임아웃까지 대기하게 됩니다.
• Nullable 컬럼은 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가 항상 simple protocol로 전송합니다(conn.go: "Always use simple protocol when there are no arguments"). simple protocol은 여러 명령을 연속으로 실행할 수 있습니다. '; DELETE FROM ...; -- 와 같은 입력이 전달될 수 있습니다. 인자가 있는 Exec은 두 번째 명령을 거부합니다(cannot insert multiple commands into a prepared statement). 인자 없는 Query는 simple protocol로 전환되지 않지만, 문자열을 이어붙이는 것은 여전히 금지입니다.
• Extended protocol도 인젝션을 완전히 막지는 못합니다. 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는 열려 있는 트랜잭션에 참여(join)하며, savepoint를 생성하지 않습니다. 기본 전파(propagation)는 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
	}
	// 오류는 savepoint까지만 롤백되고, 외부 트랜잭션은 계속됩니다.
	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
}

오류 처리

드라이버 오류는 리포지토리에서 분류하고, 외부로는 도메인 오류를 반환하세요. usecase가 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)
}

예상되는 중복은 ON CONFLICT로 처리하고 23505를 잡지 마세요. 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으로 구분하세요.
• 고유성은 인덱스나 제약 조건으로 유지하세요. INSERT 전에 SELECT로 확인하지 마세요. 두 작업 사이에 병렬 트랜잭션이 값을 선점할 수 있습니다.

자주 보는 코드: 23505 unique_violation, 23503 foreign_key_violation, 23502 not_null_violation, 23P01 exclusion_violation, 40001 serialization_failure, 40P01 deadlock_detected, 55P03 lock_not_available. 마지막 세 개는 재시도 대상이며, 항상 전체 트랜잭션을 재시도해야 합니다.

Batch와 CopyFrom

• pgx.Batch — 한 번의 왕복(round-trip)에 여러 다른 쿼리를 실행합니다. SendBatch 후 각 쿼리의 결과를 읽고 BatchResults를 닫으세요.
• CopyFrom — COPY 프로토콜을 통한 대량 로드: 각 행마다 파싱하지 않고 한 번만 파싱합니다. 10만 행 기준으로 INSERT 배치보다 약 10배 빠릅니다.
• 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
Nullable	pgtype.Text / Timestamptz / ...
행 경쟁, 잠금, 큐	postgres-concurrency 스킬