fix(dispatcher): honor caller-supplied :causation_id on persisted events

.github/workflows/ci.yml

@@ -20,8 +20,8 @@ jobs:
     name: Conventional Commits
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4.1.3
-      - uses: webiny/action-conventional-commits@v1.3.0
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+      - uses: webiny/action-conventional-commits@8bc41ff4e7d423d56fa4905f6ff79209a78776c7 # v1.3.0
         with:
           allowed-commit-types: "feat,fix,chore"
 
@@ -48,17 +48,17 @@ jobs:
             otp: 27
 
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
 
       - name: Set up Elixir
         id: beam
-        uses: erlef/setup-beam@v1
+        uses: erlef/setup-beam@fc68ffb90438ef2936bbb3251622353b3dcb2f93 # v1.24.0
         with:
           elixir-version: ${{ matrix.elixir }}
           otp-version: ${{ matrix.otp }}
 
       - name: Restore dependencies cache
-        uses: actions/cache@v4
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
         with:
           path: deps
           key: ${{ runner.os }}-${{ steps.beam.outputs.otp-version }}-${{ steps.beam.outputs.elixir-version }}-mix-${{ hashFiles('**/mix.lock') }}
@@ -87,7 +87,7 @@ jobs:
         run: mix credo --strict
 
       - name: Retrieve Dialyzer PLT cache
-        uses: actions/cache@v4
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
         id: plt-cache
         with:
           path: priv/plts

lib/commanded/commands/dispatcher.ex

@@ -156,7 +156,12 @@ defmodule Commanded.Commands.Dispatcher do
   end
 
   defp to_execution_context(%Pipeline{} = pipeline, %Payload{} = payload) do
-    %Pipeline{command: command, command_uuid: command_uuid, metadata: metadata} = pipeline
+    %Pipeline{
+      command: command,
+      command_uuid: command_uuid,
+      causation_id: causation_id,
+      metadata: metadata
+    } = pipeline
 
     %Payload{
       correlation_id: correlation_id,
@@ -170,7 +175,7 @@ defmodule Commanded.Commands.Dispatcher do
 
     %ExecutionContext{
       command: command,
-      causation_id: command_uuid,
+      causation_id: causation_id || command_uuid,
       correlation_id: correlation_id,
       metadata: metadata,
       handler: handler_module,

lib/commanded/opentelemetry/aggregate.ex

@@ -47,7 +47,8 @@ defmodule Commanded.OpenTelemetry.Aggregate do
       {MessagingAttributes.messaging_destination_name(), handler_module_name},
       {MessagingAttributes.messaging_message_id(), context.causation_id},
       {MessagingAttributes.messaging_message_conversation_id(), context.correlation_id},
-      {MessagingAttributes.messaging_consumer_group_name(), Helpers.module_name(meta.application)},
+      {MessagingAttributes.messaging_consumer_group_name(),
+       Helpers.module_name(meta.application)},
       # OTel Code SemConv
       {CodeAttributes.code_function(), to_string(context.function)},
       {CodeAttributes.code_namespace(), handler_module_name},

lib/commanded/opentelemetry/event_handler.ex

@@ -81,7 +81,8 @@ defmodule Commanded.OpenTelemetry.EventHandler do
       {MessagingAttributes.messaging_destination_subscription_name(), meta.handler_name},
       {MessagingAttributes.messaging_message_id(), recorded_event.event_id},
       {MessagingAttributes.messaging_message_conversation_id(), recorded_event.correlation_id},
-      {MessagingAttributes.messaging_consumer_group_name(), Helpers.module_name(meta.application)},
+      {MessagingAttributes.messaging_consumer_group_name(),
+       Helpers.module_name(meta.application)},
       # OTel Code SemConv
       {CodeAttributes.code_function(), "handle"},
       {CodeAttributes.code_namespace(), handler_module_name},
@@ -184,7 +185,8 @@ defmodule Commanded.OpenTelemetry.EventHandler do
       {MessagingAttributes.messaging_operation_name(), "batch"},
       {MessagingAttributes.messaging_destination_name(), handler_module_name},
       {MessagingAttributes.messaging_destination_subscription_name(), meta.handler_name},
-      {MessagingAttributes.messaging_consumer_group_name(), Helpers.module_name(meta.application)},
+      {MessagingAttributes.messaging_consumer_group_name(),
+       Helpers.module_name(meta.application)},
       {MessagingAttributes.messaging_batch_message_count(), meta.event_count},
       # OTel Code SemConv
       {CodeAttributes.code_function(), "handle_batch"},

test/commands/correlation_causation_test.exs

@@ -38,17 +38,27 @@ defmodule Commanded.Commands.CorrelationCasuationTest do
       assert [^causation_id] = CommandAuditMiddleware.dispatched_commands(& &1.causation_id)
     end
 
-    test "should be set from `command_uuid` on created event" do
+    test "should fall back to `command_uuid` on created event when not provided" do
       command = %OpenAccount{account_number: "ACC123", initial_balance: 500}
 
       :ok = BankRouter.dispatch(command, application: BankApp)
 
       [command_uuid] = CommandAuditMiddleware.dispatched_commands(& &1.command_uuid)
       [event] = EventStore.stream_forward(BankApp, "ACC123") |> Enum.to_list()
 
-      # an event's `causation_id` is the dispatched command's `command_uuid`
       assert event.causation_id == command_uuid
     end
+
+    test "should be set from explicit `:causation_id` dispatch option on created event" do
+      causation_id = UUID.uuid4()
+      command = %OpenAccount{account_number: "ACC123", initial_balance: 500}
+
+      :ok = BankRouter.dispatch(command, application: BankApp, causation_id: causation_id)
+
+      [event] = EventStore.stream_forward(BankApp, "ACC123") |> Enum.to_list()
+
+      assert event.causation_id == causation_id
+    end
   end
 
   describe "`correlation_id`" do
@@ -131,17 +141,11 @@ defmodule Commanded.Commands.CorrelationCasuationTest do
       [account_opened, money_deposited] =
         EventStore.stream_forward(BankApp, "ACC123") |> Enum.to_list()
 
-      # should set command causation id as handled event id
       [_causation_id, causation_id] =
         CommandAuditMiddleware.dispatched_commands(& &1.causation_id)
 
       assert causation_id == account_opened.event_id
-
-      # should set created event causation id as the command uuid
-      [_command_uuid, command_uuid] =
-        CommandAuditMiddleware.dispatched_commands(& &1.command_uuid)
-
-      assert money_deposited.causation_id == command_uuid
+      assert money_deposited.causation_id == account_opened.event_id
     end
 
     def start_account_bonus_handler(_context) do