Repository: commanded/commanded (elixir)

Issue:
Process manager idle timeout
Allow an idle timeout to be specified for process manager processes.

Two possible approaches:

1. Define an `idle_timeout`, in milliseconds, as an option to start link allowing it to be configured when including a process manager in a supervisor:

```elixir
children = [
  {MyApp.FooManager, idle_timeout: 60_000},
]
```

2. Define a lifespan behaviour, similar to aggregate lifespan, to be defined per process manager. It would decide when to shutdown the process depending upon the recently dispatched command, handled event, or error.

Pull request description:
Process manager idle process timeout
Every started instance of a process manager runs indefinitely. To reduce memory usage this pull request adds an idle timeout configuration (in milliseconds) after which an instance process will be shutdown.

The process will be restarted on demand, whenever an event is routed to it, and its state will be rehydrated from the instance snapshot.

## Example

Define the `idle_timeout` option in the process manager module:

```elixir
defmodule ExampleProcessManager do
  use Commanded.ProcessManagers.ProcessManager,
    name: "ExampleProcessManager",
    router: ExampleRouter,
    idle_timeout: :timer.minutes(10)
end
```

Or provide the timeout on start:

```elixir
ExampleProcessManager.start_link(idle_timeout: 10_000)
```

By default the idle timeout will be configured as `:infinity` so every process manager instance will run indefinitely once started.

Fixes #148.

Patch:
diff --git a/.travis.yml b/.travis.yml
index d12ed3d..c78e5e2 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,7 +1,7 @@
 language: elixir
 
 elixir:
-  - 1.8.0
+  - 1.8.2
 
 otp_release:
   - 21.2
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 91bded2..b7ab35b 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -290,11 +290,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 3aab40b..39ecdb4 100644
--- a/lib/commanded/process_managers/process_manager.ex
+++ b/lib/commanded/process_managers/process_manager.ex
@@ -80,6 +80,24 @@ defmodule Commanded.ProcessManagers.ProcessManager do
         end
       end
 
+    # Idle process timeouts
+
+    Each instance of a process manager will run indefinitely once started. To
+    reduce memory usage you can configure an idle timeout, in milliseconds,
+    after which the process will be shutdown.
+
+    The process will be restarted whenever another event is routed to it and its
+    state will be rehydrated from the instance snapshot.
+
+    ## Example
+
+        defmodule ExampleProcessManager do
+          use Commanded.ProcessManagers.ProcessManager,
+            name: "ExampleProcessManager",
+            router: ExampleRouter,
+            idle_timeout: :timer.minutes(10)
+        end
+
   # Consistency
 
   For each process manager you can define its consistency, as one of either
@@ -207,22 +225,27 @@ defmodule Commanded.ProcessManagers.ProcessManager do
     | {:continue, commands :: list(command), context :: map()}
     | {:stop, reason :: term()}
 
+  alias Commanded.Event.Handler
+  alias Commanded.ProcessManagers.ProcessManager
+  alias Commanded.ProcessManagers.ProcessRouter
+
   @doc false
   defmacro __using__(opts) do
     quote location: :keep do
       @before_compile unquote(__MODULE__)
       @on_definition {unquote(__MODULE__), :emit_deprecated_warnings}
 
-      @behaviour Commanded.ProcessManagers.ProcessManager
+      @behaviour ProcessManager
 
       @opts unquote(opts) || []
-      @name Commanded.Event.Handler.parse_name(__MODULE__, @opts[:name])
+      @name Handler.parse_name(__MODULE__, @opts[:name])
       @router @opts[:router] || raise "#{inspect __MODULE__} expects `:router` to be given"
 
       def start_link(opts \\ []) do
-        opts = Commanded.Event.Handler.start_opts(__MODULE__, Keyword.drop(@opts, [:name, :router]), opts)
+        module_opts = Keyword.drop(@opts, [:name, :router])
+        opts = Handler.start_opts(__MODULE__, module_opts, opts, [:idle_timeout])
 
