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

Operator API for replay jobs (`Ankusa.Dispatch.Replayer`).

A replay re-sends dead delivery rows (`kind: :dlq`) or archived hooks over a
time window (`kind: :archive`) through the normal dispatch pipeline, keeping
each hook's original `id` and `dedupe_key` and stamping the delivery with the
job's `replay_id`. A third kind, `kind: :quarantine`, re-verifies hooks held
in the quarantine pen (`Ankusa.Edge.Quarantine`) against the source's
current verifier and commits the ones that now pass, as new hooks with their
original `id`; the rest stay in the pen. Every kind only ever uses dispatch
capacity live traffic leaves free: rows are dripped in at `rate` items per
second, and only while the Pipeline's oldest-due lag is at most `max_lag_ms`
and its in-flight window is not full.

A fourth kind, `kind: :lifecycle`, re-emits lifecycle events
(`Ankusa.Lifecycle`) from the audit trail over a `since`/`until` window of
the change's time, optionally for some `tenants` (route events belong to
`"default"`). It re-delivers to the node's lifecycle sinks through the
lifecycle publisher, not through dispatch: each event keeps the `id` it was
first emitted with (the id is derived from the audit row) and carries the
job's id as `replayid`, so a consumer that dedupes on `id` drops what it
already has. `rate` is rows per second; `max_lag_ms` is not used.

Jobs are durable and resumable: every job is a store record, its cursor is
committed in the same batch as the rows it moved, and a restart resumes from
the last committed page. A job that keeps dead-lettering its deliveries
pauses itself (auto-pause) instead of feeding a sink that is still down.

## Runbook for a Salesforce-sized incident

  1. One job per node — a fleet replay is one `POST /v1/replays` per node, with
     the same `job_id`: `docker-entrypoint replay --nodes … --filter …` (the
     server image) does that and aggregates the status.
  2. Start low: `rate: 100` or so, and watch the node's metrics.
  3. PATCH `rate` upward while live latency stays flat. The replay takes only
     spare capacity, so it is always safe to leave running; the only knob is
     how long it takes.
  4. A `kind: :dlq` job only touches rows dead-lettered at or before its own
     creation, so rows that die again during the replay are never picked up a
     second time. To re-send those, start another job.
  5. A `kind: :archive` job covers only what the compactor has already
     archived: `to` must be at least `storage.roll_ms + storage.interval_ms`
     in the past.
  6. A `kind: :quarantine` job, like `:dlq`, only touches entries quarantined
     at or before its own creation. Fix the source's secret first (or add the
     new one: `secret: [new, old]`), then start the job; an entry that still
     fails verification stays in the pen and counts as `skipped`.
  7. A `kind: :lifecycle` job covers the events the trail held when it was
     created, in `since`/`until`. It is the way back for a lifecycle sink
     that was down for longer than `dispatch.retry` waits, or that lost
     events: start it at the time the outage began.

# `get`

```elixir
@spec get(atom(), String.t(), [String.t()] | nil) ::
  {:ok, map()} | {:error, :not_found | :store_unavailable}
```

One job by id; a job outside `tenants` (see `list/2`) is `:not_found`.

# `list`

```elixir
@spec list(atom(), [String.t()] | nil) ::
  {:ok, [map()]} | {:error, :store_unavailable}
```

Every job, newest first. `tenants` limits the list to the jobs scoped to
tenants inside that list (`nil` is every job, including unscoped ones).

# `live_floor`

```elixir
@spec live_floor(atom(), :quarantine | :archive, :since | :from) ::
  {:ok, non_neg_integer() | :infinity} | {:error, :store_unavailable}
```

The earliest `key` (`:since` of a quarantine job, `:from` of an archive job)
among the live (`running` or `paused`) jobs of `kind`: what retention must not
delete past, because a job still reads from there. A job without the key needs
everything (`0`). `:infinity` when this node runs no replayer, so no job can
exist; `{:error, :store_unavailable}` when the replayer cannot say, and
retention then deletes nothing.

# `start`

```elixir
@spec start(atom(), map() | keyword(), keyword()) ::
  {:ok, :created | :existing, map()}
  | {:error,
     {:invalid, String.t()}
     | :too_many_replays
     | {:role_not_enabled, atom()}
     | :store_unavailable}
```

Create a replay job. `spec` (map or keyword) takes the admin API's keys as
atoms: for `kind: :dlq` and `kind: :quarantine` the optional `:source_id`,
`:id`, `:since`, `:until`; for `kind: :dlq` also `:sink` (the sink's index in
the source) and `:error_class` (`t:Ankusa.Sink.error_class/0`: only rows
dead-lettered as that class are revived); for `kind: :archive` the required
`:from`, `:to`, and the optional `:source_id` and `:sinks`; for
`kind: :lifecycle` the optional `:since` and `:until`; plus `:rate` and
`:max_lag_ms`. Every kind also takes `:tenants`, a non-empty list of tenant
ids: the job only touches hooks (a lifecycle job, events) owned by those
tenants (the admin API sets it from the caller's `tenant:` scopes), and
`:job_id`, a lowercase UUID that names the job: starting again with the same
`:job_id`, kind and filter answers `:existing` whatever the job's state, which
is how one job is fanned out to every node of a fleet and started again after
a failure (`docker-entrypoint replay`).

Returns `{:ok, :created, job}`, or `{:ok, :existing, job}` when a `running`
or `paused` job with the same kind and filter already exists (a proxy retry
is idempotent). `{:error, {:invalid, field}}` names a bad field (`"job_id"`
also when that id belongs to a job with another kind or filter),
`{:error, :too_many_replays}` the 16-job cap, and
`{:error, {:role_not_enabled, :edge}}` an archive or quarantine job on a
node without a queue writer (`{:role_not_enabled, :lifecycle}`: a lifecycle
job on a node with no lifecycle sinks).

# `to_json`

```elixir
@spec to_json(map()) :: map()
```

The admin API's Replay object for a job map (atoms to strings).

# `update`

```elixir
@spec update(atom(), String.t(), map() | keyword(), keyword()) ::
  {:ok, map()}
  | {:error,
     :not_found | :finished | {:invalid, String.t()} | :store_unavailable}
```

Patch a job: `:state` (`:running | :paused | :cancelled`), `:rate` and/or
`:max_lag_ms`. `{:error, :finished}` for a `done`, `cancelled` or `failed`
job; a job outside the `tenants:` option (see `list/2`) is `:not_found`. Resuming clears
`error` and resets the auto-pause window.

---

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