Repository: commanded/commanded
Pull request #193 (2018-08-26 22:18:01)

Process manager event handling timeout
You can configure a timeout for event handling to ensure that events are processed in a timely manner without getting stuck.

An `event_timeout` option, defined in milliseconds, may be provided when using the `Commanded.ProcessManagers.ProcessManager` macro at compile time:

```elixir
defmodule TransferMoneyProcessManager do
  use Commanded.ProcessManagers.ProcessManager,
    name: "TransferMoneyProcessManager",
    router: BankRouter,
    event_timeout: :timer.minutes(10)
end
```

Or may be configured when starting a process manager:

```elixir
{: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.


Patch:
diff --git a/guides/Process Managers.md b/guides/Process Managers.md
index fde7660..afbb59b 100644
--- a/guides/Process Managers.md
+++ b/guides/Process Managers.md
@@ -49,8 +49,33 @@ The `error/3` callback function must return one of the following responses depen
 
 - `{:stop, reason}` - stop the process manager with the given reason.
 
+## Supervision
+
+Supervise you process managers to ensure they are restarted on error.
+
+```elixir
+defmodule Bank.Payments.Supervisor do
+  use Supervisor
+
+  def start_link(_arg) do
+    Supervisor.start_link(__MODULE__, [], name: __MODULE__)
+  end
+
+  def init(_arg) do
+    Supervisor.init(
+      [
+        Bank.Payments.TransferMoneyProcessManager
+      ],
+      strategy: :one_for_one
+    )
+  end
+end
+```
+
 ### Error handling example
 
+Define an `error/3` callback function to determine how to handle errors during event handling and command dispatch.
+
 ```elixir
 defmodule ExampleProcessManager do
   use Commanded.ProcessManagers.ProcessManager,
@@ -72,9 +97,7 @@ defmodule ExampleProcessManager do
 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.
+The default behaviour if you don't provide an `error/3` callback is to stop the process manager using the same error reason returned from the failed command dispatch.
 
 ## Example process manager
 
@@ -134,3 +157,27 @@ The name given to the process manager *must* be unique. This is used when subscr
 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.
 
 Process manager instance state is persisted to storage after each handled event. This allows the process manager to resume should the host process terminate.
+
+## Event handling timeout
+
+You can configure a timeout for event handling to ensure that events are processed in a timely manner without getting stuck.
+
+An `event_timeout` option, defined in milliseconds, may be provided when using the `Commanded.ProcessManagers.ProcessManager` macro at compile time:
+
+```elixir
+defmodule TransferMoneyProcessManager do
+  use Commanded.ProcessManagers.ProcessManager,
+    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))
+```
+
+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/event/handler.ex b/lib/commanded/event/handler.ex
index 93eed51..da3439f 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -298,11 +298,11 @@ defmodule Commanded.Event.Handler do
   def parse_name(_module, name), do: inspect(name)
 
   @doc false
-  def start_opts(module, module_opts, local_opts) do
+  def start_opts(module, module_opts, local_opts, additional_allowed_opts \\ []) do
     {valid, invalid} =
       module_opts
       |> Keyword.merge(local_opts)
-      |> Keyword.split([:consistency, :start_from])
+      |> Keyword.split([:consistency, :start_from] ++ additional_allowed_opts)
 
     if Enum.any?(invalid) do
       raise "#{inspect(module)} specifies invalid options: #{inspect(Keyword.keys(invalid))}"
diff --git a/lib/commanded/process_managers/process_manager.ex b/lib/commanded/process_managers/process_manager.ex
index e21cb70..d3abc23 100644
--- a/lib/commanded/process_managers/process_manager.ex
+++ b/lib/commanded/process_managers/process_manager.ex
@@ -235,7 +235,8 @@ defmodule Commanded.ProcessManagers.ProcessManager do
           Commanded.Event.Handler.start_opts(
             __MODULE__,
             Keyword.drop(@opts, [:name, :router]),
-            opts
+            opts,
+            [:event_timeout]
           )
 
         Commanded.ProcessManagers.ProcessRouter.start_link(@name, __MODULE__, @router, opts)
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index da910d6..9adbb22 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -17,12 +17,14 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     defstruct [
       :command_dispatcher,
       :consistency,
+      :event_timeout,
       :process_manager_name,
       :process_manager_module,
       :subscribe_from,
       :supervisor,
       :subscription,
       :last_seen_event,
+      :process_event_timer,
       process_managers: %{},
       pending_acks: %{},
       pending_events: []
@@ -37,7 +39,8 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
       process_manager_module: process_manager_module,
       command_dispatcher: command_dispatcher,
       consistency: opts[:consistency] || :eventual,
-      subscribe_from: opts[:start_from] || :origin
+      subscribe_from: opts[:start_from] || :origin,
+      event_timeout: opts[:event_timeout]
     }
 
     Registration.start_link(name, __MODULE__, state)
