River Enqueues Jobs in Your Postgres Transaction. Keep It Out of Your Go Domain

2026-11-05来源 dev.to GoEnglish
River Enqueues Jobs in Your Postgres Transaction. Keep It Out of Your Go Domain

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.

A hexagonal core with plug-shaped ports on its outer ring, and a stream outside carrying small crates into one port
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.

A single sealed envelope holding two cards, an order receipt and a job ticket, closed with one wax seal
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{<...>

原文：https://www.a8q9.cn/en/article/river-enqueues-jobs-in-your-postgres-transaction-keep-it-out-of-your-go-domain-7293d8.html

相关文章