refactor(dispatcher): align aggregate failure boundary (#98)

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

guides/explanations/commands.md

@@ -193,7 +193,7 @@ open_account = %OpenAccount{
 
 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.
+If the command does not complete within the configured timeout, dispatch fails with `{:error, :aggregate_execution_timeout}`.
 
 You can configure a different timeout value during command registration by providing a `timeout` option, defined in milliseconds:
 

guides/explanations/fork-differences.md

@@ -52,6 +52,21 @@ This fork maintains an independent release cycle to introduce new features and i
 
 ## Features Added ✨
 
+### **Dispatcher Failure Boundary Normalization**
+[PR #98](https://github.com/straw-hat-team/commanded/pull/98)
+
+**Changes:**
+- Moved aggregate execution exit normalization into `Commanded.Aggregates.Aggregate.execute/5`
+- Removed the dispatcher task wrapper used only for `GenServer.call/3` failure isolation
+- Made command timeouts deterministic as `{:error, :aggregate_execution_timeout}`
+- Preserved abnormal aggregate exit reasons as `{:error, :aggregate_execution_failed, reason}` at the aggregate execution boundary
+- Kept public dispatch failures as `{:error, :aggregate_execution_failed}` while preserving `reason` internally for middleware and logging
+
+**Benefits:**
+- Keeps failure normalization in one place instead of splitting it across aggregate and dispatcher layers
+- Preserves dispatcher safety guarantees without extra runtime infrastructure
+- Gives middleware and logs better diagnostic context for abnormal aggregate exits without widening the public dispatch API
+
 ### **UUIDv7 Support**
 [PR #22](https://github.com/straw-hat-team/commanded/pull/22)
 

lib/commanded/aggregates/aggregate.ex

@@ -258,6 +258,17 @@ defmodule Commanded.Aggregates.Aggregate do
   replies, this function returns `{:exit, {:normal, :aggregate_stopped}}`.
   `Commanded.Commands.Dispatcher` handles that case by retrying when permitted.
 
+  Infrastructure failures are normalized into return values instead of exiting
+  the caller:
+
+    - `{:error, :remote_node_down}` when a remote aggregate node becomes
+      unavailable during execution.
+    - `{:error, :aggregate_execution_timeout}` when the command does not
+      complete within the configured timeout.
+    - `{:error, :aggregate_execution_failed, reason}` for other
+      execution-time exits, preserving the underlying exit reason for
+      middleware and logging.
+
     - `aggregate_version` - the updated version of the aggregate after executing
        the command.
     - `events` - events produced by the command, can be an empty list.
@@ -282,6 +293,20 @@ defmodule Commanded.Aggregates.Aggregate do
 
       :exit, {:normal, {GenServer, :call, [^name, {:execute_command, ^context}, ^timeout]}} ->
         {:exit, {:normal, :aggregate_stopped}}
+
+      :exit,
+      {{:nodedown, _node_name},
+       {GenServer, :call, [^name, {:execute_command, ^context}, ^timeout]}} ->
+        {:error, :remote_node_down}
+
+      :exit, {:timeout, {GenServer, :call, [^name, {:execute_command, ^context}, ^timeout]}} ->
+        {:error, :aggregate_execution_timeout}
+
+      :exit, {reason, {GenServer, :call, [^name, {:execute_command, ^context}, ^timeout]}} ->
+        {:error, :aggregate_execution_failed, reason}
+
+      :exit, reason ->
+        {:error, :aggregate_execution_failed, reason}
     end
   end
 

lib/commanded/application/supervisor.ex

@@ -67,15 +67,13 @@ defmodule Commanded.Application.Supervisor do
   end
 
   defp app_child_spec(name, config) do
-    task_dispatcher_name = Module.concat([name, Commanded.Commands.TaskDispatcher])
     aggregates_supervisor_name = Module.concat([name, Commanded.Aggregates.Supervisor])
     subscriptions_name = Module.concat([name, Commanded.Subscriptions])
     registry_name = Module.concat([name, Commanded.Subscriptions.Registry])
     snapshotting = Keyword.get(config, :snapshotting, %{})
     hibernate_after = Keyword.get(config, :hibernate_after, :infinity)
 
     [
-      {Task.Supervisor, name: task_dispatcher_name, hibernate_after: hibernate_after},
       {Commanded.Aggregates.Supervisor,
        name: aggregates_supervisor_name,
        application: name,

lib/commanded/commands/dispatcher.ex

@@ -120,34 +120,7 @@ defmodule Commanded.Commands.Dispatcher do
         context.metadata
       )
 
-    task_dispatcher_name = Module.concat([application, Commanded.Commands.TaskDispatcher])
-
-    task =
-      Task.Supervisor.async_nolink(task_dispatcher_name, Aggregate, :execute, [
-        application,
-        aggregate_module,
-        aggregate_uuid,
-        context,
-        timeout
-      ])
-
-    result =
-      case Task.yield(task, timeout) || Task.shutdown(task) do
-        {:ok, result} ->
-          result
-
-        {:exit, {:normal, :aggregate_stopped}} = result ->
-          result
-
-        {:exit, {{:nodedown, _node_name}, {GenServer, :call, _}}} ->
-          {:error, :remote_node_down}
-
-        {:exit, _reason} ->
-          {:error, :aggregate_execution_failed}
-
-        nil ->
-          {:error, :aggregate_execution_timeout}
-      end
+    result = Aggregate.execute(application, aggregate_module, aggregate_uuid, context, timeout)
 
     case result do
       {:ok, aggregate_version, events, aggregate_state} ->

test/aggregates/execute_command_test.exs

@@ -4,14 +4,32 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do
   import Commanded.Helpers.ProcessHelper, only: [shutdown_aggregate: 3]
 
   alias Commanded.Aggregates.{Aggregate, ExecutionContext}
+  alias Commanded.Commands.{TimeoutAggregateRoot, TimeoutCommand, TimeoutCommandHandler}
   alias Commanded.ExampleDomain.{BankAccount, BankApp, OpenAccountHandler}
   alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
   alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
   alias Commanded.Helpers.Wait
-  alias Commanded.{Registration, UUID}
+  alias Commanded.TestSupport.RetryStopOnceAggregate
+  alias Commanded.TestSupport.RetryStopOnceAggregate.Command, as: RetryStopOnceCommand
+  alias Commanded.{DefaultApp, Registration, UUID}
+
+  defmodule CrashCommand do
+    defstruct [:uuid]
+  end
+
+  defmodule CrashBeforeExecuteAggregate do
+    alias Commanded.Aggregates.ExecutionContext
+
+    defstruct []
+
+    def before_execute(_aggregate_state, %ExecutionContext{}), do: Process.exit(self(), :boom)
+    def execute(%__MODULE__{}, %CrashCommand{}), do: []
+  end
 
   setup do
     start_supervised!(BankApp)
+    start_supervised!(DefaultApp)
+    start_supervised!(RetryStopOnceAggregate.Tracker)
 
     :ok
   end
@@ -100,6 +118,68 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do
     assert state_before == Aggregate.aggregate_state(BankApp, BankAccount, account_number)
   end
 
+  test "returns aggregate_stopped when aggregate stops after being opened" do
+    aggregate_uuid = UUID.uuid4()
+
+    assert {:ok, ^aggregate_uuid} =
+             Commanded.Aggregates.Supervisor.open_aggregate(
+               DefaultApp,
+               RetryStopOnceAggregate,
+               aggregate_uuid
+             )
+
+    context = %ExecutionContext{
+      command: %RetryStopOnceCommand{uuid: aggregate_uuid},
+      handler: RetryStopOnceAggregate,
+      function: :execute,
+      before_execute: :before_execute
+    }
+
+    assert {:exit, {:normal, :aggregate_stopped}} =
+             Aggregate.execute(DefaultApp, RetryStopOnceAggregate, aggregate_uuid, context)
+  end
+
+  test "returns aggregate_execution_timeout when command execution exceeds the timeout" do
+    aggregate_uuid = UUID.uuid4()
+
+    assert {:ok, ^aggregate_uuid} =
+             Commanded.Aggregates.Supervisor.open_aggregate(
+               DefaultApp,
+               TimeoutAggregateRoot,
+               aggregate_uuid
+             )
+
+    context = %ExecutionContext{
+      command: %TimeoutCommand{aggregate_uuid: aggregate_uuid, sleep_in_ms: 200},
+      handler: TimeoutCommandHandler,
+      function: :handle
+    }
+
+    assert {:error, :aggregate_execution_timeout} =
+             Aggregate.execute(DefaultApp, TimeoutAggregateRoot, aggregate_uuid, context, 50)
+  end
+
+  test "returns aggregate_execution_failed with the exit reason when the aggregate exits abnormally" do
+    aggregate_uuid = UUID.uuid4()
+
+    assert {:ok, ^aggregate_uuid} =
+             Commanded.Aggregates.Supervisor.open_aggregate(
+               DefaultApp,
+               CrashBeforeExecuteAggregate,
+               aggregate_uuid
+             )
+
+    context = %ExecutionContext{
+      command: %CrashCommand{uuid: aggregate_uuid},
+      handler: CrashBeforeExecuteAggregate,
+      function: :execute,
+      before_execute: :before_execute
+    }
+
+    assert {:error, :aggregate_execution_failed, :boom} =
+             Aggregate.execute(DefaultApp, CrashBeforeExecuteAggregate, aggregate_uuid, context)
+  end
+
   describe "command dispatch return" do
     alias Commanded.Aggregates.ReturnValue.Command
     alias Commanded.Aggregates.ReturnValue.Event

test/application/application_supervisor_test.exs

@@ -12,7 +12,6 @@ defmodule Commanded.Aggregates.ApplicationSupervisorTest do
 
     test "children are hibernated after inactivity" do
       for module <- [
-            Commanded.Commands.TaskDispatcher,
             Commanded.Aggregates.Supervisor,
             Commanded.Subscriptions,
             Commanded.Subscriptions.Registry

test/application/telemetry_test.exs

@@ -106,16 +106,16 @@ defmodule Commanded.Application.TelemetryTest do
   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 {:error, :aggregate_execution_timeout} =
+             TimeoutRouter.dispatch(command, application: DefaultApp)
 
     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}}
+                    %{error: :aggregate_execution_timeout}}
   end
 
   defp attach_telemetry do

test/commands/command_timeout_test.exs

@@ -14,11 +14,8 @@ defmodule Commanded.Commands.CommandTimeoutTest do
     command = %TimeoutCommand{aggregate_uuid: UUID.uuid4(), sleep_in_ms: 2_000}
 
     # Handler is set to take longer than the configured timeout
-    case TimeoutRouter.dispatch(command, application: DefaultApp) do
-      {:error, :aggregate_execution_failed} -> :ok
-      {:error, :aggregate_execution_timeout} -> :ok
-      reply -> flunk("received an unexpected response: #{inspect(reply)}")
-    end
+    assert {:error, :aggregate_execution_timeout} =
+             TimeoutRouter.dispatch(command, application: DefaultApp)
   end
 
   test "should succeed when handler completes within configured timeout" do

test/middleware/middleware_test.exs

@@ -2,6 +2,7 @@ defmodule Commanded.Middleware.MiddlewareTest do
   use ExUnit.Case
 
   import Commanded.Enumerable
+  import ExUnit.CaptureLog
 
   alias Commanded.Commands.ExecutionResult
   alias Commanded.DefaultApp
@@ -21,6 +22,19 @@ defmodule Commanded.Middleware.MiddlewareTest do
   alias Commanded.TestSupport.RetryStopOnceAggregate.Command, as: RetryStopOnceCommand
   alias Commanded.UUID
 
+  defmodule CrashCommand do
+    defstruct [:uuid]
+  end
+
+  defmodule CrashBeforeExecuteAggregate do
+    alias Commanded.Aggregates.ExecutionContext
+
+    defstruct []
+
+    def before_execute(_aggregate_state, %ExecutionContext{}), do: Process.exit(self(), :boom)
+    def execute(%__MODULE__{}, %CrashCommand{}), do: []
+  end
+
   defmodule FirstMiddleware do
     @behaviour Commanded.Middleware
 
@@ -82,6 +96,17 @@ defmodule Commanded.Middleware.MiddlewareTest do
       before_execute: :before_execute
   end
 
+  defmodule CrashRouter do
+    use Commanded.Commands.Router
+
+    middleware Commanded.Middleware.Logger
+
+    dispatch [CrashCommand],
+      to: CrashBeforeExecuteAggregate,
+      identity: :uuid,
+      before_execute: :before_execute
+  end
+
   setup do
     start_supervised!(CommandAuditMiddleware)
     start_supervised!(RetryStopOnceAggregate.Tracker)
@@ -150,12 +175,8 @@ defmodule Commanded.Middleware.MiddlewareTest do
   test "should execute middleware failure callback when aggregate process dies" do
     command = %Timeout{aggregate_uuid: UUID.uuid4()}
 
-    # Force command handling to timeout so the aggregate process is terminated
-    :ok =
-      case Router.dispatch(command, application: DefaultApp, timeout: 50) do
-        {:error, :aggregate_execution_timeout} -> :ok
-        {:error, :aggregate_execution_failed} -> :ok
-      end
+    assert {:error, :aggregate_execution_timeout} =
+             Router.dispatch(command, application: DefaultApp, timeout: 50)
 
     {dispatched, succeeded, failed} = CommandAuditMiddleware.count_commands()
 
@@ -164,6 +185,19 @@ defmodule Commanded.Middleware.MiddlewareTest do
     assert failed == 1
   end
 
+  test "should preserve abnormal aggregate exit reason for middleware logging" do
+    command = %CrashCommand{uuid: UUID.uuid4()}
+
+    log =
+      capture_log(fn ->
+        assert {:error, :aggregate_execution_failed} =
+                 CrashRouter.dispatch(command, application: DefaultApp)
+      end)
+
+    assert log =~ "failed :aggregate_execution_failed"
+    assert log =~ "due to: :boom"
+  end
+
   test "should let a middleware update the metadata" do
     command = %IncrementCount{aggregate_uuid: UUID.uuid4(), by: 1}