fix(dispatcher): preserve retry failure guarantees

guides/explanations/commands.md

@@ -191,7 +191,9 @@ open_account = %OpenAccount{
 
 ### Timeouts
 
-A command handler has a default timeout of 5 seconds. The same default as a `GenServer.call/3` process call. It must handle the command in this period, otherwise the call fails and the caller process exits.
+A command handler has a default timeout of 5 seconds, the same default as a `GenServer.call/3` process call. It must handle the command in this period; otherwise, dispatch fails with an error tuple.
+
+Depending on where the timeout is observed, this may currently be reported as either `{:error, :aggregate_execution_timeout}` or `{:error, :aggregate_execution_failed}`. Both indicate that command execution did not complete within the configured timeout.
 
 You can configure a different timeout value during command registration by providing a `timeout` option, defined in milliseconds:
 
@@ -216,6 +218,22 @@ open_account = %OpenAccount{account_number: "ACC123", initial_balance: 1_000}
 :ok = BankApp.dispatch(open_account, timeout: 2_000)
 ```
 
+Timeouts are not retried automatically. If aggregate hydration or command handling legitimately needs longer, increase the dispatch timeout explicitly.
+
+### Automatic retries
+
+Command dispatch uses the `retry_attempts` budget to retry a small set of failures automatically:
+
+- `{:error, :wrong_expected_version}` while appending events
+- an aggregate stopping after it has been located but before it can execute the command
+- a remote aggregate node becoming unavailable after the aggregate process has been located
+
+The default retry budget is `10`, and you can override it per dispatch:
+
+```elixir
+:ok = BankApp.dispatch(command, retry_attempts: 5)
+```
+
 ### Multi-command registration
 
 Command routers support multi command registration so you can group related command handlers into the same module:

lib/application.ex

@@ -358,6 +358,12 @@ defmodule Commanded.Application do
 
         - `timeout` - as described above.
 
+        - `retry_attempts` - the number of automatic retries permitted for
+          optimistic concurrency conflicts while appending events, and
+          infrastructure races where an aggregate stops after it has been
+          located or a remote aggregate node becomes unavailable. The default
+          value is 10. Timeouts are not retried automatically.
+
   Returns `:ok` on success unless the `:returning` option is specified where
   it returns one of `{:ok, aggregate_state}`, `{:ok, aggregate_version}`,
   `{:ok, events}`, or `{:ok, %Commanded.Commands.ExecutionResult{}}`.

lib/commanded/aggregates/aggregate.ex

@@ -254,6 +254,10 @@ defmodule Commanded.Aggregates.Aggregate do
   Returns `{:ok, aggregate_version, events}` on success, or `{:error, error}`
   on failure.
 
+  If the aggregate process stops after it has been located but before it
+  replies, this function returns `{:exit, {:normal, :aggregate_stopped}}`.
+  `Commanded.Commands.Dispatcher` handles that case by retrying when permitted.
+
     - `aggregate_version` - the updated version of the aggregate after executing
        the command.
     - `events` - events produced by the command, can be an empty list.

lib/commanded/aggregates/execution_context.ex

@@ -8,7 +8,9 @@ defmodule Commanded.Aggregates.ExecutionContext do
       (e.g. `%OpenBankAccount{...}`).
 
     - `retry_attempts` - the number of retries permitted if an
-      `{:error, :wrong_expected_version}` is encountered when appending events.
+      `{:error, :wrong_expected_version}` is encountered when appending events,
+      or if command dispatch must retry after the aggregate stops or a remote
+      aggregate node becomes unavailable.
 
     - `causation_id` - the UUID assigned to the dispatched command.
 

lib/commanded/commands/dispatcher.ex

@@ -175,15 +175,10 @@ defmodule Commanded.Commands.Dispatcher do
         maybe_retry(pipeline, payload, context)
 
       {:error, error} ->
-        pipeline
-        |> Pipeline.respond({:error, error})
-        |> after_failure(payload)
+        respond_with_failure(pipeline, payload, error)
 
       {:error, error, reason} ->
-        pipeline
-        |> Pipeline.assign(:error_reason, reason)
-        |> Pipeline.respond({:error, error})
-        |> after_failure(payload)
+        respond_with_failure(pipeline, payload, error, reason)
     end
   end
 
@@ -278,13 +273,26 @@ defmodule Commanded.Commands.Dispatcher do
     }
   end
 
+  defp respond_with_failure(%Pipeline{} = pipeline, %Payload{} = payload, error) do
+    pipeline
+    |> Pipeline.respond({:error, error})
+    |> after_failure(payload)
+  end
+
+  defp respond_with_failure(%Pipeline{} = pipeline, %Payload{} = payload, error, reason) do
+    pipeline
+    |> Pipeline.assign(:error_reason, reason)
+    |> Pipeline.respond({:error, error})
+    |> after_failure(payload)
+  end
+
   defp maybe_retry(pipeline, payload, context) do
     case ExecutionContext.retry(context) do
       {:ok, context} ->
         execute(pipeline, payload, context)
 
-      reply ->
-        reply
+      {:error, error} ->
+        respond_with_failure(pipeline, payload, error)
     end
   end
 end

test/aggregates/support/aggregate_telemetry_support.ex

@@ -0,0 +1,36 @@
+defmodule Commanded.Aggregates.AggregateTelemetrySupport do
+  @moduledoc false
+
+  defmodule Commands do
+    defmodule Ok do
+      defstruct [:message]
+    end
+
+    defmodule Error do
+      defstruct [:message]
+    end
+
+    defmodule RaiseException do
+      defstruct [:message]
+    end
+  end
+
+  defmodule Event do
+    @derive Jason.Encoder
+    defstruct [:message]
+  end
+
+  defmodule ExampleAggregate do
+    alias Commanded.Aggregates.AggregateTelemetrySupport.Commands.{Error, Ok, RaiseException}
+    alias Commanded.Aggregates.AggregateTelemetrySupport.Event
+
+    defstruct [:message]
+
+    def execute(%ExampleAggregate{}, %Ok{message: message}), do: %Event{message: message}
+    def execute(%ExampleAggregate{}, %Error{message: message}), do: {:error, message}
+    def execute(%ExampleAggregate{}, %RaiseException{message: message}), do: raise(message)
+
+    def apply(%ExampleAggregate{}, %Event{message: message}),
+      do: %ExampleAggregate{message: message}
+  end
+end

test/application/telemetry_test.exs

@@ -2,12 +2,16 @@ defmodule Commanded.Application.TelemetryTest do
   use ExUnit.Case
 
   alias Commanded.DefaultApp
+  alias Commanded.Commands.{TimeoutCommand, TimeoutRouter}
   alias Commanded.Middleware.Commands.Fail
   alias Commanded.Middleware.Commands.IncrementCount
   alias Commanded.Middleware.Commands.RaiseError
+  alias Commanded.TestSupport.RetryStopOnceAggregate
+  alias Commanded.TestSupport.RetryStopOnceAggregate.Command, as: RetryStopOnceCommand
   alias Commanded.UUID
 
   setup do
+    start_supervised!(RetryStopOnceAggregate.Tracker)
     start_supervised!(DefaultApp)
     attach_telemetry()
 
@@ -31,6 +35,18 @@ defmodule Commanded.Application.TelemetryTest do
       identity: :aggregate_uuid
   end
 
+  defmodule RetryExhaustionRouter do
+    use Commanded.Commands.Router
+
+    alias Commanded.TestSupport.RetryStopOnceAggregate
+    alias Commanded.TestSupport.RetryStopOnceAggregate.Command
+
+    dispatch [Command],
+      to: RetryStopOnceAggregate,
+      identity: :uuid,
+      before_execute: :before_execute
+  end
+
   test "emit `[:commanded, :application, :dispatch, :start | :stop]` event" do
     command = %IncrementCount{aggregate_uuid: UUID.uuid4()}
 
@@ -73,6 +89,35 @@ defmodule Commanded.Application.TelemetryTest do
              meta
   end
 
+  test "emit dispatch stop telemetry when dispatcher retries are exhausted" do
+    aggregate_uuid = UUID.uuid4()
+
+    command = %RetryStopOnceCommand{uuid: aggregate_uuid}
+
+    assert {:error, :too_many_attempts} =
+             RetryExhaustionRouter.dispatch(command, application: DefaultApp, retry_attempts: 0)
+
+    assert_receive {[:commanded, :application, :dispatch, :start], _, _, _}
+
+    assert_receive {[:commanded, :application, :dispatch, :stop], _, _,
+                    %{error: :too_many_attempts}}
+  end
+
+  test "emit a single aggregate execute attempt when command execution times out" do
+    command = %TimeoutCommand{aggregate_uuid: UUID.uuid4(), sleep_in_ms: 2_000}
+
+    assert {:error, error} = TimeoutRouter.dispatch(command, application: DefaultApp)
+    assert error in [:aggregate_execution_failed, :aggregate_execution_timeout]
+
+    assert_receive {[:commanded, :application, :dispatch, :start], _, _, _}
+    assert_receive {[:commanded, :aggregate, :execute, :start], _, _, _}
+
+    refute_receive {[:commanded, :aggregate, :execute, :start], _, _, _}, 200
+
+    assert_receive {[:commanded, :application, :dispatch, :stop], _, _,
+                    %{error: ^error}}
+  end
+
   defp attach_telemetry do
     agent = start_supervised!({Agent, fn -> 1 end})
 

test/commands/distributed_dispatch_test.exs

@@ -0,0 +1,88 @@
+defmodule Commanded.Commands.DistributedDispatchTest do
+  use ExUnit.Case
+
+  alias Commanded.Aggregates.Supervisor
+
+  alias Commanded.Commands.{
+    RemoteNodeDownAggregate,
+    RemoteNodeDownCommand,
+    RemoteNodeDownRouter
+  }
+
+  alias Commanded.{DistributedApp, Registration, UUID}
+
+  @moduletag :distributed
+
+  setup do
+    {"", 0} = System.cmd("epmd", ["-daemon"])
+    :ok = LocalCluster.start()
+
+    {:ok, cluster} =
+      LocalCluster.start_link(1,
+        prefix: "commanded",
+        applications: [:commanded]
+      )
+
+    {:ok, [remote_node]} = LocalCluster.nodes(cluster)
+
+    start_supervised!(DistributedApp)
+    start_distributed_app(remote_node)
+
+    [current_node: node(), remote_node: remote_node]
+  end
+
+  test "should retry command execution when the aggregate node goes down", %{
+    current_node: current_node,
+    remote_node: remote_node
+  } do
+    aggregate_uuid = UUID.uuid4()
+    aggregate_name = {DistributedApp, RemoteNodeDownAggregate, aggregate_uuid}
+
+    command = %RemoteNodeDownCommand{
+      aggregate_uuid: aggregate_uuid,
+      notify: self(),
+      sleep_in_ms: 500
+    }
+
+    assert {:ok, ^aggregate_uuid} =
+             :rpc.call(remote_node, Supervisor, :open_aggregate, [
+               DistributedApp,
+               RemoteNodeDownAggregate,
+               aggregate_uuid
+             ])
+
+    aggregate_pid = Registration.whereis_name(DistributedApp, aggregate_name)
+
+    assert is_pid(aggregate_pid)
+    assert node(aggregate_pid) == remote_node
+
+    :erlang.monitor_node(remote_node, true)
+
+    dispatch_task =
+      Task.async(fn ->
+        RemoteNodeDownRouter.dispatch(command,
+          application: DistributedApp,
+          retry_attempts: 1,
+          timeout: 5_000
+        )
+      end)
+
+    assert_receive {:remote_command_started, ^aggregate_uuid, ^remote_node}, 5_000
+
+    :rpc.cast(remote_node, :erlang, :halt, [])
+
+    assert_receive {:nodedown, ^remote_node}, 5_000
+    assert_receive {:remote_command_started, ^aggregate_uuid, ^current_node}, 5_000
+    assert :ok = Task.await(dispatch_task, 6_000)
+
+    :erlang.monitor_node(remote_node, false)
+  end
+
+  defp start_distributed_app(remote_node) do
+    reply_to = self()
+
+    Node.spawn(remote_node, Commanded.DistributedTestHelper, :start_distributed_app, [reply_to])
+
+    assert_receive {:distributed_app_started, ^remote_node}, 5_000
+  end
+end

test/commands/support/remote_node_down_aggregate.ex

@@ -0,0 +1,4 @@
+defmodule Commanded.Commands.RemoteNodeDownAggregate do
+  @moduledoc false
+  defstruct []
+end

test/commands/support/remote_node_down_command.ex

@@ -0,0 +1,4 @@
+defmodule Commanded.Commands.RemoteNodeDownCommand do
+  @moduledoc false
+  defstruct aggregate_uuid: nil, notify: nil, sleep_in_ms: nil
+end

test/commands/support/remote_node_down_handler.ex

@@ -0,0 +1,19 @@
+defmodule Commanded.Commands.RemoteNodeDownHandler do
+  @moduledoc false
+  @behaviour Commanded.Commands.Handler
+
+  alias Commanded.Commands.{RemoteNodeDownAggregate, RemoteNodeDownCommand}
+
+  def handle(%RemoteNodeDownAggregate{}, %RemoteNodeDownCommand{} = command) do
+    %RemoteNodeDownCommand{
+      aggregate_uuid: aggregate_uuid,
+      notify: notify,
+      sleep_in_ms: sleep_in_ms
+    } = command
+
+    send(notify, {:remote_command_started, aggregate_uuid, node()})
+    :timer.sleep(sleep_in_ms)
+
+    []
+  end
+end

test/commands/support/remote_node_down_router.ex

@@ -0,0 +1,15 @@
+defmodule Commanded.Commands.RemoteNodeDownRouter do
+  @moduledoc false
+  use Commanded.Commands.Router
+
+  alias Commanded.Commands.{
+    RemoteNodeDownAggregate,
+    RemoteNodeDownCommand,
+    RemoteNodeDownHandler
+  }
+
+  dispatch RemoteNodeDownCommand,
+    to: RemoteNodeDownHandler,
+    aggregate: RemoteNodeDownAggregate,
+    identity: :aggregate_uuid
+end