The Outbox Pattern
Insert the job in the same transaction as the data, then let a relay publish it, so the two cannot disagree.
BEGIN; INSERT INTO orders ...; INSERT INTO outbox (topic, payload) VALUES (...); COMMIT;An order is saved, and a confirmation email has to be sent. That is two systems: the database and the queue. Write to both and there is a gap between the two writes. A crash in that gap leaves an order with no email, or an email for an order that was rolled back. This is the dual write problem. Retrying does not fix it, because the crash can land anywhere, including after the publish and before you knew it succeeded.
The pattern
Write to one system only: the database. Add a table called the outbox, with columns for an id, a topic, a payload and published_at. In the same transaction that saves the order, insert one outbox row describing the job. Commit. Now either both rows exist or neither does, because that is what a transaction means.
A separate loop, the relay, selects rows where published_at is null, publishes each one to the real queue, and then sets published_at. The relay is the only code that talks to the queue.
What trips people up
Order matters inside the relay. Publish first, then mark the row. The other way round loses messages when the process dies between the two steps. Publishing first means a crash after the publish will publish again, so consumers must tolerate duplicates.
Put enough in the payload that the consumer does not have to re-read the row. And prune the table: an outbox nobody deletes from becomes the largest table in the database.
Example
package main
import "fmt"
// db holds the committed rows. tx holds writes until commit, the way a SQL
// transaction does: in real code tx is *sql.Tx and each write is an INSERT.
type db struct{ rows []string }
type tx struct {
db *db
pending []string
}
func (t *tx) insert(row string) { t.pending = append(t.pending, row) }
func (t *tx) commit() { t.db.rows = append(t.db.rows, t.pending...) }
func main() {
d := &db{}
t1 := &tx{db: d}
t1.insert("orders: order-1")
t1.insert("outbox: order_placed order-1") // same transaction as the order
t1.commit()
t2 := &tx{db: d}
t2.insert("orders: order-2")
t2.insert("outbox: order_placed order-2")
// This request failed before commit, so neither row lands. There is no
// state in which the order exists and its job does not.
for _, r := range d.rows {
fmt.Println(r)
}
}
// Output:
// orders: order-1
// outbox: order_placed order-1
When to use it
- Guarantee that every committed order produces exactly one confirmation job.
- Publish a 'user deleted' event with no window where the user is gone and nothing was announced.
- Survive a queue outage: jobs wait as rows until the queue comes back.
- Keep a search index in step with the database without a second write from the handler.
More examples
Writing the order and its job together
The producing half of the pattern. Use it in any handler whose work has to be visible to another service: the handler's only job is to commit, and nothing it does can leave the database and the queue disagreeing.
// placeOrder saves the order and the job that announces it in one transaction.
func placeOrder(ctx context.Context, db *sql.DB, userID, total int64) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback() // no-op after a successful Commit
var orderID int64
if err := tx.QueryRowContext(ctx,
`INSERT INTO orders (user_id, total) VALUES ($1, $2) RETURNING id`,
userID, total).Scan(&orderID); err != nil {
return fmt.Errorf("insert order: %w", err)
}
// The payload carries what the consumer needs, so it never has to
// re-read the order and see a later version of it.
if _, err := tx.ExecContext(ctx,
`INSERT INTO outbox (topic, payload) VALUES ($1, $2)`,
"order_placed", fmt.Sprintf(`{"order_id":%d,"total":%d}`, orderID, total)); err != nil {
return fmt.Errorf("insert outbox: %w", err)
}
return tx.Commit()
}The relay loop
The consuming half. Rows are read in id order and marked only after a successful publish, which makes the relay lose nothing and occasionally repeat. If the queue is down, rows simply accumulate and go out when it returns.
type outboxMsg struct {
id int64
topic string
payload string
}
// relayOnce publishes one batch. Call it from a ticker.
func relayOnce(ctx context.Context, db *sql.DB, pub func(topic, payload string) error) error {
rows, err := db.QueryContext(ctx, `SELECT id, topic, payload FROM outbox
WHERE published_at IS NULL ORDER BY id LIMIT 100`)
if err != nil {
return err
}
defer rows.Close()
var batch []outboxMsg
for rows.Next() {
var m outboxMsg
if err := rows.Scan(&m.id, &m.topic, &m.payload); err != nil {
return err
}
batch = append(batch, m)
}
if err := rows.Err(); err != nil { // the loop can end on an error, silently
return err
}
for _, m := range batch {
if err := pub(m.topic, m.payload); err != nil {
return err // published_at stays null, so this row is retried next tick
}
if _, err := db.ExecContext(ctx,
`UPDATE outbox SET published_at = now() WHERE id = $1`, m.id); err != nil {
return err
}
}
return nil
}The dual write this replaces
Worth keeping in mind because it is the obvious way to write the code and it looks correct in review. Recognising this shape in an existing service is usually the reason somebody goes looking for the outbox pattern.
// Do not do this. Two writes, two systems, a gap in the middle.
if err := saveOrder(ctx, db, o); err != nil {
return err
}
// A crash on this line ships an order that nobody is ever told about.
if err := queue.Publish(ctx, "order_placed", o.ID); err != nil {
// This error has no good handling. The order is already committed, so you
// cannot undo it, and a publish that timed out may still have arrived.
return err
}
Discussion