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(defaultfalse): 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, asAnkusa.Dedupereads one) or{mod, fun, args}(called asapply(mod, fun, [envelope | args]), returning a binary). Without a key every hook of the sink is in one lane.ordering_keywithoutordered: truekeeps its meaning onAnkusa.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.
Summary
Types
How one sink's lane key is read.
One sink's lane of a hook: the lane id and the key stamped into the row.
One sink's ordering: nil (unordered) or {:ordered, key spec or nil}.
Functions
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.
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).
The lanes of env for source alone (see resolve/3).
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.
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.
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.
Boot check: check/2 for every statically configured source. Raises ArgumentError.
Types
Functions
@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.
@spec lane_id(non_neg_integer(), String.t(), binary() | nil) :: <<_::128>>
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).
@spec lanes(atom(), Ankusa.Source.t(), Ankusa.Envelope.t()) :: [lane() | nil] | nil
The lanes of env for source alone (see resolve/3).
@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.
@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.
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.
@spec validate_config!(Ankusa.Config.t()) :: :ok
Boot check: check/2 for every statically configured source. Raises ArgumentError.