From e95b7efb45f9be91384ba8a9a7b1e1e4dae107e3 Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 20 Apr 2026 20:36:11 -0400
Subject: [PATCH] feat(telemetry): add EventStore adapter-level OTel
 instrumentation

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/commanded/opentelemetry.ex                |  19 +-
 .../opentelemetry/commanded_attributes.ex     |  30 ++
 lib/commanded/opentelemetry/event_store.ex    |   7 +-
 .../event_store/adapters/event_store.ex       | 201 +++++++++
 test/opentelemetry/aggregate_test.exs         |   5 -
 .../event_store/adapters/event_store_test.exs | 398 ++++++++++++++++++
 test/support/opentelemetry_case.ex            |  10 +-
 7 files changed, 661 insertions(+), 9 deletions(-)
 create mode 100644 lib/commanded/opentelemetry/event_store/adapters/event_store.ex
 create mode 100644 test/opentelemetry/event_store/adapters/event_store_test.exs

diff --git a/lib/commanded/opentelemetry.ex b/lib/commanded/opentelemetry.ex
index ba66abb2..89f0c366 100644
--- a/lib/commanded/opentelemetry.ex
+++ b/lib/commanded/opentelemetry.ex
@@ -126,7 +126,19 @@ defmodule Commanded.OpenTelemetry do
                      doc: "Event handler tracing configuration. Use `:disabled` to disable."
                    ],
                    event_store: [
-                     type: {:in, [:disabled, []]},
+                     type:
+                       {:or,
+                        [
+                          {:in, [:disabled]},
+                          keyword_list: [
+                            adapter: [
+                              type: {:in, [:enabled, :disabled]},
+                              default: :disabled,
+                              doc:
+                                "Hook into the telemetry events emitted by the event store adapter. Use `:enabled` to enable."
+                            ]
+                          ]
+                        ]},
                      default: [],
                      doc: "Event store tracing configuration. Use `:disabled` to disable."
                    ]
@@ -174,6 +186,9 @@ defmodule Commanded.OpenTelemetry do
       # Disable event store tracing
       Commanded.OpenTelemetry.setup(event_store: :disabled)
 
+      # Enable event store adapter tracing (hooks into adapter-level telemetry)
+      Commanded.OpenTelemetry.setup(event_store: [adapter: :enabled])
+
       # Use parent-child relationships for event handlers
       Commanded.OpenTelemetry.setup(event_handler: [span_relationship: :child])
 
@@ -209,7 +224,7 @@ defmodule Commanded.OpenTelemetry do
 
     case opts[:event_store] do
       :disabled -> :ok
-      _config -> EventStore.setup()
+      config -> EventStore.setup(config)
     end
 
     :ok
