feat(test): make TelemetryProcessor handling work with async tests

Closes the gap that left buffered events (logs and metrics) emitted
from processes registered via Sentry.Test.allow_sentry_reports/2
silently dropped instead of routed to the test's collector under
async tests.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: solnic

lib/sentry/application.ex

@@ -53,15 +53,15 @@ defmodule Sentry.Application do
         []
       end
 
-    telemetry_processor =
+    telemetry_processor_opts =
       [
-        {Sentry.TelemetryProcessor,
-         [
-           buffer_capacities: Config.telemetry_buffer_capacities(),
-           scheduler_weights: Config.telemetry_scheduler_weights(),
-           transport_capacity: Config.transport_capacity()
-         ]}
+        buffer_capacities: Config.telemetry_buffer_capacities(),
+        scheduler_weights: Config.telemetry_scheduler_weights(),
+        transport_capacity: Config.transport_capacity()
       ]
+      |> maybe_put_test_processor_resolver()
+
+    telemetry_processor = [{Sentry.TelemetryProcessor, telemetry_processor_opts}]
 
     children =
       maybe_test_registry ++
@@ -162,4 +162,12 @@ defmodule Sentry.Application do
   else
     defp maybe_rate_limiter, do: [Sentry.Transport.RateLimiter]
   end
+
+  defp maybe_put_test_processor_resolver(opts) do
+    if Config.test_mode?() do
+      Keyword.put(opts, :processor_resolver, &Sentry.Test.Registry.lookup_processor_for/1)
+    else
+      opts
+    end
+  end
 end

lib/sentry/telemetry_processor.ex

@@ -54,6 +54,9 @@ defmodule Sentry.TelemetryProcessor do
           | {:scheduler_weights, %{Category.priority() => pos_integer()}}
           | {:on_envelope, (Sentry.Envelope.t() -> any())}
           | {:transport_capacity, pos_integer()}
+          | {:processor_resolver, (pid() -> atom() | nil) | nil}
+
+  @resolver_key {__MODULE__, :resolver}
 
   ## Public API
 
@@ -74,6 +77,11 @@ defmodule Sentry.TelemetryProcessor do
     * `:scheduler_weights` - Map of priority to weight override (optional)
     * `:on_envelope` - Callback function invoked when envelopes are ready to send (optional)
     * `:transport_capacity` - Maximum number of items the transport queue can hold (default: 1000). For log envelopes, each log event counts as one item.
+    * `:processor_resolver` - Optional `(pid() -> atom() | nil)` function used by `add/1` to discover
+      the per-pid processor name when no value is set in the process dictionary. The resolver is
+      stored in `:persistent_term` and shared across all processor instances; passing the same
+      function from every `start_link/1` call is safe and idempotent. Defaults to `nil`, in which
+      case `add/1` falls back to the default processor name.
 
   ## Examples
 
@@ -294,6 +302,11 @@ defmodule Sentry.TelemetryProcessor do
     on_envelope = Keyword.get(opts, :on_envelope)
     transport_capacity = Keyword.get(opts, :transport_capacity, 1000)
 
+    case Keyword.get(opts, :processor_resolver) do
+      nil -> :ok
+      resolver when is_function(resolver, 1) -> :persistent_term.put(@resolver_key, resolver)
+    end
+
     buffer_names =
       for category <- Category.all(), into: %{} do
         {category, buffer_name(processor_name, category)}
@@ -374,7 +387,20 @@ defmodule Sentry.TelemetryProcessor do
   end
 
   defp processor_name do
-    Process.get(:sentry_telemetry_processor, @default_name)
+    case Process.get(:sentry_telemetry_processor) do
+      name when is_atom(name) and not is_nil(name) ->
+        name
+
+      _ ->
+        resolve_processor_name() || @default_name
+    end
+  end
+
+  defp resolve_processor_name do
+    case :persistent_term.get(@resolver_key, nil) do
+      nil -> nil
+      resolver when is_function(resolver, 1) -> resolver.(self())
+    end
   end
 
   defp maybe_add_opt(opts, _key, nil), do: opts

lib/sentry/test.ex

@@ -163,32 +163,35 @@ defmodule Sentry.Test do
   def setup_telemetry_processor do
     case Process.get(:sentry_telemetry_processor) do
       name when is_atom(name) and not is_nil(name) ->
-        if processor_alive?(name), do: name, else: do_setup_telemetry_processor()
+        if processor_alive?(name), do: name, else: start_telemetry_processor()
 
       _ ->
-        do_setup_telemetry_processor()
+        start_telemetry_processor()
     end
   end
 
-  defp do_setup_telemetry_processor do
+  defp start_telemetry_processor do
     uid = System.unique_integer([:positive])
     processor_name = :"test_telemetry_processor_#{uid}"
 
     ExUnit.Callbacks.start_supervised!(
-      {Sentry.TelemetryProcessor, name: processor_name},
+      {Sentry.TelemetryProcessor,
+       name: processor_name, processor_resolver: &Sentry.Test.Registry.lookup_processor_for/1},
       id: processor_name
     )
 
