Ankusa.Replay (ankusa v0.7.0)

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

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.
Start low: rate: 100 or so, and watch the node's metrics.
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.
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.
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.
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.
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.

Summary

Functions

get(instance, id, tenants \\ nil)

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

list(instance, tenants \\ nil)

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(instance, kind, key)

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(instance, spec, opts \\ [])

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

to_json(job)

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

update(instance, id, patch, opts \\ [])

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.

Functions

get(instance, id, tenants \\ nil)

@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(instance, tenants \\ nil)

@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(instance, kind, key)

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

start(instance, spec, opts \\ [])

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

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(job)

@spec to_json(map()) :: map()

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

update(instance, id, patch, opts \\ [])

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