Delivers lifecycle events (Ankusa.Lifecycle) to config.lifecycle.sinks:
the reader of the outbox.
One supervised process per instance. There is no queue to lose: the events
are the audit rows (Ankusa.Audit) flagged publish: true, and this process
keeps one number, the seq of the last row every sink has dealt with, in
m:lifecycle_cursor (written sync: true on every advance). Booted for the
first time on a store, it starts at the end of the trail.
It wakes when Ankusa.Audit commits a flagged row (and once a second, as a
safety net), reads the trail from the cursor with Ankusa.Audit.page/3, and
takes the next flagged row. Each sink gets that row's event as its own job, so
a sink that is down never keeps another from receiving it; but the cursor
only moves once every sink confirmed the row or gave up on it, so rows are
delivered in seq order and one failing sink holds the events behind that row
back from all of them. A failed job is retried after the delay
config.dispatch.retry gives (an Ankusa.RetryPolicy), until the policy gives
up or the job is retry.max_elapsed_ms (default 24 hours) past its first
attempt: the instance's dispatch.retry, not a source's, so that horizon is
how long one failing sink can hold the cursor. Then
the event is dropped for that sink (and can be replayed). An attempt that
has not returned after config.dispatch.attempt_timeout_ms is killed and
counts as a failed attempt.
A node killed after a sink confirmed a row and before the cursor moved
delivers the row again; the event id is the same, so a consumer dedupes.
replay/4 is the other way in: a replay job (Ankusa.Replay, kind: :lifecycle) hands a page of rows to re-emit, tagged with its id. They share
the concurrency (at most 8 attempts run at a time) but not the
cursor, are tried 3 times, and queue behind the live events. At most
1000 replayed deliveries are pending: past that replay/4 says
{:error, :busy} and the job asks again.
Every drop is logged and emitted as [:ankusa, :lifecycle, :dropped] with a
:reason of :gave_up (the sink's retries ran out) or :unreadable (a row
or its entity could not be read, and the cursor moved past it); a confirmed
delivery is [:ankusa, :lifecycle, :delivered].
Sinks are called without a ctx.claim, so a sink with an inline threshold
checks an oversized body in itself (Ankusa.Sink.Message.encode/3).
Summary
Functions
Returns a specification to start this module under a supervisor.
Re-emit rows ({audit_row, entity} pairs) for the replay job replay_id.
Start the publisher for opts[:instance].
Tell the publisher the trail has a new row to look at. :ok whether or not it
runs: the row is in the store either way.
Types
@type job() :: %{ env: Ankusa.Envelope.t(), type: String.t(), subject: String.t(), sink: {module(), keyword()}, attempt: pos_integer(), first_attempt_at: integer(), row: pos_integer(), batch: reference() | nil, replay: String.t() | nil }
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec replay(atom(), String.t(), [{Ankusa.Audit.row(), map()}], pid()) :: :ok | {:error, :busy | :not_running | :no_audit_key}
Re-emit rows ({audit_row, entity} pairs) for the replay job replay_id.
Returns once they are queued. When every delivery has been confirmed or
given up on, to receives {:lifecycle_replayed, replay_id, %{delivered: n, dead: m}} (counts of sink deliveries, one per row and sink).
{:error, :busy} when the replay backlog is full.
Start the publisher for opts[:instance].
@spec wake(atom()) :: :ok
Tell the publisher the trail has a new row to look at. :ok whether or not it
runs: the row is in the store either way.