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 isAnkusa.Net.ClientIP's, withroutes.trusted_proxiesdeciding whoseX-Forwarded-Foris 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 declaredContent-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_bytesdefaults to one second's worth and needsbytes_per_second). A body bigger thanburst_bytesis admitted only into a full bucket and empties it.max_body_bytes— the biggest body this tenant may send, at most the instance'smax_body_bytes(body_limit/2; the router answers413).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 is429 tenant_overload, never the global503 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):
- a runtime override set through the admin API (
PUT /v1/tenants/:tenant/rate-limit), persisted to this node'sAnkusa.Storeand node-local, exactly likeAnkusa.SourceStore.Persistent's sources; rate_limits.tenants[tenant];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.
Summary
Types
What acquire_queue/3 hands back: pass it to release_queue/2. nil is nothing counted.
Functions
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.
The biggest body tenant_id may send, when its limit sets max_body_bytes
(otherwise nil: the instance's max_body_bytes alone applies).
Returns a specification to start this module under a supervisor.
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:.
The limit that applies to tenant_id on this node, with where it came from.
{nil, :none} means unlimited.
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.
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).
Charge one hook to the client address's bucket (rate_limits.per_ip).
Charge one hook to the node's bucket (rate_limits.node).
Charge bytes of body to source_id's bucket
(rate_limits.sources.default.bytes_per_second).
Every limit this node knows: the config's tenants unioned with the runtime overrides, sorted by tenant id, plus the configured default.
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.
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.
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.
Give back what acquire_queue/3 counted. Safe to call with nil.
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.
Types
@type denied() :: {:error, {:rate_limited, pos_integer()}}
@type lease() :: {String.t(), non_neg_integer()} | nil
What acquire_queue/3 hands back: pass it to release_queue/2. nil is nothing counted.
@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() }
@type source() :: :override | :config | :default
Functions
@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.
@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).
@spec child_spec(keyword()) :: Supervisor.child_spec()
Returns a specification to start this module under a supervisor.
See Supervisor.
@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:.
The limit that applies to tenant_id on this node, with where it came from.
{nil, :none} means unlimited.
@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.
@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).
@spec hit_ip(atom(), Plug.Conn.t()) :: :ok | denied()
Charge one hook to the client address's bucket (rate_limits.per_ip).
Charge one hook to the node's bucket (rate_limits.node).
@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).
@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.
@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.
@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.
@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.
Give back what acquire_queue/3 counted. Safe to call with nil.
@spec start_link(keyword()) :: GenServer.on_start()
@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.