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

The hook queue: the public face of the store's hook and delivery machinery.

Every hook is written once, with a strictly increasing `seq` assigned by
`Ankusa.Queue.Writer`, and carries one delivery row per sink of its source at
ack time. Rows advance through the `:index` column family (due → inflight →
delivered/dead) and are owned by `Ankusa.Dispatch.Pipeline`; the DLQ is the
set of dead rows. A hook is deleted once its last obligation — every
delivery row and, while `:storage` runs, the archive — is cleared, whatever
roles this node runs.

The seq is set on the committed envelope returned by `enqueue/2`; on disk it
comes from the key, so an envelope read back is always stamped from its key.

# `entry`

```elixir
@type entry() :: %{
  :envelope => Ankusa.Envelope.t(),
  :sinks => [{module(), keyword()}],
  optional(:dedupe_ttl_ms) => pos_integer() | nil,
  optional(:replay_id) => String.t(),
  optional(:lanes) => [Ankusa.Ordering.lane() | nil] | nil
}
```

# `redrive_entry`

```elixir
@type redrive_entry() :: %{
  optional(:lanes) => [Ankusa.Ordering.lane() | nil] | nil,
  bin: binary(),
  size: non_neg_integer(),
  sinks: [{non_neg_integer(), module()}],
  replay_id: String.t(),
  source_id: String.t(),
  tenant_id: String.t() | nil
}
```

# `check_durable`

```elixir
@spec check_durable(Ankusa.Config.t(), Ankusa.Source.t()) ::
  :ok | {:error, String.t()}
```

`:ok` unless `config` runs `wal: :none` and none of `source`'s sinks is
durable (`Ankusa.Sink.durable?/2`) — such a source would answer the provider
`2xx` with no durable copy anywhere. Any other WAL mode always passes.

# `dead`

```elixir
@spec dead(atom(), keyword()) ::
  {:ok,
   %{
     total: non_neg_integer() | nil,
     entries: [map()],
     next_cursor: binary() | nil
   }}
  | {:error, term()}
```

One page of the dead-lettered deliveries (the DLQ), newest first, in a bounded
number of store reads: no call examines more than 10 000 dead keys, whatever
the DLQ holds. Returns `%{total, entries, next_cursor}`; each entry is
`%{envelope, reason, at, sink, error_class}` with `reason` the failure as text,
`sink` the sink's index and `error_class` its `t:Ankusa.Sink.error_class/0`
(`nil` for a row written before dead letters carried one).

Options: `:limit`, `:cursor` (the `next_cursor` of the previous page; a page
resumes below it), `:source_id`, `:since` (a unix-ms lower bound on the
dead-letter time), `:sink` (an index), `:error_class`, and `:tenants` (a list
of tenant ids: only dead letters of those tenants' hooks, read from each
tenant's range of the by-tenant dead index with no hook read; `nil`, the
default, is every tenant).

`next_cursor` is `nil` on the last page. A filter that matches little can make
a page short, or empty, with `next_cursor` set: ask again with it until it is
`nil`. `total` is a counter read (the node's, or the named tenants') when
there is no filter but `:tenants`, the number of matches when a filtered scan
reached the end of the DLQ within the page (no `:cursor`), and `nil` when it
cannot be known without a scan. A dead key whose row or hook is gone
(corruption) is logged and not listed. `{:error, :invalid_cursor}` for a
cursor this function did not return.

# `dead_for_tenant`

```elixir
@spec dead_for_tenant(atom(), String.t()) ::
  {:ok, %{dead: integer()}} | {:error, term()}
```

`%{dead: n}`: the dead rows of one tenant, from its counter: O(1).

# `enqueue`

```elixir
@spec enqueue(atom(), [entry()], integer() | :infinity) ::
  {:ok,
   committed: Ankusa.Envelope.t(),
   duplicate: Ankusa.Envelope.t(),
   error: :tenant_limit | :tenant_erasing | :dedupe_quota,
   error: :intake_unavailable | :intake_pending | :store_unavailable,
   error: {:half, :intake_unavailable | :store_unavailable, String.t()}}
  | {:error, term()}
```

Commits `entries` durably to the store (one synced batch), assigning each a
fresh seq, and returns the committed envelopes. The batch is atomic: it is
either fully durable or nothing was acked.

`deadline` is a `System.monotonic_time(:millisecond)` value. The writer
refuses with `{:error, :deadline_exceeded}` a batch it could not *start*
before then, and with `{:error, :caller_gone}` one whose caller died while it
waited; in both nothing is written and no seq is consumed. Once the writer has
started the batch the call waits for the commit's outcome however long it
takes, so a returned `{:error, _}` other than those two may mean the commit
failed, and `{:ok, _}` means it is durable. The call is never abandoned
half-way: an abandoned call would commit anyway, unacknowledged.

An entry's reply is `{:error, :tenant_limit}` instead when its tenant would
be one past `tenants.max`, or `{:error, :tenant_erasing}` when an erase of
its tenant is running (`Ankusa.Tenant`): that hook is not written. It is
`{:error, :dedupe_quota}` when its dedupe key would be a new live key past
`dedupe.max_entries` or `dedupe.max_entries_per_tenant`: the hook is not
written either, and never stored without its key.

## Group commit

