feat(tests): support allowance: [Broadway] with metadata-tagged routing

Adds Broadway to the :allowance dispatch. The handler subscribes to
[:broadway, :processor, :start] and [:broadway, :batch_processor, :start],
reads the first :sentry_test_owner found in metadata.messages[*].metadata,
and calls allow_sentry_reports/2 for that test pid.

This follows the same shape Broadway documents for Ecto sandbox
testing (metadata: %{ecto_sandbox: self()}) — no wrapper around
Broadway.test_message/3, no hidden state, and async-safe by design
because each message carries its origin test pid. Messages without
:sentry_test_owner are silently skipped.

Integration coverage adds a minimal Broadway pipeline
(PhoenixApp.TestBroadway) and a 3-test broadway_test.exs in
phoenix_app proving: (1) tagged + allowance captures, (2) untagged +
allowance does not, (3) tagged + no allowance does not. async: true
on the describe block to validate cross-test isolation through a
shared pipeline.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: solnic

lib/sentry/test.ex

@@ -52,7 +52,7 @@ defmodule Sentry.Test do
 
   @moduledoc since: "10.2.0"
 
-  @compile {:no_warn_undefined, [Bypass, Plug.Conn, :telemetry]}
+  @compile {:no_warn_undefined, [Bypass, Plug.Conn, :telemetry, Broadway]}
 
   @ownership_server Sentry.Test.OwnershipServer
 
@@ -75,9 +75,9 @@ defmodule Sentry.Test do
 
   ## Options
 
-    * `:allowance` - a list of integration module atoms (currently `Oban`)
-      to enable automatic `Sentry.Test.allow_sentry_reports/2` wiring for.
-      See the "Oban tests" section below.
+    * `:allowance` - a list of integration module atoms (currently `Oban`
+      and `Broadway`) to enable automatic `Sentry.Test.allow_sentry_reports/2`
+      wiring for. See the "Oban tests" and "Broadway tests" sections below.
 
   Any other key is forwarded to the per-test Sentry config (e.g.,
   `dedup_events: false`, `traces_sample_rate: 1.0`).
@@ -125,6 +125,41 @@ defmodule Sentry.Test do
   jobs) are not auto-tagged and require manual
   `Sentry.Test.allow_sentry_reports/2`.
 
+  ## Broadway tests
+
+  To route events from a Broadway processor or batch-processor back
+  to your test, pass `:sentry_test_owner` in the message metadata
+  when injecting messages via `Broadway.test_message/3` or
+  `Broadway.test_batch/3`:
+
+      setup do
+        Sentry.Test.setup_sentry(allowance: [Broadway])
+        start_supervised!(MyPipeline)
+        :ok
+      end
+
+      test "captures events from the processor" do
+        ref =
+          Broadway.test_message(MyPipeline, payload,
+            metadata: %{sentry_test_owner: self()}
+          )
+
+        assert_receive {:ack, ^ref, [_succeeded], []}
+
+        assert [%Sentry.Event{}] = Sentry.Test.pop_sentry_reports()
+      end
+
+  This mirrors the [Ecto sandbox pattern documented in
+  Broadway](https://hexdocs.pm/broadway/Broadway.html#module-testing-with-ecto):
+  the test owner travels with the message itself, so two `async: true`
+  tests racing through the same pipeline are routed independently.
+
+  Messages submitted without the `:sentry_test_owner` metadata are not
+  auto-allowed — the handler silently skips them. For production
+  producers (Kafka, SQS, etc.) that need the same routing, attach the
+  same key to the messages they emit; the handler reads it regardless
+  of source.
+
   ## Examples
 
       setup do
@@ -950,6 +985,17 @@ defmodule Sentry.Test do
     ]
   end
 
+  defp allowance_handlers(Broadway) do
+    # Both events fire once per worker invocation (per batch) in the
+    # processor / batch-processor pid, with metadata.messages giving
+    # the full batch. Reading the owner from message metadata is the
+    # documented Broadway pattern (same shape as `ecto_sandbox`).
+    [
+      {[:broadway, :processor, :start], {__MODULE__, :__handle_broadway_batch_start__}},
+      {[:broadway, :batch_processor, :start], {__MODULE__, :__handle_broadway_batch_start__}}
+    ]
+  end
+
   defp allowance_handlers(_other), do: :unknown
 
   # ── Oban allowance handlers ──
@@ -998,6 +1044,33 @@ defmodule Sentry.Test do
 
   def __handle_oban_job_finish__(_event, _measurements, _metadata, _config), do: :ok
 
