Merge pull request #268 from oliveigah/main

feat: implement handle_cancelled/2 pool callback

lib/finch/http1/pool.ex

@@ -270,6 +270,15 @@ defmodule Finch.HTTP1.Pool do
     {:ok, pool_state}
   end
 
+  @impl NimblePool
+  def handle_cancelled(:checked_out, pool_state) do
+    {_name, _shp, _pool_idx, metric_ref, _opts} = pool_state
+    PoolMetrics.maybe_add(metric_ref, in_use_connections: -1)
+    :ok
+  end
+
+  def handle_cancelled(:queued, _pool_state), do: :ok
+
   defp transfer_if_open(conn, state, {pid, _} = from) do
     if Conn.open?(conn) do
       if state == :fresh do

mix.exs

@@ -35,7 +35,7 @@ defmodule Finch.MixProject do
     [
       {:mint, "~> 1.3"},
       {:castore, "~> 0.1 or ~> 1.0"},
-      {:nimble_pool, "~> 0.2.6 or ~> 1.0"},
+      {:nimble_pool, "~> 1.1"},
       {:nimble_options, "~> 0.4 or ~> 1.0"},
       {:telemetry, "~> 0.4 or ~> 1.0"},
       {:mime, "~> 1.0 or ~> 2.0"},
@@ -45,7 +45,8 @@ defmodule Finch.MixProject do
       {:bypass, "~> 2.0", only: :test},
       {:cowboy, "~> 2.7", only: [:dev, :test]},
       {:plug_cowboy, "~> 2.0", only: [:dev, :test]},
-      {:x509, "~> 0.8", only: [:dev, :test]}
+      {:x509, "~> 0.8", only: [:dev, :test]},
+      {:mimic, "~> 1.7", only: :test}
     ]
   end
 

mix.lock

@@ -17,10 +17,11 @@
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.1", "cc9e3ca312f1cfeccc572b37a09980287e243648108384b97ff2b76e505c3555", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "e127a341ad1b209bd80f7bd1620a15693a9908ed780c3b763bccf7d200c767c6"},
   "makeup_erlang": {:hex, :makeup_erlang, "0.1.3", "d684f4bac8690e70b06eb52dad65d26de2eefa44cd19d64a8095e1417df7c8fd", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "b78dc853d2e670ff6390b605d807263bf606da3c82be37f9d7f68635bd886fc9"},
   "mime": {:hex, :mime, "2.0.5", "dc34c8efd439abe6ae0343edbb8556f4d63f178594894720607772a041b04b02", [:mix], [], "hexpm", "da0d64a365c45bc9935cc5c8a7fc5e49a0e0f9932a761c55d6c52b142780a05c"},
