Транзакции в Go — pgx.Tx на handler'е

Опирается на правила: R-SQLC-TX-1 … R-SQLC-TX-4 и R-SQLC-TX-X1 … R-SQLC-TX-X3 из sqlc Style Guide → раздел 4. Транзакции.

Важно знать

• Граница транзакции — на Handler. pool.Begin(ctx) вызывается в Handler, не в репозитории.
• Репозиторий предоставляет WithTx(tx pgx.Tx) *Postgres<X>Repository — метод, возвращающий копию с q.WithTx(tx) внутри.
• defer tx.Rollback(ctx) — обязательная страховка: при успешном tx.Commit pgx сделает откат холостым, при панике или ошибке — откатит транзакцию.
• tx.Commit(ctx) вызывается явно после всей бизнес-логики; ошибка коммита обрабатывается и оборачивается через %w.
• Read-only handler — без транзакции или с pgx.TxOptions{AccessMode: pgx.ReadOnly}.
• Передача pgx.Tx через context.Value — запрещена. WithTx — единственный способ.
• Begin/Commit/Rollback в репозитории — запрещены. Репозиторий не знает про границы бизнес-операции.

Транзакция — это граница атомарного куска работы с БД. В UCP бизнес-операция = use case (PlaceOrder, CancelOrder, ApplyPayment). Use case реализуется одним Handler'ом. Поэтому транзакция = Handler. Это даёт чёткое правило без «а где граница?» — и снимает целый класс багов, когда два метода репозитория работают в разных транзакциях.

В Go нет аннотаций вроде @Transactional — транзакция открывается явно через pgxpool.Pool.Begin, закрывается через Commit/Rollback. Это нагляднее, чем AOP-прокси: в коде Handler'а видно ровно то, что происходит.

Открытие транзакции и defer tx.Rollback

R-SQLC-TX-1 и R-SQLC-TX-3: Handler открывает транзакцию, регистрирует defer tx.Rollback, выполняет работу, коммитит.

// core/order/handler/place_order_handler.go
package handler

type PlaceOrderHandler struct {
    pool *pgxpool.Pool
    repo port.OrderRepository
    products port.ProductRepository
}

