Repository: commanded/commanded (elixir)

Issue:
Catch exceptions in Process Manager apply/2
Hi,

I have a PM that has defined the `error/3` callback but when throwing an exception during the `apply/2` command does not trigger the `errors/3` callback

Looking at the code I figured it out that we are not rescuing on apply/2

Handle processing:
https://github.com/commanded/commanded/blob/4a6df4a3fc3e318805574ca9cab5cef0a83617e3/lib/commanded/process_managers/process_manager_instance.ex#L235-L243

Apply processing:
https://github.com/commanded/commanded/blob/4a6df4a3fc3e318805574ca9cab5cef0a83617e3/lib/commanded/process_managers/process_manager_instance.ex#L323

The `interested/3` is also catching exceptions:
https://github.com/commanded/commanded/blob/4a6df4a3fc3e318805574ca9cab5cef0a83617e3/lib/commanded/process_managers/process_router.ex#L292-L398


If you think this is something that we should fix I can work on a PR.


Pull request description:
Capture exception on Process Manager apply/2 and call error/3 callback
Closes #379 

Patch:
diff --git a/lib/commanded/process_managers/process_manager_instance.ex b/lib/commanded/process_managers/process_manager_instance.ex
index 7b52250..da152f8 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -203,18 +203,24 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
         opts = [causation_id: event_id, correlation_id: correlation_id, returning: false]
 
         with :ok <- commands |> List.wrap() |> dispatch_commands(opts, state, event) do
-          process_state = mutate_state(event, state)
+          case mutate_state(event, state) do
+            {:error, error, stacktrace} ->
+              failure_context = build_failure_context(event, context, stacktrace, state)
 
-          state = %State{
-            state
-            | process_state: process_state,
-              last_seen_event: event_number
-          }
+              handle_event_error({:error, error}, event, failure_context, state)
+
+            process_state ->
+              state = %State{
+                state
+                | process_state: process_state,
+                  last_seen_event: event_number
+              }
 
-          :ok = persist_state(event_number, state)
-          :ok = ack_event(event, state)
+              :ok = persist_state(event_number, state)
+              :ok = ack_event(event, state)
 
-          {:noreply, state, idle_timeout}
+              {:noreply, state, idle_timeout}
+          end
         else
           {:stop, reason} ->
             {:stop, reason, state}
@@ -320,7 +326,15 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
       process_state: process_state
     } = state
 
-    process_manager_module.apply(process_state, data)
+    try do
+      process_manager_module.apply(process_state, data)
+    rescue
+      error ->
+        stacktrace = __STACKTRACE__
+        Logger.error(fn -> Exception.format(:error, error, stacktrace) end)
+
+        {:error, error, stacktrace}
+    end
   end
 
   defp dispatch_commands(commands, opts, state, last_event, context \\ %{})
@@ -343,9 +357,15 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
             " failed to dispatch command " <> inspect(command) <> " due to: " <> inspect(error)
         end)
 
+        process_manager_state =
+          case mutate_state(last_event, state) do
+            {:error, _, _} -> state
+            process_manager_state -> process_manager_state
+          end
+
         failure_context = %FailureContext{
           pending_commands: pending_commands,
-          process_manager_state: mutate_state(last_event, state),
+          process_manager_state: process_manager_state,
           last_event: last_event,
           context: context
         }


Test patch:
diff --git a/test/process_managers/process_manager_error_handling_test.exs b/test/process_managers/process_manager_error_handling_test.exs
index ba81808..cc8f572 100644
--- a/test/process_managers/process_manager_error_handling_test.exs
+++ b/test/process_managers/process_manager_error_handling_test.exs
@@ -8,6 +8,7 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
   alias Commanded.ProcessManagers.ErrorAggregate.Events.{
     ProcessError,
     ProcessException,
+    ProcessApplyException,
     ProcessDispatchException
   }
 
@@ -84,6 +85,20 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
     end
   end
 
