# `Ankusa.Lifecycle.Publisher`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/lifecycle/publisher.ex#L1)

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`).

# `job`

```elixir
@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
}
```

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `replay`

```elixir
@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_link`

Start the publisher for `opts[:instance]`.

# `wake`

```elixir
@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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*