+  # ── Broadway allowance handler ──
+  #
+  # The handler walks the batch's messages looking for a
+  # `:sentry_test_owner` metadata entry — the documented Broadway test
+  # pattern, identical in shape to the `:ecto_sandbox` example in the
+  # Broadway testing guide. Tests submit messages via
+  # `Broadway.test_message/3` with
+  # `metadata: %{sentry_test_owner: self()}` (or any custom producer
+  # that propagates `:metadata` onto `%Broadway.Message{}`).
+  @doc false
+  def __handle_broadway_batch_start__(_event, _measurements, %{messages: messages}, _config)
+      when is_list(messages) do
+    case find_broadway_owner(messages) do
+      nil -> :ok
+      owner_pid -> safe_allow(owner_pid, self())
+    end
+  end
+
+  def __handle_broadway_batch_start__(_event, _measurements, _metadata, _config), do: :ok
+
+  defp find_broadway_owner(messages) do
+    Enum.find_value(messages, fn
+      %{metadata: %{sentry_test_owner: pid}} when is_pid(pid) -> pid
+      _ -> nil
+    end)
+  end
+
   # Best-effort allow used by the Oban / Broadway dispatch handlers.
   # Swallows the `ArgumentError` that `allow_sentry_reports/2` raises
   # when:

test/sentry/test_test.exs

@@ -553,6 +553,134 @@ defmodule Sentry.TestTest do
     end
   end
 
+  describe "setup_sentry/1 with allowance: [Broadway] (synthetic events)" do
+    setup do
+      SentryTest.setup_sentry(allowance: [Broadway])
+    end
+
+    test "processor batch start with a tagged message routes the worker" do
+      test_pid = self()
+      worker_done = make_ref()
+
+      worker =
+        spawn(fn ->
+          messages = [%{metadata: %{sentry_test_owner: test_pid}}]
+
+          :telemetry.execute(
+            [:broadway, :processor, :start],
+            %{},
+            %{messages: messages}
+          )
+
+          captured =
+            case Sentry.capture_message("from broadway processor", result: :sync) do
+              {:ok, _} -> :captured
+              other -> {:unexpected, other}
+            end
+
+          send(test_pid, {worker_done, captured})
+        end)
+
+      ref = Process.monitor(worker)
+      assert_receive {^worker_done, :captured}, 5_000
+      assert_receive {:DOWN, ^ref, :process, ^worker, _}, 5_000
+
+      assert [%Sentry.Event{message: %{formatted: "from broadway processor"}}] =
+               SentryTest.pop_sentry_reports()
+    end
+
+    test "batch_processor start with a tagged message routes the worker" do
+      test_pid = self()
+      worker_done = make_ref()
+
+      worker =
+        spawn(fn ->
+          messages = [%{metadata: %{sentry_test_owner: test_pid}}]
+
+          :telemetry.execute(
+            [:broadway, :batch_processor, :start],
+            %{},
+            %{messages: messages}
+          )
+
+          captured =
+            case Sentry.capture_message("from broadway batch", result: :sync) do
+              {:ok, _} -> :captured
+              other -> {:unexpected, other}
+            end
+
+          send(test_pid, {worker_done, captured})
+        end)
+
+      ref = Process.monitor(worker)
+      assert_receive {^worker_done, :captured}, 5_000
+      assert_receive {:DOWN, ^ref, :process, ^worker, _}, 5_000
+
+      assert [%Sentry.Event{message: %{formatted: "from broadway batch"}}] =
+               SentryTest.pop_sentry_reports()
+    end
+
+    test "messages without :sentry_test_owner metadata are not auto-allowed" do
+      test_pid = self()
+      worker_done = make_ref()
+
+      worker =
+        spawn(fn ->
+          messages = [%{metadata: %{some_other_key: :foo}}]
+
+          :telemetry.execute(
+            [:broadway, :processor, :start],
+            %{},
+            %{messages: messages}
+          )
+
+          captured =
+            case Sentry.capture_message("untagged broadway", result: :sync) do
+              {:ok, _} -> :captured
+              other -> {:unexpected, other}
+            end
+
+          send(test_pid, {worker_done, captured})
+        end)
+
+      ref = Process.monitor(worker)
+      assert_receive {^worker_done, :captured}, 5_000
+      assert_receive {:DOWN, ^ref, :process, ^worker, _}, 5_000
+
+      assert [] == SentryTest.pop_sentry_reports()
+    end
+
+    test "uses the first tagged message in a mixed batch" do
+      test_pid = self()
+      worker_done = make_ref()
+
+      worker =
+        spawn(fn ->
+          messages = [
+            %{metadata: %{some_other_key: :foo}},
+            %{metadata: %{sentry_test_owner: test_pid}},
+            %{metadata: %{}}
+          ]
+
+          :telemetry.execute(
+            [:broadway, :processor, :start],
+            %{},
+            %{messages: messages}
+          )
+
+          Sentry.capture_message("mixed batch", result: :sync)
+          send(test_pid, worker_done)
+        end)
+
+      ref = Process.monitor(worker)
+      assert_receive ^worker_done, 5_000
+      assert_receive {:DOWN, ^ref, :process, ^worker, _}, 5_000
+
+      assert [%Sentry.Event{message: %{formatted: "mixed batch"}}] =
+               SentryTest.pop_sentry_reports()
+    end
+  end
+
   describe "before_send wrapping" do
     test "wraps existing before_send callback" do
       test_pid = self()