-        Commanded.ProcessManagers.ProcessRouter.start_link(@name, __MODULE__, @router, opts)
+        ProcessRouter.start_link(@name, __MODULE__, @router, opts)
       end
 
       @doc """
diff --git a/lib/commanded/process_managers/process_manager_instance.ex b/lib/commanded/process_managers/process_manager_instance.ex
index c40c359..d865619 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -16,7 +16,8 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   }
 
   defstruct [
-    command_dispatcher: nil,
+    :command_dispatcher,
+    :idle_timeout,
     process_manager_name: nil,
     process_manager_module: nil,
     process_uuid: nil,
@@ -24,18 +25,24 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
     last_seen_event: nil,
   ]
 
-  def start_link(command_dispatcher, process_manager_name, process_manager_module, process_uuid) do
-    GenServer.start_link(__MODULE__, %ProcessManagerInstance{
-      command_dispatcher: command_dispatcher,
-      process_manager_name: process_manager_name,
+  def start_link(opts) do
+    process_manager_module = Keyword.fetch!(opts, :process_manager_module)
+
+    state = %ProcessManagerInstance{
+      command_dispatcher: Keyword.fetch!(opts, :command_dispatcher),
+      idle_timeout: Keyword.fetch!(opts, :idle_timeout),
+      process_manager_name: Keyword.fetch!(opts, :process_manager_name),
       process_manager_module: process_manager_module,
-      process_uuid: process_uuid,
+      process_uuid: Keyword.fetch!(opts, :process_uuid),
       process_state: struct(process_manager_module),
-    })
+    }
+
+    GenServer.start_link(__MODULE__, state)
   end
 
   def init(%ProcessManagerInstance{} = state) do
     GenServer.cast(self(), :fetch_state)
+
     {:ok, state}
   end
 
@@ -66,13 +73,15 @@ 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
 
   @doc false
-  def handle_call(:process_state, _from, %ProcessManagerInstance{process_state: process_state} = state) do
-    {:reply, process_state, state}
+  def handle_call(:process_state, _from, %ProcessManagerInstance{} = state) do
+    %ProcessManagerInstance{idle_timeout: idle_timeout, process_state: process_state} = state
+
+    {:reply, process_state, state, idle_timeout}
   end
 
   @doc """
@@ -102,6 +111,13 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
     end
   end
 
+  @doc false
+  def handle_info(:timeout, %ProcessManagerInstance{} = state) do
+    Logger.debug(fn -> describe(state) <> " stopping due to inactivity timeout" end)
+
+    {:stop, :normal, state}
+  end
+
   defp event_already_seen?(
     %RecordedEvent{event_number: event_number},
     %ProcessManagerInstance{last_seen_event: last_seen_event})
@@ -110,13 +126,18 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   end
 
   defp process_seen_event(event = %RecordedEvent{}, process_router, state) do
+    %ProcessManagerInstance{idle_timeout: idle_timeout} = state
+
     # already seen event, so just ack
     :ok = ack_event(event, process_router)
 
-    {:noreply, state}
+    {:noreply, state, idle_timeout}
   end
 
-  defp process_unseen_event(%RecordedEvent{correlation_id: correlation_id, event_id: event_id, event_number: event_number} = event, process_router, %ProcessManagerInstance{} = state) do
+  defp process_unseen_event(%RecordedEvent{} = event, process_router, %ProcessManagerInstance{} = state) do
+    %RecordedEvent{correlation_id: correlation_id, event_id: event_id, event_number: event_number} = event
+    %ProcessManagerInstance{idle_timeout: idle_timeout} = state
+
     case handle_event(event, state) do
       {:error, reason} ->
         Logger.error(fn -> describe(state) <> " failed to handle event #{inspect event_number} due to: #{inspect reason}" end)
@@ -138,7 +159,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
           :ok = persist_state(state, event_number)
           :ok = ack_event(event, process_router)
 
-          {:noreply, state}
+          {:noreply, state, idle_timeout}
         else
           {:stop, reason} ->
             {:stop, reason, state}
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index c7dffc2..a381868 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -21,6 +21,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     defstruct [
       command_dispatcher: nil,
       consistency: nil,
+      idle_timeout: nil,
       process_manager_name: nil,
       process_manager_module: nil,
       subscribe_from: nil,
@@ -41,6 +42,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
       command_dispatcher: command_dispatcher,
       consistency: opts[:consistency] || :eventual,
       subscribe_from: opts[:start_from] || :origin,
+      idle_timeout: opts[:idle_timeout] || :infinity
     }
 
     Registration.start_link(name, __MODULE__, state)
@@ -143,9 +145,9 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   def handle_info({:subscribed, subscription}, %State{subscription: subscription} = state) do
     Logger.debug(fn -> describe(state) <> " has successfully subscribed to event store" end)
 
