feat: add in use verification on handle ping

Author: oliveigah

lib/finch/http1/pool.ex

@@ -176,33 +176,33 @@ defmodule Finch.HTTP1.Pool do
     # to determine the correct pool module to use to make the request
     {:ok, _} = Registry.register(registry, shp, __MODULE__)
 
-    {:ok, {registry, shp, pool_idx, metric_ref, opts, System.monotonic_time(:millisecond)}}
+    {:ok, {registry, shp, pool_idx, metric_ref, opts, init_activity_info()}}
   end
 
   @impl NimblePool
   def init_worker(
-        {_name, {scheme, host, port}, _pool_idx, _metric_ref, opts, _last_checkout_at} =
+        {_name, {scheme, host, port}, _pool_idx, _metric_ref, opts, _actt_info} =
           pool_state
       ) do
     {:ok, Conn.new(scheme, host, port, opts, self()), pool_state}
   end
 
   @impl NimblePool
   def handle_checkout(:checkout, _, %{mint: nil} = conn, pool_state) do
-    {_name, _shp, _pool_idx, metric_ref, _opts, _last_checkout_ts} = pool_state
+    {_name, _shp, _pool_idx, metric_ref, _opts, _actt_info} = pool_state
     idle_time = System.monotonic_time() - conn.last_checkin
     PoolMetrics.maybe_add(metric_ref, in_use_connections: 1)
     {:ok, {:fresh, conn, idle_time}, conn, pool_state}
   end
 
   def handle_checkout(:checkout, _from, conn, pool_state) do
     idle_time = System.monotonic_time() - conn.last_checkin
-    {_name, {scheme, host, port}, _pool_idx, metric_ref, _opts, _last_checkout_ts} = pool_state
+    {_name, {scheme, host, port}, _pool_idx, metric_ref, _opts, _actt_info} = pool_state
 
     with true <- Conn.reusable?(conn, idle_time),
          {:ok, conn} <- Conn.set_mode(conn, :passive) do
       PoolMetrics.maybe_add(metric_ref, in_use_connections: 1)
