feat(tests): support allowance: [Oban] with async-safe job tagging

Installs telemetry handlers that capture the test pid at Oban job
insert time and route the worker's captured events back to that test
on job start. The pairing makes auto-allowance safe under async: true
even when multiple tests share an Oban supervisor, because every job
is uniquely tied to the test that scheduled it (rather than to
whichever telemetry handler happened to fire first).

The tag store lives in Sentry.Test.Registry as a public ETS table
(:sentry_test_oban_job_tags) alongside the existing scope-allow
table. Tags are dropped on :stop/:exception and defensively on test
exit so jobs that crash before completion don't leave stale entries
behind.

Handlers guard on `is_integer(job.id)` so synthetic jobs from inline
mode or ad-hoc telemetry simulations (no persisted id) are skipped
silently. The :inline / :manual Oban testing modes run jobs in the
test pid anyway, so this auto-allowance is a no-op for them — it
only adds capability for the production-like worker case from
issue #1052.

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

Author: solnic

lib/sentry/test.ex

@@ -75,10 +75,9 @@ defmodule Sentry.Test do
 
   ## Options
 
-    * `:allowance` - a list of integration module atoms to enable automatic
-      `Sentry.Test.allow_sentry_reports/2` wiring for. The integrations land
-      in follow-up commits; see the integration-specific sections below for
-      supported entries.
+    * `: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.
 
   Any other key is forwarded to the per-test Sentry config (e.g.,
   `dedup_events: false`, `traces_sample_rate: 1.0`).
@@ -99,6 +98,32 @@ defmodule Sentry.Test do
 
   This collapses the common `bypass = setup_sentry(...); ref =
   setup_bypass_envelope_collector(bypass)` two-step into one call.
+  ## Oban tests
+
+  When you run Oban in `:inline` or `:manual` mode (per the
+  [Oban testing guide](https://hexdocs.pm/oban/testing.html)), jobs
+  execute synchronously in the calling process and `Sentry.Test`
+  captures their events automatically — no `:allowance` option needed.
+
+  Use `allowance: [Oban]` when your test exercises a real Oban
+  supervisor with worker processes (the production-like setup that
+  issue #1052 was filed for). The option installs telemetry handlers
+  that tag jobs at insert time and route the worker's captured events
+  back to the inserting test:
+
+      setup do
+        Sentry.Test.setup_sentry(allowance: [Oban])
+      end
+
+      test "captures events from a real Oban worker" do
+        {:ok, _} = Oban.insert(MyWorker.new(%{}))
+        # ... wait for the worker to run ...
+        assert [%Sentry.Event{}] = Sentry.Test.pop_sentry_reports()
+      end
+
+  Jobs inserted by other processes (cron plugins, jobs scheduling
+  jobs) are not auto-tagged and require manual
+  `Sentry.Test.allow_sentry_reports/2`.
 
   ## Examples
 
@@ -914,10 +939,82 @@ defmodule Sentry.Test do
 
   # Returns the list of `{event_path, {module, function}}` handler pairs for
   # a given integration atom, or `:unknown` for unsupported entries (the
-  # caller turns that into an `ArgumentError`). Commits 2 and 3 prepend
-  # clauses for `Oban` and `Broadway` respectively.
+  # caller turns that into an `ArgumentError`).
+  defp allowance_handlers(Oban) do
+    [
+      {[:oban, :engine, :insert_job, :stop], {__MODULE__, :__handle_oban_insert_job__}},
+      {[:oban, :engine, :insert_all_jobs, :stop], {__MODULE__, :__handle_oban_insert_all_jobs__}},
+      {[:oban, :job, :start], {__MODULE__, :__handle_oban_job_start__}},
+      {[:oban, :job, :stop], {__MODULE__, :__handle_oban_job_finish__}},
+      {[:oban, :job, :exception], {__MODULE__, :__handle_oban_job_finish__}}
+    ]
+  end
+
   defp allowance_handlers(_other), do: :unknown
 
+  # ── Oban allowance handlers ──
+  #
+  # Guards on `is_integer(id)` so synthetic jobs from `:inline` mode or
+  # ad-hoc telemetry simulations (no persisted id) are silently skipped —
+  # keeps the handlers safe to install in any test config.
+
+  @doc false
+  def __handle_oban_insert_job__(_event, _measurements, %{job: %{id: id}}, _config)
+      when is_integer(id) do
+    Sentry.Test.Registry.tag_oban_job(id, self())
+  end
+
+  def __handle_oban_insert_job__(_event, _measurements, _metadata, _config), do: :ok
+
+  @doc false
+  def __handle_oban_insert_all_jobs__(_event, _measurements, %{jobs: jobs}, _config)
+      when is_list(jobs) do
+    pid = self()
+
+    Enum.each(jobs, fn
+      %{id: id} when is_integer(id) -> Sentry.Test.Registry.tag_oban_job(id, pid)
+      _ -> :ok
+    end)
+  end
+
+  def __handle_oban_insert_all_jobs__(_event, _measurements, _metadata, _config), do: :ok
+
+  @doc false
+  def __handle_oban_job_start__(_event, _measurements, %{job: %{id: id}}, _config)
+      when is_integer(id) do
+    case Sentry.Test.Registry.lookup_oban_job(id) do
+      nil -> :ok
+      test_pid -> safe_allow(test_pid, self())
+    end
+  end
+
+  def __handle_oban_job_start__(_event, _measurements, _metadata, _config), do: :ok
+
+  @doc false
+  def __handle_oban_job_finish__(_event, _measurements, %{job: %{id: id}}, _config)
+      when is_integer(id) do
+    Sentry.Test.Registry.untag_oban_job(id)
+  end
+
+  def __handle_oban_job_finish__(_event, _measurements, _metadata, _config), do: :ok
+
+  # Best-effort allow used by the Oban / Broadway dispatch handlers.
+  # Swallows the `ArgumentError` that `allow_sentry_reports/2` raises
+  # when:
+  #
+  #   * the would-be worker pid is already allowed by another live test
+  #     (concurrent tests racing on a shared worker — first wins);
+  #   * the worker pid is the test pid itself (e.g. Oban :manual mode
+  #     with `drain_queue/2` — `$callers` already routes the events);
+  #   * the owner pid is no longer collecting (test exited between
+  #     insert and start).
+  defp safe_allow(owner_pid, allowed_pid)
+       when is_pid(owner_pid) and is_pid(allowed_pid) do
+    allow_sentry_reports(owner_pid, allowed_pid)
+  rescue
+    ArgumentError -> :ok
+  end
+
   # Sets up collection infrastructure (ETS table, before_send wrapping, config)
   # without opening a new Bypass. When no :dsn is provided in extra_config,
   # falls back to the default Bypass DSN from Registry.
@@ -982,7 +1079,11 @@ defmodule Sentry.Test do
     # 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 not
-    # leave stale rows pointing at a stopped per-test processor.
+    # leave stale rows pointing at a stopped per-test processor. Also
+    # defensively drop any Oban job tags owned by this test in case a
+    # job crashed before emitting a `:stop` / `:exception` event — leaves a
+    # stale row pointing at a dead pid otherwise.
+    test_pid = self()
     processor_name = Process.get(:sentry_telemetry_processor)
 
     ExUnit.Callbacks.on_exit(fn ->
@@ -993,6 +1094,8 @@ defmodule Sentry.Test do
       if is_atom(processor_name) and not is_nil(processor_name) do
         Sentry.Test.Registry.drop_processor_routing_for(processor_name)
       end
+
+      Sentry.Test.Registry.drop_oban_tags_for(test_pid)
     end)
 
     :ok

lib/sentry/test/registry.ex

@@ -20,6 +20,10 @@ defmodule Sentry.Test.Registry do
   # `tag_processor_for/2` sets it.
   @routing_table :sentry_test_pid_routing
 
+  # Separate ETS table tagging Oban job ids to the test pid that
+  # scheduled them, used by the Oban auto-allowance integration.
+  @oban_jobs_table :sentry_test_oban_job_tags
+
   @spec start_link(keyword()) :: GenServer.on_start()
   def start_link([] = _opts) do
     GenServer.start_link(__MODULE__, nil, name: __MODULE__)
@@ -236,9 +240,77 @@ defmodule Sentry.Test.Registry do
     :ok
   end
 
+  @doc """
+  Tags an inserted Oban job with the pid of the process that scheduled
+  it. Used by the `allowance: [Oban]` telemetry handlers in
+  `Sentry.Test` to route a worker's captured events back to the
+  inserting test under `async: true`.
+
+  Direct ETS write — atomic, no GenServer round-trip.
+  """
+  @spec tag_oban_job(integer(), pid()) :: :ok
+  def tag_oban_job(job_id, owner_pid)
+      when is_integer(job_id) and is_pid(owner_pid) do
+    if :ets.whereis(@oban_jobs_table) != :undefined do
+      :ets.insert(@oban_jobs_table, {job_id, owner_pid})
+    end
+
+    :ok
+  end
+
+  @doc """
+  Returns the pid that tagged `job_id`, or `nil` if the tag is missing
+  or the tagging pid is no longer alive.
+  """
+  @spec lookup_oban_job(integer()) :: pid() | nil
+  def lookup_oban_job(job_id) when is_integer(job_id) do
+    case :ets.whereis(@oban_jobs_table) do
+      :undefined ->
+        nil
+
+      _ ->
+        case :ets.lookup(@oban_jobs_table, job_id) do
+          [{^job_id, pid}] when is_pid(pid) ->
+            if Process.alive?(pid), do: pid, else: nil
+
+          [] ->
+            nil
+        end
+    end
+  end
+
+  @doc """
+  Removes the tag for `job_id`. Called from the `:oban, :job, :stop`
+  and `:oban, :job, :exception` handlers.
+  """
+  @spec untag_oban_job(integer()) :: :ok
+  def untag_oban_job(job_id) when is_integer(job_id) do
+    if :ets.whereis(@oban_jobs_table) != :undefined do
+      :ets.delete(@oban_jobs_table, job_id)
+    end
+
+    :ok
+  end
+
+  @doc """
+  Removes every tag whose owner is `owner_pid`. Used by
+  `setup_collector/1`'s `on_exit/1` cleanup so jobs that crashed
+  before emitting a `:stop`/`:exception` event don't leave stale tags
+  behind.
+  """
+  @spec drop_oban_tags_for(pid()) :: :ok
+  def drop_oban_tags_for(owner_pid) when is_pid(owner_pid) do
+    if :ets.whereis(@oban_jobs_table) != :undefined do
+      :ets.match_delete(@oban_jobs_table, {:_, owner_pid})
+    end
+
+    :ok
+  end
+
   @impl true
   def init(nil) do
     _routing_table = :ets.new(@routing_table, [:named_table, :public, :set])
+    _oban_jobs_table = :ets.new(@oban_jobs_table, [:named_table, :public, :set])
     maybe_start_default_bypass()
     {:ok, %{owner_monitors: %{}}}
   end

test/sentry/test_test.exs

@@ -416,6 +416,143 @@ defmodule Sentry.TestTest do
     end
   end
 
+  describe "setup_sentry/1 with allowance: [Oban] (synthetic events)" do
+    setup do
+      SentryTest.setup_sentry(allowance: [Oban])
+    end
+
+    test "tags the job at insert time and routes the worker on start" do
+      test_pid = self()
+      job = %{id: System.unique_integer([:positive])}
+
+      :telemetry.execute([:oban, :engine, :insert_job, :stop], %{}, %{job: job})
+      assert Sentry.Test.Registry.lookup_oban_job(job.id) == test_pid
+
+      worker_done = make_ref()
+
+      # Raw spawn/1 — does NOT propagate $callers, so the worker has no
+      # caller-chain link back to the test. The tag store is the only
+      # path that can route this worker's events to the test's collector.
+      worker =
+        spawn(fn ->
+          :telemetry.execute([:oban, :job, :start], %{}, %{job: job})
+
+          captured =
+            case Sentry.capture_message("oban hello", 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: "oban hello"}}] =
+               SentryTest.pop_sentry_reports()
+    end
+
+    test "ignores jobs that were not tagged at insert time" do
+      test_pid = self()
+      job = %{id: System.unique_integer([:positive])}
+      worker_done = make_ref()
+
+      worker =
+        spawn(fn ->
+          :telemetry.execute([:oban, :job, :start], %{}, %{job: job})
+
+          captured =
+            case Sentry.capture_message("untagged", 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 "untags the job on :stop" do
+      job = %{id: System.unique_integer([:positive])}
+      :telemetry.execute([:oban, :engine, :insert_job, :stop], %{}, %{job: job})
+      assert Sentry.Test.Registry.lookup_oban_job(job.id) == self()
+
+      :telemetry.execute([:oban, :job, :stop], %{}, %{job: job})
+      refute Sentry.Test.Registry.lookup_oban_job(job.id)
+    end
+
+    test "untags the job on :exception" do
+      job = %{id: System.unique_integer([:positive])}
+      :telemetry.execute([:oban, :engine, :insert_job, :stop], %{}, %{job: job})
+      assert Sentry.Test.Registry.lookup_oban_job(job.id) == self()
+
+      :telemetry.execute([:oban, :job, :exception], %{}, %{job: job})
+      refute Sentry.Test.Registry.lookup_oban_job(job.id)
+    end
+
+    test "insert_all_jobs tags every job in the batch" do
+      jobs = [
+        %{id: System.unique_integer([:positive])},
+        %{id: System.unique_integer([:positive])}
+      ]
+
+      :telemetry.execute([:oban, :engine, :insert_all_jobs, :stop], %{}, %{jobs: jobs})
+
+      for job <- jobs do
+        assert Sentry.Test.Registry.lookup_oban_job(job.id) == self()
+      end
+    end
+
+    test "silently ignores synthetic jobs without an integer id" do
+      # :inline mode jobs / ad-hoc telemetry simulations may carry no id.
+      :telemetry.execute([:oban, :engine, :insert_job, :stop], %{}, %{job: %{id: nil}})
+      :telemetry.execute([:oban, :job, :start], %{}, %{job: %{id: nil}})
+      :ok
+    end
+
+    test "two concurrent test scopes are routed independently" do
+      test_pid = self()
+      job_for_me = %{id: System.unique_integer([:positive])}
+      job_for_peer = %{id: System.unique_integer([:positive])}
+
+      # Spawn a peer that acts as a separate live owner via NimbleOwnership,
+      # tags its own Oban job, and reports back when ready.
+      peer =
+        spawn(fn ->
+          {:ok, _} =
+            NimbleOwnership.get_and_update(
+              Sentry.Test.OwnershipServer,
+              self(),
+              :sentry_test_collector,
+              fn _ -> {:ok, :peer_table} end
+            )
+
+          Sentry.Test.Registry.tag_oban_job(job_for_peer.id, self())
+          send(test_pid, :claimed)
+
+          receive do
+            :exit -> :ok
+          end
+        end)
+
+      on_exit(fn -> Process.exit(peer, :kill) end)
+      assert_receive :claimed, 5_000
+
+      # Tag my own job.
+      :telemetry.execute([:oban, :engine, :insert_job, :stop], %{}, %{job: job_for_me})
+
+      assert Sentry.Test.Registry.lookup_oban_job(job_for_me.id) == test_pid
+      assert Sentry.Test.Registry.lookup_oban_job(job_for_peer.id) == peer
+    end
+  end
+
   describe "before_send wrapping" do
     test "wraps existing before_send callback" do
       test_pid = self()