From bae29e6cdf01c19405da6293368c0c07ac4d3dbf Mon Sep 17 00:00:00 2001
From: Peter Solnica <peter@solnica.online>
Date: Thu, 15 Oct 2026 11:23:24 +0000
Subject: [PATCH] feat(rails): delegate ActiveJob to backend extensions

---
 sentry-rails/lib/sentry/rails.rb              |  23 +
 sentry-rails/lib/sentry/rails/active_job.rb   | 246 +++-----
 .../sentry/rails/active_job/job_extension.rb  | 123 ++++
 .../lib/sentry/rails/active_job/producer.rb   |  42 ++
 .../tracing/consumer_transaction.rb           |  39 +-
 .../tracing/messaging_span_data.rb            |  18 +
 .../shared_examples/tracing/producer_span.rb  |  20 +
 .../tracing/trace_propagation.rb              |  47 +-
 .../spec/active_job/support/harness.rb        |   6 +
 .../spec/active_job/test_adapter_spec.rb      |  23 +
 .../rails/active_job_job_extensions_spec.rb   | 551 ++++++++++++++++++
 .../sentry/rails/active_job_ownership_spec.rb |  25 +
 sentry-rails/spec/spec_helper.rb              |   2 +
 13 files changed, 975 insertions(+), 190 deletions(-)
 create mode 100644 sentry-rails/lib/sentry/rails/active_job/job_extension.rb
 create mode 100644 sentry-rails/lib/sentry/rails/active_job/producer.rb
 create mode 100644 sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb
 create mode 100644 sentry-rails/spec/sentry/rails/active_job_ownership_spec.rb

diff --git a/sentry-rails/lib/sentry/rails.rb b/sentry-rails/lib/sentry/rails.rb
index 8af39f372..896968b2b 100644
--- a/sentry-rails/lib/sentry/rails.rb
+++ b/sentry-rails/lib/sentry/rails.rb
@@ -13,5 +13,28 @@ module Sentry
   module Rails
     extend Integrable
     register_integration name: "rails", version: Sentry::Rails::VERSION
+
+    # Returns whether sentry-rails' ActiveJob integration reports jobs run
+    # through the given queue adapter.
+    #
+    # When it returns true, the integration captures errors from those jobs,
+    # records their queue.process transactions and queue.publish spans,
+    # propagates traces, and consults extensions registered with
+    # {Sentry.register_job_extension}. A backend gem should then not report the
+    # same ActiveJob jobs itself.
+    #
+    # The result depends on {Sentry.initialized?} and the current configuration,
+    # so it can change after {Sentry.init}. Call it when a job is handled rather
+    # than caching the result at load time.
+    #
+    # @param adapter_class_name [String, Class] the ActiveJob queue adapter's
+    #   class name, such as "ActiveJob::QueueAdapters::SidekiqAdapter", or the
+    #   adapter class itself. It is compared against
+    #   +config.rails.skippable_job_adapters+.
+    # @return [Boolean] true if the SDK is initialized and the adapter is not
+    #   listed in +config.rails.skippable_job_adapters+, false otherwise
+    def self.active_job_owns_adapter?(adapter_class_name)
+      Sentry.initialized? && !Sentry.configuration.rails.skippable_job_adapters.include?(adapter_class_name.to_s)
+    end
   end
 end
diff --git a/sentry-rails/lib/sentry/rails/active_job.rb b/sentry-rails/lib/sentry/rails/active_job.rb
index 69d48a1dd..05b4ed158 100644
--- a/sentry-rails/lib/sentry/rails/active_job.rb
+++ b/sentry-rails/lib/sentry/rails/active_job.rb
@@ -3,16 +3,17 @@
 require "set"
 require "sentry/rails/serializer"
 require "sentry/rails/error_reporter_context"
+require "sentry/rails/active_job/job_extension"
+require "sentry/rails/active_job/producer"
 
 module Sentry
   module Rails
     module ActiveJobExtensions
       SENTRY_PAYLOAD_KEY = "_sentry"
 
-      USER_FIELDS_ALLOWLIST = %w[id email username].freeze
-
       def self.prepended(base)
         base.attr_accessor :_sentry
+        base.attr_reader :_sentry_job_data
       end
 
       def perform_now
@@ -20,10 +21,16 @@ def perform_now
           super
         else
           data = _sentry || {}
+          current_scope = Sentry.get_current_scope
+          scope = current_scope if data["execution_scope_id"] == current_scope.object_id
+          transaction = scope&.get_span
           SentryReporter.record(
             self,
             trace_headers: data["trace_propagation_headers"],
-            user: data["user"]
+            user: data["user"],
+            payload: _sentry_job_data,
+            transaction: transaction,
+            scope: scope
           ) { super }
         end
       end
@@ -31,26 +38,7 @@ def perform_now
       def serialize
         payload = super
         return payload if !Sentry.initialized? || already_supported_by_sentry_integration?
