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

Ingest admission, enforced in this node's memory. Two layers, in the order a
request meets them:

## Before the body is read: CPU protection

Charged by `Ankusa.Edge.Router` after the source is looked up and before a
byte of the body is read, so none of them depends on verification and a forged
flood is refused cheaply. They protect this node, they are not a quota.

  * `rate_limits.per_ip` — hooks per second per client address
    (`hit_ip/2`). The address is `Ankusa.Net.ClientIP`'s, with
    `routes.trusted_proxies` deciding whose `X-Forwarded-For` is believed; an
    IPv6 address counts as its `/64`. Addresses hash into a fixed 2^20 buckets
    with a per-boot salt, so the table has a hard ceiling however many
    addresses a flood invents.
  * `rate_limits.node` — hooks per second for the whole node (`hit_node/1`).
  * `rate_limits.sources.default.bytes_per_second` — a body-bytes-per-second
    ceiling per source (`hit_source_bytes/3`), 100 MB/s unless changed. It is
    charged with the declared `Content-Length`; a body with no declared length
    is charged when it has been read.

A denial is `429` with `Retry-After`; `pre_read/4` says which one answered.

## After verification: the tenant's quota

A limit is `%{rate: number, burst: pos_integer}`: `rate` is hooks per second
(fractions are allowed — `0.5` means one every two seconds) and `burst` is the
most hooks admitted back to back. It may also carry:

  * `bytes_per_second`, `burst_bytes` — body bytes, charged with the hook
    (`burst_bytes` defaults to one second's worth and needs
    `bytes_per_second`). A body bigger than `burst_bytes` is admitted only
    into a full bucket and empties it.
  * `max_body_bytes` — the biggest body this tenant may send, at most the
    instance's `max_body_bytes` (`body_limit/2`; the router answers `413`).
  * `max_queue`, `max_queue_bytes` — how many hooks and body bytes the tenant
    may hold in this node's batchers at once, buffered and in flight
    (`acquire_queue/3`, `release_queue/2`). Past either is
    `429 tenant_overload`, never the global `503 overload`, so one tenant
    filling its share cannot starve the rest.

