Ankusa.ClaimCheck.Reader (ankusa v0.7.0)

Copy Markdown View Source

The one process every store read of a :claim_check node passes through, so the gateway stays fast while an object store throttles or hangs:

  • Coalescing — concurrent reads of one key (a pack's index, a claim's bytes, one range of them) share one store request, and every caller gets its answer.
  • An adaptive limit — per store, at most limit store requests run at once. It starts at claim_check.read_max_inflight, halves when the store throttles (429, 503) or cannot answer — at most once per 100 ms, so a wave of requests failing together halves it once — and grows by one after limit successes, back up to that ceiling (AIMD). A missing key is a success; a 403 changes nothing.
  • A bounded queue — a read that finds no free slot waits for one, up to claim_check.read_queue_max reads per store; past that it is shed at once with {:unavailable, :overloaded}.
  • A deadline — every caller passes one (claim_check.read_deadline_ms after its request started) and gets {:unavailable, :deadline} once it passes; a queued read whose deadline passed is dropped without a store request. A store request that outlives its callers still runs to the end and fills Ankusa.ClaimCheck.Cache, so the client's retry finds it.
  • A node-wide request gate — admit/2 and release/1 bound how many gateway requests the node serves at once (claim_check.max_concurrent_reads), cache hits included, so a flood of cheap reads can't starve the node. The gateway answers 503 + Retry-After past it, before it locates anything.

The limit is per store, not per key prefix. S3 throttles per prefix, and claim_check.key_layout: :sharded is what spreads a tenant's packs over more of them; this bounds what one node sends a store as a whole. Every answer is charged to the store a fetch was addressed to: an index fetch searches every store (Ankusa.ClaimCheck.Stores.read_range/6) but is charged to the pack's home (Ankusa.ClaimCheck.Stores.home/2).

Only a :claim_check node runs it, beside the gateway. A read anywhere else — a dispatch node calling Ankusa.ClaimCheck.redeem/3 — goes straight to the store, bounded only by the store's read timeout.

Emits [:ankusa, :claim_check, :read] once per caller (see fetch/5), [:ankusa, :claim_check, :read_limit] on every change of a store's limit, and [:ankusa, :claim_check, :read_state] every second per store it has read from (see Ankusa.Telemetry).

Summary

Types

key()

What a fetch reads: coalescing is by key.

stats()

Functions

admit(instance, max)

Count the calling process as one of the gateway's requests in flight, unless max already are (:shed; the caller answers 503 and does not release/1). max is claim_check.max_concurrent_reads; 0 admits everything, and so does a node whose reader is not running, which has no gate to count at.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

fetch(instance, key, store, fun, deadline)

Run fun — one read of key from claim store store — under the store's limit, or join a fetch of key already running. deadline is an absolute System.monotonic_time(:millisecond): past it the caller gets {:unavailable, :deadline}, whatever the fetch is doing; the fetch it started or joined still runs to the end and fills the cache.

release(instance)

Give back the slot admit/2 counted for the calling process.

start_link(opts)
stats(instance)

Each store's limit, store requests running, and reads queued.

Types

key()

@type key() ::
  {:index, tenant_id :: String.t(), pack_id :: String.t()}
  | {:manifest, String.t(), String.t()}
  | {:bytes, String.t(), String.t(), index :: non_neg_integer()}
  | {:range, String.t(), String.t(), non_neg_integer(),
     first :: non_neg_integer(), length :: pos_integer()}

What a fetch reads: coalescing is by key.

stats()

@type stats() :: %{
  required(non_neg_integer()) => %{
    limit: pos_integer(),
    inflight: non_neg_integer(),
    queued: non_neg_integer()
  }
}

Functions

admit(instance, max)

@spec admit(atom(), non_neg_integer()) :: :ok | :shed

Count the calling process as one of the gateway's requests in flight, unless max already are (:shed; the caller answers 503 and does not release/1). max is claim_check.max_concurrent_reads; 0 admits everything, and so does a node whose reader is not running, which has no gate to count at.

The gate is a set of request processes, not a counter, so a request process that is killed before it releases is dropped by the reader's next tick instead of leaking a slot for good.

child_spec(init_arg)

@spec child_spec(keyword()) :: Supervisor.child_spec()

Returns a specification to start this module under a supervisor.

See Supervisor.

fetch(instance, key, store, fun, deadline)

@spec fetch(atom(), key(), non_neg_integer(), (-> result), integer()) ::
  result | {:error, {:unavailable, :overloaded | :deadline | {:exit, term()}}}
when result: {:ok, term()} | {:error, term()}

Run fun — one read of key from claim store store — under the store's limit, or join a fetch of key already running. deadline is an absolute System.monotonic_time(:millisecond): past it the caller gets {:unavailable, :deadline}, whatever the fetch is doing; the fetch it started or joined still runs to the end and fills the cache.

Every caller is counted once on [:ankusa, :claim_check, :read]: the reader counts the caller whose fetch went to the store (:store, when the fetch finishes, whether or not that caller is still waiting); every other caller counts itself, as :coalesced when it got the fetch's answer and :shed when it got none (the queue was full, its deadline passed, or the reader is down).

release(instance)

@spec release(atom()) :: :ok

Give back the slot admit/2 counted for the calling process.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

stats(instance)

@spec stats(atom()) :: stats()

Each store's limit, store requests running, and reads queued.