-
-        begin
-          sentry_data = {}
-          if Sentry.configuration.rails.active_job_propagate_traces
-            headers = Sentry.get_trace_propagation_headers
-            sentry_data["trace_propagation_headers"] = headers if headers && !headers.empty?
-          end
-
-          if Sentry.configuration.data_collection.user_info
-            user = Sentry.get_current_scope.user
-            allowed = user.transform_keys(&:to_s).slice(*USER_FIELDS_ALLOWLIST)
-            sentry_data["user"] = allowed unless allowed.empty?
-          end
-
-          payload[SENTRY_PAYLOAD_KEY] = sentry_data unless sentry_data.empty?
-        rescue StandardError => e
-          Sentry.sdk_logger&.error("sentry-rails: failed to inject _sentry payload: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
-        end
-
-        payload
+        SentryReporter.producer(self).prepare_payload(self, payload)
       end
 
       def deserialize(job_data)
@@ -58,6 +46,7 @@ def deserialize(job_data)
         return if !Sentry.initialized? || already_supported_by_sentry_integration?
 
         begin
+          @_sentry_job_data = job_data
           self._sentry = job_data[SENTRY_PAYLOAD_KEY]
         rescue StandardError => e
           Sentry.sdk_logger&.error("sentry-rails: failed to extract _sentry payload: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
@@ -65,25 +54,30 @@ def deserialize(job_data)
       end
 
       def already_supported_by_sentry_integration?
-        Sentry.configuration.rails.skippable_job_adapters.include?(self.class.queue_adapter.class.to_s)
+        !Sentry::Rails.active_job_owns_adapter?(self.class.queue_adapter.class)
       end
 
       class SentryReporter
-        OP_NAME = "queue.process"
-        SPAN_ORIGIN = "auto.queue.active_job"
-
-        # Emitted as messaging.system when the configured queue adapter's
-        # identity cannot be resolved from the job. ActiveJob is adapter-agnostic
-        # and supports arbitrary third-party backends, so the concrete value is
-        # normally derived from the adapter itself; this is only the last resort.
-        MESSAGING_SYSTEM_FALLBACK = "activejob"
-
         EVENT_HANDLERS = {
           "enqueue_retry.active_job" => :retry_handler
         }
 
         class << self
-          include ErrorReporterContext
+          def default_extension
+            @default_extension ||= JobExtension.new
+          end
+
+          def consumer
+            @consumer ||= Sentry::JobExtensions::Consumer.new(integration: Sentry::Rails, origin: JobExtension::SPAN_ORIGIN, registry: job_extensions)
+          end
+
+          def producer(job)
+            job_extensions.call(job_extensions.for(job), :producer, job: job)
+          end
+
+          def job_extensions
+            default_extension.registry
+          end
 
           def producer_callback_registered?
             @producer_callback_registered ||= false
@@ -95,144 +89,51 @@ def producer_callback_registered!
 
           def record_producer_span(job, &enqueue)
             return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration?
+            producer(job).record_span(job, &enqueue)
+          end
 
-            enqueued = false
+          def record(job, trace_headers: nil, user: nil, payload: nil, transaction: nil, scope: nil, &block)
+            options = { trace_headers: trace_headers, user: user, payload: payload, transaction: transaction }
+            return record_in_scope(scope, job, **options, backend_scope: true, &block) if scope
 
-            run_enqueue = lambda do
-              enqueued = true
-              enqueue.call
+            with_isolated_scope do |isolated_scope|
+              record_in_scope(isolated_scope, job, **options, &block)
             end
+          end
 
+          def record_in_scope(scope, job, trace_headers: nil, user: nil, payload: nil, transaction: nil, backend_scope: false)
             begin
-              Sentry.with_child_span(op: "queue.publish", description: job.class.name) do |span|
-                if span
-                  span.set_origin(SPAN_ORIGIN)
-                  span.set_data(Sentry::Span::DataConventions::MESSAGING_SYSTEM, messaging_system(job))
-                  span.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
-                  span.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)
-
-                  if (count = retry_count(job))
-                    span.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count)
-                  end
-                end
-
-                run_enqueue.call
-              end
-            rescue StandardError => e
-              raise if enqueued
+              scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?
 
-              log_producer_span_error(e)
+              consumer.apply_scope_data(scope, job)
 
-              run_enqueue.call
-            end
-          end
+              scope.set_transaction_name(consumer.transaction_name(job), source: :task)
 
-          def log_producer_span_error(e)
-            Sentry.sdk_logger&.error("sentry-rails: failed to record producer span: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
-          end
-
-          def record(job, trace_headers: nil, user: nil, &block)
-            original_hub = Sentry.get_current_hub_internal
-            Sentry.clone_hub_to_current_thread
+              transaction = consumer.start_transaction(
+                scope, job,
+                transaction: transaction.is_a?(Sentry::Transaction) ? transaction : nil,
+                trace_headers: trace_headers,
+                payload: payload
+              )
 
-            Sentry.with_scope do |scope|
-              begin
-                scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?
-
-                scope.set_transaction_name(job.class.name, source: :task)
-                scope.set_tags(queue: job.queue_name)
-
-                scope.set_contexts(active_job: {
-                  job_class: job.class.name,
-                  job_id: job.job_id,
-                  queue: job.queue_name,
-                  provider_job_id: job.provider_job_id
-                })
-
-                transaction_options = {
-                  name: scope.transaction_name,
-                  source: scope.transaction_source,
-                  op: OP_NAME,
-                  origin: SPAN_ORIGIN
-                }
-
-                transaction = if trace_headers && !trace_headers.empty?
-                  continued = Sentry.continue_trace(trace_headers, **transaction_options)
-                  Sentry.start_transaction(transaction: continued, **transaction_options)
-                else
-                  Sentry.start_transaction(**transaction_options)
-                end
-
-                if transaction
-                  set_messaging_data(transaction, job)
-                  scope.set_span(transaction)
-                end
-
-                yield.tap do
-                  finish_sentry_transaction(transaction, 200)
-                end
-              rescue Exception => e # rubocop:disable Lint/RescueException
-                finish_sentry_transaction(transaction, 500)
+              yield.tap do
+                consumer.finish_transaction(transaction, 200)
+              end
+            rescue Exception => e # rubocop:disable Lint/RescueException
+              consumer.finish_transaction(transaction, 500)
 
+              if backend_scope
+                scope.update_from_options(**default_extension.error_data(job: job, exception: e))
+              elsif consumer.report_exception?(job, e)
                 capture_exception(job, e)
-
-                raise
               end
-            end
-          ensure
-            Sentry.set_current_hub_internal(original_hub)
-          end
-
-          def set_messaging_data(transaction, job)
-            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_SYSTEM, messaging_system(job))
-            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
-            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)
-            if (count = retry_count(job))
-              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count)
-            end
 
-            if (latency = compute_latency(job))
-              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
+              raise
             end
           end
 
-          def messaging_system(job)
-            name = job.class.queue_adapter_name if job.class.respond_to?(:queue_adapter_name)
-            name = name.to_s
-            name.empty? ? MESSAGING_SYSTEM_FALLBACK : name
-          end
-
-          # Number of retries the job has already gone through, as observed at
-          # the moment the span is opened. On the consumer the span opens before
-          # ActiveJob increments +executions+, so a job's first attempt reads 0,
-          # its first retry reads 1, and so on. On the producer the retry enqueue
-          # runs after the failed attempt bumped +executions+, so it observes the
-          # same progression.
-          def retry_count(job)
-            job.executions.to_i if job.respond_to?(:executions)
-          end
-
-          def compute_latency(job)
-            return unless job.respond_to?(:enqueued_at) && job.enqueued_at
-
-            enqueued_time = job.enqueued_at.is_a?(String) ? Time.parse(job.enqueued_at) : job.enqueued_at
-            (Time.now.to_f - enqueued_time.to_f) * 1000.0
-          end
-
           def capture_exception(job, e)