Limits come from three places, in precedence order (the first one found
applies *whole*; its keys do not merge with the next's):

  1. a runtime override set through the admin API (`PUT
     /v1/tenants/:tenant/rate-limit`), persisted to this node's `Ankusa.Store`
     and node-local, exactly like `Ankusa.SourceStore.Persistent`'s sources;
  2. `rate_limits.tenants[tenant]`;
  3. `rate_limits.default`.

No limit at all (`default: nil` with no tenant entry, the default config)
means unlimited. There is no "unlimited" value — exempt a single tenant from a
default by giving it a high limit. A tenant-carrying URL scheme
(`Ankusa.RouteResolver.TenantPath`) lets the sender name the tenant, so
`validate_config!/1` refuses it without a `default`.

A hook that clears the hook bucket and is then denied on bytes has spent its
hook token; a forged request spends none of these, because the tenant charge
comes after verification.

## Enforcement

One ETS row per bucket holds its GCRA theoretical arrival time, and `hit/3`
and friends update it with a compare-and-swap (`:ets.select_replace/2`), so
concurrent requests on one bucket can never overshoot the limit. A denial
writes nothing. The rows are spread over 16 tables, a bucket in the one its
key hashes to (`:erlang.phash2(tenant, 16)`; a tenant's hook and byte buckets
share one). Rows whose TAT has passed are a full bucket — identical to having
no row — so a periodic sweep deletes them, which also bounds memory when the
route resolver lets senders mint tenant ids. Each tick sweeps one shard, in
turn, every 3.75 s, so a table of a million rows is never scanned whole in
one tick.

Queue occupancy is two counters per tenant (`:ets.update_counter/4`: charge,
compare, undo on refusal), held only while records are in flight and deleted
at zero, so the table is bounded by the batchers' own caps.

The tables live on the edge node and nowhere else: with the default
`rate_limits.scope: :node` a fleet of N edge nodes admits up to N × the limit,
and an override set against one node's admin port is that node's alone.
`rate_limits.scope: :fleet` moves the tenant's buckets (hooks and bytes) into
`fleet.store` (`Ankusa.Edge.Fleet`, `fleet_charges/3`), so the fleet admits one
limit; this node's buckets then only answer when the store cannot. The override
is still read from this node (set it on every node), and `per_ip`, `node` and
`sources` stay per node under either scope: they protect this node's CPU, they
are not a quota. Per-tenant queue limits cover the group-commit batcher
(`wal: disk`); `wal: none` publishes inside the request and has no queue.

# `denied`

```elixir
@type denied() :: {:error, {:rate_limited, pos_integer()}}
```

# `lease`

```elixir
@type lease() :: {String.t(), non_neg_integer()} | nil
```

What `acquire_queue/3` hands back: pass it to `release_queue/2`. `nil` is nothing counted.

# `limit`

```elixir
@type limit() :: %{
  :rate => number(),
  :burst => pos_integer(),
  optional(:bytes_per_second) => number(),
  optional(:burst_bytes) => pos_integer(),
  optional(:max_body_bytes) => pos_integer(),
  optional(:max_queue) => pos_integer(),
  optional(:max_queue_bytes) => pos_integer()
}
```

# `source`

```elixir
@type source() :: :override | :config | :default
```

# `acquire_queue`

```elixir
@spec acquire_queue(atom(), String.t(), non_neg_integer()) ::
  {:ok, lease()} | {:error, {:tenant_overload, pos_integer()}}
```

Count one hook of `bytes` body bytes against `tenant_id`'s queue share, when
its limit sets `max_queue` or `max_queue_bytes`. Returns a lease to hand back
to `release_queue/2` once the batcher has answered, or
`{:error, {:tenant_overload, retry_after_ms}}` with nothing counted.

As in the batcher itself, a hook bigger than `max_queue_bytes` on its own is
admitted when the tenant holds nothing else.

# `body_limit`

```elixir
@spec body_limit(atom(), String.t()) :: pos_integer() | nil
```

The biggest body `tenant_id` may send, when its limit sets `max_body_bytes`
(otherwise `nil`: the instance's `max_body_bytes` alone applies).

# `child_spec`

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

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `delete_override`

```elixir
@spec delete_override(atom(), term(), keyword()) ::
  :ok | {:error, :not_found} | {:error, :store_unavailable}
```

Remove `tenant_id`'s runtime override, so the config's limit applies again.
Resets the buckets too, for the same reason a `put` does. Options: `audit:`.

# `effective`

```elixir
@spec effective(atom(), String.t()) :: {limit(), source()} | {nil, :none}
```

The limit that applies to `tenant_id` on this node, with where it came from.
`{nil, :none}` means unlimited.

# `fleet_charges`

```elixir
@spec fleet_charges(atom(), String.t(), non_neg_integer()) :: [
  Ankusa.RateLimiter.Store.charge()
]
```

The buckets `hit/3` would charge for one hook of `bytes` body bytes, as
`Ankusa.RateLimiter.Store` charges (GCRA increment and capacity in
microseconds) — `[]` when the tenant has no limit. Reads the effective limit
on this node and charges nothing.

# `hit`

```elixir
@spec hit(atom(), String.t(), non_neg_integer()) :: :ok | denied()
```

Charge one hook of `bytes` body bytes to `tenant_id`'s buckets. `:ok` admits
it; a denial carries how long the client should wait, in milliseconds. The
byte bucket is charged only when the limit has `bytes_per_second`; `bytes`
defaults to `0` (a hook-count charge only).

# `hit_ip`

```elixir
@spec hit_ip(atom(), Plug.Conn.t()) :: :ok | denied()
```

Charge one hook to the client address's bucket (`rate_limits.per_ip`).

# `hit_node`

```elixir
@spec hit_node(atom()) :: :ok | denied()
```

Charge one hook to the node's bucket (`rate_limits.node`).

# `hit_source_bytes`

```elixir
@spec hit_source_bytes(atom(), String.t(), non_neg_integer()) :: :ok | denied()
```

Charge `bytes` of body to `source_id`'s bucket
(`rate_limits.sources.default.bytes_per_second`).

# `list`

```elixir
@spec list(atom()) :: %{
  default: limit() | nil,
  tenants: [{String.t(), limit(), :override | :config}]
}
```

Every limit this node knows: the config's tenants unioned with the runtime
overrides, sorted by tenant id, plus the configured default.

# `parse_limit`

```elixir
@spec parse_limit(term(), pos_integer()) ::
  {:ok, limit()} | {:error, :invalid, String.t()}
```

Parse a limit from JSON-shaped attrs (`%{"rate" => …, "burst" => …}` plus any
of the optional keys, atom keys accepted). `rate` and `burst` are required;
neither is inferred. `max_body_bytes` is the instance's, the ceiling a
tenant's `max_body_bytes` may not pass.

# `pre_read`

```elixir
@spec pre_read(atom(), Plug.Conn.t(), String.t(), non_neg_integer() | nil) ::
  :ok | {:error, {:ip | :node | :source, pos_integer()}}
```

The checks that run before a request's body is read: the client's address
(`rate_limits.per_ip`), then the node (`rate_limits.node`), then the source's
declared `bytes` (`rate_limits.sources`; `nil` is a body with no declared
length, charged later with `hit_source_bytes/3`). The first to refuse answers,
with the scope that did and how long to wait in milliseconds. An address
refused here has not touched the node's bucket.

# `put_override`

```elixir
@spec put_override(atom(), term(), term(), keyword()) ::
  {:ok, limit()} | {:error, :invalid, String.t()} | {:error, :store_unavailable}
```

Set `tenant_id`'s runtime override, resetting its buckets on success: without
the reset a raised limit would stay blocked until the old limit's accumulated
debt drained. The override replaces the tenant's limit whole, optional keys
included. Options: `audit:`, the `Ankusa.Audit.Context` of the change.

# `release_queue`

```elixir
@spec release_queue(atom(), lease()) :: :ok
```

Give back what `acquire_queue/3` counted. Safe to call with `nil`.

# `start_link`

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

# `validate_config!`

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

Validate the `rate_limits` section of a config, raising `ArgumentError` on the
first problem. Called on every node and role at boot, so a typo fails the
boot rather than the first limited hook.

A config whose route resolver is `Ankusa.RouteResolver.TenantPath` must set a
`rate_limits.default`: the URL names the tenant, so without one every tenant
the config does not list is unlimited.

---

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