fix(event_store): defer stream_forward telemetry until stream consumed

lib/commanded/event_store.ex

@@ -57,16 +57,40 @@ defmodule Commanded.EventStore do
 
   @doc """
   Streams events from the given stream, in the order in which they were originally written.
+
+  Telemetry `[:commanded, :event_store, :stream_forward, :stop]` (and matching OpenTelemetry
+  spans) fire when enumeration finishes for adapters that return **lazy streams**,
+  so duration reflects read-from-store work. If the adapter returns a **plain list**,
+  `:stop` runs when `stream_forward` returns, as before.
+
+  For `{:error, _}` results, `:stop` still runs immediately after the failed call.
+
+  Adapter resolution and `stream_forward` are covered by the same span: if either raises,
+  `[:commanded, :event_store, :stream_forward, :exception]` is emitted (after `:start`)
+  with `kind`, `reason`, and `stacktrace` metadata, matching `:telemetry.span/3` behaviour.
+
+  Exceptions raised during **enumeration** of a lazy stream (i.e. inside `Enum.to_list/1`
+  or similar) are **not** wrapped by this telemetry — they are the caller's responsibility.
   """
   def stream_forward(application, stream_uuid, start_version \\ 0, read_batch_size \\ 1_000) do
-    meta = %{
+    base_meta = %{
       application: application,
       stream_uuid: stream_uuid,
       start_version: start_version,
       read_batch_size: read_batch_size
     }
 
-    span(:stream_forward, meta, fn ->
+    start_monotonic = :erlang.monotonic_time()
+    system_time = :erlang.system_time()
+    meta = Map.put(base_meta, :telemetry_span_context, make_ref())
+
+    :telemetry.execute(
+      [:commanded, :event_store, :stream_forward, :start],
+      %{monotonic_time: start_monotonic, system_time: system_time},
+      meta
+    )
+
+    try do
       {adapter, adapter_meta} = Application.event_store_adapter(application)
 
       case adapter.stream_forward(
@@ -76,12 +100,21 @@ defmodule Commanded.EventStore do
              read_batch_size
            ) do
         {:error, _error} = error ->
+          stream_forward_stop(start_monotonic, meta)
           error
 
-        stream ->
+        stream when is_list(stream) ->
+          stream_forward_stop(start_monotonic, meta)
           stream
+
+        stream ->
+          wrap_stream_forward_telemetry(stream, meta, start_monotonic)
       end
-    end)
+    catch
+      kind, reason ->
+        stream_forward_exception(start_monotonic, meta, kind, reason, __STACKTRACE__)
+        :erlang.raise(kind, reason, __STACKTRACE__)
+    end
   end
 
   @doc """
@@ -288,4 +321,44 @@ defmodule Commanded.EventStore do
       {func.(), meta}
     end)
   end
+
+  defp stream_forward_stop(start_monotonic, meta) do
+    stop_monotonic = :erlang.monotonic_time()
+
+    :telemetry.execute(
+      [:commanded, :event_store, :stream_forward, :stop],
+      %{
+        duration: stop_monotonic - start_monotonic,
+        monotonic_time: stop_monotonic,
+        system_time: :erlang.system_time()
+      },
+      meta
+    )
+  end
+
+  defp stream_forward_exception(start_monotonic, meta, kind, reason, stacktrace) do
+    stop_monotonic = :erlang.monotonic_time()
+
+    :telemetry.execute(
+      [:commanded, :event_store, :stream_forward, :exception],
+      %{
+        duration: stop_monotonic - start_monotonic,
+        monotonic_time: stop_monotonic,
+        system_time: :erlang.system_time()
+      },
+      Map.merge(meta, %{kind: kind, reason: reason, stacktrace: stacktrace})
+    )
+  end
+
+  # Deferred :stop only; :start was already emitted by stream_forward/4.
+  defp wrap_stream_forward_telemetry(stream, meta, start_monotonic) do
+    Stream.transform(
+      stream,
+      fn -> nil end,
+      fn elem, acc -> {[elem], acc} end,
+      fn _acc ->
+        stream_forward_stop(start_monotonic, meta)
+      end
+    )
+  end
 end

test/event_store/telemetry_test.exs

@@ -1,13 +1,17 @@
 defmodule Commanded.EventStore.TelemetryTest do
   use ExUnit.Case
 
+  import Mox
+
   alias Commanded.DefaultApp
   alias Commanded.EventStore
+  alias Commanded.EventStore.Adapters.Mock, as: MockEventStore
   alias Commanded.EventStore.EventData
   alias Commanded.EventStore.RecordedEvent
   alias Commanded.EventStore.SnapshotData
   alias Commanded.Middleware.Commands.IncrementCount
   alias Commanded.Middleware.Commands.RaiseError
+  alias Commanded.MockedApp
   alias Commanded.UUID
 
   setup do
@@ -78,6 +82,99 @@ defmodule Commanded.EventStore.TelemetryTest do
                read_batch_size: 1_000
              } = meta
     end
+
+    test "emit stream_forward start/stop when adapter returns a list (before caller enumerates)" do
+      uuid = UUID.uuid4()
+      assert :ok = EventStore.append_to_stream(DefaultApp, uuid, 0, [%EventData{}])
+
+      assert_receive {[:commanded, :event_store, :append_to_stream, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :append_to_stream, :stop], 2, _meas, _meta}
+
+      stream = EventStore.stream_forward(DefaultApp, uuid, 0)
+
+      assert_receive {[:commanded, :event_store, :stream_forward, :start], 3, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :stream_forward, :stop], 4, _meas, meta}
+
+      assert %{
+               application: DefaultApp,
+               stream_uuid: ^uuid,
+               start_version: 0,
+               read_batch_size: 1_000
+             } = meta
+
+      assert is_list(stream)
+      assert length(stream) == 1
+    end
+
+    test "deferred :stop fires only after lazy stream is enumerated" do
+      {_app, _handler} = start_mocked_lazy_stream([:a, :b])
+
+      stream = EventStore.stream_forward(Commanded.MockedApp, "stream-uuid", 0)
+
+      assert_receive {:deferred, [:commanded, :event_store, :stream_forward, :start], _start_meas,
+                      %{telemetry_span_context: ctx}}
+
+      refute_receive {:deferred, [:commanded, :event_store, :stream_forward, :stop], _, _}, 50
+
+      assert [:a, :b] = Enum.to_list(stream)
+
+      assert_receive {:deferred, [:commanded, :event_store, :stream_forward, :stop], stop_meas,
+                      stop_meta}
+
+      assert stop_meta.telemetry_span_context == ctx
+      assert is_integer(stop_meas.duration)
+      assert stop_meas.duration >= 0
+    end
+
+    test "deferred :stop fires when lazy stream is halted early" do
+      {_app, _handler} = start_mocked_lazy_stream([:a, :b, :c])
+
+      stream = EventStore.stream_forward(Commanded.MockedApp, "stream-uuid", 0)
+
+      assert_receive {:deferred, [:commanded, :event_store, :stream_forward, :start], _, _}
+      refute_receive {:deferred, [:commanded, :event_store, :stream_forward, :stop], _, _}, 50
+
+      assert [:a] = Enum.take(stream, 1)
+
+      assert_receive {:deferred, [:commanded, :event_store, :stream_forward, :stop], stop_meas, _}
+      assert is_integer(stop_meas.duration)
+    end
+
+    test "emit stream_forward :exception after :start when application lookup raises" do
+      missing = :stream_forward_telemetry_missing_commanded_app_xx
+      handler = :"stream_forward_ex-#{System.unique_integer([:positive])}"
+
+      :telemetry.attach_many(
+        handler,
+        [
+          [:commanded, :event_store, :stream_forward, :start],
+          [:commanded, :event_store, :stream_forward, :exception]
+        ],
+        fn event, measurements, metadata, reply_to ->
+          send(reply_to, {:stream_forward_telemetry, event, measurements, metadata})
+        end,
+        self()
+      )
+
+      on_exit(fn -> :telemetry.detach(handler) end)
+
+      assert_raise RuntimeError, fn ->
+        EventStore.stream_forward(missing, "stream-uuid", 0)
+      end
+
+      assert_receive {:stream_forward_telemetry,
+                      [:commanded, :event_store, :stream_forward, :start], _start_meas,
+                      start_meta}
+
+      assert_receive {:stream_forward_telemetry,
+                      [:commanded, :event_store, :stream_forward, :exception], ex_meas, ex_meta}
+
+      assert start_meta.telemetry_span_context == ex_meta.telemetry_span_context
+      assert ex_meta.kind == :error
+      assert %RuntimeError{} = ex_meta.reason
+      assert is_list(ex_meta.stacktrace)
+      assert is_integer(ex_meas.duration)
+    end
   end
 
   describe "ack_event telemetry events" do
@@ -155,6 +252,37 @@ defmodule Commanded.EventStore.TelemetryTest do
     end
   end
 
+  defp start_mocked_lazy_stream(elements) do
+    set_mox_global()
+
+    stub(MockEventStore, :child_spec, fn _app, _cfg -> {:ok, [], %{}} end)
+    stub(MockEventStore, :subscribe, fn _meta, _stream -> :ok end)
+
+    stub(MockEventStore, :stream_forward, fn _meta, _uuid, _from, _batch ->
+      Stream.map(elements, & &1)
+    end)
+
+    start_supervised!(MockedApp)
+
+    handler = :"deferred-#{System.unique_integer([:positive])}"
+
+    :telemetry.attach_many(
+      handler,
+      [
+        [:commanded, :event_store, :stream_forward, :start],
+        [:commanded, :event_store, :stream_forward, :stop]
+      ],
+      fn event, measurements, metadata, reply_to ->
+        send(reply_to, {:deferred, event, measurements, metadata})
+      end,
+      self()
+    )
+
+    on_exit(fn -> :telemetry.detach(handler) end)
+
+    {MockedApp, handler}
+  end
+
   defp attach_telemetry do
     agent = start_supervised!({Agent, fn -> 1 end})
     handler = :"#{__MODULE__}-handler"
@@ -177,7 +305,8 @@ defmodule Commanded.EventStore.TelemetryTest do
       Enum.flat_map(events, fn event ->
         [
           [:commanded, :event_store, event, :start],
-          [:commanded, :event_store, event, :stop]
+          [:commanded, :event_store, event, :stop],
+          [:commanded, :event_store, event, :exception]
         ]
       end),
       fn event_name, measurements, metadata, reply_to ->