From 10c250a01ba06dc68f6dcebfa425239a04be8c69 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 13 Mar 2026 15:13:25 -0400 Subject: [PATCH] feat: track wrong_expected_version count in aggregate and telemetry metadata Signed-off-by: Yordis Prieto --- guides/explanations/fork-differences.md | 19 +- lib/commanded/aggregates/aggregate.ex | 18 +- lib/commanded/opentelemetry/aggregate.ex | 11 + .../opentelemetry/commanded_attributes.ex | 6 + test/aggregates/aggregate_telemetry_test.exs | 5 + test/commands/correlation_causation_test.exs | 8 +- test/opentelemetry/aggregate_test.exs | 249 +++++++++++++++++- 7 files changed, 294 insertions(+), 22 deletions(-) diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index 51982fc8..653b23ab 100644 --- a/guides/explanations/fork-differences.md +++ b/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), [#58](https://github.com/straw-hat-team/commanded/pull/58) +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), [#60](https://github.com/straw-hat-team/commanded/pull/60), [#61](https://github.com/straw-hat-team/commanded/pull/61) **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), aggregate load (PR #58), 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), aggregate populate (PR #47), aggregate snapshots (PR #60), and wrong_expected_version span event (PR #61) - Added `opentelemetry_api`, `opentelemetry_telemetry`, and `opentelemetry_semantic_conventions` as required dependencies **Usage:** @@ -257,3 +257,18 @@ end **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 + +### **OpenTelemetry Aggregate Snapshots** +[PR #60](https://github.com/straw-hat-team/commanded/pull/60) + +**Changes:** +- Added OTel spans for aggregate snapshot operations (`commanded.aggregate.snapshot`) +- Spans fire when taking snapshots during aggregate execution + +### **OpenTelemetry wrong_expected_version Span Event** +[PR #61](https://github.com/straw-hat-team/commanded/pull/61) + +**Changes:** +- 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 diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index 33cd7334..ad0ed416 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -29,7 +29,8 @@ defmodule Commanded.Aggregates.Aggregate do caller: pid(), execution_context: Commanded.Aggregates.ExecutionContext.t(), events: [map()], - error: nil | any()} + error: nil | any(), + wrong_expected_version_count: non_neg_integer()} """ }) @@ -194,7 +195,8 @@ defmodule Commanded.Aggregates.Aggregate do :aggregate_state, :snapshotting, aggregate_version: 0, - lifespan_timeout: :infinity + lifespan_timeout: :infinity, + wrong_expected_version_count: 0 ] def start_link(config, opts) do @@ -372,6 +374,7 @@ defmodule Commanded.Aggregates.Aggregate do def handle_call({:execute_command, %ExecutionContext{} = context}, from, %Aggregate{} = state) do %ExecutionContext{lifespan: lifespan, command: command} = context + state = %Aggregate{state | wrong_expected_version_count: 0} telemetry_metadata = telemetry_metadata(context, from, state) start_time = telemetry_start(telemetry_metadata) @@ -623,6 +626,11 @@ defmodule Commanded.Aggregates.Aggregate do {:error, :wrong_expected_version} -> telemetry_wrong_expected_version(context, from, state) + state = %Aggregate{ + state + | wrong_expected_version_count: state.wrong_expected_version_count + 1 + } + # Fetch missing events from event store state = AggregateStateBuilder.rebuild_from_events(state) @@ -750,7 +758,8 @@ defmodule Commanded.Aggregates.Aggregate do application: application, aggregate_uuid: aggregate_uuid, aggregate_state: aggregate_state, - aggregate_version: aggregate_version + aggregate_version: aggregate_version, + wrong_expected_version_count: wrong_expected_version_count } = state {pid, _ref} = from @@ -761,7 +770,8 @@ defmodule Commanded.Aggregates.Aggregate do aggregate_state: aggregate_state, aggregate_version: aggregate_version, caller: pid, - execution_context: context + execution_context: context, + wrong_expected_version_count: wrong_expected_version_count } end diff --git a/lib/commanded/opentelemetry/aggregate.ex b/lib/commanded/opentelemetry/aggregate.ex index a630b86c..9a44880a 100644 --- a/lib/commanded/opentelemetry/aggregate.ex +++ b/lib/commanded/opentelemetry/aggregate.ex @@ -96,6 +96,17 @@ defmodule Commanded.OpenTelemetry.Aggregate do Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error))) end + wrong_expected_version_count = Map.get(meta, :wrong_expected_version_count, 0) + + if wrong_expected_version_count > 0 do + Span.add_event(ctx, "commanded.aggregate.wrong_expected_version", [ + {CommandedAttributes.commanded_aggregate_version(), meta.aggregate_version}, + {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid}, + {CommandedAttributes.commanded_wrong_expected_version_count(), + wrong_expected_version_count} + ]) + end + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) end diff --git a/lib/commanded/opentelemetry/commanded_attributes.ex b/lib/commanded/opentelemetry/commanded_attributes.ex index b841691b..c0c48aae 100644 --- a/lib/commanded/opentelemetry/commanded_attributes.ex +++ b/lib/commanded/opentelemetry/commanded_attributes.ex @@ -165,4 +165,10 @@ defmodule Commanded.OpenTelemetry.CommandedAttributes do """ @spec commanded_snapshot_module_version() :: :"commanded.snapshot.module_version" def commanded_snapshot_module_version, do: :"commanded.snapshot.module_version" + + @doc """ + Number of wrong_expected_version conflicts during a single command execution. + """ + @spec commanded_wrong_expected_version_count() :: :"commanded.wrong_expected_version.count" + def commanded_wrong_expected_version_count, do: :"commanded.wrong_expected_version.count" end diff --git a/test/aggregates/aggregate_telemetry_test.exs b/test/aggregates/aggregate_telemetry_test.exs index 8660883a..dded409a 100644 --- a/test/aggregates/aggregate_telemetry_test.exs +++ b/test/aggregates/aggregate_telemetry_test.exs @@ -301,6 +301,11 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do assert metadata.caller == self() assert metadata.execution_context.command == %Ok{message: "ok"} assert metadata.execution_context.retry_attempts == 0 + + assert_receive {[:commanded, :aggregate, :execute, :stop], _stop_measurements, + stop_metadata} + + assert stop_metadata.wrong_expected_version_count == 1 end end diff --git a/test/commands/correlation_causation_test.exs b/test/commands/correlation_causation_test.exs index 5a92f0bd..95055db2 100644 --- a/test/commands/correlation_causation_test.exs +++ b/test/commands/correlation_causation_test.exs @@ -10,7 +10,6 @@ defmodule Commanded.Commands.CorrelationCasuationTest do alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited alias Commanded.ExampleDomain.BankRouter alias Commanded.Helpers.CommandAuditMiddleware - alias Commanded.Helpers.ProcessHelper alias Commanded.UUID setup do @@ -146,11 +145,8 @@ defmodule Commanded.Commands.CorrelationCasuationTest do end def start_account_bonus_handler(_context) do - {:ok, handler} = OpenAccountBonusHandler.start_link() - - on_exit(fn -> - ProcessHelper.shutdown(handler) - end) + _handler = start_supervised!(OpenAccountBonusHandler) + :ok end end end diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs index de419a26..8d0fe54d 100644 --- a/test/opentelemetry/aggregate_test.exs +++ b/test/opentelemetry/aggregate_test.exs @@ -1,10 +1,15 @@ defmodule Commanded.OpenTelemetry.AggregateTest do use Commanded.OpenTelemetryCase, async: false + use Commanded.MockEventStoreCase + alias Commanded.Aggregates.AggregateTelemetryTest + alias Commanded.Aggregates.{Aggregate, ExecutionContext} alias Commanded.DefaultApp + alias Commanded.EventStore.Adapters.Mock, as: MockEventStore alias Commanded.Middleware.Commands.IncrementCount alias Commanded.Middleware.Commands.RaiseError - alias Commanded.OpenTelemetry.Aggregate + alias Commanded.MockedApp + alias Commanded.OpenTelemetry.Aggregate, as: OTelAggregate alias Commanded.OpenTelemetry.TestRouter alias Commanded.TestSupport.Factory alias Commanded.UUID @@ -18,7 +23,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do detach_handlers() detach_event_store_handlers() - Aggregate.setup() + OTelAggregate.setup() :ok end @@ -27,7 +32,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do test "attaches telemetry handlers for aggregate execute events" do detach_handlers() - Aggregate.setup() + OTelAggregate.setup() for event <- [ [:commanded, :aggregate, :execute, :start], @@ -38,7 +43,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do assert Enum.any?( handlers, - &match?(%{id: {Aggregate, :execute}}, &1) + &match?(%{id: {OTelAggregate, :execute}}, &1) ), "Expected handler for event #{inspect(event)}" end @@ -47,13 +52,13 @@ defmodule Commanded.OpenTelemetry.AggregateTest do test "calling setup twice raises MatchError (fail fast)" do detach_handlers() - :ok = Aggregate.setup() + :ok = OTelAggregate.setup() handlers = :telemetry.list_handlers([:commanded, :aggregate, :execute, :start]) assert length(handlers) == 1 assert_raise MatchError, fn -> - Aggregate.setup() + OTelAggregate.setup() end end end @@ -61,7 +66,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do describe "attribute completeness" do setup do detach_handlers() - Aggregate.setup() + OTelAggregate.setup() :ok end @@ -260,7 +265,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do describe "error handling" do setup do detach_handlers() - Aggregate.setup() + OTelAggregate.setup() :ok end @@ -409,7 +414,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do describe "trace context propagation" do setup do detach_handlers() - Aggregate.setup() + OTelAggregate.setup() :ok end @@ -537,7 +542,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do describe "edge cases" do setup do detach_handlers() - Aggregate.setup() + OTelAggregate.setup() :ok end @@ -597,6 +602,230 @@ defmodule Commanded.OpenTelemetry.AggregateTest do end end + describe "wrong_expected_version span event" do + test "span has wrong_expected_version event when conflict occurs" do + aggregate_uuid = UUID.uuid4() + + expect(MockEventStore, :subscribe, fn _event_store_meta, ^aggregate_uuid -> + assert is_binary(aggregate_uuid) + :ok + end) + + expect(MockEventStore, :append_to_stream, fn _meta, + ^aggregate_uuid, + _exp_ver, + _event_data, + _opts -> + {:error, :wrong_expected_version} + end) + + expect(MockEventStore, :stream_forward, 2, fn _meta, ^aggregate_uuid, _from, _batch_size -> + [] + end) + + assert {:ok, _pid} = start_aggregate(aggregate_uuid, application: MockedApp) + + assert {:error, :too_many_attempts} = + Aggregate.execute( + MockedApp, + AggregateTelemetryTest.ExampleAggregate, + aggregate_uuid, + %ExecutionContext{ + command: + struct!(Commanded.Aggregates.AggregateTelemetryTest.Commands.Ok, + message: "ok" + ), + function: :execute, + handler: AggregateTelemetryTest.ExampleAggregate, + retry_attempts: 0 + } + ) + + assert_receive {:span, + span( + name: + "execute Commanded.Aggregates.AggregateTelemetryTest.ExampleAggregate", + kind: :consumer, + events: events + )}, + 1000 + + events_list = :otel_events.list(events) + + wrong_version_event = + Enum.find(events_list, fn event -> + elem(event, 2) == "commanded.aggregate.wrong_expected_version" + end) + + assert wrong_version_event != nil, + "Expected span to contain commanded.aggregate.wrong_expected_version event, got: #{inspect(events_list)}" + + {:event, _timestamp, "commanded.aggregate.wrong_expected_version", attrs_tuple} = + wrong_version_event + + {:attributes, _, _, _, attrs_map} = attrs_tuple + + assert attrs_map[:"commanded.aggregate.uuid"] == aggregate_uuid + assert attrs_map[:"commanded.aggregate.version"] == 0 + assert attrs_map[:"commanded.wrong_expected_version.count"] == 1 + end + + test "span has wrong_expected_version event with count when retry succeeds" do + aggregate_uuid = UUID.uuid4() + + expect(MockEventStore, :subscribe, fn _event_store_meta, ^aggregate_uuid -> + assert is_binary(aggregate_uuid) + :ok + end) + + expect(MockEventStore, :append_to_stream, 2, fn + _meta, ^aggregate_uuid, 0, _event_data, _opts -> + {:error, :wrong_expected_version} + + _meta, ^aggregate_uuid, 1, _event_data, _opts -> + :ok + end) + + stream_forward_calls = :counters.new(1, []) + + expect(MockEventStore, :stream_forward, 2, fn _meta, ^aggregate_uuid, 1, _batch_size -> + n = :counters.get(stream_forward_calls, 1) + :counters.add(stream_forward_calls, 1, 1) + + if n == 0 do + [] + else + [ + %Commanded.EventStore.RecordedEvent{ + event_id: UUID.uuid4(), + event_number: 1, + stream_id: aggregate_uuid, + stream_version: 1, + correlation_id: nil, + causation_id: nil, + event_type: "Elixir.Commanded.Aggregates.AggregateTelemetryTest.Event", + data: struct!(Commanded.Aggregates.AggregateTelemetryTest.Event, message: "event"), + metadata: nil, + created_at: DateTime.utc_now() + } + ] + end + end) + + assert {:ok, _pid} = start_aggregate(aggregate_uuid, application: MockedApp) + + assert {:ok, 2, _events, _aggregate_state} = + Aggregate.execute( + MockedApp, + AggregateTelemetryTest.ExampleAggregate, + aggregate_uuid, + %ExecutionContext{ + command: + struct!(Commanded.Aggregates.AggregateTelemetryTest.Commands.Ok, + message: "ok" + ), + function: :execute, + handler: AggregateTelemetryTest.ExampleAggregate, + retry_attempts: 1 + } + ) + + assert_receive {:span, + span( + name: + "execute Commanded.Aggregates.AggregateTelemetryTest.ExampleAggregate", + kind: :consumer, + events: events + )}, + 1000 + + events_list = :otel_events.list(events) + + wrong_version_event = + Enum.find(events_list, fn event -> + elem(event, 2) == "commanded.aggregate.wrong_expected_version" + end) + + assert wrong_version_event != nil, + "Expected span to contain commanded.aggregate.wrong_expected_version event, got: #{inspect(events_list)}" + + {:event, _timestamp, "commanded.aggregate.wrong_expected_version", attrs_tuple} = + wrong_version_event + + {:attributes, _, _, _, attrs_map} = attrs_tuple + + assert attrs_map[:"commanded.wrong_expected_version.count"] == 1 + end + + test "no wrong_expected_version event when count is 0" do + aggregate_uuid = UUID.uuid4() + + expect(MockEventStore, :subscribe, fn _event_store_meta, ^aggregate_uuid -> + :ok + end) + + expect(MockEventStore, :append_to_stream, fn _meta, + ^aggregate_uuid, + _exp_ver, + _event_data, + _opts -> + :ok + end) + + expect(MockEventStore, :stream_forward, fn _meta, ^aggregate_uuid, _from, _batch_size -> + [] + end) + + assert {:ok, _pid} = start_aggregate(aggregate_uuid, application: MockedApp) + + assert {:ok, 1, _events, _aggregate_state} = + Aggregate.execute( + MockedApp, + AggregateTelemetryTest.ExampleAggregate, + aggregate_uuid, + %ExecutionContext{ + command: + struct!(Commanded.Aggregates.AggregateTelemetryTest.Commands.Ok, + message: "ok" + ), + function: :execute, + handler: AggregateTelemetryTest.ExampleAggregate + } + ) + + assert_receive {:span, + span( + name: + "execute Commanded.Aggregates.AggregateTelemetryTest.ExampleAggregate", + kind: :consumer, + events: events + )}, + 1000 + + events_list = :otel_events.list(events) + + wrong_version_event = + Enum.find(events_list, fn event -> + elem(event, 2) == "commanded.aggregate.wrong_expected_version" + end) + + assert wrong_version_event == nil, + "Expected no commanded.aggregate.wrong_expected_version event when count is 0, got: #{inspect(events_list)}" + end + end + + defp start_aggregate(aggregate_uuid, opts) do + aggregate = AggregateTelemetryTest.ExampleAggregate + app = Keyword.fetch!(opts, :application) + name = Aggregate.name(app, aggregate, aggregate_uuid) + + Aggregate.start_link([application: app], + aggregate_module: aggregate, + aggregate_uuid: aggregate_uuid, + name: Commanded.Registration.via_tuple(app, name) + ) + end + defp encode_traceparent(span_ctx) do trace_id = :otel_span.trace_id(span_ctx) span_id = :otel_span.span_id(span_ctx)