Repository: slashdotdash/commanded
Pull request #94 (2017-10-13 13:52:51)

Process manager command dispatch error handling
Implements process manager error handling as described in #93.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a720b31..739d397 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,12 @@
 # Changelog
 
+## Next release
+
+### Enhancements
+
+- Process manager command dispatch error handling ([#93](https://github.com/slashdotdash/commanded/issues/93)).
+- Event handlers may define an `init/0` callback function to start any related processes. It must return `:ok`, otherwise the handler process will be stopped.
+
 ## v0.14.0
 
 ### Enhancements
diff --git a/guides/Commands.md b/guides/Commands.md
index ad664dd..df98a0c 100644
--- a/guides/Commands.md
+++ b/guides/Commands.md
@@ -248,7 +248,7 @@ defmodule BankingRouter do
   middleware AuthorizeCommand
 
   identify BankAccount, by: :account_number
-  
+
   dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount
   dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount
 end
diff --git a/guides/Events.md b/guides/Events.md
index a835ac7..af7f2a8 100644
--- a/guides/Events.md
+++ b/guides/Events.md
@@ -22,6 +22,12 @@ Use pattern matching to match on each type of event you are interested in. A cat
 defmodule AccountBalanceHandler do
   use Commanded.Event.Handler, name: "account_balance"
 
+  def init do
+    with {:ok, _} <- Agent.start_link(fn -> 0 end, name: AccountBalance) do
+      :ok
+    end
+  end
+
   def handle(%BankAccountOpened{initial_balance: initial_balance}, _metadata) do
     Agent.update(AccountBalance, fn _ -> initial_balance end)
   end
@@ -39,7 +45,6 @@ end
 The name given to the event handler **must be** unique and remain unchanged between releases. It is used when subscribing to the event store to track which events the handler has seen during restarts.
 
 ```elixir
-{:ok, _balance} = Agent.start_link(fn -> 0 end, name: AccountBalance)
 {:ok, _handler} = AccountBalanceHandler.start_link()
 ```
 
diff --git a/guides/Process Managers.md b/guides/Process Managers.md
index 5925400..0efb7ca 100644
--- a/guides/Process Managers.md
+++ b/guides/Process Managers.md
@@ -2,16 +2,16 @@
 
 A process manager is responsible for coordinating one or more aggregate roots. It handles events and dispatches commands in response. Process managers have state that can be used to track which aggregate roots are being orchestrated.
 
-Use the `Commanded.ProcessManagers.ProcessManager` macro in your process manager module and implement the three callback functions defined in the behaviour: `interested?/1`, `handle/2`, and `apply/2`.
+Use the `Commanded.ProcessManagers.ProcessManager` macro in your process manager module and implement the callback functions defined in the behaviour: `interested?/1`, `handle/2`, `apply/2`, and `error/4`.
 
 ## `interested?/1`
 
-The `interested?/1` function is used to indicate which events the process manager receives. The response is used to route the event to an existing instance or start a new process instance.
+The `interested?/1` function is used to indicate which events the process manager receives. The response is used to route the event to an existing instance or start a new process instance:
 
-- Return `{:start, process_uuid}` to create a new instance of the process manager.
-- Return `{:continue, process_uuid}` to continue execution of an existing process manager.
-- Return `{:stop, process_uuid}` to stop an existing process manager and shutdown its process.
-- Return `false` to ignore the event.
+- `{:start, process_uuid}` - create a new instance of the process manager.
+- `{:continue, process_uuid}` - continue execution of an existing process manager.
+- `{:stop, process_uuid}` - stop an existing process manager and shutdown its process.
+- `false` - ignore the event.
 
 ## `handle/2`
 
@@ -21,6 +21,53 @@ A `handle/2` function must exist for each `:start` and `:continue` tagged event
 
 The `apply/2` function is used to mutate the process manager's state. It receives its current state and the interested event. It must return the modified state.
 
+## `error/4`
+
+You can define an `error/4` callback function to handle any errors returned from command dispatch. The function is passed the command dispatch error (e.g. `{:error, :failure}`), the failed command, any pending commands, and a context map containing state passed between retries. Use pattern matching on the error and/or failed command to explicitly handle certain errors or commands.
+
+You can return one of the following responses depending upon the error severity:
+
+- `{:retry, context}` - retry the failed command, provide a context map containing any state passed to subsequent failures. This could be used to count the number of retries, failing after too many attempts.
+
+- `{:retry, delay, context}` - retry the failed command, after sleeping for the requested delay (in milliseconds). Context is a map as described in `{:retry, context}` above.
+
+- `{:skip, :discard_pending}` - discard the failed command and any pending commands.
+
+- `{:skip, :continue_pending}` - skip the failed command, but continue dispatching any pending commands.
+
+- `{:continue, commands, context}` - continue dispatching the given commands. This allows you to retry the failed command, modify it and retry, drop it, or drop all pending commands by passing an empty list `[]`.
+
+- `{:stop, reason}` - stop the process manager with the given reason.
+
+### Error handling example
+
+```elixir
+defmodule ExampleProcessManager do
+  use Commanded.ProcessManagers.ProcessManager,
+    name: "ExampleProcessManager",
+    router: ExampleRouter
+
+  # stop process manager after three attempts
+  def error({:error, _failure}, _failed_command, _pending_commands, %{attempts: attempts} = context)
+    when attempts >= 2
+  do
+    {:stop, :too_many_attempts}
+  end
+
+  # retry command, record attempt count in context map
+  def error({:error, _failure}, _failed_command, _pending_commands, context) do
+    context = Map.update(context, :attempts, 1, fn attempts -> attempts + 1 end)
+    {:retry, context}
+  end
+end
+```
+
+The default behaviour if you don't provide an `error/4` callback is to stop the process manager using the same error reason returned from the failed command dispatch.
+
+You should supervise process managers to ensure they are correctly restarted on error.
+
+## Example process manager
+
 ```elixir
 defmodule TransferMoneyProcessManager do
   use Commanded.ProcessManagers.ProcessManager,
diff --git a/guides/Read Model Projections.md b/guides/Read Model Projections.md
index 56a9e0d..2b58150 100644
--- a/guides/Read Model Projections.md
+++ b/guides/Read Model Projections.md
@@ -36,7 +36,6 @@ defmodule MyApp.ExampleProjector do
     name: "ExampleProjector",
     consistency: :strong
 
-
   # ...
 end
 ```
diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex
index baa3564..ffa798f 100644
--- a/lib/commanded/commands/router.ex
+++ b/lib/commanded/commands/router.ex
@@ -266,7 +266,7 @@ defmodule Commanded.Commands.Router do
         metadata = Keyword.get(opts, :metadata) || @default_metadata
         timeout = Keyword.get(opts, :timeout) || unquote(timeout) || @default_dispatch_timeout
         include_aggregate_version = Keyword.get(opts, :include_aggregate_version) || @include_aggregate_version
-        lifespan = unquote(lifespan) || @default_lifespan
+        lifespan = Keyword.get(opts, :lifespan) || unquote(lifespan) || @default_lifespan
 
         default_identity = unquote(identity)
         default_identity_prefix = unquote(identity_prefix)
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 07880ef..e290851 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -87,12 +87,21 @@ defmodule Commanded.Event.Handler do
   @type subscribe_from :: :origin | :current | non_neg_integer
   @type consistency :: :eventual | :strong
 
+  @doc """
+  Optional initialisation callback function called when the handler starts.
+
+  Can be used to start any related processes when the event handler is started.
+
+  Return `:ok` on success, or `{:stop, reason}` to stop the handler process.
+  """
+  @callback init() :: :ok | {:stop, reason :: any()}
+
   @doc """
   Event handler behaviour to handle a domain event and its metadata
 
   Return `:ok` on success, `{:error, :already_seen_event}` to ack and skip the event, or `{:error, reason}` on failure.
   """
-  @callback handle(domain_event, metadata) :: :ok | {:error, reason :: atom}
+  @callback handle(domain_event, metadata) :: :ok | {:error, reason :: any()}
 
   @doc """
   Macro as a convenience for defining an event handler.
@@ -102,6 +111,11 @@ defmodule Commanded.Event.Handler do
       defmodule ExampleHandler do
         use Commanded.Event.Handler, name: "ExampleHandler"
 
+        def init do
+          # ... optional initialisation
+          :ok
+        end
+
         def handle(%AnEvent{...}, _metadata) do
           # ...
         end
@@ -110,6 +124,7 @@ defmodule Commanded.Event.Handler do
   Start event handler process (or configure as a worker inside a [supervisor](supervision.html)):
 
       {:ok, handler} = ExampleHandler.start_link()
+
   """
   defmacro __using__(opts) do
     quote location: :keep do
@@ -129,6 +144,11 @@ defmodule Commanded.Event.Handler do
 
         Commanded.Event.Handler.start_link(@name, __MODULE__, opts)
       end
+
+      @doc false
+      def init, do: :ok
+
+      defoverridable [init: 0]
     end
   end
 
@@ -160,30 +180,34 @@ defmodule Commanded.Event.Handler do
     })
   end
 