-            options = {
-              extra: sentry_context(job),
-              tags: {
-                job_id: job.job_id,
-                provider_job_id: job.provider_job_id
-              },
-              # Send synchronously: a worker process may exit before the async
-              # background worker flushes its queue, which would drop the event.
-              hint: { background: false }
-            }
-            options[:contexts] = execution_context if Sentry.configuration.data_collection.queues
-
-            Sentry::Rails.capture_exception(e, **options)
+            consumer.capture_exception(job, e, **default_extension.error_data(job: job, exception: e))
           end
 
           def register_event_handlers
@@ -255,7 +156,7 @@ def retry_handler(*args)
             handle_error_event(*args) do |job, error|
               return if error.nil?
               return if !Sentry.initialized? || job.already_supported_by_sentry_integration?
-              return unless Sentry.configuration.rails.active_job_report_on_retry_error
+              return unless consumer.report_retried_exception?(job, error, default: nil)
 
               capture_exception(job, error)
             end
@@ -266,35 +167,18 @@ def handle_error_event(*args)
             yield(event.payload[:job], event.payload[:error])
           end
 
-          def finish_sentry_transaction(transaction, status)
-            return unless transaction
-
-            transaction.set_http_status(status)
-            transaction.finish
-          end
+          private :record_in_scope
 
-          def sentry_context(job)
-            context = {
-              active_job: job.class.name,
-              scheduled_at: job.scheduled_at,
-              job_id: job.job_id,
-              provider_job_id: job.provider_job_id,
-              locale: job.locale
-            }
-
-            if Sentry.configuration.data_collection.queues
-              context[:arguments] = sentry_serialize_arguments(job.arguments)
-            end
-
-            context
-          end
+          private
 
-          def sentry_serialize_arguments(argument)
-            Sentry::Rails::Serializer.serialize(argument)
+          def with_isolated_scope(&block)
+            original_hub = Sentry.get_current_hub_internal
+            Sentry.clone_hub_to_current_thread
+            Sentry.with_scope(&block)
+          ensure
+            Sentry.set_current_hub_internal(original_hub)
           end
 
-          private
-
           def subscribers
             @__subscribers__ ||= Set.new
           end