@@ -187,6 +190,26 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, state}
   end
 
+  # 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
+
+    case Map.get(pending_acks, event_number, []) do
+      [] ->
+        {:noreply, state}
+
+      pending ->
+        Logger.error(fn ->
+          describe(state) <>
+            " has taken longer than " <>
+            inspect(event_timeout) <>
+            "ms to process event #" <> inspect(event_number) <> " and is now stopping"
+        end)
+
+        {:stop, :event_timeout, state}
+    end
+  end
+
   # 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
@@ -211,9 +234,10 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   end
 
   # ignore already seen event
-  defp event_already_seen?(%RecordedEvent{event_number: event_number}, %State{
-         last_seen_event: last_seen_event
-       }) do
+  defp event_already_seen?(
+         %RecordedEvent{event_number: event_number},
+         %State{last_seen_event: last_seen_event}
+       ) do
     not is_nil(last_seen_event) and event_number <= last_seen_event
   end
 
@@ -405,7 +429,31 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
         pending -> [process_instance | pending]
       end)
 
-    %State{state | pending_acks: pending_acks}
+    state = %State{state | pending_acks: pending_acks}
+
+    start_event_timer(event_number, state)
+  end
+
+  # Event timeout not configured
+  defp start_event_timer(_event_number, %State{event_timeout: nil} = state), do: state
+
+  defp start_event_timer(event_number, %State{process_event_timer: process_event_timer} = state)
+       when is_reference(process_event_timer) do
+    Process.cancel_timer(process_event_timer)
+
+    state = %State{state | process_event_timer: nil}
+
+    start_event_timer(event_number, state)
+  end
+
+  defp start_event_timer(event_number, %State{event_timeout: event_timeout} = state)
+       when is_integer(event_timeout) do
+    %State{event_timeout: event_timeout} = state
+
+    process_event_timer =
+      Process.send_after(self(), {:event_timeout, event_number}, event_timeout)
+
+    %State{state | process_event_timer: process_event_timer}
   end
 
   defp describe(%State{process_manager_module: process_manager_module}),


Test patch:
diff --git a/test/process_managers/process_manager_timeout_test.exs b/test/process_managers/process_manager_timeout_test.exs
new file mode 100644
index 0000000..b06f717
--- /dev/null
+++ b/test/process_managers/process_manager_timeout_test.exs
@@ -0,0 +1,45 @@
+defmodule Commanded.ProcessManagers.ProcessManagerTimeoutTest do
+  use Commanded.StorageCase
+
+  alias Commanded.ProcessManagers.ProcessRouter
+  alias Commanded.ProcessManagers.{ExampleRouter, ExampleProcessManager}
+  alias Commanded.ProcessManagers.ExampleAggregate.Commands.{Pause, Start}
+
+  test "should not timeout and shutdown process manager by default" do
+    aggregate_uuid = UUID.uuid4()
+
+    {:ok, process_router} = ExampleProcessManager.start_link()
+    router_ref = Process.monitor(process_router)
+
+    :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
+
+    process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
+    instance_ref = Process.monitor(process_instance)
+
+    :ok = ExampleRouter.dispatch(%Pause{aggregate_uuid: aggregate_uuid})
+
+    # Should not shutdown process manager or instance
+    refute_receive {:DOWN, ^router_ref, _, _, _}
+    refute_receive {:DOWN, ^instance_ref, _, _, _}
+  end
+
+  test "should timeout and shutdown process manager when `event_timeout` configured" do
+    aggregate_uuid = UUID.uuid4()
+
+    {:ok, process_router} = ExampleProcessManager.start_link(event_timeout: 10)
+    router_ref = Process.monitor(process_router)
+
+    Process.unlink(process_router)
+
+    :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
+
+    process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
+    instance_ref = Process.monitor(process_instance)
+
+    :ok = ExampleRouter.dispatch(%Pause{aggregate_uuid: aggregate_uuid})
+
+    # Should shutdown process manager and instance
+    assert_receive {:DOWN, ^router_ref, _, _, :event_timeout}
+    assert_receive {:DOWN, ^instance_ref, _, _, :shutdown}
+  end
+end
diff --git a/test/process_managers/support/example_aggregate.ex b/test/process_managers/support/example_aggregate.ex
index ee65ad7..1c267fb 100644
--- a/test/process_managers/support/example_aggregate.ex
+++ b/test/process_managers/support/example_aggregate.ex
@@ -9,6 +9,7 @@ defmodule Commanded.ProcessManagers.ExampleAggregate do
   defmodule Commands do
     defmodule(Start, do: defstruct([:aggregate_uuid]))
     defmodule(Publish, do: defstruct([:aggregate_uuid, :interesting, :uninteresting]))