-  def init(%Handler{} = state) do
-    GenServer.cast(self(), {:subscribe_to_events})
-    {:ok, state}
+  def init(%Handler{handler_module: handler_module} = state) do
+    GenServer.cast(self(), :subscribe_to_events)
+
+    reply =
+      case handler_module.init() do
+        :ok -> :ok
+        {:stop, _reason} = reply -> reply
+      end
+
+    {reply, state}
   end
 
-  def handle_cast({:subscribe_to_events}, %Handler{} = state) do
+  def handle_cast(:subscribe_to_events, %Handler{} = state) do
     {:noreply, subscribe_to_all_streams(state)}
   end
 
-  def handle_info({:events, events}, state) do
-    Logger.debug(fn -> "event handler received events: #{inspect events}" end)
-
-    state = Enum.reduce(events, state, fn (event, state) ->
-      event_number = extract_event_number(event)
-      data = extract_data(event)
-      metadata = extract_metadata(event)
-
-      case handle_event(event_number, data, metadata, state) do
-        :ok -> confirm_receipt(event, state)
-        {:error, :already_seen_event} -> confirm_receipt(event, state)
-      end
-    end)
-
-    {:noreply, state}
+  def handle_info({:events, events}, %Handler{} = state) do
+    Logger.debug(fn -> describe(state) <> " received events: #{inspect events}" end)
+
+    try do
+      state = Enum.reduce(events, state, &handle_event/2)
+      
+      {:noreply, state}
+    catch
+      {:error, reason} ->
+        # stop after event handling returned an error
+        {:stop, reason, state}
+    end
   end
 
   defp subscribe_to_all_streams(%Handler{consistency: consistency, handler_name: handler_name, subscribe_from: subscribe_from} = state) do
@@ -196,21 +220,30 @@ defmodule Commanded.Event.Handler do
   end
 
   # ignore already seen events
-  defp handle_event(event_number, _data, _metadata, %Handler{last_seen_event: last_seen_event})
+  defp handle_event(%RecordedEvent{event_number: event_number} = event, %Handler{last_seen_event: last_seen_event} = state)
     when not is_nil(last_seen_event) and event_number <= last_seen_event
   do
-    Logger.debug(fn -> "event handler has already seen event: #{inspect event_number}" end)
-    {:error, :already_seen_event}
+    Logger.debug(fn -> describe(state) <> " has already seen event ##{inspect event_number}" end)
+
+    confirm_receipt(event, state)
   end
 
   # delegate event to handler module
-  defp handle_event(_event_number, data, metadata, %Handler{handler_module: handler_module}) do
-    handler_module.handle(data, metadata)
+  defp handle_event(%RecordedEvent{data: data} = event, %Handler{handler_module: handler_module} = state) do
+    case handler_module.handle(data, enrich_metadata(event)) do
+      :ok ->
+        confirm_receipt(event, state)
+
+      {:error, reason} = error ->
+        Logger.error(fn -> describe(state) <> " failed to handle event #{inspect event} due to: #{inspect reason}" end)
+
+        throw(error)
+    end
   end
 
   # confirm receipt of event
   defp confirm_receipt(%RecordedEvent{event_number: event_number} = event, %Handler{} = state) do
-    Logger.debug(fn -> "event handler confirming receipt of event: #{inspect event_number}" end)
+    Logger.debug(fn -> describe(state) <> " confirming receipt of event ##{inspect event_number}" end)
 
     ack_event(event, state)
 
@@ -222,12 +255,12 @@ defmodule Commanded.Event.Handler do
     Subscriptions.ack_event(handler_name, consistency, event)
   end
 
-  defp extract_event_number(%RecordedEvent{event_number: event_number}), do: event_number
-
-  defp extract_data(%RecordedEvent{data: data}), do: data
-
-  defp extract_metadata(%RecordedEvent{metadata: nil} = event), do: extract_metadata(%RecordedEvent{event | metadata: %{}})
-  defp extract_metadata(%RecordedEvent{event_number: event_number, stream_id: stream_id, stream_version: stream_version, metadata: metadata, created_at: created_at}) do
-    Map.merge(%{event_number: event_number, stream_id: stream_id, stream_version: stream_version, created_at: created_at}, metadata)
+  defp enrich_metadata(%RecordedEvent{metadata: metadata} = event) do
+    event
+    |> Map.from_struct()
+    |> Map.take([:event_number, :stream_id, :stream_version, :created_at])
+    |> Map.merge(metadata || %{})
   end
+
+  defp describe(%Handler{handler_module: handler_module}), do: inspect(handler_module)
 end
diff --git a/lib/commanded/process_managers/process_manager.ex b/lib/commanded/process_managers/process_manager.ex
index 05803a2..a812e2e 100644
--- a/lib/commanded/process_managers/process_manager.ex
+++ b/lib/commanded/process_managers/process_manager.ex
@@ -7,19 +7,19 @@ defmodule Commanded.ProcessManagers.ProcessManager do
   state that can be used to track which aggregate roots are being orchestrated.
 
   Use the `Commanded.ProcessManagers.ProcessManager` macro in your process
-  manager module and implement the three callback functions defined in the
-  behaviour:
+  manager module and implement the callback functions defined in the behaviour:
 
   - `c:interested?/1`
   - `c:handle/2`
   - `c:apply/2`
+  - `c:error/4`
 
   ## Example
 
       defmodule ExampleProcessManager do
         use Commanded.ProcessManagers.ProcessManager,
           name: "ExampleProcessManager",
-          router: BankRouter
+          router: ExampleRouter
 
         def interested?(%AnEvent{...}) do
           # ...
@@ -32,12 +32,53 @@ defmodule Commanded.ProcessManagers.ProcessManager do
         def apply(%ExampleProcessManager{...}, %AnEvent{...}) do
           # ...
         end
+
+        def error({:error, failure}, %ExampleCommand{}, _pending_commands, %{} = context) do
+          # retry, skip, ignore, or stop process manager on error dispatching command
+        end
       end
 
   Start the process manager (or configure as a worker inside a [Supervisor](supervision.html))
 
       {:ok, process_manager} = ExampleProcessManager.start_link()
 
