From 57ed7630f882d0c5197cbea132acc7d29381e226 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 31 Mar 2026 21:24:38 -0400 Subject: [PATCH] fix(event_store): defer stream_forward telemetry until stream consumed Signed-off-by: Yordis Prieto --- lib/commanded/event_store.ex | 81 ++++++++++++++++- test/event_store/telemetry_test.exs | 131 +++++++++++++++++++++++++++- 2 files changed, 207 insertions(+), 5 deletions(-) diff --git a/lib/commanded/event_store.ex b/lib/commanded/event_store.ex index d2a5f9f3..50b79d65 100644 --- a/lib/commanded/event_store.ex +++ b/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 diff --git a/test/event_store/telemetry_test.exs b/test/event_store/telemetry_test.exs index 3253f0a9..35eec7fe 100644 --- a/test/event_store/telemetry_test.exs +++ b/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 ->