Raxol.Core.TokenBucket (Raxol Core v2.7.0)

Copy Markdown View Source

ETS-backed token bucket, the shared rate-limiting primitive.

A bucket holds at most :capacity tokens and regains them at :refill_per_second. take/3 removes tokens and refuses once the bucket cannot cover the cost, so a caller may burst up to :capacity and then proceeds at the refill rate. This is the algorithm to reach for when the limit being respected belongs to somebody else, because a fixed-window counter admits two full windows worth of calls across a window boundary, which is exactly the burst an upstream penalises.

There is no owning process. All state lives in a public ETS table created by the caller, so buckets are keyed per upstream origin (or per user, or per action) and two callers never contend on a single mailbox.

alias Raxol.Core.TokenBucket

table = TokenBucket.new()
limit = [capacity: 30, refill_per_second: 5.0]

case TokenBucket.take(table, "trongrid", limit) do
  {:ok, _remaining} ->
    call_upstream()

  {:error, :rate_limited} ->
    {:error, {:retry_after, TokenBucket.retry_after(table, "trongrid", limit)}}
end

The limit belongs to the caller

:capacity and :refill_per_second are required on every call and are never read from application environment. One table therefore holds buckets with different limits, which is the normal case when each key is a different upstream. The consequence is that the caller owns the limit for a key and must pass the same one consistently; passing a smaller :capacity later simply clamps that key's stored tokens down to it.

Time

Refill is computed from elapsed monotonic milliseconds, and fractional tokens survive across calls. Pass :now_ms to supply the clock yourself, which is how the tests stay deterministic without sleeping. A :now_ms that moves backwards is clamped to zero elapsed, so it can never drain a bucket.

Concurrency

take/3 is atomic. It reads the bucket, then commits with :ets.select_replace/2, which only succeeds if nothing changed underneath, and retries on a lost race. Under concurrent takes against an empty-refill bucket, exactly :capacity calls are admitted.

Summary

Functions

Create a bucket table.

Tokens currently available, refilled for the time elapsed but not consumed.

Forget a bucket, so the next take/3 starts from a full one.

Forget every bucket in the table.

Milliseconds until :cost tokens are available, rounded up.

Take tokens from a bucket, refilling it for the time elapsed first.

Types

key()

@type key() :: term()

table()

@type table() :: :ets.table()

Functions

new(name \\ :raxol_token_buckets)

@spec new(atom()) :: :ets.table()

Create a bucket table.

The table is public and owned by the calling process, so it lives exactly as long as that process. Pass the returned reference to take/3 and friends.

peek(table, key, opts \\ [])

@spec peek(table(), key(), keyword()) :: float()

Tokens currently available, refilled for the time elapsed but not consumed.

Takes the same options as take/3, except :cost, which is irrelevant to a read.

reset(table, key)

@spec reset(table(), key()) :: :ok

Forget a bucket, so the next take/3 starts from a full one.

reset_all(table)

@spec reset_all(table()) :: :ok

Forget every bucket in the table.

retry_after(table, key, opts \\ [])

@spec retry_after(table(), key(), keyword()) :: non_neg_integer()

Milliseconds until :cost tokens are available, rounded up.

Returns 0 when the bucket can already cover the cost, which makes it safe to use as a Retry-After hint without a branch. Takes the same options as take/3.

take(table, key, opts \\ [])

@spec take(table(), key(), keyword()) :: {:ok, float()} | {:error, :rate_limited}

Take tokens from a bucket, refilling it for the time elapsed first.

Returns {:ok, remaining} when the bucket covered the cost, {:error, :rate_limited} when it did not. A bucket that has never been seen starts full.

Options

  • :capacity (required) - burst size, a positive number of tokens
  • :refill_per_second (required) - tokens regained per second, a positive number
  • :cost - tokens this call consumes, defaults to 1, must not exceed :capacity
  • :now_ms - monotonic milliseconds, defaults to System.monotonic_time(:millisecond)