func (h *PlaceOrderHandler) Handle(ctx context.Context, cmd PlaceOrderCommand) (*order.Order, error) {
    tx, err := h.pool.Begin(ctx)
    if err != nil {
        return nil, fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(ctx)

    orderRepo := h.repo.WithTx(tx)
    productRepo := h.products.WithTx(tx)

    product, err := productRepo.FindByID(ctx, cmd.ProductID)
    if err != nil {
        return nil, fmt.Errorf("find product: %w", err)
    }

    ord := order.New(cmd.CustomerID, product, cmd.Quantity)
    if err := orderRepo.Save(ctx, ord); err != nil {
        return nil, fmt.Errorf("save order: %w", err)
    }

    if err := tx.Commit(ctx); err != nil {
        return nil, fmt.Errorf("commit tx: %w", err)
    }
    return ord, nil
}

Что здесь важно:

• defer tx.Rollback(ctx) регистрируется сразу после pool.Begin — до любой ошибки. Если Handler вернёт ошибку или случится паника, транзакция откатится автоматически. После успешного tx.Commit pgx игнорирует Rollback (транзакция уже закрыта).
• tx.Commit(ctx) вызывается явно в конце. Ошибка коммита — не технический мусор: сеть могла оборваться в момент фиксации, данные могли не сохраниться. Оборачиваем и возвращаем вверх.
• Каждый репозиторий получает свой WithTx(tx) — все операции в одной транзакции.

WithTx на репозитории

R-SQLC-TX-2: репозиторий предоставляет метод WithTx(tx pgx.Tx) *Postgres<X>Repository, который возвращает копию с q.WithTx(tx) внутри. Это сгенерированный sqlc-метод — *db.Queries.WithTx оборачивает запросы в переданную транзакцию.

// adapters/out/persistence/postgres_order_repository.go

func (r *PostgresOrderRepository) WithTx(tx pgx.Tx) *PostgresOrderRepository {
    return &PostgresOrderRepository{
        q:    r.q.WithTx(tx),
        pool: r.pool,
    }
}

r.q.WithTx(tx) — метод из сгенерированного кода sqlc. Он создаёт новый *db.Queries, в котором все SQL-запросы выполняются через tx, а не через pgxpool.Pool. pool остаётся прежним — он нужен для CopyFrom при bulk-операциях.

Метод WithTx в интерфейсе порта — опционально. Если интерфейс в core/ не знает про pgx.Tx, Handler может работать с конкретным типом для открытия транзакции:

// core/order/port/order_repository.go
type OrderRepository interface {
    Save(ctx context.Context, o *order.Order) error
    FindByID(ctx context.Context, id uuid.UUID) (*order.Order, error)
}

// adapters/out/persistence/postgres_order_repository.go — WithTx только на конкретном типе
func (r *PostgresOrderRepository) WithTx(tx pgx.Tx) *PostgresOrderRepository { ... }

В этом случае Handler знает о *PostgresOrderRepository для открытия транзакции, но передаёт port.OrderRepository в бизнес-логику. Это осознанный выбор: порт остаётся чистым.

Несколько репозиториев в одной транзакции

Типовой сценарий — бизнес-операция затрагивает несколько агрегатов. Пример: оформление заказа списывает остатки Product и создаёт Order атомарно.

// core/order/handler/place_order_handler.go

func (h *PlaceOrderHandler) Handle(ctx context.Context, cmd PlaceOrderCommand) (*order.Order, error) {
    tx, err := h.pool.Begin(ctx)
    if err != nil {
        return nil, fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(ctx)

    orderRepo := h.orderRepo.WithTx(tx)
    productRepo := h.productRepo.WithTx(tx)

    product, err := productRepo.FindByID(ctx, cmd.ProductID)
    if err != nil {
        return nil, fmt.Errorf("find product %s: %w", cmd.ProductID, err)
    }

    product.Reserve(cmd.Quantity)
    if err := productRepo.Update(ctx, product); err != nil {
        return nil, fmt.Errorf("update product stock: %w", err)
    }

    ord := order.New(cmd.CustomerID, product, cmd.Quantity)
    if err := orderRepo.Save(ctx, ord); err != nil {
        return nil, fmt.Errorf("save order: %w", err)
    }

    if err := tx.Commit(ctx); err != nil {
        return nil, fmt.Errorf("commit place order: %w", err)
    }
    return ord, nil
}

Если productRepo.Update упадёт с ошибкой unique violation — defer tx.Rollback откатит и Save, и Update. Без транзакции можно было бы создать Order при недостаточных остатках или списать остатки без создания заказа.

Read-only транзакция

R-SQLC-TX-4: read-only Handler — без транзакции или с pgx.TxOptions{AccessMode: pgx.ReadOnly}. Read-only аннотирует намерение: запись невозможна на уровне PostgreSQL, что страхует от случайного изменения данных внутри Handler'а.

// core/order/handler/get_orders_query_handler.go

func (h *GetOrdersQueryHandler) Handle(ctx context.Context, q GetOrdersQuery) ([]*order.Order, error) {
    tx, err := h.pool.BeginTx(ctx, pgx.TxOptions{
        AccessMode: pgx.ReadOnly,
    })
    if err != nil {
        return nil, fmt.Errorf("begin read-only tx: %w", err)
    }
    defer tx.Rollback(ctx)

    orders, err := h.repo.WithTx(tx).ListByCustomer(ctx, q.CustomerID, q.Limit, q.Offset)
    if err != nil {
        return nil, fmt.Errorf("list orders: %w", err)
    }

    if err := tx.Commit(ctx); err != nil {
        return nil, fmt.Errorf("commit read-only tx: %w", err)
    }
    return orders, nil
}

Для простых read-only запросов без требований к консистентному снимку транзакция может не открываться — sqlc-метод вызывается напрямую через h.repo (без WithTx). pgx.ReadOnly полезен, когда Handler читает несколько таблиц и важна единая точка видимости данных.

Пример: загрузка корзины покупателя в Сбере — агрегат Cart + Product-ы для обогащения данных. Без транзакции два запроса могут увидеть разное состояние Product, если между ними произошло обновление.

Пример с outbox в одной транзакции

Outbox-событие пишется в outbox в той же транзакции, что и основные данные. Это гарантирует, что событие существует тогда и только тогда, когда существуют данные.

// core/order/handler/place_order_handler.go

func (h *PlaceOrderHandler) Handle(ctx context.Context, cmd PlaceOrderCommand) (*order.Order, error) {
    tx, err := h.pool.Begin(ctx)
    if err != nil {
        return nil, fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(ctx)

    orderRepo := h.orderRepo.WithTx(tx)
    outboxRepo := h.outboxRepo.WithTx(tx)

    ord := order.New(cmd.CustomerID, cmd.ProductID, cmd.Quantity)
    if err := orderRepo.Save(ctx, ord); err != nil {
        return nil, fmt.Errorf("save order: %w", err)
    }

    event := outbox.New("order.placed", ord.ID, ord)
    if err := outboxRepo.Save(ctx, event); err != nil {
        return nil, fmt.Errorf("save outbox event: %w", err)
    }

    if err := tx.Commit(ctx); err != nil {
        return nil, fmt.Errorf("commit place order: %w", err)
    }
    return ord, nil
}

Если tx.Commit упал — ни Order, ни outbox-событие не сохранились. Relay-горутина не опубликует фантомное событие в Kafka.

Что запрещено

Куда дальше

• Repository pattern в sqlc — доменный порт, PostgresOrderRepository, WithTx как часть реализации адаптера.
• Ошибки и pgx — как pgx.ErrNoRows и pgconn.PgError превращаются в доменные ошибки; почему defer tx.Rollback не скрывает ошибку репозитория.
• PostgreSQL: ACID и уровни изоляции — когда поднимать isolation level выше READ COMMITTED, retry на SQLSTATE 40001, write skew.