diff --git a/guides/04_task_async.md b/guides/04_task_async.md index 1d8e5f8..878a51f 100644 --- a/guides/04_task_async.md +++ b/guides/04_task_async.md @@ -132,30 +132,40 @@ large_dataset ### Retry with Backoff +`Patterns.TaskAsync.retry/2` uses the [`jitter`](https://hexdocs.pm/jitter) package for +exponential backoff. Full jitter (`:full`) is the recommended default; pass +`:decorrelated` when many clients retry the same resource at once. + ```elixir -def retry_with_backoff(fun, opts \\ []) do - max_attempts = Keyword.get(opts, :max_attempts, 3) - base_delay = Keyword.get(opts, :base_delay, 100) +# Built-in retry with full jitter (default) +{:ok, result} = + Patterns.TaskAsync.retry(fn -> flaky_api_call() end, + max_attempts: 5, + base_delay: 100, + max_delay: 30_000, + jitter: :full + ) - Enum.reduce_while(1..max_attempts, nil, fn attempt, _acc -> - task = Task.async(fn -> - try do - {:ok, fun.()} - rescue - e -> {:error, e} - end - end) +# Deterministic delays for tests +Patterns.TaskAsync.retry(fn -> raise "fail" end, + max_attempts: 3, + base_delay: 50, + jitter: false +) +``` - case Task.await(task, 10_000) do - {:ok, result} -> {:halt, {:ok, result}} - {:error, _} when attempt < max_attempts -> - delay = base_delay * :math.pow(2, attempt - 1) - Process.sleep(trunc(delay)) - {:cont, nil} - {:error, reason} -> {:halt, {:error, reason}} - end - end) -end +For a custom pipeline, compose [`Jitter`](https://hexdocs.pm/jitter/Jitter.html) +directly: + +```elixir +# Single delay after attempt 2 (0-indexed) +Jitter.full(2, base: 100, cap: 30_000) + +# Or precompute delays for a manual retry loop +delays = + Jitter.full_stream(base: 100, cap: 30_000) + |> Stream.take(max_attempts - 1) + |> Enum.to_list() ``` ### Pipeline Processing diff --git a/guides/08_circuit_breaker.md b/guides/08_circuit_breaker.md index a3ec1a2..05a73b1 100644 --- a/guides/08_circuit_breaker.md +++ b/guides/08_circuit_breaker.md @@ -228,7 +228,7 @@ When the breaker is open, return a degraded-but-functional response (cached data | **Process Pool** | Bound concurrency to a resource | | **Bulkhead** | Isolate failures to a subset of resources | -Circuit breakers pair well with retries (retry within closed state) and fallbacks (serve alternatives when open). +Circuit breakers pair well with retries (retry within closed state) and fallbacks (serve alternatives when open). For retry delays, see [Task.async — Retry with Backoff](04_task_async.md#retry-with-backoff) and the [`jitter`](https://hexdocs.pm/jitter) package. ## Supervision Considerations diff --git a/lib/patterns/registry_dynamic_supervisor.ex b/lib/patterns/registry_dynamic_supervisor.ex index c629575..3028f64 100644 --- a/lib/patterns/registry_dynamic_supervisor.ex +++ b/lib/patterns/registry_dynamic_supervisor.ex @@ -78,18 +78,34 @@ defmodule Patterns.RegistryDynamicSupervisor do @spec stop_worker(Supervisor.supervisor(), worker_key()) :: :ok | {:error, :not_found | term()} def stop_worker(supervisor, key) do - %{registry: registry} = components(supervisor) + %{registry: registry, dynamic: dynamic} = components(supervisor) case Registry.lookup(registry, key) do [{pid, _}] -> - GenServer.stop(pid) - :ok + with :ok <- DynamicSupervisor.terminate_child(dynamic, pid), + :ok <- wait_until_unregistered(registry, key) do + :ok + end [] -> {:error, :not_found} end end + defp wait_until_unregistered(registry, key, attempts \\ 100) do + case Registry.lookup(registry, key) do + [] -> + :ok + + _ when attempts > 0 -> + Process.sleep(5) + wait_until_unregistered(registry, key, attempts - 1) + + _ -> + {:error, :timeout} + end + end + @doc """ Looks up a worker pid by key. """ diff --git a/lib/patterns/task_async.ex b/lib/patterns/task_async.ex index f7918d6..2810417 100644 --- a/lib/patterns/task_async.ex +++ b/lib/patterns/task_async.ex @@ -155,15 +155,27 @@ defmodule Patterns.TaskAsync do @doc """ Retry a task with exponential backoff. + + Delay between attempts uses the [`jitter`](https://hexdocs.pm/jitter) package, + which implements the strategies from Marc Brooker's + [Exponential Backoff and Jitter](https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/). + + ## Options + + - `:max_attempts` — total tries (default: `3`) + - `:base_delay` — base delay in milliseconds (default: `100`) + - `:max_delay` — cap on delay in milliseconds (default: `5000`) + - `:jitter` — `:full` (default), `:equal`, `:decorrelated`, or `false` for no jitter. + `true` is accepted as an alias for `:full`. """ @spec retry((-> term()), keyword()) :: {:ok, term()} | {:error, Exception.t()} def retry(fun, opts \\ []) do max_attempts = Keyword.get(opts, :max_attempts, 3) base_delay = Keyword.get(opts, :base_delay, 100) max_delay = Keyword.get(opts, :max_delay, 5000) - jitter = Keyword.get(opts, :jitter, true) + jitter = normalize_jitter(Keyword.get(opts, :jitter, :full)) - do_retry(fun, max_attempts, base_delay, max_delay, jitter, 1) + do_retry(fun, max_attempts, base_delay, max_delay, jitter, 1, base_delay) end @doc """ @@ -261,39 +273,57 @@ defmodule Patterns.TaskAsync do end end + @type jitter_strategy :: false | :full | :equal | :decorrelated + + @spec normalize_jitter(boolean() | jitter_strategy()) :: jitter_strategy() + defp normalize_jitter(true), do: :full + defp normalize_jitter(false), do: false + defp normalize_jitter(strategy) when strategy in [:full, :equal, :decorrelated], do: strategy + @spec do_retry( (-> term()), pos_integer(), pos_integer(), pos_integer(), - boolean(), + jitter_strategy(), + pos_integer(), pos_integer() ) :: {:ok, term()} | {:error, term()} - defp do_retry(fun, max_attempts, base_delay, max_delay, jitter, attempt) do + defp do_retry(fun, max_attempts, base_delay, max_delay, jitter, attempt, prev_delay) do result = fun.() {:ok, result} rescue e -> if attempt < max_attempts do - delay = calculate_backoff_delay(base_delay, max_delay, jitter, attempt) + delay = backoff_delay(jitter, attempt, base_delay, max_delay, prev_delay) Process.sleep(delay) - do_retry(fun, max_attempts, base_delay, max_delay, jitter, attempt + 1) + do_retry(fun, max_attempts, base_delay, max_delay, jitter, attempt + 1, delay) else {:error, e} end end - @spec calculate_backoff_delay(pos_integer(), pos_integer(), boolean(), pos_integer()) :: + @spec backoff_delay( + jitter_strategy(), + pos_integer(), + pos_integer(), + pos_integer(), pos_integer() - defp calculate_backoff_delay(base_delay, max_delay, jitter, attempt) do - delay = min(base_delay * :math.pow(2, attempt - 1), max_delay) - - if jitter do - jitter_amount = trunc(delay * 0.1) - trunc(delay + (:rand.uniform() - 0.5) * jitter_amount) - else - trunc(delay) - end + ) :: pos_integer() + defp backoff_delay(false, attempt, base_delay, max_delay, _prev_delay) do + Jitter.no_jitter(attempt - 1, base: base_delay, cap: max_delay) + end + + defp backoff_delay(:full, attempt, base_delay, max_delay, _prev_delay) do + Jitter.full(attempt - 1, base: base_delay, cap: max_delay) + end + + defp backoff_delay(:equal, attempt, base_delay, max_delay, _prev_delay) do + Jitter.equal(attempt - 1, base: base_delay, cap: max_delay) + end + + defp backoff_delay(:decorrelated, _attempt, base_delay, max_delay, prev_delay) do + Jitter.decorrelated(prev_delay, base: base_delay, cap: max_delay) end defmodule Utils do diff --git a/mix.exs b/mix.exs index 031f151..eff6a8e 100644 --- a/mix.exs +++ b/mix.exs @@ -54,6 +54,8 @@ defmodule ElixirDesignPatterns.MixProject do defp deps do [ + {:jitter, "~> 0.1"}, + # Development and testing {:credo, "~> 1.7", only: [:dev, :test], runtime: false}, {:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false}, diff --git a/mix.lock b/mix.lock index 561c933..0151c80 100644 --- a/mix.lock +++ b/mix.lock @@ -8,6 +8,7 @@ "excoveralls": {:hex, :excoveralls, "0.18.5", "e229d0a65982613332ec30f07940038fe451a2e5b29bce2a5022165f0c9b157e", [:mix], [{:castore, "~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "523fe8a15603f86d64852aab2abe8ddbd78e68579c8525ae765facc5eae01562"}, "file_system": {:hex, :file_system, "1.1.1", "31864f4685b0148f25bd3fbef2b1228457c0c89024ad67f7a81a3ffbc0bbad3a", [:mix], [], "hexpm", "7a15ff97dfe526aeefb090a7a9d3d03aa907e100e262a0f8f7746b78f8f87a5d"}, "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "jitter": {:hex, :jitter, "0.1.0", "7a956b672e63204e2747101349b6cec5bee4262d8defdd74a33ac64858faee3e", [:mix], [], "hexpm", "bdc489666a21822feb208a2f669b8354ae54076dcfa16b04fdc9e0dfece71552"}, "makeup": {:hex, :makeup, "1.2.1", "e90ac1c65589ef354378def3ba19d401e739ee7ee06fb47f94c687016e3713d1", [:mix], [{:nimble_parsec, "~> 1.4", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "d36484867b0bae0fea568d10131197a4c2e47056a6fbe84922bf6ba71c8d17ce"}, "makeup_elixir": {:hex, :makeup_elixir, "1.0.1", "e928a4f984e795e41e3abd27bfc09f51db16ab8ba1aebdba2b3a575437efafc2", [: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", "7284900d412a3e5cfd97fdaed4f5ed389b8f2b4cb49efc0eb3bd10e2febf9507"}, "makeup_erlang": {:hex, :makeup_erlang, "1.1.0", "835f7e60792e08824cda445639555d7bf1bbbddb1b60b306e33cb6f6db24dc74", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "1cd6780fb1dd1a03979abaed0fe82712b0625118fd5257d3ebbf73f960c73c3c"},