-      {:ok, {:reuse, conn, idle_time}, conn, update_last_checkout_ts(pool_state)}
+      {:ok, {:reuse, conn, idle_time}, conn, update_activity_info(:checkout, pool_state)}
     else
       false ->
         meta = %{
@@ -225,15 +225,19 @@ defmodule Finch.HTTP1.Pool do
 
   @impl NimblePool
   def handle_checkin(checkin, _from, _old_conn, pool_state) do
-    {_name, _shp, _pool_idx, metric_ref, _opts, _last_checkout_ts} = pool_state
+    {_name, _shp, _pool_idx, metric_ref, _opts, _actt_info} = pool_state
     PoolMetrics.maybe_add(metric_ref, in_use_connections: -1)
 
     with {:ok, conn} <- checkin,
          {:ok, conn} <- Conn.set_mode(conn, :active) do
-      {:ok, %{conn | last_checkin: System.monotonic_time()}, pool_state}
+      {
+        :ok,
+        %{conn | last_checkin: System.monotonic_time()},
+        update_activity_info(:checkin, pool_state)
+      }
     else
       _ ->
-        {:remove, :closed, pool_state}
+        {:remove, :closed, update_activity_info(:checkin, pool_state)}
     end
   end
 
@@ -253,14 +257,24 @@ defmodule Finch.HTTP1.Pool do
 
   @impl NimblePool
   def handle_ping(conn, pool_state) do
-    {_name, {scheme, host, port}, _pool_idx, _metric_ref, opts, last_checkout_ts} = pool_state
+    {_name, {scheme, host, port}, _pool_idx, _metric_ref, opts, activity_info} = pool_state
 
     max_idle_time = Map.get(opts, :pool_max_idle_time, :infinity)
     now = System.monotonic_time(:millisecond)
-    diff_from_last_checkout = now - last_checkout_ts
+    diff_from_last_checkout = now - activity_info.last_checkout_ts
+
+    is_idle? = diff_from_last_checkout > max_idle_time
+    max_idle_time_configured? = is_number(max_idle_time)
+    any_connection_in_use? = activity_info.in_use_count > 0
 
     cond do
-      is_number(max_idle_time) and diff_from_last_checkout > max_idle_time ->
+      not max_idle_time_configured? ->
+        {:ok, conn}
+
+      any_connection_in_use? ->
+        {:ok, conn}
+
+      is_idle? ->
         meta = %{
           scheme: scheme,
           host: host,
@@ -285,7 +299,7 @@ defmodule Finch.HTTP1.Pool do
 
   @impl NimblePool
   def handle_cancelled(:checked_out, pool_state) do
-    {_name, _shp, _pool_idx, metric_ref, _opts, _last_checkout_ts} = pool_state
+    {_name, _shp, _pool_idx, metric_ref, _opts, _actt_info} = pool_state
     PoolMetrics.maybe_add(metric_ref, in_use_connections: -1)
     :ok
   end
@@ -315,6 +329,22 @@ defmodule Finch.HTTP1.Pool do
   defp pool_idle_timeout(:infinity), do: nil
   defp pool_idle_timeout(pool_max_idle_time), do: pool_max_idle_time
 
-  defp update_last_checkout_ts(pool_state),
-    do: put_elem(pool_state, 5, System.monotonic_time(:millisecond))
+  defp init_activity_info(),
+    do: %{in_use_count: 0, last_checkout_ts: System.monotonic_time(:millisecond)}
+
+  defp update_activity_info(:checkout, pool_state) do
+    info = %{in_use_count: count} = elem(pool_state, 5)
+
+    put_elem(pool_state, 5, %{
+      info
+      | in_use_count: count + 1,
+        last_checkout_ts: System.monotonic_time(:millisecond)
+    })
+  end
+
+  defp update_activity_info(:checkin, pool_state) do
+    info = %{in_use_count: count} = elem(pool_state, 5)
+
+    put_elem(pool_state, 5, %{info | in_use_count: max(count - 1, 0)})
+  end
 end

test/finch/http1/pool_test.exs

@@ -125,6 +125,68 @@ defmodule Finch.HTTP1.PoolTest do
     assert [] = DynamicSupervisor.which_children(:"#{finch_name}.PoolSupervisor")
   end
 
+  # @tag capture_log: true
+  test "should not terminate if a connection is checked out", %{
+    bypass: bypass,
+    finch_name: finch_name
+  } do
+    parent = self()
+
+    start_supervised!(
+      {Finch,
+       name: finch_name,
+       pools: %{
+         default: [count: 1, size: 2, pool_max_idle_time: 100]
+       }}
+    )
+
+    Bypass.expect(bypass, fn conn ->
+      {"delay", str_delay} =
+        Enum.find(conn.req_headers, fn h -> match?({"delay", _}, h) end)
+
+      Process.sleep(String.to_integer(str_delay))
+      Plug.Conn.send_resp(conn, 200, "OK")
+    end)
+
+    delay_exec = fn ref, delay ->
+      send(parent, {ref, :start})
+
+      resp =
+        Finch.build(:get, endpoint(bypass), [{"delay", "#{delay}"}])
+        |> Finch.request(finch_name)
+
+      send(parent, {ref, :done})
+
+      resp
+    end
+
+    ref1 = make_ref()
+    ref2 = make_ref()
+
+    Task.async(fn -> delay_exec.(ref1, 10) end)
+    Task.async(fn -> delay_exec.(ref2, 10) end)
+
+    assert_receive {^ref1, :done}
+    assert_receive {^ref2, :done}
+
+    assert [{pool, _pool_mod}] = Registry.lookup(finch_name, shp(bypass))
+
+    Process.monitor(pool)
+
+    ref2 = make_ref()
+    Task.async(fn -> delay_exec.(ref2, 1000) end)
+
+    assert_receive {^ref2, :start}
+
+    refute_receive {:DOWN, _, :process, ^pool, {:shutdown, :idle_timeout}}, 1000
+
+    assert_receive {^ref2, :done}
+
+    assert_receive {:DOWN, _, :process, ^pool, {:shutdown, :idle_timeout}}, 200
+
+    assert [] = DynamicSupervisor.which_children(:"#{finch_name}.PoolSupervisor")
+  end
+
   describe "async_request" do
     @describetag bypass: false