From caf345aa39a4430dc9689fd0eca2e0bc433cccf5 Mon Sep 17 00:00:00 2001
From: Peter Solnica <peter@solnica.online>
Date: Wed, 4 Nov 2026 12:41:51 +0000
Subject: [PATCH 1/2] feat(rails): run ActiveJob reporting through job
 extensions

Closes #3119
---
 sentry-rails/lib/sentry/rails.rb              |  45 ++
 sentry-rails/lib/sentry/rails/active_job.rb   | 231 ++-----
 .../lib/sentry/rails/active_job_extension.rb  | 189 ++++++
 .../shared_examples/error_context.rb          |  13 +
 .../tracing/consumer_transaction.rb           |  39 +-
 .../tracing/messaging_span_data.rb            |  18 +
 .../shared_examples/tracing/producer_span.rb  |  20 +
 .../tracing/trace_propagation.rb              |  42 ++
 .../spec/active_job/support/harness.rb        |   6 +
 .../spec/active_job/test_adapter_spec.rb      |  23 +
 .../rails/active_job_job_extensions_spec.rb   | 569 ++++++++++++++++++
 .../sentry/rails/active_job_ownership_spec.rb |  25 +
 sentry-rails/spec/spec_helper.rb              |   2 +
 13 files changed, 1035 insertions(+), 187 deletions(-)
 create mode 100644 sentry-rails/lib/sentry/rails/active_job_extension.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 203709b40..462a035fe 100644
--- a/sentry-rails/lib/sentry/rails.rb
+++ b/sentry-rails/lib/sentry/rails.rb
@@ -5,6 +5,7 @@
 require "sentry/integrable"
 require "sentry/rails/tracing"
 require "sentry/rails/configuration"
+require "sentry/rails/active_job_extension"
 require "sentry/rails/structured_logging"
 require "sentry/rails/engine"
 require "sentry/rails/railtie"
@@ -14,5 +15,49 @@ module Sentry
   module Rails
     extend Integrable
     register_integration name: "rails", version: Sentry::Rails::VERSION
+
+    DEFAULT_ACTIVE_JOB_EXTENSION = ActiveJobExtension.new
+    private_constant :DEFAULT_ACTIVE_JOB_EXTENSION
+
+    # Returns whether sentry-rails reports ActiveJob jobs run through the given
+    # queue adapter, in which case the backend gem should not report them itself.
+    # The result depends on the current configuration, so check it when a job is
+    # handled rather than caching it.
+    #
+    # @param adapter_class_name [String, Class]
+    # @return [Boolean]
+    def self.active_job_owns_adapter?(adapter_class_name)
+      Sentry.initialized? && !Sentry.configuration.rails.skippable_job_adapters.include?(adapter_class_name.to_s)
+    end
+
+    # Registers the extension for ActiveJob jobs run through the given queue
+    # adapter. Raises ArgumentError if it is not an {ActiveJobExtension}.
+    #
+    # @param adapter [Symbol] the adapter name, such as `:sidekiq`
+    # @param extension [ActiveJobExtension]
+    # @return [ActiveJobExtension]
+    def self.register_active_job_extension(adapter, extension)
+      unless extension.is_a?(ActiveJobExtension)
+        raise ArgumentError, "ActiveJob extension must be an instance of #{ActiveJobExtension}, got #{extension.inspect}"
+      end
+
+      active_job_extensions[adapter.to_sym] = extension
+    end
+
+    # Returns the extension for the job class' queue adapter, or the default
+    # {ActiveJobExtension} when none is registered.
+    #
+    # @param job_class [Class]
+    # @return [ActiveJobExtension]
+    def self.active_job_extension_for(job_class)
+      active_job_extensions.fetch(ActiveJobExtension.adapter_name(job_class), DEFAULT_ACTIVE_JOB_EXTENSION)
+    end
+
+    # Returns the registered extensions by adapter name.
+    #
+    # @return [Hash{Symbol => ActiveJobExtension}]
+    def self.active_job_extensions
+      @active_job_extensions ||= {}
+    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..6137f43ac 100644
--- a/sentry-rails/lib/sentry/rails/active_job.rb
+++ b/sentry-rails/lib/sentry/rails/active_job.rb
@@ -3,6 +3,7 @@
 require "set"
 require "sentry/rails/serializer"
 require "sentry/rails/error_reporter_context"
