Packages
fnord
0.9.35
0.9.40
0.9.39
0.9.38
0.9.37
0.9.36
0.9.35
0.9.34
0.9.33
0.9.32
0.9.31
0.9.30
0.9.29
0.9.28
0.9.27
0.9.26
0.9.25
0.9.24
0.9.23
0.9.22
0.9.21
0.9.20
0.9.19
0.9.18
0.9.17
0.9.16
0.9.15
0.9.14
0.9.13
0.9.12
0.9.11
0.9.10
0.9.9
0.9.8
0.9.7
0.9.6
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.99
0.8.98
0.8.97
0.8.96
0.8.95
0.8.94
0.8.93
0.8.92
0.8.91
0.8.90
0.8.89
0.8.88
0.8.87
0.8.86
0.8.85
0.8.84
0.8.83
0.8.82
0.8.81
0.8.80
0.8.79
0.8.78
0.8.77
0.8.76
0.8.75
0.8.74
0.8.73
0.8.72
0.8.71
0.8.70
0.8.69
0.8.68
0.8.67
0.8.66
0.8.65
0.8.64
0.8.63
0.8.62
0.8.61
0.8.60
0.8.59
0.8.58
0.8.57
0.8.56
0.8.55
0.8.54
0.8.53
0.8.52
0.8.51
0.8.50
0.8.49
0.8.48
0.8.47
0.8.46
0.8.45
0.8.44
0.8.43
0.8.42
0.8.41
0.8.40
0.8.39
0.8.38
0.8.37
0.8.36
0.8.35
0.8.34
0.8.33
0.8.32
0.8.31
0.8.30
0.8.29
0.8.27
0.8.26
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.1
0.8.0
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.3
0.7.2
0.7.1
0.7.0
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.1
0.6.0
0.5.9
0.5.8
0.5.7
0.5.6
0.5.5
0.5.4
0.5.3
0.5.2
0.5.1
0.5.0
0.4.44
0.4.43
0.4.42
0.4.41
0.4.40
0.4.39
0.4.38
0.4.37
0.4.36
0.4.35
0.4.34
0.4.33
0.4.32
0.4.30
0.4.29
0.4.28
0.4.27
0.4.26
0.4.25
0.4.24
0.4.23
0.4.22
0.4.21
0.4.20
0.4.19
0.4.18
0.4.17
0.4.16
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.0
0.1.0
AI code archaeology
Current section
Files
Jump to
Current section
Files
lib/ai/endpoint.ex
defmodule AI.Endpoint do
@moduledoc """
API endpoint abstraction.
This behaviour centralizes HTTP JSON POST calls via `Http.post_json/3` and
applies provider-agnostic retry/backoff. Provider endpoint modules implement
`endpoint_path/0` and a small error classifier that inspects the raw HTTP/transport
outcome and returns a normalized decision (`:ok`, `{:retry, reason, wait_ms}` or
`{:fail, reason, human}`), allowing provider-specific error shapes (e.g.,
OpenAI, Venice, Cloudflare plaintext) to be handled without leaking details
into this module.
Callers implement `endpoint_path/0` and then call `AI.Endpoint.post_json/3`.
"""
@type headers :: list({String.t(), String.t()})
@type payload :: map()
@type http_status :: integer()
@type http_error :: {:http_error, {http_status, String.t()}}
@type transport_error :: {:transport_error, any()}
@type success :: {:ok, %{body: map(), headers: headers(), status: http_status()}}
@type response :: success | http_error | transport_error
@type endpoint :: module()
# Cap on retries per request. Each attempt waits min/max with the server's
# "try again in Xs" hint (when present), so 8 attempts spread out to
# ~30-60s in the worst case - enough to outwait a saturated 60s rolling
# TPM window. Smaller values (we had 3) gave up before the bucket
# drained on jumbo requests.
@retry_limit 8
@backoff_base_ms 200
@backoff_cap_ms 2_000
# ----------------------------------------------------------------------------
# Behaviour Callbacks
# ----------------------------------------------------------------------------
@callback endpoint_path() :: String.t()
@doc """
Classify a non-success HTTP/transport result.
This callback receives either an HTTP status + body (for HTTP errors) or a
transport reason (for transport errors). Providers should return one of:
- `:ok` — no retry; Endpoint returns the original result
- `{:retry, reason, wait_ms | nil}`— retry; optional provider-suggested delay (ms)
- `{:fail, reason, human}` — stop retrying; human-friendly message for logs
`headers` is currently `nil` for error cases as the lower-level HTTP client
does not expose them. This can be extended in the future without breaking
existing implementations.
"""
@callback endpoint_error_classify(
status :: integer | nil,
body :: binary | nil,
headers :: list | nil,
transport_reason :: term | nil
) ::
:ok
| {:retry, reason :: atom, wait_ms :: non_neg_integer | nil}
| {:fail, reason :: atom, human :: binary}
# ----------------------------------------------------------------------------
# API
# ----------------------------------------------------------------------------
@doc "The fixed retry limit for API calls."
@spec retry_limit() :: pos_integer()
def retry_limit, do: @retry_limit
@doc """
Perform a JSON POST request against the endpoint module's `endpoint_path/0`.
Retries up to `#{@retry_limit}` times when the server indicates throttling.
"""
@spec post_json(endpoint, headers, payload) :: response
def post_json(endpoint_module, headers, payload) when is_atom(endpoint_module) do
url = endpoint_module.endpoint_path()
do_post_json(endpoint_module, url, headers, payload, 1)
end
defp do_post_json(endpoint_module, url, headers, payload, attempt) do
result = Http.post_json(url, headers, payload)
model = model_from_payload(payload)
Store.APIUsage.record_for_model(model, result)
case result do
{:ok, %{body: _body} = payload} ->
Services.BgIndexingControl.note_success(model)
{:ok, payload}
{:http_error, {status, body}} ->
classify =
if function_exported?(endpoint_module, :endpoint_error_classify, 4) do
endpoint_module.endpoint_error_classify(status, body, nil, nil)
else
default_error_classify({:http_error, {status, body}}, nil)
end
handle_classification(
classify,
endpoint_module,
url,
headers,
payload,
attempt,
model,
{:http_error, {status, body}}
)
{:transport_error, reason} ->
classify =
if function_exported?(endpoint_module, :endpoint_error_classify, 4) do
endpoint_module.endpoint_error_classify(nil, nil, nil, reason)
else
default_error_classify(nil, reason)
end
handle_classification(
classify,
endpoint_module,
url,
headers,
payload,
attempt,
model,
{:transport_error, reason}
)
end
end
defp handle_classification(
classify,
endpoint_module,
url,
headers,
payload,
attempt,
model,
original_result
) do
case classify do
{:retry, _reason, wait_ms} ->
if attempt < retry_limit() do
Services.BgIndexingControl.note_throttle(model)
delay =
[wait_ms, usage_wait_ms(model), backoff_delay_ms(attempt)]
|> Enum.reject(&is_nil/1)
|> Enum.max()
UI.warn(
"[AI.Endpoint] Retrying, model=#{inspect(model)}, attempt #{attempt}/#{retry_limit()}, retrying in #{delay}ms"
)
maybe_sleep(delay)
do_post_json(endpoint_module, url, headers, payload, attempt + 1)
else
original_result
end
_ ->
original_result
end
end
@spec default_error_classify({:http_error, {http_status, String.t()}} | nil, any()) ::
:ok
| {:retry, reason :: atom, wait_ms :: non_neg_integer | nil}
| {:fail, reason :: atom, human :: binary}
defp default_error_classify({:http_error, {status, body}}, _transport_reason) do
cond do
# Retry throttled 429s by default when the body indicates OpenAI-style throttling
default_throttled?(status) and default_throttle_code?(body) ->
{:retry, :throttled, default_try_again_ms(body)}
status >= 500 ->
{:retry, :server_error, nil}
true ->
:ok
end
end
defp default_error_classify(nil, _transport_reason) do
# By default, do not retry transport errors; pass them through unchanged
:ok
end
@spec default_throttled?(integer() | nil) :: boolean()
defp default_throttled?(429), do: true
defp default_throttled?(_), do: false
@spec default_throttle_code?(binary()) :: boolean()
defp default_throttle_code?(body) do
with {:ok, %{"error" => %{"code" => code}}} <- Jason.decode(body) do
code in ["rate_limit_exceeded", "rate_limit"]
else
_ -> false
end
end
# OpenAI's rate-limit body uses either "try again in 250ms" or "try again
# in 1.566s" depending on the wait. Both formats may carry a decimal point
# (e.g. 1.566s, 1500.5ms). The retry scheduler is in ms.
@spec default_try_again_ms(binary()) :: non_neg_integer | nil
defp default_try_again_ms(body) do
cond do
match = Regex.run(~r/try\s+again\s+in\s+(\d+(?:\.\d+)?)\s*ms/i, body) ->
[_, ms_str] = match
max(1, ms_str |> parse_number() |> round())
match = Regex.run(~r/try\s+again\s+in\s+(\d+(?:\.\d+)?)\s*s/i, body) ->
[_, s_str] = match
max(1, (parse_number(s_str) * 1000) |> round())
true ->
nil
end
end
defp parse_number(str) do
case Float.parse(str) do
{n, _} -> n
:error -> 0.0
end
end
defp model_from_payload(payload) when is_map(payload) do
cond do
is_binary(Map.get(payload, :model)) -> Map.get(payload, :model)
is_binary(Map.get(payload, "model")) -> Map.get(payload, "model")
true -> nil
end
end
defp usage_wait_ms(nil), do: nil
defp usage_wait_ms(model) when is_binary(model) do
case Store.APIUsage.check(model) do
:ok -> 0
{:wait, ms} when is_integer(ms) and ms >= 0 -> ms
{:error, _} -> nil
end
end
# Exponential backoff with jitter (+/- 20%), capped at 2 seconds.
defp backoff_delay_ms(attempt) when attempt >= 1 do
jitter = 0.8 + :rand.uniform() * 0.4
@backoff_cap_ms
|> min(@backoff_base_ms * :math.pow(2, attempt - 1))
|> trunc()
|> then(&(&1 * jitter))
|> min(@backoff_cap_ms)
|> max(1)
|> round()
end
defp maybe_sleep(ms) do
if Services.Globals.get_env(:fnord, :http_retry_skip_sleep, false) do
:ok
else
Process.sleep(ms)
end
end
end