Repository: commanded/commanded (elixir)

Issue:
Process manager enforce `:start` and `:continue`
Pull request #181 adds validation to ensure that when a process manager returns `{:continue, process_uuid}` it will guarantee the instance exists, otherwise the event is ignored. The same validation should be applied to `{:start, process_uuid}` to ensure the process does not yet exist.

Currently both `:start` and `:continue` behave as though they are "start or continue" where a process manager instance is resumed, regardless of whether it had been started or not. The above validation will affect the current behaviour and is potentially a breaking change to existing process managers. 

To address this I propose either:

### Start or continue

Adding a `{:start_or_continue, process_uuid}` option which replicates the current behaviour (a process is started or resumed regardless of previous state).

This will be a breaking change and existing process managers will need to be updated.

### "Strict mode"

Extend `:start` and continue to allow an optional strict mode enforcement: 

- `{:start, process_uuid, strict: true}` - validate process does not already exist.
- `{:continue, process_uuid, strict: true}` - validate process already exists.
- `{:start, process_uuid}` and `{:continue, process_uuid}` - the default behaviour without specifying strict mode will be as currently implemented.

Won't affect existing process managers, but allows consumers to opt-in to strict checking.

### Append `!`

Append `!` to start/continue to indicate strict mode following existing Elixir function naming style (per @fdbeirao's suggestion below):

- `{:start!, process_uuid}` - validate process does not already exist.
- `{:continue!, process_uuid}` - validate process already exists.
- `{:start, process_uuid}` and `{:continue, process_uuid}` - the default behaviour without specifying strict mode will be as currently implemented.

Won't affect existing process managers, but allows consumers to opt-in to strict checking.

--- 

Any thoughts or preferences on the above approaches, or suggestions for another options. This requires addressing before the next release can be published.

Pull request description:
Strict process manager routing
Add `:start!` and `:continue!` for strict process routing replies to process manager `interested?/1` callback.

Strict routing enforces the following validation checks:

- `{:start!, process_uuid}` - validate process does not already exist.
- `{:continue!, process_uuid}` - validate process already exists.

If the check fails an error will be passed to the `error/3` callback function:

- `{:error, {:start!, :process_already_started}}`
- `{:error, {:continue!, :process_not_started}}`

The `error/3` function can choose to `:stop` the process or `:skip` the  problematic event.

Fixes #212.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7e38e9c..794e12f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,7 +4,6 @@
 
 ### Enhancements
 
-- Do not start new process manager instance on `:continue` ([#181](https://github.com/commanded/commanded/pull/181)).
 - Rename `uuid` dependency to `elixir_uuid` ([#178](https://github.com/commanded/commanded/pull/178)).
 - Allow aggregate identity to be of any type that implements the `String.Chars` protocol ([#166](https://github.com/commanded/commanded/pull/166)).
 - Process manager and event handler error & exception handling ([#192](https://github.com/commanded/commanded/pull/192)).
@@ -14,8 +13,8 @@
 - Export `Commanded.Commands.Router` macros in `.formatter.exs` file ([#204](https://github.com/commanded/commanded/pull/204)).
 - Generate specs and docs for Router dispatch functions only once ([#206](https://github.com/commanded/commanded/pull/206)).
 - Allow two-arity predicate function in `wait_for_event` receiving both event data and recorded event struct ([#213](https://github.com/commanded/commanded/pull/213)).
-- Make poison an optional dependency ([#215](https://github.com/commanded/commanded/pull/215)).
 - Allow `:infinity` timeout on command dispatch ([#227](https://github.com/commanded/commanded/pull/227))
+- Strict process manager routing ([#243](https://github.com/commanded/commanded/pull/243)).
 
 ### Bug fixes
 
diff --git a/README.md b/README.md
index 662c36b..e1a1fb3 100644
--- a/README.md
+++ b/README.md
@@ -23,7 +23,6 @@ Please refer to the [CHANGELOG](CHANGELOG.md) for features, bug fixes, and any u
 
 - [Changelog](CHANGELOG.md)
 - [Wiki](https://github.com/commanded/commanded/wiki)
-- [What is CQRS/ES?](https://kalele.io/blog-posts/really-simple-cqrs/)
 - [Frequently asked questions](https://github.com/commanded/commanded/wiki/FAQ)
 - [Getting help](https://github.com/commanded/commanded/wiki/Getting-help)
 - [Latest published Hex package](https://hex.pm/packages/commanded) & [documentation](https://hexdocs.pm/commanded/)
diff --git a/guides/Process Managers.md b/guides/Process Managers.md
index 4dd9123..27db614 100644
--- a/guides/Process Managers.md
+++ b/guides/Process Managers.md
@@ -2,18 +2,35 @@
 
 A process manager is responsible for coordinating one or more aggregates. It handles events and dispatches commands in response. You can think of a process manager as the opposite of an aggregate: aggregates handle commands and create events; process managers handle events and create commands. Process managers have state that can be used to track which aggregates are being orchestrated.
 
-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`.
+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/3`.
 
 ## `interested?/1`
 
 The `interested?/1` function is used to indicate which events the process manager handles. The response is used to route the event to an existing instance or start a new process instance:
 
 - `{:start, process_uuid}` - create a new instance of the process manager.
+- `{:start!, process_uuid}` - create a new instance of the process manager (strict).
 - `{:continue, process_uuid}` - continue execution of an existing process manager.
-- `{:stop, process_uuid}` - stop an existing process manager, shutdown its process, and delete its persisted state.
+- `{:continue!, process_uuid}` - continue execution of an existing process manager (strict).
+- `{:stop, process_uuid}` - stop an existing process manager, shutdown its
+  process, and delete its persisted state.
 - `false` - ignore the event.
 
-You can return a list of process identifiers when a single domain event must be handled by multiple process instances.
+You can return a list of process identifiers when a single domain event is to be handled by multiple process instances.
+
+### Strict process routing
+
+Using strict routing, with `:start!` or `:continue`, enforces the following validation checks:
+
+- `{:start!, process_uuid}` - validate process does not already exist.
+- `{:continue!, process_uuid}` - validate process already exists.
+
+If the check fails an error will be passed to the `error/3` callback function:
+
+- `{:error, {:start!, :process_already_started}}`
+- `{:error, {:continue!, :process_not_started}}`
+
+The `error/3` function can choose to `:stop` the process or `:skip` the problematic event.
 
 ## `handle/2`
 
@@ -33,7 +50,7 @@ You can define an `c:error/3` callback function to handle any errors or exceptio
 
 Use pattern matching on the error and/or failed event/command to explicitly handle certain errors, events, or commands. You can choose to retry, skip, ignore, or stop the process manager after a command dispatch error.
 
-The default behaviour, if you don't provide an `c:error/3` callback, is to stop the process manager using the exact error reason returned from the event handler function or command dispatch. You should supervise your process managers to ensure they are restarted on error.
+The default behaviour, if you don't provide an `c:error/3` callback, is to stop the process manager using the exact error reason returned from the event handler function or command dispatch.
 
 The `error/3` callback function must return one of the following responses depending upon the severity of error and how you choose to handle it:
 
@@ -72,6 +89,29 @@ defmodule Bank.Payments.Supervisor do
 end
 ```
 
+### Supervision caveats
+
+The default error handling strategy is to stop the process manager. When supervised, the process will be restarted and will attempt to handle the same event again, which will likely result in the same error. This could lead to too many restarts of the supervisor, which may eventually cause the application to stop, depending upon your supervision tree.
+
+To prevent this you can choose to define the default error handling strategy as `:skip` to skip over any problematic events.
+
+```elixir
+defmodule ExampleProcessManager do
+  use Commanded.ProcessManagers.ProcessManager, name: __MODULE__, router: Router
+
+  require Logger
+
+  # By default skip any problematic events
+  def error(error, _command_or_event, _failure_context) do
+    Logger.error(fn -> "#{__MODULE__} encountered an error: " <> inspect(error) end)
+
+    :skip
+  end
+end
+```
+
+Alternatively you can define the restart strategy of your process manager as `:temporary` to prevent it from being restarted on termination. This approach will require manual intervention to fix the stopped process manager, but ensures that it won't miss any events nor crash the application.
+
 ### Error handling example
 
 Define an `error/3` callback function to determine how to handle errors during event handling and command dispatch.
@@ -83,14 +123,14 @@ defmodule ExampleProcessManager do
     router: ExampleRouter
 
   # Stop process manager after three failures
-  def error({:error, _failure}, _failed_command, %{context: %{failures: failures}})
+  def error({:error, _failure}, _failed_message, %{context: %{failures: failures}})
     when failures >= 2
   do
     {:stop, :too_many_failures}
   end
 
   # Retry command, record failure count in context map
-  def error({:error, _failure}, _failed_command, %{context: context}) do
+  def error({:error, _failure}, _failed_message, %{context: context}) do
     context = Map.update(context, :failures, 1, fn failures -> failures + 1 end)
     {:retry, context}
   end
@@ -116,22 +156,32 @@ defmodule TransferMoneyProcessManager do
     :status
   ]
 
+  # Process routing
+
   def interested?(%MoneyTransferRequested{transfer_uuid: transfer_uuid}), do: {:start, transfer_uuid}
   def interested?(%MoneyWithdrawn{transfer_uuid: transfer_uuid}), do: {:continue, transfer_uuid}
   def interested?(%MoneyDeposited{transfer_uuid: transfer_uuid}), do: {:stop, transfer_uuid}
   def interested?(_event), do: false
 
-  def handle(%TransferMoneyProcessManager{}, %MoneyTransferRequested{transfer_uuid: transfer_uuid, debit_account: debit_account, amount: amount}) do
+  # Command dispatch
+
+  def handle(%TransferMoneyProcessManager{}, %MoneyTransferRequested{} = event) do
+    %MoneyTransferRequested{transfer_uuid: transfer_uuid, debit_account: debit_account, amount: amount} = event
+
     %WithdrawMoney{account_number: debit_account, transfer_uuid: transfer_uuid, amount: amount}
   end
 
-  def handle(%TransferMoneyProcessManager{transfer_uuid: transfer_uuid, credit_account: credit_account, amount: amount}, %MoneyWithdrawn{}) do
+  def handle(%TransferMoneyProcessManager{} = pm, %MoneyWithdrawn{}) do
+    %TransferMoneyProcessManager{transfer_uuid: transfer_uuid, credit_account: credit_account, amount: amount} = pm
+
     %DepositMoney{account_number: credit_account, transfer_uuid: transfer_uuid, amount: amount}
   end
 
   # State mutators
 
-  def apply(%TransferMoneyProcessManager{} = transfer, %MoneyTransferRequested{transfer_uuid: transfer_uuid, debit_account: debit_account, credit_account: credit_account, amount: amount}) do
+  def apply(%TransferMoneyProcessManager{} = transfer, %MoneyTransferRequested{} = event) do
+    %MoneyTransferRequested{transfer_uuid: transfer_uuid, debit_account: debit_account, credit_account: credit_account, amount: amount} = event
+
     %TransferMoneyProcessManager{transfer |
       transfer_uuid: transfer_uuid,
       debit_account: debit_account,
@@ -152,7 +202,7 @@ end
 The name given to the process manager *must* be unique. This is used when subscribing to events from the event store to track the last seen event and ensure they are only received once.
 
 ```elixir
-{:ok, _} = TransferMoneyProcessManager.start_link(start_from: :current)
+{:ok, _pid} = TransferMoneyProcessManager.start_link(start_from: :current)
 ```
 
 You can choose to start the process router's event store subscription from the `:origin`, `:current` position or an exact event number using the `start_from` option. The default is to use the origin so it will receive all events. You typically use `:current` when adding a new process manager to an already deployed system containing historical events.
@@ -171,14 +221,13 @@ defmodule TransferMoneyProcessManager do
     name: "TransferMoneyProcessManager",
     router: BankRouter,
     event_timeout: :timer.minutes(10)
-
 end
 ```
 
 Or may be configured when starting a process manager:
 
 ```elixir
-{:ok, _} = TransferMoneyProcessManager.start_link(event_timeout: :timer.hours(1))
+{:ok, _pid} = TransferMoneyProcessManager.start_link(event_timeout: :timer.hours(1))
 ```
 
 After the timeout has elapsed, indicating the process manager has not processed an event within the configured period, the process manager is stopped. The process manager will be restarted if supervised and will retry the event, this should help resolve transient problems.
diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex
index 26525cf..84f54a9 100644
--- a/lib/commanded/commands/router.ex
+++ b/lib/commanded/commands/router.ex
@@ -139,6 +139,10 @@ defmodule Commanded.Commands.Router do
       :ok = BankRouter.dispatch(command, metadata: %{"ip_address" => "127.0.0.1"})
 
   """
+
+  @callback dispatch(struct, keyword()) ::
+              :ok | {:ok, non_neg_integer()} | {:ok, struct} | {:error, term}
+
   defmacro __using__(_opts) do
     quote do
       require Logger
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 39fadf4..d043b62 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -535,7 +535,7 @@ defmodule Commanded.Event.Handler do
 
       invalid ->
         Logger.warn(fn ->
-          describe(state) <> " returned an invalid error reponse: #{inspect(invalid)}"
+          describe(state) <> " returned an invalid error response: #{inspect(invalid)}"
         end)
 
         # Stop event handler with original error
diff --git a/lib/commanded/process_managers/failure_context.ex b/lib/commanded/process_managers/failure_context.ex
index 7caf03d..349b093 100644
--- a/lib/commanded/process_managers/failure_context.ex
+++ b/lib/commanded/process_managers/failure_context.ex
@@ -13,16 +13,16 @@ defmodule Commanded.ProcessManagers.FailureContext do
   """
 
   @type t :: %__MODULE__{
-    pending_commands: [struct()],
-    process_manager_state: struct(),
-    last_event: struct(),
-    context: map()
-  }
+          pending_commands: [struct()],
+          process_manager_state: struct(),
+          last_event: struct(),
+          context: map()
+        }
 
   defstruct [
-    pending_commands: nil,
-    process_manager_state: nil,
-    last_event: nil,
-    context: nil
+    :process_manager_state,
+    :last_event,
+    context: %{},
+    pending_commands: []
   ]
 end
diff --git a/lib/commanded/process_managers/process_manager.ex b/lib/commanded/process_managers/process_manager.ex
index a44ac2f..906ec88 100644
--- a/lib/commanded/process_managers/process_manager.ex
+++ b/lib/commanded/process_managers/process_manager.ex
@@ -132,17 +132,38 @@ defmodule Commanded.ProcessManagers.ProcessManager do
   instance or start a new process instance:
 
   - `{:start, process_uuid}` - create a new instance of the process manager.
+  - `{:start!, process_uuid}` - create a new instance of the process manager (strict).
   - `{:continue, process_uuid}` - continue execution of an existing process manager.
+  - `{:continue!, process_uuid}` - continue execution of an existing process manager (strict).
   - `{:stop, process_uuid}` - stop an existing process manager, shutdown its
     process, and delete its persisted state.
   - `false` - ignore the event.
 
-  You can return a list of process identifiers when a single domain event must
+  You can return a list of process identifiers when a single domain event is to
   be handled by multiple process instances.
+
+  ## Strict process routing
+
+  Using strict routing, with `:start!` or `:continue`, enforces the following
+  validation checks:
+
+  - `{:start!, process_uuid}` - validate process does not already exist.
+  - `{:continue!, process_uuid}` - validate process already exists.
+
+  If the check fails an error will be passed to the `error/3` callback function:
+
+  - `{:error, {:start!, :process_already_started}}`
+  - `{:error, {:continue!, :process_not_started}}`
+
+  The `error/3` function can choose to `:stop` the process or `:skip` the
+  problematic event.
+
   """
   @callback interested?(domain_event) ::
               {:start, process_uuid}
+              | {:start!, process_uuid}
               | {:continue, process_uuid}
+              | {:continue!, process_uuid}
               | {:stop, process_uuid}
               | false
 
diff --git a/lib/commanded/process_managers/process_manager_instance.ex b/lib/commanded/process_managers/process_manager_instance.ex
index 35a47e6..088786f 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -78,7 +78,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   def handle_call(:stop, _from, %ProcessManagerInstance{} = state) do
     :ok = delete_state(state)
 
-    # stop the process with a normal reason
+    # Stop the process with a normal reason
     {:stop, :normal, :ok, state}
   end
 
@@ -126,11 +126,14 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
     end
   end
 
-  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
+  defp event_already_seen?(%RecordedEvent{}, %ProcessManagerInstance{last_seen_event: nil}),
+    do: false
+
+  defp event_already_seen?(%RecordedEvent{} = event, %ProcessManagerInstance{} = state) do
+    %RecordedEvent{event_number: event_number} = event
+    %ProcessManagerInstance{last_seen_event: last_seen_event} = state
+
+    event_number <= last_seen_event
   end
 
   # Already seen event, so just ack
@@ -196,8 +199,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
     try do
       process_manager_module.handle(process_state, data)
     rescue
-      e ->
-        {:error, e}
+      e -> {:error, e}
     end
   end
 
@@ -245,7 +247,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
 
       invalid ->
         Logger.warn(fn ->
-          describe(state) <> " returned an invalid error reponse: #{inspect(invalid)}"
+          describe(state) <> " returned an invalid error response: #{inspect(invalid)}"
         end)
 
         # Stop process manager with original error
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 71969ed..f394d67 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -8,7 +8,10 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
   alias Commanded.EventStore
   alias Commanded.EventStore.RecordedEvent
-  alias Commanded.ProcessManagers.{ProcessManagerInstance, ProcessRouter, Supervisor}
+  alias Commanded.ProcessManagers.ProcessManagerInstance
+  alias Commanded.ProcessManagers.ProcessRouter
+  alias Commanded.ProcessManagers.Supervisor
+  alias Commanded.ProcessManagers.FailureContext
   alias Commanded.Subscriptions
 
   defmodule State do
@@ -59,21 +62,17 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     GenServer.cast(process_router, {:ack_event, event, instance})
   end
 
-  @doc """
-  Fetch the pid of an individual process manager instance identified by the
-  given `process_uuid`
-  """
+  @doc false
   def process_instance(process_router, process_uuid) do
     GenServer.call(process_router, {:process_instance, process_uuid})
   end
 
-  @doc """
-  Fetch the `process_uuid` and pid of all process manager instances
-  """
+  @doc false
   def process_instances(process_router) do
     GenServer.call(process_router, :process_instances)
   end
 
+  @doc false
   def handle_call(:process_instances, _from, %State{} = state) do
     %State{process_managers: process_managers} = state
 
@@ -82,6 +81,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:reply, reply, state}
   end
 
+  @doc false
   def handle_call({:process_instance, process_uuid}, _from, %State{} = state) do
     %State{process_managers: process_managers} = state
 
@@ -94,6 +94,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:reply, reply, state}
   end
 
+  @doc false
   def handle_cast({:ack_event, event, instance}, %State{} = state) do
     %State{pending_acks: pending_acks} = state
     %RecordedEvent{event_number: event_number} = event
@@ -117,16 +118,16 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, state}
   end
 
-  @doc """
-  Subscribe the process router to all events
-  """
+  @doc false
   def handle_cast(:subscribe_to_events, %State{} = state) do
     {:noreply, subscribe_to_all_streams(state)}
   end
 
+  @doc false
   def handle_cast(:process_pending_events, %State{pending_events: []} = state),
     do: {:noreply, state}
 
+  @doc false
   def handle_cast(:process_pending_events, %State{} = state) do
     %State{pending_events: [event | pending_events]} = state
 
@@ -141,9 +142,10 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
         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}}
+    case handle_event(event, state) do
+      %State{} = state -> {:noreply, %State{state | pending_events: pending_events}}
+      reply -> reply
+    end
   end
 
   @doc false
@@ -165,6 +167,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, %State{state | supervisor: supervisor}}
   end
 
+  @doc false
   def handle_info({:events, events}, %State{pending_events: pending_events} = state) do
     Logger.debug(fn -> describe(state) <> " received #{length(events)} event(s)" end)
 
@@ -190,6 +193,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, state}
   end
 
+  @doc false
   # Shutdown process manager when processing an event has taken too long.
   def handle_info({:event_timeout, event_number}, %State{} = state) do
     %State{pending_acks: pending_acks, event_timeout: event_timeout} = state
@@ -210,6 +214,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     end
   end
 
+  @doc false
   # Remove a process manager instance that has stopped with a normal exit reason.
   def handle_info({:DOWN, _ref, :process, pid, :normal}, %State{} = state) do
     %State{process_managers: process_managers} = state
@@ -217,6 +222,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, %State{state | process_managers: remove_process_manager(process_managers, pid)}}
   end
 
+  @doc false
   # 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)
@@ -242,87 +248,135 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   end
 
   defp handle_event(%RecordedEvent{} = event, %State{} = state) do
-    %RecordedEvent{
-      event_number: event_number,
-      data: data,
-      stream_id: stream_id,
-      stream_version: stream_version
-    } = event
-
+    %RecordedEvent{data: data} = event
     %State{process_manager_module: process_manager_module} = state
 
-    case process_manager_module.interested?(data) do
-      {:start, process_uuid} ->
-        Logger.debug(fn ->
-          describe(state) <>
-            " is interested in event: #{inspect(event_number)} (#{inspect(stream_id)}@#{
-              inspect(stream_version)
-            })"
-        end)
+    try do
+      case process_manager_module.interested?(data) do
+        {:start, process_uuid} ->
+          Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
 
-        process_uuid
-        |> List.wrap()
-        |> Enum.reduce(state, fn process_uuid, state ->
-          {process_instance, state} = start_process_manager(process_uuid, state)
+          process_uuid
+          |> List.wrap()
+          |> Enum.reduce(state, fn process_uuid, state ->
+            {process_instance, state} = start_or_continue_process_manager(process_uuid, state)
+
+            delegate_event(process_instance, event, state)
+          end)
+
+        {:start!, process_uuid} ->
+          Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
+
+          {state, process_instances} =
+            process_uuid
+            |> List.wrap()
+            |> Enum.reduce({state, []}, fn process_uuid, {state, process_instances} ->
+              {process_instance, state} = start_or_continue_process_manager(process_uuid, state)
+
+              if ProcessManagerInstance.new?(process_instance) do
+                {state, [process_instance | process_instances]}
+              else
+                throw(
+                  handle_routing_error(
+                    {:error, {:start!, :process_already_started}},
+                    event,
+                    state
+                  )
+                )
+              end
+            end)
+
+          process_instances
+          |> Enum.reverse()
+          |> Enum.reduce(state, &delegate_event(&1, event, &2))
+
+        {:continue, process_uuid} ->
+          Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
 
-          delegate_event(process_instance, event, state)
-        end)
+          process_uuid
+          |> List.wrap()
+          |> Enum.reduce(state, fn process_uuid, state ->
+            {process_instance, state} = start_or_continue_process_manager(process_uuid, state)
+
+            delegate_event(process_instance, event, state)
+          end)
+
+        {:continue!, process_uuid} ->
+          Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
+
+          {state, process_instances} =
+            process_uuid
+            |> List.wrap()
+            |> Enum.reduce({state, []}, fn process_uuid, {state, process_instances} ->
+              {process_instance, state} = start_or_continue_process_manager(process_uuid, state)
+
+              if ProcessManagerInstance.new?(process_instance) do
+                throw(
+                  handle_routing_error(
+                    {:error, {:continue!, :process_not_started}},
+                    event,
+                    state
+                  )
+                )
+              else
+                {state, [process_instance | process_instances]}
+              end
+            end)
+
+          process_instances
+          |> Enum.reverse()
+          |> Enum.reduce(state, &delegate_event(&1, event, &2))
+
+        {:stop, process_uuid} ->
+          Logger.debug(fn ->
+            describe(state) <> " has been stopped by event " <> describe(event)
+          end)
+
+          state =
+            process_uuid
+            |> List.wrap()
+            |> Enum.reduce(state, &stop_process_manager/2)
+
+          ack_and_continue(event, state)
+
+        false ->
+          Logger.debug(fn ->
+            describe(state) <> " is not interested in event " <> describe(event)
+          end)
+
+          ack_and_continue(event, state)
+      end
+    rescue
+      e -> handle_routing_error({:error, e}, event, state)
+    catch
+      reply -> reply
+    end
+  end
 
-      {:continue, process_uuid} ->
-        Logger.debug(fn ->
-          describe(state) <>
-            " is interested in event: #{inspect(event_number)} (#{inspect(stream_id)}@#{
-              inspect(stream_version)
-            })"
-        end)
+  defp handle_routing_error(error, %RecordedEvent{} = failed_event, %State{} = state) do
+    %RecordedEvent{data: data} = failed_event
+    %State{process_manager_module: process_manager_module} = state
 
-        process_uuid
-        |> List.wrap()
-        |> Enum.reduce(state, fn process_uuid, state ->
-          {process_instance, state} = continue_process_manager(process_uuid, state)
-
-          case ProcessManagerInstance.new?(process_instance) do
-            false ->
-              delegate_event(process_instance, event, state)
-
-            true ->
-              Logger.debug(fn ->
-                describe(state) <>
-                  " is not interested in event: #{inspect(event_number)} (#{inspect(stream_id)}@#{
-                    inspect(stream_version)
-                  })"
-              end)
-
-              stop_process_manager(process_uuid, state)
-
-              ack_and_continue(event, state)
-          end
-        end)
+    failure_context = %FailureContext{last_event: failed_event}
 
-      {:stop, process_uuid} ->
-        Logger.debug(fn ->
-          describe(state) <>
-            " has been stopped by event: #{inspect(event_number)} (#{inspect(stream_id)}@#{
-              inspect(stream_version)
-            })"
-        end)
+    case process_manager_module.error(error, data, failure_context) do
+      :skip ->
+        # Skip the problematic event by confirming receipt
+        Logger.info(fn -> describe(state) <> " is skipping event" end)
 
-        state =
-          process_uuid
-          |> List.wrap()
-          |> Enum.reduce(state, &stop_process_manager/2)
+        ack_and_continue(failed_event, state)
 
-        ack_and_continue(event, state)
+      {:stop, reason} ->
+        Logger.warn(fn -> describe(state) <> " has requested to stop: #{inspect(error)}" end)
 
-      false ->
-        Logger.debug(fn ->
-          describe(state) <>
-            " is not interested in event: #{inspect(event_number)} (#{inspect(stream_id)}@#{
-              inspect(stream_version)
-            })"
+        {:stop, reason, state}
+
+      invalid ->
+        Logger.warn(fn ->
+          describe(state) <> " returned an invalid error response: #{inspect(invalid)}"
         end)
 
-        ack_and_continue(event, state)
+        {:stop, error, state}
     end
   end
 
@@ -344,6 +398,18 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     %State{state | last_seen_event: event_number}
   end
 
+  defp start_or_continue_process_manager(process_uuid, %State{} = state) do
+    %State{process_managers: process_managers} = state
+
+    case Map.get(process_managers, process_uuid) do
+      process_manager when is_pid(process_manager) ->
+        {process_manager, state}
+
+      nil ->
+        start_process_manager(process_uuid, state)
+    end
+  end
+
   defp start_process_manager(process_uuid, %State{} = state) do
     %State{
       process_managers: process_managers,
@@ -370,18 +436,6 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {process_manager, state}
   end
 
-  defp continue_process_manager(process_uuid, %State{} = state) do
-    %State{process_managers: process_managers} = state
-
-    case Map.get(process_managers, process_uuid) do
-      nil ->
-        start_process_manager(process_uuid, state)
-
-      process_manager ->
-        {process_manager, state}
-    end
-  end
-
   defp stop_process_manager(process_uuid, %State{} = state) do
     %State{process_managers: process_managers} = state
 
@@ -458,4 +512,14 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
   defp describe(%State{process_manager_module: process_manager_module}),
     do: inspect(process_manager_module)
+
+  defp describe(%RecordedEvent{} = event) do
+    %RecordedEvent{
+      event_number: event_number,
+      stream_id: stream_id,
+      stream_version: stream_version
+    } = event
+
+    "#{inspect(event_number)} (#{inspect(stream_id)}@#{inspect(stream_version)})"
+  end
 end


Test patch:
diff --git a/test/example_domain/money_transfer/transfer_money_process_manager.ex b/test/example_domain/money_transfer/transfer_money_process_manager.ex
index eb1ecf6..26ebe19 100644
--- a/test/example_domain/money_transfer/transfer_money_process_manager.ex
+++ b/test/example_domain/money_transfer/transfer_money_process_manager.ex
@@ -1,56 +1,59 @@
 defmodule Commanded.ExampleDomain.TransferMoneyProcessManager do
   @moduledoc false
+
   use Commanded.ProcessManagers.ProcessManager,
-    name: "transfer_money_process_manager",
+    name: __MODULE__,
     router: Commanded.ExampleDomain.BankRouter
 
   @derive Jason.Encoder
-  defstruct transfer_uuid: nil,
-            debit_account: nil,
-            credit_account: nil,
-            amount: nil,
-            status: nil
+  defstruct [:transfer_uuid, :debit_account, :credit_account, :amount, :status]
 
   alias Commanded.ExampleDomain.TransferMoneyProcessManager
-  alias Commanded.ExampleDomain.MoneyTransfer.Events.{MoneyTransferRequested}
-  alias Commanded.ExampleDomain.BankAccount.Events.{MoneyDeposited, MoneyWithdrawn}
-  alias Commanded.ExampleDomain.BankAccount.Commands.{DepositMoney, WithdrawMoney}
+  alias Commanded.ExampleDomain.MoneyTransfer.Events.MoneyTransferRequested
+  alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited
+  alias Commanded.ExampleDomain.BankAccount.Events.MoneyWithdrawn
+  alias Commanded.ExampleDomain.BankAccount.Commands.DepositMoney
+  alias Commanded.ExampleDomain.BankAccount.Commands.WithdrawMoney
 
   def interested?(%MoneyTransferRequested{transfer_uuid: transfer_uuid}),
     do: {:start, transfer_uuid}
 
-  def interested?(%MoneyWithdrawn{transfer_uuid: transfer_uuid}), do: {:continue, transfer_uuid}
-  def interested?(%MoneyDeposited{transfer_uuid: transfer_uuid}), do: {:continue, transfer_uuid}
+  def interested?(%MoneyWithdrawn{transfer_uuid: transfer_uuid}),
+    do: {:continue, transfer_uuid}
+
+  def interested?(%MoneyDeposited{transfer_uuid: transfer_uuid}),
+    do: {:continue, transfer_uuid}
+
+  def handle(%TransferMoneyProcessManager{}, %MoneyTransferRequested{} = event) do
+    %MoneyTransferRequested{
+      transfer_uuid: transfer_uuid,
+      debit_account: debit_account,
+      amount: amount
+    } = event
 
-  def handle(%TransferMoneyProcessManager{}, %MoneyTransferRequested{
-        transfer_uuid: transfer_uuid,
-        debit_account: debit_account,
-        amount: amount
-      }) do
     %WithdrawMoney{account_number: debit_account, transfer_uuid: transfer_uuid, amount: amount}
   end
 
-  def handle(
-        %TransferMoneyProcessManager{
-          transfer_uuid: transfer_uuid,
-          credit_account: credit_account,
-          amount: amount
-        },
-        %MoneyWithdrawn{}
-      ) do
+  def handle(%TransferMoneyProcessManager{} = pm, %MoneyWithdrawn{}) do
+    %TransferMoneyProcessManager{
+      transfer_uuid: transfer_uuid,
+      credit_account: credit_account,
+      amount: amount
+    } = pm
+
     %DepositMoney{account_number: credit_account, transfer_uuid: transfer_uuid, amount: amount}
   end
 
-  def handle(%TransferMoneyProcessManager{}, %MoneyDeposited{}), do: []
+  ## State mutators
 
-  ## state mutators
+  def apply(%TransferMoneyProcessManager{} = transfer, %MoneyTransferRequested{} = event) do
+    %MoneyTransferRequested{
+      transfer_uuid: transfer_uuid,
+      debit_account: debit_account,
+      credit_account: credit_account,
+      amount: amount
+    } = event
 
-  def apply(%TransferMoneyProcessManager{} = transfer, %MoneyTransferRequested{
-        transfer_uuid: transfer_uuid,
-        debit_account: debit_account,
-        credit_account: credit_account,
-        amount: amount
-      }) do
     %TransferMoneyProcessManager{
       transfer
       | transfer_uuid: transfer_uuid,
diff --git a/test/helpers/event_factory.ex b/test/helpers/event_factory.ex
index 0cdf0c7..50b2f51 100644
--- a/test/helpers/event_factory.ex
+++ b/test/helpers/event_factory.ex
@@ -2,7 +2,7 @@ defmodule Commanded.Helpers.EventFactory do
   @moduledoc false
   alias Commanded.EventStore.RecordedEvent
 
-  def map_to_recorded_events(events) do
+  def map_to_recorded_events(events, initial_event_number \\ 1) do
     stream_id = UUID.uuid4()
     causation_id = UUID.uuid4()
     correlation_id = UUID.uuid4()
@@ -10,7 +10,7 @@ defmodule Commanded.Helpers.EventFactory do
 
     events
     |> Commanded.Event.Mapper.map_to_event_data(fields)
-    |> Enum.with_index(1)
+    |> Enum.with_index(initial_event_number)
     |> Enum.map(fn {event, index} ->
       %RecordedEvent{
         event_id: UUID.uuid4(),
diff --git a/test/process_managers/process_manager_instance_test.exs b/test/process_managers/process_manager_instance_test.exs
index abc5a9a..550769d 100644
--- a/test/process_managers/process_manager_instance_test.exs
+++ b/test/process_managers/process_manager_instance_test.exs
@@ -42,7 +42,8 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstanceTest do
   end
 
   test "should provide `__name__/0` function" do
-    assert TransferMoneyProcessManager.__name__() == "transfer_money_process_manager"
+    assert TransferMoneyProcessManager.__name__() ==
+             "Commanded.ExampleDomain.TransferMoneyProcessManager"
   end
 
   test "should ensure a process manager name is provided" do
diff --git a/test/process_managers/process_manager_integration_test.exs b/test/process_managers/process_manager_integration_test.exs
new file mode 100644
index 0000000..6fbde80
--- /dev/null
+++ b/test/process_managers/process_manager_integration_test.exs
@@ -0,0 +1,63 @@
+defmodule Commanded.ProcessManagers.ProcessManagerIntegrationTest do
+  use Commanded.StorageCase
+
+  import Commanded.Assertions.EventAssertions
+
+  alias Commanded.ExampleDomain.BankRouter
+  alias Commanded.ExampleDomain.TransferMoneyProcessManager
+  alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
+  alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited
+  alias Commanded.ExampleDomain.BankAccount.Events.MoneyWithdrawn
+  alias Commanded.ExampleDomain.MoneyTransfer.Commands.TransferMoney
+  alias Commanded.ExampleDomain.MoneyTransfer.Events.MoneyTransferRequested
+  alias Commanded.Helpers.CommandAuditMiddleware
+  alias Commanded.ProcessManagers.ProcessRouter
+
+  setup do
+    CommandAuditMiddleware.start_link()
+    :ok
+  end
+
+  test "should start a process manager in response to an event" do
+    account_number1 = UUID.uuid4()
+    account_number2 = UUID.uuid4()
+    transfer_uuid = UUID.uuid4()
+
+    {:ok, process_router} = TransferMoneyProcessManager.start_link()
+
+    # Create two bank accounts
+    :ok =
+      BankRouter.dispatch(%OpenAccount{account_number: account_number1, initial_balance: 1_000})
+
+    :ok = BankRouter.dispatch(%OpenAccount{account_number: account_number2, initial_balance: 500})
+
+    # Transfer funds between account 1 and account 2
+    :ok =
+      BankRouter.dispatch(%TransferMoney{
+        transfer_uuid: transfer_uuid,
+        debit_account: account_number1,
+        credit_account: account_number2,
+        amount: 100
+      })
+
+    assert_receive_event(MoneyTransferRequested, fn event ->
+      assert event.debit_account == account_number1
+      assert event.credit_account == account_number2
+      assert event.amount == 100
+    end)
+
+    assert_receive_event(MoneyWithdrawn, fn event ->
+      assert event.account_number == account_number1
+      assert event.amount == 100
+      assert event.balance == 900
+    end)
+
+    assert_receive_event(MoneyDeposited, fn event ->
+      assert event.account_number == account_number2
+      assert event.amount == 100
+      assert event.balance == 600
+    end)
+
+    assert [{^transfer_uuid, _}] = ProcessRouter.process_instances(process_router)
+  end
+end
diff --git a/test/process_managers/process_manager_routing_test.exs b/test/process_managers/process_manager_routing_test.exs
index f5a33b2..5dc9c0b 100644
--- a/test/process_managers/process_manager_routing_test.exs
+++ b/test/process_managers/process_manager_routing_test.exs
@@ -1,86 +1,191 @@
 defmodule Commanded.ProcessManagers.ProcessManagerRoutingTest do
-  use Commanded.StorageCase
-
-  import Commanded.Assertions.EventAssertions
-
-  alias Commanded.ExampleDomain.BankRouter
-  alias Commanded.ExampleDomain.TransferMoneyProcessManager
-  alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
-  alias Commanded.ExampleDomain.BankAccount.Events.{MoneyDeposited,MoneyWithdrawn}
-  alias Commanded.ExampleDomain.MoneyTransfer.Commands.TransferMoney
-  alias Commanded.ExampleDomain.BankAccount.Commands.WithdrawMoney
-  alias Commanded.ExampleDomain.MoneyTransfer.Events.MoneyTransferRequested
-  alias Commanded.Helpers.CommandAuditMiddleware
+  use Commanded.MockEventStoreCase
+
+  alias Commanded.Helpers.EventFactory
+  alias Commanded.Helpers.Wait
   alias Commanded.ProcessManagers.ProcessRouter
+  alias Commanded.ProcessManagers.RoutingProcessManager
+  alias Commanded.ProcessManagers.RoutingProcessManager.Continued
+  alias Commanded.ProcessManagers.RoutingProcessManager.Errored
+  alias Commanded.ProcessManagers.RoutingProcessManager.Started
+  alias Commanded.ProcessManagers.RoutingProcessManager.Stopped
 
   setup do
-    CommandAuditMiddleware.start_link()
-    :ok
-  end
+    expect(MockEventStore, :subscribe_to, fn
+      :all, _name, pid, :origin ->
+        send(pid, {:subscribed, self()})
+
+        {:ok, self()}
+    end)
+
+    stub(MockEventStore, :read_snapshot, fn _snapshot_uuid -> {:error, :snapshot_not_found} end)
+    stub(MockEventStore, :record_snapshot, fn _snapshot -> :ok end)
+    stub(MockEventStore, :ack_event, fn _pid, _event -> :ok end)
+
+    {:ok, pid} = RoutingProcessManager.start_link()
 
-  test "should start a process manager in response to an event" do
-    account_number1 = UUID.uuid4
-    account_number2 = UUID.uuid4
-    transfer_uuid = UUID.uuid4
+    [pid: pid, process_uuid: UUID.uuid4()]
+  end
 
-    {:ok, process_router} = TransferMoneyProcessManager.start_link()
+  describe "process manager routing" do
+    test "should start instance on `:start`", %{pid: pid, process_uuid: process_uuid} do
+      send_events(pid, [%Started{process_uuid: process_uuid, reply_to: self()}])
 
-    assert ProcessRouter.process_instances(process_router) == []
+      instance = wait_for_instance(pid, process_uuid)
 
-    :timer.sleep 500
+      assert_receive {:started, ^instance}
+    end
 
-    # create two bank accounts
-    :ok = BankRouter.dispatch(%OpenAccount{account_number: account_number1, initial_balance: 1_000})
-    :ok = BankRouter.dispatch(%OpenAccount{account_number: account_number2, initial_balance:  500})
+    test "should continue existing instance on `:start`", %{pid: pid, process_uuid: process_uuid} do
+      send_events(pid, [
+        %Started{process_uuid: process_uuid, reply_to: self()},
+        %Started{process_uuid: process_uuid, reply_to: self()}
+      ])
 
-    # transfer funds between account 1 and account 2
-    :ok = BankRouter.dispatch(%TransferMoney{
-      transfer_uuid: transfer_uuid,
-      debit_account: account_number1,
-      credit_account: account_number2,
-      amount: 100,
-    })
+      instance = wait_for_instance(pid, process_uuid)
 
-    assert_receive_event MoneyTransferRequested, fn event ->
-      assert event.debit_account == account_number1
-      assert event.credit_account == account_number2
-      assert event.amount == 100
+      assert_receive {:started, ^instance}
+      assert_receive {:started, ^instance}
     end
 
-    assert_receive_event MoneyWithdrawn, fn event ->
-      assert event.account_number == account_number1
-      assert event.amount == 100
-      assert event.balance == 900
+    test "should continue instance on `:continue`", %{pid: pid, process_uuid: process_uuid} do
+      send_events(pid, [
+        %Started{process_uuid: process_uuid, reply_to: self()},
+        %Continued{process_uuid: process_uuid, reply_to: self()}
+      ])
+
+      instance = wait_for_instance(pid, process_uuid)
+
+      assert_receive {:started, ^instance}
+      assert_receive {:continued, ^instance}
     end
 
-    assert_receive_event MoneyDeposited, fn event ->
-      assert event.account_number == account_number2
-      assert event.amount == 100
-      assert event.balance == 600
+    test "should start instance on `:continue`", %{pid: pid, process_uuid: process_uuid} do
+      send_events(pid, [%Continued{process_uuid: process_uuid, reply_to: self()}])
+
+      instance = wait_for_instance(pid, process_uuid)
+
+      refute_receive {:started, ^instance}
+      assert_receive {:continued, ^instance}
     end
 
-    assert [{^transfer_uuid, _}] = ProcessRouter.process_instances(process_router)
+    test "should stop instance on `:stop`", %{pid: pid, process_uuid: process_uuid} do
+      expect(MockEventStore, :delete_snapshot, fn _process_uuid -> :ok end)
+
+      send_events(pid, [%Started{process_uuid: process_uuid, reply_to: self()}])
+
+      instance = wait_for_instance(pid, process_uuid)
+      ref = Process.monitor(instance)
+
+      send_events(pid, [%Stopped{process_uuid: process_uuid}], 2)
+
+      assert_receive {:started, ^instance}
+      assert_receive {:DOWN, ^ref, :process, ^instance, :normal}
+    end
   end
 
-  test "should not create entirely new instance when process manager expected to continue" do
-    account_number1 = UUID.uuid4
-    transfer_uuid = UUID.uuid4
+  describe "process manager strict routing" do
+    test "should start instance on `:start!`", %{pid: pid, process_uuid: process_uuid} do
+      send_events(pid, [%Started{process_uuid: process_uuid, reply_to: self(), strict?: true}])
+
+      instance = wait_for_instance(pid, process_uuid)
 
-    {:ok, process_router} = TransferMoneyProcessManager.start_link()
+      assert_receive {:started, ^instance}
+    end
+
+    test "should error on `:start!` when instance already started", %{
+      pid: pid,
+      process_uuid: process_uuid
+    } do
+      Process.unlink(pid)
+      ref = Process.monitor(pid)
+
+      send_events(pid, [
+        %Started{process_uuid: process_uuid, reply_to: self(), strict?: true},
+        %Started{process_uuid: process_uuid, reply_to: self(), strict?: true}
+      ])
+
+      assert_receive {:started, _instance}
+      refute_receive {:started, _instance}
+      assert_receive {:error, {:start!, :process_already_started}}
+      assert_receive {:DOWN, ^ref, :process, ^pid, {:start!, :process_already_started}}
+    end
 
-    assert ProcessRouter.process_instances(process_router) == []
+    test "should continue instance on `:continue!`", %{pid: pid, process_uuid: process_uuid} do
+      send_events(pid, [
+        %Started{process_uuid: process_uuid, reply_to: self(), strict?: true},
+        %Continued{process_uuid: process_uuid, reply_to: self(), strict?: true}
+      ])
 
-    :timer.sleep 500
+      instance = wait_for_instance(pid, process_uuid)
+
+      assert_receive {:started, ^instance}
+      assert_receive {:continued, ^instance}
+    end
+
+    test "should error on `:continue` when instance not already started", %{
+      pid: pid,
+      process_uuid: process_uuid
+    } do
+      Process.unlink(pid)
+      ref = Process.monitor(pid)
+
+      send_events(pid, [%Continued{process_uuid: process_uuid, reply_to: self(), strict?: true}])
+
+      refute_receive {:continued, _instance}
+      assert_receive {:error, {:continue!, :process_not_started}}
+      assert_receive {:DOWN, ^ref, :process, ^pid, {:continue!, :process_not_started}}
+    end
+
+    test "should rescue `interested?` errors", %{
+      pid: pid,
+      process_uuid: process_uuid
+    } do
+      Process.unlink(pid)
+      ref = Process.monitor(pid)
+
+      send_events(pid, [
+        %Errored{process_uuid: process_uuid, reply_to: self(), on_error: {:stop, :error}}
+      ])
+
+      assert_receive {:error, %RuntimeError{message: "error"}}
+      assert_receive {:DOWN, ^ref, :process, ^pid, :error}
+    end
+
+    test "should continue processing events after `:skip` error", %{
+      pid: pid,
+      process_uuid: process_uuid
+    } do
+      ref = Process.monitor(pid)
+
+      send_events(pid, [
+        %Started{process_uuid: process_uuid, reply_to: self()},
+        %Errored{process_uuid: process_uuid, reply_to: self(), on_error: :skip},
+        %Continued{process_uuid: process_uuid, reply_to: self()}
+      ])
+
+      instance = wait_for_instance(pid, process_uuid)
+
+      assert_receive {:started, ^instance}
+      assert_receive {:error, %RuntimeError{message: "error"}}
+      assert_receive {:continued, ^instance}
+      refute_receive {:DOWN, ^ref, :process, ^pid, _}
+    end
+  end
+
+  defp send_events(pid, events, initial_event_number \\ 1) do
+    recorded_events = EventFactory.map_to_recorded_events(events, initial_event_number)
+
+    send(pid, {:events, recorded_events})
+  end
 
-    :ok = BankRouter.dispatch(%OpenAccount{account_number: account_number1, initial_balance: 1_000})
-    :ok = BankRouter.dispatch(%WithdrawMoney{
-      transfer_uuid: transfer_uuid,
-      account_number: account_number1,
-      amount: 100,
-    })
+  defp wait_for_instance(pid, process_uuid) do
+    Wait.until(fn ->
+      instance = ProcessRouter.process_instance(pid, process_uuid)
 
-    :timer.sleep 300
+      assert is_pid(instance)
 
-    assert [] = ProcessRouter.process_instances(process_router)
+      instance
+    end)
   end
 end
diff --git a/test/process_managers/support/routing/routing_process_manager.ex b/test/process_managers/support/routing/routing_process_manager.ex
new file mode 100644
index 0000000..63647f5
--- /dev/null
+++ b/test/process_managers/support/routing/routing_process_manager.ex
@@ -0,0 +1,85 @@
+defmodule Commanded.ProcessManagers.RoutingProcessManager do
+  use Commanded.ProcessManagers.ProcessManager,
+    name: __MODULE__,
+    router: Commanded.Commands.MockRouter
+
+  defmodule Started do
+    @enforce_keys [:process_uuid]
+    defstruct [:process_uuid, :reply_to, strict?: false]
+  end
+
+  defmodule Continued do
+    @enforce_keys [:process_uuid]
+    defstruct [:process_uuid, :reply_to, strict?: false]
+  end
+
+  defmodule Stopped do
+    @enforce_keys [:process_uuid]
+    defstruct [:process_uuid]
+  end
+
+  defmodule Errored do
+    @enforce_keys [:process_uuid]
+    defstruct [:process_uuid, :reply_to, :on_error]
+  end
+
+  alias Commanded.ProcessManagers.FailureContext
+  alias Commanded.ProcessManagers.RoutingProcessManager
+
+  defstruct [:processes]
+
+  def interested?(%Started{process_uuid: process_uuid, strict?: true}),
+    do: {:start!, process_uuid}
+
+  def interested?(%Started{process_uuid: process_uuid}),
+    do: {:start, process_uuid}
+
+  def interested?(%Continued{process_uuid: process_uuid, strict?: true}),
+    do: {:continue!, process_uuid}
+
+  def interested?(%Continued{process_uuid: process_uuid}), do: {:continue, process_uuid}
+
+  def interested?(%Stopped{process_uuid: process_uuid}), do: {:stop, process_uuid}
+
+  def interested?(%Errored{}), do: raise("error")
+
+  def handle(%RoutingProcessManager{}, %Started{} = event) do
+    %Started{reply_to: reply_to} = event
+
+    send(reply_to, {:started, self()})
+
+    []
+  end
+
+  def handle(%RoutingProcessManager{}, %Continued{} = event) do
+    %Continued{reply_to: reply_to} = event
+
+    send(reply_to, {:continued, self()})
+
+    []
+  end
+
+  def error({:error, error}, %Started{} = event, %FailureContext{}) do
+    %Started{reply_to: reply_to} = event
+
+    send(reply_to, {:error, error})
+
+    {:stop, error}
+  end
+
+  def error({:error, error}, %Continued{} = event, %FailureContext{}) do
+    %Continued{reply_to: reply_to} = event
+
+    send(reply_to, {:error, error})
+
+    {:stop, error}
+  end
+
+  def error({:error, error}, %Errored{} = event, %FailureContext{}) do
+    %Errored{reply_to: reply_to, on_error: on_error} = event
+
+    send(reply_to, {:error, error})
+
+    on_error
+  end
+end
diff --git a/test/support/mocks.ex b/test/support/mocks.ex
index e7260ea..b2bfdf0 100644
--- a/test/support/mocks.ex
+++ b/test/support/mocks.ex
@@ -1,1 +1,2 @@
 Mox.defmock(Commanded.EventStore.Adapters.Mock, for: Commanded.EventStore)
+Mox.defmock(Commanded.Commands.MockRouter, for: Commanded.Commands.Router)