@spec check_in(atom(), String.t(), [item()], keyword()) ::
  {:ok, %{required(String.t()) => 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 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).

@spec check_in_batch(atom(), [item()]) :: %{
  required(String.t()) => {: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}.

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

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

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

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

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

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

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

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

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

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

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