Carbonite v0.2.0

Carbonite.Outbox (Carbonite v0.2.0) View Source

Implements the outbox pattern to process (and evict) recorded transactions in the database, in order of insertion.

Link to this section Summary

Types

prefix()
process_fun()
process_option()
purge_option()

Functions

process(opts \\ [], process_fun)

Builds an Ecto.Multi that can be used to load Carbonite.Transaction records from database (in order of insertion) and process them one-by-one through a function.

purge(opts \\ [])

Builds an Ecto.Multi that can be used to delete Carbonite.Transaction records and their associated Carbonite.Change rows from the database once they have been successfully processed using process/2.

Link to this section Types

Link to this type

prefix()

View Source

Specs

prefix() :: binary() | atom()
Link to this type

process_fun()

View Source

Specs

process_fun() ::
  (repo :: module(), Carbonite.Transaction.t() ->
     {:ok, any()} | {:error, any()})
Link to this type

process_option()

View Source

Specs

process_option() ::
  {:batch_size, non_neg_integer()}
  | {:min_age, non_neg_integer()}
  | {:prefix, prefix()}
Link to this type

purge_option()

View Source

Specs

purge_option() :: {:prefix, prefix()}

Link to this section Functions

Link to this function

process(opts \\ [], process_fun)

View Source

Specs

process([process_option()], process_fun()) :: Ecto.Multi.t()

Builds an Ecto.Multi that can be used to load Carbonite.Transaction records from database (in order of insertion) and process them one-by-one through a function.

Processed items are marked as processed and can be purged.

Options

  • batch_size is the size of records to load in one chunk, defaults to 20
  • min_age is the minimum age of a record, defaults to 300 seconds (see below)
  • prefix is the Carbonite schema, defaults to "carbonite_default"

Long running transactions & insertion order

A warning: As Carbonite.Transaction records are inserted at the beginning of a database transaction, their inserted_at is already a bit in the past when they become visible to other connections, e.g. to your processing job. This means that in case of long running transactions, Carbonite.Transaction records with more recent inserted_at values might be processed before older ones, and hence the eventual total order of inserted_at in the processed records can not be guaranteed. To mitigate this issue, Carbonite will by default exclude records younger than min_age seconds from processing. Tweak this setting if you have even longer transactions.

Link to this function

purge(opts \\ [])

View Source

Specs

purge([purge_option()]) :: Ecto.Multi.t()

Builds an Ecto.Multi that can be used to delete Carbonite.Transaction records and their associated Carbonite.Change rows from the database once they have been successfully processed using process/2.

Options

  • prefix is the Carbonite schema, defaults to "carbonite_default"