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

Opt-in ordering lanes (ADR 0004): what a source's sink entry may say about
order, and how a hook's lane is resolved at ingest.

    sinks: [{Ankusa.Sink.Http, url: "...", ordered: true, ordering_key: {:header, "x-entity-id"}}]

Two options on a sink entry, read by core, not by the sink module:

  * `ordered: true` (default `false`): the sink's hooks of one lane are
    delivered one at a time, each after the previous returned `:ok`, in the
    order they were committed on this node. A retry blocks its lane and
    nothing else.
  * `ordering_key`: `{:header, name}` (a request header, case-insensitive),
    `{:json, "data.object.customer"}` (a dot path into the JSON body, as
    `Ankusa.Dedupe` reads one) or `{mod, fun, args}` (called as
    `apply(mod, fun, [envelope | args])`, returning a binary). Without a key
    every hook of the sink is in one lane. `ordering_key` without `ordered: true`
    keeps its meaning on `Ankusa.Sink.GooglePubSub` (a string or a 1-arity fun,
    handed to the sink) and is an error on every other sink.

A lane is `{tenant, source, sink index, key}`. It is resolved **at ingest**, from
the complete header list and the body, because the lane a row joins is fixed
before it is committed; every other path that makes delivery rows (an archive
replay, a released quarantine hook) resolves it from the stored envelope by the
same function, so a `{:header, name}` key must survive the source's
`store_headers` policy (`check/2`).

A key longer than 128 bytes (the SQS group id limit) becomes `"sha256:" <> hex`.
A missing key (no such header, a body that is not JSON, over 1 MiB, a missing or
non-string path value, an MFA that raised or returned a non-binary) puts the hook
in the sink's shared keyless lane, counted in
`ankusa_dispatch_lane_key_missing_total{sink}`: an ordered sink never silently
falls back to unordered.

# `key_spec`

```elixir
@type key_spec() ::
  {:header, String.t()} | {:json, [String.t()]} | {module(), atom(), list()}
```

How one sink's lane key is read.

# `lane`

```elixir
@type lane() :: {&lt;&lt;_::128&gt;&gt;, binary()}
```

One sink's lane of a hook: the lane id and the key stamped into the row.

# `sink_ordering`

```elixir
@type sink_ordering() :: nil | {:ordered, key_spec() | nil}
```

One sink's ordering: `nil` (unordered) or `{:ordered, key spec or nil}`.

# `check`

```elixir
@spec check(Ankusa.Config.t(), Ankusa.Source.t()) :: :ok | {:error, String.t()}
```

The checks a source needs the instance's config for: lanes need the
store-backed queue (`wal.type: none` delivers inside the request, with no lane),
and a `{:header, name}` key must survive the source's stored-header policy,
because a replay resolves the key from the stored hook.

# `lane_id`

```elixir
@spec lane_id(non_neg_integer(), String.t(), binary() | nil) :: &lt;&lt;_::128&gt;&gt;
```

The 16-byte lane id of `key` (`nil`: the keyless lane) in sink `sink` of
`source_id`: hashed so the index key has a fixed width and no delimiter can be
forged. The tenant is not in it: it is the ordinal in front of it in every lane
key. A collision would only merge two lanes (more ordering, never less).

# `lanes`

```elixir
@spec lanes(atom(), Ankusa.Source.t(), Ankusa.Envelope.t()) :: [lane() | nil] | nil
```

The lanes of `env` for `source` alone (see `resolve/3`).

# `normalize_sinks!`

```elixir
@spec normalize_sinks!(String.t(), [{module(), keyword()}]) ::
  {[{module(), keyword()}], [sink_ordering()] | nil}
```

Reads `ordered` and `ordering_key` off every sink entry of `source_id`, and
returns the entries as the pipeline hands them to a sink (both options gone,
except the legacy `ordering_key` of an unordered `Sink.GooglePubSub`) with the
per-sink ordering, or `nil` when no sink is ordered. Raises `ArgumentError`
naming the source for every form ADR 0004 section 1 rejects.

# `resolve`

```elixir
@spec resolve(atom(), Ankusa.Source.t(), Ankusa.Envelope.t()) ::
  {String.t() | nil, [lane() | nil] | nil}
```

The provider event key and the lanes of `env` for `source`, resolved from the
complete header list and the body: `{dedupe_key, lanes}`. `lanes` is `nil` for
a source with no ordered sink, else a list aligned with the source's sinks, each
`nil` (unordered) or `{lane, key}`. One JSON decode serves a `dedupe: json` and
an `ordering_key: {:json, _}` on the same source.

# `sink_key`

```elixir
@spec sink_key(binary(), String.t() | nil, String.t()) :: String.t()
```

The key a sink receives as `ctx.ordering_key` for a row stamped with `key`
(`""`: the keyless lane): the key itself, or `"tenant/source"`, the default
group of the queue adapters, for the keyless lane.

# `validate_config!`

```elixir
@spec validate_config!(Ankusa.Config.t()) :: :ok
```

Boot check: `check/2` for every statically configured source. Raises `ArgumentError`.

---

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