Feature [client]: add supervised connection pool with legacy mode opt…

grpc/config/test.exs

@@ -3,3 +3,5 @@ import Config
 config :logger, level: :info
 
 config :grpc, :dns_adapter, GRPC.Client.Resolver.DNS.MockAdapter
+
+config :grpc, pool_enabled: true

grpc/guides/advanced/connection_pool.md

@@ -0,0 +1,90 @@
+# Connection Pool
+
+The Elixir gRPC client maintains a **connection pool** for every target address. Rather than opening a fresh HTTP/2 connection for each RPC, callers share a fixed set of long-lived connections. This reduces connection overhead, improves throughput under concurrency, and enables fine-grained control over how many connections and concurrent streams your client opens.
+
+The pool is managed transparently: when you call a stub, the client checks out a connection, tracks the open stream count, and returns the connection when the call finishes. All of this happens without any additional code on your side.
+
+---
+
+## How It Works
+
+When `GRPC.Stub.connect/2` is called, a supervised pool of connections is started for the target address. Each connection is an independent HTTP/2 session capable of multiplexing many concurrent streams.
+
+On each RPC call the client:
+
+1. Picks first connection from the pool with number open streams that is still below the configured limit.
+2. Increments that connection's open-stream counter and records a lease.
+3. Runs the RPC.
+4. Decrements the counter and releases the lease when the call completes.
+
+If every connection in the pool has reached `max_streams`, the client opens an **overflow** connection up to `max_overflow` extra connections. Overflow connections aren't discarded once the load subsides, but if pool dies or connections are otherwise dropped we reset to initial count.
+
+---
+
+## Configuration
+
+Pool behaviour is controlled via the `:pool` option passed to `GRPC.Stub.connect/2` or `GRPC.Client.Connection.connect/2`.
+
+| Option                   | Type                        | Default | Description                                                                                                        |
+|:-------------------------|:----------------------------|:--------|:-------------------------------------------------------------------------------------------------------------------|
+| `:size`                  | `non_neg_integer`           | `1`     | Number of persistent connections kept open.                                                                        |
+| `:max_overflow`          | `non_neg_integer or nil`    | `0`     | Maximum number of extra connections created when the pool is fully saturated. `nil` means no client-side limit.    |
+| `:max_streams`           | `pos_integer or nil`        | `nil`   | Maximum concurrent streams per connection. `nil` means no client-side limit (server limit applies).                |
+| `:health_check_enabled`  | `boolean`                   | `false` | When `true`, a periodic gRPC health-check ping is sent on each connection every 10 minutes.                        |
+
+---
+
+## Examples
+
+### Default pool (single connection)
+
+```elixir
+iex> {:ok, channel} = GRPC.Stub.connect("localhost:50051")
+iex> {:ok, reply} = channel |> MyService.Stub.my_rpc(request)
+```
+
+### Multiple persistent connections
+
+Open three connections upfront to distribute concurrent load:
+
+```elixir
+iex> {:ok, channel} = GRPC.Stub.connect("localhost:50051", pool: %{size: 3})
+iex> {:ok, reply} = channel |> MyService.Stub.my_rpc(request)
+```
+
+### Overflow connections
+
+Keep two persistent connections and allow up to five overflow connections during traffic spikes:
+
+```elixir
+iex> {:ok, channel} =
+...>   GRPC.Stub.connect("localhost:50051",
+...>     pool: %{size: 2, max_overflow: 5}
+...>   )
+```
+
+### Limiting streams per connection
+
+Cap each connection at 100 concurrent streams. Requests beyond this limit will use a new overflow connection (if allowed):
+
+```elixir
+iex> {:ok, channel} =
+...>   GRPC.Stub.connect("localhost:50051",
+...>     pool: %{size: 2, max_overflow: 1, max_streams: 100}
+...>   )
+```
+
+---
+
+## Disconnect
+
+Calling `GRPC.Stub.disconnect/1` stops the pool supervisor and closes all underlying connections:
+
+```elixir
+iex> {:ok, channel} = GRPC.Stub.connect("localhost:50051")
+iex> {:ok, _} = GRPC.Stub.disconnect(channel)
+```
+
+The returned channel has `pool: nil`, indicating the pool is no longer active.
+
+---

