Repository: general-CbIC/poolex (elixir)

Issue:
Handle errors on worker's launch
Sourced from Elixir forum: [link](https://elixirforum.com/t/poolex-a-library-for-managing-pools-of-workers/54800/12?u=general-cbic)

> Poolex is not handling worker start errors and whole pool will die with MatchError if some extra worker fails to start. It makes this pool inapplicable for usage in environments where workers connect to external services (like databases or HTTP services which are most pooling use-cases) which can be unavailable or return 422 for example.

Pull request:
Handle workers failed start
Closes #113 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b7db3a6..858eeed 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,11 +10,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Added caching of the dependencies on CI (compilation speed up).
+- Added a new option `failed_workers_retry_interval` to the pool configuration. This option configures an interval in milliseconds between retry attempts for workers which failed to start. The value is `1 second` by default. Example:
+
+  ```elixir
+  Poolex.start_link(
+    worker_module: SomeUnstableWorker,
+    workers_count: 5,
+    failed_workers_retry_interval: 3_000
+  )
+  ```
 
 ### Changed
 
 - Due to the deprecation of support ubuntu 20.04 on GitHub Actions (read more [here](https://github.com/actions/runner-images/issues/11101)), bumped minimum required versions of Elixir to `~> 1.11` and Erlang/OTP to `~> 24`.
 
+### Fixed
+
+- Fixed the `MatchError` on pool starting error that occurs with errors on the launch of workers (read more in [issue](https://github.com/general-CbIC/poolex/issues/113)).
+
 ## [1.2.1] - 2025-04-06
 
 ### Fixed
diff --git a/lib/poolex.ex b/lib/poolex.ex
index 70ebf75..1fd42db 100644
--- a/lib/poolex.ex
+++ b/lib/poolex.ex
@@ -22,7 +22,7 @@ defmodule Poolex do
   {:ok, true}
   ```
 
-  Fore more information see [Getting Started](https://hexdocs.pm/poolex/getting-started.html)
+  For more information see [Getting Started](https://hexdocs.pm/poolex/getting-started.html)
   """
 
   use GenServer, shutdown: :infinity
@@ -35,20 +35,27 @@ defmodule Poolex do
   alias Poolex.Private.State
   alias Poolex.Private.WaitingCallers
 
+  require Logger
+
   @default_checkout_timeout :timer.seconds(5)
+
+  # Interval between retry attempts for workers that failed to start (1 second by default)
+  @default_failed_workers_retry_interval :timer.seconds(1)
+
   @poolex_options_table """
-  | Option                 | Description                                          | Example               | Default value                     |
-  |------------------------|------------------------------------------------------|-----------------------|-----------------------------------|
-  | `pool_id`              | Identifier by which you will access the pool         | `:my_pool`            | `worker_module` value             |
-  | `worker_module`        | Name of module that implements our worker            | `MyApp.Worker`        | **option is required**            |
-  | `workers_count`        | How many workers should be running in the pool       | `5`                   | **option is required**            |
-  | `max_overflow`         | How many workers can be created over the limit       | `2`                   | `0`                               |
-  | `worker_args`          | List of arguments passed to the start function       | `[:gg, "wp"]`         | `[]`                              |
-  | `worker_start_fun`     | Name of the function that starts the worker          | `:run`                | `:start_link`                     |
-  | `busy_workers_impl`    | Module that describes how to work with busy workers  | `SomeBusyWorkersImpl` | `Poolex.Workers.Impl.List`        |
-  | `idle_workers_impl`    | Module that describes how to work with idle workers  | `SomeIdleWorkersImpl` | `Poolex.Workers.Impl.List`        |
-  | `waiting_callers_impl` | Module that describes how to work with callers queue | `WaitingCallersImpl`  | `Poolex.Callers.Impl.ErlangQueue` |
-  | `pool_size_metrics`    | Whether to dispatch pool size metrics                | `true`                | `false`                           |
+  | Option                          | Description                                                        | Example               | Default value                     |
+  |---------------------------------|--------------------------------------------------------------------|-----------------------|-----------------------------------|
+  | `pool_id`                       | Identifier by which you will access the pool                       | `:my_pool`            | `worker_module` value             |
+  | `worker_module`                 | Name of module that implements our worker                          | `MyApp.Worker`        | **option is required**            |
+  | `workers_count`                 | How many workers should be running in the pool                     | `5`                   | **option is required**            |
+  | `max_overflow`                  | How many workers can be created over the limit                     | `2`                   | `0`                               |
+  | `worker_args`                   | List of arguments passed to the start function                     | `[:gg, "wp"]`         | `[]`                              |
+  | `worker_start_fun`              | Name of the function that starts the worker                        | `:run`                | `:start_link`                     |
+  | `busy_workers_impl`             | Module that describes how to work with busy workers                | `SomeBusyWorkersImpl` | `Poolex.Workers.Impl.List`        |
+  | `idle_workers_impl`             | Module that describes how to work with idle workers                | `SomeIdleWorkersImpl` | `Poolex.Workers.Impl.List`        |
+  | `waiting_callers_impl`          | Module that describes how to work with callers queue               | `WaitingCallersImpl`  | `Poolex.Callers.Impl.ErlangQueue` |
+  | `pool_size_metrics`             | Whether to dispatch pool size metrics                              | `true`                | `false`                           |
+  | `failed_workers_retry_interval` | Interval in milliseconds between retry attempts for failed workers | `5_000`               | `1_000`                           |
   """
 
   @typedoc """
@@ -70,6 +77,7 @@ defmodule Poolex do
           | {:idle_workers_impl, module()}
           | {:waiting_callers_impl, module()}
           | {:pool_size_metrics, boolean()}
+          | {:failed_workers_retry_interval, timeout()}
 
   @typedoc """
   Process id of `worker`.
@@ -216,6 +224,7 @@ defmodule Poolex do
 
       * `busy_workers_count` - how many workers are busy right now.
       * `busy_workers_pids` - list of busy workers.
+      * `failed_to_start_workers_count` - how many workers failed to start.
       * `idle_workers_count` - how many workers are ready to work.
       * `idle_workers_pids` - list of idle workers.
       * `max_overflow` - how many workers can be created over the limit.
@@ -286,10 +295,14 @@ defmodule Poolex do
     waiting_callers_impl =
       Keyword.get(opts, :waiting_callers_impl, Poolex.Callers.Impl.ErlangQueue)
 
+    failed_workers_retry_interval =
+      Keyword.get(opts, :failed_workers_retry_interval, @default_failed_workers_retry_interval)
+
     {:ok, supervisor} = Poolex.Private.Supervisor.start_link()
 
     state =
       %State{
+        failed_workers_retry_interval: failed_workers_retry_interval,
         max_overflow: max_overflow,
         pool_id: pool_id,
         supervisor: supervisor,
@@ -324,6 +337,8 @@ defmodule Poolex do
   def handle_continue(opts, state) do
     Metrics.start_poller(opts)
 
+    state = schedule_retry_failed_workers(state, state.failed_workers_retry_interval)
+
     {:noreply, state}
   end
 
@@ -333,10 +348,16 @@ defmodule Poolex do
   end
 
   defp start_workers(workers_count, state) when is_integer(workers_count) and workers_count >= 1 do
-    Enum.map_reduce(1..workers_count, state, fn _iterator, state ->
-      {:ok, worker_pid} = start_worker(state)
-      state = Monitoring.add(state, worker_pid, :worker)
-      {worker_pid, state}
+    Enum.reduce(1..workers_count, {[], state}, fn _iterator, {workers_pids, state} ->
+      case start_worker(state) do
+        {:ok, worker_pid} ->
+          state = Monitoring.add(state, worker_pid, :worker)
+          {[worker_pid | workers_pids], state}
+
+        {:error, :failed_to_start_worker} ->
+          state = %{state | failed_to_start_workers_count: state.failed_to_start_workers_count + 1}
+          {workers_pids, state}
+      end
     end)
   end
 
@@ -345,13 +366,20 @@ defmodule Poolex do
     raise ArgumentError, msg
   end
 
-  @spec start_worker(State.t()) :: {:ok, pid()}
+  @spec start_worker(State.t()) :: {:ok, pid()} | {:error, :failed_to_start_worker}
   defp start_worker(%State{} = state) do
-    DynamicSupervisor.start_child(state.supervisor, %{
-      id: make_ref(),
-      start: {state.worker_module, state.worker_start_fun, state.worker_args},
-      restart: :temporary
-    })
+    case DynamicSupervisor.start_child(state.supervisor, %{
+           id: make_ref(),
+           start: {state.worker_module, state.worker_start_fun, state.worker_args},
+           restart: :temporary
+         }) do
+      {:ok, pid} ->
+        {:ok, pid}
+
+      {:error, reason} ->
+        Logger.error("[Poolex] Failed to start worker. Reason: #{inspect(reason)}")
+        {:error, :failed_to_start_worker}
+    end
   end
 
   @spec stop_worker(Supervisor.supervisor(), pid()) :: :ok | {:error, :not_found}
@@ -392,6 +420,7 @@ defmodule Poolex do
       busy_workers_count: BusyWorkers.count(state),
       busy_workers_impl: state.busy_workers_impl,
       busy_workers_pids: BusyWorkers.to_list(state),
+      failed_to_start_workers_count: state.failed_to_start_workers_count,
       idle_workers_count: IdleWorkers.count(state),
       idle_workers_impl: state.idle_workers_impl,
       idle_workers_pids: IdleWorkers.to_list(state),
@@ -472,6 +501,22 @@ defmodule Poolex do
     end
   end
 
+  @impl GenServer
+  def handle_info(:retry_failed_workers, %{failed_workers_retry_interval: retry_interval} = state) do
+    # Try to start workers that failed to initialize
+    state =
+      if state.failed_to_start_workers_count > 0 do
+        retry_failed_workers(state)
+      else
+        state
+      end
+
+    # Schedule next check
+    state = schedule_retry_failed_workers(state, retry_interval)
+
+    {:noreply, state}
+  end
+
   @spec release_busy_worker(State.t(), worker()) :: State.t()
   defp release_busy_worker(%State{} = state, worker) do
     if BusyWorkers.member?(state, worker) do
@@ -551,4 +596,37 @@ defmodule Poolex do
       end
     end)
   end
+
+  @spec schedule_retry_failed_workers(State.t(), timeout()) :: State.t()
+  defp schedule_retry_failed_workers(state, retry_interval) do
+    Process.send_after(self(), :retry_failed_workers, retry_interval)
+    %{state | failed_workers_retry_interval: retry_interval}
+  end
+
+  @spec retry_failed_workers(State.t()) :: State.t()
+  defp retry_failed_workers(%State{} = state) do
+    workers_to_retry = state.failed_to_start_workers_count
+
+    Logger.info("[Poolex] Attempting to restart #{workers_to_retry} failed workers")
+
+    # Reset the failed workers counter
+    state = %{state | failed_to_start_workers_count: 0}
+
+    # Start the specified number of workers
+    {workers, updated_state} = start_workers(workers_to_retry, state)
+
+    # Add successfully started workers to the pool
+    Enum.reduce(workers, updated_state, fn worker, acc_state ->
+      if WaitingCallers.empty?(acc_state) do
+        # If there are no waiting callers, add to idle workers list
+        IdleWorkers.add(acc_state, worker)
+      else
+        # If there are waiting callers, give them the worker
+        acc_state
+        |> Monitoring.add(worker, :worker)
+        |> BusyWorkers.add(worker)
+        |> provide_worker_to_waiting_caller(worker)
+      end
+    end)
+  end
 end
diff --git a/lib/poolex/private/debug_info.ex b/lib/poolex/private/debug_info.ex
index bb347b1..4fb6f2c 100644
--- a/lib/poolex/private/debug_info.ex
+++ b/lib/poolex/private/debug_info.ex
@@ -8,6 +8,7 @@ defmodule Poolex.Private.DebugInfo do
   defstruct busy_workers_count: 0,
             busy_workers_impl: nil,
             busy_workers_pids: [],
+            failed_to_start_workers_count: 0,
             idle_workers_count: 0,
             idle_workers_impl: nil,
             idle_workers_pids: [],
@@ -23,6 +24,7 @@ defmodule Poolex.Private.DebugInfo do
           busy_workers_count: non_neg_integer(),
           busy_workers_impl: module(),
           busy_workers_pids: list(pid()),
+          failed_to_start_workers_count: non_neg_integer(),
           idle_workers_count: non_neg_integer(),
           idle_workers_impl: module(),
           idle_workers_pids: list(pid()),
diff --git a/lib/poolex/private/state.ex b/lib/poolex/private/state.ex
index 495dbf3..cb8dff4 100644
--- a/lib/poolex/private/state.ex
+++ b/lib/poolex/private/state.ex
@@ -6,6 +6,7 @@ defmodule Poolex.Private.State do
   """
 
   @enforce_keys [
+    :failed_workers_retry_interval,
     :max_overflow,
     :pool_id,
     :supervisor,
@@ -18,6 +19,7 @@ defmodule Poolex.Private.State do
               [
                 busy_workers_impl: nil,
                 busy_workers_state: nil,
+                failed_to_start_workers_count: 0,
                 idle_workers_impl: nil,
                 idle_workers_state: nil,
                 monitors: %{},
@@ -29,6 +31,8 @@ defmodule Poolex.Private.State do
   @type t() :: %__MODULE__{
           busy_workers_impl: module(),
           busy_workers_state: nil | Poolex.Workers.Behaviour.state(),
+          failed_to_start_workers_count: non_neg_integer(),
+          failed_workers_retry_interval: timeout() | nil,
           idle_workers_impl: module(),
           idle_workers_state: nil | Poolex.Workers.Behaviour.state(),
           max_overflow: non_neg_integer(),
diff --git a/lib/poolex/private/supervisor.ex b/lib/poolex/private/supervisor.ex
index da24a9e..893717f 100644
--- a/lib/poolex/private/supervisor.ex
+++ b/lib/poolex/private/supervisor.ex
@@ -5,7 +5,7 @@ defmodule Poolex.Private.Supervisor do
   @doc false
   @spec start_link() :: Supervisor.on_start()
   def start_link do
-    DynamicSupervisor.start_link(__MODULE__, nil)
+    DynamicSupervisor.start_link(__MODULE__, [])
   end
 
   @doc false
diff --git a/mix.exs b/mix.exs
index 4f6b698..034b01a 100644
--- a/mix.exs
+++ b/mix.exs
@@ -21,7 +21,9 @@ defmodule Poolex.MixProject do
 
   # Run "mix help compile.app" to learn about applications.
   def application do
-    []
+    [
+      extra_applications: [:logger]
+    ]
   end
 
   defp elixirc_paths(:test), do: ["lib", "test/support"]


Test patch:
diff --git a/test/poolex_test.exs b/test/poolex_test.exs
index eb7da0f..364ee0d 100644
--- a/test/poolex_test.exs
+++ b/test/poolex_test.exs
@@ -714,4 +714,33 @@ defmodule PoolexTest do
       end)
     end
   end
+
+  describe "handle errors on workers launch" do
+    @describetag capture_log: true
+    test "while starting the pool", %{pool_options: pool_options} do
+      {:ok, control_agent} = Agent.start_link(fn -> 3 end)
+
+      pool_name =
+        pool_options
+        |> Keyword.merge(
+          worker_module: SomeUnstableWorker,
+          worker_args: [[control_agent: control_agent]],
+          failed_workers_retry_interval: 100
+        )
+        |> start_pool()
+
+      debug_info = Poolex.get_debug_info(pool_name)
+      assert debug_info.failed_to_start_workers_count == 2
+      assert debug_info.idle_workers_count == 3
+
+      # Increase the agent value to allow the remaining workers to start
+      Agent.update(control_agent, fn _ -> 2 end)
+
+      :timer.sleep(150)
+
+      debug_info = Poolex.get_debug_info(pool_name)
+      assert debug_info.failed_to_start_workers_count == 0
+      assert debug_info.idle_workers_count == 5
+    end
+  end
 end
diff --git a/test/support/some_unstable_worker.ex b/test/support/some_unstable_worker.ex
new file mode 100644
index 0000000..0e684a4
--- /dev/null
+++ b/test/support/some_unstable_worker.ex
@@ -0,0 +1,50 @@
+defmodule SomeUnstableWorker do
+  @moduledoc false
+  use GenServer
+
+  def traceable_call(server, pid, msg, delay) do
+    GenServer.call(server, {:traceable, pid, msg, delay})
+  end
+
+  def start_link(args) do
+    GenServer.start_link(__MODULE__, args)
+  end
+
+  @impl true
+  def init(args) do
+    control_agent = Keyword.fetch!(args, :control_agent)
+
+    if Agent.get_and_update(control_agent, fn
+         0 -> {false, 0}
+         value -> {true, value - 1}
+       end) do
+      {:ok, nil}
+    else
+      {:error, :failed}
+    end
+  end
+
+  @impl true
+  def handle_call({:traceable, pid, msg, delay}, _pid, state) do
+    send(pid, {:traceable_start, msg, self()})
+    :timer.sleep(delay)
+    Process.send_after(pid, {:traceable_end, msg, self()}, 10)
+    {:reply, :ok, state}
+  end
+
+  def handle_call(:do_some_work, _from, state) do
+    {:reply, :some_result, state}
+  end
+
+  def handle_call({:do_some_work_with_delay, delay}, _from, state) do
+    :timer.sleep(delay)
+
+    {:reply, :some_result, state}
+  end
+
+  def handle_call(:do_raise, _from, state) do
+    raise RuntimeError
+
+    {:reply, nil, state}
+  end
+end