+require "sentry/rails/active_job_extension"
 
 module Sentry
   module Rails
@@ -13,44 +14,21 @@ module ActiveJobExtensions
 
       def self.prepended(base)
         base.attr_accessor :_sentry
+        base.attr_reader :_sentry_job_data
       end
 
       def perform_now
         if !Sentry.initialized? || already_supported_by_sentry_integration?
           super
         else
-          data = _sentry || {}
-          SentryReporter.record(
-            self,
-            trace_headers: data["trace_propagation_headers"],
-            user: data["user"]
-          ) { super }
+          SentryReporter.record(self) { super }
         end
       end
 
       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.prepare_payload(self, payload)
       end
 
       def deserialize(job_data)
@@ -58,6 +36,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,17 +44,12 @@ 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.
+        OP_NAME = Sentry::JobExtensions::Extension::PROCESS_OP
+        SPAN_ORIGIN = ActiveJobExtension::SPAN_ORIGIN
         MESSAGING_SYSTEM_FALLBACK = "activejob"
 
         EVENT_HANDLERS = {
@@ -83,7 +57,9 @@ class SentryReporter
         }
 
         class << self
-          include ErrorReporterContext
+          def prepare_payload(job, payload)
+            Sentry::Rails.active_job_extension_for(job.class).prepare_payload(job: job, payload: payload)
+          end
 
           def producer_callback_registered?
             @producer_callback_registered ||= false
@@ -96,143 +72,38 @@ def producer_callback_registered!
           def record_producer_span(job, &enqueue)
             return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration?
 
-            enqueued = false
-
-            run_enqueue = lambda do
-              enqueued = true
-              enqueue.call
-            end
-
-            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
-
-              log_producer_span_error(e)
+            Sentry::Rails.active_job_extension_for(job.class).instrument_enqueue(job: job, &enqueue)
+          end
 
-              run_enqueue.call
+          def record(job, &block)
+            with_isolated_scope do |scope|
+              record_in_scope(scope, job, &block)
             end
           end
 
-          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_in_scope(scope, job)
+            extension = Sentry::Rails.active_job_extension_for(job.class)
+            transaction = nil
 
-          def record(job, trace_headers: nil, user: nil, &block)
-            original_hub = Sentry.get_current_hub_internal
-            Sentry.clone_hub_to_current_thread
+            begin
+              transaction = extension.start_job(scope: scope, job: job)
 
-            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)
-
-                capture_exception(job, e)
-
-                raise
+              yield.tap do
+                extension.finish_job(transaction: transaction, status: 200)
               end
-            end
-          ensure
-            Sentry.set_current_hub_internal(original_hub)
-          end
+            rescue Exception => e # rubocop:disable Lint/RescueException
+              extension.finish_job(transaction: transaction, status: 500)
 
-          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 extension.safely("decide whether to report the job exception", fallback: true) { extension.report_exception?(job: job, exception: e) }
+                extension.capture_exception(job: job, exception: e)
+              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)
+            Sentry::Rails.active_job_extension_for(job.class).capture_exception(job: job, exception: e)
           end
 
           def register_event_handlers
@@ -255,9 +126,14 @@ 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
 
-              capture_exception(job, error)
+              extension = Sentry::Rails.active_job_extension_for(job.class)
+              report = extension.safely("decide whether to report the retried exception", fallback: Sentry.configuration.rails.active_job_report_on_retry_error) do
+                extension.report_retried_exception?(job: job, exception: error)
+              end
+              return unless report
+
+              extension.capture_exception(job: job, exception: error)
             end
           end
 