diff --git a/sentry-rails/lib/sentry/rails/active_job/job_extension.rb b/sentry-rails/lib/sentry/rails/active_job/job_extension.rb
new file mode 100644
index 000000000..4ca4480d7
--- /dev/null
+++ b/sentry-rails/lib/sentry/rails/active_job/job_extension.rb
@@ -0,0 +1,123 @@
+# frozen_string_literal: true
+
+require "sentry/rails/serializer"
+require "sentry/rails/error_reporter_context"
+
+module Sentry
+  module Rails
+    module ActiveJobExtensions
+      class JobExtension < Sentry::JobExtensions::Extension
+        include ErrorReporterContext
+
+        SPAN_ORIGIN = "auto.queue.active_job"
+        USER_FIELDS_ALLOWLIST = %w[id email username].freeze
+
+        attr_reader :registry
+
+        def initialize
+          super(name: "active_job")
+          @registry = Sentry.job_extensions.with_default(self)
+        end
+
+        def merge_hook(hook, default:, override:)
+          case hook
+          when :scope_data
+            contexts = override.fetch(:contexts, {}).merge(default[:contexts]) { |_name, backend, generic| backend.merge(generic) }
+            override.merge(contexts: contexts, tags: override.fetch(:tags, {}).merge(default[:tags]))
+          when :messaging_data
+            default.merge(override)
+          when :publish_messaging_data
+            override.merge(default)
+          when :error_data, :inject_trace_headers
+            override
+          else
+            super
+          end
+        end
+
+        def handles?(job)
+          job.is_a?(::ActiveJob::Base)
+        end
+
+        def producer(job:)
+          @producer ||= Producer.new(registry: registry, origin: SPAN_ORIGIN)
+        end
+
+        def scope_data(job:)
+          {
+            contexts: { active_job: {
+              job_class: job.class.name,
+              job_id: job.job_id,
+              queue: job.queue_name,
+              provider_job_id: job.provider_job_id
+            } },
+            tags: { queue: job._sentry&.[]("delivery_queue") || job.queue_name }
+          }
+        end
+
+        def error_data(job:, exception:)
+          extra = {
+            active_job: job.class.name,
+            scheduled_at: job.scheduled_at,
+            job_id: job.job_id,
+            provider_job_id: job.provider_job_id,
+            locale: job.locale
+          }
+          extra[:arguments] = Sentry::Rails::Serializer.serialize(job.arguments) if Sentry.configuration.data_collection.queues
+
+          {
+            contexts: Sentry.configuration.data_collection.queues ? execution_context : {},
+            tags: { job_id: job.job_id, provider_job_id: job.provider_job_id },
+            extra: extra
+          }
+        end
+
+        def messaging_data(job:)
+          data = publish_messaging_data(job: job)
+          data[:receive_latency] = Sentry::Utils::Latency.milliseconds_since(job.enqueued_at) if job.respond_to?(:enqueued_at)
+          data[:body_size] = body_size(job._sentry_job_data)
+          data.compact
+        end
+
+        def body_size(payload)
+          JSON.generate(payload).bytesize if payload
+        rescue StandardError
+          nil
+        end
+
+        def publish_messaging_data(job:)
+          name = job.class.queue_adapter_name if job.class.respond_to?(:queue_adapter_name)
+          {
+            system: name.to_s.empty? ? "activejob" : name.to_s,
+            message_id: job.job_id,
+            destination: job.queue_name
+          }
+        end
+
+        def retry_info(job:)
+          { count: job.executions.to_i, will_retry: false }
+        end
+
+        def report_retried_exception?(job:, exception:)
+          Sentry.configuration.rails.active_job_report_on_retry_error
+        end
+
+        def transaction_name(job:)
+          job.class.name
+        end
+
+        def publish_description(job:)
+          job.class.name
+        end
+
+        def span_origin(job:)
+          SPAN_ORIGIN
+        end
+
+        def integration(job:)
+          Sentry::Rails
+        end
+      end
+    end
+  end
+end
diff --git a/sentry-rails/lib/sentry/rails/active_job/producer.rb b/sentry-rails/lib/sentry/rails/active_job/producer.rb
new file mode 100644
index 000000000..29c83a024
--- /dev/null
+++ b/sentry-rails/lib/sentry/rails/active_job/producer.rb
@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Rails
+    module ActiveJobExtensions
+      class JobExtension
+        class Producer < Sentry::JobExtensions::Producer
+          def prepare_payload(job, payload)
+            sentry_data = {}
+            if Sentry.configuration.rails.active_job_propagate_traces
+              headers = Sentry.get_trace_propagation_headers
+              if headers && !headers.empty?
+                sentry_data["trace_propagation_headers"] = headers
+                payload = inject_trace_headers(job, payload, headers)
+              end
+            end
+
+            if Sentry.configuration.data_collection.user_info
+              user = Sentry.get_current_scope.user.transform_keys(&:to_s).slice(*USER_FIELDS_ALLOWLIST)
+              sentry_data["user"] = user unless user.empty?
+            end
+
+            sentry_data.empty? ? payload : payload.merge(SENTRY_PAYLOAD_KEY => sentry_data)
+          rescue StandardError => e
+            Sentry.sdk_logger&.error("sentry-rails: failed to inject _sentry payload: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
+            payload
+          end
+
+          def record_span(job, &enqueue)
+            super(job, on_instrumentation_error: method(:log_producer_span_error), &enqueue)
+          end
+
+          private
+
+          def log_producer_span_error(e)
+            Sentry.sdk_logger&.error("sentry-rails: failed to record producer span: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb b/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb
index 12222acaa..9d0be2ef0 100644
--- a/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb
@@ -89,14 +89,51 @@ def perform
   context "with traces_sample_rate = 0" do
     before { Sentry.configuration.traces_sample_rate = 0 }
 
-    it "does not capture a transaction" do
+    it "reports job errors with messaging data without capturing a transaction" do
       expect do
         failing_job.perform_later
         drain
       end.to raise_error(RuntimeError, /boom from failing_job spec/)
 
+      error_event = sentry_events.find { |event| event.is_a?(Sentry::ErrorEvent) }
+      expect(error_event.contexts.dig(:trace, :data)).to include(
+        "messaging.message.id" => a_kind_of(String),
+        "messaging.destination.name" => "default",
+        "messaging.system" => successful_job.queue_adapter_name,
+        "messaging.message.retry.count" => 0
+      )
+
       transactions = sentry_events.select { |e| e.is_a?(Sentry::TransactionEvent) }
       expect(transactions).to be_empty
     end
   end
+
+  context "when the transaction is unsampled" do
+    let(:configure_sentry) { proc { |config| config.traces_sample_rate = 0.0 } }
+
+    it "performs the enqueued job without reporting transactions" do
+      performed_arguments = []
+      recording_job = job_fixture do
+        define_method(:perform) { |argument| performed_arguments << argument }
+      end
+
+      within_parent_transaction { recording_job.perform_later("hello") }
+      drain
+
+      expect(performed_arguments).to eq(["hello"])
+      expect(transactions).to be_empty
+    end
+  end
+
+  context "when tracing is disabled" do
+    it "reports job errors without messaging.message.body.size" do
+      expect do
+        failing_job.perform_later("hello")
+        drain
+      end.to raise_error(RuntimeError, /boom from failing_job spec/)
+
+      trace_data = last_sentry_event.contexts.dig(:trace, :data) || {}
+      expect(trace_data).not_to have_key("messaging.message.body.size")
+    end
+  end
 end
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb b/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb
index 16af73cb3..6dfd4d9f8 100644
--- a/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb
@@ -23,6 +23,24 @@
     expect(data["messaging.system"]).to eq(successful_job.queue_adapter_name)
   end
 
+  it "records the byte length of the delivered job data encoded as JSON as messaging.message.body.size" do
+    delivered_job_data = []
+    recording_job = job_fixture do
+      define_method(:deserialize) do |job_data|
+        delivered_job_data << job_data
+        super(job_data)
+      end
+
+      def perform(_greeting); end
+    end
+
+    recording_job.perform_later("héllo")
+    drain
+
+    data = consumer_transaction.contexts.dig(:trace, :data)
+    expect(data["messaging.message.body.size"]).to eq(JSON.generate(delivered_job_data.last).bytesize)
+  end
+
   it "records messaging.message.retry.count = 0 for non-retryable jobs" do
     successful_job.perform_later
     drain
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb b/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb
index 102093ab1..278de41a9 100644
--- a/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb
@@ -62,6 +62,26 @@
     end
   end
 
+  context "when tracing is disabled" do
+    let(:routed_job) do
+      job_fixture do
+        attr_accessor :routed_queue
+
+        queue_as { routed_queue || "unrouted" }
+
+        before_enqueue { |job| job.routed_queue = "routed" }
+
+        def perform; end
+      end
+    end
+
+    it "resolves a queue_as block after the job's own enqueue callbacks" do
+      routed_job.perform_later
+
+      expect(last_enqueued_payload["queue_name"]).to eq("routed")
+    end
+  end
+
   context "when producer-span instrumentation raises" do
     let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
 
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb b/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb
index 570ae87ac..eadcbf240 100644
--- a/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb
@@ -20,15 +20,30 @@
     expect(consumer_transaction.contexts.dig(:trace, :parent_span_id)).to eq(publish_span_id)
   end
 
-  it "injects both sentry-trace and baggage propagation headers into the serialized payload" do
-    within_parent_transaction do
-      successful_job.perform_later
+  [nil, 0.0, 1.0].each do |sample_rate|
+    context "with incoming baggage at sample rate #{sample_rate.inspect}" do
+      let(:configure_sentry) { proc { |config| config.traces_sample_rate = sample_rate } }
+
+      it "preserves the incoming trace and baggage through enqueue and execution" do
+        reporting_job = job_fixture do
+          def perform
+            Sentry.capture_message("propagated", extra: { headers: Sentry.get_trace_propagation_headers })
+          end
+        end
+        trace_id = "a" * 32
+        baggage = "sentry-trace_id=#{trace_id},sentry-release=upstream,sentry-public_key=12345"
+        Sentry.continue_trace({ "HTTP_SENTRY_TRACE" => "#{trace_id}-#{'b' * 16}-1", "HTTP_BAGGAGE" => baggage })
+
+        reporting_job.perform_later
+        Sentry.get_current_scope.clear
+        drain
+
+        event = sentry_events.find { |entry| entry.is_a?(Sentry::ErrorEvent) && entry.message == "propagated" }
+        expect(event.contexts.dig(:trace, :trace_id)).to eq(trace_id)
+        expect(event.extra[:headers]["sentry-trace"]).to start_with(trace_id)
+        expect(event.extra[:headers]["baggage"]).to include("sentry-release=upstream", "sentry-trace_id=#{trace_id}")
+      end
     end
-
-    headers = last_enqueued_payload["_sentry"]["trace_propagation_headers"]
-    expect(headers.keys).to include("sentry-trace", "baggage")
-    expect(headers["sentry-trace"]).not_to be_empty
-    expect(headers["baggage"]).not_to be_empty
   end
 
   it "captures a consumer transaction without raising when no parent transaction was active at enqueue" do
@@ -98,4 +113,20 @@
       expect(publish_span[:origin]).to eq("auto.queue.active_job")
     end
   end
+
+  context "when tracing is disabled and the job carries no trace headers" do
+    let(:configure_sentry) { proc { |config| config.rails.active_job_propagate_traces = false } }
+
+    it "keeps the worker's trace on the captured error event" do
+      worker_trace_id = Sentry.get_main_hub.current_scope.propagation_context.trace_id
+
+      expect do
+        failing_job.perform_later
+        drain
+      end.to raise_error(RuntimeError, /boom from failing_job spec/)
+
+      error_event = sentry_events.find { |e| e.is_a?(Sentry::ErrorEvent) }
+      expect(error_event.contexts.dig(:trace, :trace_id)).to eq(worker_trace_id)
+    end
+  end
 end
diff --git a/sentry-rails/spec/active_job/support/harness.rb b/sentry-rails/spec/active_job/support/harness.rb
index ca4ed21f5..74aee6c2d 100644
--- a/sentry-rails/spec/active_job/support/harness.rb
+++ b/sentry-rails/spec/active_job/support/harness.rb
@@ -27,6 +27,12 @@ def perform; end
     end
   end
 
+  let(:greeting_job) do
+    job_fixture do
+      def perform(_greeting); end
+    end
+  end
+
   let(:failing_job) do
     job_fixture do
       def perform(*_args, **_kwargs)
diff --git a/sentry-rails/spec/active_job/test_adapter_spec.rb b/sentry-rails/spec/active_job/test_adapter_spec.rb
index e3093d938..d968699d6 100644
--- a/sentry-rails/spec/active_job/test_adapter_spec.rb
+++ b/sentry-rails/spec/active_job/test_adapter_spec.rb
@@ -8,4 +8,27 @@
 
   it_behaves_like "a Sentry-instrumented ActiveJob backend"
   it_behaves_like "an ActiveJob backend that supports distributed tracing"
+
+  context "when the job data cannot be encoded as JSON" do
+    let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
+    let(:unencodable_argument) { "\xFF".dup.force_encoding(Encoding::UTF_8) }
+
+    it "enqueues and performs the job and omits messaging.message.body.size" do
+      performed_arguments = []
+      recording_job = job_fixture do
+        define_method(:perform) { |argument| performed_arguments << argument }
+      end
+
+      within_parent_transaction { recording_job.perform_later(unencodable_argument) }
+      drain
+
+      publish_span = transactions.flat_map(&:spans).find { |s| s[:op] == "queue.publish" }
+      consumer_data = consumer_transaction.contexts.dig(:trace, :data)
+      expect(performed_arguments).to eq([unencodable_argument])
+      expect(publish_span[:data]).to include("messaging.destination.name" => "default")
+      expect(publish_span[:data]).not_to have_key("messaging.message.body.size")
+      expect(consumer_data).to include("messaging.destination.name" => "default")
+      expect(consumer_data).not_to have_key("messaging.message.body.size")
+    end
+  end
 end
diff --git a/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb b/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb
new file mode 100644
index 000000000..ea8927f33
--- /dev/null
+++ b/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb
@@ -0,0 +1,551 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+RSpec.describe "Sentry + ActiveJob with a registered job extension", type: :job do
+  include_context "active_job backend harness", adapter: :test
+  include_context "test adapter"
+
+  let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
+
+  def register_extension(&body)
+    klass = Class.new(Sentry::JobExtensions::Extension) do
+      def handles?(_job)
+        true
+      end
+
+      class_eval(&body) if body
+    end
+
+    Sentry.register_job_extension(klass.new(name: "test_backend"))
+  end
+
+  def perform_failing_job
+    expect do
+      failing_job.perform_later
+      drain
+    end.to raise_error(RuntimeError, /boom from failing_job spec/)
+
+    error_event
+  end
+
+  def error_event
+    sentry_events.find { |event| event.is_a?(Sentry::ErrorEvent) }
+  end
+
+  def publish_span
+    parent = transactions.find { |t| t.contexts.dig(:trace, :op) == "test" }
+    parent.spans.find { |span| span[:op] == "queue.publish" }
+  end
+
+  def register_enriching_extension
+    register_extension do
+      def scope_data(job:)
+        {
+          contexts: { backend: { "jid" => "backend-jid" }, active_job: { job_id: "backend", attempt: 3 } },
+          tags: { "backend.queue" => "low", queue: "backend-queue" },
+          extra: { backend_payload_size: 12, job_id: "backend" }
+        }
+      end
+
+      def error_data(job:, exception:)
+        {
+          contexts: { backend_error: { "retried" => true }, active_job: { job_id: "backend" } },
+          tags: { "backend.error" => "yes", job_id: "backend", provider_job_id: "backend" },
+          extra: { backend_attempts: 2, active_job: "backend" }
+        }
+      end
+
+      def messaging_data(job:)
+        { system: "backend", message_id: "backend", destination: "backend", receive_latency: 42.0 }
+      end
+    end
+  end
+
+  describe "scope data" do
+    before { register_enriching_extension }
+
+    it "adds the extension's contexts, tags and extra to the consumer transaction while keeping ActiveJob's values" do
+      successful_job.perform_later
+      drain
+
+      transaction = consumer_transaction
+
+      expect(transaction.contexts[:backend]).to eq("jid" => "backend-jid")
+      expect(transaction.contexts[:active_job]).to include(job_class: successful_job.name, attempt: 3)
+      expect(transaction.contexts[:active_job][:job_id]).not_to eq("backend")
+      expect(transaction.tags).to include("backend.queue" => "low", queue: "default")
+      expect(transaction.extra).to include(backend_payload_size: 12)
+    end
+
+    it "adds the extension's scope data to the error event while keeping ActiveJob's values" do
+      event = perform_failing_job
+
+      expect(event.contexts[:backend]).to eq("jid" => "backend-jid")
+      expect(event.tags).to include("backend.queue" => "low", queue: "default")
+      expect(event.extra).to include(backend_payload_size: 12)
+      expect(event.extra[:job_id]).not_to eq("backend")
+    end
+  end
+
+  describe "error data" do
+    before { register_enriching_extension }
+
+    it "adds the extension's error data to the captured event while keeping ActiveJob's values" do
+      event = perform_failing_job
+
+      expect(event.contexts[:backend_error]).to eq("retried" => true)
+      expect(event.tags).to include("backend.error" => "yes")
+      expect(event.extra).to include(backend_attempts: 2, active_job: failing_job.name)
+      expect(event.tags[:job_id]).to eq(event.extra[:job_id])
+      expect(event.tags[:provider_job_id]).to eq(event.extra[:provider_job_id])
+      expect(event.contexts[:active_job][:job_id]).to eq(event.extra[:job_id])
+    end
+  end
+
+  describe "error data from a changing job" do
+    it "keeps extension context values from before execution when the job fails" do
+      register_extension do
+        def scope_data(job:)
+          { contexts: { backend: { attempt: job.executions } } }
+        end
+
+        def error_data(job:, exception:)
+          { contexts: { backend: { attempt: job.executions } } }
+        end
+      end
+
+      expect(perform_failing_job.contexts[:backend]).to eq(attempt: 0)
+    end
+  end
+
+  describe "string-keyed error context" do
+    it "preserves serialized ActiveJob metadata while adding backend fields" do
+      register_extension do
+        def error_data(job:, exception:)
+          { contexts: { "active_job" => { "job_id" => "backend-id", "attempt" => 3 } } }
+        end
+      end
+
+      event = JSON.parse(JSON.generate(perform_failing_job.to_h))
+
+      expect(event.dig("contexts", "active_job")).to include(
+        "job_id" => event.dig("extra", "job_id"),
+        "job_class" => failing_job.name,
+        "queue" => "default",
+        "provider_job_id" => event.dig("extra", "provider_job_id"),
+        "attempt" => 3
+      )
+    end
+  end
+
+  describe "error data overlapping the job's scope" do
+    it "does not let the extension's error data replace tags and extra already on the job's scope" do
+      register_extension do
+        def scope_data(job:)
+          { tags: { "backend.queue" => "low" }, extra: { backend_payload_size: 12 } }
+        end
+
+        def error_data(job:, exception:)
+          { tags: { "backend.queue" => "high", queue: "backend-queue" }, extra: { backend_payload_size: 99 } }
+        end
+      end
+
+      event = perform_failing_job
+
+      expect(event.tags).to include("backend.queue" => "low", queue: "default")
+      expect(event.extra).to include(backend_payload_size: 12)
+    end
+  end
+
+  describe "error data with Rails execution context", skip: RAILS_VERSION < 7.0 do
+    context "when queue data collection is enabled" do
+      let(:configure_sentry) do
+        proc do |config|
+          config.traces_sample_rate = 1.0
+          config.data_collection.queues = true
+        end
+      end
+
+      let(:failing_job) do
+        job_fixture do
+          def perform
+            Rails.error.set_context(debug_key: "from_job")
+            raise "boom from failing_job spec"
+          end
+        end
+      end
+
+      it "keeps ActiveJob's execution context alongside the extension's contexts" do
+        register_extension do
+          def error_data(job:, exception:)
+            { contexts: { "rails.error" => { debug_key: "backend" }, backend_error: { "retried" => true } } }
+          end
+        end
+
+        event = perform_failing_job
+
+        expect(event.contexts["rails.error"]).to include(debug_key: "from_job")
+        expect(event.contexts[:backend_error]).to eq("retried" => true)
+      end
+    end
+  end
+
+  describe "messaging data" do
+    before { register_enriching_extension }
+
+    it "adds the extension's messaging data to the producer span only where ActiveJob sets none" do
+      within_parent_transaction do
+        successful_job.set(queue: "events").perform_later
+      end
+
+      expect(publish_span[:data]).to include(
+        "messaging.system" => successful_job.queue_adapter_name,
+        "messaging.destination.name" => "events",
+        "messaging.message.receive.latency" => 42.0
+      )
+      expect(publish_span[:data]["messaging.message.id"]).not_to eq("backend")
+    end
+
+    it "reports the extension's messaging data on the consumer transaction in place of ActiveJob's" do
+      successful_job.set(queue: "events").perform_later
+      drain
+
+      expect(consumer_transaction.contexts.dig(:trace, :data)).to include(
+        "messaging.system" => "backend",
+        "messaging.message.id" => "backend",
+        "messaging.destination.name" => "backend",
+        "messaging.message.receive.latency" => 42.0
+      )
+    end
+  end
+
+  describe "user" do
+    let(:configure_sentry) do
+      proc do |config|
+        config.traces_sample_rate = 1.0
+        config.send_default_pii = true
+      end
+    end
+
+    it "replaces the user the job carries with the user the extension supplies" do
+      register_extension do
+        def scope_data(job:)
+          { user: { "id" => "backend-user", "segment" => "vip" } }
+        end
+      end
+      Sentry.set_user(id: "web-user")
+
+      successful_job.perform_later
+      Sentry.set_user({})
+
+      expect(last_enqueued_payload.dig("_sentry", "user")).to eq("id" => "web-user")
+
+      drain
+
+      expect(consumer_transaction.user).to eq("id" => "backend-user", "segment" => "vip")
+    end
+  end
+
+  describe "retry info" do
+    def register_retrying_extension(count:, will_retry:)
+      register_extension do
+        define_method(:retry_info) { |job:| { count: count, will_retry: will_retry } }
+      end
+    end
+
+    it "reports the extension's retry count on the producer span and the consumer transaction" do
+      register_retrying_extension(count: 4, will_retry: false)
+
+      within_parent_transaction { successful_job.perform_later }
+      drain
+
+      expect(publish_span[:data]).to include("messaging.message.retry.count" => 4)
+      expect(consumer_transaction.contexts.dig(:trace, :data)).to include("messaging.message.retry.count" => 4)
+    end
+
+    it "does not capture the job's exception but still fails the transaction when the backend will retry" do
+      register_retrying_extension(count: 1, will_retry: true)
+
+      expect do
+        failing_job.perform_later
+        drain
+      end.to raise_error(RuntimeError, /boom from failing_job spec/)
+
+      expect(error_event).to be_nil
+      expect(consumer_transaction.contexts.dig(:trace, :status)).to eq("internal_error")
+    end
+
+    it "captures the job's exception when the backend will not retry" do
+      register_retrying_extension(count: 5, will_retry: false)
+
+      expect(perform_failing_job).not_to be_nil
+    end
+  end
+
+  describe "exception reporting" do
+    def register_deciding_extension(report:, will_retry:)
+      register_extension do
+        define_method(:retry_info) { |job:| { count: 1, will_retry: will_retry } }
+        define_method(:report_exception?) { |job:, exception:| report }
+      end
+    end
+
+    it "captures the job's exception when the extension reports it, even though the backend will retry" do
+      register_deciding_extension(report: true, will_retry: true)
+
+      expect(perform_failing_job).not_to be_nil
+    end
+
+    it "does not capture the job's exception when the extension declines it, but still fails the transaction" do
+      register_deciding_extension(report: false, will_retry: false)
+
+      expect(perform_failing_job).to be_nil
+      expect(consumer_transaction.contexts.dig(:trace, :status)).to eq("internal_error")
+    end
+  end
+
+  describe "retried exception reporting", skip: RAILS_VERSION < 6.0 do
+    let(:retrying_job) do
+      job_fixture do
+        retry_on StandardError, attempts: 2, wait: 0
+
+        def perform
+          raise "boom from retrying_job spec" if executions == 1
+        end
+      end
+    end
+
+    def register_retry_deciding_extension(report)
+      register_extension do
+        define_method(:report_retried_exception?) { |job:, exception:| report }
+      end
+    end
+
+    def perform_retrying_job
+      retrying_job.perform_later
+      drain
+      error_event
+    end
+
+    it "does not capture an exception ActiveJob retries when the extension declines it, even with active_job_report_on_retry_error" do
+      Sentry.configuration.rails.active_job_report_on_retry_error = true
+      register_retry_deciding_extension(false)
+
+      expect(perform_retrying_job).to be_nil
+    end
+
+    it "captures an exception ActiveJob retries when the extension reports it, even without active_job_report_on_retry_error" do
+      register_retry_deciding_extension(true)
+
+      expect(perform_retrying_job&.exception&.values&.first&.value).to include("boom from retrying_job spec")
+    end
+  end
+
+  describe "trace headers" do
+    let(:foreign_trace_id) { "a" * 32 }
+    let(:foreign_trace_headers) { { "sentry-trace" => "#{foreign_trace_id}-#{"b" * 16}-1" } }
+
+    def register_extension_extracting(headers)
+      register_extension do
+        define_method(:extract_trace_headers) { |job:, payload:| headers }
+      end
+    end
+
+    it "lets the extension carry the trace headers in the serialized payload alongside _sentry" do
+      register_extension do
+        def inject_trace_headers(job:, payload:, headers:)
+          payload.merge("backend_trace" => headers["sentry-trace"])
+        end
+      end
+
+      within_parent_transaction { successful_job.perform_later }
+
+      payload = last_enqueued_payload
+      expect(payload["backend_trace"]).not_to be_nil
+      expect(payload["backend_trace"]).to eq(payload["_sentry"]["trace_propagation_headers"]["sentry-trace"])
+    end
+
+    context "when the job carries no _sentry trace headers" do
+      let(:configure_sentry) do
+        proc do |config|
+          config.traces_sample_rate = 1.0
+          config.rails.active_job_propagate_traces = false
+        end
+      end
+
+      it "continues the trace the extension extracts" do
+        register_extension_extracting(foreign_trace_headers)
+
+        successful_job.perform_later
+        drain
+
+        expect(consumer_transaction.contexts.dig(:trace, :trace_id)).to eq(foreign_trace_id)
+      end
+    end
+
+    it "continues the _sentry trace even when the extension extracts different headers" do
+      register_extension_extracting(foreign_trace_headers)
+      parent_trace_id = nil
+
+      within_parent_transaction do |parent|
+        parent_trace_id = parent.trace_id
+        successful_job.perform_later
+      end
+      drain
+
+      expect(consumer_transaction.contexts.dig(:trace, :trace_id)).to eq(parent_trace_id)
+    end
+  end
+
+  describe "naming" do
+    let(:backend_integration) do
+      Module.new do
+        extend Sentry::Integrable
+
+        def self.integration_name
+          "backend"
+        end
+      end
+    end
+
+    def mechanism_type(event)
+      event.exception.values.first.mechanism.type
+    end
+
+    before do
+      integration = backend_integration
+      register_extension do
+        define_method(:transaction_name) { |job:| "Backend/#{job.class.name}" }
+        define_method(:span_origin) { |job:| "auto.queue.backend" }
+        define_method(:publish_description) { |job:| "Backend/#{job.class.name}" }
+        define_method(:integration) { |job:| integration }
+      end
+    end
+
+    it "names the consumer transaction and sets its origin as the extension supplies" do
+      successful_job.perform_later
+      drain
+
+      transaction = consumer_transaction
+
+      expect(transaction.transaction).to eq("Backend/#{successful_job.name}")
+      expect(transaction.transaction_info).to eq(source: :task)
+      expect(transaction.contexts.dig(:trace, :origin)).to eq("auto.queue.backend")
+    end
+
+    it "describes the producer span and sets its origin as the extension supplies" do
+      within_parent_transaction { successful_job.perform_later }
+
+      expect(publish_span[:description]).to eq("Backend/#{successful_job.name}")
+      expect(publish_span[:origin]).to eq("auto.queue.backend")
+    end
+
+    it "captures the job's exception under the extension's integration and transaction name" do
+      event = perform_failing_job
+
+      expect(mechanism_type(event)).to eq("backend")
+      expect(event.transaction).to eq("Backend/#{failing_job.name}")
+    end
+  end
+
+  describe "enqueue instrumentation" do
+    let(:configure_sentry) do
+      proc do |config|
+        config.traces_sample_rate = 1.0
+        config.data_collection.user_info = true
+      end
+    end
+
+    it "leaves the producer span, trace headers and user to an extension that instruments enqueueing the job" do
+      register_extension do
+        def producer(job:)
+          Sentry::JobExtensions::Producer::Delegated.new
+        end
+      end
+      Sentry.set_user(id: "42")
+
+      within_parent_transaction { successful_job.perform_later }
+
+      expect(publish_span).to be_nil
+      expect(last_enqueued_payload).not_to have_key("_sentry")
+    end
+  end
+
+  describe "when the extension adds nothing" do
+    let(:extension_data_keys) { [:backend, "backend.queue", :backend_payload_size, :backend_error, "backend.error", :backend_attempts] }
+
+    def expect_no_extension_data(event)
+      expect(event.contexts.keys & extension_data_keys).to be_empty
+      expect(event.tags.keys & extension_data_keys).to be_empty
+      expect(event.extra.keys & extension_data_keys).to be_empty
+    end
+
+    it "sends nothing from an extension that does not handle the job" do
+      register_enriching_extension.define_singleton_method(:handles?) { |_job| false }
+
+      within_parent_transaction { failing_job.perform_later }
+      expect { drain }.to raise_error(RuntimeError, /boom from failing_job spec/)
+
+      expect_no_extension_data(error_event)
+      expect_no_extension_data(consumer_transaction)
+      expect(publish_span[:data]).not_to have_key("messaging.message.receive.latency")
+    end
+
+    it "never consults the registry for jobs on a skipped adapter" do
+      consulted = []
+      Sentry.register_job_extension(
+        Class.new(Sentry::JobExtensions::Extension) do
+          define_method(:handles?) do |job|
+            consulted << job
+            true
+          end
+        end.new(name: "observer")
+      )
+      Sentry.configuration.rails.skippable_job_adapters = [failing_job.queue_adapter.class.to_s]
+
+      expect do
+        failing_job.perform_later
+        drain
+      end.to raise_error(RuntimeError, /boom from failing_job spec/)
+
+      expect(consulted).to be_empty
+    end
+
+    {
+      "raises" => proc { raise "extension failure" },
+      "returns malformed data" => proc { :malformed }
+    }.each do |behavior, hook_body|
+      context "when every hook #{behavior}" do
+        before do
+          register_extension do
+            %i[scope_data error_data messaging_data publish_messaging_data transaction_name span_origin publish_description integration producer].each do |hook|
+              define_method(hook) { |**_args| instance_exec(&hook_body) }
+            end
+          end
+        end
+
+        it "still performs the job and finishes its transaction" do
+          within_parent_transaction { successful_job.perform_later }
+          drain
+
+          expect(consumer_transaction.contexts.dig(:trace, :status)).to eq("ok")
+          expect(consumer_transaction.transaction).to eq(successful_job.name)
+          expect(consumer_transaction.contexts.dig(:trace, :origin)).to eq("auto.queue.active_job")
+          expect(publish_span[:data]).not_to have_key("messaging.message.receive.latency")
+          expect(publish_span[:description]).to eq(successful_job.name)
+          expect(publish_span[:origin]).to eq("auto.queue.active_job")
+        end
+
+        it "still captures the job's exception with ActiveJob's data" do
+          event = perform_failing_job
+
+          expect(event.extra).to include(active_job: failing_job.name)
+          expect(event.tags[:job_id]).to eq(event.extra[:job_id])
+          expect(event.transaction).to eq(failing_job.name)
+          expect(event.exception.values.first.mechanism.type).to eq("rails")
+        end
+      end
+    end
+  end
+end
diff --git a/sentry-rails/spec/sentry/rails/active_job_ownership_spec.rb b/sentry-rails/spec/sentry/rails/active_job_ownership_spec.rb
new file mode 100644
index 000000000..3cc14c575
--- /dev/null
+++ b/sentry-rails/spec/sentry/rails/active_job_ownership_spec.rb
@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+RSpec.describe "Sentry::Rails.active_job_owns_adapter?" do
+  let(:adapter_class_name) { "ActiveJob::QueueAdapters::SidekiqAdapter" }
+
+  it "is true when Sentry is initialized and the adapter is not skipped" do
+    make_basic_app
+
+    expect(Sentry::Rails.active_job_owns_adapter?(adapter_class_name)).to be(true)
+  end
+
+  it "is false when the adapter is in skippable_job_adapters" do
+    make_basic_app do |config|
+      config.rails.skippable_job_adapters = [adapter_class_name]
+    end
+
+    expect(Sentry::Rails.active_job_owns_adapter?(adapter_class_name)).to be(false)
+  end
+
+  it "is false when Sentry is not initialized" do
+    expect(Sentry::Rails.active_job_owns_adapter?(adapter_class_name)).to be(false)
+  end
+end
diff --git a/sentry-rails/spec/spec_helper.rb b/sentry-rails/spec/spec_helper.rb
index 1cdd3e617..82c56f508 100644
--- a/sentry-rails/spec/spec_helper.rb
+++ b/sentry-rails/spec/spec_helper.rb
@@ -88,6 +88,8 @@
 
     reset_sentry_globals!
 
+    Sentry.job_extensions.clear
+
     Rails.application&.cleanup!
   end