+    Process.put(:sentry_telemetry_processor, processor_name)
+
     scheduler_pid = Sentry.TelemetryProcessor.get_scheduler(processor_name)
 
     if scheduler_pid do
       # Goes through the unified `:sentry_test_scope` key, which also
       # populates the merged routing ETS row so `Config.namespace/1`
       # resolves the scheduler pid back to this test's scope.
       Sentry.Test.Registry.claim_allow(self(), scheduler_pid, :soft)
+      tag_processor_for_allowed_pid(self(), scheduler_pid)
     end
 
-    Process.put(:sentry_telemetry_processor, processor_name)
     processor_name
   end
 
@@ -227,8 +230,7 @@ defmodule Sentry.Test do
 
   > #### Deprecated {: .warning}
   >
-  > This function is deprecated and will be removed in v13.0.0.
-  > Use `setup_sentry/1` or `start_collecting_sentry_reports/0` instead.
+  > This function is deprecated and will be removed in v14.0.0. Use `setup_sentry/1` instead.
 
   The `:owner`, `:cleanup`, and `:key` options are no longer supported and are ignored.
   """
@@ -249,11 +251,11 @@ defmodule Sentry.Test do
 
   > #### Deprecated {: .warning}
   >
-  > This function is deprecated and will be removed in v13.0.0.
-  > Cleanup is now handled automatically via `on_exit` callbacks.
+  > This function is deprecated and will be removed in v14.0.0.
+  > Cleanup is now handled automatically when the owning test process exits.
   """
   @doc since: "10.2.0"
-  @doc deprecated: "Cleanup is now automatic via on_exit callbacks"
+  @doc deprecated: "Cleanup is now automatic when the owning test process exits"
   @spec cleanup(pid()) :: :ok
   def cleanup(owner_pid) when is_pid(owner_pid) do
     :ok
@@ -844,22 +846,24 @@ defmodule Sentry.Test do
 
     if scheduler_pid do
       Sentry.Test.Registry.claim_allow(self(), scheduler_pid, :soft)
+      tag_processor_for_allowed_pid(self(), scheduler_pid)
     end
 
     # Register cleanup for the collector ETS table only. NimbleOwnership
     # cleans up the key and allowances automatically when this test exits.
     # Drop any worker→processor routing rows that point at this test's
-    # processor so a test that exits before its allowed pids do does not
+    # processor so a test that exits before its allowed pids do not
     # leave stale rows pointing at a stopped per-test processor.
-    processor_name =
-      Process.get(:sentry_telemetry_processor, Sentry.TelemetryProcessor.default_name())
+    processor_name = Process.get(:sentry_telemetry_processor)
 
     ExUnit.Callbacks.on_exit(fn ->
       if :ets.whereis(collector_table) != :undefined do
         :ets.delete(collector_table)
       end
 
-      Sentry.Test.Registry.drop_processor_routing_for(processor_name)
+      if is_atom(processor_name) and not is_nil(processor_name) do
+        Sentry.Test.Registry.drop_processor_routing_for(processor_name)
+      end
     end)
 
     :ok

lib/sentry/test/registry.ex

@@ -55,21 +55,18 @@ defmodule Sentry.Test.Registry do
   end
 
   @doc """
-  Removes every routing row whose owner is `owner_pid`. Direct ETS
-  match_delete — atomic, no GenServer round-trip.
-
-  Kept for the case where a scope wants to drop its allowances
-  explicitly (e.g. `Sentry.Test.Scope.Registry.unregister/1`); the
-  `:DOWN` handler also prunes rows automatically when the owner pid
-  exits.
+  Ensures `owner_pid` is monitored by the registry so that the
+  `:DOWN` handler runs cleanup (routing-table prune + scope-state
+  erase via `Sentry.Test.Scope.Registry.handle_owner_down/1`) when
+  the owner exits. Idempotent.
+
+  Called from `Sentry.Test.Scope.Registry.update/1` on first scope
+  creation so cleanup does not depend on `claim_allow` ever being
+  invoked for this owner.
   """