+  # Error handling
+
+  You can define an `c:error/4` callback function to handle any errors returned
+  from command dispatch. The function is passed the command dispatch error (e.g.
+  `{:error, :failure}`), the failed command, any pending commands, and a context
+  map containing state passed between retries.
+
+  Use pattern matching on the error and/or failed command to explicitly handle
+  certain errors or commands. You can choose to retry, skip, ignore, or stop the
+  process manager after an error dispatching command.
+
+  The default behaviour if you don't provide an `c:error/4` callback is to stop
+  the process manager using the exact error reason returned from the command
+  dispatch. You should supervise process managers to ensure they are correctly
+  restarted on error.
+
+  ## Example
+
+      defmodule ExampleProcessManager do
+        use Commanded.ProcessManagers.ProcessManager,
+          name: "ExampleProcessManager",
+          router: ExampleRouter
+
+        # stop process manager after three attempts
+        def error({:error, _failure}, _failed_command, _pending_commands, %{attempts: attempts} = context)
+          when attempts >= 2
+        do
+          {:stop, :too_many_attempts}
+        end
+
+        # retry command, record attempt count in context map
+        def error({:error, _failure}, _failed_command, _pending_commands, context) do
+          context = Map.update(context, :attempts, 1, fn attempts -> attempts + 1 end)
+          {:retry, context}
+        end
+      end
+
   # Consistency
 
   For each process manager you can define its consistency, as one of either
@@ -78,34 +119,77 @@ defmodule Commanded.ProcessManagers.ProcessManager do
   @doc """
   Is the process manager interested in the given command?
 
-  The `c:interested?/1` function is used to indicate which events the process manager receives.
-  The response is used to route the event to an existing instance or start a new process instance.
+  The `c:interested?/1` function is used to indicate which events the process
+  manager receives. The response is used to route the event to an existing
+  instance or start a new process instance:
 
-  - Return `{:start, process_uuid}` to create a new instance of the process manager.
-  - Return `{:continue, process_uuid}` to continue execution of an existing process manager.
-  - Return `{:stop, process_uuid}` to stop an existing process manager and shutdown its process.
-  - Return `false` to ignore the event.
+  - `{:start, process_uuid}` - create a new instance of the process manager.
+  - `{:continue, process_uuid}` - continue execution of an existing process manager.
+  - `{:stop, process_uuid}` - stop an existing process manager and shutdown its process.
+  - `false` - ignore the event.
   """
   @callback interested?(domain_event) :: {:start, process_uuid} | {:continue, process_uuid} | {:stop, process_uuid} | false
 
   @doc """
-  Process manager instance handles the domain event, returning commands to dispatch
+  Process manager instance handles the domain event, returning commands to dispatch.
 
-  A `c:handle/2` function must exist for each `:start` and `:continue` tagged event previously specified.
-  It receives the process manager’s state and the event to be handled.
-  It must return the commands to be dispatched. This may be none, a single command, or many commands.
+  A `c:handle/2` function must exist for each `:start` and `:continue` tagged
+  event previously specified. It receives the process manager’s state and the
+  event to be handled. It must return the commands to be dispatched. This may be
+  none, a single command, or many commands.
   """
   @callback handle(process_manager, domain_event) :: list(command)
 
   @doc """
-  Mutate the process manager's state by applying the domain event
+  Mutate the process manager's state by applying the domain event.
 
-  The `c:apply/2` function is used to mutate the process manager’s state.
-  It receives its current state and the interested event.
-  It must return the modified state.
+  The `c:apply/2` function is used to mutate the process manager’s state. It
+  receives its current state and the interested event. It must return the
+  modified state.
   """
   @callback apply(process_manager, domain_event) :: process_manager
 
+  @doc """
+  Called when a command dispatch returns an error.
+
+  The `c:error/4` function allows you to control how command dispatch failures
+  are handled. The function is passed the command dispatch error (e.g. `{:error,
+  :failure}`), the failed command, any pending commands, and a context map
+  containing state passed between retries. The context may also be used to track
+  state between retried failures.
+
+  You can return one of the following responses depending upon the
+  error severity:
+
+  - `{:retry, context}` - retry the failed command, provide a context
+    map containing any state passed to subsequent failures. This could be used
+    to count the number of retries, failing after too many attempts.
+
+  - `{:retry, delay, context}` - retry the failed command, after sleeping for
+    the requested delay (in milliseconds). Context is a map as described in
+    `{:retry, context}` above.
+
+  - `{:skip, :discard_pending}` - discard the failed command and any pending
+    commands.
+
+  - `{:skip, :continue_pending}` - skip the failed command, but continue
+    dispatching any pending commands.
+
+  - `{:continue, commands, context}` - continue dispatching the given commands.
+    This allows you to retry the failed command, modify it and retry, drop it,
+    or drop all pending commands by passing an empty list `[]`. Context is a map
+    as described in `{:retry, context}` above.
+
+  - `{:stop, reason}` - stop the process manager with the given reason.
+
+  """
+  @callback error(error :: term(), failed_command :: command, pending_commands :: list(command), context :: map()) :: {:retry, context :: map()}
+    | {:retry, delay :: non_neg_integer(), context :: map()}
+    | {:skip, :discard_pending}
+    | {:skip, :continue_pending}
+    | {:continue, commands :: list(command), context :: map()}
+    | {:stop, reason :: term()}
+
   @doc false
   defmacro __using__(opts) do
     quote location: :keep do
@@ -140,6 +224,10 @@ defmodule Commanded.ProcessManagers.ProcessManager do
 
       @doc false
       def apply(process_manager, _event), do: process_manager
+
+      @doc false
+      def error({:error, reason}, _failed_command, _pending_commands, _context),
+        do: {:stop, reason}
     end
   end
 end
diff --git a/lib/commanded/process_managers/process_manager_instance.ex b/lib/commanded/process_managers/process_manager_instance.ex
index d9e3e11..3aadb11 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -34,7 +34,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   end
 
   def init(%ProcessManagerInstance{} = state) do
-    GenServer.cast(self(), {:fetch_state})
+    GenServer.cast(self(), :fetch_state)
     {:ok, state}
   end
 
@@ -51,31 +51,33 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   Typically called when it has reached its final state.
   """
   def stop(process_manager) do
-    GenServer.call(process_manager, {:stop})
+    GenServer.call(process_manager, :stop)
   end
 
   @doc """
   Fetch the process state of this instance
   """
   def process_state(process_manager) do
-    GenServer.call(process_manager, {:process_state})
+    GenServer.call(process_manager, :process_state)
   end
 
-  def handle_call({:stop}, _from, %ProcessManagerInstance{} = state) do
+  @doc false
+  def handle_call(:stop, _from, %ProcessManagerInstance{} = state) do
     :ok = delete_state(state)
 
     # stop the process with a normal reason
     {:stop, :normal, :ok, state}
   end
 
-  def handle_call({:process_state}, _from, %ProcessManagerInstance{process_state: process_state} = state) do
+  @doc false
+  def handle_call(:process_state, _from, %ProcessManagerInstance{process_state: process_state} = state) do
     {:reply, process_state, state}
   end
 
   @doc """
   Attempt to fetch intial process state from snapshot storage
   """
-  def handle_cast({:fetch_state}, %ProcessManagerInstance{} = state) do
+  def handle_cast(:fetch_state, %ProcessManagerInstance{} = state) do
     state = case EventStore.read_snapshot(process_state_uuid(state)) do
       {:ok, snapshot} ->
         %ProcessManagerInstance{state |
@@ -93,63 +95,123 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   Handle the given event, using the process manager module, against the current process state
   """
   def handle_cast({:process_event, %RecordedEvent{} = event, process_router}, %ProcessManagerInstance{} = state) do
-    state = case event_already_seen?(event, state) do
+    case event_already_seen?(event, state) do
       true -> process_seen_event(event, process_router, state)
       false -> process_unseen_event(event, process_router, state)
     end
-
-    {:noreply, state}
   end
 
-  defp event_already_seen?(%RecordedEvent{event_number: event_number}, %ProcessManagerInstance{last_seen_event: last_seen_event}) do
+  defp event_already_seen?(
+    %RecordedEvent{event_number: event_number},
+    %ProcessManagerInstance{last_seen_event: last_seen_event})
+  do
     not is_nil(last_seen_event) and event_number <= last_seen_event
   end
 
