jackc

River Enqueues Jobs in Your Postgres Transaction. Keep It Out of Your Go Domain - 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. !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{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. 

Your Go Fake Repository Drifts From Postgres. Run One Contract Suite Against Both - 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 --- Your checkout tests are green. One of them replays a request the way a flaky mobile client would: same order reference, sent twice. The in-memory repository stores both orders without a word, so the use case's "reference already used" branch never runs and the test passes. In production, Postgres rejects the second insert with SQLSTATE 23505, unique_violation, nothing maps it to a domain error, and the client gets a 500. Same suite, another test. The order-history endpoint should return the newest order first. The fake appends to a slice, the test happens to insert in the order it asserts, and it passes. The Postgres adapter forgot its ORDER BY, and the Postgres docs are blunt about what that means: without sorting, "the rows will be returned in an unspecified order." Neither bug is in the fake's code. The fake never agreed to the same rules as the database. Martin Fowler's definition of a fake already warns you: fakes "have working implementations, but usually take some shortcut." The fix he describes for doubles is a contract test, which checks that calls against your double "return the same results as a call to the external service would." In a hexagonal Go service you can do that with one exported function and a single call per adapter. The port, with its rules written down The contract starts as prose on the interface. If a rule isn't written here, no adapter can be held to it. package domain import ( "context" "errors" "time" ) var ( ErrNotFound = errors.New("order not found") ErrDuplicateRef = errors.New("order ref already used") ErrVersionConflict = errors.New("order changed since read") ) type Order struct { ID string Ref string CustomerID string Status string CreatedAt time.Time Version int64 } // OrderRepository is the port. Every adapter must: // - Create: store o with Version 1. ErrDuplicateRef if // another order already has o.Ref. // - Get: ErrNotFound if no order has that ID. // - ListByCustomer: newest CreatedAt first, ties broken // by ID in byte order. // - Update: write o.Status if the stored Version equals // o.Version, then bump it. ErrVersionConflict if not, // ErrNotFound if the order does not exist. type OrderRepository interface { Create(ctx context.Context, o Order) error Get(ctx context.Context, id string) (Order, error) ListByCustomer( ctx context.Context, customerID string, ) ([]Order, error) Update(ctx context.Context, o Order) error } The errors live in the domain because they are business facts. "That reference is taken" means something to a checkout use case. SQLSTATE 23505 does not, and it should never cross the port. One exported function, every adapter The suite goes in its own package, domain/ordertest, in a normal .go file rather than a _test.go file. go build ignores files that end in _test.go, so nothing declared in one can be imported by another package, and both adapter packages need to call this. Importing testing from a non-test package has a precedent in the Go project itself: nettest.TestConn "tests that a net.Conn implementation properly satisfies the interface," and it takes a constructor the same way. !One brass key ring hanging between two doors, a cardboard door and a steel door, both with the same lock One set of checks, two implementations. Each door has to open the same way. The constructor gives each subtest a fresh, empty repository, so subtests never see each other's rows. package ordertest import ( "errors" "slices" "testing" "time" "example.com/shop/domain" ) var t0 = time.Date(2026, 3, 1, 9, 0, 0, 0, time.UTC) func RunOrderRepoContract( t *testing.T, newRepo func(t *testing.T) domain.OrderRepository, ) { t.Run("missing id", func(t *testing.T) { repo := newRepo(t) _, err := repo.Get(t.Context(), "nope") if !errors.Is(err, domain.ErrNotFound) { t.Fatalf("got %v, want ErrNotFound", err) } }) t.Run("round trip", func(t *testing.T) { repo := newRepo(t) mustCreate(t, repo, ord("o-1", "R-1", "c-1", t0)) got, err := repo.Get(t.Context(), "o-1") if err != nil { t.Fatal(err) } if got.Ref != "R-1" || got.Version != 1 || !got.CreatedAt.Equal(t0) { t.Fatalf("got %+v", got) } }) t.Run("duplicate ref", func(t *testing.T) { repo := newRepo(t) ctx := t.Context() mustCreate(t, repo, ord("o-1", "R-1", "c-1", t0)) err := repo.Create(ctx, ord("o-2", "R-1", "c-2", t0)) if !errors.Is(err, domain.ErrDuplicateRef) { t.Fatalf("got %v, want ErrDuplicateRef", err) } _, err = repo.Get(ctx, "o-2") if !errors.Is(err, domain.ErrNotFound) { t.Fatalf("o-2 was stored anyway: %v", err) } }) t.Context() needs Go 1.24 or later. Timestamps are compared with Equal, because the time docs note that == "compares not just the time instant but also the Location," and a database adapter is free to hand back the same instant in a different location. The function continues with the ordering rule: t.Run("list order", func(t *testing.T) { repo := newRepo(t) later := t0.Add(time.Hour) mustCreate(t, repo, ord("o-a", "R-1", "c-1", t0)) mustCreate(t, repo, ord("o-c", "R-2", "c-1", later)) mustCreate(t, repo, ord("o-B", "R-3", "c-1", t0)) mustCreate(t, repo, ord("o-x", "R-4", "c-2", later)) got, err := repo.ListByCustomer(t.Context(), "c-1") if err != nil { t.Fatal(err) } var ids []string for _, o := range got { ids = append(ids, o.ID) } want := []string{"o-c", "o-B", "o-a"} if !slices.Equal(ids, want) { t.Fatalf("got %v, want %v", ids, want) } }) The list order subtest is built to fail a lazy fake. Insertion order is o-a, o-c, o-B, and neither that order nor its reverse matches the expected o-c, o-B, o-a. A fake that returns map iteration order fails too, at least some of the time, since the Go spec says map iteration order "is not specified and is not guaranteed to be the same from one iteration to the next." The uppercase B is there on purpose, and it comes back in the Postgres section. The last two subtests cover updates, followed by the helpers: t.Run("stale version", func(t *testing.T) { repo := newRepo(t) ctx := t.Context() mustCreate(t, repo, ord("o-1", "R-1", "c-1", t0)) a, err := repo.Get(ctx, "o-1") if err != nil { t.Fatal(err) } b := a a.Status = "paid" if err := repo.Update(ctx, a); err != nil { t.Fatal(err) } b.Status = "cancelled" err = repo.Update(ctx, b) if !errors.Is(err, domain.ErrVersionConflict) { t.Fatalf("got %v, want ErrVersionConflict", err) } got, err := repo.Get(ctx, "o-1") if err != nil { t.Fatal(err) } if got.Status != "paid" || got.Version != 2 { t.Fatalf("got %+v", got) } }) t.Run("update missing", func(t *testing.T) { repo := newRepo(t) o := ord("nope", "R-9", "c-1", t0) o.Version = 1 err := repo.Update(t.Context(), o) if !errors.Is(err, domain.ErrNotFound) { t.Fatalf("got %v, want ErrNotFound", err) } }) } func ord(id, ref, cust string, at time.Time) domain.Order { return domain.Order{ ID: id, Ref: ref, CustomerID: cust, Status: "pending", CreatedAt: at, } } func mustCreate( t *testing.T, r domain.OrderRepository, o domain.Order, ) { t.Helper() if err := r.Create(t.Context(), o); err != nil { t.Fatalf("create %s: %v", o.ID, err) } } The stale version subtest is the optimistic-locking rule from an earlier post on version conflicts in a Go repository adapter, folded into the shared suite instead of standing alone. The fake, held to it This map-backed fake passes the suite, and it is still small. package memory import ( "context" "fmt" "slices" "strings" "sync" "example.com/shop/domain" ) type OrderRepo struct { mu sync.Mutex byID map[string]domain.Order byRef map[string]string } func NewOrderRepo() *OrderRepo { return &OrderRepo{ byID: map[string]domain.Order{}, byRef: map[string]string{}, } } func (r *OrderRepo) Create( _ context.Context, o domain.Order, ) error { r.mu.Lock() defer r.mu.Unlock() if _, ok := r.byRef[o.Ref]; ok { return domain.ErrDuplicateRef } if _, ok := r.byID[o.ID]; ok { return fmt.Errorf("memory: id %s exists", o.ID) } o.Version = 1 r.byID[o.ID] = o r.byRef[o.Ref] = o.ID return nil } func (r *OrderRepo) Get( _ context.Context, id string, ) (domain.Order, error) { r.mu.Lock() defer r.mu.Unlock() o, ok := r.byID[id] if !ok { return domain.Order{}, domain.ErrNotFound } return o, nil } func (r *OrderRepo) ListByCustomer( _ context.Context, customerID string, ) ([]domain.Order, error) { r.mu.Lock() defer r.mu.Unlock() var out []domain.Order for _, o := range r.byID { if o.CustomerID == customerID { out = append(out, o) } } slices.SortFunc(out, func(a, b domain.Order) int { if c := b.CreatedAt.Compare(a.CreatedAt); c != 0 { return c } return strings.Compare(a.ID, b.ID) }) return out, nil } func (r *OrderRepo) Update( _ context.Context, o domain.Order, ) error { r.mu.Lock() defer r.mu.Unlock() cur, ok := r.byID[o.ID] if !ok { return domain.ErrNotFound } if cur.Version != o.Version { return domain.ErrVersionConflict } cur.Status = o.Status cur.Version++ r.byID[o.ID] = cur return nil } The byRef index is the fake's version of a unique constraint, and the sort is its version of ORDER BY. Without those two pieces, the duplicate and ordering bugs from the opening are exactly what you would have shipped. Wiring it to the suite is one call: package memory_test import ( "testing" "example.com/shop/adapter/memory" "example.com/shop/domain" "example.com/shop/domain/ordertest" ) func TestOrderRepoContract(t *testing.T) { ordertest.RunOrderRepoContract(t, func(*testing.T) domain.OrderRepository { return memory.NewOrderRepo() }) } The pgx adapter, mapping SQLSTATE 23505 The real adapter uses pgx v5. A failed insert carries a *pgconn.PgError with a Code (the SQLSTATE) and a ConstraintName. Checking both matters: a primary-key collision is also 23505, and it is not the same domain fact as a reused reference. package postgres import ( "context" "errors" "fmt" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" "example.com/shop/domain" ) const cols = `id, ref, customer_id, status, created_at, version` type OrderRepo struct{ pool *pgxpool.Pool } func NewOrderRepo(p *pgxpool.Pool) *OrderRepo { return &OrderRepo{pool: p} } func (r *OrderRepo) Create( ctx context.Context, o domain.Order, ) error { _, err := r.pool.Exec(ctx, ` INSERT INTO orders (`+cols+`) VALUES ($1, $2, $3, $4, $5, 1)`, o.ID, o.Ref, o.CustomerID, o.Status, o.CreatedAt) var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" && pgErr.ConstraintName == "orders_ref_key" { return domain.ErrDuplicateRef } if err != nil { return fmt.Errorf("create order %s: %w", o.ID, err) } return nil } func (r *OrderRepo) Get( ctx context.Context, id string, ) (domain.Order, error) { rows, err := r.pool.Query(ctx, `SELECT `+cols+` FROM orders WHERE id = $1`, id) if err != nil { return domain.Order{}, fmt.Errorf("get %s: %w", id, err) } o, err := pgx.CollectExactlyOneRow(rows, scanOrder) if errors.Is(err, pgx.ErrNoRows) { return domain.Order{}, domain.ErrNotFound } if err != nil { return domain.Order{}, fmt.Errorf("get %s: %w", id, err) } return o, nil } func (r *OrderRepo) ListByCustomer( ctx context.Context, customerID string, ) ([]domain.Order, error) { rows, err := r.pool.Query(ctx, ` SELECT `+cols+` FROM orders WHERE customer_id = $1 ORDER BY created_at DESC, id COLLATE "C"`, customerID) if err != nil { return nil, fmt.Errorf("list %s: %w", customerID, err) } return pgx.CollectRows(rows, scanOrder) } func (r *OrderRepo) Update( ctx context.Context, o domain.Order, ) error { tag, err := r.pool.Exec(ctx, ` UPDATE orders SET status = $3, version = version + 1 WHERE id = $1 AND version = $2`, o.ID, o.Version, o.Status) if err != nil { return fmt.Errorf("update %s: %w", o.ID, err) } if tag.RowsAffected() == 1 { return nil } if _, err := r.Get(ctx, o.ID); err != nil { return err // ErrNotFound, or a real failure } return domain.ErrVersionConflict } func scanOrder(row pgx.CollectableRow) (domain.Order, error) { var o domain.Order err := row.Scan(&o.ID, &o.Ref, &o.CustomerID, &o.Status, &o.CreatedAt, &o.Version) return o, err } CollectExactlyOneRow returns an error matching pgx.ErrNoRows when the query finds nothing, which is the one place the adapter translates to ErrNotFound. CollectRows closes the rows for you. Now the uppercase B. The fake breaks ties with strings.Compare, and the Go spec says strings "are compared lexically byte-wise." Postgres sorts text by the column's collation, and a linguistic collation such as en_US.utf8 does not have to put o-B before o-a the way byte order does. The C collation sorts "by byte values rather than natural language order," so COLLATE "C" on the tiebreak makes both adapters agree no matter what the database default is. Drop it, and the list order subtest is where you find out. Real Postgres, behind a build tag The integration test starts Postgres with the testcontainers-go Postgres module, runs the schema, and hands the suite a constructor that truncates the table. A //go:build line keeps it out of the default run. !A workbench with a small wooden model engine and a real steel engine, both wired to the same burnt-orange test gauge The model and the real engine go on the same gauge. Only one of them needs Docker. //go:build integration package postgres_test import ( "testing" "github.com/jackc/pgx/v5/pgxpool" "github.com/testcontainers/testcontainers-go" tcpg "github.com/testcontainers/testcontainers-go/modules/postgres" "example.com/shop/adapter/postgres" "example.com/shop/domain" "example.com/shop/domain/ordertest" ) const schema = ` CREATE TABLE orders ( id text PRIMARY KEY, ref text NOT NULL, customer_id text NOT NULL, status text NOT NULL, created_at timestamptz NOT NULL, version bigint NOT NULL, CONSTRAINT orders_ref_key UNIQUE (ref) )` func TestOrderRepoContract(t *testing.T) { ctx := t.Context() ctr, err := tcpg.Run(ctx, "postgres:16-alpine", tcpg.WithDatabase("shop"), tcpg.WithUsername("shop"), tcpg.WithPassword("shop"), tcpg.BasicWaitStrategies(), ) testcontainers.CleanupContainer(t, ctr) if err != nil { t.Fatal(err) } dsn, err := ctr.ConnectionString(ctx, "sslmode=disable") if err != nil { t.Fatal(err) } pool, err := pgxpool.New(ctx, dsn) if err != nil { t.Fatal(err) } t.Cleanup(pool.Close) if _, err := pool.Exec(ctx, schema); err != nil { t.Fatal(err) } ordertest.RunOrderRepoContract(t, func(t *testing.T) domain.OrderRepository { _, err := pool.Exec(t.Context(), "TRUNCATE orders") if err != nil { t.Fatal(err) } return postgres.NewOrderRepo(pool) }) } CleanupContainer goes before the error check because its docs say a nil container is a no-op, so a half-started container still gets removed. The module page also says it supplies no default wait strategy, which is why BasicWaitStrategies() is there. One container serves all six subtests, and TRUNCATE gives each one an empty table, which keeps the run to a single container start. The constraint is named explicitly, orders_ref_key, because the adapter matches on that name. In a real service, replace the inline schema with your migration runner, so the test checks the constraint names your migrations actually create. Two commands now cover both worlds: go test -race ./... # fake only, no Docker go test -race -tags integration ./... # fake and Postgres What the suite catches next The suite only knows the rules you gave it, and Postgres has more. Two worth adding as subtests before they bite: Timestamp precision. Go's time.Time holds nanoseconds, and Postgres timestamptz resolution is 1 microsecond. The suite uses whole-second timestamps, so it can't see this. A subtest that creates an order with time.Now() and compares CreatedAt after a round trip will pass on the fake and, whenever the clock reading has sub-microsecond digits, fail on Postgres. Decide in the port whether the repository truncates, then make the fake do the same. Empty results. The two adapters in this post already disagree here. The fake above returns a nil slice for a customer with no orders. The pgx adapter does not: CollectRows is return AppendRows([]T{}, rows, fn), so it starts from an empty, non-nil slice. Marshal both and the encoding/json docs say "a nil slice encodes as the null JSON value," so one adapter gives the client null and the other [], and the suite as written passes both. Add a subtest that pins the shape, or normalise it in the handler. When a third adapter shows up (a read-through cache wrapping the Postgres repo, a SQLite build for a CLI), its test is one RunOrderRepoContract call with a different constructor. If it disagrees with the other two, the suite names the rule. --- If this was useful A contract suite only works if the port says what the rules are and database errors stay in the adapter. Hexagonal Architecture in Go is the book I wrote about that layer: where repositories, adapters and their tests sit in a Go service, and how to test both sides of a port without a pile of mocks. It is part of the Thinking in Go series, with a companion book on the language itself. 

A cancelled query can destroy your database connection Two short deadlines took down Inngest's database path twice in one afternoon, and neither was misconfigured. They expired exactly on time. That is what made the failure so effective. On 10 October 2026, both of Inngest's connection poolers lost every connection to their primary database at 14:00 UTC. They lost them again at 15:30. Run scheduling and queue processing stalled for most accounts in both windows, 14:00–14:48 and 15:30–16:25, and the mitigation that followed caused intermittent delays until 04:17 the next morning. The report is worth reading because the mechanism generalises to any stack that puts a connection pooler with a wait queue between the application and Postgres. !Component schematic: a client row with an expired deadline, a PgBouncer waiting queue, a gate admitting one new connection at a time, and a rack of connection slots marked closed rather than returned to the pool The sequence, in four steps 1. A frequently used lookup carried a short deadline. 2. The deadline expired while its query was still queued inside PgBouncer, waiting for a database connection. The pgx driver cancelled the query and closed its client connection. 3. PgBouncer had nothing running to cancel, because the query had never been handed a server connection. It later ran the abandoned query on a database connection. 4. It then closed that connection entirely instead of returning it to the pool. Step 4 is the whole failure. The client's timeout cost the system a database connection, not a request, and the cost landed after the client had stopped watching. No single request looked expensive. Individually, every one of these is a wasted query. Collectively they are a pool that empties itself. Two amplifiers arrived at the same moment. Both collapses began at the top and bottom of the hour, when cron-scheduled work roughly triples the query rate, and a newly rolled out set of lookups was failing in large numbers, adding cancellations to a pool that was already busy. Inngest reproduced the behaviour in a test environment. Why restarting and resizing both failed Restarting the poolers did not reliably recover them. Raising the pool size did not help either. The report is explicit on why: opening connections was the bottleneck, not the pool limit. Each pooler opens new database connections one at a time. So the refill path is serial while the destruction path is parallel. As a steady-state sketch rather than a model of this incident, if usable connections are destroyed at d per second and a pooler needs t seconds on average to establish a replacement, the refill ceiling is 1/t and the pool drains whenever: d > 1/t The sketch assumes constant destruction and handshake times, and it ignores the point at which the pooler stops trying. Inngest published neither number, so treat this as the shape of the constraint rather than a measurement of it. To see the shape: at 100 ms per connection a single pooler refills at ten per second; at 50 ms, twenty. The second collapse ended by changing paths, not by growing the pool. The team moved all services onto poolers run alongside the database by the database provider, and processing recovered between 16:10 and 16:25. The fix that caused a second incident The new route reached the database through NAT gateways, and some connections through those gateways were silently dropped. Queries on a dropped connection hung for up to 16 minutes before failing, which stalled processing intermittently and most visibly hit cron functions. Customers reported delays at 03:01 on 11 October; the affected service moved back to the original poolers at 04:17, and gateway routing was improved on 12 October. Sixteen minutes of silent hang is worse than a fast failure for a scheduler. A connection that errors invites a retry. One that simply stops answering consumes a worker slot and a deadline, and you learn about it from a customer. A failover drill that measures successful requests will not catch this. One that measures how long a hung connection stays hung will. The contract that broke underneath it The part of the report with the longest reach is not about pools. Internal failures counted against functions' configured retries, and a step that failed permanently on an internal error left its run in Running instead of marking it failed. onFailure handlers did not fire. Functions waiting on that run through step.invoke were not notified. That is a broken guarantee, not an availability number. A durable execution engine sells the promise that every run reaches a terminal state and that failure handlers run. Under this bug the engine could hold a run in Running indefinitely, after spending the retry budget it was given — the customer's allowance consumed by the platform's own faults, and the transition that would have reported it never sent. The recovery instructions confirm the state of those runs: they will not continue on their own, and customers can cancel and replay them, with the caveat that a replay starts from the beginning, so completed steps run again. Anyone with non-idempotent side effects in a completed step had a decision the platform could not make for them. Four things to check in your own stack Alert on net connection drain, not pool utilisation. Graph opens and closes per pooler per second, and alert when the close rate exceeds the open rate for a sustained interval. A pool at 30% with a negative slope is in more danger than one at 70% that is stable, and utilisation alone shows neither. Bound the wait, not just the query. PgBouncer's query_wait_timeout disconnects a client whose query has not been assigned a server within the limit. ; pgbouncer.ini — the ordering matters more than the value query_wait_timeout = 5 ; must sit below your client's own deadline The point of setting it below the client deadline is to remove the queued entry before the client gives up. That is the reasoning, not proof that it preserves backend connections. Inngest describes the pooler running the abandoned query and then closing that connection; an independent load test from July 2025 puts the waste a step earlier, with the pooler finishing a login for a client that has already gone and discarding the fresh backend connection. The same engineer reported that lowering the timeout helped without being enough on its own. Test it against your driver's real cancellation path before adopting it. Know what your driver does on cancel. In pgx v5, the thread reports replacing the default cancellation behaviour with a context watcher handler that sets a connection deadline instead of closing immediately, and describes the effect in their load test as large. If your code path cancels queued queries, this is the boundary to inspect. Fence the lease before you force a terminal state. If you mark a run failed because its step lease expired without a heartbeat, late writes from a worker that is still running can land afterwards, and a replay produces a second copy of the side effects. Mark it failed only when the engine can reject writes from the old lease, or when the step is confirmed not to be in flight. What the report does not establish Inngest disclosed no connection counts, handshake timings, pool sizes or customer totals, so the arithmetic here is the report's timestamps and nothing more. The 44-minute, 48-minute and 55-minute windows are my subtraction on its UTC boundaries, not figures it published. The report also links pgx issue 1780 as the known interaction behind the collapse. I read that issue: it was opened on 26 October 2023 and was still open at its latest comment in February 2026. The upstream record offers a compatible description of the boundary and no confirmed root cause, so Inngest's account is a well-documented instance of an unresolved interaction rather than evidence that it is fixed. The structural change is the one to watch. Inngest is consolidating onto a single data centre acting as one cell, which removes the cross-data-centre failure class behind the 9 October incident. That trades one failure shape for another: fewer paths that can break, and more consequence when one does. Sources - Inngest: Incident report for October 9 and 10, 2026 - jackc/pgx issue 1780 Thanks for reading :)

Go's database/sql Has Been Ports and Adapters Since Go 1. Read Its Driver Port - 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 --- Drop one line from a Go service's import block and sql.Open answers with sql: unknown driver "postgres" (forgotten import?). The standard library guesses your bug for you. The missing line is import _ "github.com/lib/pq", and that underscore is the visible end of a ports-and-adapters design that has sat in Go's standard library since the language hit 1.0. The package was exp/sql until the weekly snapshot of 2012-01-20 renamed it to database/sql, and it shipped in Go 1 on 28 March 2012. Alistair Cockburn had written up the ports and adapters pattern in 2005, with the stated intent of letting an application be "developed and tested in isolation from its eventual run-time devices and databases". Fourteen years after Go 1, database/sql is still a clean blueprint of that idea, plus one design decision you should not copy. The core owns the port Start with who defines what. The interfaces live in database/sql/driver, and the package doc opens with: "Package driver defines interfaces to be implemented by database drivers as used by package sql." The consumer writes the contract. Postgres, MySQL and SQLite authors implement it. That is the rule the Go wiki's Code Review Comments gives for every interface: "Go interfaces generally belong in the package that uses values of the interface type, not the package that implements those values." database/sql follows it to the letter. Here is the required surface, abridged from the package docs: // database/sql/driver (abridged) type Driver interface { Open(name string) (Conn, error) } type Conn interface { Prepare(query string) (Stmt, error) Close() error Begin() (Tx, error) // deprecated } type Stmt interface { Close() error NumInput() int Exec(args []Value) (Result, error) // deprecated Query(args []Value) (Rows, error) // deprecated } type Rows interface { Columns() []string Close() error Next(dest []Value) error } That is small on purpose. The design goals document that sits in the package directory lists a clear split between the driver layer and the user-facing layer, type conversion handled mostly in the sql package so drivers deal with fewer types, and optional interfaces for special cases or faster paths. Your application code never touches driver.Conn. It talks to *sql.DB, *sql.Rows and *sql.Tx, which are concrete types the core controls. !A hexagon representing the core with three plug-shaped adapters docking into one socket on its edge The core defines one socket. Every database driver is a plug shaped to fit it. Adapters plug themselves in Each driver is an adapter, and it registers itself. This is the whole init in lib/pq's conn.go: func init() { sql.Register("postgres", &Driver{}) } go-sql-driver/mysql does the same with sql.Register(driverName, &MySQLDriver{}), and its package comment tells you to use it via import _ "github.com/go-sql-driver/mysql". The official Go database tutorial explains why the underscore is there: if you never call a function from the driver package directly, you need a blank import so the package still loads. import ( "database/sql" _ "github.com/lib/pq" ) func open(dsn string) (*sql.DB, error) { return sql.Open("postgres", dsn) } The blank import runs init, init writes into a map inside database/sql, and sql.Open looks the name up in that map. Your code depends only on the port. The adapter is a line in your import block. The SQLDrivers wiki page lists roughly 60 adapters built this way, from pgx to modernc.org, and marks the ones that pass a shared compatibility test suite. Optional interfaces: capabilities without a fat port Here is the part worth borrowing first. The required port stayed tiny, but the protocol grew a lot after 2012: context cancellation, named parameters, session resets, connection validation. The Go team did not add those methods to driver.Conn, which would have broken every driver. They added separate, optional interfaces and let the core detect them at runtime. Go 1.8 brought QueryerContext, ExecerContext, ConnBeginTx and Pinger. Go 1.9 brought NamedValueChecker. Go 1.10 brought Connector, DriverContext and SessionResetter. Go 1.15 brought Validator. An old driver that implements none of them still compiles and still works. The detection is a type assertion with a fallback. A simplified sketch of what queryDC in sql.go does: // simplified sketch of database/sql's queryDC queryerCtx, ok := dc.ci.(driver.QueryerContext) var queryer driver.Queryer if !ok { queryer, ok = dc.ci.(driver.Queryer) } if ok { rows, err := ctxDriverQuery(ctx, queryerCtx, queryer, query, args) if err != driver.ErrSkip { return rows, err // adapter handled it } } // fallback: Prepare, Query, Close the statement The driver package docs spell out the chain: if a Conn implements neither QueryerContext nor Queryer, DB.Query prepares the statement, runs it, and closes it. And driver.ErrSkip, whose message reads "driver: skip fast-path; continue as if unimplemented", lets an adapter say "I have this capability, but not for this particular call." Argument checking follows the same idea. According to convert.go, the core looks for NamedValueChecker on the statement first, then on the connection, and falls back to the statement's column converter or its own default checker when neither exists or when the checker returns ErrSkip. pgx's stdlib adapter implements CheckNamedValue, ResetSession and QueryContext. A smaller driver can skip all three and still get correct behaviour from the core. That gives you a port that can grow for a decade without a breaking change. The pool lives in the core The other half of the design is what adapters do not have to write. The sql.OpenDB doc comment says the returned DB "maintains its own pool of idle connections". The DB struct in that same file holds freeConn, connRequests, maxOpen, maxLifetime and maxIdleTime. Idle connections default to defaultMaxIdleConns = 2, and SetMaxOpenConns documents that "The default is 0 (unlimited)." No driver reimplements pooling, lifetimes or waiting for a free connection. The adapter contributes hooks the core calls at the right moment. If a Conn implements Validator, the driver docs say IsValid runs before the connection goes back to the pool. If it implements SessionResetter, ResetSession runs before reuse. And driver.ErrBadConn tells the core to retry on a fresh connection, with a warning not to return it if the server might already have done the work. Policy in the core, mechanism in the adapter. When you design your own ports, retries, timeouts, circuit breaking and pooling belong on the core side for the same reason: otherwise every adapter writes its own slightly different version. The half you should not copy: a global registry Now the part that breaks the hexagonal rules. In sql.go, the registry is a package-level map behind a mutex: var driversMu sync.RWMutex var drivers = make(map[string]driver.Driver) Adapters are wired by import side effects into global state, not by an explicit composition root, which Mark Seemann describes as "a (preferably) unique location in an application where modules are composed together." Three costs follow from that. First, the wiring fails at runtime, not compile time. That is the error from the top of this post: the message has to guess at the forgotten import because the compiler never sees the dependency. Second, names collide globally. Register panics with sql: Register called twice for driver on a duplicate name. pgx works around that in its init: it registers "pgx" only if sql.Drivers() does not already contain it, and always registers "pgx/v5". Third, the map is now hard to change. The comment above it in sql.go says it should be an internal detail, but that other packages reach into it via linkname, and asks that its type signature not change. !Left side shows tangled import arrows writing into a single global map; right side shows one main function passing adapters directly into a core Side-effect registration on the left, an explicit composition root on the right. Go 1.10 gave you the exit. sql.OpenDB takes a driver.Connector value instead of a name string, so main hands the adapter to the core directly: // pgx is github.com/jackc/pgx/v5 // stdlib is github.com/jackc/pgx/v5/stdlib (pgx's adapter) cfg, err := pgx.ParseConfig(os.Getenv("DATABASE_URL")) if err != nil { log.Fatal(err) } db := sql.OpenDB(stdlib.GetConnector(*cfg)) MySQL has the same shape through mysql.NewConnector, which the Go tutorial uses as its example. Importing the package still runs its init, so the registry entry still exists, but your code no longer depends on a magic string or on import order. Apply the good half to your own ports Take the three habits that aged well and leave the registry behind. Say a billing service sends notifications. The service package owns the port, a richer capability is optional, and there is always a fallback: package billing // Sender is the port. billing owns it. type Sender interface { Send(ctx context.Context, m Message) error } // BatchSender is optional. Detected, never required. type BatchSender interface { SendBatch(ctx context.Context, ms []Message) error } func (s *Service) notifyAll( ctx context.Context, ms []Message, ) error { if bs, ok := s.sender.(BatchSender); ok { return bs.SendBatch(ctx, ms) } for _, m := range ms { if err := s.sender.Send(ctx, m); err != nil { return err } } return nil } Retries and rate limits go in Service, not in each adapter, the same way sql.DB owns the pool. And the wiring happens in one visible place: func main() { db := mustOpenDB() repo := postgres.NewInvoiceRepo(db) sender := smtp.NewSender(loadSMTPConfig()) svc := billing.NewService(repo, sender) runHTTP(svc) } No init, no name strings, no blank imports. If you forget an adapter, the code does not compile. For tests, pass a fake Sender into NewService, which is exactly the isolation Cockburn's 2005 write-up was after. Next time you type import _ for a database driver, read it as the stdlib showing you the pattern, and the one place it cut a corner. Wire your own adapters in main. --- If this was useful database/sql gets the port and the capability detection right, and gets the wiring wrong. Hexagonal Architecture in Go works through both halves for your own services: where ports live in the package layout, how persistence adapters stay behind them, and how to build a composition root in main that the compiler can check. If the billing example above looks like the direction you want your codebase to go, that book is the long version of it. 

How to Implement Idempotency Keys in Go with PostgreSQL --- title: How to Implement Idempotency Keys in Go with PostgreSQL published: true date: 2026-10-12 12:12:37 UTC tags: DistributedSystems,Go,IdempotentAPIs,Payments canonical_url: shahidyousuf.dev cover_image: shahidyousuf.dev published_at: "2026-10-13T03:15:23Z" --- Payment requests travel over networks that drop things. A phone on a train loses signal for a second, or a mobile SDK stops waiting. The request can reach your server, your server can charge the card, and the answer can vanish on the way back. The client sees an error and does the sensible thing: it sends the request again. Retrying is the right reaction: the client can't tell a request that never arrived from one whose answer got lost. The trouble is on the server, which without an idempotency key can't tell a retry from a new payment. A payment is a POST, and a server that treats every POST as new work charges the card once per attempt. So the customer taps Pay once, sees one receipt, and finds two charges of 25.00 on their statement. Your logs show two successful payments a second apart, each correct on its own, and the fix is a refund, an apology and possibly a chargeback, a dispute raised through the customer's bank. The general fix is an idempotency key : a unique value the client generates once per payment and sends with every attempt. The server does the work the first time it sees a key, records the result, and answers every repeat with that same result. The key needs a home that every copy of your server can reach and that can settle a race between two requests carrying it. If your payments already live in PostgreSQL, a table with a primary key on the idempotency key does both. In this tutorial, you'll implement idempotency keys in Go with PostgreSQL for a small payments API, break each version on purpose, and load-test the result against a fake card provider that keeps its own ledger. It's a sequel to Distributed Locks in Redis with Go. Everything was tested with Go 1.27.2, pgx v5.11.0 and PostgreSQL 18.6. Prerequisites To follow along, you need: - Go 1.25 or later, which pgx v5.11.0, the PostgreSQL driver, requires. The code also uses crypto/rand.Text (Go 1.24) and method patterns such as "POST /payments" in http.HandleFunc, which arrived in Go 1.22. - Docker, to run PostgreSQL locally. - Working knowledge of net/http, context.Context and SQL. No payments experience is needed. - Ports 5432, 8080 and 8090 free, and three terminal windows open in the project directory. Step 1: Set Up a Fake Card Provider and a Checkout Client This step builds a fake card provider on port 8090 and a checkout client that retries like a phone. The provider and the API each keep a ledger, because the bug lives in the gap between what the shop believes it charged and what the card was actually charged. Start PostgreSQL and wait until it reports accepting connections: docker run -d --name pg -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:18 docker exec pg pg_isready -U postgres Output /var/run/postgresql:5432 - accepting connections The fallback password is for this container only The API reads DATABASE_URL and the provider reads PROVIDER_DATABASE_URL. When they're unset, both fall back to this container's postgres:postgres URL, so every command here runs as shown. In a real deployment, set the variables from a secret store. Create one database per ledger: docker exec pg psql -U postgres -c "CREATE DATABASE shop" -c "CREATE DATABASE provider" Output CREATE DATABASE CREATE DATABASE Create the Go module and add pgx: mkdir payments && cd payments go mod init example.com/payments go get github.com/jackc/pgx/v5@v5.11.0 Output go: creating new go.mod: module example.com/payments go: to add module requirements and sums: go mod tidy go: added github.com/jackc/pgpassfile v1.0.0 go: added github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 go: added github.com/jackc/pgx/v5 v5.11.0 go: added golang.org/x/text v0.29.0 In the shop's schema, charges is the ledger and idempotency_keys holds one row per key. Its primary key is (customer_id, key), so two customers who send the same key never see each other's payments. schema.sql -- The shop's own ledger: one row per payment it believes it took. CREATE TABLE charges ( id bigserial PRIMARY KEY, customer_id text NOT NULL, amount bigint NOT NULL CHECK (amount > 0), provider_id text NOT NULL, created_at timestamptz NOT NULL DEFAULT now() ); -- One row per idempotency key: who claimed it, for which request, and what we answered. CREATE TABLE idempotency_keys ( customer_id text NOT NULL, key text NOT NULL, request_hash bytea NOT NULL, status text NOT NULL DEFAULT 'started', response_code int, response_body bytea, locked_at timestamptz NOT NULL DEFAULT now(), created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (customer_id, key) ); The provider's table is its record of what it actually charged. Its idempotency_key column is UNIQUE, and because a UNIQUE column treats NULLs as distinct by default, a charge sent without a key never conflicts with anything: provider.sql -- The fake card processor's own records: what it actually charged. CREATE TABLE card_charges ( id text PRIMARY KEY DEFAULT 'ch_' || substr(md5(random()::text), 1, 10), idempotency_key text UNIQUE, card text NOT NULL, amount bigint NOT NULL, created_at timestamptz NOT NULL DEFAULT now() ); Load both schemas: docker exec -i pg psql -U postgres -d shop < schema.sql docker exec -i pg psql -U postgres -d provider < provider.sql Output CREATE TABLE CREATE TABLE CREATE TABLE The fake provider takes one second per charge and declines the test card tok_declined with 402 Payment Required. With an Idempotency-Key, it behaves like Stripe: it takes a PostgreSQL advisory lock (a lock PostgreSQL holds on a number, here a hash of the key, until the transaction ends), answers 409 if another request with the same key is still running, and returns the original charge if the key was already used. cmd/provider/main.go // Command provider is a stand-in for a card processor such as Stripe or Adyen. // It charges slowly, declines one test card, and honors an Idempotency-Key. package main import ( "context" "encoding/json" "log" "net/http" "os" "time" "github.com/jackc/pgx/v5/pgxpool" ) type chargeRequest struct { Amount int64 `json:"amount"` Card string `json:"card"` } func main() { url := os.Getenv("PROVIDER_DATABASE_URL") if url == "" { url = "postgres://postgres:postgres@localhost:5432/provider?pool_max_conns=50" // the local Docker container; one connection per charge in flight } db, err := pgxpool.New(context.Background(), url) if err != nil { log.Fatal(err) } http.HandleFunc("POST /charges", func(w http.ResponseWriter, r *http.Request) { var req chargeRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "bad request", http.StatusBadRequest) return } ctx := r.Context() tx, err := db.Begin(ctx) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } defer tx.Rollback(ctx) // Like Stripe, answer 409 while another request with the same key is running, // and return the original charge once it has finished. var key *string if k := r.Header.Get("Idempotency-Key"); k != "" { key = &k var locked bool tx.QueryRow(ctx, `SELECT pg_try_advisory_xact_lock(hashtext($1))`, k).Scan(&locked) if !locked { writeJSON(w, http.StatusConflict, map[string]string{"error": "request_in_progress"}) return } var id string if tx.QueryRow(ctx, `SELECT id FROM card_charges WHERE idempotency_key = $1`, k).Scan(&id) == nil { writeJSON(w, http.StatusCreated, map[string]string{"id": id}) return } } time.Sleep(time.Second) // the card network is slow if req.Card == "tok_declined" { writeJSON(w, http.StatusPaymentRequired, map[string]string{"error": "card_declined"}) return } // Without a key (NULL), nothing conflicts: every call is a new charge. var id string err = tx.QueryRow(ctx, ` INSERT INTO card_charges (idempotency_key, card, amount) VALUES ($1, $2, $3) RETURNING id`, key, req.Card, req.Amount).Scan(&id) if err == nil { err = tx.Commit(ctx) } if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } writeJSON(w, http.StatusCreated, map[string]string{"id": id}) }) log.Println("provider listening on :8090") log.Fatal(http.ListenAndServe(":8090", nil)) } func writeJSON(w http.ResponseWriter, code int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) json.NewEncoder(w).Encode(v) } payments/server.go holds the API's shared pieces. A payment carries an amount in cents and a card token, a string the provider's card form issues so your server never sees a card number. customerID reads a header as a stand-in for authentication; a real API answers 401 when there's no authenticated customer and caps the body with http.MaxBytesReader. payments/server.go // Package payments is a small payments API that charges a card through a provider // and records each payment in PostgreSQL. package payments import ( "bytes" "context" "encoding/json" "errors" "fmt" "net/http" "time" "github.com/jackc/pgx/v5/pgxpool" ) // PaymentRequest is what the checkout page sends. Amount is in cents; Card is a // token from the provider's card form, never a card number. type PaymentRequest struct { Amount int64 `json:"amount"` Card string `json:"card"` } // Payment is what the API returns for a successful charge. type Payment struct { ID int64 `json:"id"` Amount int64 `json:"amount"` ProviderID string `json:"provider_charge_id"` } // Server holds the API's dependencies. type Server struct { DB *pgxpool.Pool Provider *Provider } // ErrDeclined means the provider refused the card. That is a final answer, not a fault. var ErrDeclined = errors.New("card declined") // ErrInProgress means the provider is still working on a charge with the same key. var ErrInProgress = errors.New("charge in progress at the provider") // Provider is a client for the card processor. type Provider struct { URL string HTTP *http.Client } // Charge asks the provider to take amount from card. A non-empty key is sent as // the provider's Idempotency-Key, so repeating the call cannot charge twice. func (p *Provider) Charge(ctx context.Context, amount int64, card, key string) (string, error) { body, _ := json.Marshal(PaymentRequest{Amount: amount, Card: card}) req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.URL+"/charges", bytes.NewReader(body)) if err != nil { return "", err } req.Header.Set("Content-Type", "application/json") if key != "" { req.Header.Set("Idempotency-Key", key) } resp, err := p.HTTP.Do(req) if err != nil { return "", err } defer resp.Body.Close() switch resp.StatusCode { case http.StatusCreated: var out struct{ ID string } err := json.NewDecoder(resp.Body).Decode(&out) return out.ID, err case http.StatusPaymentRequired: return "", ErrDeclined case http.StatusConflict: return "", ErrInProgress default: return "", fmt.Errorf("provider returned %s", resp.Status) } } // NewProvider returns a client whose timeout bounds every provider call. func NewProvider(url string) *Provider { return &Provider{URL: url, HTTP: &http.Client{Timeout: 5 * time.Second}} } // customerID stands in for authentication: a real API takes it from the session or token. func customerID(r *http.Request) string { return r.Header.Get("Customer-ID") } func writeJSON(w http.ResponseWriter, code int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) json.NewEncoder(w).Encode(v) } func writeError(w http.ResponseWriter, code int, msg string) { writeJSON(w, code, map[string]string{"error": msg}) } func decode(r *http.Request) (PaymentRequest, error) { var req PaymentRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { return req, err } if req.Amount <= 0 || req.Card == "" { return req, errors.New("amount and card are required") } return req, nil } The checkout client plays the phone. Its first attempt gives up after 600 ms; later attempts wait 3 seconds, with a 300 ms backoff that doubles, for up to 4 attempts. -key sends one key per payment (the default), one per attempt (new) or none, and -parallel fires identical copies at once. cmd/pay/main.go // Command pay is the checkout client: it sends one payment and retries the way a // mobile app does when the network is slow. package main import ( "bytes" "crypto/rand" "flag" "fmt" "io" "net/http" "strings" "sync" "time" ) func main() { amount := flag.Int("amount", 2500, "amount in cents") card := flag.String("card", "tok_visa", "card token") keyMode := flag.String("key", "once", "once: one key per payment; new: a new key per attempt; none: no key") keyValue := flag.String("key-value", "", "use this key instead of a random one") firstTimeout := flag.Duration("first-timeout", 600*time.Millisecond, "the first attempt gives up after this long (the phone loses signal)") timeout := flag.Duration("timeout", 3*time.Second, "how long later attempts wait for an answer") attempts := flag.Int("attempts", 4, "attempts before giving up") parallel := flag.Int("parallel", 1, "send this many copies at the same moment") flag.Parse() key := *keyValue if key == "" { key = rand.Text() // 26 random characters from crypto/rand } body := fmt.Sprintf(`{"amount": %d, "card": %q}`, *amount, *card) send := func(attempt int) (int, string, error) { req, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/payments", strings.NewReader(body)) req.Header.Set("Content-Type", "application/json") req.Header.Set("Customer-ID", "cus_1001") switch *keyMode { case "once": req.Header.Set("Idempotency-Key", key) case "new": req.Header.Set("Idempotency-Key", rand.Text()) } wait := *timeout if attempt == 1 { wait = *firstTimeout } resp, err := (&http.Client{Timeout: wait}).Do(req) if err != nil { return 0, "", err } defer resp.Body.Close() b, _ := io.ReadAll(resp.Body) extra := "" if resp.Header.Get("Idempotent-Replayed") == "true" { extra = " (replayed)" } return resp.StatusCode, string(bytes.TrimSpace(b)) + extra, nil } // One payment, retried with exponential backoff until it gets a final answer. pay := func(id int) { backoff := 300 * time.Millisecond for attempt := 1; attempt <= *attempts; attempt++ { code, msg, err := send(attempt) prefix := fmt.Sprintf("attempt %d:", attempt) if *parallel > 1 { prefix = fmt.Sprintf("copy %d %s", id, prefix) } switch { case err != nil: fmt.Println(prefix, "no answer:", shorten(err)) case code == http.StatusConflict || code >= 500: fmt.Println(prefix, code, msg) default: fmt.Println(prefix, code, msg) return } time.Sleep(backoff) backoff *= 2 } } var wg sync.WaitGroup for i := 1; i <= *parallel; i++ { wg.Add(1) go func() { defer wg.Done(); pay(i) }() } wg.Wait() } func shorten(err error) string { if strings.Contains(err.Error(), "Client.Timeout") { return "timed out" } return err.Error() } pgx's connection pool needs two modules that go get didn't record, so let Go fill them in before the first build: go mod tidy In terminal 1, start the provider and leave it running. It logs provider listening on :8090: go run ./cmd/provider Between steps, you'll empty both ledgers with these two commands: docker exec pg psql -U postgres -d shop -c "TRUNCATE charges, idempotency_keys RESTART IDENTITY" docker exec pg psql -U postgres -d provider -c "TRUNCATE card_charges" Step 2: Reproduce the Double Charge This step builds the handler most payment endpoints start as. Create it: payments/naive.go package payments import ( "context" "errors" "net/http" ) // CreatePaymentNaive charges the card and records the payment. Every call charges again. func (s *Server) CreatePaymentNaive(w http.ResponseWriter, r *http.Request) { req, err := decode(r) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } // Finish the payment even if the caller hangs up: a charge must not be half done. ctx := context.WithoutCancel(r.Context()) providerID, err := s.Provider.Charge(ctx, req.Amount, req.Card, "") if errors.Is(err, ErrDeclined) { writeError(w, http.StatusPaymentRequired, "card declined") return } if err != nil { writeError(w, http.StatusBadGateway, "payment provider unavailable") return } var p Payment err = s.DB.QueryRow(ctx, ` INSERT INTO charges (customer_id, amount, provider_id) VALUES ($1, $2, $3) RETURNING id, amount, provider_id`, customerID(r), req.Amount, providerID).Scan(&p.ID, &p.Amount, &p.ProviderID) if err != nil { writeError(w, http.StatusInternalServerError, "could not record payment") return } writeJSON(w, http.StatusCreated, p) } One line matters before you run it. net/http cancels a request's context when the client's connection closes. A call cancelled mid-charge leaves you not knowing whether the card was charged, so context.WithoutCancel keeps the context's values but drops the cancellation, and the server finishes what it started. That's right for a payment, and it's also what makes the retry dangerous: the first attempt keeps going after the client has moved on. The API command is a teaching rig: one binary that serves a version of the handler picked with -mode. This first version knows only the naive one; Steps 3 and 4 add the others. cmd/api/main.go (first version) // Command api serves POST /payments in one of three versions, chosen with -mode. package main import ( "context" "flag" "log" "net/http" "os" "time" "example.com/payments/payments" "github.com/jackc/pgx/v5/pgxpool" ) func main() { mode := flag.String("mode", "idempotent", "naive, checkfirst or idempotent") flag.Parse() db, err := pgxpool.New(context.Background(), env("DATABASE_URL", "postgres://postgres:postgres@localhost:5432/shop")) if err != nil { log.Fatal(err) } s := &payments.Server{DB: db, Provider: payments.NewProvider("http://localhost:8090")} var h http.HandlerFunc switch *mode { case "naive": h = s.CreatePaymentNaive default: log.Fatalf("unknown -mode %q", *mode) } http.HandleFunc("POST /payments", h) srv := &http.Server{Addr: ":8080", ReadHeaderTimeout: 5 * time.Second} log.Printf("api (%s) listening on :8080", *mode) log.Fatal(srv.ListenAndServe()) } // env returns the variable's value, or a default that suits the local Docker container. func env(name, fallback string) string { if v := os.Getenv(name); v != "" { return v } return fallback } In terminal 2, start the API in naive mode: go run ./cmd/api -mode naive In terminal 3, pay once with no idempotency key. Expect the first attempt to time out and the retry to succeed: go run ./cmd/pay -key none Output attempt 1: no answer: timed out attempt 2: 201 {"id":2,"amount":2500,"provider_charge_id":"ch_abd3fed44e"} From the customer's side, this is one payment: a spinner, then a receipt. Now look at both ledgers: docker exec pg psql -U postgres -d shop -c "SELECT id, amount, provider_id FROM charges" docker exec pg psql -U postgres -d provider -c "SELECT id, idempotency_key, amount FROM card_charges" Output id | amount | provider_id ----+--------+--------------- 1 | 2500 | ch_d03cc74dad 2 | 2500 | ch_abd3fed44e id | idempotency_key | amount ---------------+-----------------+-------- ch_d03cc74dad | | 2500 ch_abd3fed44e | | 2500 The customer saw one success and was charged twice. The receipt gives it away: payment id 2, because row 1 came from the attempt the phone gave up on. Nothing in either ledger marks the rows as the same payment. The server did exactly what HTTP allows. POST isn't idempotent, meaning sending it twice isn't guaranteed to have the same effect as sending it once, and RFC 9110 says so in as many words: > Of the request methods defined by this specification, PUT, DELETE, and safe request methods are idempotent. > > A client SHOULD NOT automatically retry a request with a non-idempotent method unless it has some means to know that the request semantics are actually idempotent, regardless of the method, or some means to detect that the original request was never applied. > > RFC 9110, section 9.2.2 The client broke that rule, but refusing to retry would strand customers on every dropped connection. An idempotency key gives it "some means to know". Step 3: Add an Idempotency Key and Check It First This step adds the fix most people write first: look the key up, and charge only if it's new. The client already sends a key by default, 26 random characters from rand.Text, generated once per payment and repeated on every attempt. Create the check-first handler: payments/checkfirst.go package payments import ( "context" "encoding/json" "errors" "net/http" "github.com/jackc/pgx/v5" ) // CreatePaymentCheckFirst looks the key up, and only charges if it has never seen it. // The lookup and the save are separate steps, so two requests can both miss. func (s *Server) CreatePaymentCheckFirst(w http.ResponseWriter, r *http.Request) { key := r.Header.Get("Idempotency-Key") if key == "" { writeError(w, http.StatusBadRequest, "Idempotency-Key header is required") return } req, err := decode(r) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } ctx := context.WithoutCancel(r.Context()) // 1. Have we answered this key before? If so, send the same answer again. var code int var body []byte err = s.DB.QueryRow(ctx, ` SELECT response_code, response_body FROM idempotency_keys WHERE customer_id = $1 AND key = $2`, customerID(r), key).Scan(&code, &body) if err == nil { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) w.Write(body) return } if !errors.Is(err, pgx.ErrNoRows) { writeError(w, http.StatusInternalServerError, "could not read idempotency key") return } // 2. New key: charge and record the payment. providerID, err := s.Provider.Charge(ctx, req.Amount, req.Card, "") if err != nil { writeError(w, http.StatusBadGateway, "payment provider unavailable") return } var p Payment err = s.DB.QueryRow(ctx, ` INSERT INTO charges (customer_id, amount, provider_id) VALUES ($1, $2, $3) RETURNING id, amount, provider_id`, customerID(r), req.Amount, providerID).Scan(&p.ID, &p.Amount, &p.ProviderID) if err != nil { writeError(w, http.StatusInternalServerError, "could not record payment") return } // 3. Remember the answer for next time. body, _ = json.Marshal(p) _, err = s.DB.Exec(ctx, ` INSERT INTO idempotency_keys (customer_id, key, request_hash, status, response_code, response_body) VALUES ($1, $2, '', 'completed', 201, $3) ON CONFLICT DO NOTHING`, customerID(r), key, body) if err != nil { writeError(w, http.StatusInternalServerError, "could not save idempotency key") return } writeJSON(w, http.StatusCreated, p) } It reads well, but the provider's one-second charge sits between step 1, the lookup, and step 3, the save. Add its case to the switch in cmd/api/main.go, under the naive case: cmd/api/main.go (excerpt) case "checkfirst": h = s.CreatePaymentCheckFirst In terminal 2, stop the API with <kbd>Ctrl</kbd>+<kbd>C</kbd> and start it in check-first mode. Reset both ledgers with the two TRUNCATE commands from Step 1, then pay with the default key behavior: go run ./cmd/api -mode checkfirst Then run: go run ./cmd/pay Output attempt 1: no answer: timed out attempt 2: 201 {"id":2,"amount":2500,"provider_charge_id":"ch_0ee40e9309"} id 2 again, and both ledgers hold two rows, as in Step 2. The key table holds one row, because the second save hit ON CONFLICT DO NOTHING: docker exec pg psql -U postgres -d shop -c "SELECT key, status, response_code FROM idempotency_keys" Output key | status | response_code ----------------------------+-----------+--------------- G2D7ZUOEGVRORUF253FGPBOFKF | completed | 201 (1 row) The retry arrived at about 0.9 s (600 ms timeout plus 300 ms backoff), while the first request was at the provider until about 1.0 s and hadn't saved the key, so the lookup found nothing: View this diagram in the original post _The retry looks the key up while the first request is still at the provider. Both lookups miss, and both requests charge._ This is a check-then-act race : a decision based on a read that's stale by the time you act on it. The key is only checked, never claimed, and ownership has to come from the one component that sees every request in order: the database. Step 4: Claim the Idempotency Key Atomically This step makes the first request to arrive own the key before it does any work. Create payments/idempotent.go from the collapsed block; this step explains the claim and the stored answer, Steps 5 and 6 the rest. payments/idempotent.go (the full file) payments/idempotent.go package payments import ( "bytes" "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "net/http" "os" "strings" "time" "github.com/jackc/pgx/v5" ) // Options switch on the protections one at a time, so each step of the tutorial // can show what the previous version gets wrong. type Options struct { Lease time.Duration // 0: a started key is never taken over PassKeyToCharge bool // send our key to the provider as its Idempotency-Key CrashAfterCharge bool // exit after the provider charges, before we record it } type keyRecord struct { status string hash []byte lockedAt time.Time // when this request claimed the key; proves it still owns it } // fingerprint identifies the request a key was first used with. Hashing the decoded // struct, not the raw bytes, ignores whitespace and field order. func fingerprint(r *http.Request, req PaymentRequest) []byte { canonical, _ := json.Marshal(req) sum := sha256.Sum256(append([]byte(r.Method+" "+r.URL.Path+"\n"), canonical...)) return sum[:] } // claim tries to make this request the owner of the key. It returns true, and the // claim's locked_at, if this request should do the work. Otherwise it returns the // key's current record. func (s *Server) claim(ctx context.Context, customer, key string, hash []byte, opt Options) (bool, keyRecord, error) { var rec keyRecord var err error if opt.Lease == 0 { err = s.DB.QueryRow(ctx, ` INSERT INTO idempotency_keys (customer_id, key, request_hash) VALUES ($1, $2, $3) ON CONFLICT (customer_id, key) DO NOTHING RETURNING locked_at`, customer, key, hash).Scan(&rec.lockedAt) } else { // Also take over a key whose owner started long ago and never finished. err = s.DB.QueryRow(ctx, ` INSERT INTO idempotency_keys (customer_id, key, request_hash) VALUES ($1, $2, $3) ON CONFLICT (customer_id, key) DO UPDATE SET locked_at = now() WHERE idempotency_keys.status = 'started' AND idempotency_keys.request_hash = EXCLUDED.request_hash AND idempotency_keys.locked_at < now() - make_interval(secs => $4) RETURNING locked_at`, customer, key, hash, opt.Lease.Seconds()).Scan(&rec.lockedAt) } if err == nil { return true, rec, nil } if !errors.Is(err, pgx.ErrNoRows) { return false, keyRecord{}, err } // Someone else holds the key. Read what they have so far. err = s.DB.QueryRow(ctx, ` SELECT status, request_hash FROM idempotency_keys WHERE customer_id = $1 AND key = $2`, customer, key).Scan(&rec.status, &rec.hash) if errors.Is(err, pgx.ErrNoRows) { // Released a moment ago: report it as in progress so the client retries. rec.status, rec.hash = "started", hash return false, rec, nil } return false, rec, err } // CreatePayment charges at most once per Idempotency-Key and replays the first answer. func (s *Server) CreatePayment(opt Options) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { customer := customerID(r) key := idempotencyKey(r) if key == "" || len(key) > 255 { writeError(w, http.StatusBadRequest, "Idempotency-Key header is required (at most 255 characters)") return } req, err := decode(r) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } hash := fingerprint(r, req) ctx := context.WithoutCancel(r.Context()) claimed, rec, err := s.claim(ctx, customer, key, hash, opt) switch { case err != nil: writeError(w, http.StatusInternalServerError, "could not claim idempotency key") return case claimed: // This request owns the key; carry on below. case !bytes.Equal(rec.hash, hash): writeError(w, http.StatusUnprocessableEntity, "Idempotency-Key was already used with a different request") return case rec.status == "completed": s.replay(ctx, w, customer, key) return default: w.Header().Set("Retry-After", "1") writeError(w, http.StatusConflict, "a request with this Idempotency-Key is still in progress") return } chargeKey := "" if opt.PassKeyToCharge { // Derived from our key, so every retry and takeover sends the same one. // Hashed to 64 hex characters to fit providers' length limits. sum := sha256.Sum256([]byte(customer + ":" + key)) chargeKey = hex.EncodeToString(sum[:]) } providerID, err := s.Provider.Charge(ctx, req.Amount, req.Card, chargeKey) if errors.Is(err, ErrDeclined) { // A decline is a final answer: store it so a retry gets the same one. body, _ := json.Marshal(map[string]string{"error": "card declined"}) err := s.finish(ctx, customer, key, http.StatusPaymentRequired, body, nil) if errors.Is(err, errLostKey) { s.replay(ctx, w, customer, key) return } if err != nil { writeError(w, http.StatusInternalServerError, "could not save result") return } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusPaymentRequired) w.Write(body) return } if errors.Is(err, ErrInProgress) { // The provider is still charging under this key (an earlier claim we took // over is mid-call). Not a fault: keep the key and let the client retry. w.Header().Set("Retry-After", "1") writeError(w, http.StatusConflict, "a request with this Idempotency-Key is still in progress") return } if err != nil { // A fault is not an answer. Release the key so a retry can run the charge // again, but only if it is still our claim and not a later takeover's. s.DB.Exec(ctx, ` DELETE FROM idempotency_keys WHERE customer_id = $1 AND key = $2 AND status = 'started' AND locked_at = $3`, customer, key, rec.lockedAt) writeError(w, http.StatusBadGateway, "payment provider unavailable") return } if opt.CrashAfterCharge { os.Exit(1) // simulate a deploy or OOM kill at the worst moment } var p Payment record := func(tx pgx.Tx) ([]byte, error) { err := tx.QueryRow(ctx, ` INSERT INTO charges (customer_id, amount, provider_id) VALUES ($1, $2, $3) RETURNING id, amount, provider_id`, customer, req.Amount, providerID).Scan(&p.ID, &p.Amount, &p.ProviderID) if err != nil { return nil, err } return json.Marshal(p) } err = s.finish(ctx, customer, key, http.StatusCreated, nil, record) if errors.Is(err, errLostKey) { s.replay(ctx, w, customer, key) // a takeover finished first; answer as it did return } if err != nil { writeError(w, http.StatusInternalServerError, "could not record payment") return } writeJSON(w, http.StatusCreated, p) } } // errLostKey means another request finished this key first. var errLostKey = errors.New("idempotency key already finished") // finish saves the answer for a key, and the payment row when there is one, in one // transaction. Only a request that still sees the key as started may finish it. func (s *Server) finish(ctx context.Context, customer, key string, code int, body []byte, record func(pgx.Tx) ([]byte, error)) error { tx, err := s.DB.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) var status string err = tx.QueryRow(ctx, ` SELECT status FROM idempotency_keys WHERE customer_id = $1 AND key = $2 FOR UPDATE`, customer, key).Scan(&status) if err != nil { return err } if status != "started" { return errLostKey } if record != nil { if body, err = record(tx); err != nil { return err } } _, err = tx.Exec(ctx, ` UPDATE idempotency_keys SET status = 'completed', response_code = $3, response_body = $4 WHERE customer_id = $1 AND key = $2`, customer, key, code, body) if err != nil { return err } return tx.Commit(ctx) } // replay sends the stored answer for a finished key. func (s *Server) replay(ctx context.Context, w http.ResponseWriter, customer, key string) { var code int var body []byte err := s.DB.QueryRow(ctx, ` SELECT response_code, response_body FROM idempotency_keys WHERE customer_id = $1 AND key = $2`, customer, key).Scan(&code, &body) if err != nil { writeError(w, http.StatusInternalServerError, "could not read stored result") return } w.Header().Set("Content-Type", "application/json") w.Header().Set("Idempotent-Replayed", "true") w.WriteHeader(code) w.Write(body) } // idempotencyKey reads the header. The IETF draft sends it as a quoted string, // most providers send it bare; accept both. func idempotencyKey(r *http.Request) string { k := strings.TrimSpace(r.Header.Get("Idempotency-Key")) if len(k) >= 2 && k[0] == '"' && k[len(k)-1] == '"' { k = k[1 : len(k)-1] } return k } Then replace cmd/api/main.go with its final version, which defaults to the idempotent handler and adds three flags for Step 6: cmd/api/main.go // Command api serves POST /payments in one of three versions, chosen with -mode. package main import ( "context" "flag" "log" "net/http" "os" "time" "example.com/payments/payments" "github.com/jackc/pgx/v5/pgxpool" ) func main() { mode := flag.String("mode", "idempotent", "naive, checkfirst or idempotent") lease := flag.Duration("lease", 0, "take over a started key after this long (0 = never)") passKey := flag.Bool("pass-key", false, "send the idempotency key to the provider") crash := flag.Bool("crash-after-charge", false, "exit after the provider charges") flag.Parse() db, err := pgxpool.New(context.Background(), env("DATABASE_URL", "postgres://postgres:postgres@localhost:5432/shop")) if err != nil { log.Fatal(err) } s := &payments.Server{DB: db, Provider: payments.NewProvider("http://localhost:8090")} var h http.HandlerFunc switch *mode { case "naive": h = s.CreatePaymentNaive case "checkfirst": h = s.CreatePaymentCheckFirst case "idempotent": h = s.CreatePayment(payments.Options{Lease: *lease, PassKeyToCharge: *passKey, CrashAfterCharge: *crash}) default: log.Fatalf("unknown -mode %q", *mode) } http.HandleFunc("POST /payments", h) srv := &http.Server{Addr: ":8080", ReadHeaderTimeout: 5 * time.Second} log.Printf("api (%s) listening on :8080", *mode) log.Fatal(srv.ListenAndServe()) } // env returns the variable's value, or a default that suits the local Docker container. func env(name, fallback string) string { if v := os.Getenv(name); v != "" { return v } return fallback } PostgreSQL's INSERT ... ON CONFLICT does the check and the claim in one statement, arbitrated by the primary key. With DO NOTHING, a conflicting row isn't inserted, and RETURNING reports only what changed: > Only rows that were successfully inserted or updated will be returned. > > PostgreSQL 18 documentation, INSERT So the claim returns the new key's locked_at if this request inserted it, and zero rows if another request got there first. In pgx, zero rows shows up as pgx.ErrNoRows from Scan, the "lost the claim" signal: payments/idempotent.go (excerpt) // claim tries to make this request the owner of the key. It returns true, and the // claim's locked_at, if this request should do the work. Otherwise it returns the // key's current record. func (s *Server) claim(ctx context.Context, customer, key string, hash []byte, opt Options) (bool, keyRecord, error) { var rec keyRecord var err error if opt.Lease == 0 { err = s.DB.QueryRow(ctx, ` INSERT INTO idempotency_keys (customer_id, key, request_hash) VALUES ($1, $2, $3) ON CONFLICT (customer_id, key) DO NOTHING RETURNING locked_at`, customer, key, hash).Scan(&rec.lockedAt) } else { // Also take over a key whose owner started long ago and never finished. err = s.DB.QueryRow(ctx, ` INSERT INTO idempotency_keys (customer_id, key, request_hash) VALUES ($1, $2, $3) ON CONFLICT (customer_id, key) DO UPDATE SET locked_at = now() WHERE idempotency_keys.status = 'started' AND idempotency_keys.request_hash = EXCLUDED.request_hash AND idempotency_keys.locked_at < now() - make_interval(secs => $4) RETURNING locked_at`, customer, key, hash, opt.Lease.Seconds()).Scan(&rec.lockedAt) } if err == nil { return true, rec, nil } if !errors.Is(err, pgx.ErrNoRows) { return false, keyRecord{}, err } // Someone else holds the key. Read what they have so far. err = s.DB.QueryRow(ctx, ` SELECT status, request_hash FROM idempotency_keys WHERE customer_id = $1 AND key = $2`, customer, key).Scan(&rec.status, &rec.hash) if errors.Is(err, pgx.ErrNoRows) { // Released a moment ago: report it as in progress so the client retries. rec.status, rec.hash = "started", hash return false, rec, nil } return false, rec, err } The else branch adds a lease for Step 6. If the key vanishes between the two queries (released after a fault, as in Step 5), the request reports it as in progress so the client retries. The claim runs in its own implicit transaction, so it commits the moment the statement finishes. That's deliberate. If the claim and the charge shared one transaction, a retry's INSERT would wait on the unique index for the whole provider call, holding a database connection, and never get its 409. Commit the claim on its own and keep the slow work outside any transaction. The handler acts on what claim reports: payments/idempotent.go (excerpt) // CreatePayment charges at most once per Idempotency-Key and replays the first answer. func (s *Server) CreatePayment(opt Options) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { customer := customerID(r) key := idempotencyKey(r) if key == "" || len(key) > 255 { writeError(w, http.StatusBadRequest, "Idempotency-Key header is required (at most 255 characters)") return } req, err := decode(r) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } hash := fingerprint(r, req) ctx := context.WithoutCancel(r.Context()) claimed, rec, err := s.claim(ctx, customer, key, hash, opt) switch { case err != nil: writeError(w, http.StatusInternalServerError, "could not claim idempotency key") return case claimed: // This request owns the key; carry on below. case !bytes.Equal(rec.hash, hash): writeError(w, http.StatusUnprocessableEntity, "Idempotency-Key was already used with a different request") return case rec.status == "completed": s.replay(ctx, w, customer, key) return default: w.Header().Set("Retry-After", "1") writeError(w, http.StatusConflict, "a request with this Idempotency-Key is still in progress") return } | What claim reports | Response | | --- | --- | | This request owns the key | Charge the card and record the payment | | Key exists with a different request | 422 Unprocessable Content (Step 5) | | Key exists and is completed | Replay the stored status and body | | Key exists and is started | 409 Conflict with Retry-After: 1 | Retry-After is a standard HTTP header that tells the client how many seconds to wait. The codes follow the Idempotency-Key header draft from the IETF, the body that publishes internet standards: 400 for a missing key, 422 for a key reused with a different payload, and 409 when "A request is outstanding for this Idempotency-Key". The draft expired in April 2026 without becoming an RFC, so treat it as a convention. It also says a retry of a completed request gets the previous result, success or error. After a successful charge, finish saves the shop's ledger row and the key's stored answer in one transaction: payments/idempotent.go (excerpt) var p Payment record := func(tx pgx.Tx) ([]byte, error) { err := tx.QueryRow(ctx, ` INSERT INTO charges (customer_id, amount, provider_id) VALUES ($1, $2, $3) RETURNING id, amount, provider_id`, customer, req.Amount, providerID).Scan(&p.ID, &p.Amount, &p.ProviderID) if err != nil { return nil, err } return json.Marshal(p) } err = s.finish(ctx, customer, key, http.StatusCreated, nil, record) if errors.Is(err, errLostKey) { s.replay(ctx, w, customer, key) // a takeover finished first; answer as it did return } if err != nil { writeError(w, http.StatusInternalServerError, "could not record payment") return } writeJSON(w, http.StatusCreated, p) payments/idempotent.go (excerpt) // errLostKey means another request finished this key first. var errLostKey = errors.New("idempotency key already finished") // finish saves the answer for a key, and the payment row when there is one, in one // transaction. Only a request that still sees the key as started may finish it. func (s *Server) finish(ctx context.Context, customer, key string, code int, body []byte, record func(pgx.Tx) ([]byte, error)) error { tx, err := s.DB.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) var status string err = tx.QueryRow(ctx, ` SELECT status FROM idempotency_keys WHERE customer_id = $1 AND key = $2 FOR UPDATE`, customer, key).Scan(&status) if err != nil { return err } if status != "started" { return errLostKey } if record != nil { if body, err = record(tx); err != nil { return err } } _, err = tx.Exec(ctx, ` UPDATE idempotency_keys SET status = 'completed', response_code = $3, response_body = $4 WHERE customer_id = $1 AND key = $2`, customer, key, code, body) if err != nil { return err } return tx.Commit(ctx) } The AWS Builders' Library article on making retries safe says recording the token and the mutations "needs to be an 'all or nothing' or atomic process"; otherwise a crash between the two writes leaves a payment no retry can see. The status check and FOR UPDATE matter in Step 6. replay returns the stored answer with Idempotent-Replayed: true, a header Stripe uses so a client can tell a replay from fresh work. No standard defines it. payments/idempotent.go (excerpt) // replay sends the stored answer for a finished key. func (s *Server) replay(ctx context.Context, w http.ResponseWriter, customer, key string) { var code int var body []byte err := s.DB.QueryRow(ctx, ` SELECT response_code, response_body FROM idempotency_keys WHERE customer_id = $1 AND key = $2`, customer, key).Scan(&code, &body) if err != nil { writeError(w, http.StatusInternalServerError, "could not read stored result") return } w.Header().Set("Content-Type", "application/json") w.Header().Set("Idempotent-Replayed", "true") w.WriteHeader(code) w.Write(body) } Restart the API in terminal 2 with no flags, reset both ledgers, and pay with a fixed key: go run ./cmd/api Then run: go run ./cmd/pay -key-value order-7731 docker exec pg psql -U postgres -d shop -c "SELECT key, status, response_code, convert_from(response_body, 'UTF8') AS response_body FROM idempotency_keys" Output attempt 1: no answer: timed out attempt 2: 409 {"error":"a request with this Idempotency-Key is still in progress"} attempt 3: 201 {"id":1,"amount":2500,"provider_charge_id":"ch_b2d0620c57"} (replayed) key | status | response_code | response_body ------------+-----------+---------------+------------------------------------

pgx 5.11 parses connection strings like libpq. In a URI, ?application_name=my+app gave "my app" on 5.10 and gives "my+app" on 5.11 in our run; psql 18.6 agrees with the new reading. Use %20 for a space. github.com/jackc/pgx/blob/master/CHANGELOG.md