@@ -266,35 +142,18 @@ def handle_error_event(*args)
             yield(event.payload[:job], event.payload[:error])
           end
 
-          def finish_sentry_transaction(transaction, status)
-            return unless transaction
+          private :record_in_scope
 
-            transaction.set_http_status(status)
-            transaction.finish
-          end
-
-          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_extension.rb b/sentry-rails/lib/sentry/rails/active_job_extension.rb
new file mode 100644
index 000000000..af3e02672
--- /dev/null
+++ b/sentry-rails/lib/sentry/rails/active_job_extension.rb
@@ -0,0 +1,189 @@
+# frozen_string_literal: true
+
+require "sentry/rails/serializer"
+require "sentry/rails/error_reporter_context"
+
+module Sentry
+  module Rails
+    # The job extension sentry-rails uses for ActiveJob jobs. Backend gems
+    # subclass it to enrich ActiveJob jobs run through their queue adapter, and
+    # register it with {Sentry::Rails.register_active_job_extension}.
+    class ActiveJobExtension < Sentry::JobExtensions::Extension
+      include ErrorReporterContext
+
+      SPAN_ORIGIN = "auto.queue.active_job"
+
+      # @param serializer [#serialize] serializes job arguments for error events
+      # @see Sentry::JobExtensions::Extension#initialize
+      def initialize(name: :active_job, integration: Sentry::Rails, origin: SPAN_ORIGIN, serializer: Sentry::Rails::Serializer)
+        super(name: name, integration: integration, origin: origin)
+        @serializer = serializer
+      end
+
+      # Returns the name of the job class' queue adapter.
+      #
+      # @param job_class [Class]
+      # @return [Symbol]
+      def self.adapter_name(job_class)
+        name = job_class.queue_adapter_name if job_class.respond_to?(:queue_adapter_name)
+        (name || job_class.queue_adapter.class.name.demodulize.delete_suffix("Adapter").underscore).to_sym
+      end
+
+      # Returns the job's serialized payload carrying Sentry's trace propagation
+      # headers and user data.
+      #
+      # @param job [ActiveJob::Base]
+      # @param payload [Hash]
+      # @return [Hash]
+      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: job, payload: payload, headers: headers)
+          end
+        end
+
+        if Sentry.configuration.data_collection.user_info
+          user = Sentry.get_current_scope.user.transform_keys(&:to_s).slice(*ActiveJobExtensions::USER_FIELDS_ALLOWLIST)
+          sentry_data["user"] = user unless user.empty?
+        end
+
+        sentry_data.empty? ? payload : payload.merge(ActiveJobExtensions::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
+
+      # Enqueues the job even when recording its `queue.publish` span fails.
+      #
+      # @see Sentry::JobExtensions::Extension#instrument_enqueue
+      def instrument_enqueue(job:, **context, &enqueue)
+        enqueued = false
+        super(job: job, **context) do
+          enqueued = true
+          enqueue.call
+        end
+      rescue StandardError => e
+        raise if enqueued
+
+        log_failure("record producer span", e)
+        enqueue.call
+      end
+
+      # Also applies the user and ActiveJob context the job carries.
+      #
+      # @see Sentry::JobExtensions::Extension#start_job
+      def start_job(scope:, job:, **context)
+        user = job._sentry&.[]("user")
+        scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?
+        scope.set_transaction_name(job.class.name, source: :task)
+
+        super.tap { apply_active_job_context(scope, job) }
+      end
+
+      # Also adds the job's ActiveJob data to the event.
+      #
+      # @see Sentry::JobExtensions::Extension#capture_exception
+      def capture_exception(job:, exception:, contexts: {}, tags: {}, extra: {}, **context)
+        data = active_job_error_data(job)
+
+        super(
+          job: job,
+          exception: exception,
+          contexts: data[:contexts].merge(contexts),
+          tags: data[:tags].merge(tags),
+          extra: data[:extra].merge(extra),
+          **context
+        )
+      end
+
+      # Reads the trace propagation headers stored in the job's Sentry payload.
+      #
+      # @see Sentry::JobExtensions::Extension#extract_trace_headers
+      def extract_trace_headers(job:, **)
+        job._sentry&.[]("trace_propagation_headers") || {}
+      end
+
+      # Extends {#publish_messaging_data} with the job's receive latency and body size.
+      #
+      # @see Sentry::JobExtensions::Extension#messaging_data
+      def messaging_data(job:, **context)
+        data = publish_messaging_data(job: job, **context)
+        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) if Sentry.get_current_scope.get_span&.sampled
+        data.compact
+      end
+
+      # Describes the job with its ActiveJob messaging attributes.
+      #
+      # @see Sentry::JobExtensions::Extension#publish_messaging_data
+      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,
+          retry_count: (job.executions.to_i if job.respond_to?(:executions))
+        }.compact
+      end
+
+      # Follows `config.rails.active_job_report_on_retry_error`.
+      #
+      # @see Sentry::JobExtensions::Extension#report_retried_exception?
+      def report_retried_exception?(job:, exception:, **)
+        Sentry.configuration.rails.active_job_report_on_retry_error
+      end
+
+      private
+
+      def log_failure(context, error)
+        Sentry.sdk_logger&.error("sentry-rails: job extension #{name.inspect} failed to #{context}: #{error.class}: #{error.message}\n  #{Array(error.backtrace).first(5).join("\n  ")}")
+      end
+
+      def body_size(payload)
+        JSON.generate(payload).bytesize if payload
+      rescue StandardError
+        nil
+      end
+
+      def apply_active_job_context(scope, job)
+        required = {
+          job_class: job.class.name,
+          job_id: job.job_id,
+          queue: job.queue_name,
+          provider_job_id: job.provider_job_id
+        }
+        scope.set_contexts(active_job: scope.contexts.fetch(:active_job, {}).merge(required))
+        scope.set_tags(queue: job.queue_name)
+      end
+
+      def active_job_error_data(job)
+        queues = Sentry.configuration.data_collection.queues
+        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
+        }
+        arguments = serialize_arguments(job.arguments) if queues
+        extra[:arguments] = arguments unless arguments.nil?
+
+        {
+          contexts: queues ? execution_context : {},
+          tags: { job_id: job.job_id, provider_job_id: job.provider_job_id },
+          extra: extra
+        }
+      end
+
+      def serialize_arguments(arguments)
+        @serializer.serialize(arguments)
+      rescue StandardError => e
+        Sentry.sdk_logger&.error("sentry-rails: failed to serialize job arguments: #{e.class}: #{e.message}")
+        nil
+      end
+    end
+  end
+end
diff --git a/sentry-rails/spec/active_job/shared_examples/error_context.rb b/sentry-rails/spec/active_job/shared_examples/error_context.rb
index 69edbb7ee..58e322255 100644
--- a/sentry-rails/spec/active_job/shared_examples/error_context.rb
+++ b/sentry-rails/spec/active_job/shared_examples/error_context.rb
@@ -58,6 +58,19 @@ def perform(_argument)
 
       expect(last_sentry_event.extra[:arguments]).to eq(["sensitive argument"])
     end
