scary-production/claude-plugins

pgx + go-transaction-manager

核心要點。 Repository 透過 DefaultTrOrDB 取得連線，交易邊界由 service 透過 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。 此版本修復了一個安全漏洞：在使用 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 中，移除了驅動程式中的背景 goroutine awaitDone：取消 context 不再會從另一個 goroutine 執行 rollback（pgx.Tx 不具備執行緒安全性，在正在執行的查詢上這樣做會導致 panic）。現在 rollback 會在交易函式返回後才執行。v2.0.5 則修復了 manager.Init 在發生錯誤時的 context 洩漏問題。模組版本使用類似 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 不具備執行緒安全性：從多個 goroutine 使用它會導致 conn busy 錯誤和資料競爭。pgx.Conn 僅適用於在專用連線上執行 LISTEN/NOTIFY 或一次性腳本。
• NewWithConfig 不會立即連線到資料庫：連線池是惰性的。資料庫在啟動時不可用，只有在第一次 Ping 時才會被發現。如果沒有設定連線超時，pgx 會等待作業系統的 TCP 超時（在 Linux 上約為兩分鐘）。關於帶有重試機制的 Ping、網路錯誤分類、PrepareConn 和 AfterRelease 鉤子，請參閱 references/pool-startup.md。

交易管理器與 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))

Repository 不需要知道它是否在交易中：DefaultTrOrDB 會根據 context 返回一個交易（如果存在），否則返回連線池。

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，在三次洩漏後，第四個請求會一直掛起直到 context 超時。
• 可為 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 總是使用 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 和排序方向無法作為參數傳遞。應從白名單中獲取它們：使用一個 map 將 API 傳入的值映射到預先準備好的 SQL 片段，對於未知的值則返回錯誤。

透過 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)
	}) // 錯誤則 rollback，nil 則 commit
}

巢狀的 Do 會加入已開啟的交易，而非建立 savepoint。 預設的傳播行為是 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
}

錯誤處理

在 Repository 層解析驅動程式的錯誤，對外返回領域錯誤：use case 不應該知道 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 返回零行表示記錄未建立：該行已存在或剛被並行事務提交。這不是成功。
• 違反其他唯一約束仍會返回 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 — 在一次往返中執行多個不同的查詢。呼叫 SendBatch 後，需要讀取每個查詢的結果並關閉 BatchResults。
• CopyFrom — 透過 COPY 協定進行大量資料載入：只解析一次，而不是每行都解析。對於 10 萬行資料，它比 INSERT 的 batch 快大約一個數量級。
• 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()
Repository 中的查詢	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 技能