grpc/lib/grpc/channel.ex

@@ -27,5 +27,6 @@ defmodule GRPC.Channel do
             interceptors: [],
             compressor: nil,
             accepted_compressors: [],
-            headers: []
+            headers: [],
+            pool: nil
 end

grpc/lib/grpc/client/adapters/gun/connection_process.ex

@@ -110,21 +110,21 @@ defmodule GRPC.Client.Adapters.Gun.ConnectionProcess do
     {:stop, :normal, :ok, state}
   end
 
-  def handle_call({:request, path, headers, body}, _from, %{gun_pid: gun_pid} = state) do
+  def handle_call({:request, path, headers, body}, {caller_pid, _}, %{gun_pid: gun_pid} = state) do
     with {:ok, response_pid} <- start_response_process(),
          stream_ref <- :gun.post(gun_pid, path, headers, body, %{reply_to: response_pid}) do
       {:reply, {:ok, %{stream_ref: stream_ref, response_pid: response_pid}},
-       put_response_pid(state, stream_ref, response_pid)}
+       put_response_pid(state, stream_ref, response_pid, caller_pid)}
     else
       {:error, reason} -> {:reply, {:error, reason}, state}
     end
   end
 
-  def handle_call({:open_stream, path, headers}, _from, %{gun_pid: gun_pid} = state) do
+  def handle_call({:open_stream, path, headers}, {caller_pid, _}, %{gun_pid: gun_pid} = state) do
     with {:ok, response_pid} <- start_response_process(),
          stream_ref <- :gun.post(gun_pid, path, headers, %{reply_to: response_pid}) do
       {:reply, {:ok, %{stream_ref: stream_ref, response_pid: response_pid}},
-       put_response_pid(state, stream_ref, response_pid)}
+       put_response_pid(state, stream_ref, response_pid, caller_pid)}
     else
       {:error, reason} -> {:reply, {:error, reason}, state}
     end
@@ -164,15 +164,57 @@ defmodule GRPC.Client.Adapters.Gun.ConnectionProcess do
   # Gun is gone for good (reconnect retries exhausted or a crash). Fail all
   # in-flight streams and stop
   def handle_info({:DOWN, _monitor_ref, :process, gun_pid, reason}, %{gun_pid: gun_pid} = state) do