+  describe "process manager event applying error" do
+    test "should call `error/3` callback on exception", %{
+      process_router: process_router,
+      ref: ref
+    } do
+      send_apply_exception_event(process_router)
+
+      expected_runtime_error = %RuntimeError{message: "an exception"}
+
+      assert_receive {:error, ^expected_runtime_error, _failure_context}
+      assert_receive {:DOWN, ^ref, :process, ^process_router, ^expected_runtime_error}
+    end
+  end
+
   describe "process manager dispatch command error" do
     test "should retry the event until process manager requests stop" do
       {:ok, process_router} = ErrorHandlingProcessManager.start_link()
@@ -209,6 +224,18 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
     ])
   end
 
+  defp send_apply_exception_event(process_router) do
+    process_uuid = UUID.uuid4()
+
+    send_events_to_process_router(process_router, [
+      %ProcessApplyException{
+        process_uuid: process_uuid,
+        reply_to: reply_to(),
+        message: "an exception"
+      }
+    ])
+  end
+
   defp send_dispatch_exception_event(process_router) do
     process_uuid = UUID.uuid4()
 
diff --git a/test/process_managers/support/error/error_aggregate.ex b/test/process_managers/support/error/error_aggregate.ex
index 588f6d9..dcccf35 100644
--- a/test/process_managers/support/error/error_aggregate.ex
+++ b/test/process_managers/support/error/error_aggregate.ex
@@ -51,6 +51,11 @@ defmodule Commanded.ProcessManagers.ErrorAggregate do
       defstruct [:process_uuid, :message, :reply_to]
     end
 
+    defmodule ProcessApplyException do
+      @derive Jason.Encoder
+      defstruct [:process_uuid, :message, :reply_to]
+    end
+
     defmodule ProcessDispatchException do
       @derive Jason.Encoder
       defstruct [:process_uuid, :message, :reply_to]
diff --git a/test/process_managers/support/error/error_handling_process_manager.ex b/test/process_managers/support/error/error_handling_process_manager.ex
index bc12277..49138f1 100644
--- a/test/process_managers/support/error/error_handling_process_manager.ex
+++ b/test/process_managers/support/error/error_handling_process_manager.ex
@@ -15,6 +15,7 @@ defmodule Commanded.ProcessManagers.ErrorHandlingProcessManager do
     ProcessContinued,
     ProcessDispatchException,
     ProcessError,
+    ProcessApplyException,
     ProcessException,
     ProcessStarted
   }
@@ -28,6 +29,7 @@ defmodule Commanded.ProcessManagers.ErrorHandlingProcessManager do
 
   def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
   def interested?(%ProcessError{process_uuid: process_uuid}), do: {:start, process_uuid}
+  def interested?(%ProcessApplyException{process_uuid: process_uuid}), do: {:start, process_uuid}
   def interested?(%ProcessException{process_uuid: process_uuid}), do: {:start, process_uuid}
   def interested?(%ProcessContinued{process_uuid: process_uuid}), do: {:continue, process_uuid}
 
@@ -79,6 +81,13 @@ defmodule Commanded.ProcessManagers.ErrorHandlingProcessManager do
     []
   end
 
+  # Simulate an exception applying an event.
+  def apply(%ErrorHandlingProcessManager{}, %ProcessApplyException{} = event) do
+    %ProcessApplyException{message: message} = event
+
+    raise message
+  end
+
   # Skip events causing errors during event handling
   def error({:error, error}, %ProcessError{} = event, failure_context) do
     %ProcessError{reply_to: reply_to} = event
@@ -97,6 +106,15 @@ defmodule Commanded.ProcessManagers.ErrorHandlingProcessManager do
     {:stop, error}
   end
 
+  # Stop on exceptions during event applying
+  def error({:error, error}, %ProcessApplyException{} = event, failure_context) do
+    %ProcessApplyException{reply_to: reply_to} = event
+
+    reply(reply_to, {:error, error, failure_context})
+
+    {:stop, error}
+  end
+
   # Stop on exceptions during command dispatch.
   def error({:error, error}, %RaiseException{} = command, failure_context) do
     %RaiseException{reply_to: reply_to} = command