+    defmodule(Pause, do: defstruct([:aggregate_uuid]))
     defmodule(Stop, do: defstruct([:aggregate_uuid]))
     defmodule(Error, do: defstruct([:aggregate_uuid]))
     defmodule(Raise, do: defstruct([:aggregate_uuid]))
@@ -19,6 +20,7 @@ defmodule Commanded.ProcessManagers.ExampleAggregate do
     defmodule(Interested, do: defstruct([:aggregate_uuid, :index]))
     defmodule(Uninterested, do: defstruct([:aggregate_uuid, :index]))
     defmodule(Stopped, do: defstruct([:aggregate_uuid]))
+    defmodule(Paused, do: defstruct([:aggregate_uuid]))
     defmodule(Errored, do: defstruct([:aggregate_uuid]))
     defmodule(Raised, do: defstruct([:aggregate_uuid]))
   end
@@ -34,6 +36,10 @@ defmodule Commanded.ProcessManagers.ExampleAggregate do
     )
   end
 
+  def pause(%ExampleAggregate{uuid: aggregate_uuid}) do
+    %Events.Paused{aggregate_uuid: aggregate_uuid}
+  end
+
   def stop(%ExampleAggregate{uuid: aggregate_uuid}) do
     %Events.Stopped{aggregate_uuid: aggregate_uuid}
   end
@@ -70,6 +76,9 @@ defmodule Commanded.ProcessManagers.ExampleAggregate do
   def apply(%ExampleAggregate{items: items} = state, %Events.Interested{index: index}),
     do: %ExampleAggregate{state | items: items ++ [index]}
 
+  def apply(%ExampleAggregate{} = state, %Events.Paused{}),
+    do: %ExampleAggregate{state | state: :paused}
+
   def apply(%ExampleAggregate{} = state, %Events.Errored{}),
     do: %ExampleAggregate{state | state: :errored}
 
