Outbox and side effects

A use case that changes data and then sends an email has a gap between the two. If the process dies after the commit, the email is never sent. If the email goes first and the commit fails, the user gets a link for a change that never happened.

The outbox closes the gap. The use case doesn’t send the email: it adds a message to an outbox table in the same transaction as the change. The message exists exactly when the change does. A relay then delivers it.

Adding a message

// NewMessage returns a message for topic with payload encoded as JSON.
func NewMessage(topic string, payload any, now time.Time) (Message, error) {
  data, err := json.Marshal(payload)
  if err != nil {
    return Message{}, fmt.Errorf("outbox: encode %s: %w", topic, err)
  }
  return Message{
    ID:        uuid.NewV7(),
    Topic:     topic,
    Payload:   data,
    CreatedAt: now.UTC().Truncate(time.Millisecond),
  }, nil
}

A message has a topic, a JSON payload and an ID. Its store has the same interface on every backend, and Add takes the same uow.Tx as the event store:

// Message is one side effect to deliver.
type Message struct {
  ID    uuid.UUID
  Topic string
  // Payload is JSON.
  Payload   []byte
  CreatedAt time.Time
  // Attempts counts the times the message was claimed, this one included.
  // Store.Claim sets it.
  Attempts int
}

// Store keeps messages until they are delivered.
type Store interface {
  // Add stores messages as part of tx. Each becomes due at its CreatedAt.
  Add(ctx context.Context, tx uow.Tx, msgs ...Message) error
  // Claim leases up to limit due messages, earliest due first, until now+lease,
  // and increments their attempts. A due message is one that isn't buried,
  // whose retry time has passed and whose lease, if any, has ended, so a
  // message claimed by a relay that crashed is claimed again. Concurrent
  // claims never return the same message.
  Claim(ctx context.Context, now time.Time, lease time.Duration, limit int) ([]Message, error)
  // Ack removes a delivered message.
  Ack(ctx context.Context, id uuid.UUID) error
  // Retry ends a message's lease and makes it due again at at, keeping
  // reason as its last error.
  Retry(ctx context.Context, id uuid.UUID, at time.Time, reason string) error
  // Bury stops delivering a message and keeps it, with reason, for an
  // operator to inspect.
  Bury(ctx context.Context, id uuid.UUID, reason string) error
}

Delivery

The relay polls for due messages, claims a batch under a lease, and hands each to the handler for its topic. A handler that returns an error gets the message again later:

// Handler delivers one message. Returning an error makes the relay retry
// the message later.
type Handler func(ctx context.Context, m Message) error

// RelayConfig tunes a Relay. Zero fields take the defaults shown.
type RelayConfig struct {
  // Interval is the wait between polls once no message is due.
  Interval time.Duration `env:"INTERVAL" envDefault:"1s" desc:"Wait between outbox polls once no message is due."`
  // Lease is how long a claimed message stays claimed. It must outlast
  // Timeout, or a slow delivery can overlap its own retry.
  Lease time.Duration `env:"LEASE" envDefault:"2m" desc:"How long a claimed message stays claimed; must outlast the timeout."`
  // Timeout bounds one handler call.
  Timeout time.Duration `env:"TIMEOUT" envDefault:"30s" desc:"Longest one delivery may take."`
  // Batch is how many messages one claim takes.
  Batch int `env:"BATCH" envDefault:"10" desc:"Messages claimed per poll."`
  // MaxAttempts is how many times a message is tried before it is buried.
  MaxAttempts int `env:"MAX_ATTEMPTS" envDefault:"8" desc:"Deliveries tried before a message is buried."`
}

Each RelayConfig field is read from the environment with the service’s OUTBOX_ prefix: OUTBOX_INTERVAL, OUTBOX_LEASE, OUTBOX_TIMEOUT, OUTBOX_BATCH and OUTBOX_MAX_ATTEMPTS.

Retries back off from one second, doubling each time, up to five minutes:

// Backoff is the wait before retrying a message that failed on its
// attempt-th try: one second, doubling each time, at most five minutes.
func Backoff(attempt int) time.Duration {
  const ceiling = 5 * time.Minute
  if attempt < 1 {
    attempt = 1
  }
  if attempt > 10 {
    return ceiling
  }
  return min(time.Second<<(attempt-1), ceiling)
}

The test drives the relay with an injected clock, so it needs no sleeps:

func TestRelayRetriesUntilDelivered(t *testing.T) {
  f := newFixture(t)
  var calls []int
  var delivered []uuid.UUID
  f.handle = func(_ context.Context, m outbox.Message) error {
    calls = append(calls, m.Attempts)
    if len(calls) < 3 {
      return errors.New("smtp: connection refused")
    }
    delivered = append(delivered, m.ID)
    return nil
  }
  m := f.add(t, "mail")

  f.drain(t) // fails, retry in 1s
  f.drain(t) // not due yet
  f.now = f.now.Add(outbox.Backoff(1))
  f.drain(t) // fails, retry in 2s
  f.now = f.now.Add(outbox.Backoff(2))
  f.drain(t) // succeeds
  f.now = f.now.Add(time.Hour)
  f.drain(t) // nothing left

  if !slices.Equal(calls, []int{1, 2, 3}) {
    t.Errorf("handler saw attempts %v, want [1 2 3]", calls)
  }
  if !slices.Equal(delivered, []uuid.UUID{m.ID}) {
    t.Errorf("delivered %v, want exactly %s", delivered, m.ID)
  }
}

At least once, so handlers are idempotent

A relay can crash after the handler succeeded but before it acknowledged the message. When the lease runs out, another relay claims the message and delivers it again. So delivery is at least once, and every handler must make a repeat harmless.

The message ID is the tool for that. The mail handler sets the email’s Message-ID header from it, <message-id@domain>, so a mail system that sees a resend can tell it is the same message. A link that has expired by the time its message is delivered is dropped rather than sent, since it would only fail.

When a message keeps failing

After MaxAttempts failures, the relay buries the message: it stops trying and keeps it, with the last error, for someone to look at. It does the same for a topic with no handler, and for a message that was claimed more often than allowed without finishing, because that message may be what crashes the relay.

Two metrics count the outcomes: outbox.delivered and outbox.buried, both tagged with the topic. Alert on outbox.buried.

Topics

Topic	Sent when	Handler
`mail.login_link`	Someone asks for a sign-in link	Emails the link with its one-time code
`mail.email_change`	A user asks to change their email	Emails a confirmation link to the new address

The topic names are constants in internal/user/mail, which also holds the handlers.

Where a broker would go

Nothing in this system needs a message broker. If another system should hear about a change, publish to NATS, Kafka or a webhook from an outbox handler. The message is committed with the change and retried until the publish succeeds, which is the guarantee a broker integration needs.