- Book: Hexagonal Architecture in Go
- Also by me: Pocket Guides for Developers — the 8-book series
- My project: Hermes IDE | GitHub — an IDE for developers who ship with Claude Code and other AI coding tools
- Me: xgabriel.com | GitHub
A customer places an order. Your handler inserts the row, commits, then pushes a "send confirmation" job to Redis. The Redis call times out. The order exists, the email never goes out, and nothing retries it because nothing knows the job was supposed to exist. Swap the order of operations and you get the opposite bug: the job lands first, a worker grabs it, and the order it points to is still uncommitted, or rolled back for good.
River's own docs tell this story with a user signup: the worker fetched the job from Redis before the Postgres transaction committed, and "the worker could not yet see that row when it queried for it." Their fix is the reason to pick River at all. The queue lives in the same Postgres database as your data, so the job can commit in the same transaction as the order.
That fix comes with a cost. The obvious way to use it puts pgx and River inside your use case. The version below keeps both out.
InsertTx is the feature, and the trap
River is a Go job queue backed by Postgres. Its Client is generic over the transaction type (type Client[TTx any] struct), and with the pgx v5 driver that type is pgx.Tx. The method that matters is InsertTx. Its doc comment in client.go spells out the guarantee: "the job will not be worked until the transaction has committed, and if the transaction rolls back, so too is the inserted job."
Here is how that tends to end up in a codebase:
func (uc PlaceOrder) Execute(ctx context.Context, o Order) error {
tx, err := uc.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
// ... INSERT INTO orders using tx ...
_, err = uc.river.InsertTx(ctx, tx, ConfirmArgs{ID: o.ID}, nil)
if err != nil {
return err
}
return tx.Commit(ctx)
}
It works, and it's atomic. It also means your application layer imports github.com/jackc/pgx/v5 and github.com/riverqueue/river. Testing PlaceOrder now needs a Postgres instance. Swapping the queue later means editing every use case that enqueues. The business rule ("placing an order schedules a confirmation") is tangled with the plumbing that makes it atomic.
The goal: keep the atomicity, move the plumbing to the edge.
The use case talks to ports. River and pgx live outside the ring.
The ports: a scheduler that knows nothing about Postgres
Start with the domain. It has no imports beyond the standard library:
package order
import "errors"
var ErrInvalid = errors.New("order: invalid")
type Order struct {
ID string
Email string
Total int64 // cents
}
func New(id, email string, total int64) (Order, error) {
if id == "" || email == "" || total <= 0 {
return Order{}, ErrInvalid
}
return Order{ID: id, Email: email, Total: total}, nil
}
The application layer declares what it needs. Four ports: a unit of work, a repository, a job scheduler, and a mailer.
package app
import (
"context"
"example.com/shop/internal/order"
)
type UnitOfWork interface {
Do(ctx context.Context, fn func(context.Context) error) error
}
type Orders interface {
Save(ctx context.Context, o order.Order) error
ByID(ctx context.Context, id string) (order.Order, error)
}
type JobScheduler interface {
ScheduleConfirmation(ctx context.Context, id string) error
}
type Mailer interface {
SendConfirmation(ctx context.Context, o order.Order) error
}
JobScheduler says what should happen later, in domain words. No job kinds, no InsertOpts, no queue names. The use cases read like the business rule:
package app
import (
"context"
"example.com/shop/internal/order"
)
type PlaceOrder struct {
UoW UnitOfWork
Orders Orders
Jobs JobScheduler
}
func (uc PlaceOrder) Execute(
ctx context.Context, id, email string, total int64,
) error {
o, err := order.New(id, email, total)
if err != nil {
return err
}
return uc.UoW.Do(ctx, func(ctx context.Context) error {
if err := uc.Orders.Save(ctx, o); err != nil {
return err
}
return uc.Jobs.ScheduleConfirmation(ctx, o.ID)
})
}
type SendConfirmation struct {
Orders Orders
Mailer Mailer
}
func (uc SendConfirmation) Execute(
ctx context.Context, orderID string,
) error {
o, err := uc.Orders.ByID(ctx, orderID)
if err != nil {
return err
}
return uc.Mailer.SendConfirmation(ctx, o)
}
PlaceOrder states that saving and scheduling belong to one unit of work. It doesn't know what a unit of work is made of.
The adapter: where the transaction has to bend
Here's the one place the abstraction can't stay clean. InsertTx needs the exact pgx.Tx the repository wrote with. The port doesn't mention a transaction, so something has to carry it from the unit of work to the River adapter. This version carries it in context.Context, behind an unexported key. Only code that imports the postgres adapter can get it back out, and the app layer never imports it:
package postgres
import (
"context"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
type txKey struct{}
// UnitOfWork runs fn inside one pgx transaction.
type UnitOfWork struct{ Pool *pgxpool.Pool }
func (u UnitOfWork) Do(
ctx context.Context, fn func(context.Context) error,
) error {
if _, ok := TxFrom(ctx); ok {
return fn(ctx) // nested call joins the outer tx
}
return pgx.BeginFunc(ctx, u.Pool, func(tx pgx.Tx) error {
return fn(context.WithValue(ctx, txKey{}, tx))
})
}
// TxFrom is for adapters. The app layer never calls it.
func TxFrom(ctx context.Context) (pgx.Tx, bool) {
tx, ok := ctx.Value(txKey{}).(pgx.Tx)
return tx, ok
}
pgx.BeginFunc commits if fn returns nil and rolls back if it returns an error, so the use case never sees Commit or Rollback. The repository picks the transaction when one is present and falls back to the pool otherwise:
package postgres
import (
"context"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool"
"example.com/shop/internal/order"
)
type querier interface {
Exec(context.Context, string, ...any) (
pgconn.CommandTag, error)
QueryRow(context.Context, string, ...any) pgx.Row
}
type Orders struct{ Pool *pgxpool.Pool }
func (r Orders) q(ctx context.Context) querier {
if tx, ok := TxFrom(ctx); ok {
return tx
}
return r.Pool
}
func (r Orders) Save(ctx context.Context, o order.Order) error {
_, err := r.q(ctx).Exec(ctx,
`INSERT INTO orders (id, email, total_cents)
VALUES ($1, $2, $3)`,
o.ID, o.Email, o.Total)
return err
}
func (r Orders) ByID(
ctx context.Context, id string,
) (order.Order, error) {
var o order.Order
err := r.q(ctx).QueryRow(ctx,
`SELECT id, email, total_cents
FROM orders WHERE id = $1`,
id).Scan(&o.ID, &o.Email, &o.Total)
return o, err
}
And the River adapter implements JobScheduler by pulling the same transaction:
package riverq
import (
"context"
"errors"
"github.com/jackc/pgx/v5"
"github.com/riverqueue/river"
"example.com/shop/internal/adapter/postgres"
)
type ConfirmationArgs struct {
OrderID string `json:"order_id"`
}
func (ConfirmationArgs) Kind() string {
return "order_confirmation"
}
var ErrNoTx = errors.New("riverq: no transaction in context")
type Scheduler struct {
Client *river.Client[pgx.Tx]
}
func (s Scheduler) ScheduleConfirmation(
ctx context.Context, orderID string,
) error {
tx, ok := postgres.TxFrom(ctx)
if !ok {
return ErrNoTx
}
_, err := s.Client.InsertTx(ctx, tx,
ConfirmationArgs{OrderID: orderID}, nil)
return err
}
Two decisions in there are worth defending.
The scheduler fails when there's no transaction instead of falling back to client.Insert. A silent fallback would bring back the exact gap you adopted River to close, and you'd only find out from a missing email. If a caller outside a unit of work ever needs to enqueue, give it a separate method that says so in its name.
The riverq package imports postgres. That's adapter-to-adapter coupling, and it's fine: both sit in the outer ring, and both exist because you chose Postgres. The application layer still compiles without either of them.
Carrying a transaction in a context hides a dependency, and some Go developers reject it for that reason. The other shape that works is a unit of work that hands fn a bundle of tx-bound ports (fn func(Ports) error), where Ports holds an Orders and a JobScheduler already bound to the transaction. It's more explicit and costs a constructor per bundle. Either way the bend stays in the adapters.
One transaction, two rows: the order and the job commit or roll back together.
Workers are driving adapters
A River worker is another way into your application, like an HTTP handler. So treat it like one: unpack the input and call a use case.
package riverq
import (
"context"
"github.com/riverqueue/river"
"example.com/shop/internal/app"
)
type ConfirmationWorker struct {
river.WorkerDefaults[ConfirmationArgs]
UC app.SendConfirmation
}
func (w *ConfirmationWorker) Work(
ctx context.Context, job *river.Job[ConfirmationArgs],
) error {
return w.UC.Execute(ctx, job.Args.OrderID)
}
The HTTP side calls the same SendConfirmation use case for a "resend confirmation" button, using the method patterns and PathValue that Go 1.22 added to net/http:
package httpapi
import (
"encoding/json"
"net/http"
"example.com/shop/internal/app"
)
type Handler struct {
Place app.PlaceOrder
Resend app.SendConfirmation
}
func (h Handler) Routes() *http.ServeMux {
mux := http.NewServeMux()
mux.HandleFunc("POST /orders", h.place)
mux.HandleFunc("POST /orders/{id}/confirmation", h.resend)
return mux
}
type placeReq struct {
ID string `json:"id"`
Email string `json:"email"`
TotalCents int64 `json:"total_cents"`
}
func (h Handler) place(w http.ResponseWriter, r *http.Request) {
var req placeReq
err := json.NewDecoder(r.Body).Decode(&req)
if err != nil {
http.Error(w, "bad json", http.StatusBadRequest)
return
}
err = h.Place.Execute(
r.Context(), req.ID, req.Email, req.TotalCents)
if err != nil {
http.Error(w, "could not place order", 500)
return
}
w.WriteHeader(http.StatusCreated)
}
func (h Handler) resend(w http.ResponseWriter, r *http.Request) {
err := h.Resend.Execute(r.Context(), r.PathValue("id"))
if err != nil {
http.Error(w, "could not resend", 500)
return
}
w.WriteHeader(http.StatusAccepted)
}
When the worker returns an error, River retries it. Per the retry docs, the default is 25 total attempts with a backoff of "attempts ^ 4 + rand(±10%)". The use case doesn't know any of that, which is correct: retry policy is a property of how the work was triggered. Because a job can run more than once, make the mailer idempotent, for example with a provider idempotency key or a sent_at check. JobCompleteTx won't help with email, since an email can't roll back. It's for workers whose database writes must commit with the job's completion.
Wiring happens in main, the only file that sees every package. The order matters only because PlaceOrder needs the client:
package main
import (
"context"
"log"
"net/http"
"os"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/riverqueue/river"
"github.com/riverqueue/river/riverdriver/riverpgxv5"
"example.com/shop/internal/adapter/httpapi"
"example.com/shop/internal/adapter/postgres"
"example.com/shop/internal/adapter/riverq"
"example.com/shop/internal/app"
"example.com/shop/internal/order"
)
type logMailer struct{}
func (logMailer) SendConfirmation(
_ context.Context, o order.Order,
) error {
log.Printf("confirmation for %s", o.ID)
return nil
}
logMailer stands in for a real mailer. The rest of main builds the adapters and hands them to the use cases:
func main() {
ctx := context.Background()
pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
defer pool.Close()
orders := postgres.Orders{Pool: pool}
send := app.SendConfirmation{
Orders: orders, Mailer: logMailer{},
}
workers := river.NewWorkers()
river.AddWorker(workers,
&riverq.ConfirmationWorker{UC: send})
client, err := river.NewClient(riverpgxv5.New(pool),
&river.Config{
Queues: map[string]river.QueueConfig{
river.QueueDefault: {MaxWorkers: 10},
},
Workers: workers,
})
if err != nil {
log.Fatal(err)
}
if err := client.Start(ctx); err != nil {
log.Fatal(err)
}
h := httpapi.Handler{
Place: app.PlaceOrder{
UoW: postgres.UnitOfWork{Pool: pool},
Orders: orders,
Jobs: riverq.Scheduler{Client: client},
},
Resend: send,
}
log.Fatal(http.ListenAndServe(":8080", h.Routes()))
}
Before it runs you need River's tables (river migrate-up --database-url "$DATABASE_URL", per the getting started guide) and your own orders table with id text PRIMARY KEY, email text NOT NULL and total_cents bigint NOT NULL. For production, add signal handling and call client.Stop, which River's source documents as a graceful shutdown that "waits for any fetched or in-progress jobs to complete before exiting."
Testing: a fake for the rule, one real test for the atomicity
The payoff of the port shows up in the use-case tests. No database, no River:
package app_test
import (
"context"
"errors"
"testing"
"example.com/shop/internal/app"
"example.com/shop/internal/order"
)
type fakeUoW struct{}
func (fakeUoW) Do(
ctx context.Context, fn func(context.Context) error,
) error {
return fn(ctx)
}
type memOrders struct {
saved map[string]order.Order
saveErr error
}
func (m *memOrders) Save(_ context.Context, o order.Order) error {
if m.saveErr != nil {
return m.saveErr
}
if m.saved == nil {
m.saved = map[string]order.Order{}
}
m.saved[o.ID] = o
return nil
}
func (m *memOrders) ByID(
_ context.Context, id string,
) (order.Order, error) {
return m.saved[id], nil
}
type memJobs struct{ scheduled []string }
func (m *memJobs) ScheduleConfirmation(
_ context.Context, id string,
) error {
m.scheduled = append(m.scheduled, id)
return nil
}
Three fakes, none of them aware of transactions. The tests sit in the same file:
func TestPlaceOrderSchedulesConfirmation(t *testing.T) {
jobs := &memJobs{}
uc := app.PlaceOrder{
UoW: fakeUoW{}, Orders: &memOrders{}, Jobs: jobs,
}
err := uc.Execute(context.Background(),
"ord_1", "alice@example.com", 4200)
if err != nil {
t.Fatal(err)
}
if len(jobs.scheduled) != 1 || jobs.scheduled[0] != "ord_1" {
t.Fatalf("scheduled = %v", jobs.scheduled)
}
}
func TestPlaceOrderSkipsJobWhenSaveFails(t *testing.T) {
boom := errors.New("boom")
jobs := &memJobs{}
uc := app.PlaceOrder{
UoW: fakeUoW{},
Orders: &memOrders{saveErr: boom},
Jobs: jobs,
}
err := uc.Execute(context.Background(),
"ord_2", "alice@example.com", 4200)
if !errors.Is(err, boom) || len(jobs.scheduled) != 0 {
t.Fatalf("err = %v, scheduled = %v", err, jobs.scheduled)
}
}
These tests prove the rule: a placed order schedules a confirmation, and a failed save doesn't. They can't prove the job and the order share a transaction, because the fake unit of work has no transaction. That claim belongs to the adapter, so test it there against a real database with rivertest, whose RequireInsertedTx fails the test when no job of that kind was inserted on the given transaction:
package riverq_test
import (
"context"
"errors"
"testing"
"github.com/riverqueue/river"
"github.com/riverqueue/river/riverdriver/riverpgxv5"
"github.com/riverqueue/river/rivertest"
"example.com/shop/internal/adapter/postgres"
"example.com/shop/internal/adapter/riverq"
)
func TestSchedulerUsesUoWTransaction(t *testing.T) {
ctx := context.Background()
pool := testPool(t) // your helper: a migrated test DB
workers := river.NewWorkers()
river.AddWorker(workers, &riverq.ConfirmationWorker{})
client, err := river.NewClient(riverpgxv5.New(pool),
&river.Config{Workers: workers})
if err != nil {
t.Fatal(err)
}
uow := postgres.UnitOfWork{Pool: pool}
sched := riverq.Scheduler{Client: client}
errUndo := errors.New("roll back on purpose")
err = uow.Do(ctx, func(ctx context.Context) error {
err := sched.ScheduleConfirmation(ctx, "ord_9")
if err != nil {
return err
}
tx, _ := postgres.TxFrom(ctx)
rivertest.RequireInsertedTx[*riverpgxv5.Driver](
ctx, t, tx, &riverq.ConfirmationArgs{}, nil)
return errUndo
})
if !errors.Is(err, errUndo) {
t.Fatalf("err = %v", err)
}
}
The client here never calls Start. River's Config docs say Queues "may be omitted for a program that's only queueing jobs rather than working them", and registering the worker lets the client check that the job kind has a worker. Returning errUndo rolls the transaction back, so the test leaves no rows behind.
River appears in three non-test files: the scheduler, the worker, and main (plus the adapter's integration test). app imports order and the standard library. If you move to another Postgres-backed queue, you rewrite riverq and the wiring. If you move to a queue outside Postgres, the port stays the same, but you lose InsertTx and need an outbox table to get the atomicity back. app doesn't change either way.
If this was useful
The transaction is where hexagonal Go codebases tend to leak, and this post is one instance of that. Hexagonal Architecture in Go works through that problem at book length: where ports end, how adapters share infrastructure without pulling it inward, how persistence and units of work fit the layout, and how to test each ring on its own. If River (or any Postgres-backed queue) is about to land in your codebase, it's the book I'd want you to have read first.
Top comments (0)