Repository: commanded/commanded
Pull request #605 (2024-11-21 18:55:31)

Application-wide event handler error handling
Addresses #604 

## What

Introduce a mechanism for Commanded application level error handling strategies in general. Includes `:stop` and `:backoff` as supported strategies. Also includes a mechanism for custom error handling.

## Why
A classic stumbling block for newer Commanded user and people first going to production is that a single problem with an event, or handler, can result in their entire application crashing. 

We can't change the default behaviour without some nasty surprises, so this PR includes what is hopefully a reasonable strategy that can be opted into for all handlers in an application.


Patch:
diff --git a/lib/application.ex b/lib/application.ex
index b166e3b..ecd9957 100644
--- a/lib/application.ex
+++ b/lib/application.ex
@@ -398,4 +398,9 @@ defmodule Commanded.Application do
   @doc false
   @spec registry_adapter(Commanded.Application.t()) :: {module, map}
   def registry_adapter(application), do: Config.get(application, :registry)
+
+  @doc false
+  @spec on_event_handler_error(Commanded.Application.t()) :: atom | module
+  def on_event_handler_error(application),
+    do: Config.get(application, :on_event_handler_error)
 end
diff --git a/lib/commanded/application/config.ex b/lib/commanded/application/config.ex
index a71373b..6553747 100644
--- a/lib/commanded/application/config.ex
+++ b/lib/commanded/application/config.ex
@@ -15,6 +15,14 @@ defmodule Commanded.Application.Config do
     application |> lookup() |> Keyword.get(setting)
   end
 
+  @doc """
+  Put a value into the application's config. This is test induced damage and you
+  probably shouldn't be using it.
+  """
+  def __put__(application, setting, value) when is_atom(application) and is_atom(setting) do
+    GenServer.call(__MODULE__, {:put, application, setting, value})
+  end
+
   @impl GenServer
   def init(:ok) do
     table = :ets.new(__MODULE__, [:named_table, read_concurrency: true])
@@ -30,6 +38,17 @@ defmodule Commanded.Application.Config do
     {:reply, :ok, table}
   end
 
+  @impl GenServer
+  def handle_call({:put, application, settings, value}, _from, table) do
+    config =
+      lookup(application)
+      |> Keyword.put(settings, value)
+
+    true = :ets.update_element(table, application, {4, config})
+
+    {:reply, :ok, table}
+  end
+
   @impl GenServer
   def handle_info({:DOWN, ref, _type, pid, _reason}, table) do
     [[application]] = :ets.match(table, {:"$1", pid, ref, :_})
diff --git a/lib/commanded/event/error_handler.ex b/lib/commanded/event/error_handler.ex
new file mode 100644
index 0000000..187cd92
--- /dev/null
+++ b/lib/commanded/event/error_handler.ex
@@ -0,0 +1,39 @@
+defmodule Commanded.Event.ErrorHandler do
+  alias Commanded.Event.FailureContext
+
+  @doc """
+    Stop the `Commanded.Event.Handler` with the reason given.
+  """
+  def stop_on_error({:error, reason}, _failed_event, _failure_context) do
+    {:stop, reason}
+  end
+
+  @doc """
+    Backoff exponentially when an error is encountered.
+    * Adds up to 1 second of jitter.
+    * Minimum of 1 second delay
+    * Maximum of 24 hour delay
+  """
+  def backoff(
+        _error,
+        _event,
+        %FailureContext{} = failure_context,
+        jitter_fn \\ &one_second_jitter/0
+      ) do
+    %FailureContext{context: context} = failure_context
+    context = Map.update(context, :failures, 1, &(&1 + 1))
+    failures = Map.fetch!(context, :failures)
+
+    base_delay = failures ** 2 * 1000
+    delay = base_delay + jitter_fn.()
+    delay = max(delay, :timer.seconds(1))
+    delay = min(delay, :timer.hours(24))
+
+    failure_context = %FailureContext{context: context}
+    {:retry, delay, failure_context}
+  end
+
+  defp one_second_jitter do
+    :rand.uniform(1000)
+  end
+end
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 7f23d2a..0df44af 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -331,6 +331,7 @@ defmodule Commanded.Event.Handler do
 
   require Logger
 
+  alias Commanded.Event.ErrorHandler
   alias Commanded.Event.FailureContext
   alias Commanded.Event.Handler
   alias Commanded.Event.Upcast
@@ -679,13 +680,10 @@ defmodule Commanded.Event.Handler do
 
   @doc false
   defmacro __before_compile__(_env) do
-    # Include default `handle/2` and `error/3` callback functions in module
+    # Include default `handle/2` callback function in module
     quote generated: true do
       @doc false
       def handle(_event, _metadata), do: :ok
-
-      @doc false
-      def error({:error, reason}, _failed_event, _failure_context), do: {:stop, reason}
     end
   end
 
@@ -1026,9 +1024,8 @@ defmodule Commanded.Event.Handler do
          %Handler{} = state
        ) do
     %RecordedEvent{data: data} = failed_event
-    %Handler{handler_module: handler_module} = state
 
