A transactional outbox for Ecto.

You write a message to the outbox table in the same database transaction as your business data. Outboxer then relays it to Kafka, SQS, Oban or your own sink, in order, retrying until it gets through. The message exists if and only if the transaction committed, which is the guarantee that "write to the database, then publish to the broker" cannot give you.

Getting started

1. Create the table (see Outboxer.Migration):

   def up, do: Outboxer.Migration.up()
   def down, do: Outboxer.Migration.down()

2. Define a message (see Outboxer.Message):

   defmodule MyApp.Events.OrderPlaced do
     use Outboxer.Message, outbox: :orders
     defstruct [:order_id, :total]
   
     @impl true
     def group_key(%__MODULE__{order_id: id}), do: "order:#{id}"
   end

3. Add Outboxer to your supervision tree, after the repo. Outboxer has no application of its own; this child is the whole runtime:

   children = [
     MyApp.Repo,
     {Outboxer,
      repo: MyApp.Repo,
      outboxes: [
        orders: [relay: {Outboxer.Relay.Kafka, client: :my_brod_client, topic: "orders"}]
      ]}
   ]

4. Publish inside your transaction:

   MyApp.Repo.transaction(fn ->
     order = MyApp.Repo.insert!(changeset)
     {:ok, _envelope} = Outboxer.publish(%OrderPlaced{order_id: order.id, total: order.total})
     order
   end)

Options

• :outboxes (keyword/0) - The outboxes, as name: options. See "Outbox options" below. The default value is [].

• :name (atom/0) - Name of this instance. Only needed when you run more than one. The default value is Outboxer.

• :repo (atom/0) - Required. The Ecto.Repo that holds the outbox table. It must be the repo your business writes use.

• :table (String.t/0) - Name of the table used by every outbox that does not set its own :table. Must match what was given to Outboxer.Migration. The default value is "outboxer_messages".

• :prefix - Postgres schema of the outbox tables. The default value is nil.

• :storage (atom/0) - An Outboxer.Storage module. Detected from the repo's adapter when omitted.

• :notify (boolean/0) - Postgres only. Wake workers through LISTEN/NOTIFY when a publishing transaction commits, instead of waiting for the next poll. Uses one extra database connection. The default value is false.

• :relay (boolean/0) - Whether this node delivers messages. false makes the instance publish-only: nothing is started, no relay is initialised (so the outboxes need no :relay and the node needs no broker dependency), and publish/2 and the admin functions still work. Use it on web nodes when a separate node, started with the same outboxes and relay: true, does the delivering. The default value is true.

• :testing - For your test environment. :manual starts no workers: messages are stored but only delivered when you call Outboxer.Testing.drain/2. The default value is :disabled.

• :require_transaction (boolean/0) - Raise when Outboxer.publish/2 is called outside a repo transaction. The default value is false.

• :prune_interval (pos_integer/0) - Milliseconds between pruning passes, for outboxes using retention: {:keep, seconds}. The default value is 60000.

Outbox options

Each entry under :outboxes accepts:

• :relay - The relay that delivers this outbox's messages: a module implementing Outboxer.Relay, or {module, opts}. Required, except on an instance started with relay: false, where it is ignored.

• :table (String.t/0) - Table holding this outbox's messages, for example "order_outbox". Defaults to the instance's :table. Create it with Outboxer.Migration.up(table: "order_outbox").

• :ordering - :group keeps order between messages that share a group key (messages without one are unordered). :total keeps one global order for the whole outbox (forces a single partition; one failing message blocks everything behind it). :none ignores group keys entirely. The default value is :group.

• :partitions (pos_integer/0) - How many partitions, and so how many concurrent delivery workers, the outbox has. A group always lands in the same partition. Do not change this while the outbox holds undelivered messages. The default value is 1.

• :batch_size (pos_integer/0) - Maximum number of messages fetched and handed to the relay at once. The default value is 100.

• :poll_interval (pos_integer/0) - Milliseconds an idle worker waits before looking for new messages. The default value is 1000.

• :max_attempts - :infinity retries a failing message forever, which blocks its group and never breaks order. A number moves the message to the dead state after that many failed attempts and lets the group continue. The default value is :infinity.

• :backoff - Delay before a failed message is retried. Either [min: ms, max: ms] for exponential backoff with jitter, or a function taking the number of failed attempts and returning milliseconds. The default value is [min: 1000, max: 60000].

• :retention - :delete removes a message as soon as it is delivered (smallest table, recommended). {:keep, seconds} marks it delivered and prunes it after that many seconds. The default value is :delete.

• :serialize_writes (boolean/0) - Postgres only. Take a transaction-scoped advisory lock on the group key when publishing, so that concurrent transactions writing to the same group commit in id order. Only needed when nothing else (such as a row lock on the aggregate) already serialises those writers. The default value is false.

What gets started

Only what the configuration needs:

Outboxer (Supervisor)
├── Registry                       worker lookup
├── Outboxer.Notifier              only with notify: true
├── one supervisor per outbox
│   ├── the relay's own children   e.g. a brod client, if the relay starts one
│   └── one worker per partition
└── Outboxer.Pruner                only if an outbox uses retention: {:keep, _}

With relay: false or testing: :manual none of these are started and publish/2 still works.

Publishing and relaying on different nodes

Start the same outboxes everywhere and switch delivery off where it is not wanted, for example on web pods:

{Outboxer,
 repo: MyApp.Repo,
 relay: System.get_env("OUTBOXER_RELAY", "true") == "true",
 outboxes: [...]}

A relay: false instance only publishes. It ignores the :relay of its outboxes, so it needs neither broker settings nor the broker's dependency. See the operations guide for what the two sides must agree on.

One table per outbox

Every outbox can live in its own table:

outboxes: [
  orders: [table: "order_outbox", relay: ...],
  invoices: [table: "invoice_outbox", relay: ...]
]

Create each one with Outboxer.Migration.up(table: "order_outbox"). Outboxes without :table share the instance's table. Message ids are unique within a table, not across tables.

Several instances

All functions take a :name option, defaulting to Outboxer. Run a second instance (for another repo, say) by giving it a name:

{Outboxer, name: MyApp.BillingOutbox, repo: MyApp.BillingRepo, outboxes: [...]}

Outboxer.publish(message, name: MyApp.BillingOutbox)

Telemetry