-    %State{command_dispatcher: command_dispatcher} = state
+    %State{command_dispatcher: command_dispatcher, idle_timeout: idle_timeout} = state
 
-    {:ok, supervisor} = Supervisor.start_link(command_dispatcher)
+    {:ok, supervisor} = Supervisor.start_link()
 
     {:noreply, %State{state | supervisor: supervisor}}
   end
@@ -274,19 +276,23 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
   defp start_process_manager(process_uuid, %State{} = state) do
     %State{
+      command_dispatcher: command_dispatcher,
+      idle_timeout: idle_timeout,
       process_managers: process_managers,
       process_manager_name: process_manager_name,
       process_manager_module: process_manager_module,
       supervisor: supervisor
     } = state
 
-    {:ok, process_manager} =
-      Supervisor.start_process_manager(
-        supervisor,
-        process_manager_name,
-        process_manager_module,
-        process_uuid
-      )
+    opts = [
+      command_dispatcher: command_dispatcher,
+      idle_timeout: idle_timeout,
+      process_manager_name: process_manager_name,
+      process_manager_module: process_manager_module,
+      process_uuid: process_uuid
+    ]
+
+    {:ok, process_manager} = Supervisor.start_process_manager(supervisor, opts)
 
     Process.monitor(process_manager)
 
diff --git a/lib/commanded/process_managers/supervisor.ex b/lib/commanded/process_managers/supervisor.ex
index 02c78bb..265fef7 100644
--- a/lib/commanded/process_managers/supervisor.ex
+++ b/lib/commanded/process_managers/supervisor.ex
@@ -1,21 +1,19 @@
 defmodule Commanded.ProcessManagers.Supervisor do
   @moduledoc false
+
   use Supervisor
-  require Logger
 
-  def start_link(command_dispatcher) do
-    Supervisor.start_link(__MODULE__, command_dispatcher)
+  def start_link do
+    Supervisor.start_link(__MODULE__, :ok)
   end
 
-  def start_process_manager(supervisor, process_manager_name, process_manager_module, process_uuid) do
-    Logger.debug(fn -> "Starting process manager process for `#{process_manager_module}` with uuid #{process_uuid}" end)
-
-    Supervisor.start_child(supervisor, [process_manager_name, process_manager_module, process_uuid])
+  def start_process_manager(supervisor, opts) do
+    Supervisor.start_child(supervisor, [opts])
   end
 
-  def init(command_dispatcher) do
+  def init(:ok) do
     children = [
-      worker(Commanded.ProcessManagers.ProcessManagerInstance, [command_dispatcher], restart: :temporary),
+      worker(Commanded.ProcessManagers.ProcessManagerInstance, [], restart: :temporary),
     ]
 
     supervise(children, strategy: :simple_one_for_one)