-  # already seen event, so just ack
   defp process_seen_event(event = %RecordedEvent{}, process_router, state) do
+    # already seen event, so just ack
     :ok = ack_event(event, process_router)
-    state
+
+    {:noreply, state}
   end
 
-  defp process_unseen_event(%RecordedEvent{event_number: event_number} = event, process_router, %ProcessManagerInstance{command_dispatcher: command_dispatcher, process_manager_module: process_manager_module, process_state: process_state} = state) do
-    case handle_event(process_manager_module, process_state, event) do
+  defp process_unseen_event(%RecordedEvent{event_number: event_number} = event, process_router, %ProcessManagerInstance{} = state) do
+    case handle_event(event, state) do
       {:error, reason} ->
-        Logger.warn(fn -> "process manager instance failed to handle event #{inspect event_number} due to: #{inspect reason}" end)
-	      state
-
-      commands ->
-        :ok = dispatch_commands(List.wrap(commands), command_dispatcher)
-
-        process_state = mutate_state(process_manager_module, process_state, event)
-
-        state = %ProcessManagerInstance{state |
-          process_state: process_state,
-          last_seen_event: event_number,
-        }
+        Logger.error(fn -> describe(state) <> " failed to handle event #{inspect event_number} due to: #{inspect reason}" end)
 
-        :ok = persist_state(state, event_number)
-        :ok = ack_event(event, process_router)
+	      {:stop, reason, state}
 
-	      state
+      commands ->
+        with :ok <- commands |> List.wrap() |> dispatch_commands(state) do
+          process_state = mutate_state(event, state)
+
+          state = %ProcessManagerInstance{state |
+            process_state: process_state,
+            last_seen_event: event_number,
+          }
+
+          :ok = persist_state(state, event_number)
+          :ok = ack_event(event, process_router)
+
+          {:noreply, state}
+        else
+          {:stop, reason} ->
+            {:stop, reason, state}
+        end
     end
   end
 
   # process instance is given the event and returns applicable commands (may be none, one or many)
-  defp handle_event(process_manager_module, process_state, %RecordedEvent{data: data}) do
+  defp handle_event(%RecordedEvent{data: data}, %ProcessManagerInstance{process_manager_module: process_manager_module, process_state: process_state}) do
     process_manager_module.handle(process_state, data)
   end
 
   # update the process instance's state by applying the event
-  defp mutate_state(process_manager_module, process_state, %RecordedEvent{data: data}) do
+  defp mutate_state(%RecordedEvent{data: data}, %ProcessManagerInstance{process_manager_module: process_manager_module, process_state: process_state}) do
     process_manager_module.apply(process_state, data)
   end
 
-  defp dispatch_commands([], _command_dispatcher), do: :ok
-  defp dispatch_commands(commands, command_dispatcher) when is_list(commands) do
-    Enum.each(commands, fn command ->
-      Logger.debug(fn -> "process manager instance attempting to dispatch command: #{inspect command}" end)
-      :ok = command_dispatcher.dispatch(command)
-    end)
+  defp dispatch_commands(commands, state, context \\ %{})
+  defp dispatch_commands([], _state, _context), do: :ok
+  defp dispatch_commands([command | pending_commands], %ProcessManagerInstance{command_dispatcher: command_dispatcher} = state, context) do
+    Logger.debug(fn -> describe(state) <> " attempting to dispatch command: #{inspect command}" end)
+
+    case command_dispatcher.dispatch(command) do
+      :ok ->
+        dispatch_commands(pending_commands, state)
+
+      error ->
+        Logger.warn(fn -> describe(state) <> " failed to dispatch command #{inspect command} due to: #{inspect error}" end)
+
+        dispatch_failure(error, command, pending_commands, context, state)
+    end
+  end
+
+  defp dispatch_failure(error, failed_command, pending_commands, context, %ProcessManagerInstance{process_manager_module: process_manager_module} = state) do
+    case process_manager_module.error(error, failed_command, pending_commands, context) do
+      {:continue, commands, context} ->
+        # continue dispatching the given commands
+        Logger.info(fn -> describe(state) <> " is continuing with modified command(s)" end)
+
+        dispatch_commands(commands, state, context)
+
+      {:retry, context} ->
+        # retry the failed command immediately
+        Logger.info(fn -> describe(state) <> " is retrying failed command" end)
+
+        dispatch_commands([failed_command | pending_commands], state, context)
+
+      {:retry, delay, context} ->
+        # retry the failed command after waiting for the given delay, in milliseconds
+        Logger.info(fn -> describe(state) <> " is retrying failed command after #{inspect delay}ms" end)
+
+        :timer.sleep(delay)
+
+        dispatch_commands([failed_command | pending_commands], state, context)
+
+      {:skip, :discard_pending} ->
+        # skip the failed command and discard any pending commands
+        Logger.info(fn -> describe(state) <> " is skipping event and #{length(pending_commands)} pending command(s)" end)
+
+        :ok
+
+      {:skip, :continue_pending} ->
+        # skip the failed command, but continue dispatching any pending commands
+        Logger.info(fn -> describe(state) <> " is ignoring error dispatching command" end)
+
+        dispatch_commands(pending_commands, state)
+
+      {:stop, reason} = reply ->
+        # stop process manager
+        Logger.warn(fn -> describe(state) <> " has requested to stop: #{inspect reason}" end)
+
+        reply
+    end
+  end
+
+  defp describe(%ProcessManagerInstance{process_manager_module: process_manager_module}) do
+    inspect(process_manager_module)
   end
 
   defp persist_state(%ProcessManagerInstance{process_manager_module: process_manager_module, process_state: process_state} = state, source_version) do
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 015749f..5ae7fcf 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -42,7 +42,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
     state = %State{state | supervisor: supervisor}
 
-    GenServer.cast(self(), {:subscribe_to_events})
+    GenServer.cast(self(), :subscribe_to_events)
 
     {:ok, state}
   end
@@ -74,7 +74,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     state = confirm_receipt(event, state)
 
     # continue processing any pending events
-    GenServer.cast(self(), {:process_pending_events})
+    GenServer.cast(self(), :process_pending_events)
 
     {:noreply, state}
   end
@@ -82,21 +82,25 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   @doc """
   Subscribe the process router to all events
   """
-  def handle_cast({:subscribe_to_events}, %State{} = state) do
+  def handle_cast(:subscribe_to_events, %State{} = state) do
     {:noreply, subscribe_to_all_streams(state)}
   end
 
-  def handle_cast({:process_pending_events}, %State{pending_events: []} = state), do: {:noreply, state}
-  def handle_cast({:process_pending_events}, %State{pending_events: [event | pending_events], process_manager_name: process_manager_name} = state) do
-    Logger.debug(fn ->  "process router \"#{process_manager_name}\" has #{length(pending_events)} pending event(s) to process" end)
+  def handle_cast(:process_pending_events, %State{pending_events: []} = state), do: {:noreply, state}
+  def handle_cast(:process_pending_events, %State{pending_events: [event | pending_events]} = state) do
+    case length(pending_events) do
+      0 -> :ok
+      1 -> Logger.debug(fn -> describe(state) <> " has 1 pending event to process" end)
+      count -> Logger.debug(fn -> describe(state) <> " has #{count} pending events to process" end)
+    end
 
     state = handle_event(event, state)
 
     {:noreply, %State{state | pending_events: pending_events}}
   end
 
