The behaviour between Outboxer and the database that holds the outbox.
Two adapters ship with the library, both built on Ecto:
Outboxer.Storage.Postgres- multi-node safe. Each delivery batch runs in a short transaction that holds a transaction-scoped advisory lock on the partition.Outboxer.Storage.SQLite- single node. The one worker per partition is the only owner; delivery happens outside any transaction because SQLite has a single writer.
The adapter is chosen from the repo's Ecto adapter, or set explicitly with
the :storage option of Outboxer.
Writing another adapter
The contract an adapter has to honour:
insert/2must take part in the caller's database transaction. That is the whole point of the pattern: the message commits or rolls back with the business write. A store that cannot do this (Redis, for example) can still implement the behaviour, but what you get is a best-effort queue, not a transactional outbox.with_partition/4must guarantee that at most one process in the whole cluster runs the function for a given outbox partition at a time, and must release that ownership if the process dies.fetch/5must return ready messages in id order and must leave out every group that has a failed message waiting for its retry.
Summary
Callbacks
Records successful delivery: deletes the rows or marks them delivered.
Records a failed attempt and schedules the retry.
Returns up to limit deliverable messages of a partition, oldest first.
Inserts rows in the caller's transaction and returns their envelopes.
Moves messages to the dead state after their last allowed attempt.
Serialises concurrent publishers of the same group until their transaction ends. A no-op where the database already has a single writer.
Tells other nodes that a partition has new messages. Optional.
Deletes up to limit delivered messages older than cutoff. Returns how many.
Runs fun as the exclusive owner of a partition, or returns :busy.
Types
@type conf() :: %{ :repo => module(), :outboxes => %{ optional(atom()) => %{:table => String.t(), optional(atom()) => term()} }, :prefix => String.t() | nil, :name => atom(), optional(atom()) => term() }
The instance configuration. An adapter can rely on these fields:
:repo- theEcto.Repomodule:prefix- the schema of the outbox tables, ornil:name- the Outboxer instance name:outboxes- a map of outbox name to its configuration;.tableis the table that outbox lives in
Message ids are unique within a table only, which is why every callback that takes ids also takes the outbox they belong to.
@type outbox() :: atom()
Callbacks
@callback ack( conf(), outbox(), ids :: [pos_integer()], retention :: :delete | {:keep, pos_integer()}, now :: DateTime.t() ) :: :ok
Records successful delivery: deletes the rows or marks them delivered.
@callback fail( conf(), outbox(), ids :: [pos_integer()], retry_at :: DateTime.t(), error :: String.t() ) :: :ok
Records a failed attempt and schedules the retry.
@callback fetch( conf(), outbox(), partition :: non_neg_integer(), limit :: pos_integer(), now :: DateTime.t() ) :: [Outboxer.Envelope.t()]
Returns up to limit deliverable messages of a partition, oldest first.
@callback insert(conf(), [row()]) :: [Outboxer.Envelope.t()]
Inserts rows in the caller's transaction and returns their envelopes.
@callback kill( conf(), outbox(), ids :: [pos_integer()], error :: String.t(), now :: DateTime.t() ) :: :ok
Moves messages to the dead state after their last allowed attempt.
Serialises concurrent publishers of the same group until their transaction ends. A no-op where the database already has a single writer.
@callback notify(conf(), outbox(), partition :: non_neg_integer()) :: :ok
Tells other nodes that a partition has new messages. Optional.
@callback prune(conf(), outbox(), cutoff :: DateTime.t(), limit :: pos_integer()) :: non_neg_integer()
Deletes up to limit delivered messages older than cutoff. Returns how many.
@callback with_partition(conf(), outbox(), partition :: non_neg_integer(), (-> result)) :: {:ok, result} | :busy when result: term()
Runs fun as the exclusive owner of a partition, or returns :busy.