Test patch:
diff --git a/test/process_managers/process_manager_idle_timeout_test.exs b/test/process_managers/process_manager_idle_timeout_test.exs
new file mode 100644
index 0000000..70b0849
--- /dev/null
+++ b/test/process_managers/process_manager_idle_timeout_test.exs
@@ -0,0 +1,80 @@
+defmodule Commanded.ProcessManagers.ProcessManagerIdleTimeoutTest do
+  use Commanded.StorageCase
+
+  alias Commanded.ProcessManagers.ExampleAggregate.Commands.Start
+  alias Commanded.ProcessManagers.ExampleAggregate.Commands.Stop
+  alias Commanded.ProcessManagers.ExampleRouter
+  alias Commanded.ProcessManagers.ProcessRouter
+  alias Commanded.ProcessManagers.TimeoutProcessManager
+  alias Commanded.Helpers.Wait
+
+  describe "process manager idle timeout" do
+    test "should shutdown instance after inactivity" do
+      {:ok, pm} = TimeoutProcessManager.start_link(idle_timeout: 50)
+
+      aggregate_uuid = UUID.uuid4()
+
+      :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
+
+      instance = wait_for_process_instance(pm, aggregate_uuid)
+
+      ref = Process.monitor(instance)
+      assert_receive {:DOWN, ^ref, :process, ^instance, :normal}
+    end
+
+    test "should stop instance on demand" do
+      {:ok, pm} = TimeoutProcessManager.start_link(idle_timeout: 10_000)
+
+      aggregate_uuid = UUID.uuid4()
+
+      :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
+
+      instance = wait_for_process_instance(pm, aggregate_uuid)
+
+      :ok = ExampleRouter.dispatch(%Stop{aggregate_uuid: aggregate_uuid})
+
+      ref = Process.monitor(instance)
+      assert_receive {:DOWN, ^ref, :process, ^instance, :normal}
+    end
+  end
+
+  describe "process manager `:infinity` idle timeout" do
+    test "should not shutdown instance" do
+      {:ok, pm} = TimeoutProcessManager.start_link(idle_timeout: :infinity)
+
+      aggregate_uuid = UUID.uuid4()
+
+      :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
+
+      instance = wait_for_process_instance(pm, aggregate_uuid)
+
+      ref = Process.monitor(instance)
+      refute_receive {:DOWN, ^ref, :process, ^instance, :normal}
+    end
+
+    test "should stop instance on demand" do
+      {:ok, pm} = TimeoutProcessManager.start_link(idle_timeout: :infinity)
+
+      aggregate_uuid = UUID.uuid4()
+
+      :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
+
+      instance = wait_for_process_instance(pm, aggregate_uuid)
+
+      :ok = ExampleRouter.dispatch(%Stop{aggregate_uuid: aggregate_uuid})
+
+      ref = Process.monitor(instance)
+      assert_receive {:DOWN, ^ref, :process, ^instance, :normal}
+    end
+  end
+
+  defp wait_for_process_instance(pm, process_uuid) do
+    Wait.until(fn ->
+      with instance <- ProcessRouter.process_instance(pm, process_uuid) do
+        assert is_pid(instance)
+
+        instance
+      end
+    end)
+  end
+end
diff --git a/test/process_managers/process_manager_instance_test.exs b/test/process_managers/process_manager_instance_test.exs
index 444416b..e4e7b7d 100644
--- a/test/process_managers/process_manager_instance_test.exs
+++ b/test/process_managers/process_manager_instance_test.exs
@@ -11,13 +11,14 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstanceTest do
     transfer_uuid = UUID.uuid4()
     account1_uuid = UUID.uuid4()
     account2_uuid = UUID.uuid4()
-
+    
     {:ok, process_manager} =
       ProcessManagerInstance.start_link(
-        NullRouter,
-        "TransferMoneyProcessManager",
-        TransferMoneyProcessManager,
-        transfer_uuid
+        command_dispatcher: NullRouter,
+        idle_timeout: :infinity,
+        process_manager_name: "TransferMoneyProcessManager",
+        process_manager_module: TransferMoneyProcessManager,
+        process_uuid: transfer_uuid
       )
 
     event = %RecordedEvent{
diff --git a/test/process_managers/support/example_router.ex b/test/process_managers/support/example_router.ex
index 23966fa..be3654f 100644
--- a/test/process_managers/support/example_router.ex
+++ b/test/process_managers/support/example_router.ex
@@ -1,5 +1,6 @@
 defmodule Commanded.ProcessManagers.ExampleRouter do
   @moduledoc false
+  
   use Commanded.Commands.Router
 
   alias Commanded.ProcessManagers.{ExampleAggregate,ExampleCommandHandler}
diff --git a/test/process_managers/support/timeout_process_manager.ex b/test/process_managers/support/timeout_process_manager.ex
new file mode 100644
index 0000000..3212977
--- /dev/null
+++ b/test/process_managers/support/timeout_process_manager.ex
@@ -0,0 +1,24 @@
+defmodule Commanded.ProcessManagers.TimeoutProcessManager do
+  @moduledoc false
+
+  alias Commanded.ProcessManagers.ExampleAggregate.Events.Started
+  alias Commanded.ProcessManagers.ExampleAggregate.Events.Stopped
+  alias Commanded.ProcessManagers.ExampleRouter
+  alias Commanded.ProcessManagers.TimeoutProcessManager
+
+  use Commanded.ProcessManagers.ProcessManager,
+    name: "TimeoutProcessManager",
+    router: ExampleRouter,
+    idle_timeout: :timer.minutes(1)
+
+  defstruct [:status]
+
+  def interested?(%Started{aggregate_uuid: aggregate_uuid}), do: {:start, aggregate_uuid}
+  def interested?(%Stopped{aggregate_uuid: aggregate_uuid}), do: {:stop, aggregate_uuid}
+
+  ## State mutators
+
+  def apply(%TimeoutProcessManager{} = pm, %Started{}) do
+    %TimeoutProcessManager{pm | status: :started}
+  end
+end