test_integrations/phoenix_app/mix.exs

@@ -75,7 +75,8 @@ defmodule PhoenixApp.MixProject do
       {:opentelemetry_ecto, "~> 1.2"},
       {:opentelemetry_logger_metadata, "~> 0.2.0"},
       {:hackney, "~> 1.18"},
-      {:oban, "~> 2.10"}
+      {:oban, "~> 2.10"},
+      {:broadway, "~> 1.0", only: [:test]}
     ]
   end
 

test_integrations/phoenix_app/mix.lock

@@ -1,6 +1,7 @@
 %{
   "acceptor_pool": {:hex, :acceptor_pool, "1.0.1", "d88c2e8a0be9216cf513fbcd3e5a4beb36bee3ff4168e85d6152c6f899359cdb", [:rebar3], [], "hexpm", "f172f3d74513e8edd445c257d596fc84dbdd56d2c6fa287434269648ae5a421e"},
   "bandit": {:hex, :bandit, "1.10.3", "1e5d168fa79ec8de2860d1b4d878d97d4fbbe2fdbe7b0a7d9315a4359d1d4bb9", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "99a52d909c48db65ca598e1962797659e3c0f1d06e825a50c3d75b74a5e2db18"},
+  "broadway": {:hex, :broadway, "1.3.0", "f75f6376159b74f55c5ba2629dac613e4fd79d9e71148ab5fbac8fdd7c999d2a", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "bef3b4c5512d0072917b70239cbecf8f76a2587465a5b7c3e2b9ae18b4bc405b"},
   "bypass": {:hex, :bypass, "2.1.0", "909782781bf8e20ee86a9cabde36b259d44af8b9f38756173e8f5e2e1fabb9b1", [:mix], [{:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.0", [hex: :plug_cowboy, repo: "hexpm", optional: false]}, {:ranch, "~> 1.3", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "d9b5df8fa5b7a6efa08384e9bbecfe4ce61c77d28a4282f79e02f1ef78d96b80"},
   "castore": {:hex, :castore, "1.0.17", "4f9770d2d45fbd91dcf6bd404cf64e7e58fed04fadda0923dc32acca0badffa2", [:mix], [], "hexpm", "12d24b9d80b910dd3953e165636d68f147a31db945d2dcb9365e441f8b5351e5"},
   "cc_precompiler": {:hex, :cc_precompiler, "0.1.11", "8c844d0b9fb98a3edea067f94f616b3f6b29b959b6b3bf25fee94ffe34364768", [:mix], [{:elixir_make, "~> 0.7", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "3427232caf0835f94680e5bcf082408a70b48ad68a5f5c0b02a3bea9f3a075b9"},
@@ -24,6 +25,7 @@
   "finch": {:hex, :finch, "0.21.0", "b1c3b2d48af02d0c66d2a9ebfb5622be5c5ecd62937cf79a88a7f98d48a8290c", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.6.2 or ~> 1.7", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "87dc6e169794cb2570f75841a19da99cfde834249568f2a5b121b809588a4377"},
   "fine": {:hex, :fine, "0.1.4", "b19a89c1476c7c57afb5f9314aed5960b5bc95d5277de4cb5ee8e1d1616ce379", [:mix], [], "hexpm", "be3324cc454a42d80951cf6023b9954e9ff27c6daa255483b3e8d608670303f5"},
   "floki": {:hex, :floki, "0.38.0", "62b642386fa3f2f90713f6e231da0fa3256e41ef1089f83b6ceac7a3fd3abf33", [:mix], [], "hexpm", "a5943ee91e93fb2d635b612caf5508e36d37548e84928463ef9dd986f0d1abd9"},
+  "gen_stage": {:hex, :gen_stage, "1.3.2", "7c77e5d1e97de2c6c2f78f306f463bca64bf2f4c3cdd606affc0100b89743b7b", [:mix], [], "hexpm", "0ffae547fa777b3ed889a6b9e1e64566217413d018cabd825f786e843ffe63e7"},
   "gettext": {:hex, :gettext, "0.26.2", "5978aa7b21fada6deabf1f6341ddba50bc69c999e812211903b169799208f2a8", [:mix], [{:expo, "~> 0.5.1 or ~> 1.0", [hex: :expo, repo: "hexpm", optional: false]}], "hexpm", "aa978504bcf76511efdc22d580ba08e2279caab1066b76bb9aa81c4a1e0a32a5"},
   "gproc": {:hex, :gproc, "0.9.1", "f1df0364423539cf0b80e8201c8b1839e229e5f9b3ccb944c5834626998f5b8c", [:rebar3], [], "hexpm", "905088e32e72127ed9466f0bac0d8e65704ca5e73ee5a62cb073c3117916d507"},
   "grpcbox": {:hex, :grpcbox, "0.17.1", "6e040ab3ef16fe699ffb513b0ef8e2e896da7b18931a1ef817143037c454bcce", [:rebar3], [{:acceptor_pool, "~> 1.0.0", [hex: :acceptor_pool, repo: "hexpm", optional: false]}, {:chatterbox, "~> 0.15.1", [hex: :ts_chatterbox, repo: "hexpm", optional: false]}, {:ctx, "~> 0.6.0", [hex: :ctx, repo: "hexpm", optional: false]}, {:gproc, "~> 0.9.1", [hex: :gproc, repo: "hexpm", optional: false]}], "hexpm", "4a3b5d7111daabc569dc9cbd9b202a3237d81c80bf97212fbc676832cb0ceb17"},

test_integrations/phoenix_app/test/phoenix_app/broadway_test.exs

@@ -0,0 +1,53 @@
+defmodule PhoenixApp.BroadwayTest do
+  # async: true — the auto-allowance design routes per-message via the
+  # :sentry_test_owner metadata, which Broadway propagates onto the
+  # %Broadway.Message{} struct. Two of these tests racing against each
+  # other on the same shared pipeline still produce the right per-test
+  # results because each message carries its origin test pid.
+  use ExUnit.Case, async: true
+
+  describe "setup_sentry/1 with allowance: [Broadway]" do
+    setup do
+      Sentry.Test.setup_sentry(allowance: [Broadway])
+      start_supervised!(PhoenixApp.TestBroadway)
+      :ok
+    end
+
+    test "events from a Broadway processor are captured when tagged via metadata" do
+      ref =
+        Broadway.test_message(PhoenixApp.TestBroadway, :capture,
+          metadata: %{sentry_test_owner: self()}
+        )
+
+      assert_receive {:ack, ^ref, [_succeeded], []}, 5_000
+
+      assert [%Sentry.Event{message: %{formatted: "from broadway"}}] =
+               Sentry.Test.pop_sentry_reports()
+    end
+
+    test "raw Broadway.test_message without :sentry_test_owner is not auto-allowed" do
+      ref = Broadway.test_message(PhoenixApp.TestBroadway, :capture)
+
+      assert_receive {:ack, ^ref, [_succeeded], []}, 5_000
+      assert [] == Sentry.Test.pop_sentry_reports()
+    end
+  end
+
+  describe "without allowance" do
+    setup do
+      Sentry.Test.setup_sentry()
+      start_supervised!(PhoenixApp.TestBroadway)
+      :ok
+    end
+
+    test "tagged messages are still dropped without allowance: [Broadway]" do
+      ref =
+        Broadway.test_message(PhoenixApp.TestBroadway, :capture,
+          metadata: %{sentry_test_owner: self()}
+        )
+
+      assert_receive {:ack, ^ref, [_succeeded], []}, 5_000
+      assert [] == Sentry.Test.pop_sentry_reports()
+    end
+  end
+end

test_integrations/phoenix_app/test/support/test_broadway.ex

@@ -0,0 +1,29 @@
+defmodule PhoenixApp.TestBroadway do
+  @moduledoc false
+
+  use Broadway
+
+  def start_link(_opts \\ []) do
+    Broadway.start_link(__MODULE__,
+      name: __MODULE__,
+      producer: [module: {Broadway.DummyProducer, []}],
+      processors: [default: [concurrency: 1]]
+    )
+  end
+
+  @impl true
+  def handle_message(_processor, %Broadway.Message{data: data} = message, _context) do
+    case data do
+      :capture ->
+        Sentry.capture_message("from broadway", result: :sync)
+
+      :raise ->
+        raise "broadway boom"
+
+      _ ->
+        :ok
+    end
+
+    message
+  end
+end