Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 31 additions & 21 deletions guides/04_task_async.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion guides/08_circuit_breaker.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
22 changes: 19 additions & 3 deletions lib/patterns/registry_dynamic_supervisor.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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.
"""
Expand Down
62 changes: 46 additions & 16 deletions lib/patterns/task_async.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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 """
Expand Down Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions mix.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand Down
1 change: 1 addition & 0 deletions mix.lock
Original file line number Diff line number Diff line change
Expand Up @@ -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"},
Expand Down
Loading