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。 该版本修复了一个安全漏洞：在使用简单协议 (simple protocol) 时，通过美元引号字面量 (dollar-quoted literal) 进行的 SQL 注入攻击。v5.9.1 版本修复了在缓存预编译查询时导致批量结果损坏的问题。请通过 go list -m -u github.com/jackc/pgx/v5 查看最新的补丁版本。
trm 和 pgxv5 驱动版本不低于 v2.0.4，最好使用 v2.0.5。 在 2.0.4 版本中，移除了驱动中的后台协程 awaitDone：上下文取消操作不再会从另一个协程触发回滚（pgx.Tx 并非线程安全，在正在执行的查询上执行此操作会导致 panic）。现在，回滚会在事务函数返回后执行。2.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 的 CPU 核心数
cfg.MinConns = 2
cfg.ConnConfig.ConnectTimeout = 5 * time.Second
pool, err := pgxpool.NewWithConfig(context.Background(), cfg)

必须显式设置 MaxConns。 默认值为 max(4, runtime.NumCPU())，即机器的 CPU 核心数，但连接池大小应取决于数据库的承受能力。所有服务和副本的 MaxConns 总和应不超过 PostgreSQL 的 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 会等待操作系统的 TCP 超时（Linux 下大约两分钟）。带重试的 Ping、网络错误分类、PrepareConn 和 AfterRelease 钩子，请参考 参考文献/连接池启动.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 时，泄漏三次后，第四个请求将一直挂起直到上下文超时。
可空列应使用 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"` // 可空
}

查询参数 — 仅使用 `$1, $2`

任何值 — 电话号码、姓名、搜索字符串 — 都必须作为 $N 参数传递。不要通过 + 或 fmt.Sprintf 拼接 SQL，即使字符串看起来是安全的。

不带参数的 Exec 调用，pgx 始终通过简单协议 (simple protocol) 发送（参见 conn.go：“当没有参数时，始终使用简单协议”），而简单协议可以一次执行多条命令：'; DELETE FROM ...; -- 这样的注入会成功。带参数的 Exec 会拒绝第二条命令（cannot insert multiple commands into a prepared statement）。不带参数的 Query 不会切换到简单协议，但同样不允许拼接字符串。
扩展协议 (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)
    }) // 返回错误 => 回滚，返回 nil => 提交
}

嵌套的 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 就足够了，只是它不会隔离任何东西。

事务重试

关于何时需要重试（可重复读和可序列化隔离级别、死锁、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": // 序列化失败、死锁检测到、锁不可用
        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" { // 违反唯一约束
    return 0, fmt.Errorf("create user %q: %w", email, ErrConflict)
}

预期中的重复数据应通过 ON CONFLICT 处理，而不是捕获 23505 错误。 在 Do 内部发生的任何错误（包括 23505）都会终止事务：之后的每条命令都会收到 25P02（“当前事务已中止”）。如果重复是预期中的结果（例如，重复提交、双击）：

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) { // 未创建：邮箱已被占用
    return 0, fmt.Errorf("create user %q: %w", email, ErrConflict)
}

冲突目标 (列名) 必须始终指定。如果不指定，DO NOTHING 会抑制任何唯一约束的违反，你将无法知道是哪个约束被触发了。
RETURNING 返回零行意味着记录未创建：记录已存在，或者刚刚被另一个并发事务提交了。这并非成功。
违反其他唯一约束时，仍然会收到 23505 错误 — 可以通过 pgErr.ConstraintName 区分。
唯一性应通过索引或约束来保证，而不是使用 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。只有最后三个错误值得重试，并且必须重试整个事务。

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 (`参考文献/连接池启动.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`
可空字段	`pgtype.Text` / `Timestamptz` / ...
行竞争、锁、队列	参考 `postgres-concurrency` 技能

pgx + go-transaction-manager