From f7b3adf2c00f3844ef85c97f7169f749bea2c8cc Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 28 Apr 2026 22:54:43 -0400 Subject: [PATCH 1/2] refactor(dispatcher): align aggregate failure boundary Signed-off-by: Yordis Prieto --- guides/explanations/commands.md | 2 +- lib/commanded/aggregates/aggregate.ex | 20 +++++ lib/commanded/application/supervisor.ex | 2 - lib/commanded/commands/dispatcher.ex | 29 +------ test/aggregates/execute_command_test.exs | 82 ++++++++++++++++++- .../application_supervisor_test.exs | 1 - test/application/telemetry_test.exs | 6 +- test/commands/command_timeout_test.exs | 7 +- test/middleware/middleware_test.exs | 8 +- 9 files changed, 110 insertions(+), 47 deletions(-) diff --git a/guides/explanations/commands.md b/guides/explanations/commands.md index 500453f2..bc404529 100644 --- a/guides/explanations/commands.md +++ b/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: diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index 31686ef1..4abe0ddc 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -258,6 +258,15 @@ 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}` for other execution-time exits. + - `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 +291,17 @@ 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 -> + {:error, :aggregate_execution_failed} end end diff --git a/lib/commanded/application/supervisor.ex b/lib/commanded/application/supervisor.ex index 71632e5b..5a61feb7 100644 --- a/lib/commanded/application/supervisor.ex +++ b/lib/commanded/application/supervisor.ex @@ -67,7 +67,6 @@ 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]) @@ -75,7 +74,6 @@ defmodule Commanded.Application.Supervisor do 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, diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex index c3d3b24c..034e6ac7 100644 --- a/lib/commanded/commands/dispatcher.ex +++ b/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} -> diff --git a/test/aggregates/execute_command_test.exs b/test/aggregates/execute_command_test.exs index 44885721..a810ec5c 100644 --- a/test/aggregates/execute_command_test.exs +++ b/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 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} = + Aggregate.execute(DefaultApp, CrashBeforeExecuteAggregate, aggregate_uuid, context) + end + describe "command dispatch return" do alias Commanded.Aggregates.ReturnValue.Command alias Commanded.Aggregates.ReturnValue.Event diff --git a/test/application/application_supervisor_test.exs b/test/application/application_supervisor_test.exs index 98a5f396..681b1ffe 100644 --- a/test/application/application_supervisor_test.exs +++ b/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 diff --git a/test/application/telemetry_test.exs b/test/application/telemetry_test.exs index fd98f779..b1d685b4 100644 --- a/test/application/telemetry_test.exs +++ b/test/application/telemetry_test.exs @@ -106,8 +106,8 @@ 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], _, _, _} @@ -115,7 +115,7 @@ defmodule Commanded.Application.TelemetryTest do refute_receive {[:commanded, :aggregate, :execute, :start], _, _, _}, 200 assert_receive {[:commanded, :application, :dispatch, :stop], _, _, - %{error: ^error}} + %{error: :aggregate_execution_timeout}} end defp attach_telemetry do diff --git a/test/commands/command_timeout_test.exs b/test/commands/command_timeout_test.exs index fe16f86d..d907e48e 100644 --- a/test/commands/command_timeout_test.exs +++ b/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 diff --git a/test/middleware/middleware_test.exs b/test/middleware/middleware_test.exs index 72b0ea1e..6ce37c8e 100644 --- a/test/middleware/middleware_test.exs +++ b/test/middleware/middleware_test.exs @@ -150,12 +150,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() From 5c4979e83dda06117c15561657b2abfa046b493c Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 28 Apr 2026 23:17:38 -0400 Subject: [PATCH 2/2] fix(dispatcher): keep aggregate exit reasons observable Signed-off-by: Yordis Prieto --- guides/explanations/fork-differences.md | 15 ++++++++++ lib/commanded/aggregates/aggregate.ex | 11 +++++-- test/aggregates/execute_command_test.exs | 4 +-- test/middleware/middleware_test.exs | 38 ++++++++++++++++++++++++ 4 files changed, 63 insertions(+), 5 deletions(-) diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index a3fad976..573c72b1 100644 --- a/guides/explanations/fork-differences.md +++ b/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) diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index 4abe0ddc..5e321873 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -265,7 +265,9 @@ defmodule Commanded.Aggregates.Aggregate do unavailable during execution. - `{:error, :aggregate_execution_timeout}` when the command does not complete within the configured timeout. - - `{:error, :aggregate_execution_failed}` for other execution-time exits. + - `{: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. @@ -300,8 +302,11 @@ defmodule Commanded.Aggregates.Aggregate do :exit, {:timeout, {GenServer, :call, [^name, {:execute_command, ^context}, ^timeout]}} -> {:error, :aggregate_execution_timeout} - :exit, _reason -> - {:error, :aggregate_execution_failed} + :exit, {reason, {GenServer, :call, [^name, {:execute_command, ^context}, ^timeout]}} -> + {:error, :aggregate_execution_failed, reason} + + :exit, reason -> + {:error, :aggregate_execution_failed, reason} end end diff --git a/test/aggregates/execute_command_test.exs b/test/aggregates/execute_command_test.exs index a810ec5c..889d9cf6 100644 --- a/test/aggregates/execute_command_test.exs +++ b/test/aggregates/execute_command_test.exs @@ -159,7 +159,7 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do Aggregate.execute(DefaultApp, TimeoutAggregateRoot, aggregate_uuid, context, 50) end - test "returns aggregate_execution_failed when the aggregate exits abnormally" do + test "returns aggregate_execution_failed with the exit reason when the aggregate exits abnormally" do aggregate_uuid = UUID.uuid4() assert {:ok, ^aggregate_uuid} = @@ -176,7 +176,7 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do before_execute: :before_execute } - assert {:error, :aggregate_execution_failed} = + assert {:error, :aggregate_execution_failed, :boom} = Aggregate.execute(DefaultApp, CrashBeforeExecuteAggregate, aggregate_uuid, context) end diff --git a/test/middleware/middleware_test.exs b/test/middleware/middleware_test.exs index 6ce37c8e..03663ab9 100644 --- a/test/middleware/middleware_test.exs +++ b/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) @@ -160,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}