-    case handler_module.error(error, data, failure_context) do
+    case on_error(state, error, data, failure_context) do
       {:retry, %FailureContext{context: context}} when is_map(context) ->
         # Retry the failed event
         Logger.info(describe(state) <> " is retrying failed event")
@@ -1080,6 +1077,25 @@ defmodule Commanded.Event.Handler do
     end
   end
 
+  defp on_error(%Handler{} = state, error, data, failure_context) do
+    %Handler{application: application, handler_module: handler_module} = state
+
+    if function_exported?(handler_module, :error, 3) do
+      handler_module.error(error, data, failure_context)
+    else
+      case Commanded.Application.on_event_handler_error(application) do
+        default when default in [nil, :stop] ->
+          ErrorHandler.stop_on_error(error, data, failure_context)
+
+        :backoff ->
+          ErrorHandler.backoff(error, data, failure_context)
+
+        module when is_atom(module) ->
+          module.error(error, data, failure_context)
+      end
+    end
+  end
+
   defp log_event_error(error, %RecordedEvent{} = failed_event, %Handler{} = state) do
     reason =
       case error do


Test patch:
diff --git a/test/commanded/event/error_handler_test.exs b/test/commanded/event/error_handler_test.exs
new file mode 100644
index 0000000..cbc2905
--- /dev/null
+++ b/test/commanded/event/error_handler_test.exs
@@ -0,0 +1,82 @@
+defmodule Commanded.Event.ErrorHandlerTest do
+  use ExUnit.Case
+
+  alias Commanded.Event.ErrorHandler
+  alias Commanded.Event.FailureContext
+
+  defmodule Event do
+    defstruct [:id]
+  end
+
+  test "Stop on error" do
+    context = %FailureContext{}
+    error = {:error, "an error"}
+
+    result = ErrorHandler.stop_on_error(error, an_event(), context)
+    assert result == {:stop, "an error"}
+  end
+
+  describe "backoff" do
+    test "the first delay is 1 second" do
+      context = %FailureContext{context: %{}}
+
+      {:retry, delay, context} =
+        ErrorHandler.backoff(an_error(), an_event(), context, no_jitter())
+
+      assert context.context.failures == 1
+      assert delay == :timer.seconds(1)
+    end
+
+    test "applies jitter to delay" do
+      jitter_fn = fn -> 234 end
+
+      {:retry, delay, _} =
+        ErrorHandler.backoff(an_error(), an_event(), failure_context(), jitter_fn)
+
+      assert delay == 1234
+    end
+
+    test "backs off exponentially-ish" do
+      failure_context = %FailureContext{context: %{}}
+
+      expectations = [
+        {1, 1_000},
+        {2, 4_000},
+        {3, 9_000},
+        {4, 16_000},
+        {5, 25_000},
+        {6, 36_000},
+        {7, 49_000},
+        {8, 64_000},
+        {9, 81_000},
+        {10, 100_000}
+      ]
+
+      Enum.reduce(expectations, failure_context, fn {expected_failures, expected_delay}, ctx ->
+        {:retry, actual_delay, context} =
+          ErrorHandler.backoff(an_error(), an_event(), ctx, no_jitter())
+
+        assert actual_delay == expected_delay
+        actual_failures = Map.fetch!(context.context, :failures)
+        assert actual_failures == expected_failures
+
+        context
+      end)
+    end
+
+    test "maxes out at 1 day" do
+      # We should hit the max after 294 failures
+      context = %FailureContext{context: %{failures: 500}}
+
+      {:retry, actual_delay, _context} =
+        ErrorHandler.backoff(an_error(), an_event(), context, no_jitter())
+
+      assert actual_delay == :timer.hours(24)
+    end
+  end
+
+  defp an_event, do: %Event{id: 123}
+  defp an_error, do: {:error, "Invalid pizza toppings"}
+  defp no_jitter, do: fn -> 0 end
+  defp failure_context, do: %FailureContext{context: %{}}
+end
diff --git a/test/event/event_handler_error_handling_test.exs b/test/event/event_handler_error_handling_test.exs
index fb5f2d0..079d61f 100644
--- a/test/event/event_handler_error_handling_test.exs
+++ b/test/event/event_handler_error_handling_test.exs
@@ -3,6 +3,10 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
 
   import ExUnit.CaptureLog
 
+  alias Commanded.Application.Config, as: AppConfig
+
+  alias Commanded.DefaultApp
+
   alias Commanded.Event.ErrorAggregate.Events.{
     ErrorEvent,
     ExceptionEvent,
@@ -12,6 +16,8 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
   alias Commanded.Event.ErrorEventHandler
   alias Commanded.Event.FailureContext
   alias Commanded.Event.Handler
+  alias Commanded.Event.SimpleErrorEventHandler
+  alias Commanded.Event.ThreeStrikesErrorHandler
   alias Commanded.Helpers.EventFactory
 
   setup do
@@ -25,6 +31,53 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
     ]
   end
 
