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

Claim check: store payloads too large to ride inline in a queue message, and
hand each one back by reference.

Ankusa's dispatch pipeline is the only writer. It writes directly to the
instance's claim stores (`Ankusa.ClaimCheck.Stores`: one or more object
stores, with failover between them), packing every claim a WAL read batch
needs for one tenant into one object (`Ankusa.ClaimCheck.Pack`), so one
`PUT` covers many claims. Each claim gets an `Ankusa.ClaimCheck.Ref` (a URN
naming its tenant and claim id — and, with `claim_check.ref_signing_keys`,
an expiry and an HMAC over both) and the sha256 of its bytes, which travel
together as a `t:claim/0`.

Readers either call `redeem/3` in-process, fetch
`GET /v1/claims/<tenant>/<claim_id>` from a `:claim_check`-role node
(`Ankusa.ClaimCheck.Router`), or read the objects straight from the bucket.
The gateway authenticates and authorizes every request (`claim_check.auth`,
`claim_check.authorize`) and holds the ref to `verify_ref/2`; `read/4`,
`locate/4` and `read_bytes/5` below do neither, so code calling them has
already decided who may read what.

Objects are written once under a freshly minted id and never rewritten, so
every read is safe to cache forever.

The gateway checks what it serves against the pack's manifest
(`claim_check.integrity_check: :manifest`, the default): see `digest/3`.

Permanent errors (`:not_found`, `:integrity_mismatch`, `:invalid_ref`,
`:invalid_tenant`, `:invalid_id`, `:unsigned_ref`, `:bad_signature`,
`:ref_expired`, `:too_large`) mean retrying won't help — callers should dead-letter.
`{:unavailable, reason}` is transient, and so is `:forbidden` (the object
store refused this node's credential: a configuration fault to fix, not a
reason to drop the claim).

# `claim`

```elixir
@type claim() :: %{ref: Ankusa.ClaimCheck.Ref.t(), sha256: String.t()}
```

A checked-in claim: its ref, and the lowercase hex sha256 of its bytes.

# `item`

```elixir
@type item() :: %{
  :id =&gt; String.t(),
  :body =&gt; binary(),
  optional(:tenant_id) =&gt; String.t(),
  optional(:content_type) =&gt; String.t() | nil,
  optional(:received_at) =&gt; integer() | nil
}
```

# `location`

```elixir
@type location() :: %{
  tenant_id: String.t(),
  pack_id: String.t(),
  index: non_neg_integer(),
  store: non_neg_integer(),
  key: String.t(),
  offset: non_neg_integer(),
  length: non_neg_integer(),
  claims: pos_integer() | nil,
  claims_end: non_neg_integer() | nil
}
```

A located claim: its tenant, pack and index in the pack, the index of the
claim store the pack was found in, its object key there, and the claim's
offset and length in it; and how many claims the pack holds (`nil` if its
index does not say) and the offset where its last claim ends, when the index
bytes read have that row (`nil` otherwise).

# `reason`

```elixir
@type reason() ::
  :invalid_ref
  | :invalid_tenant
  | :invalid_id
  | :not_found
  | :integrity_mismatch
  | :unsigned_ref
  | :bad_signature
  | :ref_expired
  | :forbidden
  | :too_large
  | {:unavailable, term()}
```

# `check_in`

```elixir
@spec check_in(atom(), String.t(), [item()], keyword()) ::
  {:ok, %{required(String.t()) =&gt; claim()}} | {:error, reason()}
```

Store `items` for one tenant as a single pack object, with one conditional
`PUT` (`Ankusa.BlobStore.put_if_absent/4`): a claim store that already holds
an object under the pack's key refuses, and `check_in/4` writes the pack
under a **fresh pack id** (at most 3 attempts), so a reused or colliding id
can never overwrite claims that are already referenced. The refs it returns
name the id the pack was written under.

Returns a `t:claim/0` per item id once the write is durable in one of the
claim stores (`Ankusa.ClaimCheck.Stores.write/6`). The whole call, every
store tried, every attempt and multipart part, and every pack-id retry,
shares one deadline (`Ankusa.ClaimCheck.Stores.write_deadline/1`, the largest
claim store `op_deadline_ms`): past it the call returns an error rather than
keep a dispatch task and its connections busy. A store that throttles with a
`Retry-After` that does not fit is `{:error, {:unavailable, {:throttled,
seconds}}}` at once, for dispatch's own backoff to schedule. Items that no
single pack can hold (more than 65 000 claims, or about 4 GiB of them) are
`{:error, :too_large}` before anything is built; `check_in_batch/2` splits
its packs inside both limits. opts:

  * `:pack_id` — the first pack id tried (`Ankusa.ClaimCheck.Ref.pack_id/2`);
    default a freshly minted one (`Ref.new_pack_id/0`: now plus 64 random bits).