-  def handle_info({:events, events}, %State{process_manager_name: process_manager_name, pending_events: pending_events} = state) do
-    Logger.debug(fn -> "process router \"#{process_manager_name}\" received #{length(events)} event(s)" end)
+  def handle_info({:events, events}, %State{pending_events: pending_events} = state) do
+    Logger.debug(fn -> describe(state) <> " received #{length(events)} event(s)" end)
 
     unseen_events = Enum.reject(events, &event_already_seen?(&1, state))
 
@@ -107,7 +111,8 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
       {[], _} ->
         # no pending events, but some unseen events so start processing them
-        GenServer.cast(self(), {:process_pending_events})
+        GenServer.cast(self(), :process_pending_events)
+
         %State{state | pending_events: unseen_events}
 
       {_, _} ->
@@ -118,14 +123,17 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, state}
   end
 
-  def handle_info({:DOWN, _ref, :process, pid, reason}, %State{process_managers: process_managers} = state) do
-    if reason != :normal do
-      Logger.warn(fn -> "process manager process down due to: #{inspect reason}" end)
-    end
-
+  # remove a process manager instance that has stopped with a normal exit reason
+  def handle_info({:DOWN, _ref, :process, pid, :normal}, %State{process_managers: process_managers} = state) do
     {:noreply, %State{state | process_managers: remove_process_manager(process_managers, pid)}}
   end
 
+  # stop process router when a process manager instance terminates abnormally
+  def handle_info({:DOWN, _ref, :process, _pid, reason}, %State{} = state) do
+    Logger.warn(fn -> describe(state) <> " is stopping due to: #{inspect reason}" end)
+    {:stop, reason, state}
+  end
+
   defp subscribe_to_all_streams(%State{consistency: consistency, process_manager_name: process_manager_name, subscribe_from: subscribe_from} = state) do
     {:ok, subscription} = EventStore.subscribe_to_all_streams(process_manager_name, self(), subscribe_from)
 
@@ -140,7 +148,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     not is_nil(last_seen_event) and event_number <= last_seen_event
   end
 
-  defp handle_event(%RecordedEvent{event_number: event_number, data: data, stream_id: stream_id, stream_version: stream_version} = event, %State{process_manager_name: process_manager_name, process_manager_module: process_manager_module, process_managers: process_managers} = state) do
+  defp handle_event(%RecordedEvent{event_number: event_number, data: data, stream_id: stream_id, stream_version: stream_version} = event, %State{process_manager_module: process_manager_module, process_managers: process_managers} = state) do
     {process_uuid, process_manager} = case process_manager_module.interested?(data) do
       {:start, process_uuid} -> {process_uuid, start_process_manager(process_uuid, state)}
       {:continue, process_uuid} -> {process_uuid, continue_process_manager(process_uuid, state)}
@@ -150,17 +158,17 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
     case process_uuid do
       nil ->
-        Logger.debug(fn -> "process router \"#{process_manager_name}\" is not interested in event: #{inspect event_number} (#{inspect stream_id}@#{inspect stream_version})" end)
+        Logger.debug(fn -> describe(state) <> " is not interested in event: #{inspect event_number} (#{inspect stream_id}@#{inspect stream_version})" end)
 
         ack_and_continue(event, state)
 
       :stopped ->
-        Logger.debug(fn -> "process manager instance \"#{process_manager_name}\" has been stopped by event: #{inspect event_number} (#{inspect stream_id}@#{inspect stream_version})" end)
+        Logger.debug(fn -> describe(state) <> " has been stopped by event: #{inspect event_number} (#{inspect stream_id}@#{inspect stream_version})" end)
 
         ack_and_continue(event, state)
 
       _ ->
-        Logger.debug(fn -> "process router \"#{process_manager_name}\" is interested in event: #{inspect event_number} (#{inspect stream_id}@#{inspect stream_version})" end)
+        Logger.debug(fn -> describe(state) <> " is interested in event: #{inspect event_number} (#{inspect stream_id}@#{inspect stream_version})" end)
 
         # delegate event to process instance who will ack event processing on success
         :ok = delegate_event(process_manager, event)
@@ -171,14 +179,14 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
   # continue processing any pending events and confirm receipt of the given event id
   defp ack_and_continue(%RecordedEvent{} = event, %State{} = state) do
-    GenServer.cast(self(), {:process_pending_events})
+    GenServer.cast(self(), :process_pending_events)
 
     confirm_receipt(event, state)
   end
 
   # confirm receipt of given event
-  defp confirm_receipt(%RecordedEvent{event_number: event_number} = event, %State{process_manager_name: process_manager_name} = state) do
-    Logger.debug(fn -> "process router \"#{process_manager_name}\" confirming receipt of event: #{inspect event_number}" end)
+  defp confirm_receipt(%RecordedEvent{event_number: event_number} = event, %State{} = state) do
+    Logger.debug(fn -> describe(state) <> " confirming receipt of event: #{inspect event_number}" end)
 
     do_ack_event(event, state)
 
@@ -223,4 +231,8 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   defp delegate_event(process_manager, %RecordedEvent{} = event) do
     ProcessManagerInstance.process_event(process_manager, event, self())
   end
+
+  defp describe(%State{process_manager_module: process_manager_module}) do
+    inspect(process_manager_module)
+  end
 end


Test patch:
diff --git a/test/commands/command_timeout_test.exs b/test/commands/command_timeout_test.exs
index e0d6111..be3d6fb 100644
--- a/test/commands/command_timeout_test.exs
+++ b/test/commands/command_timeout_test.exs
@@ -23,8 +23,7 @@ defmodule Commanded.Commands.CommandTimeoutTest do
   test "should allow timeout to be specified during command registration" do
     # handler is set to take longer than the configured timeout
     case TimeoutRouter.dispatch(%TimeoutCommand{aggregate_uuid: UUID.uuid4, sleep_in_ms: 2_000}) do
-      {:error, :aggregate_execution_failed} -> :ok
-      {:error, :aggregate_execution_failed, _reason} -> :ok
+      {:error, :aggregate_execution_failed}  -> :ok
       {:error, :aggregate_execution_timeout} -> :ok
       reply -> flunk("received an unexpected response: #{inspect reply}")
     end
diff --git a/test/event/event_handler_error_handling_test.exs b/test/event/event_handler_error_handling_test.exs
new file mode 100644
index 0000000..438dca3
--- /dev/null
+++ b/test/event/event_handler_error_handling_test.exs
@@ -0,0 +1,29 @@
+defmodule Commanded.Event.EventHandlerErrorHandlingTest do
+  use Commanded.StorageCase
+
+  defmodule ErrorEventHandler do
+    use Commanded.Event.Handler, name: "ErrorEventHandler"
+
+    alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
+
+    def handle(%BankAccountOpened{}, _metadata), do: {:error, :failed}
+  end
+
+  alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
+  alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
+  alias Commanded.ExampleDomain.BankRouter
+
+  test "should stop event handler on error" do
+    account_number = UUID.uuid4()
+
+    {:ok, handler} = ErrorEventHandler.start_link()
+
+    Process.unlink(handler)
+    ref = Process.monitor(handler)
+
+    :ok = BankRouter.dispatch(%OpenAccount{account_number: account_number, initial_balance: 1_000})
+
+    assert_receive {:DOWN, ^ref, _, _, _}
+    refute Process.alive?(handler)
+  end
+end
diff --git a/test/event/event_handler_macro_test.exs b/test/event/event_handler_macro_test.exs
index 7b68690..da4fab4 100644
--- a/test/event/event_handler_macro_test.exs
+++ b/test/event/event_handler_macro_test.exs
@@ -1,37 +1,22 @@
 defmodule Commanded.Event.EventHandlerMacroTest do
   use Commanded.StorageCase
 