+  describe "Configured error handling" do
+    setup [:listen_for_telemetry_events, :start_simple_error_handler]
+
+    test ":stop stops the handler", %{handler: handler, ref: ref} do
+      AppConfig.__put__(DefaultApp, :on_event_handler_error, :stop)
+
+      send_error_event(handler)
+
+      assert_receive {:DOWN, ^ref, :process, ^handler, :failed}
+    end
+
+    test ":backoff delays the next attempt", %{handler: handler} do
+      AppConfig.__put__(DefaultApp, :on_event_handler_error, :backoff)
+
+      # When we sent the error event
+      send_error_event(handler)
+
+      # Then the first attempt and failure occur
+      assert_receive {[:commanded, :event, :handle, :start], 1, _, %{context: %{}}}
+      assert_receive {[:commanded, :event, :handle, :stop], 2, _, %{context: %{}}}
+
+      # And then the next one is received up to 2 seconds later
+      assert_receive {[:commanded, :event, :handle, :start], 3, _, %{context: %{failures: 1}}},
+                     2100
+
+      assert_receive {[:commanded, :event, :handle, :stop], 4, _, %{context: %{failures: 1}}},
+                     2100
+    end
+
+    test "error handler can be a custom module", %{handler: handler, ref: ref} do
+      AppConfig.__put__(DefaultApp, :on_event_handler_error, ThreeStrikesErrorHandler)
+
+      send_error_event(handler)
+
+      assert_receive {[:commanded, :event, :handle, :start], 1, _, %{context: %{}}}
+      assert_receive {[:commanded, :event, :handle, :stop], 2, _, %{context: %{}}}
+
+      assert_receive {[:commanded, :event, :handle, :start], 3, _, %{context: %{attempts: 1}}}
+      assert_receive {[:commanded, :event, :handle, :stop], 4, _, %{context: %{attempts: 1}}}
+
+      assert_receive {[:commanded, :event, :handle, :start], 5, _, %{context: %{attempts: 2}}}
+      assert_receive {[:commanded, :event, :handle, :stop], 6, _, %{context: %{attempts: 2}}}
+
+      assert_receive {:DOWN, ^ref, :process, ^handler, :too_many}
+    end
+  end
+
   describe "event handling exception handling" do
     test "should print the stack trace", %{handler: handler, ref: ref} do
       send_error_message = fn ->
@@ -180,4 +233,45 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
   end
 
   defp reply_to, do: :erlang.pid_to_list(self())
+
+  def listen_for_telemetry_events(_) do
+    agent = start_supervised!({Agent, fn -> 1 end})
+    handler_id = :"#{__MODULE__}-handler"
+
+    events = [
+      [:commanded, :event, :handle, :start],
+      [:commanded, :event, :handle, :stop],
+      [:commanded, :event, :handle, :exception]
+    ]
+
+    increment = fn n -> {n, n + 1} end
+
+    :telemetry.attach_many(
+      handler_id,
+      events,
+      fn event_name, measurements, metadata, reply_to ->
+        if Process.alive?(agent) do
+          num = Agent.get_and_update(agent, increment)
+          send(reply_to, {event_name, num, measurements, metadata})
+        end
+      end,
+      self()
+    )
+
+    on_exit(fn ->
+      :telemetry.detach(handler_id)
+    end)
+  end
+
+  defp start_simple_error_handler(_) do
+    start_supervised!(DefaultApp)
+    handler = start_supervised!(SimpleErrorEventHandler)
+    true = Process.unlink(handler)
+    ref = Process.monitor(handler)
+
+    [
+      handler: handler,
+      ref: ref
+    ]
+  end
 end
diff --git a/test/event/support/error/simple_error_event_handler.ex b/test/event/support/error/simple_error_event_handler.ex
new file mode 100644
index 0000000..7798d43
--- /dev/null
+++ b/test/event/support/error/simple_error_event_handler.ex
@@ -0,0 +1,27 @@
+defmodule Commanded.Event.SimpleErrorEventHandler do
+  @moduledoc false
+  use Commanded.Event.Handler,
+    application: Commanded.DefaultApp,
+    name: __MODULE__
+
+  alias Commanded.Event.ErrorAggregate.Events.{
+    ErrorEvent,
+    ExceptionEvent,
+    InvalidReturnValueEvent
+  }
+
+  # Simulate event handling error reply
+  def handle(%ErrorEvent{}, _metadata) do
+    {:error, :failed}
+  end
+
+  # Simulate event handling exception
+  def handle(%ExceptionEvent{}, _metadata) do
+    raise "exception"
+  end
+
+  # Simulate event handling returning an invalid value
+  def handle(%InvalidReturnValueEvent{}, _metadata) do
+    nil
+  end
+end
diff --git a/test/event/support/error/three_strikes_error_handler.ex b/test/event/support/error/three_strikes_error_handler.ex
new file mode 100644
index 0000000..f206c41
--- /dev/null
+++ b/test/event/support/error/three_strikes_error_handler.ex
@@ -0,0 +1,11 @@
+defmodule Commanded.Event.ThreeStrikesErrorHandler do
+  def error({:error, :failed}, _event, %{context: context}) do
+    attempts = Map.get(context, :attempts, 1)
+
+    if attempts >= 3 do
+      {:stop, :too_many}
+    else
+      {:retry, Map.update(context, :attempts, 1, &(&1 + 1))}
+    end
+  end
+end