diff --git a/lib/commanded/opentelemetry/commanded_attributes.ex b/lib/commanded/opentelemetry/commanded_attributes.ex
index 9fa5b9fd..0176df53 100644
--- a/lib/commanded/opentelemetry/commanded_attributes.ex
+++ b/lib/commanded/opentelemetry/commanded_attributes.ex
@@ -240,4 +240,34 @@ defmodule Commanded.OpenTelemetry.CommandedAttributes do
   """
   @spec commanded_stream_batch_size() :: :"commanded.stream.batch_size"
   def commanded_stream_batch_size, do: :"commanded.stream.batch_size"
+
+  @doc """
+  Number of events read from a stream.
+  """
+  @spec eventstore_read_count() :: :"eventstore.read.count"
+  def eventstore_read_count, do: :"eventstore.read.count"
+
+  @doc """
+  The version number to start reading a stream from (EventStore adapter).
+  """
+  @spec eventstore_stream_start_version() :: :"eventstore.stream.start_version"
+  def eventstore_stream_start_version, do: :"eventstore.stream.start_version"
+
+  @doc """
+  The direction of a stream read (:forward or :backward).
+  """
+  @spec eventstore_stream_direction() :: :"eventstore.stream.direction"
+  def eventstore_stream_direction, do: :"eventstore.stream.direction"
+
+  @doc """
+  The batch size used when reading events from a stream (EventStore adapter).
+  """
+  @spec eventstore_stream_batch_size() :: :"eventstore.stream.batch_size"
+  def eventstore_stream_batch_size, do: :"eventstore.stream.batch_size"
+
+  @doc """
+  The type of stream deletion (soft or hard).
+  """
+  @spec eventstore_stream_delete_type() :: :"eventstore.stream.delete_type"
+  def eventstore_stream_delete_type, do: :"eventstore.stream.delete_type"
 end
diff --git a/lib/commanded/opentelemetry/event_store.ex b/lib/commanded/opentelemetry/event_store.ex
index a5d56c2e..c112fb46 100644
--- a/lib/commanded/opentelemetry/event_store.ex
+++ b/lib/commanded/opentelemetry/event_store.ex
@@ -3,6 +3,7 @@ defmodule Commanded.OpenTelemetry.EventStore do
 
   alias Commanded.Application, as: CommandedApplication
   alias Commanded.OpenTelemetry.CommandedAttributes
+  alias Commanded.OpenTelemetry.EventStore.Adapters
   alias Commanded.OpenTelemetry.Helpers
   alias OpenTelemetry.SemConv.ErrorAttributes
   alias OpenTelemetry.SemConv.Incubating.CodeAttributes
@@ -20,7 +21,7 @@ defmodule Commanded.OpenTelemetry.EventStore do
     stream_forward
   )a
 
-  def setup do
+  def setup(config \\ []) do
     for event <- @events do
       :ok =
         :telemetry.attach_many(
@@ -35,6 +36,10 @@ defmodule Commanded.OpenTelemetry.EventStore do
         )
     end
 
+    if config[:adapter] == :enabled do
+      Adapters.EventStore.setup()
+    end
+
     :ok
   end
 
diff --git a/lib/commanded/opentelemetry/event_store/adapters/event_store.ex b/lib/commanded/opentelemetry/event_store/adapters/event_store.ex
new file mode 100644
index 00000000..3af7622f
--- /dev/null
+++ b/lib/commanded/opentelemetry/event_store/adapters/event_store.ex
@@ -0,0 +1,201 @@
+defmodule Commanded.OpenTelemetry.EventStore.Adapters.EventStore do
+  @moduledoc false
+
+  alias Commanded.OpenTelemetry.CommandedAttributes
+  alias Commanded.OpenTelemetry.Helpers
+  alias OpenTelemetry.SemConv.ErrorAttributes
+  alias OpenTelemetry.SemConv.Incubating.CodeAttributes
+  alias OpenTelemetry.SemConv.Incubating.DBAttributes
+  alias OpenTelemetry.SemConv.Incubating.MessagingAttributes
+  alias OpenTelemetry.Span
+
+
+  @tracer_id __MODULE__
+
+  @events ~w(
+    delete_stream
+    delete_subscription
+    link_to_stream
+    paginate_streams
+    read_stream_backward
+    read_stream_forward
+    stream_batch_read
+  )a
+
+  def setup do
+    for event <- @events do
+      :ok =
+        :telemetry.attach_many(
+          {__MODULE__, event},
+          [
+            [:eventstore, event, :start],
+            [:eventstore, event, :stop],
+            [:eventstore, event, :exception]
+          ],
+          &__MODULE__.handle_telemetry_event/4,
+          %{}
+        )
+    end
+
+    :ok
+  end
+
+  def handle_telemetry_event(
+        [:eventstore, action, :start],
+        _measurements,
+        meta,
+        _config
+      ) do
+    operation_type = operation_type_for(action)
+    action_name = to_string(action)
+    destination_name = event_store_destination_name(meta)
+
+    conn_config = resolve_connection_config(meta)
+
+    attributes =
+      [
+        {MessagingAttributes.messaging_system(), "eventstore"},
+        {MessagingAttributes.messaging_operation_name(), action_name},
+        {CodeAttributes.code_function(), action_name},
+        {DBAttributes.db_system(), :postgresql}
+      ]
+      |> Helpers.maybe_add_connection_attributes(conn_config)
+      |> Helpers.maybe_add_operation_type(operation_type)
+      |> Helpers.maybe_add_destination_name(destination_name)
+      |> Helpers.maybe_add_stream_uuid(meta[:stream_uuid])
+      |> Helpers.maybe_add_expected_version(meta[:expected_version])
+      |> Helpers.maybe_add_event_count(meta[:event_count])
+      |> Helpers.maybe_add_subscription_name(meta[:subscription_name])
+      |> Helpers.maybe_add_source_uuid(meta[:source_uuid])
+      |> maybe_add_count(meta[:count])
+      |> maybe_add_start_version(meta[:start_version])
+      |> maybe_add_direction(meta[:direction])
+      |> maybe_add_batch_size(meta[:requested_batch_size])
+      |> maybe_add_delete_type(meta[:delete_type])
+
+    span_name =
+      case destination_name do
+        nil -> action_name
+        name -> "#{action_name} #{name}"
+      end
+
+    OpentelemetryTelemetry.start_telemetry_span(
+      @tracer_id,
+      span_name,
+      meta,
+      %{
+        kind: :client,
+        attributes: attributes
+      }
+    )
+  end
+
+  def handle_telemetry_event(
+        [:eventstore, _action, :stop],
+        _measurements,
+        meta,
+        _config
+      ) do
+    ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta)
+
+    maybe_set_stop_event_count(ctx, meta)
+
+    case meta[:result] do
+      {:error, reason} ->
+        Span.set_attribute(
+          ctx,
+          ErrorAttributes.error_type(),
+          Helpers.to_error_type(reason, @tracer_id)
+        )
+
+        Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(reason)))
+
+      _ ->
+        :ok
+    end
+
+    OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta)
+  end
+
+  def handle_telemetry_event(
+        [:eventstore, _action, :exception],
+        _measurements,
+        %{kind: kind, reason: reason, stacktrace: stacktrace} = meta,
+        _config
+      ) do
+    ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta)
+
+    Span.set_attribute(ctx, :"erlang.exception.kind", kind)
+
+    exception = Exception.normalize(kind, reason, stacktrace)
+
+    Span.set_attribute(
+      ctx,
+      ErrorAttributes.error_type(),
+      Helpers.to_error_type(exception, @tracer_id)
+    )
+
+    Span.record_exception(ctx, exception, stacktrace)
+
+    Span.set_status(
+      ctx,
+      OpenTelemetry.status(:error, Exception.format_banner(kind, reason, stacktrace))
+    )
+
+    OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta)
+  end
+
+  defp operation_type_for(:link_to_stream), do: :publish
+  defp operation_type_for(:read_stream_forward), do: :receive
+  defp operation_type_for(:read_stream_backward), do: :receive
+  defp operation_type_for(:stream_batch_read), do: :receive
+  defp operation_type_for(:delete_stream), do: nil
+  defp operation_type_for(:delete_subscription), do: nil
+  defp operation_type_for(:paginate_streams), do: nil
+  defp operation_type_for(_), do: nil
+
+  defp event_store_destination_name(meta) do
+    name = meta[:name] || meta[:event_store]
+    Helpers.to_destination_name(name)
+  end
+
+  defp resolve_connection_config(meta) do
+    name = meta[:name] || meta[:event_store]
+
+    EventStore.Config.lookup(name)
+  rescue
+    _ -> []
+  end
+
+  defp maybe_add_count(attrs, nil), do: attrs
+
+  defp maybe_add_count(attrs, count),
+    do: [{CommandedAttributes.eventstore_read_count(), count} | attrs]
+
+  defp maybe_add_start_version(attrs, nil), do: attrs
+
+  defp maybe_add_start_version(attrs, version),
+    do: [{CommandedAttributes.eventstore_stream_start_version(), version} | attrs]
+
+  defp maybe_add_direction(attrs, nil), do: attrs
+
+  defp maybe_add_direction(attrs, direction),
+    do: [{CommandedAttributes.eventstore_stream_direction(), direction} | attrs]
+
+  defp maybe_add_batch_size(attrs, nil), do: attrs
+
+  defp maybe_add_batch_size(attrs, size),
+    do: [{CommandedAttributes.eventstore_stream_batch_size(), size} | attrs]
+
+  defp maybe_add_delete_type(attrs, nil), do: attrs
+
+  defp maybe_add_delete_type(attrs, type),
+    do: [{CommandedAttributes.eventstore_stream_delete_type(), type} | attrs]
+
+  # stream_batch_read includes event_count only in stop metadata
+  defp maybe_set_stop_event_count(ctx, %{event_count: count}) when is_integer(count) do
+    Span.set_attribute(ctx, CommandedAttributes.commanded_event_count(), count)
+  end
+
+  defp maybe_set_stop_event_count(_ctx, _meta), do: :ok
+end
diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs
index 8e8f19a3..5669f3b3 100644
--- a/test/opentelemetry/aggregate_test.exs
+++ b/test/opentelemetry/aggregate_test.exs
@@ -860,16 +860,11 @@ defmodule Commanded.OpenTelemetry.AggregateTest do
 
   defp detach_event_store_handlers do
     event_store_events = ~w(
-      ack_event
       append_to_stream
       delete_snapshot
-      delete_subscription
       read_snapshot
       record_snapshot
       stream_forward
-      subscribe
-      subscribe_to
-      unsubscribe
     )a
 
     for event <- event_store_events,
diff --git a/test/opentelemetry/event_store/adapters/event_store_test.exs b/test/opentelemetry/event_store/adapters/event_store_test.exs
new file mode 100644
index 00000000..ed51a46e
--- /dev/null
+++ b/test/opentelemetry/event_store/adapters/event_store_test.exs
@@ -0,0 +1,398 @@
+defmodule Commanded.OpenTelemetry.EventStore.Adapters.EventStoreTest do
+  @moduledoc """
+  Tests for the EventStore library OpenTelemetry instrumentation.
+
+  Hooks into [:eventstore, operation, suffix] telemetry events emitted
+  by the eventstore library itself.
+  """
+
+  use Commanded.OpenTelemetryCase, async: false
+
+  alias Commanded.OpenTelemetry.EventStore.Adapters.EventStore, as: OTelEventstoreAdapter
+
+  @events ~w(
+    delete_stream
+    delete_subscription
+    link_to_stream
+    paginate_streams
+    read_stream_backward
+    read_stream_forward
+    stream_batch_read
+  )a
+
+  setup do
+    detach_handlers()
+    OTelEventstoreAdapter.setup()
+
+    :ok
+  end
+
+  describe "setup/0" do
+    test "attaches telemetry handlers for all eventstore operations" do
+      detach_handlers()
+
+      OTelEventstoreAdapter.setup()
+
+      for event <- @events do
+        for suffix <- [:start, :stop, :exception] do
+          handlers = :telemetry.list_handlers([:eventstore, event, suffix])
+
+          assert Enum.any?(
+                   handlers,
+                   &match?(%{id: {OTelEventstoreAdapter, ^event}}, &1)
+                 ),
+                 "Expected handler for event [:eventstore, #{event}, #{suffix}]"
+        end
+      end
+    end
+
+    test "calling setup twice raises MatchError (fail fast)" do
+      detach_handlers()
+
+      :ok = OTelEventstoreAdapter.setup()
+
+      assert_raise MatchError, fn ->
+        OTelEventstoreAdapter.setup()
+      end
+    end
+  end
+
+  describe "link_to_stream" do
+    test "creates span with correct attributes" do
+      meta =
+        emit_start(:link_to_stream, %{
+          event_store: TestEventStore,
+          stream_uuid: "target-stream",
+          expected_version: 0,
+          event_count: 2
+        })
+
+      emit_stop(:link_to_stream, Map.put(meta, :result, :ok))
+
+      assert span(kind: :client, attributes: attributes) =
+               assert_receive_span_named("link_to_stream TestEventStore")
+
+      attrs = :otel_attributes.map(attributes)
+      assert attrs[:"messaging.system"] == "eventstore"
+      assert attrs[:"messaging.operation.type"] == :publish
+      assert attrs[:"commanded.stream.uuid"] == "target-stream"
+      assert attrs[:"commanded.event.count"] == 2
+    end
+  end
+
+  describe "read_stream_forward" do
+    test "creates span with correct attributes" do
+      meta =
+        emit_start(:read_stream_forward, %{
+          event_store: TestEventStore,
+          stream_uuid: "stream-123",
+          count: 100,
+          start_version: 0
+        })
+
+      emit_stop(:read_stream_forward, Map.put(meta, :result, :ok))
+
+      assert span(kind: :client, attributes: attributes) =
+               assert_receive_span_named("read_stream_forward TestEventStore")
+
+      assert :otel_attributes.map(attributes) == %{
+               "messaging.system": "eventstore",
+               "messaging.operation.type": :receive,
+               "messaging.operation.name": "read_stream_forward",
+               "messaging.destination.name": "TestEventStore",
+               "code.function": "read_stream_forward",
+               "commanded.stream.uuid": "stream-123",
+               "eventstore.read.count": 100,
+               "eventstore.stream.start_version": 0,
+               "db.system": :postgresql
+             }
+    end
+  end
+
+  describe "read_stream_backward" do
+    test "creates span with correct attributes" do
+      meta =
+        emit_start(:read_stream_backward, %{
+          event_store: TestEventStore,
+          stream_uuid: "stream-123",
+          count: 50,
+          start_version: -1
+        })
+
+      emit_stop(:read_stream_backward, Map.put(meta, :result, :ok))
+
+      assert span(kind: :client, attributes: attributes) =
+               assert_receive_span_named("read_stream_backward TestEventStore")
+
+      attrs = :otel_attributes.map(attributes)
+      assert attrs[:"messaging.operation.type"] == :receive
+      assert attrs[:"eventstore.read.count"] == 50
+      assert attrs[:"eventstore.stream.start_version"] == -1
+    end
+  end
+
+  describe "delete_stream" do
+    test "creates span with correct attributes" do
+      meta =
+        emit_start(:delete_stream, %{
+          event_store: TestEventStore,
+          stream_uuid: "stream-123",
+          expected_version: :stream_exists,
+          delete_type: :soft
+        })
+
+      emit_stop(:delete_stream, Map.put(meta, :result, :ok))
+
+      assert span(kind: :client, attributes: attributes) =
+               assert_receive_span_named("delete_stream TestEventStore")
+
+      assert :otel_attributes.map(attributes) == %{
+               "messaging.system": "eventstore",
+               "messaging.operation.name": "delete_stream",
+               "messaging.destination.name": "TestEventStore",
+               "code.function": "delete_stream",
+               "commanded.stream.uuid": "stream-123",
+               "commanded.expected_version": :stream_exists,
+               "eventstore.stream.delete_type": :soft,
+               "db.system": :postgresql
+             }
+    end
+  end
+
+  describe "delete_subscription" do
+    test "creates span with correct attributes" do
+      meta =
+        emit_start(:delete_subscription, %{
+          event_store: TestEventStore,
+          stream_uuid: "stream-123",
+          subscription_name: "my-subscription"
+        })
+
+      emit_stop(:delete_subscription, Map.put(meta, :result, :ok))
+
+      assert span(kind: :client, attributes: attributes) =
+               assert_receive_span_named("delete_subscription TestEventStore")
+
+      attrs = :otel_attributes.map(attributes)
+      refute Map.has_key?(attrs, :"messaging.operation.type")
+      assert attrs[:"commanded.subscription.name"] == "my-subscription"
+    end
+  end
+
+  describe "paginate_streams" do
+    test "creates span with correct attributes" do
+      meta = emit_start(:paginate_streams, %{event_store: TestEventStore})
+      emit_stop(:paginate_streams, Map.put(meta, :result, :ok))
+
+      assert span(kind: :client, attributes: attributes) =
+               assert_receive_span_named("paginate_streams TestEventStore")
+
+      assert :otel_attributes.map(attributes) == %{
+               "messaging.system": "eventstore",
+               "messaging.operation.name": "paginate_streams",
+               "messaging.destination.name": "TestEventStore",
+               "code.function": "paginate_streams",
+               "db.system": :postgresql
+             }
+    end
+  end
+
+  describe "stream_batch_read" do
+    test "creates span with direction and batch attributes" do
+      meta =
+        emit_start(:stream_batch_read, %{
+          event_store: TestEventStore,
+          stream_uuid: "stream-123",
+          direction: :forward,
+          requested_batch_size: 100,
+          start_version: 1
+        })
+
+      emit_stop(:stream_batch_read, Map.merge(meta, %{result: :ok, event_count: 50}))
+
+      assert span(kind: :client, attributes: attributes) =
+               assert_receive_span_named("stream_batch_read TestEventStore")
+
+      assert :otel_attributes.map(attributes) == %{
+               "messaging.system": "eventstore",
+               "messaging.operation.type": :receive,
+               "messaging.operation.name": "stream_batch_read",
+               "messaging.destination.name": "TestEventStore",
+               "code.function": "stream_batch_read",
+               "commanded.stream.uuid": "stream-123",
+               "commanded.event.count": 50,
+               "eventstore.stream.direction": :forward,
+               "eventstore.stream.batch_size": 100,
+               "eventstore.stream.start_version": 1,
+               "db.system": :postgresql
+             }
+    end
+  end
+
+  describe "destination name" do
+    test "uses event_store module when name is not present" do
+      meta = emit_start(:link_to_stream, %{event_store: MyApp.EventStore, stream_uuid: "s1"})
+      emit_stop(:link_to_stream, Map.put(meta, :result, :ok))
+
+      assert span(attributes: attributes) =
+               assert_receive_span_named("link_to_stream MyApp.EventStore")
+
+      assert :otel_attributes.map(attributes)[:"messaging.destination.name"] ==
+               "MyApp.EventStore"
+    end
+
+    test "prefers name over event_store module" do
+      meta =
+        emit_start(:link_to_stream, %{
+          event_store: MyApp.EventStore,
+          name: :my_named_store,
+          stream_uuid: "s1"
+        })
+
+      emit_stop(:link_to_stream, Map.put(meta, :result, :ok))
+
+      assert span(attributes: attributes) =
+               assert_receive_span_named("link_to_stream :my_named_store")
+
+      assert :otel_attributes.map(attributes)[:"messaging.destination.name"] == ":my_named_store"
+    end
+
+    test "omits destination when neither name nor event_store is present" do
+      meta = emit_start(:link_to_stream, %{stream_uuid: "s1"})
+      emit_stop(:link_to_stream, Map.put(meta, :result, :ok))
+
+      assert span(name: "link_to_stream", attributes: attributes) =
+               assert_receive_span_named("link_to_stream")
+
+      refute Map.has_key?(:otel_attributes.map(attributes), :"messaging.destination.name")
+    end
+  end
+
+  describe "exception handling" do
+    test "sets error status and records exception" do
+      meta =
+        emit_start(:read_stream_forward, %{
+          event_store: TestEventStore,
+          stream_uuid: "stream-123"
+        })
+
+      emit_exception(:read_stream_forward, meta, %{
+        kind: :error,
+        reason: %RuntimeError{message: "connection lost"},
+        stacktrace: []
+      })
+
+      assert span(
+               status: {:status, :error, error_message},
+               attributes: attributes,
+               events: events
+             ) = assert_receive_span_named("read_stream_forward TestEventStore")
+
+      assert error_message == "** (RuntimeError) connection lost"
+
+      attrs = :otel_attributes.map(attributes)
+      assert attrs[:"erlang.exception.kind"] == :error
+      assert attrs[:"error.type"] == "RuntimeError"
+
+      assert_exception_event(events, "Elixir.RuntimeError", "connection lost")
+    end
+  end
+
+  describe "stop error result" do
+    test "no error status when result is :ok" do
+      meta = emit_start(:link_to_stream, %{event_store: TestEventStore, stream_uuid: "s1"})
+      emit_stop(:link_to_stream, Map.put(meta, :result, :ok))
+
+      assert span(status: status, attributes: attributes) =
+               assert_receive_span_named("link_to_stream TestEventStore")
+
+      refute match?({:status, :error, _}, status)
+      refute Map.has_key?(:otel_attributes.map(attributes), :"error.type")
+    end
+
+    test "sets error status when result is {:error, reason}" do
+      meta = emit_start(:link_to_stream, %{event_store: TestEventStore, stream_uuid: "s1"})
+      emit_stop(:link_to_stream, Map.put(meta, :result, {:error, :stream_not_found}))
+
+      assert span(
+               status: {:status, :error, ":stream_not_found"},
+               attributes: attributes
+             ) = assert_receive_span_named("link_to_stream TestEventStore")
+
+      assert :otel_attributes.map(attributes)[:"error.type"] == ":stream_not_found"
+    end
+  end
+
+  defp emit_start(operation, meta) do
+    meta = Map.put_new(meta, :telemetry_span_context, make_ref())
+
+    :telemetry.execute(
+      [:eventstore, operation, :start],
+      %{system_time: System.system_time(), monotonic_time: System.monotonic_time()},
+      meta
+    )
+
+    meta
+  end
+
+  defp emit_stop(operation, meta) do
+    :telemetry.execute(
+      [:eventstore, operation, :stop],
+      %{duration: 1000, monotonic_time: System.monotonic_time()},
+      meta
+    )
+  end
+
+  defp emit_exception(operation, meta, exception_fields) do
+    meta = Map.merge(meta, exception_fields)
+
+    :telemetry.execute(
+      [:eventstore, operation, :exception],
+      %{duration: 100, monotonic_time: System.monotonic_time()},
+      meta
+    )
+  end
+
+  defp assert_receive_span_named(name, timeout \\ 1000) do
+    deadline = System.monotonic_time(:millisecond) + timeout
+    receive_span_named(name, deadline)
+  end
+
+  defp receive_span_named(name, deadline) do
+    timeout = max(deadline - System.monotonic_time(:millisecond), 0)
+
+    receive do
+      {:span, span(name: ^name) = span_record} ->
+        span_record
+
+      {:span, _other} ->
+        receive_span_named(name, deadline)
+    after
+      timeout ->
+        flunk("Expected span #{inspect(name)}")
+    end
+  end
+
+  defp assert_exception_event(events, exception_type, exception_message) do
+    events_list = :otel_events.list(events)
+    exception_event = Enum.find(events_list, &(elem(&1, 2) == :exception))
+
+    assert exception_event
+
+    {:event, _timestamp, :exception, attrs_tuple} = exception_event
+    {:attributes, _, _, _, attrs_map} = attrs_tuple
+
+    assert attrs_map[:"exception.type"] == exception_type
+    assert attrs_map[:"exception.message"] == exception_message
+  end
+
+  defp detach_handlers do
+    for event <- @events do
+      for suffix <- [:start, :stop, :exception] do
+        for handler <- :telemetry.list_handlers([:eventstore, event, suffix]) do
+          :telemetry.detach(handler.id)
+        end
+      end
+    end
+  end
+end
diff --git a/test/support/opentelemetry_case.ex b/test/support/opentelemetry_case.ex
index 33689f21..37d648c1 100644
--- a/test/support/opentelemetry_case.ex
+++ b/test/support/opentelemetry_case.ex
@@ -59,7 +59,15 @@ defmodule Commanded.OpenTelemetryCase do
         [:commanded, :application, :dispatch, :exception]
       ]
 
-      for event <- commanded_events,
+      eventstore_operations =
+        ~w(append_to_stream delete_snapshot delete_stream delete_subscription link_to_stream paginate_streams read_snapshot read_stream_backward read_stream_forward record_snapshot stream_batch_read subscribe_to_stream)a
+
+      eventstore_events =
+        for op <- eventstore_operations, suffix <- [:start, :stop, :exception] do
+          [:eventstore, op, suffix]
+        end
+
+      for event <- commanded_events ++ eventstore_events,
           handler <- :telemetry.list_handlers(event) do
         :telemetry.detach(handler.id)
       end