fix: emit populate telemetry for new aggregates (stream_not_found)

guides/explanations/fork-differences.md

@@ -218,11 +218,11 @@ end
 With a dedicated protocol, the API response format and the event store stream ID format are properly separated and can evolve independently.
 
 ### **OpenTelemetry Integration**
-PRs: [#37](https://github.com/straw-hat-team/commanded/pull/37), [#41](https://github.com/straw-hat-team/commanded/pull/41), [#45](https://github.com/straw-hat-team/commanded/pull/45), [#46](https://github.com/straw-hat-team/commanded/pull/46), [#47](https://github.com/straw-hat-team/commanded/pull/47)
+PRs: [#37](https://github.com/straw-hat-team/commanded/pull/37), [#41](https://github.com/straw-hat-team/commanded/pull/41), [#45](https://github.com/straw-hat-team/commanded/pull/45), [#46](https://github.com/straw-hat-team/commanded/pull/46), [#47](https://github.com/straw-hat-team/commanded/pull/47), [#58](https://github.com/straw-hat-team/commanded/pull/58)
 
 **Changes:**
 - Added `Commanded.OpenTelemetry` module for distributed tracing
-- Creates spans for event handlers (PR #41), EventStore operations (PR #37), aggregate execution (PR #45), application dispatch (PR #46), and aggregate populate (PR #47)
+- Creates spans for event handlers (PR #41), EventStore operations (PR #37), aggregate execution (PR #45), application dispatch (PR #46), aggregate load (PR #58), and aggregate populate (PR #47)
 - Added `opentelemetry_api`, `opentelemetry_telemetry`, and `opentelemetry_semantic_conventions` as required dependencies
 
 **Usage:**
@@ -243,3 +243,17 @@ end
 - Visualize event handler execution in your tracing backend
 - Correlate event processing with command dispatch using span links
 - Configurable span relationships (`:link`, `:child`, `:none`)
+
+### **Aggregate Load Telemetry**
+
+[PR #58](https://github.com/straw-hat-team/commanded/pull/58)
+
+**Changes:**
+- Added `[:commanded, :aggregate, :load]` telemetry for full event store load (stream_forward + consumption)
+- Load spans fire for both new aggregates (`stream_not_found`, `count: 0`) and existing aggregates
+- Populate spans remain unchanged: only fire when applying events to rebuild state
+- Trace hierarchy: `commanded.aggregate.load` (parent) → `commanded.aggregate.populate` (child, when events exist)
+
+**Benefits:**
+- Measure event store read latency as a whole, including the "not found" case
+- Separate load (event store I/O) from populate (state rebuild) in traces

lib/commanded/aggregates/aggregate_state_builder.ex

@@ -7,9 +7,34 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
   alias Commanded.Snapshotting
   alias Commanded.Telemetry
 
+  telemetry_event(%{
+    event: [:commanded, :aggregate, :load, :start],
+    description:
+      "Emitted when an aggregate begins loading from the event store (stream_forward + consumption)",
+    measurements: "%{system_time: integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      aggregate_uuid: String.t(),
+      aggregate_state: struct(),
+      aggregate_version: non_neg_integer()}
+    """
+  })
+
+  telemetry_event(%{
+    event: [:commanded, :aggregate, :load, :stop],
+    description: "Emitted when an aggregate completes loading from the event store",
+    measurements: "%{duration: non_neg_integer(), count: non_neg_integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      aggregate_uuid: String.t(),
+      aggregate_state: struct(),
+      aggregate_version: non_neg_integer()}
+    """
+  })
+
   telemetry_event(%{
     event: [:commanded, :aggregate, :populate, :start],
-    description: "Emitted when an aggregate begins loading from the event store",
+    description: "Emitted when an aggregate begins applying events to rebuild state",
     measurements: "%{system_time: integer()}",
     metadata: """
     %{application: Commanded.Application.t(),
@@ -21,7 +46,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
 
   telemetry_event(%{
     event: [:commanded, :aggregate, :populate, :stop],
-    description: "Emitted when an aggregate completes loading from the event store",
+    description: "Emitted when an aggregate completes applying events to rebuild state",
     measurements: "%{duration: non_neg_integer(), count: non_neg_integer()}",
     metadata: """
     %{application: Commanded.Application.t(),
@@ -65,25 +90,35 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
   Load events from the event store, in batches, to rebuild the aggregate state
   """
   def rebuild_from_events(%Aggregate{} = state) do
+    load_prefix = [:commanded, :aggregate, :load]
+    load_start = Telemetry.start(load_prefix, telemetry_metadata(state))
+
     %Aggregate{
       application: application,
       aggregate_uuid: aggregate_uuid,
       aggregate_version: aggregate_version
     } = state
 
-    case EventStore.stream_forward(
-           application,
-           aggregate_uuid,
-           aggregate_version + 1,
-           @read_event_batch_size
-         ) do
-      {:error, :stream_not_found} ->
-        # aggregate does not exist, return initial state
-        state
-
-      event_stream ->
-        rebuild_from_event_stream(event_stream, state)
-    end
+    result =
+      case EventStore.stream_forward(
+             application,
+             aggregate_uuid,
+             aggregate_version + 1,
+             @read_event_batch_size
+           ) do
+        {:error, :stream_not_found} ->
+          # aggregate does not exist, return initial state
+          Telemetry.stop(load_prefix, load_start, telemetry_metadata(state), %{count: 0})
+          {state, 0}
+
+        event_stream ->
+          {state, count} = rebuild_from_event_stream(event_stream, state)
+          Telemetry.stop(load_prefix, load_start, telemetry_metadata(state), %{count: count})
+          {state, count}
+      end
+
+    {state, _count} = result
+    state
   end
 
   # Rebuild aggregate state from a `Stream` of its events.
@@ -107,7 +142,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
 
     Telemetry.stop(telemetry_prefix, start_time, telemetry_metadata(state), %{count: count})
 
-    state
+    {state, count}
   end
 
   defp telemetry_metadata(%Aggregate{} = state) do

lib/commanded/opentelemetry/aggregate_populate.ex

@@ -11,8 +11,10 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do
   def setup do
     :ok =
       :telemetry.attach_many(
-        {__MODULE__, :populate},
+        {__MODULE__, :load_populate},
         [
+          [:commanded, :aggregate, :load, :start],
+          [:commanded, :aggregate, :load, :stop],
           [:commanded, :aggregate, :populate, :start],
           [:commanded, :aggregate, :populate, :stop]
         ],
@@ -21,6 +23,53 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do
       )
   end
 
+  def handle_telemetry_event(
+        [:commanded, :aggregate, :load, :start],
+        _measurements,
+        meta,
+        _config
+      ) do
+    attributes = [
+      {MessagingAttributes.messaging_system(), "commanded"},
+      {MessagingAttributes.messaging_operation_type(), :receive},
+      {MessagingAttributes.messaging_operation_name(), "load"},
+      {CodeAttributes.code_function(), "load"},
+      {CommandedAttributes.commanded_application(), meta.application},
+      {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid},
+      {CommandedAttributes.commanded_aggregate_version(), meta.aggregate_version}
+    ]
+
+    OpentelemetryTelemetry.start_telemetry_span(
+      @tracer_id,
+      "commanded.aggregate.load",
+      meta,
+      %{
+        kind: :internal,
+        attributes: attributes
+      }
+    )
+  end
+
+  def handle_telemetry_event(
+        [:commanded, :aggregate, :load, :stop],
+        measurements,
+        meta,
+        _config
+      ) do
+    ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta)
+
+    event_count = Map.get(measurements, :count, 0)
+    Span.set_attribute(ctx, CommandedAttributes.commanded_event_count(), event_count)
+
+    Span.set_attribute(
+      ctx,
+      CommandedAttributes.commanded_aggregate_version(),
+      meta.aggregate_version
+    )
+
+    OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta)
+  end
+
   def handle_telemetry_event(
         [:commanded, :aggregate, :populate, :start],
         _measurements,

test/aggregates/aggregate_telemetry_test.exs

@@ -2,7 +2,7 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do
   use Commanded.MockEventStoreCase
 
   alias Commanded.Aggregates.{Aggregate, ExecutionContext}
-  alias Commanded.{DefaultApp, UUID}
+  alias Commanded.{DefaultApp, MockedApp, UUID}
 
   defmodule Commands do
     defmodule Ok do
@@ -193,7 +193,30 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do
       refute_received {[:commanded, :aggregate, :execute, :stop], _measurements, _metadata}
     end
 
-    test "emit `[:commanded, :aggregate, :populate]` events",
+    test "emit `[:commanded, :aggregate, :load]` events for new aggregate (stream_not_found)",
+         %{aggregate_uuid: aggregate_uuid} do
+      # Setup already started the aggregate. For a new aggregate, stream_forward returns
+      # {:error, :stream_not_found}. Load telemetry fires with count: 0; populate does not fire.
+      assert_receive {[:commanded, :aggregate, :load, :start], _measurements, _metadata}
+      assert_receive {[:commanded, :aggregate, :load, :stop], measurements, metadata}
+
+      assert match?(%{count: 0}, measurements)
+
+      assert match?(
+               %{
+                 aggregate_state: %ExampleAggregate{},
+                 aggregate_uuid: ^aggregate_uuid,
+                 aggregate_version: 0,
+                 application: DefaultApp
+               },
+               metadata
+             )
+
+      refute_received {[:commanded, :aggregate, :populate, :start], _, _}
+      refute_received {[:commanded, :aggregate, :populate, :stop], _, _}
+    end
+
+    test "emit `[:commanded, :aggregate, :load]` and `[:commanded, :aggregate, :populate]` for existing aggregate (reload)",
          %{aggregate_uuid: aggregate_uuid, pid: pid} do
       context = %ExecutionContext{
         command: %Ok{message: "ok"},
@@ -210,13 +233,16 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do
 
       Process.exit(pid, :normal)
 
-      # Do the reload, we should now have telemetry
-      start_aggregate(aggregate_uuid)
+      # Do the reload. Consume initial load (count: 0) from setup, then reload load + populate.
+      assert_receive {[:commanded, :aggregate, :load, :start], _, _}
+      assert_receive {[:commanded, :aggregate, :load, :stop], %{count: 0}, _}
 
-      assert_receive {[:commanded, :aggregate, :populate, :start], _measurements, _metadata}
-      assert_receive {[:commanded, :aggregate, :populate, :stop], measurements, metadata}
+      start_aggregate(aggregate_uuid)
 
-      assert match?(%{count: ^count}, measurements)
+      assert_receive {[:commanded, :aggregate, :load, :start], _, _}
+      assert_receive {[:commanded, :aggregate, :populate, :start], _, _}
+      assert_receive {[:commanded, :aggregate, :populate, :stop], %{count: ^count}, metadata}
+      assert_receive {[:commanded, :aggregate, :load, :stop], %{count: ^count}, _}
 
       assert match?(
                %{
@@ -278,6 +304,63 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do
     end
   end
 
+  @tag :unit
+  describe "load/populate telemetry (unit: MockedApp + expect)" do
+    setup do
+      attach_telemetry()
+      :ok
+    end
+
+    test "load only when stream_forward returns stream_not_found" do
+      aggregate_uuid = UUID.uuid4()
+
+      expect(MockEventStore, :subscribe, fn _meta, ^aggregate_uuid -> :ok end)
+
+      expect(MockEventStore, :stream_forward, fn _meta, ^aggregate_uuid, _from, _batch_size ->
+        {:error, :stream_not_found}
+      end)
+
+      assert {:ok, _pid} = start_aggregate(aggregate_uuid, application: MockedApp)
+
+      assert_receive {[:commanded, :aggregate, :load, :start], _, _}
+      assert_receive {[:commanded, :aggregate, :load, :stop], %{count: 0}, _}
+
+      refute_received {[:commanded, :aggregate, :populate, :start], _, _}
+      refute_received {[:commanded, :aggregate, :populate, :stop], _, _}
+    end
+
+    test "load + populate when stream_forward returns events" do
+      aggregate_uuid = UUID.uuid4()
+      count = 2
+
+      expect(MockEventStore, :subscribe, fn _meta, ^aggregate_uuid -> :ok end)
+
+      expect(MockEventStore, :stream_forward, fn _meta, ^aggregate_uuid, _from, _batch_size ->
+        for i <- 1..count do
+          %Commanded.EventStore.RecordedEvent{
+            event_id: UUID.uuid4(),
+            event_number: i,
+            stream_id: aggregate_uuid,
+            stream_version: i,
+            correlation_id: nil,
+            causation_id: nil,
+            event_type: "Elixir.Commanded.Aggregates.AggregateTelemetryTest.Event",
+            data: %Event{message: "event#{i}"},
+            metadata: nil,
+            created_at: DateTime.utc_now()
+          }
+        end
+      end)
+
+      assert {:ok, _pid} = start_aggregate(aggregate_uuid, application: MockedApp)
+
+      assert_receive {[:commanded, :aggregate, :load, :start], _, _}
+      assert_receive {[:commanded, :aggregate, :populate, :start], _, _}
+      assert_receive {[:commanded, :aggregate, :populate, :stop], %{count: ^count}, _}
+      assert_receive {[:commanded, :aggregate, :load, :stop], %{count: ^count}, _}
+    end
+  end
+
   def start_aggregate(aggregate_uuid) do
     Aggregate.start_link([application: DefaultApp],
       aggregate_module: ExampleAggregate,
@@ -305,6 +388,8 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do
         [:commanded, :aggregate, :execute, :stop],
         [:commanded, :aggregate, :execute, :exception],
         [:commanded, :aggregate, :execute, :wrong_expected_version],
+        [:commanded, :aggregate, :load, :start],
+        [:commanded, :aggregate, :load, :stop],
         [:commanded, :aggregate, :populate, :start],
         [:commanded, :aggregate, :populate, :stop]
       ],