From 6da285af7da23a9c3a7bcb8fdffc72f29fab733a Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sun, 5 Apr 2026 11:58:16 -0400 Subject: [PATCH] feat(telemetry): add processing_latency_ms to event handler stop measurements Signed-off-by: Yordis Prieto --- guides/explanations/fork-differences.md | 12 ++ lib/commanded/event/handler.ex | 101 ++++++++-- .../event_handler_batch_ordering_test.exs | 73 ++++++++ .../event_handler_batch_telemetry_test.exs | 138 -------------- test/event/event_handler_telemetry_test.exs | 177 ++++++++++++++++++ 5 files changed, 346 insertions(+), 155 deletions(-) create mode 100644 test/event/event_handler_batch_ordering_test.exs delete mode 100644 test/event/event_handler_batch_telemetry_test.exs create mode 100644 test/event/event_handler_telemetry_test.exs diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index 653b23ab..b2011318 100644 --- a/guides/explanations/fork-differences.md +++ b/guides/explanations/fork-differences.md @@ -272,3 +272,15 @@ end - Track `wrong_expected_version_count` in aggregate struct and `:stop` telemetry metadata - OTel execute span records `commanded.aggregate.wrong_expected_version` event when count > 0 - Enables alerting on optimistic concurrency conflicts via telemetry + +### **Event Handler Processing Latency Telemetry** +[PR #66](https://github.com/straw-hat-team/commanded/pull/66) + +**Changes:** +- Added `processing_latency_ms` measurement to `[:commanded, :event, :handle, :stop]` and `[:commanded, :event, :batch, :stop]` telemetry events +- `processing_latency_ms` is the elapsed milliseconds between `RecordedEvent.created_at` and handler completion — how long it took the system to fully process the event from creation to done +- For batch handlers, reflects the oldest event in the batch (worst-case processing latency) + +**Benefits:** +- All event handlers (projectors, sagas, notification handlers) get processing latency visibility for free +- Enables SLA dashboards and alerting without any application-level code diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex index 8b65b079..e545f14b 100644 --- a/lib/commanded/event/handler.ex +++ b/lib/commanded/event/handler.ex @@ -20,7 +20,7 @@ defmodule Commanded.Event.Handler do telemetry_event(%{ event: [:commanded, :event, :handle, :stop], description: "Emitted when an event handler stops handling an event", - measurements: "%{duration: non_neg_integer()}", + measurements: "%{duration: non_neg_integer(), processing_latency_ms: non_neg_integer()}", metadata: """ %{:application => Commanded.Application.t(), :context => map(), @@ -68,7 +68,7 @@ defmodule Commanded.Event.Handler do telemetry_event(%{ event: [:commanded, :event, :batch, :stop], description: "Emitted when an event handler stops handling a batch of events", - measurements: "%{duration: non_neg_integer()}", + measurements: "%{duration: non_neg_integer(), processing_latency_ms: non_neg_integer()}", metadata: """ %{application: Commanded.Application.t(), context: map(), @@ -1046,7 +1046,12 @@ defmodule Commanded.Event.Handler do case delegate_event_to_handler(event, state) do :ok -> - telemetry_stop(start_time, telemetry_metadata, :handle) + telemetry_stop( + start_time, + telemetry_metadata, + :handle, + processing_latency_measurements(event) + ) confirm_receipt(event, state) @@ -1054,7 +1059,8 @@ defmodule Commanded.Event.Handler do telemetry_stop( start_time, Map.put(telemetry_metadata, :handler_state, handler_state), - :handle + :handle, + processing_latency_measurements(event) ) confirm_receipt(event, %Handler{state | handler_state: handler_state}) @@ -1063,14 +1069,21 @@ defmodule Commanded.Event.Handler do telemetry_stop( start_time, Map.put(telemetry_metadata, :error, :already_seen_event), - :handle + :handle, + processing_latency_measurements(event) ) confirm_receipt(event, state) {:error, reason} = error -> log_event_error(error, event, state) - telemetry_stop(start_time, Map.put(telemetry_metadata, :error, reason), :handle) + + telemetry_stop( + start_time, + Map.put(telemetry_metadata, :error, reason), + :handle, + processing_latency_measurements(event) + ) failure_context = build_failure_context(event, context, state) retry_fun = fn context, state -> handle_event(event, context, state) end @@ -1078,7 +1091,16 @@ defmodule Commanded.Event.Handler do {:error, reason, stacktrace} -> log_event_error({:error, reason, stacktrace}, event, state) - telemetry_exception(start_time, :error, reason, stacktrace, telemetry_metadata, :handle) + + telemetry_exception( + start_time, + :error, + reason, + stacktrace, + telemetry_metadata, + :handle, + processing_latency_measurements(event) + ) failure_context = build_failure_context(event, context, stacktrace, state) retry_fun = fn context, state -> handle_event(event, context, state) end @@ -1097,7 +1119,8 @@ defmodule Commanded.Event.Handler do telemetry_stop( start_time, Map.put(telemetry_metadata, :error, :invalid_return_value), - :handle + :handle, + processing_latency_measurements(event) ) error = {:error, :invalid_return_value} @@ -1137,16 +1160,34 @@ defmodule Commanded.Event.Handler do case delegate_event_to_handler(events, state) do :ok -> - telemetry_stop(start_time, telemetry_metadata, :batch) + telemetry_stop( + start_time, + telemetry_metadata, + :batch, + processing_latency_measurements(events) + ) + confirm_receipt(events, state) {:ok, handler_state} -> - telemetry_stop(start_time, %{telemetry_metadata | handler_state: state}, :batch) + telemetry_stop( + start_time, + %{telemetry_metadata | handler_state: handler_state}, + :batch, + processing_latency_measurements(events) + ) + confirm_receipt(events, %Handler{state | handler_state: handler_state}) {:error, reason} = error -> log_batch_error(error, events, state) - telemetry_stop(start_time, Map.put(telemetry_metadata, :error, reason), :batch) + + telemetry_stop( + start_time, + Map.put(telemetry_metadata, :error, reason), + :batch, + processing_latency_measurements(events) + ) failure_context = build_failure_context(nil, context, state) retry_fun = fn context, state -> handle_batch(events, context, state) end @@ -1154,7 +1195,16 @@ defmodule Commanded.Event.Handler do {:error, reason, stacktrace} -> log_batch_error({:error, reason, stacktrace}, events, state) - telemetry_exception(start_time, :error, reason, stacktrace, telemetry_metadata, :batch) + + telemetry_exception( + start_time, + :error, + reason, + stacktrace, + telemetry_metadata, + :batch, + processing_latency_measurements(events) + ) failure_context = build_failure_context(nil, context, stacktrace, state) retry_fun = fn context, state -> handle_batch(events, context, state) end @@ -1170,7 +1220,8 @@ defmodule Commanded.Event.Handler do telemetry_stop( start_time, Map.put(telemetry_metadata, :error, :invalid_return_value), - :batch + :batch, + processing_latency_measurements(events) ) failure_context = build_failure_context(nil, context, state) @@ -1428,8 +1479,13 @@ defmodule Commanded.Event.Handler do Telemetry.start([:commanded, :event, telemetry_type], telemetry_metadata) end - defp telemetry_stop(start_time, telemetry_metadata, telemetry_type) do - Telemetry.stop([:commanded, :event, telemetry_type], start_time, telemetry_metadata) + defp telemetry_stop(start_time, telemetry_metadata, telemetry_type, additional_measurements) do + Telemetry.stop( + [:commanded, :event, telemetry_type], + start_time, + telemetry_metadata, + additional_measurements + ) end defp telemetry_exception( @@ -1438,7 +1494,8 @@ defmodule Commanded.Event.Handler do reason, stacktrace, telemetry_metadata, - telemetry_type + telemetry_type, + additional_measurements ) do Telemetry.exception( [:commanded, :event, telemetry_type], @@ -1446,10 +1503,20 @@ defmodule Commanded.Event.Handler do kind, reason, stacktrace, - telemetry_metadata + telemetry_metadata, + additional_measurements ) end + defp processing_latency_measurements(%RecordedEvent{created_at: %DateTime{} = created_at}) do + %{processing_latency_ms: DateTime.diff(DateTime.utc_now(), created_at, :millisecond)} + end + + defp processing_latency_measurements([%RecordedEvent{} = first | _]), + do: processing_latency_measurements(first) + + defp processing_latency_measurements(_), do: %{} + defp batch_telemetry_metadata(recorded_events, context, %Handler{} = state) when is_list(recorded_events) do first_event = List.first(recorded_events) diff --git a/test/event/event_handler_batch_ordering_test.exs b/test/event/event_handler_batch_ordering_test.exs new file mode 100644 index 00000000..9494df1b --- /dev/null +++ b/test/event/event_handler_batch_ordering_test.exs @@ -0,0 +1,73 @@ +defmodule Commanded.Event.EventHandlerBatchOrderingTest do + use ExUnit.Case, async: false + + @moduletag :eventstore_adapter + + # Verifies that events returned by the EventStore arrive ordered by + # event_number ascending (first = oldest, last = newest), which is the + # assumption behind using List.first/1 to compute processing_latency_ms + # for the worst-case (oldest) event in a batch. + + alias Commanded.UUID + + defmodule TestEvent do + @derive Jason.Encoder + defstruct [:index] + end + + defmodule App do + use Commanded.Application, + otp_app: :commanded, + event_store: [ + adapter: Commanded.EventStore.Adapters.EventStore, + event_store: TestEventStore + ], + pubsub: :local, + registry: :local + end + + setup do + alias Commanded.EventStore.Adapters.EventStore.Storage + + config = Storage.config() + + on_exit(fn -> + {:ok, conn} = Storage.connect(config) + Storage.reset!(conn, config) + end) + + start_supervised!(App) + :ok + end + + test "stream_forward returns events ordered ascending by event_number (first = oldest)" do + stream_id = UUID.uuid4() + + events = + Enum.map(1..3, fn index -> + %Commanded.EventStore.EventData{ + event_type: Atom.to_string(TestEvent), + data: %TestEvent{index: index}, + metadata: %{} + } + end) + + :ok = Commanded.EventStore.append_to_stream(App, stream_id, 0, events) + + recorded_events = + App + |> Commanded.EventStore.stream_forward(stream_id) + |> Enum.to_list() + + assert length(recorded_events) == 3 + + event_numbers = Enum.map(recorded_events, & &1.event_number) + created_ats = Enum.map(recorded_events, & &1.created_at) + + assert event_numbers == Enum.sort(event_numbers), + "expected events ordered ascending by event_number, got: #{inspect(event_numbers)}" + + assert created_ats == Enum.sort(created_ats, DateTime), + "expected events ordered ascending by created_at, got: #{inspect(created_ats)}" + end +end diff --git a/test/event/event_handler_batch_telemetry_test.exs b/test/event/event_handler_batch_telemetry_test.exs deleted file mode 100644 index 8c994e41..00000000 --- a/test/event/event_handler_batch_telemetry_test.exs +++ /dev/null @@ -1,138 +0,0 @@ -defmodule Commanded.Event.EventHandlerBatchTelemetryTest do - use ExUnit.Case - - alias Commanded.Application.Config - alias Commanded.Event.Handler - alias Commanded.EventStore.Subscription - - # Test support code - alias Commanded.Helpers.EventFactory - alias Commanded.Event.{BatchHandler, ErrorHandlingBatchHandler} - alias Commanded.Event.EventHandlerBatchTelemetryTest.MockAdapter - alias Commanded.Event.ReplyEvent - - setup do - attach_telemetry() - :ok - end - - describe "batch event handler telemetry" do - test "should emit `[:commanded, :event, :batch, :stop]` telemetry from :ok" do - events = [ - %ReplyEvent{reply_to: self(), value: 1}, - %ReplyEvent{reply_to: self(), value: 2} - ] - - metadata = %{"key" => "value"} - recorded_events = EventFactory.map_to_recorded_events(events, 1, metadata: metadata) - state = setup_state(BatchHandler) - - {:noreply, _state} = Handler.handle_info({:events, recorded_events}, state) - - assert_receive {[:commanded, :event, :batch, :start], _measurements, _metadata} - - assert_receive {[:commanded, :event, :batch, :stop], _measurements, metadata} - assert metadata.event_count == 2 - - refute_received {[:commanded, :event, :batch, :exception], _measurements, _metadata} - end - - test "should emit `[:commanded, :event, :batch, :stop]` telemetry from :error" do - events = [ - %ReplyEvent{reply_to: self(), value: :error}, - %ReplyEvent{reply_to: self(), value: :error} - ] - - metadata = %{"key" => "value"} - recorded_events = EventFactory.map_to_recorded_events(events, 1, metadata: metadata) - state = setup_state(BatchHandler) - - {:stop, :bad_value, _state} = Handler.handle_info({:events, recorded_events}, state) - - assert_receive {[:commanded, :event, :batch, :start], _measurements, _metadata} - assert_receive {[:commanded, :event, :batch, :stop], _measurements, _metadata} - refute_received {[:commanded, :event, :batch, :exception], _measurements, _metadata} - end - - test "should include count of successfully processed events when erroring on a specific event" do - events = [ - %ReplyEvent{reply_to: self(), value: 1}, - %ReplyEvent{reply_to: self(), value: 2}, - %ReplyEvent{reply_to: self(), value: :error}, - %ReplyEvent{reply_to: self(), value: 4} - ] - - metadata = %{"key" => "value"} - recorded_events = EventFactory.map_to_recorded_events(events, 1, metadata: metadata) - state = setup_state(BatchHandler) - - Handler.handle_info({:events, recorded_events}, state) - - assert_receive {[:commanded, :event, :batch, :start], _measurements, metadata} - assert metadata.event_count == 4 - - assert_receive {[:commanded, :event, :batch, :stop], _measurements, metadata} - assert metadata.event_count == 4 - end - - test "should emit `[:commanded, :event, :batch, :exception]` telemetry from thrown exception" do - events = [ - %ReplyEvent{reply_to: self(), value: :raise}, - %ReplyEvent{reply_to: self(), value: :raise} - ] - - metadata = %{"key" => "value"} - recorded_events = EventFactory.map_to_recorded_events(events, 1, metadata: metadata) - state = setup_state(ErrorHandlingBatchHandler) - - {:stop, _, _} = Handler.handle_info({:events, recorded_events}, state) - - assert_receive {[:commanded, :event, :batch, :start], _measurements, _metadata} - refute_received {[:commanded, :event, :batch, :stop], _measurements, _metadata} - assert_receive {[:commanded, :event, :batch, :exception], _measurements, _metadata} - end - end - - defp setup_state(handler_module) do - Config.associate(self(), __MODULE__, event_store: {MockAdapter, nil}) - - %Handler{ - subscription: - struct(Subscription, - application: __MODULE__, - subscription_pid: self() - ), - handler_callback: :batch, - handler_module: handler_module, - application: __MODULE__, - consistency: :eventual, - last_seen_event: 0 - } - end - - defp attach_telemetry do - :telemetry.attach_many( - "test-handler", - [ - [:commanded, :event, :batch, :start], - [:commanded, :event, :batch, :stop], - [:commanded, :event, :batch, :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 - - defmodule MockAdapter do - def ack_event(nil, subscription_pid, event) do - send(subscription_pid, {:acked, event}) - :ok - end - end -end diff --git a/test/event/event_handler_telemetry_test.exs b/test/event/event_handler_telemetry_test.exs new file mode 100644 index 00000000..bcd085ba --- /dev/null +++ b/test/event/event_handler_telemetry_test.exs @@ -0,0 +1,177 @@ +defmodule Commanded.Event.EventHandlerTelemetryTest do + use ExUnit.Case + + alias Commanded.Application.Config + alias Commanded.Event.Handler + alias Commanded.EventStore.Subscription + + alias Commanded.Helpers.EventFactory + alias Commanded.Event.{BatchHandler, EchoHandler, ErrorHandlingBatchHandler} + alias Commanded.Event.EventHandlerTelemetryTest.MockAdapter + alias Commanded.Event.ReplyEvent + + setup do + attach_telemetry() + :ok + end + + describe "single-event handler" do + test "emits :start and :stop" do + recorded_events = EventFactory.map_to_recorded_events([reply_event(1)]) + state = setup_state(EchoHandler, :event) + + Handler.handle_info({:events, recorded_events}, state) + + assert_receive {[:commanded, :event, :handle, :start], _measurements, _metadata} + assert_receive {[:commanded, :event, :handle, :stop], _measurements, _metadata} + refute_received {[:commanded, :event, :handle, :exception], _measurements, _metadata} + end + + test "includes processing_latency_ms in :stop measurements" do + recorded_events = EventFactory.map_to_recorded_events([reply_event(1)]) + state = setup_state(EchoHandler, :event) + + Handler.handle_info({:events, recorded_events}, state) + + assert_receive {[:commanded, :event, :handle, :stop], measurements, _metadata} + assert is_integer(measurements.processing_latency_ms) + assert measurements.processing_latency_ms >= 0 + end + end + + describe "batch event handler" do + test "emits :start and :stop from :ok" do + recorded_events = + EventFactory.map_to_recorded_events([batch_reply_event(1), batch_reply_event(2)], 1) + + state = setup_state(BatchHandler, :batch) + + {:noreply, _state} = Handler.handle_info({:events, recorded_events}, state) + + assert_receive {[:commanded, :event, :batch, :start], _measurements, _metadata} + assert_receive {[:commanded, :event, :batch, :stop], _measurements, metadata} + assert metadata.event_count == 2 + refute_received {[:commanded, :event, :batch, :exception], _measurements, _metadata} + end + + test "includes processing_latency_ms in :stop measurements" do + recorded_events = + EventFactory.map_to_recorded_events([batch_reply_event(1), batch_reply_event(2)], 1) + + state = setup_state(BatchHandler, :batch) + + {:noreply, _state} = Handler.handle_info({:events, recorded_events}, state) + + assert_receive {[:commanded, :event, :batch, :stop], measurements, _metadata} + assert is_integer(measurements.processing_latency_ms) + assert measurements.processing_latency_ms >= 0 + end + + test "emits :stop from :error" do + recorded_events = + EventFactory.map_to_recorded_events( + [batch_reply_event(:error), batch_reply_event(:error)], + 1 + ) + + state = setup_state(BatchHandler, :batch) + + {:stop, :bad_value, _state} = Handler.handle_info({:events, recorded_events}, state) + + assert_receive {[:commanded, :event, :batch, :start], _measurements, _metadata} + assert_receive {[:commanded, :event, :batch, :stop], _measurements, _metadata} + refute_received {[:commanded, :event, :batch, :exception], _measurements, _metadata} + end + + test "emits :exception from raised exception" do + recorded_events = + EventFactory.map_to_recorded_events( + [batch_reply_event(:raise), batch_reply_event(:raise)], + 1 + ) + + state = setup_state(ErrorHandlingBatchHandler, :batch) + + {:stop, _, _} = Handler.handle_info({:events, recorded_events}, state) + + assert_receive {[:commanded, :event, :batch, :start], _measurements, _metadata} + refute_received {[:commanded, :event, :batch, :stop], _measurements, _metadata} + assert_receive {[:commanded, :event, :batch, :exception], _measurements, _metadata} + end + + test "includes event_count in metadata when erroring on a specific event" do + recorded_events = + EventFactory.map_to_recorded_events( + [ + batch_reply_event(1), + batch_reply_event(2), + batch_reply_event(:error), + batch_reply_event(4) + ], + 1 + ) + + state = setup_state(BatchHandler, :batch) + + Handler.handle_info({:events, recorded_events}, state) + + assert_receive {[:commanded, :event, :batch, :start], _measurements, metadata} + assert metadata.event_count == 4 + + assert_receive {[:commanded, :event, :batch, :stop], _measurements, metadata} + assert metadata.event_count == 4 + end + end + + # EchoHandler expects reply_to as a charlist (:erlang.list_to_pid/1 is called internally) + defp reply_event(value), do: %ReplyEvent{reply_to: :erlang.pid_to_list(self()), value: value} + + # BatchHandler uses reply_to directly as a pid + defp batch_reply_event(value), do: %ReplyEvent{reply_to: self(), value: value} + + defp setup_state(handler_module, callback) do + Config.associate(self(), __MODULE__, event_store: {MockAdapter, nil}) + + %Handler{ + subscription: + struct(Subscription, + application: __MODULE__, + subscription_pid: self() + ), + handler_callback: callback, + handler_module: handler_module, + application: __MODULE__, + consistency: :eventual, + last_seen_event: 0 + } + end + + defp attach_telemetry do + :telemetry.attach_many( + "test-handler", + [ + [:commanded, :event, :handle, :start], + [:commanded, :event, :handle, :stop], + [:commanded, :event, :handle, :exception], + [:commanded, :event, :batch, :start], + [:commanded, :event, :batch, :stop], + [:commanded, :event, :batch, :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 + + defmodule MockAdapter do + def ack_event(nil, subscription_pid, event) do + send(subscription_pid, {:acked, event}) + :ok + end + end +end