-    Enum.each(state.response_processes, fn {_stream_ref, {response_pid, _monitor_ref}} ->
-      send(response_pid, {:connection_down, reason})
+    Enum.each(state.response_processes, fn {_stream_ref, entry} ->
+      pid =
+        case entry do
+          %{response_pid: p} -> p
+          {p, _} -> p
+          _ -> nil
+        end
+
+      if is_pid(pid) do
+        send(pid, {:connection_down, reason})
+      end
     end)
 
     {:stop, {:shutdown, {:gun_down, reason}}, state}
   end
 
-  def handle_info({:DOWN, monitor_ref, :process, _pid, _reason}, state) do
-    {:noreply, drop_response_pid_by_monitor(state, monitor_ref)}
+  def handle_info({:DOWN, monitor_ref, :process, _pid, _reason}, %{gun_pid: gun_pid} = state) do
+    case find_stream_by_monitor(state, monitor_ref) do
+      {stream_ref, entry} ->
+        :ok = :gun.cancel(gun_pid, stream_ref)
+
+        response_pid =
+          case entry do
+            %{response_pid: pid} -> pid
+            {pid, _} -> pid
+            _ -> nil
+          end
+
+        caller_monitor =
+          case entry do
+            %{caller_monitor: mon} -> mon
+            _ -> nil
+          end
+
+        if monitor_ref == caller_monitor and is_pid(response_pid) and Process.alive?(response_pid) do
+          GenServer.stop(response_pid, :normal)
+        end
+
+        {:noreply, drop_response_pid(state, stream_ref)}
+
+      nil ->
+        {:noreply, state}
+    end
+  end
+
+  def handle_info({:gun_error, _conn_pid, stream_ref, reason}, state) do
+    Logger.debug(
+      "#{inspect(__MODULE__)} received gun_error for stream #{inspect(stream_ref)}: #{inspect(reason)}"
+    )
+
+    {:noreply, drop_response_pid(state, stream_ref)}
   end
 
   def handle_info(msg, state) do
@@ -200,13 +242,32 @@ defmodule GRPC.Client.Adapters.Gun.ConnectionProcess do
     StreamResponseProcess.start_link()
   end
 
-  defp put_response_pid(%{response_processes: processes} = state, stream_ref, response_pid) do
-    monitor_ref = Process.monitor(response_pid)
-    %{state | response_processes: Map.put(processes, stream_ref, {response_pid, monitor_ref})}
+  defp put_response_pid(
+         %{response_processes: processes} = state,
+         stream_ref,
+         response_pid,
+         caller_pid
+       ) do
+    response_monitor = Process.monitor(response_pid)
+    caller_monitor = if is_pid(caller_pid), do: Process.monitor(caller_pid), else: nil
+
+    entry = %{
+      response_pid: response_pid,
+      response_monitor: response_monitor,
+      caller_pid: caller_pid,
+      caller_monitor: caller_monitor
+    }
+
+    %{state | response_processes: Map.put(processes, stream_ref, entry)}
   end
 
   defp drop_response_pid(%{response_processes: processes} = state, stream_ref) do
     case Map.pop(processes, stream_ref) do
+      {%{response_monitor: r_mon, caller_monitor: c_mon}, remaining} ->
+        if r_mon, do: Process.demonitor(r_mon, [:flush])
+        if c_mon, do: Process.demonitor(c_mon, [:flush])
+        %{state | response_processes: remaining}
+
       {{_response_pid, monitor_ref}, remaining} ->
         Process.demonitor(monitor_ref, [:flush])
         %{state | response_processes: remaining}
@@ -216,15 +277,24 @@ defmodule GRPC.Client.Adapters.Gun.ConnectionProcess do
     end
   end
 
-  defp drop_response_pid_by_monitor(%{response_processes: processes} = state, monitor_ref) do
-    case Enum.find(processes, fn {_stream_ref, {_response_pid, ref}} -> ref == monitor_ref end) do
-      {stream_ref, _entry} -> %{state | response_processes: Map.delete(processes, stream_ref)}
-      nil -> state
-    end
+  defp find_stream_by_monitor(%{response_processes: processes}, monitor_ref) do
+    Enum.find(processes, fn {_stream_ref, entry} ->
+      case entry do
+        %{response_monitor: r_mon, caller_monitor: c_mon} ->
+          r_mon == monitor_ref or c_mon == monitor_ref
+
+        {_pid, r_mon} ->
+          r_mon == monitor_ref
+
+        _ ->
+          false
+      end
+    end)
   end
 
   defp response_pid(%{response_processes: processes}, stream_ref) do
     case Map.get(processes, stream_ref) do
+      %{response_pid: response_pid} -> response_pid
       {response_pid, _monitor_ref} -> response_pid
       nil -> nil
     end

grpc/lib/grpc/client/adapters/mint/connection_process/connection_process.ex

@@ -117,10 +117,13 @@ if Code.ensure_loaded?(Mint.HTTP) do
         ) do
       case Mint.HTTP.request(state.conn, method, path, headers, :stream) do
         {:ok, conn, request_ref} ->
+          stream_response_pid = opts[:stream_response_pid]
+          monitor_ref = if is_pid(stream_response_pid), do: Process.monitor(stream_response_pid)
+
           new_state =
             state
             |> State.update_conn(conn)
-            |> State.put_empty_ref_state(request_ref, opts[:stream_response_pid])
+            |> State.put_empty_ref_state(request_ref, stream_response_pid, monitor_ref)
 
           {:reply, {:ok, %{request_ref: request_ref}}, new_state}
 
@@ -138,12 +141,14 @@ if Code.ensure_loaded?(Mint.HTTP) do
       case Mint.HTTP.request(state.conn, method, path, headers, :stream) do
         {:ok, conn, request_ref} ->
           queue = :queue.in({request_ref, body, nil}, state.request_stream_queue)
