LLMProxy.ConcurrencyLimiter (llm_proxy v0.2.0)

In-memory concurrent-request admission for LLMProxy API keys.

Add LLMProxy.Limit.concurrent_requests/1 to a key's budget_limits to enable the gate. Admission and release do not query storage. Limits apply to one LLMProxy runtime instance; deployments with multiple instances must size each instance accordingly.

A lease follows a stream into the process that consumes it. The lease is released when the request or stream ends, when stream enumeration halts, or when the process that owns the lease exits.

Summary

Types

lease()

status()

Functions

acquire(api_key, owner \\ self())

Tries to acquire one request lease for an API key.

child_spec(opts)

error_message()

Returns the stable public message for an admission refusal.

release(lease)

Releases a request lease. Repeated release calls are safe.

retry_after_seconds()

Returns the retry delay advertised by HTTP admission refusals.

run(api_key, fun)

Runs a non-streaming request while holding its configured lease.

start_link(opts \\ [])

status()

Returns aggregate content-free counters for limited requests.

status(api_key)

Returns the current active count and configured limit for one key.

wrap_stream(stream, lease)

Wraps a stream so its lease follows the consumer and is always released.

Types

lease()

@type lease() :: LLMProxy.ConcurrencyLimiter.Lease.t() | :unlimited

status()

@type status() :: %{
  active: non_neg_integer(),
  admitted: non_neg_integer(),
  rejected: non_neg_integer(),
  released: non_neg_integer()
}

Functions

acquire(api_key, owner \\ self())

@spec acquire(map(), pid()) ::
  {:ok, lease()} | {:error, {:limit_exceeded, non_neg_integer()}}

Tries to acquire one request lease for an API key.

Keys without a concurrent-request limit receive an unlimited no-op lease.

child_spec(opts)

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

error_message()

@spec error_message() :: String.t()

Returns the stable public message for an admission refusal.

release(lease)

@spec release(lease()) :: :ok

Releases a request lease. Repeated release calls are safe.

retry_after_seconds()

@spec retry_after_seconds() :: pos_integer()

Returns the retry delay advertised by HTTP admission refusals.

run(api_key, fun)

@spec run(map(), (-> result)) ::
  result | {:error, {:limit_exceeded, non_neg_integer()}}
when result: term()

Runs a non-streaming request while holding its configured lease.

start_link(opts \\ [])

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

status()

@spec status() :: status()

Returns aggregate content-free counters for limited requests.

status(api_key)

@spec status(map() | String.t()) :: map()

Returns the current active count and configured limit for one key.

wrap_stream(stream, lease)

@spec wrap_stream(Enumerable.t(), lease()) :: Enumerable.t()

Wraps a stream so its lease follows the consumer and is always released.