Ankusa for Elixir

The same pipeline the container runs — a durable local store, verification, dispatch, sinks — as a library in your own supervision tree. If you would rather run Ankusa as a container, or you don't write Elixir, start at the README.

Install

def deps do
  [
    {:ankusa, "~> 0.7"},
    # add these as you scale out:
    {:ankusa_kafka, "~> 0.7"},     # deliver to Kafka
    {:ankusa_rabbitmq, "~> 0.7"}   # deliver to RabbitMQ
  ]
end

ankusa depends on rocksdb, a NIF built from source when you run mix deps.compile: you need cmake >= 3.12, a C++20 compiler, and zstd + OpenSSL development headers (on Alpine, also linux-headers).

Configure a source

config :ankusa,
  autostart: true,
  sources: %{
    "stripe" => [
      verifier: {Ankusa.Verifier.Hmac, scheme: :stripe, secret: System.get_env("STRIPE_WHSEC")},
      sinks: [{Ankusa.Sink.Http, url: "https://example.internal/stripe", secret: System.get_env("SINK_WHSEC")}]
    ]
  }

Stripe now posts to /webhooks/stripe, and every accepted hook is delivered to your sinks at least once.

Try it from a checkout

Requires Elixir 1.20+ / OTP 29 (elixir --version).

1. Install deps and start the server

mix deps.get
iex -S mix               # or: mix run --no-halt

You'll see the durability banner and the listener come up:

[ankusa] starting instance default roles=[:edge, :dispatch, :storage] port=4000 data_dir=./data
[ankusa] store at ./data/default/store. Durable to power loss on THIS host only.
Running Ankusa.Edge.Router with Bandit 1.12.5 at 0.0.0.0:4000 (http)

A zero-config demo source is preconfigured (accepts anything, logs it). Override the port with PORT=4055 iex -S mix or config :ankusa, port: <n>.

2. Ingest a webhook

From another terminal:

curl -XPOST localhost:4000/webhooks/demo -H 'content-type: application/json' \
  -d '{"id":"evt_1","event":"push"}'
# => {"id":"01a0...","status":"accepted"}

The 201 returns only after the payload is committed to the store with one synced batch, that's the core invariant, not a formality. The dispatch pipeline then delivers it, visible in the server log:

[info] hook delivered id=01a0... source=demo attempt=1

Under wal.type: dual it also means the hook was confirmed in the Kafka intake log, and under wal.type: none the same 201 means the sinks confirmed inside the request instead: there is no queue, no dispatch pipeline, and the provider's retry is the retry.

3. Without dedupe, every accepted POST is stored

Without a source dedupe: setting, posting the same body again is a new hook with a new id:

curl -XPOST localhost:4000/webhooks/demo -d '{"id":"evt_1"}'
# => {"id":"01b1...","status":"accepted"}

A provider retry after a lost ack lands the same way, stored and delivered again. Consumers are idempotent receivers, so absorbing the redelivery is their job; see integrations.md. A source with dedupe: collapses hooks sharing a provider event key within the TTL: the repeat gets 201 with the original id and "duplicate": true, and nothing new is stored.

4. Inspect state

curl localhost:4000/health
# => {"status":"ok","instance":"default"}

curl localhost:4002/v1/wal      # admin.enabled: true
# => {"instance":"default","wal":{"next_seq":..,"hooks":..,"deliveries":..,"disk_bytes":..}}

hooks and deliveries are RocksDB key estimates; next_seq is exact. wal is {} when the store cannot be read, and /v1/wal is 409 wal_disabled under wal.type: none.

Durable state on disk — the store, plus the segments the compactor rolls out of it:

find data/default -type f
# data/default/store/...        # RocksDB: hooks, delivery rows, catalogue
# data/default/segments/archive/v1/dt=2026-10-19/<node>/seg-00000000000000000001-...seg
# data/default/segments/archive/v1/dt=2026-10-19/<node>/seg-00000000000000000001-...idx

State lives under ./data/<instance>/: store/ holds the queue, DLQ, quarantine, sources and rate limits; segments/ holds the object store's segment and index objects.

Endpoints

The ingest listener serves the catch URL, plus one read-only endpoint:

Method	Path	Description
POST	(catch URL)	Ingest. Path scheme is set by the configured Ankusa.RouteResolver (default /webhooks/:source_id; TenantPath gives /webhooks/:tenant/:source, see multi-tenancy.md). Raw body kept verbatim; verified inline; the 2xx is returned only after the durable accept — the store commit (wal.type: disk), the store commit plus a confirmed intake-log publish (dual), or every sink's confirm (none): 201 accepted is the only committed response (there is no 200). Without a source dedupe: setting every accepted POST is a new hook with a new id, so a provider retry after a lost ack is stored and delivered again; with it, a repeat inside the TTL gets 201 with the original id and "duplicate": true. 201 accepted (or duplicate) / 202 quarantined / 400 body unreadable / 400 invalid_header / 401 verification failed / 403 route guard (when routes.ip_denied_status is 403) / 404 unknown source / 413 too large / 429 rate_limited or quarantine_rate_limited (with Retry-After) / 503 overloaded, undeliverable or quarantine_full.
GET	/health	Liveness. Always 200 while the listener is up.

The operator API on its own port, /metrics, the dead-letter queue, replay, quarantine, is admin.enabled: true; see the admin API section in configuration.md.

Roles from code

Ankusa.Instance's init/1 starts children conditionally on config.roles. The store and the queue writer, for example:

defp store_children(config, opts) do
  if Enum.any?([:edge, :dispatch, :storage], &Config.role?(config, &1)) or
       match?({Ankusa.SourceStore.Persistent, _}, config.source_store) do
    [{Ankusa.Store, opts}]
  else
    []
  end
end

defp writer_children(%Config{wal: wal}, opts) when wal in [:disk, :dual],
  do: [{Ankusa.Queue.Writer, opts}]

defp writer_children(_config, _opts), do: []

defp claim_check_children(config) do
  if Config.role?(config, :claim_check) do
    # ...
    isolated(config, :claim_check, [bandit_child(Ankusa.ClaimCheck.Router, ...)])
  else
    []
  end
end

The writer is not role-checked itself: it lives in the :edge subtree, which only starts when the edge role is on, and only under wal: :disk or wal: :dual. The edge, dispatch, storage and claim_check children follow the same if Config.role?(config, role) shape.

One Mix release, many deployments: the same compiled artifact runs all-in-one on a laptop or as a node in a fleet, because which children start is a runtime config decision, never a build-time one.

edge, dispatch, and storage share the one local RocksDB store at <data_dir>/<instance>/store, and RocksDB is single-process, so every store-backed role has to run in the same BEAM node: you can't split them across processes or hosts. claim_check is stateless and can run anywhere:

ANKUSA_ROLES=edge,dispatch,storage mix run --no-halt   # the store-backed shape
ANKUSA_ROLES=claim_check mix run --no-halt             # claim-check gateway, any node

Ankusa.Application reads ANKUSA_ROLES (comma-separated) and PORT on top of whatever config.exs sets, see configuration.md#runtime-environment-overrides.

The queue from code

The default wal: :disk commits every accepted hook to Ankusa.Queue; wal: :dual runs the same commit and a confirmed intake-log publish. The batcher does it on the ingest path, and you can read the queue and manage the DLQ directly:

# one entry per hook: the envelope plus the source's sinks at ack time
{:ok, [{:committed, env}]} =
  Ankusa.Queue.enqueue(:default, [%{envelope: env, sinks: [{Ankusa.Sink.Http, url: "..."}]}])

{:ok, hooks} = Ankusa.Queue.hooks(:default, 0, 100)     # seq > 0, ascending
{:ok, stats} = Ankusa.Queue.stats(:default)             # %{next_seq, hooks, deliveries, disk_bytes}
{:ok, %{total: n, entries: entries}} = Ankusa.Queue.dead(:default, source_id: "stripe", limit: 100)

# A durable, paced replay job instead of a one-shot flip:
{:ok, :created, job} = Ankusa.Replay.start(:default, kind: :dlq, source_id: "stripe", rate: 1_000)
{:ok, job} = Ankusa.Replay.get(:default, job.id)        # watch moved/delivered/state
{:ok, job} = Ankusa.Replay.update(:default, job.id, state: :paused)   # or :cancelled / a new :rate

Ankusa.Replay.start/2 creates a job that drips dead rows back to pending at rate items per second — only while dispatch has spare capacity — and the pipeline delivers them asynchronously; a row that fails again is dead-lettered again. The job is durable (its cursor commits with the rows it moved) and resumable, and it pauses itself if its deliveries keep dead-lettering. Under wal.type: none nothing is committed, so there is no queue and no DLQ.

Deploying your own wrapper app

If you embed Ankusa rather than run the published image, deployment shape is a choice the operator makes, so the worked example (examples/rabbitmq-consumer/ingest_app/Dockerfile) shows the pattern rather than prescribing one image for every use case:

• Dev-mode image (what the example uses): elixir:1.20.4-alpine, mix deps.get && mix compile, CMD ["mix", "run", "--no-halt"]. Simpler, faster to build, right-sized for an example or a low-traffic deployment.
• Release image (what a real production deployment should build instead): multi-stage, mix release in the build stage, a slim runtime base in the final stage. Not shipped here: the framework doesn't prescribe a release config because that's genuinely deployment-specific (env-var vs. runtime.exs-based config, which roles per image, etc.).

A build stage needs the RocksDB toolchain from Install — cmake, a C++20 compiler, zstd and OpenSSL headers — the same packages the shipped example installs in its build stage.

A deployable wrapper app (like ingest_app/) is the intended pattern: a tiny Mix project that depends on ankusa (+ whichever adapter packages it needs), reads its own env vars, and calls Ankusa.Config.new/1 + Ankusa.Instance.start_link/1 directly. ankusa core stays a library; the wrapper is where "how do I actually deploy this" config lives.

Stopping the node (SIGTERM) is the wrapper's to make graceful too. :ankusa's own prep_stop/1 drains every instance in its supervision tree (the autostarted default instance, or one you start with Supervisor.start_child(Ankusa.Supervisor, {Ankusa.Instance, config})). An instance your app supervises itself is not reachable from there: call Ankusa.Instance.Drain.run(instance) from your application's prep_stop/1, which answers /ready with 503 draining, waits listener.drain_delay_ms, closes the ingest listener after the requests in flight and waits for the batchers (see Deployment). Set the pod's terminationGracePeriodSeconds to cover it.

If that wrapper depends on ankusa directly and transitively through an adapter package, mark your direct entry override: true: the reason, and the Dockerfile/compose context that goes with it, is in packaging.md#building-an-app-against-the-path-deps.

Where next

• Every config key, struct and YAML: configuration.md
• Embedding Ankusa and Oban in one app: integrations.md
• The local store and object stores: storage.md
• Sinks, retries, dead letters: delivery.md
• Module reference and callback signatures: HexDocs