+          stream_response_pid = opts[:stream_response_pid]
+          monitor_ref = if is_pid(stream_response_pid), do: Process.monitor(stream_response_pid)
 
           new_state =
             state
             |> State.update_conn(conn)
             |> State.update_request_stream_queue(queue)
-            |> State.put_empty_ref_state(request_ref, opts[:stream_response_pid])
+            |> State.put_empty_ref_state(request_ref, stream_response_pid, monitor_ref)
 
           {:reply, {:ok, %{request_ref: request_ref}}, new_state,
            {:continue, :process_request_stream_queue}}
@@ -155,12 +160,19 @@ if Code.ensure_loaded?(Mint.HTTP) do
     end
 
     def handle_call({:stream_body, request_ref, :eof}, _from, state) do
-      case Mint.HTTP.stream_request_body(state.conn, request_ref, :eof) do
-        {:ok, conn} ->
-          {:reply, :ok, State.update_conn(state, conn)}
+      if State.has_request_ref(state, request_ref) do
+        case Mint.HTTP.stream_request_body(state.conn, request_ref, :eof) do
+          {:ok, conn} ->
+            {:reply, :ok, State.update_conn(state, conn)}
 
-        {:error, conn, error} ->
-          {:reply, {:error, error}, State.update_conn(state, conn)}
+          {:error, conn, %Mint.HTTPError{reason: :unknown_request_to_stream}} ->
+            {:reply, :ok, State.update_conn(state, conn)}
+
+          {:error, conn, error} ->
+            {:reply, {:error, error}, State.update_conn(state, conn)}
+        end
+      else
+        {:reply, :ok, state}
       end
     end
 
@@ -190,6 +202,17 @@ if Code.ensure_loaded?(Mint.HTTP) do
       attempt_reconnect(state)
     end
 
+    def handle_info({:DOWN, monitor_ref, :process, pid, reason}, state) do
+      case State.find_request_ref_by_monitor_ref(state, monitor_ref) do
+        nil ->
+          {:noreply, state}
+
+        request_ref ->
+          new_state = cancel_dead_stream(state, request_ref, pid, reason)
+          {:noreply, new_state}
+      end
+    end
+
     def handle_info(message, state) do
       case Mint.HTTP.stream(state.conn, message) do
         :unknown ->

grpc/lib/grpc/client/adapters/mint/connection_process/state.ex

@@ -37,14 +37,21 @@ if Code.ensure_loaded?(Mint.HTTP) do
       %{state | request_stream_queue: queue}
     end
 
-    def put_empty_ref_state(state, ref, response_pid) do
+    def put_empty_ref_state(state, ref, response_pid, monitor_ref \\ nil) do
       put_in(state.requests[ref], %{
         stream_response_pid: response_pid,
+        monitor_ref: monitor_ref,
         done: false,
         response: %{}
       })
     end
 
+    def find_request_ref_by_monitor_ref(state, monitor_ref) do
+      Enum.find_value(state.requests, fn {ref, req} ->
+        if req[:monitor_ref] == monitor_ref, do: ref
+      end)
+    end
+
     def update_response_status(state, ref, status) do
       put_in(state.requests[ref].response[:status], status)
     end
@@ -64,7 +71,13 @@ if Code.ensure_loaded?(Mint.HTTP) do
     defguard has_request_ref(state, ref) when is_map_key(state.requests, ref)
 
     def pop_ref(state, ref) do
-      pop_in(state.requests[ref])
+      {request_data, new_requests} = Map.pop(state.requests, ref)
+
+      if is_map(request_data) and is_reference(request_data[:monitor_ref]) do
+        Process.demonitor(request_data[:monitor_ref], [:flush])
+      end
+
+      {request_data, %{state | requests: new_requests}}
     end
 
     def append_response_data(state, ref, new_data) do

grpc/lib/grpc/client/application.ex

@@ -7,6 +7,7 @@ defmodule GRPC.Client.Application do
 
     children = [
       {Registry, [keys: :unique, name: GRPC.Client.Registry]},
+      {Registry, [keys: :unique, name: GRPC.Client.Pool.Registry]},
       {DynamicSupervisor,
        [
          name: GRPC.Client.Supervisor,