-  @spec drop_allows_for(pid()) :: :ok
-  def drop_allows_for(owner_pid) when is_pid(owner_pid) do
-    if :ets.whereis(@routing_table) != :undefined do
-      :ets.match_delete(@routing_table, {:_, owner_pid, :_})
-    end
-
-    :ok
+  @spec monitor_owner(pid()) :: :ok
+  def monitor_owner(owner_pid) when is_pid(owner_pid) do
+    GenServer.call(__MODULE__, {:monitor_owner, owner_pid})
   end
 
   @doc """
@@ -166,38 +163,63 @@ defmodule Sentry.Test.Registry do
     {:ok, %{owner_monitors: %{}}}
   end
 
+  # Serialization note: every claim funnels through this single named
+  # GenServer and holds it across TWO blocking round-trips to the
+  # ownership server — `ensure_scope_owner/1`'s
+  # `NimbleOwnership.get_and_update/4` and `NimbleOwnership.allow/4`.
+  # This is the deliberate price of atomicity (no two concurrent async
+  # tests can both pass a check-then-write race for the same
+  # `allowed_pid`). It is acceptable because claims happen at test
+  # setup, not per event, and the hot config/buffer read paths
+  # (`lookup_allow_owner/1`, `lookup_processor_for/1`) bypass this
+  # GenServer with lock-free direct ETS reads.
   @impl true
   def handle_call({:claim_allow, owner_pid, allowed_pid, mode}, _from, state) do
     state = ensure_owner_monitored(state, owner_pid)
-    ensure_scope_owner(owner_pid)
 
     reply =
-      case NimbleOwnership.allow(@ownership_server, owner_pid, allowed_pid, @scope_key) do
+      case ensure_scope_owner(owner_pid) do
+        {:error, {:taken, existing_owner}} ->
+          if mode == :strict, do: {:error, {:taken, existing_owner}}, else: :skipped
+
         :ok ->
-          upsert_owner(allowed_pid, owner_pid)
-          :ok
+          case NimbleOwnership.allow(@ownership_server, owner_pid, allowed_pid, @scope_key) do
+            :ok ->
+              upsert_owner(allowed_pid, owner_pid)
+              :ok
 
-        {:error, %{reason: {:already_allowed, ^owner_pid}}} ->
-          upsert_owner(allowed_pid, owner_pid)
-          :ok
+            {:error, %{reason: {:already_allowed, ^owner_pid}}} ->
+              upsert_owner(allowed_pid, owner_pid)
+              :ok
 
-        {:error, %{reason: {:already_allowed, other}}} ->
-          if mode == :strict, do: {:error, {:taken, other}}, else: :skipped
+            {:error, %{reason: {:already_allowed, other}}} ->
+              if mode == :strict, do: {:error, {:taken, other}}, else: :skipped
 
-        {:error, %{reason: :already_an_owner}} ->
-          # `allowed_pid` is itself a scope owner — treat as a conflict.
-          if mode == :strict, do: {:error, {:taken, allowed_pid}}, else: :skipped
+            {:error, %{reason: :already_an_owner}} ->
+              # `allowed_pid` is itself a scope owner — treat as a conflict.
+              if mode == :strict, do: {:error, {:taken, allowed_pid}}, else: :skipped
+
+            {:error, %{reason: :not_allowed}} ->
+              if mode == :strict, do: {:error, {:taken, allowed_pid}}, else: :skipped
+          end
       end
 
     {:reply, reply, state}
   end
 
+  def handle_call({:monitor_owner, owner_pid}, _from, state) do
+    state = ensure_owner_monitored(state, owner_pid)
+    {:reply, :ok, state}
+  end
+
   @impl true
   def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
     if :ets.whereis(@routing_table) != :undefined do
       :ets.match_delete(@routing_table, {:_, pid, :_})
     end
 
+    Sentry.Test.Scope.Registry.handle_owner_down(pid)
+
     {:noreply, %{state | owner_monitors: Map.delete(state.owner_monitors, pid)}}
   end
 
@@ -220,21 +242,38 @@ defmodule Sentry.Test.Registry do
   # `Sentry.Test.setup_collector/1` (e.g. a test that uses
   # `Sentry.Test.Config.put/1` standalone). When the owner already
   # owns the key, the existing metadata is preserved.
+  #
+  # INVARIANT: the `:sentry_test_scope` key's metadata is overloaded —
+  # `Sentry.Test.setup_collector/1` stores the per-test collector ETS
+  # table name (an atom) under it, while this function stores a bare
+  # `%{}` marker for collector-less scopes. `Sentry.Test`'s
+  # `owner_collecting?/1` distinguishes the two purely by value type
+  # (atom = collecting, map = not). Therefore the update fun below MUST
+  # preserve an existing value (`current -> {:ok, current}`) and MUST
+  # NOT overwrite it with `%{}`; doing so would silently turn a
+  # collecting scope into a non-collecting one with no type error.
   defp ensure_scope_owner(owner_pid) do
     case NimbleOwnership.get_and_update(
            @ownership_server,
            owner_pid,
            @scope_key,
            # Metadata MUST be non-nil so that NimbleOwnership treats
            # `owner_pid` as a key owner (its `cond` in `allow/4` checks
-           # truthiness of the metadata). Preserve any existing value.
+           # truthiness of the metadata). Preserve any existing value
+           # (see the INVARIANT above — never clobber a collector atom).
            fn
              nil -> {:ok, %{}}
              current -> {:ok, current}
            end
          ) do
-      {:ok, _} -> :ok
-      {:error, _} -> :ok
+      {:ok, _} ->
+        :ok
+
+      {:error, %{reason: {:already_allowed, existing_owner}}} ->
+        {:error, {:taken, existing_owner}}
+
+      {:error, _} ->
+        :ok
     end
   end