diff --git a/config/runtime.exs b/config/runtime.exs index 3fe2fbb9..f298db15 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -177,7 +177,34 @@ config :wanderer_app, ], extra_characters_50: map_subscription_extra_characters_50_price, extra_hubs_10: map_subscription_extra_hubs_10_price - } + }, + # Finch pool configuration - separate pools for different services + # ESI Character Tracking pool - high capacity for bulk character operations + # With 30+ TrackerPools × ~100 concurrent tasks, need large pool + finch_esi_character_pool_size: + System.get_env("WANDERER_FINCH_ESI_CHARACTER_POOL_SIZE", "200") |> String.to_integer(), + finch_esi_character_pool_count: + System.get_env("WANDERER_FINCH_ESI_CHARACTER_POOL_COUNT", "4") |> String.to_integer(), + # ESI General pool - standard capacity for general ESI operations + finch_esi_general_pool_size: + System.get_env("WANDERER_FINCH_ESI_GENERAL_POOL_SIZE", "50") |> String.to_integer(), + finch_esi_general_pool_count: + System.get_env("WANDERER_FINCH_ESI_GENERAL_POOL_COUNT", "4") |> String.to_integer(), + # Webhooks pool - isolated from ESI rate limits + finch_webhooks_pool_size: + System.get_env("WANDERER_FINCH_WEBHOOKS_POOL_SIZE", "25") |> String.to_integer(), + finch_webhooks_pool_count: + System.get_env("WANDERER_FINCH_WEBHOOKS_POOL_COUNT", "2") |> String.to_integer(), + # Default pool - everything else (email, license manager, etc.) + finch_default_pool_size: + System.get_env("WANDERER_FINCH_DEFAULT_POOL_SIZE", "25") |> String.to_integer(), + finch_default_pool_count: + System.get_env("WANDERER_FINCH_DEFAULT_POOL_COUNT", "2") |> String.to_integer(), + # Character tracker concurrency settings + # Location updates need high concurrency for <2s response with 3000+ characters + location_concurrency: + System.get_env("WANDERER_LOCATION_CONCURRENCY", "#{System.schedulers_online() * 12}") + |> String.to_integer() config :ueberauth, Ueberauth, providers: [ diff --git a/lib/wanderer_app/application.ex b/lib/wanderer_app/application.ex index ec1bf3f3..728e7672 100644 --- a/lib/wanderer_app/application.ex +++ b/lib/wanderer_app/application.ex @@ -16,15 +16,48 @@ defmodule WandererApp.Application do WandererApp.Vault, WandererApp.Repo, {Phoenix.PubSub, name: WandererApp.PubSub, adapter_name: Phoenix.PubSub.PG2}, + # Multiple Finch pools for different services to prevent connection pool exhaustion + # ESI Character Tracking pool - high capacity for bulk character operations + { + Finch, + name: WandererApp.Finch.ESI.CharacterTracking, + pools: %{ + default: [ + size: Application.get_env(:wanderer_app, :finch_esi_character_pool_size, 100), + count: Application.get_env(:wanderer_app, :finch_esi_character_pool_count, 4) + ] + } + }, + # ESI General pool - standard capacity for general ESI operations + { + Finch, + name: WandererApp.Finch.ESI.General, + pools: %{ + default: [ + size: Application.get_env(:wanderer_app, :finch_esi_general_pool_size, 50), + count: Application.get_env(:wanderer_app, :finch_esi_general_pool_count, 4) + ] + } + }, + # Webhooks pool - isolated from ESI rate limits + { + Finch, + name: WandererApp.Finch.Webhooks, + pools: %{ + default: [ + size: Application.get_env(:wanderer_app, :finch_webhooks_pool_size, 25), + count: Application.get_env(:wanderer_app, :finch_webhooks_pool_count, 2) + ] + } + }, + # Default pool - everything else (email, license manager, etc.) { Finch, name: WandererApp.Finch, pools: %{ default: [ - # number of connections per pool - size: 50, - # number of pools (so total 50 connections) - count: 4 + size: Application.get_env(:wanderer_app, :finch_default_pool_size, 25), + count: Application.get_env(:wanderer_app, :finch_default_pool_count, 2) ] } }, diff --git a/lib/wanderer_app/character.ex b/lib/wanderer_app/character.ex index ea6672d7..caab0d63 100644 --- a/lib/wanderer_app/character.ex +++ b/lib/wanderer_app/character.ex @@ -331,7 +331,7 @@ defmodule WandererApp.Character do do: {:ok, Enum.map(eve_ids, fn eve_id -> - Task.async(fn -> apply(WandererApp.Esi.ApiClient, method, [eve_id]) end) + Task.async(fn -> apply(WandererApp.Esi, method, [eve_id]) end) end) # 145000 == Timeout in milliseconds |> Enum.map(fn task -> Task.await(task, 145_000) end) diff --git a/lib/wanderer_app/character/tracker_pool.ex b/lib/wanderer_app/character/tracker_pool.ex index cea846c1..5f64638f 100644 --- a/lib/wanderer_app/character/tracker_pool.ex +++ b/lib/wanderer_app/character/tracker_pool.ex @@ -8,7 +8,8 @@ defmodule WandererApp.Character.TrackerPool do :tracked_ids, :uuid, :characters, - server_online: false + server_online: false, + last_location_duration: 0 ] @name __MODULE__ @@ -23,6 +24,12 @@ defmodule WandererApp.Character.TrackerPool do @update_info_interval :timer.minutes(2) @update_wallet_interval :timer.minutes(10) + # Per-operation concurrency limits + # Location updates are critical and need high concurrency (100 chars in ~200ms) + @location_concurrency Application.compile_env(:wanderer_app, :location_concurrency, System.schedulers_online() * 12) + # Other operations can use lower concurrency + @standard_concurrency System.schedulers_online() * 2 + @logger Application.compile_env(:wanderer_app, :logger) def new(), do: __struct__() @@ -106,14 +113,23 @@ defmodule WandererApp.Character.TrackerPool do "server_status" ) - Process.send_after(self(), :update_online, 100) - Process.send_after(self(), :update_location, 300) - Process.send_after(self(), :update_ship, 500) - Process.send_after(self(), :update_info, 1500) + # Stagger pool startups to distribute load across multiple pools + # Critical location updates get minimal stagger (0-500ms) + # Other operations get wider stagger (0-10s) to reduce thundering herd + location_stagger = :rand.uniform(500) + online_stagger = :rand.uniform(10_000) + ship_stagger = :rand.uniform(10_000) + info_stagger = :rand.uniform(60_000) + + Process.send_after(self(), :update_online, 100 + online_stagger) + Process.send_after(self(), :update_location, 300 + location_stagger) + Process.send_after(self(), :update_ship, 500 + ship_stagger) + Process.send_after(self(), :update_info, 1500 + info_stagger) Process.send_after(self(), :check_offline_characters, @check_offline_characters_interval) if WandererApp.Env.wallet_tracking_enabled?() do - Process.send_after(self(), :update_wallet, 1000) + wallet_stagger = :rand.uniform(120_000) + Process.send_after(self(), :update_wallet, 1000 + wallet_stagger) end {:noreply, state} @@ -163,7 +179,7 @@ defmodule WandererApp.Character.TrackerPool do fn character_id -> WandererApp.Character.Tracker.update_online(character_id) end, - max_concurrency: System.schedulers_online() * 4, + max_concurrency: @standard_concurrency, on_timeout: :kill_task, timeout: :timer.seconds(5) ) @@ -226,7 +242,7 @@ defmodule WandererApp.Character.TrackerPool do WandererApp.Character.Tracker.check_offline(character_id) end, timeout: :timer.seconds(15), - max_concurrency: System.schedulers_online() * 4, + max_concurrency: @standard_concurrency, on_timeout: :kill_task ) |> Enum.each(fn @@ -254,26 +270,52 @@ defmodule WandererApp.Character.TrackerPool do ) do Process.send_after(self(), :update_location, @update_location_interval) + start_time = System.monotonic_time(:millisecond) + try do characters |> Task.async_stream( fn character_id -> WandererApp.Character.Tracker.update_location(character_id) end, - max_concurrency: System.schedulers_online() * 4, + max_concurrency: @location_concurrency, on_timeout: :kill_task, timeout: :timer.seconds(5) ) |> Enum.each(fn _result -> :ok end) + + # Emit telemetry for location update performance + duration = System.monotonic_time(:millisecond) - start_time + + :telemetry.execute( + [:wanderer_app, :tracker_pool, :location_update], + %{duration: duration, character_count: length(characters)}, + %{pool_uuid: state.uuid} + ) + + # Warn if location updates are falling behind (taking > 800ms for 100 chars) + if duration > 800 do + Logger.warning( + "[Tracker Pool] Location updates falling behind: #{duration}ms for #{length(characters)} chars (pool: #{state.uuid})" + ) + + :telemetry.execute( + [:wanderer_app, :tracker_pool, :location_lag], + %{duration: duration, character_count: length(characters)}, + %{pool_uuid: state.uuid} + ) + end + + {:noreply, %{state | last_location_duration: duration}} rescue e -> Logger.error(""" [Tracker Pool] update_location => exception: #{Exception.message(e)} #{Exception.format_stacktrace(__STACKTRACE__)} """) - end - {:noreply, state} + {:noreply, state} + end end def handle_info( @@ -289,32 +331,48 @@ defmodule WandererApp.Character.TrackerPool do :update_ship, %{ characters: characters, - server_online: true + server_online: true, + last_location_duration: location_duration } = state ) do Process.send_after(self(), :update_ship, @update_ship_interval) - try do - characters - |> Task.async_stream( - fn character_id -> - WandererApp.Character.Tracker.update_ship(character_id) - end, - max_concurrency: System.schedulers_online() * 4, - on_timeout: :kill_task, - timeout: :timer.seconds(5) + # Backpressure: Skip ship updates if location updates are falling behind + if location_duration > 1000 do + Logger.debug( + "[Tracker Pool] Skipping ship update due to location lag (#{location_duration}ms)" ) - |> Enum.each(fn _result -> :ok end) - rescue - e -> - Logger.error(""" - [Tracker Pool] update_ship => exception: #{Exception.message(e)} - #{Exception.format_stacktrace(__STACKTRACE__)} - """) - end - {:noreply, state} + :telemetry.execute( + [:wanderer_app, :tracker_pool, :ship_skipped], + %{count: 1}, + %{pool_uuid: state.uuid, reason: :location_lag} + ) + + {:noreply, state} + else + try do + characters + |> Task.async_stream( + fn character_id -> + WandererApp.Character.Tracker.update_ship(character_id) + end, + max_concurrency: @standard_concurrency, + on_timeout: :kill_task, + timeout: :timer.seconds(5) + ) + |> Enum.each(fn _result -> :ok end) + rescue + e -> + Logger.error(""" + [Tracker Pool] update_ship => exception: #{Exception.message(e)} + #{Exception.format_stacktrace(__STACKTRACE__)} + """) + end + + {:noreply, state} + end end def handle_info( @@ -330,35 +388,51 @@ defmodule WandererApp.Character.TrackerPool do :update_info, %{ characters: characters, - server_online: true + server_online: true, + last_location_duration: location_duration } = state ) do Process.send_after(self(), :update_info, @update_info_interval) - try do - characters - |> Task.async_stream( - fn character_id -> - WandererApp.Character.Tracker.update_info(character_id) - end, - timeout: :timer.seconds(15), - max_concurrency: System.schedulers_online() * 4, - on_timeout: :kill_task + # Backpressure: Skip info updates if location updates are severely falling behind + if location_duration > 1500 do + Logger.debug( + "[Tracker Pool] Skipping info update due to location lag (#{location_duration}ms)" ) - |> Enum.each(fn - {:ok, _result} -> :ok - error -> Logger.error("Error in update_info: #{inspect(error)}") - end) - rescue - e -> - Logger.error(""" - [Tracker Pool] update_info => exception: #{Exception.message(e)} - #{Exception.format_stacktrace(__STACKTRACE__)} - """) - end - {:noreply, state} + :telemetry.execute( + [:wanderer_app, :tracker_pool, :info_skipped], + %{count: 1}, + %{pool_uuid: state.uuid, reason: :location_lag} + ) + + {:noreply, state} + else + try do + characters + |> Task.async_stream( + fn character_id -> + WandererApp.Character.Tracker.update_info(character_id) + end, + timeout: :timer.seconds(15), + max_concurrency: @standard_concurrency, + on_timeout: :kill_task + ) + |> Enum.each(fn + {:ok, _result} -> :ok + error -> Logger.error("Error in update_info: #{inspect(error)}") + end) + rescue + e -> + Logger.error(""" + [Tracker Pool] update_info => exception: #{Exception.message(e)} + #{Exception.format_stacktrace(__STACKTRACE__)} + """) + end + + {:noreply, state} + end end def handle_info( @@ -387,7 +461,7 @@ defmodule WandererApp.Character.TrackerPool do WandererApp.Character.Tracker.update_wallet(character_id) end, timeout: :timer.minutes(5), - max_concurrency: System.schedulers_online() * 4, + max_concurrency: @standard_concurrency, on_timeout: :kill_task ) |> Enum.each(fn diff --git a/lib/wanderer_app/esi/api_client.ex b/lib/wanderer_app/esi/api_client.ex index c03180ae..a895aa6c 100644 --- a/lib/wanderer_app/esi/api_client.ex +++ b/lib/wanderer_app/esi/api_client.ex @@ -17,6 +17,17 @@ defmodule WandererApp.Esi.ApiClient do @logger Application.compile_env(:wanderer_app, :logger) + # Pool selection for different operation types + # Character tracking operations use dedicated high-capacity pool + @character_tracking_pool WandererApp.Finch.ESI.CharacterTracking + # General ESI operations use standard pool + @general_pool WandererApp.Finch.ESI.General + + # Helper function to get Req options with appropriate Finch pool + defp req_options_for_pool(pool) do + [base_url: "https://esi.evetech.net", finch: pool] + end + def get_server_status, do: do_get("/status", [], @cache_opts) def set_autopilot_waypoint(add_to_beginning, clear_other_waypoints, destination_id, opts \\ []), @@ -38,10 +49,13 @@ defmodule WandererApp.Esi.ApiClient do do: do_post_esi( "/characters/affiliation/", - json: character_eve_ids, - params: %{ - datasource: "tranquility" - } + [ + json: character_eve_ids, + params: %{ + datasource: "tranquility" + } + ], + @character_tracking_pool ) def get_routes_custom(hubs, origin, params), @@ -289,14 +303,18 @@ defmodule WandererApp.Esi.ApiClient do character_id = opts |> Keyword.get(:character_id, nil) + # Use character tracking pool for character operations + pool = @character_tracking_pool + if not is_access_token_expired?(character_id) do do_get( path, auth_opts, - opts |> with_refresh_token() + opts |> with_refresh_token(), + pool ) else - do_get_retry(path, auth_opts, opts |> with_refresh_token()) + do_get_retry(path, auth_opts, opts |> with_refresh_token(), :forbidden, pool) end end @@ -330,19 +348,19 @@ defmodule WandererApp.Esi.ApiClient do defp with_cache_opts(opts), do: opts |> Keyword.merge(@cache_opts) |> Keyword.merge(cache_dir: System.tmp_dir!()) - defp do_get(path, api_opts \\ [], opts \\ []) do + defp do_get(path, api_opts \\ [], opts \\ [], pool \\ @general_pool) do case Cachex.get(:api_cache, path) do {:ok, cached_data} when not is_nil(cached_data) -> {:ok, cached_data} _ -> - do_get_request(path, api_opts, opts) + do_get_request(path, api_opts, opts, pool) end end - defp do_get_request(path, api_opts \\ [], opts \\ []) do + defp do_get_request(path, api_opts \\ [], opts \\ [], pool \\ @general_pool) do try do - @req_esi_options + req_options_for_pool(pool) |> Req.new() |> Req.get( api_opts @@ -433,12 +451,48 @@ defmodule WandererApp.Esi.ApiClient do {:ok, %{status: status, headers: headers}} -> {:error, "Unexpected status: #{status}"} - {:error, _reason} -> + {:error, %Mint.TransportError{reason: :timeout}} -> + # Emit telemetry for pool timeout + :telemetry.execute( + [:wanderer_app, :finch, :pool_timeout], + %{count: 1}, + %{method: "GET", path: path, pool: pool} + ) + + {:error, :pool_timeout} + + {:error, reason} -> + # Check if this is a Finch pool error + if is_exception(reason) and Exception.message(reason) =~ "unable to provide a connection" do + :telemetry.execute( + [:wanderer_app, :finch, :pool_exhausted], + %{count: 1}, + %{method: "GET", path: path, pool: pool} + ) + end + {:error, "Request failed"} end rescue e -> - Logger.error(Exception.message(e)) + error_msg = Exception.message(e) + + # Emit telemetry for pool exhaustion errors + if error_msg =~ "unable to provide a connection" do + :telemetry.execute( + [:wanderer_app, :finch, :pool_exhausted], + %{count: 1}, + %{method: "GET", path: path, pool: pool} + ) + + Logger.error("FINCH_POOL_EXHAUSTED: #{error_msg}", + method: "GET", + path: path, + pool: inspect(pool) + ) + else + Logger.error(error_msg) + end {:error, "Request failed"} end @@ -527,13 +581,13 @@ defmodule WandererApp.Esi.ApiClient do end end - defp do_post_esi(url, opts) do + defp do_post_esi(url, opts, pool \\ @general_pool) do try do req_opts = (opts |> with_user_agent_opts() |> Keyword.merge(@retry_opts)) ++ [params: opts[:params] || []] - Req.new(@req_esi_options ++ req_opts) + Req.new(req_options_for_pool(pool) ++ req_opts) |> Req.post(url: url) |> case do {:ok, %{status: status, body: body}} when status in [200, 201] -> @@ -611,18 +665,54 @@ defmodule WandererApp.Esi.ApiClient do {:ok, %{status: status}} -> {:error, "Unexpected status: #{status}"} + {:error, %Mint.TransportError{reason: :timeout}} -> + # Emit telemetry for pool timeout + :telemetry.execute( + [:wanderer_app, :finch, :pool_timeout], + %{count: 1}, + %{method: "POST_ESI", path: url, pool: pool} + ) + + {:error, :pool_timeout} + {:error, reason} -> + # Check if this is a Finch pool error + if is_exception(reason) and Exception.message(reason) =~ "unable to provide a connection" do + :telemetry.execute( + [:wanderer_app, :finch, :pool_exhausted], + %{count: 1}, + %{method: "POST_ESI", path: url, pool: pool} + ) + end + {:error, reason} end rescue e -> - @logger.error(Exception.message(e)) + error_msg = Exception.message(e) + + # Emit telemetry for pool exhaustion errors + if error_msg =~ "unable to provide a connection" do + :telemetry.execute( + [:wanderer_app, :finch, :pool_exhausted], + %{count: 1}, + %{method: "POST_ESI", path: url, pool: pool} + ) + + @logger.error("FINCH_POOL_EXHAUSTED: #{error_msg}", + method: "POST_ESI", + path: url, + pool: inspect(pool) + ) + else + @logger.error(error_msg) + end {:error, "Request failed"} end end - defp do_get_retry(path, api_opts, opts, status \\ :forbidden) do + defp do_get_retry(path, api_opts, opts, status \\ :forbidden, pool \\ @general_pool) do refresh_token? = opts |> Keyword.get(:refresh_token?, false) retry_count = opts |> Keyword.get(:retry_count, 0) character_id = opts |> Keyword.get(:character_id, nil) @@ -637,7 +727,8 @@ defmodule WandererApp.Esi.ApiClient do do_get( path, api_opts |> Keyword.merge(auth_opts), - opts |> Keyword.merge(retry_count: retry_count + 1) + opts |> Keyword.merge(retry_count: retry_count + 1), + pool ) {:error, _error} -> diff --git a/lib/wanderer_app/external_events/webhook_dispatcher.ex b/lib/wanderer_app/external_events/webhook_dispatcher.ex index 05a455eb..675677f7 100644 --- a/lib/wanderer_app/external_events/webhook_dispatcher.ex +++ b/lib/wanderer_app/external_events/webhook_dispatcher.ex @@ -292,7 +292,7 @@ defmodule WandererApp.ExternalEvents.WebhookDispatcher do request = Finch.build(:post, url, headers, payload) - case Finch.request(request, WandererApp.Finch, timeout: 30_000) do + case Finch.request(request, WandererApp.Finch.Webhooks, timeout: 30_000) do {:ok, %Finch.Response{status: status}} -> {:ok, status} diff --git a/lib/wanderer_app_web/telemetry.ex b/lib/wanderer_app_web/telemetry.ex index 8d1864f5..c5de1e7d 100644 --- a/lib/wanderer_app_web/telemetry.ex +++ b/lib/wanderer_app_web/telemetry.ex @@ -78,7 +78,36 @@ defmodule WandererAppWeb.Telemetry do summary("vm.memory.total", unit: {:byte, :kilobyte}), summary("vm.total_run_queue_lengths.total"), summary("vm.total_run_queue_lengths.cpu"), - summary("vm.total_run_queue_lengths.io") + summary("vm.total_run_queue_lengths.io"), + + # Finch Pool Metrics + counter("wanderer_app.finch.pool_exhausted.count", + tags: [:pool, :method], + description: "Count of Finch pool exhaustion errors" + ), + counter("wanderer_app.finch.pool_timeout.count", + tags: [:pool, :method], + description: "Count of Finch pool timeout errors" + ), + + # Character Tracker Pool Metrics + summary("wanderer_app.tracker_pool.location_update.duration", + unit: :millisecond, + tags: [:pool_uuid], + description: "Time taken to update all character locations in a pool" + ), + counter("wanderer_app.tracker_pool.location_lag.count", + tags: [:pool_uuid], + description: "Count of location updates falling behind (>800ms)" + ), + counter("wanderer_app.tracker_pool.ship_skipped.count", + tags: [:pool_uuid, :reason], + description: "Count of ship updates skipped due to backpressure" + ), + counter("wanderer_app.tracker_pool.info_skipped.count", + tags: [:pool_uuid, :reason], + description: "Count of info updates skipped due to backpressure" + ) ] end