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

What happens to a delivered hook. The interesting extension point: it lets the
framework be a library inside a Phoenix app, a standalone forwarding gateway,
or the front door to someone else's pipeline — with the same ingest guarantees.

Delivery is at-least-once. Return `:ok` on success; `{:error, reason}` triggers
the source's `Ankusa.RetryPolicy`.

## Error classes

Three shapes of `reason` change what dispatch does with a failure:

  * `{:permanent, term}` — retrying cannot help (a `410 Gone`, a record the
    broker will never take). The row is dead-lettered after this attempt,
    whatever attempts the retry policy had left, and the failure does not
    count towards the sink's circuit breaker.
  * `{:permanent, {:config, term}}` — a permanent failure that is the
    operator's to fix, not the hook's: a credential the destination refuses
    (`AccessDenied`, `PERMISSION_DENIED`), a queue or topic that does not
    exist. It dead-letters like any `{:permanent, _}` and also counts in
    `ankusa_dispatch_config_errors_total{sink}`, the alert that says a source
    needs a config fix and then a DLQ replay.
  * `{:retry_after, ms, term}` — the sink asked for a pause (`Retry-After`, or
    a throttle: AWS `Throttling`, Pub/Sub `RESOURCE_EXHAUSTED`). The next
    attempt is no sooner than `ms` (capped at `dispatch.max_retry_after_ms`,
    default one hour) or the policy's own backoff, whichever is later; the
    policy still decides when to give up.

Every other `reason` is transient and follows the retry policy. Whatever the
class, a delivery that is still failing `max_elapsed_ms` after its first
attempt (default 24 hours, per source: `retry:`) is dead-lettered
(`Ankusa.RetryPolicy`).

Deliveries are **not ordered**: hooks for one sink may be delivered in any
order, and a retry runs after whatever is due before it. A Kafka partition key
or an AMQP routing key only keeps the order hooks were *published* in, so it
does not restore one. A consumer that needs order has to rebuild it from data
it receives (the provider's own event timestamp or sequence number in the
body; `received_at` is on every message, but ties are possible within a
millisecond) and tolerate redelivery.

# `class`

```elixir
@type class() ::
  {:permanent, term()}
  | {:retry_after, pos_integer(), term()}
  | {:transient, term()}
```

What `classify/1` answers.

# `ctx`

```elixir
@type ctx() :: %{
  :instance => atom(),
  :source_id => String.t(),
  :tenant_id => String.t() | nil,
  :attempt => pos_integer(),
  optional(:claim) => Ankusa.ClaimCheck.claim(),
  optional(:replay_id) => String.t(),
  optional(:forward_headers) => :default | [String.t()],
  optional(:batch) => boolean(),
  optional(:max_headers) => pos_integer(),
  optional(:forward_headers_max_bytes) => pos_integer(),
  optional(atom()) => term()
}
```

# `error_class`

```elixir
@type error_class() :: :permanent | :throttled | :transient
```

Why a delivery was dead-lettered, as an operator filters on it
(`error_class` of `GET /v1/dlq` and of a replay job): `:permanent` (the sink
said it will never work, or the hook or its destination is gone), `:throttled`
(the destination asked for a pause, `{:retry_after, ms, reason}`, and the
retry horizon ended it) or `:transient` (anything else that kept failing until
the retry policy gave up).

# `failure`

```elixir
@type failure() ::
  {:permanent, term()} | {:retry_after, pos_integer(), term()} | term()
```

What `c:deliver/3` may return; see "Error classes".

# `batched?`
*optional* 

```elixir
@callback batched?(opts :: keyword()) :: boolean()
```

Does this sink batch deliveries (`Ankusa.Sink.Batcher`)?

Dispatch runs a batching sink's attempts without counting them against
`dispatch.sink_concurrency`, a tenant's `concurrency` or the global
`dispatch.concurrency`: an attempt waits in the batcher for its batch's
response, so those limits cannot count messages and still let a batch fill.
The sink's batcher bounds what it holds (a queue cap, a per-tenant share) and
counts *requests* against `sink_concurrency`; the claimed-row windows
(`dispatch.max_inflight`, `dispatch.tenants`) bound the attempts.

Defaults to `false`. A sink that returns `true` must batch only for a delivery
whose ctx carries `batch: true`, which dispatch sets and `Ankusa.Edge.Publish`
and `Ankusa.Lifecycle.Publisher` do not.

# `deliver`

```elixir
@callback deliver(Ankusa.Envelope.t(), ctx(), opts :: keyword()) ::
  :ok | {:error, failure()}
```

# `describe`
*optional* 

```elixir
@callback describe(
  subject :: %{source_id: String.t(), tenant_id: String.t() | nil},
  opts :: keyword()
) ::
  Ankusa.Sink.Description.t()
```

Where this sink publishes, for the AsyncAPI document (`Ankusa.AsyncApi`).

`subject` names the hook the description is for: `source_id`, and `tenant_id`
(`nil` when the tenant varies per hook, i.e. a resolver that reads it from the
URL). Messaging sinks describe the channel they publish the envelope to;
`Ankusa.Sink.Http` describes its destination with `payload: :body` and says
which options it `computed` per hook, and `Ankusa.Sink.Log` has no channel to
advertise. Only `payload: :message` descriptions reach the AsyncAPI document.

