Async dispatch pipeline: a scheduler over delivery rows. Every committed
hook has one row per sink of its source (Ankusa.Queue); this process claims
the rows that are due, runs each delivery in a Task.Supervisor task, and
records the outcome as a single store batch. At-least-once: a row is only
deleted after its sink answered :ok, and a restart makes every claimed row
due again.
What wakes it
A commit sends it :wake, so a new hook is claimed at once, not at the next
poll. Otherwise it sleeps until the earliest due row: a retry is a row due at
now + backoff, so waiting costs no slot and no process — a sink that is
down for an hour holds up nothing but its own rows.
Concurrency
Up to dispatch.concurrency deliveries run at a time. A window
(dispatch.max_inflight rows, dispatch.max_inflight_bytes of stored hook)
bounds claimed-but-unfinished work, so a dead sink cannot walk the pipeline
into an OOM. Deliveries are not ordered unless a sink entry opts in: two hooks
for one sink may run in either order, and a retry runs after whatever is due
before it.
An attempt that has not returned after dispatch.attempt_timeout_ms (default
30 s) is killed and counts as a failed attempt, {:attempt_timeout, ms}, so a
hung sink frees its slot; the fallback claim check-in runs inside the same
deadline.
Ordering lanes
A sink entry with ordered: true (Ankusa.Ordering, ADR 0004) delivers the
hooks of one lane ({tenant, source, sink, key}) one at a time, in the order
they were committed on this node (seq). Only a lane's head is a due row or
a claim; the hooks behind it are members (Deliveries.Lanes), in the store
and nowhere else: a waiting lane holds no slot, no window room and no process.
A retry or a breaker park rewrites the head's due row and blocks its lane
alone. Every terminal transition of a head (delivered, dead-lettered,
{:permanent, _}, :sink_gone, an orphaned row) releases its lane in the
batch that writes it: the next member becomes the head, so a poison hook never
holds its key forever. Once per visit of a tenant, before the due fold, a sweep
promotes members that have no head (a writer committed one into an idle lane)
and heals a head whose row is gone; it runs only while the node holds a member
at all (lane_members), so a deployment with no ordered sink pays one counter
read per scan. Only this process moves lane state, except the queue writer's
blind puts of a new member and its marker.
Per-tenant fairness
A scan claims from the tenants that have work due
(Deliveries.active_tenants/3) by deficit round-robin: the page of
dispatch.batch rows is split into equal shares, a tenant with less due than
its share leaves the rest to the others (further passes), and a tenant the page
did not reach goes first in the next scan, so a tenant with a million queued
rows is one share of a page, not the head of the line. Each tenant also has its
own window beside the global one: dispatch.tenants.default and
dispatch.tenants[tenant_id] set max_inflight and max_inflight_bytes of
claimed, unfinished rows and concurrency attempts running at once. A tenant
whose sink is slow fills its own window and nobody else's; a tenant at its
window is skipped by the scan and by the wake-up timer until an outcome frees
room. Slots go round-robin across tenants first, then across each tenant's
keys.
Per-sink isolation
Claimed jobs wait in one queue per sink key {tenant_id, source_id, sink index, sink module}, and a tenant's slots go round-robin across the keys that
have work, so one source's backlog cannot queue ahead of every other source's.
With dispatch.sink_concurrency set (default 8), one key never runs more than
that many attempts at once; the key carries the tenant, so the cap and the
breaker below are per tenant.
Each key has a circuit breaker. dispatch.breaker_failures consecutive
failures (0 disables breakers; a {:permanent, _} error does not count, see
Ankusa.Sink) open it: the key's queued rows and every row of it claimed
while it is open are parked — written back due when the breaker closes,
with their attempt count unchanged — for dispatch.breaker_open_ms, doubling
per consecutive open up to dispatch.breaker_max_open_ms. After that one
attempt runs as a probe (rows claimed meanwhile are parked for
breaker_open_ms): success closes the breaker, failure opens it again for
longer. Breakers live in this process's memory; a restart closes them all.
Parked rows spend no attempts, but not forever: a row's retry horizon
(retry.max_elapsed_ms, default 24 h, from its first attempt) also ends a
park. A parked row comes due no later than its horizon, and one that is parked
again at or past it is dead-lettered as {:retry_horizon, module}.
Batching sinks
A sink that batches (Ankusa.Sink.batched?/1: Sink.SQS, Sink.GooglePubSub)
has its attempts wait in the destination's Ankusa.Sink.Batcher for the
batch's response, so counting them against dispatch.sink_concurrency, a
tenant's concurrency or the global dispatch.concurrency would cap a batch
at that many messages (8, not the 10 of an SQS batch, let alone Pub/Sub's
1 000). Their attempts are not counted: the claimed-row windows
(dispatch.max_inflight, dispatch.tenants) bound how many are waiting, the
batcher bounds its queue (per tenant too) and counts sink_concurrency in
in-flight requests. Each attempt is still its own: it settles its own row
on its own entry's result, is timed out by dispatch.attempt_timeout_ms, and
moves its own key's breaker, so a tenant whose breaker is open never has an
entry in another tenant's batch.
Binding
A row binds to (sink index, module) at ack time. Its opts always come from
the source as it is now, so a config fix applies to the backlog and no fun
or secret is ever persisted. If the source's sinks were reordered the row
falls back to the one sink with its module; if it cannot be bound it is
dead-lettered as {:sink_gone, index, module}, and a deleted source as
{:source_gone, source_id}.
Claim check
A body larger than a sink's Ankusa.Sink.inline_max_bytes/1 is checked in
once per hook, before any of its sinks run: the claims of a scan are
packed per tenant and uploaded together (Ankusa.ClaimCheck.check_in_batch/2)
in a task, and the ref is stored with the hook so every sink and every retry
reuses it. If a pack fails, its jobs still run: the first attempt checks the
body in on its own.
start_link/1 opts: :instance and :config.
Summary
Functions
Returns a specification to start this module under a supervisor.
Dead-letter every pending row of source_id now, as {:source_gone, source_id} (what dispatch does to a deleted source's rows), and return how
many; for deleting a source that still has a backlog. A row whose hook is
gone is dropped and not counted. Rows claimed by a running attempt are left
to settle: one that comes back pending after the source is deleted is
dead-lettered when it is next claimed. Runs inside this process, the only
one that moves pending rows, so no claim can race it. {:error, :unavailable} while the pipeline is down or busy past the call's deadline.
The spare capacity a replay job may use right now: the lag of the oldest due
row (lag_ms, 0 when nothing is due) and whether the in-flight window is
full. {:error, :unavailable} while the pipeline is down or the store
cannot answer.
Take dispatch.concurrency, dispatch.sink_concurrency and dispatch.tenants
from the instance's current config (Ankusa.Reload), without a restart: the
tenants' limits are rebuilt, and rows already claimed or running are never
dropped (a lowered limit only stops the next start until the count falls below
it). :ok also when the pipeline is down: it reads the config when it starts.
Block until nothing is claimed, running, or due; returns the rows settled (delivered or dead-lettered) during the call. A row waiting out a backoff is not due, so this does not wait for it.
Check the config.dispatch keys this module reads; raises ArgumentError
naming the key.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec dead_letter_source(atom(), String.t()) :: {:ok, non_neg_integer()} | {:error, term()}
Dead-letter every pending row of source_id now, as {:source_gone, source_id} (what dispatch does to a deleted source's rows), and return how
many; for deleting a source that still has a backlog. A row whose hook is
gone is dropped and not counted. Rows claimed by a running attempt are left
to settle: one that comes back pending after the source is deleted is
dead-lettered when it is next claimed. Runs inside this process, the only
one that moves pending rows, so no claim can race it. {:error, :unavailable} while the pipeline is down or busy past the call's deadline.
@spec pressure(atom()) :: {:ok, %{lag_ms: non_neg_integer(), window_full: boolean()}} | {:error, :unavailable}
The spare capacity a replay job may use right now: the lag of the oldest due
row (lag_ms, 0 when nothing is due) and whether the in-flight window is
full. {:error, :unavailable} while the pipeline is down or the store
cannot answer.
@spec reconfigure(atom()) :: :ok
Take dispatch.concurrency, dispatch.sink_concurrency and dispatch.tenants
from the instance's current config (Ankusa.Reload), without a restart: the
tenants' limits are rebuilt, and rows already claimed or running are never
dropped (a lowered limit only stops the next start until the count falls below
it). :ok also when the pipeline is down: it reads the config when it starts.
@spec tick(atom()) :: {:ok, non_neg_integer()}
Block until nothing is claimed, running, or due; returns the rows settled (delivered or dead-lettered) during the call. A row waiting out a backoff is not due, so this does not wait for it.
@spec validate_config!(Ankusa.Config.t()) :: :ok
Check the config.dispatch keys this module reads; raises ArgumentError
naming the key.