From 3af552a3caf1f260909473fd9d6b54965f7bdeb5 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 28 Apr 2026 21:09:13 -0400 Subject: [PATCH 1/3] fix(dispatcher): preserve retry failure guarantees Signed-off-by: Yordis Prieto --- guides/explanations/commands.md | 20 ++++- lib/application.ex | 6 ++ lib/commanded/aggregates/aggregate.ex | 4 + lib/commanded/aggregates/execution_context.ex | 4 +- lib/commanded/commands/dispatcher.ex | 26 ++++-- test/application/telemetry_test.exs | 83 +++++++++++++++++ test/commands/distributed_dispatch_test.exs | 88 +++++++++++++++++++ .../support/remote_node_down_aggregate.ex | 4 + .../support/remote_node_down_command.ex | 4 + .../support/remote_node_down_handler.ex | 19 ++++ .../support/remote_node_down_router.ex | 15 ++++ test/middleware/middleware_test.exs | 55 ++++++++++++ test/support/distributed_test_helper.ex | 9 ++ 13 files changed, 326 insertions(+), 11 deletions(-) create mode 100644 test/commands/distributed_dispatch_test.exs create mode 100644 test/commands/support/remote_node_down_aggregate.ex create mode 100644 test/commands/support/remote_node_down_command.ex create mode 100644 test/commands/support/remote_node_down_handler.ex create mode 100644 test/commands/support/remote_node_down_router.ex create mode 100644 test/support/distributed_test_helper.ex diff --git a/guides/explanations/commands.md b/guides/explanations/commands.md index cf76fbf0..434bc3c9 100644 --- a/guides/explanations/commands.md +++ b/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: diff --git a/lib/application.ex b/lib/application.ex index 39fcbea5..cfd94242 100644 --- a/lib/application.ex +++ b/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{}}`. diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index 2001aac2..31686ef1 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/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. diff --git a/lib/commanded/aggregates/execution_context.ex b/lib/commanded/aggregates/execution_context.ex index 487e0a87..25f9a3a3 100644 --- a/lib/commanded/aggregates/execution_context.ex +++ b/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. diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex index 4169d0b2..c3d3b24c 100644 --- a/lib/commanded/commands/dispatcher.ex +++ b/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 diff --git a/test/application/telemetry_test.exs b/test/application/telemetry_test.exs index e325aab5..563ad05c 100644 --- a/test/application/telemetry_test.exs +++ b/test/application/telemetry_test.exs @@ -1,6 +1,8 @@ defmodule Commanded.Application.TelemetryTest do use ExUnit.Case + alias Commanded.Aggregates.{Lifespan, LifespanAggregate} + alias Commanded.Aggregates.LifespanAggregate.Command, as: LifespanCommand alias Commanded.DefaultApp alias Commanded.Middleware.Commands.Fail alias Commanded.Middleware.Commands.IncrementCount @@ -31,6 +33,19 @@ defmodule Commanded.Application.TelemetryTest do identity: :aggregate_uuid end + defmodule RetryExhaustionRouter do + use Commanded.Commands.Router + + alias Commanded.Aggregates.Lifespan + alias Commanded.Aggregates.LifespanAggregate + alias Commanded.Aggregates.LifespanAggregate.Command + + dispatch [Command], + to: LifespanAggregate, + identity: :uuid, + lifespan: Lifespan + end + test "emit `[:commanded, :application, :dispatch, :start | :stop]` event" do command = %IncrementCount{aggregate_uuid: UUID.uuid4()} @@ -73,6 +88,46 @@ defmodule Commanded.Application.TelemetryTest do meta end + test "emit dispatch stop telemetry when dispatcher retries are exhausted" do + aggregate_uuid = UUID.uuid4() + + {:ok, ^aggregate_uuid} = + Commanded.Aggregates.Supervisor.open_aggregate( + DefaultApp, + LifespanAggregate, + aggregate_uuid + ) + + command = %LifespanCommand{uuid: aggregate_uuid, action: :noop, lifespan: :stop} + + results = + dispatch_concurrently(fn -> + RetryExhaustionRouter.dispatch(command, application: DefaultApp, retry_attempts: 0) + end) + + assert Enum.any?(results, &match?({:ok, {:error, :too_many_attempts}}, &1)) + + assert Enum.all?(results, fn + {:ok, :ok} -> true + {:ok, {:error, :too_many_attempts}} -> true + _ -> false + end) + + events = collect_dispatch_events(length(results)) + + starts = + Enum.count(events, &match?({[:commanded, :application, :dispatch, :start], _, _, _}, &1)) + + stop_errors = + for {[:commanded, :application, :dispatch, :stop], _, _, %{error: error}} <- events do + error + end + + assert starts == length(results) + assert length(stop_errors) == length(results) + assert :too_many_attempts in stop_errors + end + defp attach_telemetry do agent = start_supervised!({Agent, fn -> 1 end}) @@ -96,4 +151,32 @@ defmodule Commanded.Application.TelemetryTest do :telemetry.detach("test-handler") end) end + + defp collect_dispatch_events(expected_stops, events \\ []) do + stop_count = + Enum.count(events, fn + {[:commanded, :application, :dispatch, :stop], _, _, _} -> true + _ -> false + end) + + if stop_count == expected_stops do + Enum.reverse(events) + else + assert_receive event, 1_000 + + case event do + {[:commanded, :application, :dispatch, _], _, _, _} -> + collect_dispatch_events(expected_stops, [event | events]) + + _ -> + collect_dispatch_events(expected_stops, events) + end + end + end + + defp dispatch_concurrently(fun, count \\ 10) do + 1..count + |> Task.async_stream(fn _ -> fun.() end, ordered: false, timeout: 5_000) + |> Enum.to_list() + end end diff --git a/test/commands/distributed_dispatch_test.exs b/test/commands/distributed_dispatch_test.exs new file mode 100644 index 00000000..e853d799 --- /dev/null +++ b/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 diff --git a/test/commands/support/remote_node_down_aggregate.ex b/test/commands/support/remote_node_down_aggregate.ex new file mode 100644 index 00000000..852749cd --- /dev/null +++ b/test/commands/support/remote_node_down_aggregate.ex @@ -0,0 +1,4 @@ +defmodule Commanded.Commands.RemoteNodeDownAggregate do + @moduledoc false + defstruct [] +end diff --git a/test/commands/support/remote_node_down_command.ex b/test/commands/support/remote_node_down_command.ex new file mode 100644 index 00000000..dd9c10cd --- /dev/null +++ b/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 diff --git a/test/commands/support/remote_node_down_handler.ex b/test/commands/support/remote_node_down_handler.ex new file mode 100644 index 00000000..a8fe3ae1 --- /dev/null +++ b/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 diff --git a/test/commands/support/remote_node_down_router.ex b/test/commands/support/remote_node_down_router.ex new file mode 100644 index 00000000..5ea7e800 --- /dev/null +++ b/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 diff --git a/test/middleware/middleware_test.exs b/test/middleware/middleware_test.exs index d2858c09..cab68906 100644 --- a/test/middleware/middleware_test.exs +++ b/test/middleware/middleware_test.exs @@ -4,6 +4,8 @@ defmodule Commanded.Middleware.MiddlewareTest do import Commanded.Enumerable alias Commanded.Commands.ExecutionResult + alias Commanded.Aggregates.{Lifespan, LifespanAggregate} + alias Commanded.Aggregates.LifespanAggregate.Command, as: LifespanCommand alias Commanded.DefaultApp alias Commanded.Helpers.CommandAuditMiddleware @@ -66,6 +68,21 @@ defmodule Commanded.Middleware.MiddlewareTest do identity: :aggregate_uuid end + defmodule RetryExhaustionRouter do + use Commanded.Commands.Router + + alias Commanded.Aggregates.Lifespan + alias Commanded.Aggregates.LifespanAggregate + alias Commanded.Aggregates.LifespanAggregate.Command + + middleware CommandAuditMiddleware + + dispatch [Command], + to: LifespanAggregate, + identity: :uuid, + lifespan: Lifespan + end + setup do start_supervised!(CommandAuditMiddleware) start_supervised!(DefaultApp) @@ -163,4 +180,42 @@ defmodule Commanded.Middleware.MiddlewareTest do "updated_by" => "ModifyMetadataMiddleware" } end + + test "should execute middleware failure callback when dispatcher retries are exhausted" do + aggregate_uuid = UUID.uuid4() + + {:ok, ^aggregate_uuid} = + Commanded.Aggregates.Supervisor.open_aggregate( + DefaultApp, + LifespanAggregate, + aggregate_uuid + ) + + command = %LifespanCommand{uuid: aggregate_uuid, action: :noop, lifespan: :stop} + + results = + dispatch_concurrently(fn -> + RetryExhaustionRouter.dispatch(command, application: DefaultApp, retry_attempts: 0) + end) + + assert Enum.any?(results, &match?({:ok, {:error, :too_many_attempts}}, &1)) + + assert Enum.all?(results, fn + {:ok, :ok} -> true + {:ok, {:error, :too_many_attempts}} -> true + _ -> false + end) + + error_count = Enum.count(results, &match?({:ok, {:error, :too_many_attempts}}, &1)) + success_count = Enum.count(results, &match?({:ok, :ok}, &1)) + + assert CommandAuditMiddleware.count_commands() == + {length(results), success_count, error_count} + end + + defp dispatch_concurrently(fun, count \\ 10) do + 1..count + |> Task.async_stream(fn _ -> fun.() end, ordered: false, timeout: 5_000) + |> Enum.to_list() + end end diff --git a/test/support/distributed_test_helper.ex b/test/support/distributed_test_helper.ex new file mode 100644 index 00000000..be5b5ed1 --- /dev/null +++ b/test/support/distributed_test_helper.ex @@ -0,0 +1,9 @@ +defmodule Commanded.DistributedTestHelper do + alias Commanded.DistributedApp + + def start_distributed_app(reply_to) do + {:ok, _pid} = DistributedApp.start_link() + send(reply_to, {:distributed_app_started, node()}) + :timer.sleep(:infinity) + end +end From ce516ca65be1b75257de1b39730e4cc0768bc80d Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 28 Apr 2026 21:42:57 -0400 Subject: [PATCH 2/3] test(dispatcher): lock down retry guarantees Signed-off-by: Yordis Prieto --- .../support/aggregate_telemetry_support.ex | 36 +++++ test/application/telemetry_test.exs | 86 +++-------- test/middleware/middleware_test.exs | 49 ++---- test/opentelemetry/aggregate_test.exs | 33 ++-- test/opentelemetry/dispatcher_retry_test.exs | 143 ++++++++++++++++++ test/support/retry_stop_once_aggregate.ex | 41 +++++ 6 files changed, 272 insertions(+), 116 deletions(-) create mode 100644 test/aggregates/support/aggregate_telemetry_support.ex create mode 100644 test/opentelemetry/dispatcher_retry_test.exs create mode 100644 test/support/retry_stop_once_aggregate.ex diff --git a/test/aggregates/support/aggregate_telemetry_support.ex b/test/aggregates/support/aggregate_telemetry_support.ex new file mode 100644 index 00000000..39833ac2 --- /dev/null +++ b/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 diff --git a/test/application/telemetry_test.exs b/test/application/telemetry_test.exs index 563ad05c..fd98f779 100644 --- a/test/application/telemetry_test.exs +++ b/test/application/telemetry_test.exs @@ -1,15 +1,17 @@ defmodule Commanded.Application.TelemetryTest do use ExUnit.Case - alias Commanded.Aggregates.{Lifespan, LifespanAggregate} - alias Commanded.Aggregates.LifespanAggregate.Command, as: LifespanCommand 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() @@ -36,14 +38,13 @@ defmodule Commanded.Application.TelemetryTest do defmodule RetryExhaustionRouter do use Commanded.Commands.Router - alias Commanded.Aggregates.Lifespan - alias Commanded.Aggregates.LifespanAggregate - alias Commanded.Aggregates.LifespanAggregate.Command + alias Commanded.TestSupport.RetryStopOnceAggregate + alias Commanded.TestSupport.RetryStopOnceAggregate.Command dispatch [Command], - to: LifespanAggregate, + to: RetryStopOnceAggregate, identity: :uuid, - lifespan: Lifespan + before_execute: :before_execute end test "emit `[:commanded, :application, :dispatch, :start | :stop]` event" do @@ -91,41 +92,30 @@ defmodule Commanded.Application.TelemetryTest do test "emit dispatch stop telemetry when dispatcher retries are exhausted" do aggregate_uuid = UUID.uuid4() - {:ok, ^aggregate_uuid} = - Commanded.Aggregates.Supervisor.open_aggregate( - DefaultApp, - LifespanAggregate, - aggregate_uuid - ) + command = %RetryStopOnceCommand{uuid: aggregate_uuid} - command = %LifespanCommand{uuid: aggregate_uuid, action: :noop, lifespan: :stop} + assert {:error, :too_many_attempts} = + RetryExhaustionRouter.dispatch(command, application: DefaultApp, retry_attempts: 0) - results = - dispatch_concurrently(fn -> - RetryExhaustionRouter.dispatch(command, application: DefaultApp, retry_attempts: 0) - end) + assert_receive {[:commanded, :application, :dispatch, :start], _, _, _} - assert Enum.any?(results, &match?({:ok, {:error, :too_many_attempts}}, &1)) + assert_receive {[:commanded, :application, :dispatch, :stop], _, _, + %{error: :too_many_attempts}} + end - assert Enum.all?(results, fn - {:ok, :ok} -> true - {:ok, {:error, :too_many_attempts}} -> true - _ -> false - 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} - events = collect_dispatch_events(length(results)) + assert {:error, error} = TimeoutRouter.dispatch(command, application: DefaultApp) + assert error in [:aggregate_execution_failed, :aggregate_execution_timeout] - starts = - Enum.count(events, &match?({[:commanded, :application, :dispatch, :start], _, _, _}, &1)) + assert_receive {[:commanded, :application, :dispatch, :start], _, _, _} + assert_receive {[:commanded, :aggregate, :execute, :start], _, _, _} - stop_errors = - for {[:commanded, :application, :dispatch, :stop], _, _, %{error: error}} <- events do - error - end + refute_receive {[:commanded, :aggregate, :execute, :start], _, _, _}, 200 - assert starts == length(results) - assert length(stop_errors) == length(results) - assert :too_many_attempts in stop_errors + assert_receive {[:commanded, :application, :dispatch, :stop], _, _, + %{error: ^error}} end defp attach_telemetry do @@ -151,32 +141,4 @@ defmodule Commanded.Application.TelemetryTest do :telemetry.detach("test-handler") end) end - - defp collect_dispatch_events(expected_stops, events \\ []) do - stop_count = - Enum.count(events, fn - {[:commanded, :application, :dispatch, :stop], _, _, _} -> true - _ -> false - end) - - if stop_count == expected_stops do - Enum.reverse(events) - else - assert_receive event, 1_000 - - case event do - {[:commanded, :application, :dispatch, _], _, _, _} -> - collect_dispatch_events(expected_stops, [event | events]) - - _ -> - collect_dispatch_events(expected_stops, events) - end - end - end - - defp dispatch_concurrently(fun, count \\ 10) do - 1..count - |> Task.async_stream(fn _ -> fun.() end, ordered: false, timeout: 5_000) - |> Enum.to_list() - end end diff --git a/test/middleware/middleware_test.exs b/test/middleware/middleware_test.exs index cab68906..72b0ea1e 100644 --- a/test/middleware/middleware_test.exs +++ b/test/middleware/middleware_test.exs @@ -4,8 +4,6 @@ defmodule Commanded.Middleware.MiddlewareTest do import Commanded.Enumerable alias Commanded.Commands.ExecutionResult - alias Commanded.Aggregates.{Lifespan, LifespanAggregate} - alias Commanded.Aggregates.LifespanAggregate.Command, as: LifespanCommand alias Commanded.DefaultApp alias Commanded.Helpers.CommandAuditMiddleware @@ -19,6 +17,8 @@ defmodule Commanded.Middleware.MiddlewareTest do } alias Commanded.Middleware.Pipeline + alias Commanded.TestSupport.RetryStopOnceAggregate + alias Commanded.TestSupport.RetryStopOnceAggregate.Command, as: RetryStopOnceCommand alias Commanded.UUID defmodule FirstMiddleware do @@ -71,20 +71,20 @@ defmodule Commanded.Middleware.MiddlewareTest do defmodule RetryExhaustionRouter do use Commanded.Commands.Router - alias Commanded.Aggregates.Lifespan - alias Commanded.Aggregates.LifespanAggregate - alias Commanded.Aggregates.LifespanAggregate.Command + alias Commanded.TestSupport.RetryStopOnceAggregate + alias Commanded.TestSupport.RetryStopOnceAggregate.Command middleware CommandAuditMiddleware dispatch [Command], - to: LifespanAggregate, + to: RetryStopOnceAggregate, identity: :uuid, - lifespan: Lifespan + before_execute: :before_execute end setup do start_supervised!(CommandAuditMiddleware) + start_supervised!(RetryStopOnceAggregate.Tracker) start_supervised!(DefaultApp) :ok @@ -184,38 +184,11 @@ defmodule Commanded.Middleware.MiddlewareTest do test "should execute middleware failure callback when dispatcher retries are exhausted" do aggregate_uuid = UUID.uuid4() - {:ok, ^aggregate_uuid} = - Commanded.Aggregates.Supervisor.open_aggregate( - DefaultApp, - LifespanAggregate, - aggregate_uuid - ) - - command = %LifespanCommand{uuid: aggregate_uuid, action: :noop, lifespan: :stop} - - results = - dispatch_concurrently(fn -> - RetryExhaustionRouter.dispatch(command, application: DefaultApp, retry_attempts: 0) - end) - - assert Enum.any?(results, &match?({:ok, {:error, :too_many_attempts}}, &1)) + command = %RetryStopOnceCommand{uuid: aggregate_uuid} - assert Enum.all?(results, fn - {:ok, :ok} -> true - {:ok, {:error, :too_many_attempts}} -> true - _ -> false - end) - - error_count = Enum.count(results, &match?({:ok, {:error, :too_many_attempts}}, &1)) - success_count = Enum.count(results, &match?({:ok, :ok}, &1)) - - assert CommandAuditMiddleware.count_commands() == - {length(results), success_count, error_count} - end + assert {:error, :too_many_attempts} = + RetryExhaustionRouter.dispatch(command, application: DefaultApp, retry_attempts: 0) - defp dispatch_concurrently(fun, count \\ 10) do - 1..count - |> Task.async_stream(fn _ -> fun.() end, ordered: false, timeout: 5_000) - |> Enum.to_list() + assert CommandAuditMiddleware.count_commands() == {1, 0, 1} end end diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs index e4601606..ae5c4b5a 100644 --- a/test/opentelemetry/aggregate_test.exs +++ b/test/opentelemetry/aggregate_test.exs @@ -2,7 +2,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do use Commanded.OpenTelemetryCase, async: false use Commanded.MockEventStoreCase - alias Commanded.Aggregates.AggregateTelemetryTest + alias Commanded.Aggregates.AggregateTelemetrySupport alias Commanded.Aggregates.{Aggregate, ExecutionContext} alias Commanded.DefaultApp alias Commanded.Middleware.Commands.IncrementCount @@ -678,15 +678,15 @@ defmodule Commanded.OpenTelemetry.AggregateTest do assert {:error, :too_many_attempts} = Aggregate.execute( MockedApp, - AggregateTelemetryTest.ExampleAggregate, + AggregateTelemetrySupport.ExampleAggregate, aggregate_uuid, %ExecutionContext{ command: - struct!(Commanded.Aggregates.AggregateTelemetryTest.Commands.Ok, + struct!(Commanded.Aggregates.AggregateTelemetrySupport.Commands.Ok, message: "ok" ), function: :execute, - handler: AggregateTelemetryTest.ExampleAggregate, + handler: AggregateTelemetrySupport.ExampleAggregate, retry_attempts: 0 } ) @@ -694,7 +694,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do assert_receive {:span, span( name: - "execute Commanded.Aggregates.AggregateTelemetryTest.ExampleAggregate", + "execute Commanded.Aggregates.AggregateTelemetrySupport.ExampleAggregate", kind: :consumer, events: events )}, @@ -727,8 +727,8 @@ defmodule Commanded.OpenTelemetry.AggregateTest do build_recorded_event( aggregate_uuid, 1, - struct!(Commanded.Aggregates.AggregateTelemetryTest.Event, message: "event"), - event_type: "Elixir.Commanded.Aggregates.AggregateTelemetryTest.Event" + struct!(Commanded.Aggregates.AggregateTelemetrySupport.Event, message: "event"), + event_type: "Elixir.Commanded.Aggregates.AggregateTelemetrySupport.Event" ) expect_wrong_expected_version_retry_succeeds(aggregate_uuid, event) @@ -738,15 +738,15 @@ defmodule Commanded.OpenTelemetry.AggregateTest do assert {:ok, 2, _events, _aggregate_state} = Aggregate.execute( MockedApp, - AggregateTelemetryTest.ExampleAggregate, + AggregateTelemetrySupport.ExampleAggregate, aggregate_uuid, %ExecutionContext{ command: - struct!(Commanded.Aggregates.AggregateTelemetryTest.Commands.Ok, + struct!(Commanded.Aggregates.AggregateTelemetrySupport.Commands.Ok, message: "ok" ), function: :execute, - handler: AggregateTelemetryTest.ExampleAggregate, + handler: AggregateTelemetrySupport.ExampleAggregate, retry_attempts: 1 } ) @@ -754,7 +754,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do assert_receive {:span, span( name: - "execute Commanded.Aggregates.AggregateTelemetryTest.ExampleAggregate", + "execute Commanded.Aggregates.AggregateTelemetrySupport.ExampleAggregate", kind: :consumer, events: events )}, @@ -776,6 +776,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do {:attributes, _, _, _, attrs_map} = attrs_tuple assert attrs_map[:"commanded.wrong_expected_version.count"] == 1 + refute_receive {:span, _}, 200 end test "no wrong_expected_version event when count is 0" do @@ -788,22 +789,22 @@ defmodule Commanded.OpenTelemetry.AggregateTest do assert {:ok, 1, _events, _aggregate_state} = Aggregate.execute( MockedApp, - AggregateTelemetryTest.ExampleAggregate, + AggregateTelemetrySupport.ExampleAggregate, aggregate_uuid, %ExecutionContext{ command: - struct!(Commanded.Aggregates.AggregateTelemetryTest.Commands.Ok, + struct!(Commanded.Aggregates.AggregateTelemetrySupport.Commands.Ok, message: "ok" ), function: :execute, - handler: AggregateTelemetryTest.ExampleAggregate + handler: AggregateTelemetrySupport.ExampleAggregate } ) assert_receive {:span, span( name: - "execute Commanded.Aggregates.AggregateTelemetryTest.ExampleAggregate", + "execute Commanded.Aggregates.AggregateTelemetrySupport.ExampleAggregate", kind: :consumer, events: events )}, @@ -822,7 +823,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do end defp start_aggregate(aggregate_uuid, opts) do - aggregate = AggregateTelemetryTest.ExampleAggregate + aggregate = AggregateTelemetrySupport.ExampleAggregate app = Keyword.fetch!(opts, :application) name = Aggregate.name(app, aggregate, aggregate_uuid) diff --git a/test/opentelemetry/dispatcher_retry_test.exs b/test/opentelemetry/dispatcher_retry_test.exs new file mode 100644 index 00000000..7befa98a --- /dev/null +++ b/test/opentelemetry/dispatcher_retry_test.exs @@ -0,0 +1,143 @@ +defmodule Commanded.OpenTelemetry.DispatcherRetryTest do + use Commanded.OpenTelemetryCase, async: false + + alias Commanded.OpenTelemetry.Aggregate, as: OTelAggregate + alias Commanded.OpenTelemetry.Application, as: OTelApplication + alias Commanded.TestSupport.RetryStopOnceAggregate + alias Commanded.TestSupport.RetryStopOnceAggregate.Command, as: RetryStopOnceCommand + alias Commanded.UUID + + defmodule RetryRouter do + use Commanded.Commands.Router + + alias Commanded.TestSupport.RetryStopOnceAggregate + alias Commanded.TestSupport.RetryStopOnceAggregate.Command + + middleware Commanded.Middleware.TraceContextPropagator + + dispatch [Command], + to: RetryStopOnceAggregate, + identity: :uuid, + before_execute: :before_execute + end + + defmodule App do + use Commanded.Application, + otp_app: :commanded, + event_store: [ + adapter: Commanded.EventStore.Adapters.InMemory, + serializer: Commanded.Serialization.JsonSerializer + ], + pubsub: :local, + registry: :local + + router(RetryRouter) + end + + setup do + start_supervised!(RetryStopOnceAggregate.Tracker) + start_supervised!(App) + + detach_all_handlers() + OTelApplication.setup() + OTelAggregate.setup() + + attach_attempt_spy() + + :ok + end + + test "aggregate-stop retries emit telemetry per attempt but export spans only for completed attempts" do + aggregate_uuid = UUID.uuid4() + command = %RetryStopOnceCommand{uuid: aggregate_uuid} + + assert :ok = App.dispatch(command, retry_attempts: 1) + + assert_receive {:aggregate_execute_start, ^aggregate_uuid}, 1_000 + assert_receive {:aggregate_execute_start, ^aggregate_uuid}, 1_000 + assert_receive {:aggregate_execute_stop, ^aggregate_uuid}, 1_000 + + refute_receive {:aggregate_execute_stop, ^aggregate_uuid}, 200 + refute_receive {:aggregate_execute_start, ^aggregate_uuid}, 200 + + spans = collect_spans(2) + + dispatch_spans = Enum.filter(spans, &String.starts_with?(span(&1, :name), "dispatch ")) + execute_spans = Enum.filter(spans, &String.starts_with?(span(&1, :name), "execute ")) + + assert length(dispatch_spans) == 1 + assert length(execute_spans) == 1 + + [dispatch_span] = dispatch_spans + [execute_span] = execute_spans + + assert span(execute_span, :trace_id) == span(dispatch_span, :trace_id) + assert span(execute_span, :parent_span_id) == span(dispatch_span, :span_id) + end + + defp attach_attempt_spy do + test_pid = self() + + :telemetry.attach_many( + {__MODULE__, :attempt_spy}, + [ + [:commanded, :aggregate, :execute, :start], + [:commanded, :aggregate, :execute, :stop] + ], + fn event, _measurements, meta, _config -> + case event do + [:commanded, :aggregate, :execute, :start] -> + send(test_pid, {:aggregate_execute_start, meta.aggregate_uuid}) + + [:commanded, :aggregate, :execute, :stop] -> + send(test_pid, {:aggregate_execute_stop, meta.aggregate_uuid}) + end + end, + nil + ) + + on_exit(fn -> + :telemetry.detach({__MODULE__, :attempt_spy}) + end) + end + + defp collect_spans(expected_count, timeout_ms \\ 1_500) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + collect_spans([], expected_count, deadline) + end + + defp collect_spans(spans, expected_count, _deadline) when length(spans) >= expected_count do + Enum.reverse(spans) + end + + defp collect_spans(spans, expected_count, deadline) do + remaining = max(deadline - System.monotonic_time(:millisecond), 0) + + receive do + {:span, span} -> + collect_spans([span | spans], expected_count, deadline) + after + remaining -> + flunk( + "expected #{expected_count} spans, got #{length(spans)}: " <> + inspect(Enum.map(spans, &span(&1, :name))) + ) + end + end + + defp detach_all_handlers do + commanded_events = [ + [:commanded, :application, :dispatch, :start], + [:commanded, :application, :dispatch, :stop], + [:commanded, :application, :dispatch, :exception], + [:commanded, :aggregate, :execute, :start], + [:commanded, :aggregate, :execute, :stop], + [:commanded, :aggregate, :execute, :exception] + ] + + for event <- commanded_events, + handler <- :telemetry.list_handlers(event) do + :telemetry.detach(handler.id) + end + end +end diff --git a/test/support/retry_stop_once_aggregate.ex b/test/support/retry_stop_once_aggregate.ex new file mode 100644 index 00000000..378aee0c --- /dev/null +++ b/test/support/retry_stop_once_aggregate.ex @@ -0,0 +1,41 @@ +defmodule Commanded.TestSupport.RetryStopOnceAggregate do + alias Commanded.Aggregates.ExecutionContext + + @derive Jason.Encoder + defstruct [] + + defmodule Command do + @derive Jason.Encoder + defstruct [:uuid] + end + + defmodule Tracker do + use Agent + + def start_link(_opts) do + Agent.start_link(fn -> MapSet.new() end, name: __MODULE__) + end + + def stop_once(uuid) do + Agent.get_and_update(__MODULE__, fn seen -> + if MapSet.member?(seen, uuid) do + {:ok, seen} + else + {:stop, MapSet.put(seen, uuid)} + end + end) + end + end + + def before_execute(_aggregate_state, %ExecutionContext{command: %Command{uuid: uuid}}) do + case Tracker.stop_once(uuid) do + :stop -> + Process.exit(self(), :normal) + + :ok -> + :ok + end + end + + def execute(%__MODULE__{}, %Command{}), do: [] +end From 106ffa0b766cbca52f31e7a9902862cf1d20b7d3 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 28 Apr 2026 21:52:40 -0400 Subject: [PATCH 3/3] docs(commands): clarify timeout description Signed-off-by: Yordis Prieto --- guides/explanations/commands.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/guides/explanations/commands.md b/guides/explanations/commands.md index 434bc3c9..500453f2 100644 --- a/guides/explanations/commands.md +++ b/guides/explanations/commands.md @@ -191,7 +191,7 @@ 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 dispatch fails with an error tuple. +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.