feat(telemetry): keep returned command errors out of failed spans

guides/howtos/setting-up-opentelemetry-tracing.md

@@ -61,3 +61,37 @@ Note: `setup/1` should only be called once during application startup.
 Commanded.OpenTelemetry.setup(event_handler: :disabled)
 ```
 
+## Override Returned Error Status
+
+You can override the span status used for returned `:stop` errors on application
+dispatch and aggregate execution spans:
+
+```elixir
+Commanded.OpenTelemetry.setup(
+  application: [
+    error_status: fn
+      _event_name, _measurements, %{error: :validation_failed}, _config -> :unset
+      _event_name, _measurements, _meta, _config -> :error
+    end
+  ],
+  aggregate: [
+    error_status: fn
+      _event_name, _measurements, %{error: :validation_failed}, _config -> :unset
+      _event_name, _measurements, _meta, _config -> :error
+    end
+  ]
+)
+```
+
+The callback receives the telemetry event name first, then the stop measurements,
+then the full telemetry metadata map, and finally the instrumentation config. It
+may return:
+
+- `:unset`, `:ok`, or `:error`
+- `nil` to leave the status unset
+
+When the callback returns `:error`, Commanded derives the status description from
+the returned error using `Commanded.OpenTelemetry.Helpers.format_error/1`.
+
+Exceptions are not routed through this callback; they continue to record
+exception events and set span status to error.

lib/commanded/opentelemetry.ex

@@ -55,9 +55,36 @@ defmodule Commanded.OpenTelemetry do
   """
   @type span_relationship :: :link | :child | :none
 
+  @typedoc """
+  Callback used to decide the OpenTelemetry span status for returned `:stop` errors.
+
+  The callback is only invoked when Commanded emits stop metadata containing `:error`.
+  Exception telemetry continues to use OpenTelemetry exception semantics directly.
+  """
+  @type error_status_callback ::
+          (event_name :: [atom()],
+           measurements :: map(),
+           metadata :: map(),
+           config :: keyword() ->
+             OpenTelemetry.status_code() | nil)
+
+  @error_status_options [
+    error_status: [
+      type: {:fun, 4},
+      type_doc: "`t:error_status_callback/0`",
+      doc:
+        "Override the span status for returned `:stop` errors. Return `nil` to leave the status unset."
+    ]
+  ]
+
   @nimble_schema NimbleOptions.new!(
                    aggregate: [
-                     type: {:in, [:disabled, []]},
+                     type:
+                       {:or,
+                        [
+                          {:in, [:disabled]},
+                          keyword_list: @error_status_options
+                        ]},
                      default: [],
                      doc: "Aggregate tracing configuration. Use `:disabled` to disable."
                    ],
@@ -72,7 +99,12 @@ defmodule Commanded.OpenTelemetry do
                      doc: "Aggregate snapshot tracing configuration. Use `:disabled` to disable."
                    ],
                    application: [
-                     type: {:in, [:disabled, []]},
+                     type:
+                       {:or,
+                        [
+                          {:in, [:disabled]},
+                          keyword_list: @error_status_options
+                        ]},
                      default: [],
                      doc:
                        "Application dispatch tracing configuration. Use `:disabled` to disable."
@@ -123,6 +155,16 @@ defmodule Commanded.OpenTelemetry do
       # Disable event handler tracing
       Commanded.OpenTelemetry.setup(event_handler: :disabled)
 
+      # Leave returned domain errors unset for dispatch spans
+      Commanded.OpenTelemetry.setup(
+        application: [
+          error_status: fn
+            _event_name, _measurements, %{error: :validation_failed}, _config -> :unset
+            _event_name, _measurements, _meta, _config -> :error
+          end
+        ]
+      )
+
       # Disable aggregate populate tracing
       Commanded.OpenTelemetry.setup(aggregate_populate: :disabled)
 
@@ -142,7 +184,7 @@ defmodule Commanded.OpenTelemetry do
 
     case opts[:aggregate] do
       :disabled -> :ok
-      _config -> Aggregate.setup()
+      config -> Aggregate.setup(config)
     end
 
     case opts[:aggregate_populate] do
@@ -157,7 +199,7 @@ defmodule Commanded.OpenTelemetry do
 
     case opts[:application] do
       :disabled -> :ok
-      _config -> OTelApplication.setup()
+      config -> OTelApplication.setup(config)
     end
 
     case opts[:event_handler] do

lib/commanded/opentelemetry/aggregate.ex

@@ -10,7 +10,7 @@ defmodule Commanded.OpenTelemetry.Aggregate do
 
   @tracer_id __MODULE__
 
-  def setup do
+  def setup(config \\ []) do
     :ok =
       :telemetry.attach_many(
         {__MODULE__, :execute},
@@ -20,7 +20,7 @@ defmodule Commanded.OpenTelemetry.Aggregate do
           [:commanded, :aggregate, :execute, :exception]
         ],
         &__MODULE__.handle_telemetry_event/4,
-        %{}
+        config
       )
   end
 
@@ -79,10 +79,11 @@ defmodule Commanded.OpenTelemetry.Aggregate do
 
   def handle_telemetry_event(
         [:commanded, :aggregate, :execute, :stop],
-        _measurements,
+        measurements,
         meta,
-        _config
+        config
       ) do
+    event_name = [:commanded, :aggregate, :execute, :stop]
     ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta)
 
     events = Map.get(meta, :events, [])
@@ -95,7 +96,7 @@ defmodule Commanded.OpenTelemetry.Aggregate do
         Helpers.to_error_type(error, @tracer_id)
       )
 
-      Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error)))
+      Helpers.set_error_status(ctx, error, event_name, measurements, meta, config, @tracer_id)
     end
 
     wrong_expected_version_count = Map.get(meta, :wrong_expected_version_count, 0)

lib/commanded/opentelemetry/application.ex

@@ -10,7 +10,7 @@ defmodule Commanded.OpenTelemetry.Application do
 
   @tracer_id __MODULE__
 
-  def setup do
+  def setup(config \\ []) do
     :ok =
       :telemetry.attach_many(
         {__MODULE__, :dispatch},
@@ -20,7 +20,7 @@ defmodule Commanded.OpenTelemetry.Application do
           [:commanded, :application, :dispatch, :exception]
         ],
         &__MODULE__.handle_telemetry_event/4,
-        %{}
+        config
       )
   end
 
@@ -69,10 +69,11 @@ defmodule Commanded.OpenTelemetry.Application do
 
   def handle_telemetry_event(
         [:commanded, :application, :dispatch, :stop],
-        _measurements,
+        measurements,
         meta,
-        _config
+        config
       ) do
+    event_name = [:commanded, :application, :dispatch, :stop]
     ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta)
 
     if error = meta[:error] do
@@ -82,7 +83,7 @@ defmodule Commanded.OpenTelemetry.Application do
         Helpers.to_error_type(error, @tracer_id)
       )
 
-      Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error)))
+      Helpers.set_error_status(ctx, error, event_name, measurements, meta, config, @tracer_id)
     end
 
     OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta)

lib/commanded/opentelemetry/helpers.ex

@@ -1,6 +1,8 @@
 defmodule Commanded.OpenTelemetry.Helpers do
   @moduledoc false
 
+  alias OpenTelemetry.Span
+
   def extract_propagated_ctx(nil), do: {[], :undefined}
 
   def extract_propagated_ctx(metadata) when is_map(metadata) do
@@ -65,6 +67,41 @@ defmodule Commanded.OpenTelemetry.Helpers do
   def format_error(error) when is_binary(error), do: error
   def format_error(error), do: inspect(error)
 
+  def set_error_status(ctx, error, event_name, measurements, meta, config, tracer_id) do
+    status_code =
+      case Keyword.get(config, :error_status) do
+        nil -> :error
+        fun when is_function(fun, 4) -> fun.(event_name, measurements, meta, config)
+      end
+
+    apply_error_status(ctx, status_code, error, tracer_id)
+  end
+
+  defp apply_error_status(_ctx, nil, _error, _tracer_id), do: :ok
+
+  defp apply_error_status(ctx, :error, error, _tracer_id) do
+    Span.set_status(ctx, OpenTelemetry.status(:error, format_error(error)))
+  end
+
+  defp apply_error_status(ctx, code, _error, _tracer_id) when code in [:unset, :ok] do
+    Span.set_status(ctx, OpenTelemetry.status(code))
+  end
+
+  defp apply_error_status(ctx, status_code, error, tracer_id) do
+    :telemetry.execute(
+      [:commanded, :opentelemetry, :warning],
+      %{count: 1},
+      %{
+        message: "Unknown error status encountered, falling back to error status",
+        error: error,
+        error_status: status_code,
+        tracer_id: tracer_id
+      }
+    )
+
+    Span.set_status(ctx, OpenTelemetry.status(:error, format_error(error)))
+  end
+
   def module_name(nil), do: nil
   def module_name(module) when is_atom(module), do: inspect(module)
   def module_name(_), do: nil

test/opentelemetry/aggregate_test.exs

@@ -324,6 +324,64 @@ defmodule Commanded.OpenTelemetry.AggregateTest do
              }
     end
 
+    test "allows returned aggregate error status to stay unset" do
+      detach_handlers()
+
+      OTelAggregate.setup(
+        error_status: fn _event_name, _measurements, _meta, _config -> :unset end
+      )
+
+      aggregate_uuid = UUID.uuid4()
+      causation_id = UUID.uuid4()
+      correlation_id = UUID.uuid4()
+
+      meta =
+        Factory.build_aggregate_execute_metadata(
+          aggregate_uuid: aggregate_uuid,
+          causation_id: causation_id,
+          correlation_id: correlation_id
+        )
+
+      :telemetry.execute([:commanded, :aggregate, :execute, :start], %{}, meta)
+
+      stop_meta = Map.put(meta, :error, :validation_failed)
+      :telemetry.execute([:commanded, :aggregate, :execute, :stop], %{duration: 1000}, stop_meta)
+
+      assert_receive {:span,
+                      span(
+                        name: "execute Commanded.TestSupport.TestDomain.Account",
+                        status: status,
+                        attributes: span_attrs
+                      )},
+                     1000
+
+      assert status == {:status, :unset, ""}
+      assert :otel_attributes.map(span_attrs)[:"error.type"] == "validation_failed"
+    end
+
+    test "uses the formatted aggregate error message when callback returns error" do
+      detach_handlers()
+
+      OTelAggregate.setup(
+        error_status: fn _event_name, _measurements, _meta, _config -> :error end
+      )
+
+      meta = Factory.build_aggregate_execute_metadata(aggregate_uuid: UUID.uuid4())
+
+      :telemetry.execute([:commanded, :aggregate, :execute, :start], %{}, meta)
+
+      :telemetry.execute(
+        [:commanded, :aggregate, :execute, :stop],
+        %{duration: 1000},
+        Map.put(meta, :error, :validation_failed)
+      )
+
+      assert_receive {:span, span(status: {:status, :error, error_message})},
+                     1000
+
+      assert error_message == ":validation_failed"
+    end
+
     test "handles ArgumentError exception" do
       # Commanded's rescue blocks always emit kind: :error
       {_event_name, _measurements, meta} =