# `check_in_batch`

```elixir
@spec check_in_batch(atom(), [item()]) :: %{
  required(String.t()) =&gt; {:ok, claim()} | {:error, reason()}
}
```

Store many items, possibly for many tenants: group them by tenant, split each
group into packs of at most `claim_check.pack_max_bytes`, and upload the
packs concurrently. A body larger than the cap gets a pack of its own.

Returns a result per item id. A failed pack fails only its own items: a store
that raises, or an upload process that dies, is `{:error, {:unavailable,
reason}}` for that pack's items and never takes the caller down. An item no
pack can hold (over ZIP's 4 GiB) is `{:error, :too_large}`.

# `check_retry_horizon`

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

Refuse a source whose retry horizon outlasts the bucket's lifecycle rule:
`{:error, message}` when the source routes to a claim-checking sink, claims
are in an object store, `claim_check.bucket_expiry_days` is declared, and
`max(dispatch.retry horizon, the source's) + queue.dlq.retention_days +
claim_check.redeem_grace_ms` (in days) exceeds it. A pack the rule deleted
mid-retry would be a silent loss, so this runs on every source write
(`Ankusa.SourceStore.put/6` answers `422 retry_exceeds_claim_expiry`) and
over the stored sources at boot (`validate_stored_horizons!/2`).

# `deadline`

```elixir
@spec deadline(atom() | Ankusa.Config.t()) :: integer()
```

The `:deadline` of a read starting now: `claim_check.read_deadline_ms` from
now, as an absolute `System.monotonic_time(:millisecond)`.

# `digest`

```elixir
@spec digest(atom(), location(), keyword()) :: {:ok, String.t()} | {:error, reason()}
```

The lowercase hex sha256 of a located claim, as its pack's manifest records
it (`claim_check.integrity_check: :manifest`, the gateway's check).

The pack's digests come from `Ankusa.ClaimCheck.Cache` (inside the pack's
entry, so under `claim_check.read_cache_bytes` and aged with it), else from
one ranged read of the manifest where the last claim ends (a second one for
a manifest longer than the first guessed), shared by every concurrent caller
through `Ankusa.ClaimCheck.Reader` like any other read. A manifest that does
not parse, or does not list exactly this pack's claims, is
`{:error, :integrity_mismatch}`; a pack that ends before its manifest does is
`:not_found`. `opts[:deadline]` is as for `locate/4`.

# `locate`

```elixir
@spec locate(atom(), String.t(), String.t(), keyword()) ::
  {:ok, location()} | {:error, reason()}
```

Find a claim without reading it: its pack's index, from
`Ankusa.ClaimCheck.Cache`, else one ranged read of the index's first
8 231 bytes (the first 1 024 claims' rows), or up to the claim's row when
that is further, searching every claim store under both key layouts
(`Ankusa.ClaimCheck.Stores.read_range/6`). A claim id past the end of the
index, or a pack no store has, is `:not_found`; a pack not found while some
store could not answer is that store's error.

On a `:claim_check` node the read goes through `Ankusa.ClaimCheck.Reader`,
and gives up with `{:unavailable, :deadline}` at `opts[:deadline]` (default
`deadline/1`); elsewhere it goes straight to the store.

# `read`

```elixir
@spec read(atom(), String.t(), String.t(), keyword()) ::
  {:ok, binary()} | {:error, reason()}
```

Read a claim's bytes, with no digest check: `locate/4`, then `read_bytes/5`
over the whole claim, both under one `:deadline` (default `deadline/1`). A
claim id past the end of its pack's index, a missing pack, or a pack that
ends before the claim does is `:not_found`.

# `read_bytes`

```elixir
@spec read_bytes(atom(), location(), non_neg_integer(), non_neg_integer(), keyword()) ::
  {:ok, binary()} | {:error, reason()}
```

`length` bytes of a located claim, starting `first` bytes into it: the
whole claim from `Ankusa.ClaimCheck.Cache` when it is there, else one ranged
read, through `Ankusa.ClaimCheck.Reader` on a `:claim_check` node (see
`locate/4` for `opts[:deadline]`). A whole claim of at most
`claim_check.read_cache_max_entry_bytes` is cached; a range never is. The
caller keeps the range inside the claim's `length`; a pack that ends before
the range does is `:not_found`, and forgets the cached pack (it was swept).

# `redeem`

```elixir
@spec redeem(atom(), Ankusa.ClaimCheck.Ref.t() | String.t(), String.t()) ::
  {:ok, binary()} | {:error, reason()}
```

Redeem a ref (a `%Ref{}` or its URN) for its bytes and check them against
`sha256`, the lowercase hex digest the queue message carried with the ref.
Integrity is never trusted from the store. The ref is held to the node's
policy first (`verify_ref/2`): a forged, expired or (when refused) unsigned
ref reads nothing.

# `retry_horizon_ms`

```elixir
@spec retry_horizon_ms(Ankusa.Config.t(), [Ankusa.Source.t()]) :: pos_integer()
```

The longest, in milliseconds, a hook can keep retrying: the instance's
`dispatch.retry` horizon (`max_elapsed_ms`), or a longer one of any of
`sources` that routes to a sink using the claim check. The first term of how
long a claim in an object store is needed (then `queue.dlq.retention_days` and
`claim_check.redeem_grace_ms`); the 24-hour default is what the bucket rule
has to outlast, not the 21 days the old 84-attempt policy was assumed to take.

# `validate_config!`

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

Validate `config.claim_check` at boot. Raises rather than surfacing a
misconfiguration at runtime:

  * `claim_check.pack_max_bytes` must be a positive integer
  * `claim_check.retention_days` must be `nil` or an integer >= 1, and
    `claim_check.sweep_interval_ms` an integer >= 1
  * `claim_check.blob_stores` must be `nil` or a non-empty list of
    `{module, opts}`; `claim_check.placement` `:ordered` or `:hash`;
    `claim_check.key_layout` `:flat` or `:sharded`
  * `claim_check.retention_days` set with any non-`LocalFS` claim store (the
    sweeper only ever covers `LocalFS`; S3/GCS need a bucket lifecycle rule)
  * `claim_check.bucket_expiry_days` declared with an object-store claim store
    shorter than the longest a claim can be needed: the retry horizon (the
    instance's `dispatch.retry` `max_elapsed_ms`, 24 hours by default, or the
    longest `retry.max_elapsed_ms` of a configured source that uses the claim
    check) + `queue.dlq.retention_days` + `claim_check.redeem_grace_ms`, each
    rounded up to days. A bucket lifecycle rule cannot see claim leases, so
    the declared age is the real cutoff; undeclared, boot warns with the
    minimum. Sources written later are held to it too
    (`check_retry_horizon/2`)
  * `max_body_bytes` of 4 GiB or more: a pack can't hold a body that large
  * `claim_check.ref_signing_keys` must be a list of strings of at least 16
    bytes, and not empty while `claim_check.accept_unsigned_refs` is `false`
  * `claim_check.authorize` must be `nil` or a `{module, fun, args}` the
    module exports with arity `length(args) + 1`

A `:claim_check`-only node with a `LocalFS` claim store boots with a
warning: the gateway reads its own disk, and other nodes' claims are only
there if that directory is shared with them.

# `validate_stored_horizons!`

```elixir
@spec validate_stored_horizons!(Ankusa.Config.t(), [Ankusa.Source.t()]) :: :ok
```

`check_retry_horizon/2` over the sources a writable store loaded at boot:
raises `ArgumentError` naming the first source whose horizon outlasts the
bucket rule, as `validate_config!/1` does for the configured ones.

# `verify_ref`

```elixir
@spec verify_ref(atom() | Ankusa.Config.t(), Ankusa.ClaimCheck.Ref.t()) ::
  :ok | {:error, :unsigned_ref | :bad_signature | :ref_expired}
```

Check a ref against this node's policy, as the gateway does before it reads
anything:

  * a v2 ref must carry a signature some `claim_check.ref_signing_keys` entry
    produced (`{:error, :bad_signature}` otherwise) and must not be past its
    expiry (`{:error, :ref_expired}`)
  * a v1 ref is `{:error, :unsigned_ref}` unless `claim_check.accept_unsigned_refs`
    is true (the default only while `ref_signing_keys` is empty)

---

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