-  defmodule AccountBalanceHandler do
-    use Commanded.Event.Handler, name: "account_balance_handler"
-
-    alias Commanded.ExampleDomain.BankAccount.Events.{
-      BankAccountOpened,
-      MoneyDeposited,
-    }
-
-    def handle(%BankAccountOpened{initial_balance: initial_balance}, _metadata) do
-      Agent.update(__MODULE__, fn _ -> initial_balance end)
-    end
-
-    def handle(%MoneyDeposited{balance: balance}, _metadata) do
-      Agent.update(__MODULE__, fn _ -> balance end)
-    end
-
-    def current_balance do
-      Agent.get(__MODULE__, fn balance -> balance end)
-    end
-  end
-
   defmodule IgnoredEvent do
     defstruct [name: nil]
   end
 
+  alias Commanded.ExampleDomain.AccountBalanceHandler
+  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
   alias Commanded.Helpers.EventFactory
   alias Commanded.Helpers.Wait
-  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
+
+  setup do
+    on_exit fn ->
+      Commanded.Helpers.Process.shutdown(AccountBalanceHandler)
+    end
+  end
 
   test "should handle published events" do
-    {:ok, _} = Agent.start_link(fn -> 0 end, name: AccountBalanceHandler)
     {:ok, handler} = AccountBalanceHandler.start_link()
 
     recorded_events =
diff --git a/test/event/handle_event_test.exs b/test/event/handle_event_test.exs
index b8eec6b..822109a 100644
--- a/test/event/handle_event_test.exs
+++ b/test/event/handle_event_test.exs
@@ -11,9 +11,15 @@ defmodule Commanded.Event.HandleEventTest do
   alias Commanded.ExampleDomain.AccountBalanceHandler
   alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
 
+  setup do
+    on_exit fn ->
+      Commanded.Helpers.Process.shutdown(AccountBalanceHandler)
+      Commanded.Helpers.Process.shutdown(AppendingEventHandler)
+    end
+  end
+
   test "should be notified of events" do