+  "mimic": {:hex, :mimic, "1.7.4", "cd2772ffbc9edefe964bc668bfd4059487fa639a5b7f1cbdf4fd22946505aa4f", [:mix], [], "hexpm", "437c61041ecf8a7fae35763ce89859e4973bb0666e6ce76d75efc789204447c3"},
   "mint": {:hex, :mint, "1.5.2", "4805e059f96028948870d23d7783613b7e6b0e2fb4e98d720383852a760067fd", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "d77d9e9ce4eb35941907f1d3df38d8f750c357865353e21d335bdcdf6d892a02"},
   "nimble_options": {:hex, :nimble_options, "1.1.0", "3b31a57ede9cb1502071fade751ab0c7b8dbe75a9a4c2b5bbb0943a690b63172", [:mix], [], "hexpm", "8bbbb3941af3ca9acc7835f5655ea062111c9c27bcac53e004460dfd19008a99"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
-  "nimble_pool": {:hex, :nimble_pool, "1.0.0", "5eb82705d138f4dd4423f69ceb19ac667b3b492ae570c9f5c900bb3d2f50a847", [:mix], [], "hexpm", "80be3b882d2d351882256087078e1b1952a28bf98d0a287be87e4a24a710b67a"},
+  "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},
   "plug": {:hex, :plug, "1.15.2", "94cf1fa375526f30ff8770837cb804798e0045fd97185f0bb9e5fcd858c792a3", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "02731fa0c2dcb03d8d21a1d941bdbbe99c2946c0db098eee31008e04c6283615"},
   "plug_cowboy": {:hex, :plug_cowboy, "2.6.1", "9a3bbfceeb65eff5f39dab529e5cd79137ac36e913c02067dba3963a26efe9b2", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "de36e1a21f451a18b790f37765db198075c25875c64834bcc82d90b309eb6613"},
   "plug_crypto": {:hex, :plug_crypto, "2.0.0", "77515cc10af06645abbfb5e6ad7a3e9714f805ae118fa1a70205f80d2d70fe73", [:mix], [], "hexpm", "53695bae57cc4e54566d993eb01074e4d894b65a3766f1c43e2c61a1b0f45ea9"},

test/finch/http1/pool_metrics_test.exs

@@ -1,5 +1,6 @@
 defmodule Finch.HTTP1.PoolMetricsTest do
   use FinchCase, async: true
+  use Mimic
 
   alias Finch.HTTP1.PoolMetrics
   alias Finch.PoolManager
@@ -158,6 +159,141 @@ defmodule Finch.HTTP1.PoolMetricsTest do
     assert p1_in_use_conns + p2_in_use_conns == 0
   end
 
+  test "get pool status with not reusable connections", %{bypass: bypass, finch_name: finch_name} do
+    start_supervised!(
+      {Finch,
+       name: finch_name,
+       pools: %{
+         default: [
+           protocols: [:http1],
+           start_pool_metrics?: true,
+           conn_max_idle_time: 1,
+           size: 10
+         ]
+       }}
+    )
+
+    shp = shp_from_bypass(bypass)
+
+    parent = self()
+
+    Bypass.expect(bypass, "GET", "/", fn conn ->
+      ["number", number] = String.split(conn.query_string, "=")
+      send(parent, {:ping_bypass, number})
+      Process.sleep(:timer.seconds(1))
+      Plug.Conn.send_resp(conn, 200, "OK")
+    end)
+
+    refs =
+      Enum.map(1..8, fn i ->
+        ref =
+          Finch.build(:get, endpoint(bypass, "?number=#{i}"))
+          |> Finch.async_request(finch_name)
+
+        {ref, i}
+      end)
+
+    assert_receive {:ping_bypass, "8"}, 500
+
+    assert {:ok,
+            [
+              %PoolMetrics{
+                pool_index: 1,
+                pool_size: 10,
+                available_connections: 2,
+                in_use_connections: 8
+              }
+            ]} = Finch.get_pool_status(finch_name, shp)
+
+    Enum.each(refs, fn {req_ref, _number} -> assert_receive({^req_ref, {:status, 200}}, 2000) end)
+
+    wait_connection_checkin()
+
+    assert {:ok,
+            [
+              %PoolMetrics{
+                pool_index: 1,
+                pool_size: 10,
+                available_connections: 10,
+                in_use_connections: 0
+              }
+            ]} = Finch.get_pool_status(finch_name, shp)
+
+    refs =
+      Enum.map(1..8, fn i ->
+        ref =
+          Finch.build(:get, endpoint(bypass, "?number=#{i}"))
+          |> Finch.async_request(finch_name)
+
+        {ref, i}
+      end)
+
+    assert_receive {:ping_bypass, "8"}, 500
+
+    assert {:ok,
+            [
+              %PoolMetrics{
+                pool_index: 1,
+                pool_size: 10,
+                available_connections: 2,
+                in_use_connections: 8
+              }
+            ]} = Finch.get_pool_status(finch_name, shp)
+
+    Enum.each(refs, fn {req_ref, _number} -> assert_receive({^req_ref, {:status, 200}}, 2000) end)
+
+    wait_connection_checkin()
+
+    assert {:ok,
+            [
+              %PoolMetrics{
+                pool_index: 1,
+                pool_size: 10,
+                available_connections: 10,
+                in_use_connections: 0
+              }
+            ]} = Finch.get_pool_status(finch_name, shp)
+  end
+
+  test "get pool status with raise before checkin", %{finch_name: finch_name} do
+    stub(Mint.HTTP, :request, fn _, _, _, _, _ ->
+      raise "unexpected error"
+    end)
+
+    start_supervised!(
+      {Finch,
+       name: finch_name,
+       pools: %{
+         default: [
+           protocols: [:http1],
+           start_pool_metrics?: true,
+           size: 10
+         ]
+       }}
+    )
+
+    url = "http://raise.com"
+    shp = {:http, "raise.com", 80}
+
+    Enum.map(1..20, fn _idx ->
+      assert_raise(RuntimeError, fn ->
+        Finch.build(:get, url) |> Finch.request(finch_name)
+      end)
+    end)
+
+    wait_connection_checkin()
+
+    assert {:ok,
+            [
+              %PoolMetrics{
+                pool_index: 1,
+                pool_size: 10,
+                available_connections: 10,
+                in_use_connections: 0
+              }
+            ]} = Finch.get_pool_status(finch_name, shp)
+  end
+
   defp shp_from_bypass(bypass), do: {:http, "localhost", bypass.port}
 
   defp wait_connection_checkin(), do: Process.sleep(5)

test/test_helper.exs

@@ -5,6 +5,8 @@
 Finch.HTTP2Server.start(port)
 Application.put_env(:finch, :test_https_h2_url, "https://localhost:#{port}")
 
+Mimic.copy(Mint.HTTP)
+
 ExUnit.start()
 Application.ensure_all_started(:bypass)