HTTP: Implement max request limits

config/config.exs

@@ -648,7 +648,8 @@
 
 config :pleroma, :connections_pool,
   reclaim_multiplier: 0.1,
-  checkin_timeout: 250,
+  connection_acquisition_wait: 250,
+  connection_acquisition_retries: 5,
   max_connections: 250,
   max_idle_time: 30_000,
   retry: 1,
@@ -658,23 +659,19 @@
 config :pleroma, :pools,
   federation: [
     size: 50,
-    max_overflow: 10,
+    max_waiting: 10
-    timeout: 150_000
   ],
   media: [
     size: 50,
-    max_overflow: 10,
+    max_waiting: 10
-    timeout: 150_000
   ],
   upload: [
     size: 25,
-    max_overflow: 5,
+    max_waiting: 5
-    timeout: 300_000
   ],
   default: [
     size: 10,
-    max_overflow: 2,
+    max_waiting: 2
-    timeout: 10_000
   ]
 
 config :pleroma, :hackney_pools,

lib/pleroma/application.ex

@@ -244,7 +244,8 @@ defp http_children(Tesla.Adapter.Hackney, _) do
   end
 
   defp http_children(Tesla.Adapter.Gun, _) do
-    Pleroma.Gun.ConnectionPool.children()
+    Pleroma.Gun.ConnectionPool.children() ++
+      [{Task, &Pleroma.HTTP.AdapterHelper.Gun.limiter_setup/0}]
   end
 
   defp http_children(_, _), do: []

lib/pleroma/http/adapter_helper/gun.ex

@@ -49,4 +49,25 @@ def get_conn(uri, opts) do
       err -> err
     end
   end
+
+  @prefix Pleroma.Gun.ConnectionPool
+  def limiter_setup do
+    wait = Pleroma.Config.get([:connections_pool, :connection_acquisition_wait])
+    retries = Pleroma.Config.get([:connections_pool, :connection_acquisition_retries])
+
+    :pools
+    |> Pleroma.Config.get([])
+    |> Enum.each(fn {name, opts} ->
+      max_running = Keyword.get(opts, :size, 50)
+      max_waiting = Keyword.get(opts, :max_waiting, 10)
+
+      :ok =
+        ConcurrentLimiter.new(:"#{@prefix}.#{name}", max_running, max_waiting,
+          wait: wait,
+          max_retries: retries
+        )
+    end)
+
+    :ok
+  end
 end

lib/pleroma/http/http.ex

@@ -71,7 +71,13 @@ def request(method, url, body, headers, options) when is_binary(url) do
         adapter = Application.get_env(:tesla, :adapter)
         client = Tesla.client([Pleroma.HTTP.Middleware.FollowRedirects], adapter)
 
+        maybe_limit(
+          fn ->
             request(client, request)
+          end,
+          adapter,
+          adapter_opts
+        )
 
       # Connection release is handled in a custom FollowRedirects middleware
       err ->
@@ -92,4 +98,13 @@ defp build_request(method, headers, options, url, body, params) do
     |> Builder.add_param(:query, :query, params)
     |> Builder.convert_to_keyword()
   end
+
+  @prefix Pleroma.Gun.ConnectionPool
+  defp maybe_limit(fun, Tesla.Adapter.Gun, opts) do
+    ConcurrentLimiter.limit(:"#{@prefix}.#{opts[:pool] || :default}", fun)
+  end
+
+  defp maybe_limit(fun, _, _) do
+    fun.()
+  end
 end

mix.exs

@@ -191,6 +191,9 @@ defp deps do
       {:plug_static_index_html, "~> 1.0.0"},
       {:excoveralls, "~> 0.12.1", only: :test},
       {:flake_id, "~> 0.1.0"},
+      {:concurrent_limiter,
+       git: "https://git.pleroma.social/pleroma/elixir-libraries/concurrent_limiter",
+       ref: "8eee96c6ba39b9286ec44c51c52d9f2758951365"},
       {:remote_ip,
        git: "https://git.pleroma.social/pleroma/remote_ip.git",
        ref: "b647d0deecaa3acb140854fe4bda5b7e1dc6d1c8"},

mix.lock

@@ -15,6 +15,7 @@
   "certifi": {:hex, :certifi, "2.5.2", "b7cfeae9d2ed395695dd8201c57a2d019c0c43ecaf8b8bcb9320b40d6662f340", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm", "3b3b5f36493004ac3455966991eaf6e768ce9884693d9968055aeeeb1e575040"},
   "combine": {:hex, :combine, "0.10.0", "eff8224eeb56498a2af13011d142c5e7997a80c8f5b97c499f84c841032e429f", [:mix], [], "hexpm", "1b1dbc1790073076580d0d1d64e42eae2366583e7aecd455d1215b0d16f2451b"},
   "comeonin": {:hex, :comeonin, "5.3.1", "7fe612b739c78c9c1a75186ef2d322ce4d25032d119823269d0aa1e2f1e20025", [:mix], [], "hexpm", "d6222483060c17f0977fad1b7401ef0c5863c985a64352755f366aee3799c245"},
+  "concurrent_limiter": {:git, "https://git.pleroma.social/pleroma/elixir-libraries/concurrent_limiter", "8eee96c6ba39b9286ec44c51c52d9f2758951365", [ref: "8eee96c6ba39b9286ec44c51c52d9f2758951365"]},
   "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm", "4a0850c9be22a43af9920a71ab17c051f5f7d45c209e40269a1938832510e4d9"},
   "cors_plug": {:hex, :cors_plug, "1.5.2", "72df63c87e4f94112f458ce9d25800900cc88608c1078f0e4faddf20933eda6e", [:mix], [{:plug, "~> 1.3 or ~> 1.4 or ~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "9af027d20dc12dd0c4345a6b87247e0c62965871feea0bfecf9764648b02cc69"},
   "cowboy": {:hex, :cowboy, "2.7.0", "91ed100138a764355f43316b1d23d7ff6bdb0de4ea618cb5d8677c93a7a2f115", [:rebar3], [{:cowlib, "~> 2.8.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.7.1", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "04fd8c6a39edc6aaa9c26123009200fc61f92a3a94f3178c527b70b767c6e605"},