-    {:ok, _} = AccountBalanceHandler.start_link()
-    {:ok, handler} = Commanded.Event.Handler.start_link("account_balance", AccountBalanceHandler)
+    {:ok, handler} = AccountBalanceHandler.start_link()
 
     events = [
       %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
@@ -31,8 +37,7 @@ defmodule Commanded.Event.HandleEventTest do
   defmodule UninterestingEvent, do: defstruct [field: nil]
 
   test "should ignore uninterested events" do
-    {:ok, _} = AccountBalanceHandler.start_link
-		{:ok, handler} = Commanded.Event.Handler.start_link("account_balance", AccountBalanceHandler)
+    {:ok, handler} = AccountBalanceHandler.start_link
 
     # include uninterested events within those the handler is interested in
     events = [
@@ -51,9 +56,7 @@ defmodule Commanded.Event.HandleEventTest do
     end)
   end
 
-  test "should ignore events created before the event handler's subscription when starting from `current`" do
-    {:ok, _} = AppendingEventHandler.start_link()
-
+  test "should ignore events created before the event handler's subscription when starting from `:current`" do
     stream_uuid = UUID.uuid4
     initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
     new_events = [%MoneyDeposited{amount: 50, balance: 1_050}]
@@ -62,7 +65,7 @@ defmodule Commanded.Event.HandleEventTest do
 
     wait_for_event BankAccountOpened
 
-    {:ok, _handler} = Commanded.Event.Handler.start_link("test_event_handler", AppendingEventHandler, start_from: :current)
+    {:ok, _} = AppendingEventHandler.start_link(start_from: :current)
 
     {:ok, 2} = EventStore.append_to_stream(stream_uuid, 1, Commanded.Event.Mapper.map_to_event_data(new_events, UUID.uuid4))
 
@@ -70,20 +73,24 @@ defmodule Commanded.Event.HandleEventTest do
 
     Wait.until(fn ->
       assert AppendingEventHandler.received_events == new_events
-      assert pluck(AppendingEventHandler.received_metadata, :stream_version) == [2]
+
+      [ metadata ] = AppendingEventHandler.received_metadata()
+
+      assert Map.get(metadata, :event_number) == 2
+      assert Map.get(metadata, :stream_id) == stream_uuid
+      assert Map.get(metadata, :stream_version) == 2
+      assert %NaiveDateTime{} = Map.get(metadata, :created_at)
     end)
 	end
 
-  test "should receive events created before the event handler's subscription when starting from `origin`" do
-    {:ok, _} = AppendingEventHandler.start_link()
-
+  test "should receive events created before the event handler's subscription when starting from `:origin`" do
     stream_uuid = UUID.uuid4
     initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
     new_events = [%MoneyDeposited{amount: 50, balance: 1_050}]
 
     {:ok, 1} = EventStore.append_to_stream(stream_uuid, 0, Commanded.Event.Mapper.map_to_event_data(initial_events, UUID.uuid4))
 
-    {:ok, _handler} = Commanded.Event.Handler.start_link("test_event_handler", AppendingEventHandler, start_from: :origin)
+    {:ok, _} = AppendingEventHandler.start_link(start_from: :origin)
 
     {:ok, 2} = EventStore.append_to_stream(stream_uuid, 1, Commanded.Event.Mapper.map_to_event_data(new_events, UUID.uuid4))
 
@@ -91,13 +98,21 @@ defmodule Commanded.Event.HandleEventTest do
 
     Wait.until(fn ->
       assert AppendingEventHandler.received_events == initial_events ++ new_events
-      assert pluck(AppendingEventHandler.received_metadata, :stream_version) == [1, 2]
+
+      received_metadata = AppendingEventHandler.received_metadata()
+
+      assert pluck(received_metadata, :event_number) == [1, 2]
+      assert pluck(received_metadata, :stream_version) == [1, 2]
+
+      Enum.each(received_metadata, fn metadata ->
+        assert Map.get(metadata, :stream_id) == stream_uuid
+        assert %NaiveDateTime{} = Map.get(metadata, :created_at)
+      end)
     end)
 	end
 
 	test "should ignore already seen events" do
-    {:ok, _} = AppendingEventHandler.start_link
-    {:ok, handler} = Commanded.Event.Handler.start_link("test_event_handler", AppendingEventHandler)
+    {:ok, handler} = AppendingEventHandler.start_link()
 
     events = [
       %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
diff --git a/test/example_domain/bank_account/account_balance_handler.ex b/test/example_domain/bank_account/account_balance_handler.ex
index e589ed2..9b74631 100644
--- a/test/example_domain/bank_account/account_balance_handler.ex
+++ b/test/example_domain/bank_account/account_balance_handler.ex
@@ -1,10 +1,12 @@
 defmodule Commanded.ExampleDomain.AccountBalanceHandler do
-  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
+  use Commanded.Event.Handler, name: "AccountBalanceHandler"
 
-  @behaviour Commanded.Event.Handler
+  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
 
-  def start_link do
-    Agent.start_link(fn -> 0 end, name: __MODULE__)
+  def init do
+    with {:ok, _} <- Agent.start_link(fn -> 0 end, name: __MODULE__) do
+      :ok
+    end
   end
 
   def handle(%BankAccountOpened{initial_balance: initial_balance}, _metadata) do
@@ -15,9 +17,6 @@ defmodule Commanded.ExampleDomain.AccountBalanceHandler do
     Agent.update(__MODULE__, fn _ -> balance end)
   end
 
-  # ignore any other events
-  def handle(_event, _metadata), do: :ok
-  
   def current_balance do
     Agent.get(__MODULE__, fn balance -> balance end)
   end
diff --git a/test/helpers/appending_event_handler.ex b/test/helpers/appending_event_handler.ex
index cddd168..feca8ce 100644
--- a/test/helpers/appending_event_handler.ex
+++ b/test/helpers/appending_event_handler.ex
@@ -1,9 +1,11 @@
 defmodule Commanded.Event.AppendingEventHandler do
   @moduledoc false
-  @behaviour Commanded.Event.Handler
+  use Commanded.Event.Handler, name: "AppendingEventHandler"
 
-  def start_link do
-    Agent.start_link(fn -> %{events: [], metadata: []} end, name: __MODULE__)
+  def init do
+    with {:ok, _} <- Agent.start_link(fn -> %{events: [], metadata: []} end, name: __MODULE__) do
+      :ok
+    end
   end
 
   def handle(event, event_metadata) do
diff --git a/test/process_managers/process_manager_error_handling_test.exs b/test/process_managers/process_manager_error_handling_test.exs
new file mode 100644
index 0000000..62b36f6
--- /dev/null
+++ b/test/process_managers/process_manager_error_handling_test.exs
@@ -0,0 +1,226 @@
+defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
+  use Commanded.StorageCase
+
+  defmodule ExampleAggregate do
+    @moduledoc false
+    defstruct [:process_uuid]
+
+    defmodule Commands do
+      defmodule StartProcess, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
+      defmodule AttemptProcess, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
+      defmodule ContinueProcess, do: defstruct [:process_uuid, :reply_to]
+    end
+
+    defmodule Events do
+      defmodule ProcessStarted, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
+      defmodule ProcessContinued, do: defstruct [:process_uuid, :reply_to]
+    end
+
+    alias Commands.{AttemptProcess,ContinueProcess,StartProcess}
+    alias Events.{ProcessContinued,ProcessStarted}
+
+    def execute(
+      %ExampleAggregate{},
+      %StartProcess{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to})
+    do
+      %ProcessStarted{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to}
+    end
+
+    def execute(%ExampleAggregate{}, %AttemptProcess{}),
+      do: {:error, :failed}
+
+    def execute(%ExampleAggregate{}, %ContinueProcess{process_uuid: process_uuid, reply_to: reply_to}),
+      do: %ProcessContinued{process_uuid: process_uuid, reply_to: reply_to}
+
+    def apply(%ExampleAggregate{} = aggregate, %ProcessStarted{process_uuid: process_uuid}),
+      do: %ExampleAggregate{aggregate | process_uuid: process_uuid}
+
+    def apply(%ExampleAggregate{} = aggregate, %ProcessContinued{}), do: aggregate
+  end
+
+  alias ExampleAggregate.Commands.{AttemptProcess,ContinueProcess,StartProcess}
+  alias ExampleAggregate.Events.{ProcessContinued,ProcessStarted}
+
+  defmodule ExampleRouter do
+    @moduledoc false
+    use Commanded.Commands.Router
+
+    dispatch [StartProcess,AttemptProcess,ContinueProcess],
+      to: ExampleAggregate, identity: :process_uuid
+  end
+
+  defmodule ErrorHandlingProcessManager do
+    @moduledoc false
+    use Commanded.ProcessManagers.ProcessManager,
+      name: "ErrorHandlingProcessManager",
+      router: ExampleRouter
+
+    defstruct [:process_uuid]
+
+    def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
+    def interested?(%ProcessContinued{process_uuid: process_uuid}), do: {:continue, process_uuid}
+
+    def handle(
+      %ErrorHandlingProcessManager{},
+      %ProcessStarted{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to})
+    do
+      %AttemptProcess{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to}
+    end
+
+    def handle(%ErrorHandlingProcessManager{}, %ProcessContinued{reply_to: reply_to}) do
+      send(reply_to, :process_continued)
+      []
+    end
+
+    # stop after three attempts
+    def error({:error, :failed}, %AttemptProcess{strategy: :retry, reply_to: reply_to}, _pending_commands, %{attempts: attempts} = context)
+      when attempts >= 2
+    do
+      send(reply_to, {:error, :too_many_attempts, record_attempt(context)})
+
+      {:stop, :too_many_attempts}
+    end
+
+    # retry command with delay
+    def error({:error, :failed}, %AttemptProcess{strategy: :retry, delay: delay} = failed_command, _pending_commands, context)
+      when is_integer(delay)
+    do
+      context = record_attempt(context)
+      send_failure(failed_command, Map.put(context, :delay, delay))
+
+      {:retry, delay, context}
+    end
+
+    # retry command
+    def error({:error, :failed}, %AttemptProcess{strategy: :retry} = failed_command, _pending_commands, context) do
+      context = record_attempt(context)
+      send_failure(failed_command, context)
+
+      {:retry, context}
+    end
+
+    # skip failed command, continue pending
+    def error({:error, :failed}, %AttemptProcess{strategy: :skip, reply_to: reply_to}, _pending_commands, context) do
+      send(reply_to, {:error, :failed, record_attempt(context)})
+
+      {:skip, :continue_pending}
+    end
+
+    # continue with modified command
+    def error({:error, :failed}, %AttemptProcess{strategy: :continue, process_uuid: process_uuid, reply_to: reply_to}, pending_commands, context) do
+      context = record_attempt(context)
+      send(reply_to, {:error, :failed, context})
+
+      continue = %ContinueProcess{process_uuid: process_uuid, reply_to: reply_to}
+
+      {:continue, [continue | pending_commands], context}
+    end
+
+    defp record_attempt(context) do
+      Map.update(context, :attempts, 1, fn attempts -> attempts + 1 end)
+    end
+
+    defp send_failure(%AttemptProcess{reply_to: reply_to}, context) do
+      send(reply_to, {:error, :failed, context})
+    end
+  end
+
+  test "should retry the event until process manager requests stop" do
+    process_uuid = UUID.uuid4()
+    command = %StartProcess{process_uuid: process_uuid, strategy: :retry, reply_to: self()}
+
+    {:ok, process_router} = ErrorHandlingProcessManager.start_link()
+
+    Process.unlink(process_router)
+    ref = Process.monitor(process_router)
+
+    assert :ok = ExampleRouter.dispatch(command)
+
+    assert_receive {:error, :failed, %{attempts: 1}}
+    assert_receive {:error, :failed, %{attempts: 2}}
+    assert_receive {:error, :too_many_attempts, %{attempts: 3}}
+
+    # should shutdown process router
+    assert_receive {:DOWN, ^ref, _, _, _}
+  end
+
+  test "should retry event with specified delay between attempts" do
+    process_uuid = UUID.uuid4()
+    command = %StartProcess{process_uuid: process_uuid, strategy: :retry, delay: 10, reply_to: self()}
+
+    {:ok, process_router} = ErrorHandlingProcessManager.start_link()
+
+    Process.unlink(process_router)
+    ref = Process.monitor(process_router)
+
+    assert :ok = ExampleRouter.dispatch(command)
+
+    assert_receive {:error, :failed, %{attempts: 1, delay: 10}}
+    assert_receive {:error, :failed, %{attempts: 2, delay: 10}}
+    assert_receive {:error, :too_many_attempts, %{attempts: 3}}
+
+    # should shutdown process router
+    assert_receive {:DOWN, ^ref, _, _, _}
+  end
+
+  test "should skip the event when error reply is `{:skip, :continue_pending}`" do
+    process_uuid = UUID.uuid4()
+    command = %StartProcess{process_uuid: process_uuid, strategy: :skip, reply_to: self()}
+
+    {:ok, process_router} = ErrorHandlingProcessManager.start_link()
+
+    assert :ok = ExampleRouter.dispatch(command)
+
+    assert_receive {:error, :failed, %{attempts: 1}}
+    refute_receive {:error, :failed, %{attempts: 2}}
+
+    # should not shutdown process router
+    assert Process.alive?(process_router)
+  end
+
+  test "should continue with modified command" do
+    process_uuid = UUID.uuid4()
+    command = %StartProcess{process_uuid: process_uuid, strategy: :continue, reply_to: self()}
+
+    {:ok, process_router} = ErrorHandlingProcessManager.start_link()
+
+    assert :ok = ExampleRouter.dispatch(command)
+
+    assert_receive {:error, :failed, %{attempts: 1}}
+    assert_receive :process_continued
+
+    # should not shutdown process router
+    assert Process.alive?(process_router)
+  end
+
+  defmodule DefaultErrorHandlingProcessManager do
+    @moduledoc false
+    use Commanded.ProcessManagers.ProcessManager,
+      name: "DefaultErrorHandlingProcessManager",
+      router: ExampleRouter
+
+    defstruct [:process_uuid]
+
+    def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
+
+    def handle(%ErrorHandlingProcessManager{}, %ProcessStarted{process_uuid: process_uuid}) do
+      %AttemptProcess{process_uuid: process_uuid}
+    end
+  end
+
+  test "should stop process manager on error by default" do
+    process_uuid = UUID.uuid4()
+    command = %StartProcess{process_uuid: process_uuid}
+
+    {:ok, process_router} = ErrorHandlingProcessManager.start_link()
+
+    Process.unlink(process_router)
+    ref = Process.monitor(process_router)
+
+    assert :ok = ExampleRouter.dispatch(command)
+
+    # should shutdown process router
+    assert_receive {:DOWN, ^ref, _, _, _}
+    refute Process.alive?(process_router)
+  end
+end
diff --git a/test/process_managers/process_router_process_pending_events_test.exs b/test/process_managers/process_router_process_pending_events_test.exs
index 9894d2e..34dbaaf 100644
--- a/test/process_managers/process_router_process_pending_events_test.exs
+++ b/test/process_managers/process_router_process_pending_events_test.exs
@@ -20,6 +20,7 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
       defmodule Start, do: defstruct [:aggregate_uuid]
       defmodule Publish, do: defstruct [:aggregate_uuid, :interesting, :uninteresting]
       defmodule Stop, do: defstruct [:aggregate_uuid]
+      defmodule Error, do: defstruct [:aggregate_uuid]
     end
 
     defmodule Events do
@@ -27,6 +28,7 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
       defmodule Interested, do: defstruct [:aggregate_uuid, :index]
       defmodule Uninterested, do: defstruct [:aggregate_uuid, :index]
       defmodule Stopped, do: defstruct [:aggregate_uuid]
+      defmodule Errored, do: defstruct [:aggregate_uuid]
     end
 
     def start(%ExampleAggregate{}, aggregate_uuid) do
@@ -44,6 +46,10 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
       %Events.Stopped{aggregate_uuid: aggregate_uuid}
     end
 
+    def error(%ExampleAggregate{uuid: aggregate_uuid}) do
+      %Events.Errored{aggregate_uuid: aggregate_uuid}
+    end
+
     defp publish_interesting(_aggregate_uuid, 0, _index), do: []
     defp publish_interesting(aggregate_uuid, interesting, index) do
       [
@@ -62,12 +68,13 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
 
     def apply(%ExampleAggregate{} = state, %Events.Started{aggregate_uuid: aggregate_uuid}), do: %ExampleAggregate{state | uuid: aggregate_uuid, state: :started}
     def apply(%ExampleAggregate{items: items} = state, %Events.Interested{index: index}), do: %ExampleAggregate{state | items: items ++ [index]}
+    def apply(%ExampleAggregate{} = state, %Events.Errored{}), do: %ExampleAggregate{state | state: :errored}
     def apply(%ExampleAggregate{} = state, %Events.Uninterested{}), do: state
     def apply(%ExampleAggregate{} = state, %Events.Stopped{}), do: %ExampleAggregate{state | state: :stopped}
   end
 
-  alias ExampleAggregate.Commands.{Start,Publish,Stop}
-  alias ExampleAggregate.Events.{Started,Interested,Uninterested,Stopped}
+  alias ExampleAggregate.Commands.{Start,Publish,Stop,Error}
+  alias ExampleAggregate.Events.{Started,Interested,Uninterested,Stopped,Errored}
 
   defmodule ExampleCommandHandler do
     @behaviour Commanded.Commands.Handler
@@ -75,12 +82,16 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
     def handle(%ExampleAggregate{} = aggregate, %Start{aggregate_uuid: aggregate_uuid}), do: ExampleAggregate.start(aggregate, aggregate_uuid)
     def handle(%ExampleAggregate{} = aggregate, %Publish{interesting: interesting, uninteresting: uninteresting}), do: ExampleAggregate.publish(aggregate, interesting, uninteresting)
     def handle(%ExampleAggregate{} = aggregate, %Stop{}), do: ExampleAggregate.stop(aggregate)
+    def handle(%ExampleAggregate{} = aggregate, %Error{}), do: ExampleAggregate.error(aggregate)
   end
 
   defmodule Router do
     use Commanded.Commands.Router
 
-    dispatch [Start,Publish,Stop], to: ExampleCommandHandler, aggregate: ExampleAggregate, identity: :aggregate_uuid
+    dispatch [Start,Publish,Stop,Error],
+      to: ExampleCommandHandler,
+      aggregate: ExampleAggregate,
+      identity: :aggregate_uuid
   end
 
   defmodule ExampleProcessManager do
@@ -95,12 +106,15 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
 
     def interested?(%Started{aggregate_uuid: aggregate_uuid}), do: {:start, aggregate_uuid}
     def interested?(%Interested{aggregate_uuid: aggregate_uuid}), do: {:continue, aggregate_uuid}
+    def interested?(%Errored{aggregate_uuid: aggregate_uuid}), do: {:continue, aggregate_uuid}
     def interested?(%Stopped{aggregate_uuid: aggregate_uuid}), do: {:stop, aggregate_uuid}
 
     def handle(%ExampleProcessManager{}, %Interested{index: 10, aggregate_uuid: aggregate_uuid}) do
       %Stop{aggregate_uuid: aggregate_uuid}
     end
 
+    def handle(%ExampleProcessManager{}, %Errored{}), do: {:error, :failed}
+
     ## state mutators
 
     def apply(%ExampleProcessManager{} = process_manager, %Started{}) do
@@ -208,7 +222,7 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
 
     wait_for_event Interested, fn event -> event.index == 4 end
 
-    {:ok, process_router} = ProcessRouter.start_link("example_process_manager", ExampleProcessManager, Router, start_from: :current)
+    {:ok, process_router} = ExampleProcessManager.start_link(start_from: :current)
 
     :ok = Router.dispatch(%Start{aggregate_uuid: aggregate_uuid})
     :ok = Router.dispatch(%Publish{aggregate_uuid: aggregate_uuid, interesting: 6, uninteresting: 0})
@@ -220,4 +234,19 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
     %{items: items} = ProcessManagerInstance.process_state(process_instance)
     assert items == [1, 2, 3, 4, 5, 6]
   end
+
+  test "should stop process router when handling event errors" do
+    aggregate_uuid = UUID.uuid4
+
+    {:ok, process_router} = ExampleProcessManager.start_link()
+
+    Process.unlink(process_router)
+    ref = Process.monitor(process_router)
+
+    :ok = Router.dispatch(%Start{aggregate_uuid: aggregate_uuid})
+    :ok = Router.dispatch(%Error{aggregate_uuid: aggregate_uuid})
+
+    # should shutdown process
+    assert_receive {:DOWN, ^ref, _, _, _}
+  end
 end
diff --git a/test/process_managers/resume_process_manager_test.exs b/test/process_managers/resume_process_manager_test.exs
index 01ef13b..d8be41a 100644
--- a/test/process_managers/resume_process_manager_test.exs
+++ b/test/process_managers/resume_process_manager_test.exs
@@ -98,7 +98,6 @@ defmodule Commanded.ProcessManager.ResumeProcessManagerTest do
 
     {:ok, process_router} = ExampleProcessManager.start_link()
 
-    # transfer funds between account 1 and account 2
     :ok = ExampleRouter.dispatch(%StartProcess{process_uuid: process_uuid, status: "start"})
 
     assert_receive_event ProcessStarted, fn event ->