Repository: commanded/commanded
Pull request #414 (2020-11-03 23:51:26)

Define `Commanded.Telemetry` module to emit consistent telemetry events
Follow the conventions defined in the [telemetry](https://hexdocs.pm/telemetry/) library's `span/3` function for naming convention (`:start`, `:stop`, `:exception`) and the structure of measurements and metadata.


Patch:
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 6d3a6f3..f2ab836 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -7,24 +7,47 @@ defmodule Commanded.Aggregates.Aggregate do
     event: [:commanded, :aggregate, :execute, :start],
     description: "Emitted when an aggregate starts executing a command",
     measurements: "%{system_time: integer()}",
-    metadata:
-      "%{execution_context: ExecutionContext.t(), aggregate: %__MODULE__{}, caller: pid()}"
+    metadata: """
+    %{application: Commanded.Application.t(),
+      aggregate_uuid: String.t(),
+      aggregate_state: struct(),
+      aggregate_version: non_neg_integer(),
+      caller: pid(),
+      execution_context: Commanded.Aggregates.ExecutionContext.t()}
+    """
   })
 
   telemetry_event(%{
     event: [:commanded, :aggregate, :execute, :stop],
     description: "Emitted when an aggregate stops executing a command",
-    measurements: "%{duration: non_neg_integer(), num_events: non_neg_integer()}",
-    metadata:
-      "%{execution_context: ExecutionContext.t(), aggregate: %__MODULE__{}, events: [map()], caller: pid()}"
+    measurements: "%{duration: non_neg_integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      aggregate_uuid: String.t(),
+      aggregate_state: struct(),
+      aggregate_version: non_neg_integer(),
+      caller: pid(),
+      execution_context: Commanded.Aggregates.ExecutionContext.t(),
+      events: [map()],
+      error: nil | any()}
+    """
   })
 
   telemetry_event(%{
     event: [:commanded, :aggregate, :execute, :exception],
     description: "Emitted when an aggregate raises an exception",
     measurements: "%{duration: non_neg_integer()}",
-    metadata:
-      "%{execution_context: ExecutionContext.t(), aggregate: %__MODULE__{}, error: any(), caller: pid()}"
+    metadata: """
+    %{application: Commanded.Application.t(),
+      aggregate_uuid: String.t(),
+      aggregate_state: struct(),
+      aggregate_version: non_neg_integer(),
+      caller: pid(),
+      execution_context: Commanded.Aggregates.ExecutionContext.t(),
+      kind: :throw | :error | :exit,
+      reason: any(),
+      stacktrace: list()}
+    """
   })
 
   @moduledoc """
@@ -71,6 +94,7 @@ defmodule Commanded.Aggregates.Aggregate do
   ## Telemetry
 
   #{telemetry_docs()}
+
   """
 
   require Logger
@@ -84,6 +108,7 @@ defmodule Commanded.Aggregates.Aggregate do
   alias Commanded.EventStore.SnapshotData
   alias Commanded.Registration
   alias Commanded.Snapshotting
+  alias Commanded.Telemetry
 
   @read_event_batch_size 100
 
@@ -256,22 +281,13 @@ defmodule Commanded.Aggregates.Aggregate do
   def handle_call({:execute_command, %ExecutionContext{} = context}, from, %Aggregate{} = state) do
     %ExecutionContext{lifespan: lifespan, command: command} = context
 
-    start = System.monotonic_time()
+    telemetry_metadata = telemetry_metadata(context, from, state)
+    start_time = telemetry_start(telemetry_metadata)
 
-    :telemetry.execute(
-      [:commanded, :aggregate, :execute, :start],
-      %{system_time: System.system_time()},
-      %{
-        execution_context: context,
-        aggregate: state,
-        caller: from
-      }
-    )
-
-    {reply, state} = execute_command(context, state)
+    {result, state} = execute_command(context, state)
 
     lifespan_timeout =
-      case reply do
+      case result do
         {:ok, []} ->
           aggregate_lifespan_timeout(lifespan, :after_command, command)
 
@@ -280,9 +296,12 @@ defmodule Commanded.Aggregates.Aggregate do
 
         {:error, error} ->
           aggregate_lifespan_timeout(lifespan, :after_error, error)
+
+        {:error, error, _stacktrace} ->
+          aggregate_lifespan_timeout(lifespan, :after_error, error)
       end
 
-    formatted_reply = ExecutionContext.format_reply(reply, context, state)
+    formatted_reply = ExecutionContext.format_reply(result, context, state)
 
     state = %Aggregate{state | lifespan_timeout: lifespan_timeout}
 
@@ -300,23 +319,8 @@ defmodule Commanded.Aggregates.Aggregate do
         end
       end
 
-    duration = System.monotonic_time() - start
-
-    case reply do
-      {:ok, events} ->
-        :telemetry.execute(
-          [:commanded, :aggregate, :execute, :stop],
-          %{duration: duration, num_events: Enum.count(events)},
-          %{execution_context: context, aggregate: state, events: events, caller: from}
-        )
-
-      {:error, error} ->
-        :telemetry.execute(
-          [:commanded, :aggregate, :execute, :exception],
-          %{duration: duration},
-          %{execution_context: context, aggregate: state, error: error, caller: from}
-        )
-    end
+    telemetry_metadata = telemetry_metadata(context, from, state)
+    telemetry_stop(start_time, telemetry_metadata, result)
 
     response
   end
@@ -502,6 +506,7 @@ defmodule Commanded.Aggregates.Aggregate do
 
   defp before_execute_command(aggregate_state, %ExecutionContext{} = context) do
     %ExecutionContext{handler: handler, before_execute: before_execute} = context
+
     Kernel.apply(handler, before_execute, [aggregate_state, context])
   end
 
@@ -542,7 +547,7 @@ defmodule Commanded.Aggregates.Aggregate do
     error ->
       Logger.error(fn -> Exception.format(:error, error, __STACKTRACE__) end)
 
-      {{:error, error}, state}
+      {{:error, error, __STACKTRACE__}, state}
   end
 
   defp apply_and_persist_events(pending_events, context, %Aggregate{} = state) do
@@ -621,6 +626,52 @@ defmodule Commanded.Aggregates.Aggregate do
     EventStore.append_to_stream(application, aggregate_uuid, expected_version, event_data)
   end
 
+  defp telemetry_start(telemetry_metadata) do
+    Telemetry.start([:commanded, :aggregate, :execute], telemetry_metadata)
+  end
+
+  defp telemetry_stop(start_time, telemetry_metadata, result) do
+    event_prefix = [:commanded, :aggregate, :execute]
+
+    case result do
+      {:ok, events} ->
+        Telemetry.stop(event_prefix, start_time, Map.put(telemetry_metadata, :events, events))
+
+      {:error, error} ->
+        Telemetry.stop(event_prefix, start_time, Map.put(telemetry_metadata, :error, error))
+
+      {:error, error, stacktrace} ->
+        Telemetry.exception(
+          event_prefix,
+          start_time,
+          :error,
+          error,
+          stacktrace,
+          telemetry_metadata
+        )
+    end
+  end
+
+  defp telemetry_metadata(%ExecutionContext{} = context, from, %Aggregate{} = state) do
+    %Aggregate{
+      application: application,
+      aggregate_uuid: aggregate_uuid,
+      aggregate_state: aggregate_state,
+      aggregate_version: aggregate_version
+    } = state
+
+    {pid, _ref} = from
+
+    %{
+      application: application,
+      aggregate_uuid: aggregate_uuid,
+      aggregate_state: aggregate_state,
+      aggregate_version: aggregate_version,
+      caller: pid,
+      execution_context: context
+    }
+  end
+
   defp via_name(application, aggregate_module, aggregate_uuid) do
     name = name(application, aggregate_module, aggregate_uuid)
 
diff --git a/lib/commanded/aggregates/execution_context.ex b/lib/commanded/aggregates/execution_context.ex
index ebed8cd..b69a177 100644
--- a/lib/commanded/aggregates/execution_context.ex
+++ b/lib/commanded/aggregates/execution_context.ex
@@ -69,7 +69,7 @@ defmodule Commanded.Aggregates.ExecutionContext do
     {:ok, context}
   end
 
-  def format_reply(reply, %ExecutionContext{} = context, %Aggregate{} = aggregate) do
+  def format_reply(result, %ExecutionContext{} = context, %Aggregate{} = aggregate) do
     %Aggregate{
       aggregate_uuid: aggregate_uuid,
       aggregate_state: aggregate_state,
@@ -78,7 +78,7 @@ defmodule Commanded.Aggregates.ExecutionContext do
 
     %ExecutionContext{metadata: metadata, returning: returning} = context
 
-    with {:ok, events} <- reply do
+    with {:ok, events} <- result do
       case returning do
         :aggregate_state ->
           {:ok, aggregate_version, events, aggregate_state}
@@ -100,6 +100,9 @@ defmodule Commanded.Aggregates.ExecutionContext do
         false ->
           {:ok, aggregate_version, events}
       end
+    else
+      {:error, _error} = reply -> reply
+      {:error, error, _stacktrace} -> {:error, error}
     end
   end
 end
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index d58052b..d052d69 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -8,11 +8,12 @@ defmodule Commanded.Event.Handler do
     description: "Emitted when an Event.Handler.handle/2 callback is started",
     measurements: "%{system_time: integer()}",
     metadata: """
-    %{
-      recorded_event: RecordedEvent.t(),
-      handler_state: %__MODULE__{},
-      context: map() | FailureContext.t()
-    }
+    %{application: Commanded.Application.t(),
+      context: map(),
+      handler_name: String.t(),
+      handler_module: atom(),
+      handler_state: map(),
+      recorded_event: RecordedEvent.t()}
     """
   })
 
@@ -21,12 +22,13 @@ defmodule Commanded.Event.Handler do
     description: "Emitted when an Event.Handler.handle/2 callback returns",
     measurements: "%{duration: non_neg_integer()}",
     metadata: """
-    %{
+    %{:application => Commanded.Application.t(),
+      :context => map(),
+      :handler_name => String.t(),
+      :handler_module => atom(),
+      :handler_state => map(),
       :recorded_event => RecordedEvent.t(),
-      :handler_state => %__MODULE__{},
-      :context => map() | FailureContext.t(),
-      optional(:error) => term()
-    }
+      optional(:error) => any()}
     """
   })
 
@@ -35,14 +37,15 @@ defmodule Commanded.Event.Handler do
     description: "Emitted when an Event.Handler.handle/2 raises an exception",
     measurements: "%{duration: non_neg_integer()}",
     metadata: """
-    %{
+    %{application: Commanded.Application.t(),
+      context: map(),
+      handler_name: String.t(),
+      handler_module: atom(),
+      handler_state: map(),
       recorded_event: RecordedEvent.t(),
-      handler_state: %__MODULE__{},
-      context: map() | FailureContext.t(),
-      kind: :error,
-      stacktrace: list(),
-      reason: term()
-    }
+      kind: :throw | :error | :exit,
+      reason: any(),
+      stacktrace: list()}
     """
   })
 
@@ -275,6 +278,7 @@ defmodule Commanded.Event.Handler do
   ## Telemetry
 
   #{telemetry_docs()}
+
   """
 
   require Logger
@@ -285,6 +289,7 @@ defmodule Commanded.Event.Handler do
   alias Commanded.EventStore.RecordedEvent
   alias Commanded.EventStore.Subscription
   alias Commanded.Subscriptions
+  alias Commanded.Telemetry
 
   @type domain_event :: struct()
   @type metadata :: map()
@@ -780,13 +785,13 @@ defmodule Commanded.Event.Handler do
     end
   end
 
-  defp handle_event(event, handler, context \\ %{})
+  defp handle_event(event, context \\ %{}, handler)
 
   # Ignore already seen event.
   defp handle_event(
          %RecordedEvent{event_number: event_number} = event,
-         %Handler{last_seen_event: last_seen_event} = state,
-         _context
+         _context,
+         %Handler{last_seen_event: last_seen_event} = state
        )
        when not is_nil(last_seen_event) and event_number <= last_seen_event do
     Logger.debug(fn -> describe(state) <> " has already seen event ##{inspect(event_number)}" end)
@@ -795,71 +800,43 @@ defmodule Commanded.Event.Handler do
   end
 
   # Delegate event to handler module.
-  defp handle_event(%RecordedEvent{} = event, %Handler{} = state, context) do
-    start = System.monotonic_time()
-
-    :telemetry.execute(
-      [:commanded, :event, :handle, :start],
-      %{system_time: System.system_time()},
-      %{recorded_event: event, handler_state: state, context: context}
-    )
-
-    handled_event = delegate_event_to_handler(event, state)
-    measurements = %{duration: System.monotonic_time() - start}
-    metadata = %{recorded_event: event, handler_state: state, context: context}
+  defp handle_event(%RecordedEvent{} = event, context, %Handler{} = state) do
+    telemetry_metadata = telemetry_metadata(event, context, state)
+    start_time = telemetry_start(telemetry_metadata)
 
-    case handled_event do
+    case delegate_event_to_handler(event, state) do
       :ok ->
-        :telemetry.execute([:commanded, :event, :handle, :stop], measurements, metadata)
+        telemetry_stop(start_time, telemetry_metadata)
+
         confirm_receipt(event, state)
 
       {:ok, handler_state} ->
-        metadata = %{metadata | handler_state: handler_state}
-        :telemetry.execute([:commanded, :event, :handle, :stop], measurements, metadata)
+        telemetry_stop(start_time, Map.put(telemetry_metadata, :handler_state, handler_state))
+
         confirm_receipt(event, %Handler{state | handler_state: handler_state})
 
       {:error, :already_seen_event} ->
-        :telemetry.execute(
-          [:commanded, :event, :handle, :stop],
-          measurements,
-          Map.put(metadata, :error, :already_seen_event)
-        )
+        telemetry_stop(start_time, Map.put(telemetry_metadata, :error, :already_seen_event))
 
         confirm_receipt(event, state)
 
       {:error, reason} = error ->
-        :telemetry.execute(
-          [:commanded, :event, :handle, :stop],
-          measurements,
-          Map.put(metadata, :error, reason)
-        )
-
         log_event_error(error, event, state)
+        telemetry_stop(start_time, Map.put(telemetry_metadata, :error, reason))
 
         failure_context = build_failure_context(event, context, state)
 
         handle_event_error(error, event, failure_context, state)
 
       {:error, reason, stacktrace} ->
-        :telemetry.execute(
-          [:commanded, :event, :handle, :exception],
-          measurements,
-          Map.merge(metadata, %{kind: :error, reason: reason, stacktrace: stacktrace})
-        )
-
         log_event_error({:error, reason}, event, state)
+        telemetry_exception(start_time, :error, reason, stacktrace, telemetry_metadata)
 
         failure_context = build_failure_context(event, context, stacktrace, state)
 
         handle_event_error({:error, reason}, event, failure_context, state)
 
       invalid ->
-        :telemetry.execute(
-          [:commanded, :event, :handle, :stop],
-          measurements,
-          Map.put(metadata, :error, invalid)
-        )
-
         Logger.error(fn ->
           describe(state) <>
             " failed to handle event " <>
@@ -869,14 +846,12 @@ defmodule Commanded.Event.Handler do
             ", expected `:ok` or `{:error, term}`"
         end)
 
+        telemetry_stop(start_time, Map.put(telemetry_metadata, :error, :invalid_return_value))
+
+        error = {:error, :invalid_return_value}
         failure_context = build_failure_context(event, context, state)
 
-        handle_event_error(
-          {:error, :invalid_return_value},
-          event,
-          failure_context,
-          state
-        )
+        handle_event_error(error, event, failure_context, state)
     end
   end
 
@@ -947,13 +922,13 @@ defmodule Commanded.Event.Handler do
         # Retry the failed event
         Logger.info(fn -> describe(state) <> " is retrying failed event" end)
 
-        handle_event(failed_event, state, context)
+        handle_event(failed_event, context, state)
 
       {:retry, context} when is_map(context) ->
         # Retry the failed event
         Logger.info(fn -> describe(state) <> " is retrying failed event" end)
 
-        handle_event(failed_event, state, context)
+        handle_event(failed_event, context, state)
 
       {:retry, delay, %FailureContext{context: context}}
       when is_map(context) and is_integer(delay) and delay >= 0 ->
@@ -964,7 +939,7 @@ defmodule Commanded.Event.Handler do
 
         :timer.sleep(delay)
 
-        handle_event(failed_event, state, context)
+        handle_event(failed_event, context, state)
 
       {:retry, delay, context} when is_map(context) and is_integer(delay) and delay >= 0 ->
         # Retry the failed event after waiting for the given delay, in milliseconds
@@ -974,7 +949,7 @@ defmodule Commanded.Event.Handler do
 
         :timer.sleep(delay)
 
-        handle_event(failed_event, state, context)
+        handle_event(failed_event, context, state)
 
       :skip ->
         # Skip the failed event by confirming receipt
@@ -1029,6 +1004,43 @@ defmodule Commanded.Event.Handler do
     %Handler{state | last_seen_event: event_number}
   end
 
+  defp telemetry_start(telemetry_metadata) do
+    Telemetry.start([:commanded, :event, :handle], telemetry_metadata)
+  end
+
+  defp telemetry_stop(start_time, telemetry_metadata) do
+    Telemetry.stop([:commanded, :event, :handle], start_time, telemetry_metadata)
+  end
+
+  defp telemetry_exception(start_time, kind, reason, stacktrace, telemetry_metadata) do
+    Telemetry.exception(
+      [:commanded, :event, :handle],
+      start_time,
+      kind,
+      reason,
+      stacktrace,
+      telemetry_metadata
+    )
+  end
+
+  defp telemetry_metadata(recorded_event, context, %Handler{} = state) do
+    %Handler{
+      application: application,
+      handler_name: handler_name,
+      handler_module: handler_module,
+      handler_state: handler_state
+    } = state
+
+    %{
+      application: application,
+      handler_name: handler_name,
+      handler_module: handler_module,
+      handler_state: handler_state,
+      context: context,
+      recorded_event: recorded_event
+    }
+  end
+
   defp consistency(opts) do
     case opts[:consistency] || Application.get_env(:commanded, :default_consistency, :eventual) do
       consistency when consistency in [:eventual, :strong] -> consistency
diff --git a/lib/commanded/telemetry.ex b/lib/commanded/telemetry.ex
new file mode 100644
index 0000000..7726258
--- /dev/null
+++ b/lib/commanded/telemetry.ex
@@ -0,0 +1,39 @@
+defmodule Commanded.Telemetry do
+  @moduledoc false
+
+  def start(event_prefix, metadata \\ %{}, additional_measurements \\ %{}) do
+    start_time = System.monotonic_time()
+    measurements = Map.put(additional_measurements, :system_time, System.system_time())
+
+    :telemetry.execute(event_prefix ++ [:start], measurements, metadata)
+
+    start_time
+  end
+
+  def stop(event_prefix, start_time, metadata \\ %{}, additional_measurements \\ %{}) do
+    measurements = include_duration(start_time, additional_measurements)
+
+    :telemetry.execute(event_prefix ++ [:stop], measurements, metadata)
+  end
+
+  def exception(
+        event_prefix,
+        start_time,
+        kind,
+        reason,
+        stacktrace,
+        metadata \\ %{},
+        additional_measurements \\ %{}
+      ) do
+    measurements = include_duration(start_time, additional_measurements)
+    metadata = Map.merge(metadata, %{kind: kind, reason: reason, stacktrace: stacktrace})
+
+    :telemetry.execute(event_prefix ++ [:exception], measurements, metadata)
+  end
+
+  defp include_duration(start_time, measurements) do
+    end_time = System.monotonic_time()
+
+    Map.put(measurements, :duration, end_time - start_time)
+  end
+end


Test patch:
diff --git a/test/aggregates/aggregate_telemetry_test.exs b/test/aggregates/aggregate_telemetry_test.exs
new file mode 100644
index 0000000..e17e328
--- /dev/null
+++ b/test/aggregates/aggregate_telemetry_test.exs
@@ -0,0 +1,223 @@
+defmodule Commanded.Aggregates.AggregateTelemetryTest do
+  use ExUnit.Case
+
+  alias Commanded.Aggregates.Aggregate
+  alias Commanded.Aggregates.ExecutionContext
+  alias Commanded.DefaultApp
+
+  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 Commands.{Ok, Error, RaiseException}
+
+    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
+
+  alias Commands.{Ok, Error, RaiseException}
+
+  setup do
+    start_supervised!(DefaultApp)
+    attach_telemetry()
+    :ok
+  end
+
+  describe "aggregate telemetry" do
+    setup do
+      aggregate_uuid = UUID.uuid4()
+
+      {:ok, pid} = start_aggregate(aggregate_uuid)
+
+      [aggregate_uuid: aggregate_uuid, pid: pid]
+    end
+
+    test "emit `[:commanded, :aggregate, :execute, :start]` event",
+         %{aggregate_uuid: aggregate_uuid, pid: pid} do
+      context = %ExecutionContext{
+        command: %Ok{message: "ok"},
+        function: :execute,
+        handler: ExampleAggregate
+      }
+
+      self = self()
+
+      {:ok, 1, _events} = GenServer.call(pid, {:execute_command, context})
+
+      assert_receive {[:commanded, :aggregate, :execute, :start], measurements, metadata}
+
+      assert match?(%{system_time: _system_time}, measurements)
+      assert is_number(measurements.system_time)
+
+      assert match?(
+               %{
+                 aggregate_state: %ExampleAggregate{},
+                 aggregate_uuid: ^aggregate_uuid,
+                 aggregate_version: 0,
+                 application: Commanded.DefaultApp,
+                 caller: ^self,
+                 execution_context: ^context
+               },
+               metadata
+             )
+
+      assert_receive {[:commanded, :aggregate, :execute, :stop], _measurements, _metadata}
+      refute_received {[:commanded, :aggregate, :execute, :exception], _measurements, _metadata}
+    end
+
+    test "emit `[:commanded, :aggregate, :execute, :stop]` event",
+         %{aggregate_uuid: aggregate_uuid, pid: pid} do
+      context = %ExecutionContext{
+        command: %Ok{message: "ok"},
+        function: :execute,
+        handler: ExampleAggregate
+      }
+
+      self = self()
+
+      {:ok, 1, events} = GenServer.call(pid, {:execute_command, context})
+
+      assert_receive {[:commanded, :aggregate, :execute, :start], _measurements, _metadata}
+      assert_receive {[:commanded, :aggregate, :execute, :stop], measurements, metadata}
+
+      assert match?(%{duration: _duration}, measurements)
+      assert is_number(measurements.duration)
+
+      assert match?(
+               %{
+                 aggregate_state: %ExampleAggregate{message: "ok"},
+                 aggregate_uuid: ^aggregate_uuid,
+                 aggregate_version: 1,
+                 application: Commanded.DefaultApp,
+                 caller: ^self,
+                 execution_context: ^context,
+                 events: ^events
+               },
+               metadata
+             )
+
+      refute_received {[:commanded, :aggregate, :execute, :exception], _measurements, _metadata}
+    end
+
+    test "emit `[:commanded, :aggregate, :execute, :stop]` with error event",
+         %{aggregate_uuid: aggregate_uuid, pid: pid} do
+      context = %ExecutionContext{
+        command: %Error{message: "an error"},
+        function: :execute,
+        handler: ExampleAggregate
+      }
+
+      self = self()
+
+      {:error, "an error"} = GenServer.call(pid, {:execute_command, context})
+
+      assert_receive {[:commanded, :aggregate, :execute, :start], _measurements, _metadata}
+      assert_receive {[:commanded, :aggregate, :execute, :stop], measurements, metadata}
+
+      assert match?(%{duration: _duration}, measurements)
+      assert is_number(measurements.duration)
+
+      assert match?(
+               %{
+                 aggregate_state: %ExampleAggregate{},
+                 aggregate_uuid: ^aggregate_uuid,
+                 aggregate_version: 0,
+                 application: Commanded.DefaultApp,
+                 caller: ^self,
+                 execution_context: ^context,
+                 error: "an error"
+               },
+               metadata
+             )
+
+      refute_received {[:commanded, :aggregate, :execute, :exception], _measurements, _metadata}
+    end
+
+    test "emit `[:commanded, :aggregate, :execute, :exception]` event",
+         %{aggregate_uuid: aggregate_uuid, pid: pid} do
+      context = %ExecutionContext{
+        command: %RaiseException{message: "an exception"},
+        function: :execute,
+        handler: ExampleAggregate
+      }
+
+      self = self()
+
+      Process.unlink(pid)
+
+      {:error, %RuntimeError{message: "an exception"}} =
+        GenServer.call(pid, {:execute_command, context})
+
+      assert_receive {[:commanded, :aggregate, :execute, :start], _measurements, _metadata}
+      assert_receive {[:commanded, :aggregate, :execute, :exception], measurements, metadata}
+
+      assert match?(%{duration: _duration}, measurements)
+      assert is_number(measurements.duration)
+
+      assert match?(
+               %{
+                 aggregate_state: %ExampleAggregate{},
+                 aggregate_uuid: ^aggregate_uuid,
+                 aggregate_version: 0,
+                 application: Commanded.DefaultApp,
+                 caller: ^self,
+                 execution_context: ^context,
+                 kind: :error,
+                 reason: %RuntimeError{message: "an exception"},
+                 stacktrace: _stacktrace
+               },
+               metadata
+             )
+
+      refute_received {[:commanded, :aggregate, :execute, :stop], _measurements, _metadata}
+    end
+  end
+
+  def start_aggregate(aggregate_uuid) do
+    Aggregate.start_link([application: DefaultApp],
+      aggregate_module: ExampleAggregate,
+      aggregate_uuid: aggregate_uuid
+    )
+  end
+
+  defp attach_telemetry do
+    :telemetry.attach_many(
+      "test-handler",
+      [
+        [:commanded, :aggregate, :execute, :start],
+        [:commanded, :aggregate, :execute, :stop],
+        [:commanded, :aggregate, :execute, :exception]
+      ],
+      fn event_name, measurements, metadata, reply_to ->
+        send(reply_to, {event_name, measurements, metadata})
+      end,
+      self()
+    )
+
+    on_exit(fn ->
+      :telemetry.detach("test-handler")
+    end)
+  end
+end