The description must never contain credentials or URL userinfo.

# `durable?`
*optional* 

```elixir
@callback durable?(opts :: keyword()) :: boolean()
```

Does `:ok` from this sink mean the hook is durably accepted by something that
outlives this node — a broker ack, a publisher confirm, an upstream `2xx`?

Defaults to `true`: every shipped sink except `Ankusa.Sink.Log` confirms
durably. `wal.type: none` acks the provider on this promise, so a sink that
cannot make it must say so.

# `inline_max_bytes`
*optional* 

```elixir
@callback inline_max_bytes(opts :: keyword()) :: pos_integer() | nil
```

The largest body this sink sends inline, or `nil` if it never uses the claim
check.

Dispatch checks a body in **once**, before any sink runs, when it is larger
than at least one of its source's sinks' thresholds, and hands the resulting
ref to every sink and every retry in `ctx.claim`. A sink that declares a
threshold must use that ref rather than checking the body in itself.

# `attempt`

```elixir
@spec attempt(
  module(),
  Ankusa.Envelope.t(),
  pos_integer(),
  String.t() | nil,
  (-> result)
) :: result
when result: :ok | {:error, term()} | {:ok | {:error, term()}, term()}
```

Run one delivery attempt of `env` inside an `ankusa.dispatch.attempt` span
(`Ankusa.Trace`): a child of the hook's stored trace context and linked to it,
so every attempt of a hook, however late, is in its trace. `fun` is the whole
attempt, and may return `:ok`, `{:error, reason}`, or either of those paired
with anything (`Ankusa.Dispatch.Pipeline` also reports a claim it checked in).
A failure marks the span with its class (`permanent`, `retry_after`,
`transient`), never the reason. With no OpenTelemetry SDK running this is
`fun.()`.

The attempt's process gets the hook's correlation as `Logger` metadata
(`hook_id`, `source_id`, `tenant_id`, `sink`, `attempt`, `trace_id`, and
`replay_id` for a replayed row), so whatever the sink logs while it delivers
names the hook. Every caller runs it in a process of its own (a task per
attempt).

# `batched?`

```elixir
@spec batched?(module(), keyword()) :: boolean()
```

Resolve `c:batched?/1` for `mod` with `opts`; `false` for a sink that doesn't
implement it, or whose callback raises or answers anything but `true`.

# `classify`

```elixir
@spec classify(term()) :: class()
```

The class of a sink's error `reason` (see "Error classes"): `{:permanent,
term}`, `{:retry_after, ms, term}` with a positive integer `ms`, or
`{:transient, reason}` for anything else.

# `describe`

```elixir
@spec describe(
  module(),
  %{source_id: String.t(), tenant_id: String.t() | nil},
  keyword()
) ::
  Ankusa.Sink.Description.t() | nil
```

Resolve `c:describe/2` for `mod`; `nil` for a sink that doesn't implement it.

# `durable?`

```elixir
@spec durable?(module(), keyword()) :: boolean()
```

Resolve `durable?/1` for `mod` with `opts`; `true` for sinks that don't
implement it — a sink that has not said otherwise is assumed to confirm
durably, which is the safe default for an ack path (`wal.type: none`).

# `error_class`

```elixir
@spec error_class(class()) :: error_class()
```

The `t:error_class/0` of a `classify/1` answer.

# `error_classes`

```elixir
@spec error_classes() :: [error_class()]
```

Every `t:error_class/0`.

# `inline_max_bytes`

```elixir
@spec inline_max_bytes(module(), keyword()) ::
  {:ok, pos_integer() | nil} | {:error, term()}
```

Resolve the inline threshold for `mod` with `opts`: `{:ok, nil}` for sinks
that don't implement `c:inline_max_bytes/1`.

The callback is user code. A raise, exit, throw or a value other than `nil`
or a positive integer is an `{:error, reason}` — never a crash of the caller,
which may be the dispatch pipeline or a request process.

# `safe_deliver`

```elixir
@spec safe_deliver(module(), Ankusa.Envelope.t(), ctx(), keyword()) ::
  :ok | {:error, term()}
```

Call `c:deliver/3`, turning a raise, throw, or exit into `{:error, reason}`.

A sink is user code: it may raise, throw, or exit (a `GenServer.call` into a
dead process). Any of those is a delivery failure, not a caller crash. Any
other return value is `{:error, {:bad_return, value}}`. Every ack path —
`Ankusa.Dispatch.Pipeline`, `Ankusa.Lifecycle.Publisher` and
`Ankusa.Edge.Publish` — delivers through here so they agree on what "the sink
failed" means, and each can match on `:ok | {:error, _}` exhaustively.

# `throttle_after_ms`

```elixir
@spec throttle_after_ms(pos_integer()) :: pos_integer()
```

The pause a sink asks for when its destination throttles it and says nothing
about how long (`{:retry_after, ms, reason}`): full jitter, a uniform draw
from `1..min(30_000, 500 * 2^(attempt - 1))` milliseconds, so every throttled
delivery of a source does not come back in the same instant. `attempt` is
`ctx.attempt`.

---

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