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

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

@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).

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

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

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

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

@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).

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

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

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

@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).

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