+
+    it "captures the original job error exactly once when an argument is a decimal range", skip: !defined?(::ActiveJob::Serializers::RangeSerializer) do
+      original = RuntimeError.new("boom from decimal range job")
+      job = job_fixture do
+        define_method(:perform) { |_range| raise original }
+      end
+      job.perform_later(1.5..2.5)
+
+      expect { drain }.to raise_error { |error| expect(error).to equal(original) }
+
+      expect(sentry_events.size).to eq(1)
+      expect(sentry_events.first.exception.values.first.value).to include(original.message)
+    end
   end
 
   context "with Rails.error.set_context data attached before the job raises", skip: RAILS_VERSION < 7.0 do
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..a19edb048 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,6 +20,32 @@
     expect(consumer_transaction.contexts.dig(:trace, :parent_span_id)).to eq(publish_span_id)
   end
 
+  [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
+  end
+
   it "injects both sentry-trace and baggage propagation headers into the serialized payload" do
     within_parent_transaction do
       successful_job.perform_later
@@ -98,4 +124,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..1c282183d
--- /dev/null
+++ b/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb
@@ -0,0 +1,569 @@
+# 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(adapter: :test, &body)
+    klass = Class.new(Sentry::Rails::ActiveJobExtension)
+    klass.class_eval(&body) if body
+
+    Sentry::Rails.register_active_job_extension(adapter, 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(adapter: :test)
+    register_extension(adapter: adapter) 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
+
+      def publish_messaging_data(job:, **)
+        messaging_data(job: job).merge(super)
+      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 an extension that does not call super" 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(_payload)
+          raise "boom from failing_job spec"
+        end
+      end
+    end
+
+    it "still attaches ActiveJob's job data and tags to the captured event" do
+      register_extension do
+        def error_data(job:, exception:, **)
+          { extra: { backend_attempts: 2 } }
+        end
+      end
+
+      expect do
+        failing_job.perform_later({ password: "example-password", delivery_id: "delivery-42" })
+        drain
+      end.to raise_error(RuntimeError, /boom from failing_job spec/)
+      event = error_event
+
+      expect(event.extra).to include(
+        backend_attempts: 2,
+        active_job: failing_job.name,
+        locale: "en",
+        arguments: [{ password: "example-password", delivery_id: "delivery-42" }]
+      )
+      expect(event.extra[:job_id]).not_to be_nil
+      expect(event.extra.keys).to include(:provider_job_id, :scheduled_at)
+      expect(event.tags).to include(job_id: event.extra[:job_id], provider_job_id: event.extra[:provider_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 "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 "inherited defaults" do
+    it "extends consumer metadata through super while inheriting tracing and exception reporting" do
+      register_extension do
+        def messaging_data(job:, **)
+          super.merge(system: "backend")
+        end
+      end
+      parent_trace_id = nil
+
+      within_parent_transaction do |parent|
+        parent_trace_id = parent.trace_id
+        failing_job.set(queue: "events").perform_later
+      end
+      expect { drain }.to raise_error(RuntimeError, /boom from failing_job spec/)
+      transaction = consumer_transaction
+
+      expect(transaction.contexts.dig(:trace, :data)).to include(
+        "messaging.system" => "backend",
+        "messaging.destination.name" => "events",
+        "messaging.message.id" => error_event.extra[:job_id]
+      )
+      expect(transaction.contexts.dig(:trace, :trace_id)).to eq(parent_trace_id)
+      expect(error_event.exception.values.first.mechanism.type).to eq("rails")
+    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 "retries" do
+    def register_retrying_extension(count:, will_retry:)
+      register_extension do
+        define_method(:messaging_data) { |job:, **| super(job: job).merge(retry_count: count) }
+        define_method(:publish_messaging_data) { |job:, **| super(job: job).merge(retry_count: count) }
+        define_method(:will_retry?) { |job:, **| 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(:will_retry?) { |job:, **| 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:, **| 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
+  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 to an extension whose backend records it, while still propagating the trace and user" do
+      register_extension do
+        def record_publish_span?(job:, **)
+          false
+        end
+      end
+      Sentry.set_user(id: "42")
+
+      within_parent_transaction { successful_job.perform_later }
+
+      expect(publish_span).to be_nil
+      expect(last_enqueued_payload["_sentry"]).to include("trace_propagation_headers", "user" => { "id" => "42" })
+    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 registered for another adapter" do
+      register_enriching_extension(adapter: :sidekiq)
+
+      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
+
+    context "when every hook raises" do
+      before do
+        register_extension do
+          %i[
+            record_publish_span? scope_data error_data messaging_data publish_messaging_data will_retry? report_exception?
+            report_retried_exception? extract_trace_headers inject_trace_headers
+            transaction_name span_origin publish_span_origin publish_description integration
+          ].each do |hook|
+            define_method(hook) { |**_args| raise "extension failure" }
+          end
+        end
+      end
+
+      let(:sdk_log) { StringIO.new }
+
+      before { Sentry.configuration.sdk_logger = ::Logger.new(sdk_log) }
+
+      it "still enqueues and performs the job once and logs the extension's failure" do
+        performed = []
+        job = job_fixture { define_method(:perform) { performed << job_id } }
+
+        within_parent_transaction { job.perform_later }
+        drain
+
+        expect(performed.size).to eq(1)
+        expect(sdk_log.string).to include("job extension :test_backend failed to", "extension failure")
+      end
+
+      it "still captures the job's exception with ActiveJob's data and re-raises it" do
+        event = perform_failing_job
+
+        expect(event.extra).to include(active_job: failing_job.name)
+        expect(event.tags).to include(queue: "default", job_id: event.extra[:job_id])
+        expect(event.contexts[:active_job]).to include(job_class: failing_job.name, queue: "default")
+        expect(event.transaction).to eq(failing_job.name)
+        expect(event.exception.values.first.mechanism.type).to eq("rails")
+      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..cde34bdbd 100644
--- a/sentry-rails/spec/spec_helper.rb
+++ b/sentry-rails/spec/spec_helper.rb
@@ -88,6 +88,8 @@
 
     reset_sentry_globals!
 
+    Sentry::Rails.active_job_extensions.clear
+
     Rails.application&.cleanup!
   end
 

From 15f96fd3ca77f73097e747fcac15caa58ee61129 Mon Sep 17 00:00:00 2001
From: Peter Solnica <peter@solnica.online>
Date: Wed, 4 Nov 2026 13:10:41 +0000
Subject: [PATCH 2/2] fix(rails): keep the _sentry payload when injecting trace
 headers fails

---
 .../lib/sentry/rails/active_job_extension.rb  |  2 +-
 .../rails/active_job_job_extensions_spec.rb   | 24 +++++++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/sentry-rails/lib/sentry/rails/active_job_extension.rb b/sentry-rails/lib/sentry/rails/active_job_extension.rb
index af3e02672..557d6798f 100644
--- a/sentry-rails/lib/sentry/rails/active_job_extension.rb
+++ b/sentry-rails/lib/sentry/rails/active_job_extension.rb
@@ -41,7 +41,7 @@ def prepare_payload(job:, payload:)
           headers = Sentry.get_trace_propagation_headers
           if headers && !headers.empty?
             sentry_data["trace_propagation_headers"] = headers
-            payload = inject_trace_headers(job: job, payload: payload, headers: headers)
+            payload = safely("inject trace headers", fallback: payload) { inject_trace_headers(job: job, payload: payload, headers: headers) }
           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
index 1c282183d..f52a2c675 100644
--- a/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb
+++ b/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb
@@ -414,6 +414,30 @@ def inject_trace_headers(job:, payload:, headers:, **)
       expect(payload["backend_trace"]).to eq(payload["_sentry"]["trace_propagation_headers"]["sentry-trace"])
     end
 
+    context "when the extension fails to inject the trace headers" do
+      let(:configure_sentry) do
+        proc do |config|
+          config.traces_sample_rate = 1.0
+          config.send_default_pii = true
+        end
+      end
+
+      it "keeps the trace headers and user in the serialized payload" do
+        register_extension do
+          def inject_trace_headers(**)
+            raise "extension failure"
+          end
+        end
+        Sentry.set_user(id: "web-user")
+
+        within_parent_transaction { successful_job.perform_later }
+
+        sentry_data = last_enqueued_payload["_sentry"]
+        expect(sentry_data["trace_propagation_headers"]).to include("sentry-trace")
+        expect(sentry_data["user"]).to eq("id" => "web-user")
+      end
+    end
+
     context "when the job carries no _sentry trace headers" do
       let(:configure_sentry) do
         proc do |config|