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

Group-commit batcher, one process per partition. Requests hand it an envelope
and **block on the reply**. The batcher buffers callers, then flushes the whole
batch to the queue in one `enqueue`, which the queue writer merges with the
batches of the other partitions into one group commit (one fsync). Every
blocked caller is replied to *after* the commit returns — that is what makes the
ack honest.

The append itself runs in a `Task`, so the batcher keeps accepting requests
while commits are in flight, up to `batcher.max_commits` of them (64 by
default): a request that finds a free slot is handed to the writer at once, one
that does not accumulates behind the commits ahead and goes the instant one
finishes. Keeping a second commit queued at the writer is what lets it merge
the batches of every partition into one group, and what keeps a caller waiting
for one commit rather than for the one ahead of it and then its own. Nothing
waits on a commit except the callers whose own records are in it — with
`max_delay_ms: 0` that is the steady state, and the batch is naturally as
large as concurrency allows.

The queue is bounded, and the bound counts buffered *and* in-flight records.
When it is reached the batcher sheds load: the caller gets `{:error,
:overload}`, which the edge turns into a `503` with `Retry-After`. Never ack
what you haven't saved.

The bound is the partition's, so on its own one tenant could fill it and every
other tenant would be shed with it. A tenant whose limit sets `max_queue` or
`max_queue_bytes` (`Ankusa.Edge.RateLimiter`) is counted by `commit/4`
across this node's partitions before the call reaches a partition, and is
refused `{:error, {:tenant_overload, retry_after_ms}}` (a `429`) past its
share: the partition is never asked, and the global `503` is left to the
tenants still within theirs.

## Deadlines

Every record carries a deadline: the latest moment its batch may *start*
committing. A record still buffered at its deadline (a commit ahead of it is
stalled) is answered `{:error, :store_unavailable}` and dropped, and the
writer refuses a batch whose earliest deadline passed while it queued behind
other work (`Ankusa.Queue.enqueue/3`). A batch the writer has started is never
abandoned: its callers wait for the commit's outcome, however long the disk
takes. A stall therefore never turns into a `503` for a hook that is then
committed. The one way a `503` can still cover a durable hook is a process
dying while the writer is mid-commit (the commit task killed, the batcher
crashing and taking its tasks down, or the writer crashing after the sync but
before it replies); the provider's retry then stores the hook a second time.
The commit task is supervised, not linked: a task that dies takes only its own
batch's callers with it, never the batcher or the records buffered behind it.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `commit`

```elixir
@spec commit(atom(), non_neg_integer(), Ankusa.Queue.entry(), timeout()) ::
  {:committed, Ankusa.Envelope.t()}
  | {:duplicate, Ankusa.Envelope.t()}
  | {:error,
     :overload
     | :tenant_limit
     | :tenant_erasing
     | :dedupe_quota
     | :store_unavailable}
  | {:error, :intake_unavailable | :intake_pending}
  | {:error, {:half, :intake_unavailable | :store_unavailable, String.t()}}
  | {:error, {:tenant_overload, pos_integer()}}
```

Submit a record and block until it is durably committed (or shed).

Returns `{:committed, env}` | `{:duplicate, env}` | `{:error, :overload}` |
`{:error, {:tenant_overload, retry_after_ms}}` |
`{:error, :tenant_limit}` (the hook's tenant would pass `tenants.max`) |
`{:error, :tenant_erasing}` (an erase of the hook's tenant is running) |
`{:error, :dedupe_quota}` (its dedupe key would pass `dedupe.max_entries` or
`dedupe.max_entries_per_tenant`) |
`{:error, :store_unavailable}`; under `wal: :dual` also
`{:error, :intake_unavailable}` (the intake breaker is open or its pending set is
full: nothing was stored), `{:error, :intake_pending}` (a retry that met an
original not yet confirmed in the log) and `{:error, {:half, reason, id}}` (the
hook persisted in only one place and is carried to completion by
`Ankusa.Intake.Reconciler`). The queue itself is never allowed to crash the
call: a failed commit (or a dead writer process) is reported as
`:store_unavailable`, which the edge maps to `503`.

The record's tenant share is counted before the call and given back when the
call returns or exits, whatever the outcome, so a batcher that crashes with
records buffered leaks nothing.

`timeout` (milliseconds, or `:infinity`) bounds how long the record may wait
before its batch *starts* committing. Once it has started the call waits for
the outcome, so a slow disk is never answered `:store_unavailable` for a hook
it then commits (see the moduledoc for the crash windows that still can).

# `queued`

```elixir
@spec queued(atom(), non_neg_integer()) :: non_neg_integer()
```

How many records the partition holds, buffered and in flight. A drain waits
for every partition to report `0` (`Ankusa.Instance.Drain`): each of those
records belongs to a request that has not been answered yet.

# `record_deadline_ms`

```elixir
@spec record_deadline_ms() :: pos_integer()
```

The default record deadline in milliseconds: the latest a record may wait before its batch starts committing.

# `start_link`

---

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