diff --git a/test/process_managers/support/example_command_handler.ex b/test/process_managers/support/example_command_handler.ex
index 6aebf33..1c34796 100644
--- a/test/process_managers/support/example_command_handler.ex
+++ b/test/process_managers/support/example_command_handler.ex
@@ -6,6 +6,7 @@ defmodule Commanded.ProcessManagers.ExampleCommandHandler do
 
   alias Commanded.ProcessManagers.ExampleAggregate.Commands.{
     Error,
+    Pause,
     Publish,
     Raise,
     Start,
@@ -21,6 +22,9 @@ defmodule Commanded.ProcessManagers.ExampleCommandHandler do
     ExampleAggregate.publish(aggregate, interesting, uninteresting)
   end
 
+  def handle(%ExampleAggregate{} = aggregate, %Pause{}),
+    do: ExampleAggregate.pause(aggregate)
+
   def handle(%ExampleAggregate{} = aggregate, %Stop{}),
     do: ExampleAggregate.stop(aggregate)
 
diff --git a/test/process_managers/support/example_process_manager.ex b/test/process_managers/support/example_process_manager.ex
index eb1c925..dda586a 100644
--- a/test/process_managers/support/example_process_manager.ex
+++ b/test/process_managers/support/example_process_manager.ex
@@ -1,11 +1,13 @@
 defmodule Commanded.ProcessManagers.ExampleProcessManager do
   @moduledoc false
+
   alias Commanded.ProcessManagers.{ExampleProcessManager, ExampleRouter}
   alias Commanded.ProcessManagers.ExampleAggregate.Commands.Stop
 
   alias Commanded.ProcessManagers.ExampleAggregate.Events.{
     Errored,
     Interested,
+    Paused,
     Raised,
     Started,
     Stopped
@@ -19,6 +21,7 @@ defmodule Commanded.ProcessManagers.ExampleProcessManager 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?(%Paused{aggregate_uuid: aggregate_uuid}), do: {:continue, aggregate_uuid}
   def interested?(%Errored{aggregate_uuid: aggregate_uuid}), do: {:continue, aggregate_uuid}
   def interested?(%Raised{aggregate_uuid: aggregate_uuid}), do: {:continue, aggregate_uuid}
   def interested?(%Stopped{aggregate_uuid: aggregate_uuid}), do: {:stop, aggregate_uuid}
@@ -27,6 +30,11 @@ defmodule Commanded.ProcessManagers.ExampleProcessManager do
     %Stop{aggregate_uuid: aggregate_uuid}
   end
 
+  # Simulate a "stuck" process
+  def handle(%ExampleProcessManager{}, %Paused{}) do
+    :timer.sleep(:infinity)
+  end
+
   def handle(%ExampleProcessManager{}, %Errored{}), do: {:error, :failed}
 
   def handle(%ExampleProcessManager{}, %Raised{}), do: raise("failed")
diff --git a/test/process_managers/support/example_router.ex b/test/process_managers/support/example_router.ex
index 50673c7..956551a 100644
--- a/test/process_managers/support/example_router.ex
+++ b/test/process_managers/support/example_router.ex
@@ -3,9 +3,17 @@ defmodule Commanded.ProcessManagers.ExampleRouter do
   use Commanded.Commands.Router
 
   alias Commanded.ProcessManagers.{ExampleAggregate, ExampleCommandHandler}
-  alias Commanded.ProcessManagers.ExampleAggregate.Commands.{Error, Publish, Raise, Start, Stop}
 
-  dispatch [Error, Publish, Raise, Start, Stop],
+  alias Commanded.ProcessManagers.ExampleAggregate.Commands.{
+    Error,
+    Pause,
+    Publish,
+    Raise,
+    Start,
+    Stop
+  }
+
+  dispatch [Error, Pause, Publish, Raise, Start, Stop],
     to: ExampleCommandHandler,
     aggregate: ExampleAggregate,
     identity: :aggregate_uuid
diff --git a/test/process_managers/support/resume/resume_process_manager.ex b/test/process_managers/support/resume/resume_process_manager.ex
index 213bdd5..e2974b0 100644
--- a/test/process_managers/support/resume/resume_process_manager.ex
+++ b/test/process_managers/support/resume/resume_process_manager.ex
@@ -4,11 +4,9 @@ defmodule Commanded.ProcessManagers.ResumeProcessManager do
     name: "resume-process-manager",
     router: ResumeRouter
 
-  defstruct [
-    status_history: []
-  ]
+  defstruct status_history: []
 
-  alias Commanded.ProcessManagers.ResumeAggregate.Events.{ProcessStarted,ProcessResumed}
+  alias Commanded.ProcessManagers.ResumeAggregate.Events.{ProcessStarted, ProcessResumed}
   alias Commanded.ProcessManagers.ResumeProcessManager
 
   def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
@@ -17,17 +15,19 @@ defmodule Commanded.ProcessManagers.ResumeProcessManager do
   def handle(%ResumeProcessManager{}, %ProcessStarted{}), do: []
   def handle(%ResumeProcessManager{}, %ProcessResumed{}), do: []
 
-  ## state mutators
+  # State mutators
 
-  def apply(%ResumeProcessManager{status_history: status_history} = process, %ProcessStarted{status: status}) do
-    %ResumeProcessManager{process |
-      status_history: status_history ++ [status]
-    }
+  def apply(%ResumeProcessManager{} = process, %ProcessStarted{} = event) do
+    %ResumeProcessManager{status_history: status_history} = process
+    %ProcessStarted{status: status} = event
+
+    %ResumeProcessManager{process | status_history: status_history ++ [status]}
   end
 
-  def apply(%ResumeProcessManager{status_history: status_history} = process, %ProcessResumed{status: status}) do
-    %ResumeProcessManager{process |
-      status_history: status_history ++ [status]
-    }
+  def apply(%ResumeProcessManager{} = process, %ProcessResumed{} = event) do
+    %ResumeProcessManager{status_history: status_history} = process
+    %ProcessResumed{status: status} = event
+
+    %ResumeProcessManager{process | status_history: status_history ++ [status]}
   end
 end