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

The batching worker behind `Ankusa.Sink.SQS` (`SendMessageBatch`, 10 entries)
and `Ankusa.Sink.GooglePubSub` (`publish`, 1 000 messages): many deliveries
to one destination, one HTTP request.

A sink that batches (`c:Ankusa.Sink.batched?/1`) encodes its message exactly
as it did for a single request, then hands it to `deliver/4`, which is a
`GenServer.call` to the destination's batcher and returns when *that message's*
result is known: the batch's response, mapped back to the entry. The caller is
the dispatch attempt, so a delivery is answered only after its batch's
response, its row stays claimed until then, and a batcher that dies (or an
attempt that is killed) loses nothing: the call exits, the attempt fails, the
row is retried, and the idempotency key travels as it always did. A delivery
that reaches a destination whose batcher has just died (or stopped idle) starts
its replacement: it waits for the dead batcher's registered name to be reaped
(at most `250 ms`) instead of failing on `:noproc`.

## One batcher per destination

`Ankusa.Instance` starts a `DynamicSupervisor` for them; the first delivery
to a destination starts its batcher. A destination is the codec (the sink's
`Ankusa.Sink.Batcher` implementation) plus a digest of everything that shapes
the request (`c:batch_identity/1`: queue or topic, endpoint, region,
credentials, transport options and the `batch_*` tuning), so two sources that
name different credentials never share a request, and no secret is part of a
registered name. A batcher with nothing queued or in flight for a minute stops;
the next delivery starts a new one.

## When a batch goes out

A batch is sent when the queue holds `max_entries` or `max_bytes`, or
`linger_ms` after the queue became non-empty (or after the last batch left),
whichever is first. At most `max_inflight` requests are in flight per
destination: the instance's `dispatch.sink_concurrency` (default 8) counts
*requests* for a batching sink, not messages (see `Ankusa.Dispatch.Pipeline`).
Entries are taken round-robin across tenants, so one tenant's backlog is
interleaved with, not ahead of, another's.

Entries that share a `group` (a FIFO queue's `MessageGroupId`, a Pub/Sub
ordering key) stay in the order they were admitted, and a group has at most
one request in flight: the next entry of a group waits for the previous
request's response. That is what keeps "the order Ankusa sent them" true; a
retry after a per-entry failure is, as it always was, a later send.

## Bounded (ground rule 3)

The queue holds at most `queue_max` entries, `queue_max_bytes` bytes, and
`tenant_max` entries per tenant (defaults 4 096, 128 MiB, 1 024; the instance's
dispatch windows are the same size, so these are the backstop, not the
limit). A delivery that does not fit is answered at once with
`{:error, {:retry_after, ms, {:batch_queue_full, :entries | :bytes | :tenant}}}`
(`ms` is full jitter) and counted in `ankusa_sink_batch_refused_total`; nothing
is queued. An entry whose caller died (the attempt timed out) is dropped when
the queue is next swept, which every flush does; one already in a request
cannot be recalled.

## Telemetry

  * `[:ankusa, :sink_batch, :flush]` — measurements `entries`, `bytes`,
    `requests` (HTTP requests the batch took), `bisections` (the extra
    requests spent isolating a refused message), `duration_ms`; metadata
    `instance`, `sink`.
  * `[:ankusa, :sink_batch, :refused]` — `instance`, `sink`, `reason`.
  * `[:ankusa, :sink_batch, :state]` — gauge: `batchers`, `queued`,
    `queued_bytes`, from `stats/1`.

## Writing a codec

A codec is a module implementing this behaviour; `send_batch/2` runs in a task
under the batcher's supervisor, never in the batcher itself.

# `config`

```elixir
@type config() :: %{
  linger_ms: pos_integer(),
  max_entries: pos_integer(),
  max_bytes: pos_integer(),
  queue_max: pos_integer(),
  queue_max_bytes: pos_integer(),
  tenant_max: pos_integer(),
  max_inflight: pos_integer(),
  task_timeout_ms: pos_integer()
}
```

# `entry`

```elixir
@type entry() :: %{
  tenant: String.t(),
  bytes: non_neg_integer(),
  group: String.t() | nil,
  attempt: pos_integer(),
  item: term()
}
```

What a sink hands `deliver/4`.

# `result`

```elixir
@type result() :: :ok | {:error, term()}
```

# `batch_config`

```elixir
@callback batch_config(opts :: keyword(), instance :: atom()) :: config()
```

The batching limits for `opts` (`config/3` reads the `batch_*` options).

# `batch_identity`

```elixir
@callback batch_identity(opts :: keyword()) :: term()
```

Everything in `opts` that shapes a request: two deliveries with equal identities
may share one. Include credentials and transport options; the batcher only
ever uses a digest of it.

# `send_batch`

```elixir
@callback send_batch(items :: [term()], opts :: keyword()) ::
  {[result()], %{requests: non_neg_integer(), bisections: non_neg_integer()}}
```

Send `items` (one request, or several when a refused message has to be isolated)
and answer each, in order. The second element counts `requests` and `bisections`.

# `sink`

```elixir
@callback sink() :: module()
```

The sink module this codec batches for; the `sink` label of its telemetry.

# `available?`

```elixir
@spec available?(term()) :: boolean()
```

Is `instance`'s batcher supervisor running? When it is not (a sink called
outside an instance), a batching sink sends each message in a request of its
own.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `config`

```elixir
@spec config(keyword(), atom(), keyword()) :: config()
```

The `batch_*` options of a sink's `opts`, range-checked, over the codec's
defaults. `entries` and `bytes` are `{default, maximum}`.

  * `:batch_linger_ms` (default 20), `:batch_max_entries`, `:batch_max_bytes`
  * `:batch_queue_max` (4 096), `:batch_queue_max_bytes` (128 MiB),
    `:batch_queue_tenant_max` (1 024)
  * `:batch_max_inflight` — default the instance's `dispatch.sink_concurrency`, or 8

Raises `ArgumentError` naming the option.

# `deliver`

```elixir
@spec deliver(module(), atom(), keyword(), entry()) :: result()
```

Queue `entry` for `codec`'s batcher of the destination `opts` names and wait for
its result.

# `start_link`

# `stats`

```elixir
@spec stats(atom()) :: %{
  batchers: non_neg_integer(),
  queued: non_neg_integer(),
  queued_bytes: non_neg_integer()
}
```

What `instance`'s batchers hold right now: `%{batchers, queued, queued_bytes}`,
summed from each batcher's registered counters (no scan of a queue, no call).

# `validate_opts!`

```elixir
@spec validate_opts!(module(), keyword()) :: :ok
```

Raise `ArgumentError` naming the option unless `opts`' `:batch` is a boolean and
its `batch_*` options are in range for `codec`. What `Ankusa.Sink.SQS.validate_opts!/1`
and `Ankusa.Sink.GooglePubSub.validate_opts!/1` run, so a bad value fails at
load; a delivery that reaches a batcher with one is dead-lettered with the same
message.

---

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