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