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

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

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

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

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

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

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

@spec error_class(class()) :: error_class()

The error_class/0 of a classify/1 answer.

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

Every error_class/0.

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

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

Call 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.

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