The work that needs no seq runs here, in the caller (the batcher's commit
task), not in the writer: the hook's binary, one encoded delivery row per sink
and one `multi_get` of the dedupe keys (a *prefetch*). The writer merges every
call it has queued, across partitions, into one group with one synced write,
and re-validates the prefetch before it assigns seqs (`Ankusa.Queue.Writer`),
so two partitions that read the same key as absent still commit one hook.

# `hooks`

```elixir
@spec hooks(atom(), non_neg_integer(), pos_integer()) ::
  {:ok, [Ankusa.Envelope.t()]} | {:error, term()}
```

Committed envelopes with seq > `after_seq`, ascending, `seq` set from the
key.

# `label`

```elixir
@spec label(:disk | :dual | :none) :: String.t()
```

The configured queue mode's name, for boot banners and `check-config`:
`"disk"` for the store-backed queue, `"dual"` for the store plus the intake
log (`wal: :dual`), `"none"` under `wal: :none`.

# `pending_for_source`

```elixir
@spec pending_for_source(atom(), String.t(), keyword()) ::
  {:ok, %{pending: non_neg_integer(), inflight: non_neg_integer()}}
  | {:error, term()}
```

The undelivered rows of one source in this node's queue: `pending` waiting
in the due index (due now or after a backoff) and `inflight` claimed by
dispatch. Walks both indexes (O(backlog)); rows written before delivery rows
recorded their source are matched by reading their hook. `tenant:` names the
one tenant a source owned by a tenant files its hooks under, and limits the
walk to that tenant's rows.

# `pending_for_tenant`

```elixir
@spec pending_for_tenant(atom(), String.t()) ::
  {:ok,
   %{
     pending: integer(),
     scheduled: integer(),
     inflight: integer(),
     bytes: integer()
   }}
  | {:error, term()}
```

`%{pending, scheduled, inflight, bytes}` of one tenant, from its counters:
O(1). `pending` counts first-attempt rows waiting and `scheduled` retry rows
waiting (including a retry whose backoff just ended); `bytes` is the stored
size of every delivery row the tenant holds (a hook with *n* sinks counts *n*
times).

# `purge_tenant`

```elixir
@spec purge_tenant(atom(), String.t(), [:pending | :inflight | :dead]) ::
  {:ok,
   %{
     :pending => non_neg_integer(),
     :dead => non_neg_integer(),
     optional(:inflight) => non_neg_integer()
   }}
  | {:error, term()}
```

Removes `tenant`'s `:pending` (every due row), `:inflight` (claimed rows,
for a tenant dispatch has stopped claiming: see `Ankusa.Tenant`) and/or
`:dead` rows (`kinds` a subset of `[:pending, :inflight, :dead]`) in synced
chunks of 500: idempotent, so a run that stopped part-way resumes by being
called again. Without `:inflight` it never touches claimed rows. Returns the
rows removed per kind.

# `redrive`

```elixir
@spec redrive(atom(), [redrive_entry()], [Ankusa.Store.op()]) ::
  {:ok, non_neg_integer()} | {:error, term()}
```

Re-commit archived hooks as new queue entries, for a `kind: :archive` replay
job. Each entry gets a fresh seq, its hook payload, and one pending delivery
row per bound sink carrying `replay: replay_id`. No archive obligation (the
hook is already archived) and no ingest dedupe.

`extra_ops` ride in the same synced batch as the hooks — the job's cursor
update — so the cursor and the moved hooks commit atomically. Returns the
number of entries committed. The call is `:infinity` for the same reason as
`enqueue/3`: the writer never abandons a started batch.

# `release`

```elixir
@spec release(atom(), [entry()], [Ankusa.Store.op()]) ::
  {:ok, committed: Ankusa.Envelope.t(), duplicate: Ankusa.Envelope.t()}
  | {:error, term()}
```

Commit hooks released from the quarantine pen: `enqueue/3` semantics (dedupe,
archive obligation, fresh seqs) plus `replay: replay_id` on every delivery
row, with `extra_ops` (the pen deletes and the job's cursor) in the same
synced batch — so a crash cannot leave a hook both committed and still held,
or held and gone. `extra_ops` commit even when every entry is a duplicate.
A tenant past `tenants.max` (or one being erased) is the whole call's
`{:error, :tenant_limit}` (`:tenant_erasing`):
nothing commits, the pen keeps its entries.
Never deadline-bound.

# `stats`

```elixir
@spec stats(atom()) :: {:ok, map()} | {:error, term()}
```

Store statistics: `next_seq` (exact), `hooks` and `deliveries` (RocksDB key
estimates), and `disk_bytes` (SST + blob file sizes across column families).

# `validate_config!`

```elixir
@spec validate_config!(Ankusa.Config.t()) :: :ok
```

Reject a configuration that would ack without durability.

Under `wal: :none` the provider's `2xx` means "a sink confirmed", so every
statically configured source must have at least one sink whose `:ok` means the
hook is durably accepted by something that outlives this node — see
`Ankusa.Sink.durable?/2`. Raises `ArgumentError` naming the first source that
cannot make that promise.

This covers the sources in `config.source_store`. A source written at runtime
(the admin API, `Ankusa.SourceStore.put/5`) is refused by the same rule when
it is written, by every writable source store; see `check_durable/2`.

---

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