diff --git a/sentry-sidekiq/Rakefile b/sentry-sidekiq/Rakefile
index 9207ea5fe..648ec9fe3 100644
--- a/sentry-sidekiq/Rakefile
+++ b/sentry-sidekiq/Rakefile
@@ -1,13 +1,26 @@
 # frozen_string_literal: true
 
 require "bundler/gem_tasks"
-require_relative "../lib/sentry/test/rake_tasks"
-
-Sentry::Test::RakeTasks.define_spec_tasks(
-  spec_pattern: "spec/sentry/**/*_spec.rb",
-  spec_rspec_opts: "--order rand --format progress",
-  isolated_specs_pattern: "spec/isolated/**/*_spec.rb",
-  isolated_rspec_opts: "--format progress"
-)
+require "rspec/core/rake_task"
+
+RSpec::Core::RakeTask.new(:spec) do |task|
+  task.pattern = "spec/sentry/**/*_spec.rb"
+  task.rspec_opts = "--order rand --format progress"
+end
+
+namespace :spec do
+  isolated_tasks = Dir["spec/isolated/**/*_spec.rb"].sort.map do |file|
+    task_name = "isolated:#{file.delete_prefix("spec/isolated/").delete_suffix("_spec.rb")}"
+
+    RSpec::Core::RakeTask.new(task_name) do |task|
+      task.pattern = file
+      task.rspec_opts = "--format progress"
+    end
+
+    task_name
+  end
+
+  task isolated: isolated_tasks
+end
 
 task default: [:spec, :"spec:isolated"]
diff --git a/sentry-sidekiq/lib/sentry-sidekiq.rb b/sentry-sidekiq/lib/sentry-sidekiq.rb
index 4faeffcd2..98e5e27a6 100644
--- a/sentry-sidekiq/lib/sentry-sidekiq.rb
+++ b/sentry-sidekiq/lib/sentry-sidekiq.rb
@@ -5,6 +5,7 @@
 require "sentry/integrable"
 require "sentry/sidekiq/version"
 require "sentry/sidekiq/configuration"
+require "sentry/sidekiq/job_extension"
 require "sentry/sidekiq/error_handler"
 require "sentry/sidekiq/sentry_context_middleware"
 
@@ -14,18 +15,16 @@ module Sidekiq
 
     register_integration name: "sidekiq", version: Sentry::Sidekiq::VERSION
 
-    if defined?(::Rails::Railtie)
-      class Railtie < ::Rails::Railtie
-        config.after_initialize do
-          next unless Sentry.initialized? && defined?(::Sentry::Rails)
-
-          Sentry.configuration.rails.skippable_job_adapters << "ActiveJob::QueueAdapters::SidekiqAdapter"
-        end
-      end
+    def self.job_extension
+      @job_extension ||= JobExtension.new
     end
+
+    Sentry.register_job_extension(job_extension)
   end
 end
 
+require "sentry/sidekiq/railtie" if defined?(::Rails::Railtie)
+
 Sidekiq.configure_server do |config|
   config.error_handlers << Sentry::Sidekiq::ErrorHandler.new
   config.server_middleware do |chain|
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/job_extension.rb b/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/job_extension.rb
new file mode 100644
index 000000000..59d7ca6c1
--- /dev/null
+++ b/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/job_extension.rb
@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Sidekiq
+    module ActiveJobSupport
+      module JobExtension
+        ACTIVE_JOB_ADAPTER = "ActiveJob::QueueAdapters::SidekiqAdapter"
+        SIDEKIQ_JOB_KEY = "_sentry_sidekiq_job"
+
+        def handles?(job)
+          super || active_job?(job)
+        end
+
+        def scope_data(job:)
+          return super unless active_job?(job)
+
+          sidekiq_job = handed_over_sidekiq_job(job)
+          sidekiq_job ? super(job: sidekiq_job) : {}
+        end
+
+        def messaging_data(job:)
+          return super unless active_job?(job)
+
+          sidekiq_job = handed_over_sidekiq_job(job)
+          return {} unless sidekiq_job
+
+          super(job: sidekiq_job).merge(body_size: body_size(sidekiq_job)).compact
+        end
+
+        def retry_info(job:)
+          return super unless active_job?(job)
+
+          sidekiq_job = handed_over_sidekiq_job(job)
+          super(job: sidekiq_job) if sidekiq_job
+        end
+
+        def extract_trace_headers(job:, payload:)
+          return super if sidekiq_job?(payload)
+
+          sidekiq_job = sidekiq_job_in(payload)
+          super(job: job, payload: sidekiq_job) if sidekiq_job
+        end
+
+        def transaction_name(job:)
+          active_job?(job) ? "#{self.class::SIDEKIQ_NAME}/#{job.class.name}" : super
+        end
+
+        def span_origin(job:)
+          active_job?(job) ? self.class::SPAN_ORIGIN : super
+        end
+
+        def integration(job:)
+          active_job?(job) ? Sentry::Sidekiq : super
+        end
+
+        def report_retried_exception?(job:, exception:)
+          active_job?(job) ? false : super
+        end
+
+        def performed_by_active_job?(sidekiq_job)
+          !!sidekiq_job["wrapped"] && active_job_owns_adapter?
+        end
+
+        def active_job_args(sidekiq_job, scope:)
+          payload, *arguments = sidekiq_job["args"]
+          return sidekiq_job["args"] unless payload.is_a?(Hash)
+
+          sentry_data = (payload["_sentry"] || {}).merge("delivery_queue" => sidekiq_job["queue"])
+          sentry_data["execution_scope_id"] = scope.object_id
+          [payload.merge(
+            SIDEKIQ_JOB_KEY => sidekiq_job,
+            "_sentry" => sentry_data
+          ), *arguments]
+        end
+
+        private
+
+        def active_job?(job)
+          return false unless defined?(::ActiveJob::Base) && job.is_a?(::ActiveJob::Base)
+
+          job.class.queue_adapter.class.name == ACTIVE_JOB_ADAPTER
+        end
+
+        def active_job_owns_adapter?
+          return false unless defined?(::Sentry::Rails::ActiveJobExtensions)
+          return false unless ::Sentry::Rails.respond_to?(:active_job_owns_adapter?)
+
+          ::Sentry::Rails.active_job_owns_adapter?(ACTIVE_JOB_ADAPTER)
+        end
+
+        def handed_over_sidekiq_job(job)
+          return unless active_job?(job) && job.respond_to?(:_sentry_job_data)
+
+          sidekiq_job_in(job._sentry_job_data)
+        end
+
+        def sidekiq_job_in(active_job_payload)
+          return unless active_job_payload.is_a?(Hash)
+
+          sidekiq_job = active_job_payload[SIDEKIQ_JOB_KEY]
+          sidekiq_job if sidekiq_job?(sidekiq_job)
+        end
+      end
+    end
+  end
+end
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/server_middleware.rb b/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/server_middleware.rb
new file mode 100644
index 000000000..257482261
--- /dev/null
+++ b/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/server_middleware.rb
@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Sidekiq
+    module ActiveJobSupport
+      module ServerMiddleware
+        private
+
+        def execution_trace_headers(job, sidekiq_job)
+          if job_extension.performed_by_active_job?(job) && job["args"].first.is_a?(Hash)
+            headers = job["args"].first.dig("_sentry", "trace_propagation_headers")
+            return headers if headers && !headers.empty?
+          end
+
+          super
+        end
+
+        def perform_job(job, sidekiq_job, scope:)
+          return super unless job_extension.performed_by_active_job?(job)
+
+          args = job["args"]
+          begin
+            job["args"] = job_extension.active_job_args(sidekiq_job, scope: scope)
+            super
+          ensure
+            job["args"] = args
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb b/sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb
index ff9f014c3..40890db5c 100644
--- a/sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb
+++ b/sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb
@@ -35,17 +35,7 @@ def filtered
       end
 
       def transaction_name
-        class_name = (context["wrapped"] || context["class"] ||
-                      (context[:job] && (context[:job]["wrapped"] || context[:job]["class"]))
-                    )
-
-        if class_name
-          "#{SIDEKIQ_NAME}/#{class_name}"
-        elsif context[:event]
-          "#{SIDEKIQ_NAME}/#{context[:event]}"
-        else
-          SIDEKIQ_NAME
-        end
+        Sentry::Sidekiq.job_extension.error_context_transaction_name(context)
       end
 
       private
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb b/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb
index 024364e96..cb24d6789 100644
--- a/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb
+++ b/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb
@@ -1,11 +1,17 @@
 # frozen_string_literal: true
 
 require "sentry/sidekiq/context_filter"
+require "sentry/sidekiq/failure_reporting"
+require "sentry/sidekiq/sentry_context_middleware"
 
 module Sentry
   module Sidekiq
     class ErrorHandler
-      WITH_SIDEKIQ_7 = ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("7.0")
+      def initialize(failure_reporting: FailureReporting.new, consumer: SentryContextServerMiddleware.consumer, job_extension: Sentry::Sidekiq.job_extension)
+        @failure_reporting = failure_reporting
+        @consumer = consumer
+        @job_extension = job_extension
+      end
 
       # @param ex [Exception] the exception / error that occured
       # @param context [Hash or Array] Sidekiq error context
@@ -16,83 +22,19 @@ class ErrorHandler
       def call(ex, context, sidekiq_config = nil)
         return unless Sentry.initialized?
 
-        context_filter = Sentry::Sidekiq::ContextFilter.new(context)
-
         scope = Sentry.get_current_scope
-        scope.set_transaction_name(context_filter.transaction_name, source: :task) unless scope.transaction_name
-
-        retry_lim = retry_limit(context, sidekiq_config)
-
-        # If Sentry is configured to only report an error _after_ all retries have been exhausted,
-        # and if the job is retryable, and have not exceeded the retry_limit,
-        # return early.
-        if Sentry.configuration.sidekiq.report_after_job_retries && retryable?(context)
-          retry_count = context.dig(:job, "retry_count")
-          if retry_count.nil? || retry_count < retry_lim - 1
-            return
-          end
-        end
-
-        # See if we want to ignore jobs that have dead: false
-        return if Sentry.configuration.sidekiq.report_only_dead_jobs && context.dig(:job, "dead") == false
-
-        # Check if the retry count is below the attempt_threshold
-        attempt_threshold = context.dig(:job, "attempt_threshold")
-        if attempt_threshold && retryable?(context)
-          attempt_threshold = attempt_threshold.to_i
-          retry_count = context.dig(:job, "retry_count")
-          # attempt 1 - retry_count is nil
-          # attempt 2 - this is your first retry so retry_count is 0
-          # attempt 3 - you have retried once, retry_count is 1
-          attempt = retry_count.nil? ? 1 : retry_count.to_i + 2
-          # Cap at the final attempt so jobs with fewer retries than the threshold still report.
-          effective_threshold = [attempt_threshold, retry_lim + 1].min
+        scope.set_transaction_name(@job_extension.error_context_transaction_name(context), source: :task) unless scope.transaction_name
 
-          return if attempt < effective_threshold
-        end
+        job = context[:job]
+        return unless @failure_reporting.report?(job, sidekiq_config)
 
-        Sentry::Sidekiq.capture_exception(
-          ex,
-          contexts: { sidekiq: context_filter.filtered },
-          hint: { background: false }
+        @consumer.capture_exception(
+          job, ex,
+          contexts: { sidekiq: ContextFilter.new(context).filtered }
         )
       ensure
         scope&.clear
       end
-
-      private
-
-      def retryable?(context)
-        retry_option = context.dig(:job, "retry")
-        # when `retry` is not specified, it's default is `true` and it means 25 retries.
-        retry_option.nil? || retry_option == true || (retry_option.is_a?(Integer) && retry_option.positive?)
-      end
-
-      # @return [Integer] the number of retries allowed for the job
-      # Tries to fetch the retry limit from the job config first,
-      # then falls back to Sidekiq's configuration.
-      def retry_limit(context, sidekiq_config)
-        limit = context.dig(:job, "retry")
-
-        case limit
-        when Integer
-          limit
-        when TrueClass, NilClass
-          max_retries =
-            if WITH_SIDEKIQ_7
-              # Sidekiq 7.1.5+ passes the config to the error handler, so we should use that.
-              # Sidekiq 7.0 -> 7.1.5 provides ::Sidekiq.default_configuration.
-              sidekiq_config.is_a?(::Sidekiq::Config) ?
-                sidekiq_config[:max_retries] :
-                ::Sidekiq.default_configuration[:max_retries]
-            else
-              ::Sidekiq.options[:max_retries]
-            end
-          max_retries || 25
-        else
-          0
-        end
-      end
     end
   end
 end
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/failure_reporting.rb b/sentry-sidekiq/lib/sentry/sidekiq/failure_reporting.rb
new file mode 100644
index 000000000..3798ce16f
--- /dev/null
+++ b/sentry-sidekiq/lib/sentry/sidekiq/failure_reporting.rb
@@ -0,0 +1,49 @@
+# frozen_string_literal: true
+
+require "sentry/sidekiq/retry_policy"
+
+module Sentry
+  module Sidekiq
+    class FailureReporting
+      def initialize(retry_policy: RetryPolicy.build)
+        @retry_policy = retry_policy
+      end
+
+      def report?(job, sidekiq_config = nil)
+        config = Sentry.configuration.sidekiq
+        retry_limit = @retry_policy.limit(job, sidekiq_config)
+
+        return false if config.report_after_job_retries && retries_pending?(job, retry_limit)
+        return false if config.report_only_dead_jobs && job_value(job, "dead") == false
+        return false if below_attempt_threshold?(job, retry_limit)
+
+        true
+      end
+
+      private
+
+      def retries_pending?(job, retry_limit)
+        return false unless @retry_policy.retryable?(job)
+
+        retry_count = job_value(job, "retry_count")
+        retry_count.nil? || retry_count < retry_limit - 1
+      end
+
+      def below_attempt_threshold?(job, retry_limit)
+        attempt_threshold = job_value(job, "attempt_threshold")
+        return false unless attempt_threshold && @retry_policy.retryable?(job)
+
+        [attempt_threshold.to_i, retry_limit + 1].min > attempt(job)
+      end
+
+      def attempt(job)
+        retry_count = job_value(job, "retry_count")
+        retry_count.nil? ? 1 : retry_count.to_i + 2
+      end
+
+      def job_value(job, key)
+        job && job[key]
+      end
+    end
+  end
+end
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/job_extension.rb b/sentry-sidekiq/lib/sentry/sidekiq/job_extension.rb
new file mode 100644
index 000000000..3b626433a
--- /dev/null
+++ b/sentry-sidekiq/lib/sentry/sidekiq/job_extension.rb
@@ -0,0 +1,119 @@
+# frozen_string_literal: true
+
+require "sentry/sidekiq/retry_policy"
+require "sentry/sidekiq/context_filter"
+require "sentry/sidekiq/active_job_support/job_extension"
+
+module Sentry
+  module Sidekiq
+    class JobExtension < Sentry::JobExtensions::Extension
+      prepend ActiveJobSupport::JobExtension
+
+      SPAN_ORIGIN = "auto.queue.sidekiq"
+      SIDEKIQ_NAME = "Sidekiq"
+      ENQUEUED_AT_IN_MILLISECONDS = ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("8.0.0")
+
+      def initialize(retry_policy: RetryPolicy.build)
+        super(name: "sidekiq")
+        @retry_policy = retry_policy
+      end
+
+      def handles?(job)
+        sidekiq_job?(job)
+      end
+
+      def error_context_transaction_name(context)
+        job = context["wrapped"] || context["class"] ? context : context[:job]
+        return transaction_name(job: job) if job_class(job)
+
+        context[:event] ? "#{SIDEKIQ_NAME}/#{context[:event]}" : SIDEKIQ_NAME
+      end
+
+      def scope_data(job:)
+        return {} unless sidekiq_job?(job)
+
+        data = {
+          contexts: { sidekiq: ContextFilter.new(job).filtered },
+          tags: { queue: job["queue"], jid: job["jid"] }.merge(job_tags(job))
+        }
+        user = job["sentry_user"]
+        data[:user] = user if user.is_a?(Hash)
+        data
+      end
+
+      def messaging_data(job:)
+        return {} unless sidekiq_job?(job)
+
+        {
+          system: name,
+          message_id: job["jid"],
+          destination: job["queue"],
+          receive_latency: receive_latency(job)
+        }.compact
+      end
+
+      def consumer_messaging_data(job)
+        messaging_data(job: job).merge(body_size: body_size(job)).compact
+      end
+
+      def retry_info(job:)
+        return unless sidekiq_job?(job)
+
+        { count: @retry_policy.retry_count(job) || 0, will_retry: @retry_policy.will_retry?(job) }
+      end
+
+      def inject_trace_headers(job:, payload:, headers:)
+        return payload unless sidekiq_job?(payload)
+        return payload if payload["trace_propagation_headers"]
+
+        payload.merge("trace_propagation_headers" => headers)
+      end
+
+      def extract_trace_headers(job:, payload:)
+        return unless sidekiq_job?(payload)
+
+        payload["trace_propagation_headers"] || {}
+      end
+
+      def transaction_name(job:)
+        name = job_class(job)
+        "#{SIDEKIQ_NAME}/#{name}" if name
+      end
+
+      def producer(job:)
+        @producer ||= Sentry::JobExtensions::Producer::Delegated.new
+      end
+
+      private
+
+      def sidekiq_job?(job)
+        job.is_a?(Hash) && job.key?("jid")
+      end
+
+      def job_class(job)
+        job["wrapped"] || job["class"] if job.is_a?(Hash)
+      end
+
+      def job_tags(job)
+        Array(job["tags"]).each_with_object({}) { |name, tags_hash| tags_hash[:"sidekiq.#{name}"] = true }
+      end
+
+      def body_size(job)
+        ::Sidekiq.dump_json(job).bytesize
+      rescue StandardError
+        nil
+      end
+
+      def enqueued_at(job)
+        enqueued_at = job["enqueued_at"]
+        return unless enqueued_at
+
+        ENQUEUED_AT_IN_MILLISECONDS ? Time.at(Rational(enqueued_at, 1000)) : Time.at(enqueued_at)
+      end
+
+      def receive_latency(job)
+        Sentry::Utils::Latency.milliseconds_since(enqueued_at(job))&.round
+      end
+    end
+  end
+end
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/railtie.rb b/sentry-sidekiq/lib/sentry/sidekiq/railtie.rb
new file mode 100644
index 000000000..ecdd60140
--- /dev/null
+++ b/sentry-sidekiq/lib/sentry/sidekiq/railtie.rb
@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Sidekiq
+    class Railtie < ::Rails::Railtie
+      config.after_initialize do
+        next unless Sentry.initialized? && defined?(::Sentry::Rails)
+
+        Sentry.configuration.rails.skippable_job_adapters << ActiveJobSupport::JobExtension::ACTIVE_JOB_ADAPTER
+      end
+    end
+  end
+end
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/retry_policy.rb b/sentry-sidekiq/lib/sentry/sidekiq/retry_policy.rb
new file mode 100644
index 000000000..796132b26
--- /dev/null
+++ b/sentry-sidekiq/lib/sentry/sidekiq/retry_policy.rb
@@ -0,0 +1,100 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Sidekiq
+    class RetryPolicy
+      DEFAULT_MAX_RETRIES = 25
+
+      def self.build
+        if ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("7.0")
+          Sidekiq7.new
+        elsif ::Sidekiq.respond_to?(:[])
+          Sidekiq6.new
+        else
+          Legacy.new
+        end
+      end
+
+      attr_reader :default_config
+
+      def initialize(default_config)
+        @default_config = default_config
+      end
+
+      def retryable?(job)
+        retry_option = retry_option(job)
+        # when `retry` is not specified, it's default is `true` and it means 25 retries.
+        retry_option.nil? || retry_option == true || (retry_option.is_a?(Integer) && retry_option.positive?)
+      end
+
+      # @return [Integer] the number of retries allowed for the job
+      # Tries to fetch the retry limit from the job config first,
+      # then falls back to Sidekiq's configuration.
+      def limit(job, sidekiq_config = nil)
+        retry_option = retry_option(job)
+
+        case retry_option
+        when Integer
+          retry_option
+        when TrueClass, NilClass
+          max_retries(sidekiq_config) || DEFAULT_MAX_RETRIES
+        else
+          0
+        end
+      end
+
+      def retry_count(job)
+        job && job["retry_count"]
+      end
+
+      def will_retry?(job)
+        count = retry_count(job)
+        next_count = count.nil? ? 0 : count + 1
+        retryable?(job) && next_count < limit(job)
+      end
+
+      private
+
+      def retry_option(job)
+        job && job["retry"]
+      end
+
+      def max_retries(_sidekiq_config)
+        default_config[:max_retries]
+      end
+
+      class Sidekiq7 < RetryPolicy
+        def initialize(default_config = ::Sidekiq.default_configuration)
+          super
+        end
+
+        private
+
+        # Sidekiq 7.1.5+ passes the config to the error handler, so we should use that.
+        # Sidekiq 7.0 -> 7.1.5 provides ::Sidekiq.default_configuration.
+        def max_retries(sidekiq_config)
+          config = sidekiq_config.is_a?(::Sidekiq::Config) ? sidekiq_config : default_config
+          config[:max_retries]
+        end
+      end
+
+      class Sidekiq6 < RetryPolicy
+        def initialize(default_config = ::Sidekiq)
+          super
+        end
+      end
+
+      class Legacy < RetryPolicy
+        def initialize(default_config = ::Sidekiq)
+          super
+        end
+
+        private
+
+        def max_retries(_sidekiq_config)
+          default_config.options[:max_retries]
+        end
+      end
+    end
+  end
+end
diff --git a/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb b/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb
index 8a06b38d5..11f22ce2b 100644
--- a/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb
+++ b/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb
@@ -1,109 +1,74 @@
 # frozen_string_literal: true
 
-require "sentry/sidekiq/context_filter"
+require "sentry/sidekiq/job_extension"
+require "sentry/sidekiq/active_job_support/server_middleware"
 
 module Sentry
   module Sidekiq
-    module Helpers
-      def set_span_data(span, id:, queue:, latency: nil, retry_count: nil)
-        return unless span
-
-        span.set_data(Span::DataConventions::MESSAGING_MESSAGE_ID, id)
-        span.set_data(Span::DataConventions::MESSAGING_DESTINATION_NAME, queue)
-        span.set_data(Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) if latency
-        span.set_data(Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, retry_count) if retry_count
-      end
+    class SentryContextServerMiddleware
+      prepend ActiveJobSupport::ServerMiddleware
 
-      if ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("8.0.0")
-        def calculate_latency(job)
-          now_in_ms - job["enqueued_at"] if job["enqueued_at"]
-        end
-      else
-        def calculate_latency(job)
-          ((Time.now.to_f - job["enqueued_at"]) * 1000).round if job["enqueued_at"]
-        end
-      end
+      SPAN_ORIGIN = JobExtension::SPAN_ORIGIN
 
-      def now_in_ms
-        ::Process.clock_gettime(::Process::CLOCK_REALTIME, :millisecond)
+      def self.consumer
+        @consumer ||= Sentry::JobExtensions::Consumer.new(integration: Sentry::Sidekiq, origin: SPAN_ORIGIN)
       end
-    end
-
-    class SentryContextServerMiddleware
-      include Sentry::Sidekiq::Helpers
-
-      OP_NAME = "queue.process"
-      SPAN_ORIGIN = "auto.queue.sidekiq"
 
       def call(worker, job, queue)
         return yield unless Sentry.initialized?
 
-        context_filter = Sentry::Sidekiq::ContextFilter.new(job)
+        sidekiq_job = job.merge("queue" => queue)
 
         Sentry.clone_hub_to_current_thread
         scope = Sentry.get_current_scope
-        if (user = job["sentry_user"])
-          scope.set_user(user)
-        end
-        scope.set_tags(queue: queue, jid: job["jid"])
-        scope.set_tags(build_tags(job["tags"]))
-        scope.set_contexts(sidekiq: job.merge("queue" => queue))
-        scope.set_transaction_name(context_filter.transaction_name, source: :task)
-        transaction = start_transaction(scope, job["trace_propagation_headers"])
-
-        if transaction
-          scope.set_span(transaction)
-
-          latency = calculate_latency(job)
-
-          set_span_data(
-            transaction,
-            id: job["jid"],
-            queue: queue,
-            latency: latency,
-            retry_count: job["retry_count"] || 0
-          )
-        end
+        consumer.apply_scope_data(scope, sidekiq_job, default_scope_data: job_extension.scope_data(job: sidekiq_job))
+        scope.set_transaction_name(job_extension.transaction_name(job: job), source: :task)
+        trace_headers = execution_trace_headers(job, sidekiq_job)
+        transaction = consumer.start_transaction(
+          scope, sidekiq_job,
+          payload: sidekiq_job,
+          trace_headers: trace_headers,
+          messaging_data: -> { job_extension.consumer_messaging_data(sidekiq_job) },
+          default_retry_count: job_extension.retry_info(job: sidekiq_job)[:count],
+          always_continue_trace: true
+        )
 
         begin
-          yield
+          perform_job(job, sidekiq_job, scope: scope) { yield }
         rescue
-          finish_transaction(transaction, 500)
+          consumer.finish_transaction(transaction, 500)
           raise
         end
 
-        finish_transaction(transaction, 200)
+        consumer.finish_transaction(transaction, 200)
         # don't need to use ensure here
         # if the job failed, we need to keep the scope for error handler. and the scope will be cleared there
         scope.clear
       end
 
-      def build_tags(tags)
-        Array(tags).each_with_object({}) { |name, tags_hash| tags_hash[:"sidekiq.#{name}"] = true }
-      end
+      private
 
-      def start_transaction(scope, env)
-        options = {
-          name: scope.transaction_name,
-          source: scope.transaction_source,
-          op: OP_NAME,
-          origin: SPAN_ORIGIN
-        }
+      def execution_trace_headers(job, sidekiq_job)
+        job_extension.extract_trace_headers(job: sidekiq_job, payload: sidekiq_job)
+      end
 
-        transaction = Sentry.continue_trace(env, **options)
-        Sentry.start_transaction(transaction: transaction, **options)
+      def perform_job(job, sidekiq_job, scope:)
+        yield
       end
 
-      def finish_transaction(transaction, status)
-        return unless transaction
+      def consumer
+        self.class.consumer
+      end
 
-        transaction.set_http_status(status)
-        transaction.finish
+      def job_extension
+        Sentry::Sidekiq.job_extension
       end
     end
 
     class SentryContextClientMiddleware
-      include Sentry::Sidekiq::Helpers
+      def self.producer
+        @producer ||= Sentry::JobExtensions::Producer.new
+      end
 
       def call(worker_class, job, queue, _redis_pool)
         return yield unless Sentry.initialized?
@@ -114,14 +79,25 @@ def call(worker_class, job, queue, _redis_pool)
         end
 
         if Sentry.configuration.sidekiq.propagate_traces
-          job["trace_propagation_headers"] ||= Sentry.get_trace_propagation_headers
+          headers = Sentry.get_trace_propagation_headers
+          payload = Sentry::Sidekiq.job_extension.inject_trace_headers(job: job, payload: job, headers: headers)
+          job.replace(producer.inject_trace_headers(job, payload, headers))
         end
 
-        Sentry.with_child_span(op: "queue.publish", description: worker_class.to_s) do |span|
-          set_span_data(span, id: job["jid"], queue: queue)
+        producer.record_span(
+          job,
+          description: worker_class.to_s,
+          extension_messaging_keys: %i[system message_id],
+          default_messaging_data: { system: Sentry::Sidekiq.job_extension.name, message_id: job["jid"] },
+          messaging_data: { destination: queue },
+          retry_count: false
+        ) { yield }
+      end
+
+      private
 
-          yield
-        end
+      def producer
+        self.class.producer
       end
     end
   end
diff --git a/sentry-sidekiq/spec/isolated/active_job_spec.rb b/sentry-sidekiq/spec/isolated/active_job_spec.rb
new file mode 100644
index 000000000..8e34564c5
--- /dev/null
+++ b/sentry-sidekiq/spec/isolated/active_job_spec.rb
@@ -0,0 +1,698 @@
+# frozen_string_literal: true
+
+require_relative "support/rails_app"
+require "active_job/railtie"
+require "sidekiq/api"
+
+RSpec.describe "ActiveJob jobs on Sidekiq" do
+  let(:sidekiq_adapter) { "ActiveJob::QueueAdapters::SidekiqAdapter" }
+  let(:processor) { new_processor }
+  let(:transport) { Sentry.get_current_client.transport }
+  let(:traces_sample_rate) { 1.0 }
+
+  let(:parity_job) do
+    stub_const("ParityJob", Class.new(ActiveJob::Base) do
+      self.queue_adapter = :sidekiq
+      queue_as :events
+
+      def perform(*); end
+    end)
+  end
+
+  let(:reporting_job) do
+    stub_const("ReportingParityJob", Class.new(ActiveJob::Base) do
+      self.queue_adapter = :sidekiq
+
+      def perform
+        Sentry.capture_message("reported from the job")
+      end
+    end)
+  end
+
+  before do
+    sample_rate = traces_sample_rate
+    make_basic_app { |config| config.traces_sample_rate = sample_rate }
+  end
+
+  def enqueued_at
+    WITH_SIDEKIQ_8 ? now_in_ms : Time.now.to_f
+  end
+
+  def sidekiq_message(job_class, *arguments, **attributes)
+    job = job_class.new(*arguments)
+    {
+      "class" => ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper.to_s,
+      "wrapped" => job_class.name,
+      "queue" => job.queue_name,
+      "args" => [job.serialize],
+      "jid" => SecureRandom.hex(12),
+      "retry" => true,
+      "created_at" => enqueued_at,
+      "enqueued_at" => enqueued_at
+    }.merge(attributes.transform_keys(&:to_s))
+  end
+
+  def skip_sidekiq_adapter(skipped)
+    adapters = Sentry.configuration.rails.skippable_job_adapters
+    skipped ? adapters |= [sidekiq_adapter] : adapters -= [sidekiq_adapter]
+    Sentry.configuration.rails.skippable_job_adapters = adapters
+  end
+
+  def process(message, adapter_skipped:)
+    skip_sidekiq_adapter(adapter_skipped)
+    transport.events.clear
+    process_work(processor, Sidekiq::BasicFetch::UnitOfWork.new("queue:#{message["queue"]}", Sidekiq.dump_json(message)))
+    transport.events.dup
+  end
+
+  def consumer_transactions(events)
+    events.select { |event| event.is_a?(Sentry::TransactionEvent) && event.contexts.dig(:trace, :op) == "queue.process" }
+  end
+
+  def consumer_transaction(message, adapter_skipped:)
+    consumer_transactions(process(message, adapter_skipped: adapter_skipped)).first
+  end
+
+  context "with the transaction sentry-sidekiq sends when the adapter is skipped" do
+    let(:message) do
+      sidekiq_message(
+        parity_job,
+        queue: "retry_events",
+        tags: ["marvel", "dc"],
+        retry_count: 2,
+        sentry_user: { "id" => "42", "segment" => "beta" }
+      )
+    end
+
+    let(:sidekiq_transactions) { consumer_transactions(process(message, adapter_skipped: true)) }
+    let(:active_job_transactions) { consumer_transactions(process(message, adapter_skipped: false)) }
+    let(:sidekiq_transaction) { sidekiq_transactions.first }
+    let(:active_job_transaction) { active_job_transactions.first }
+
+    it "sends one transaction with the same name and origin when the adapter is not skipped" do
+      expect(sidekiq_transactions.size).to eq(1)
+      expect(active_job_transactions.size).to eq(1)
+      expect(sidekiq_transaction.transaction).to eq("Sidekiq/ParityJob")
+      expect(active_job_transaction.transaction).to eq(sidekiq_transaction.transaction)
+      expect(active_job_transaction.transaction_info).to eq(sidekiq_transaction.transaction_info)
+      expect(active_job_transaction.contexts.dig(:trace, :origin)).to eq(sidekiq_transaction.contexts.dig(:trace, :origin))
+    end
+
+    it "sends the same sidekiq context, tags and user, plus the active_job context, when the adapter is not skipped" do
+      expect(active_job_transaction.contexts[:sidekiq]).to eq(sidekiq_transaction.contexts[:sidekiq])
+      expect(active_job_transaction.tags).to include(sidekiq_transaction.tags)
+      expect(active_job_transaction.user).to eq(sidekiq_transaction.user)
+      expect(active_job_transaction.contexts[:active_job]).to include(job_class: "ParityJob", queue: "events")
+    end
+
+    it "sends the same messaging data and retry count when the adapter is not skipped" do
+      latency = "messaging.message.receive.latency"
+      sidekiq_data = sidekiq_transaction.contexts.dig(:trace, :data)
+      active_job_data = active_job_transaction.contexts.dig(:trace, :data)
+
+      expect(active_job_data).to include(sidekiq_data.except(latency))
+      expect(active_job_data[latency]).to be_within(1000).of(sidekiq_data[latency])
+    end
+  end
+
+  [nil, 0.0, 1.0].each do |sample_rate|
+    context "with incoming baggage at sample rate #{sample_rate.inspect}" do
+      let(:traces_sample_rate) { sample_rate }
+
+      it "matches native trace and baggage propagation when the adapter is not skipped" do
+        job_class = stub_const("PropagationParityJob", Class.new(ActiveJob::Base) do
+          self.queue_adapter = :sidekiq
+
+          def perform
+            Sentry.capture_message("propagated", extra: { headers: Sentry.get_trace_propagation_headers })
+          end
+        end)
+        trace_id = "a" * 32
+        headers = {
+          "sentry-trace" => "#{trace_id}-#{"b" * 16}-1",
+          "baggage" => "sentry-trace_id=#{trace_id},sentry-release=upstream,sentry-public_key=12345"
+        }
+
+        message = sidekiq_message(job_class, trace_propagation_headers: headers)
+        native_event = process(message, adapter_skipped: true).grep(Sentry::ErrorEvent).first
+        active_job_event = process(message, adapter_skipped: false).grep(Sentry::ErrorEvent).first
+
+        expect(active_job_event.contexts[:trace].slice(:trace_id, :parent_span_id))
+          .to eq(native_event.contexts[:trace].slice(:trace_id, :parent_span_id))
+        expect(active_job_event.extra[:headers]["sentry-trace"].split("-").first)
+          .to eq(native_event.extra[:headers]["sentry-trace"].split("-").first)
+        expect(active_job_event.extra[:headers]["baggage"]).to eq(native_event.extra[:headers]["baggage"])
+      end
+    end
+  end
+
+  context "without trace propagation headers when tracing is disabled" do
+    let(:traces_sample_rate) { nil }
+
+    def reported_trace_id(adapter_skipped:)
+      events = process(sidekiq_message(reporting_job), adapter_skipped: adapter_skipped)
+      events.find { |event| event.is_a?(Sentry::ErrorEvent) }.contexts.dig(:trace, :trace_id)
+    end
+
+    it "starts a new trace for the job instead of keeping the worker's, whether the adapter is skipped or not" do
+      worker_trace_id = Sentry.get_main_hub.current_scope.propagation_context.trace_id
+
+      expect(reported_trace_id(adapter_skipped: true)).not_to eq(worker_trace_id)
+      expect(reported_trace_id(adapter_skipped: false)).not_to eq(worker_trace_id)
+    end
+  end
+
+  context "when the job fails" do
+    let(:failing_job) do
+      stub_const("FailingParityJob", Class.new(ActiveJob::Base) do
+        self.queue_adapter = :sidekiq
+        queue_as :events
+
+        def perform(*)
+          raise "job failed"
+        end
+      end)
+    end
+
+    def error_events(events)
+      events.select { |event| event.is_a?(Sentry::ErrorEvent) }
+    end
+
+    def reported_events(adapter_skipped:, retry_counts:, **attributes)
+      retry_counts.map do |retry_count|
+        message = sidekiq_message(failing_job, retry_count: retry_count, **attributes)
+        error_events(process(message, adapter_skipped: adapter_skipped))
+      end
+    end
+
+    def expect_same_reported_attempts(retry_counts:, **attributes)
+      sidekiq_events = reported_events(adapter_skipped: true, retry_counts: retry_counts, **attributes)
+      active_job_events = reported_events(adapter_skipped: false, retry_counts: retry_counts, **attributes)
+
+      expect(active_job_events.map(&:size)).to eq(sidekiq_events.map(&:size))
+      expect(sidekiq_events.map(&:size).uniq).to contain_exactly(0, 1)
+      expect(active_job_events.flatten).to all(satisfy { |event| event.contexts.key?(:active_job) })
+    end
+
+    context "with the error event sentry-sidekiq sends when the adapter is skipped" do
+      let(:message) do
+        sidekiq_message(
+          failing_job,
+          { "sku" => "abc" },
+          queue: "retry_events",
+          tags: ["marvel"],
+          sentry_user: { "id" => "42" }
+        )
+      end
+
+      let(:sidekiq_events) { error_events(process(message, adapter_skipped: true)) }
+      let(:active_job_events) { error_events(process(message, adapter_skipped: false)) }
+      let(:sidekiq_event) { sidekiq_events.first }
+      let(:active_job_event) { active_job_events.first }
+
+      it "captures it once under the sidekiq integration with the same transaction name when the adapter is not skipped" do
+        expect(sidekiq_events.size).to eq(1)
+        expect(active_job_events.size).to eq(1)
+        expect(active_job_event.exception.values.first.mechanism.type).to eq("sidekiq")
+        expect(active_job_event.transaction).to eq(sidekiq_event.transaction)
+        expect(active_job_event.transaction_info).to eq(sidekiq_event.transaction_info)
+      end
+
+      it "sends the same tags, user and filtered sidekiq context, plus the ActiveJob data, when the adapter is not skipped" do
+        expect(active_job_event.tags).to include(sidekiq_event.tags)
+        expect(active_job_event.user).to eq(sidekiq_event.user)
+        expect(active_job_event.contexts[:sidekiq]).to eq(sidekiq_event.contexts[:sidekiq])
+        expect(active_job_event.contexts[:active_job]).to include(job_class: "FailingParityJob", queue: "events")
+        expect(active_job_event.extra).to include(active_job: "FailingParityJob")
+      end
+    end
+
+    [false, true].each do |report_after_retries|
+      [true, false].each do |adapter_skipped|
+        it "clears the failed delivery scope with report_after_job_retries=#{report_after_retries} and adapter_skipped=#{adapter_skipped}" do
+          Sentry.configuration.sidekiq.report_after_job_retries = report_after_retries
+          message = sidekiq_message(failing_job, retry: 2)
+          hub = Sentry.get_current_hub
+          Sentry.set_tags(worker: "outside")
+          Sentry.set_user(id: "worker-user")
+          Sentry.add_breadcrumb(Sentry::Breadcrumb.new(message: "worker context"))
+
+          events = process(message, adapter_skipped: adapter_skipped)
+
+          expect(error_events(events).size).to eq(report_after_retries ? 0 : 1)
+          expect(Sentry.get_current_scope.tags).to be_empty
+          expect(Sentry.get_current_scope.user).to be_empty
+          expect(Sentry.get_current_scope.breadcrumbs).to be_empty
+          expect(Sentry.get_current_scope.get_span).to be_nil
+          expect(hub.current_scope.tags).to eq(worker: "outside")
+        end
+      end
+    end
+
+    it "restores the payload while leaving the failed delivery scope for the error handler" do
+      skip_sidekiq_adapter(false)
+      message = sidekiq_message(failing_job, queue: "retry_events")
+      original_args = message["args"]
+      original_payload = Marshal.load(Marshal.dump(original_args))
+      hub = Sentry.get_current_hub
+      Sentry.set_tags(worker: "outside")
+
+      expect do
+        Sentry::Sidekiq::SentryContextServerMiddleware.new.call(nil, message, "retry_events") do
+          ActiveJob::Base.execute(message["args"].first)
+        end
+      end.to raise_error(RuntimeError, "job failed")
+
+      expect(message["args"]).to equal(original_args)
+      expect(message["args"]).to eq(original_payload)
+      expect(Sentry.get_current_scope.tags).to include(queue: "retry_events", jid: message["jid"])
+      expect(hub.current_scope.tags).to eq(worker: "outside")
+    end
+
+    context "with config.report_after_job_retries = true" do
+      before { Sentry.configuration.sidekiq.report_after_job_retries = true }
+
+      it "reports the same attempts whether the adapter is skipped or not when retry is a limit" do
+        expect_same_reported_attempts(retry: 3, retry_counts: [nil, 0, 1, 2])
+      end
+
+      context "when Sidekiq sets max_retries" do
+        let(:sidekiq_options) { WITH_SIDEKIQ_7 ? Sidekiq.default_configuration : Sidekiq.options }
+
+        around do |example|
+          previous_max_retries = sidekiq_options[:max_retries] || Sidekiq::JobRetry::DEFAULT_MAX_RETRY_ATTEMPTS
+          sidekiq_options[:max_retries] = 3
+          example.run
+        ensure
+          sidekiq_options[:max_retries] = previous_max_retries
+        end
+
+        it "reports the same attempts whether the adapter is skipped or not" do
+          expect_same_reported_attempts(retry: true, retry_counts: [nil, 0, 1, 2])
+        end
+      end
+    end
+
+    context "with config.report_only_dead_jobs = true" do
+      before { Sentry.configuration.sidekiq.report_only_dead_jobs = true }
+
+      it "reports the same attempts whether the adapter is skipped or not" do
+        sidekiq_events = [{}, { dead: false }].flat_map do |attributes|
+          reported_events(adapter_skipped: true, retry_counts: [nil], retry: 0, **attributes)
+        end
+        active_job_events = [{}, { dead: false }].flat_map do |attributes|
+          reported_events(adapter_skipped: false, retry_counts: [nil], retry: 0, **attributes)
+        end
+
+        expect(active_job_events.map(&:size)).to eq(sidekiq_events.map(&:size))
+        expect(sidekiq_events.map(&:size)).to eq([1, 0])
+        expect(active_job_events.flatten).to all(satisfy { |event| event.contexts.key?(:active_job) })
+      end
+    end
+
+    context "with attempt_threshold" do
+      it "reports the same attempts whether the adapter is skipped or not" do
+        expect_same_reported_attempts(attempt_threshold: 3, retry_counts: [nil, 0, 1, 2])
+      end
+
+      it "reports the same attempts whether the adapter is skipped or not when the retry limit is below the threshold" do
+        expect_same_reported_attempts(attempt_threshold: 3, retry: 1, retry_counts: [nil, 0])
+      end
+    end
+
+    context "with ActiveJob retry_on and discard_on" do
+      let(:retried_error) { stub_const("RetriedParityError", Class.new(StandardError)) }
+      let(:discarded_error) { stub_const("DiscardedParityError", Class.new(StandardError)) }
+
+      let(:retrying_job) do
+        error = retried_error
+        stub_const("RetryingParityJob", Class.new(ActiveJob::Base) do
+          self.queue_adapter = :sidekiq
+          retry_on error, attempts: 2, wait: 0
+
+          define_method(:perform) { raise error }
+        end)
+      end
+
+      let(:discarding_job) do
+        error = discarded_error
+        stub_const("DiscardingParityJob", Class.new(ActiveJob::Base) do
+          self.queue_adapter = :sidekiq
+          discard_on error
+
+          define_method(:perform) { raise error }
+        end)
+      end
+
+      def message_after_executions(job_class, error, executions)
+        message = sidekiq_message(job_class)
+        message["args"].first.merge!(
+          "executions" => executions,
+          "exception_executions" => { [error].to_s => executions }
+        )
+        message
+      end
+
+      def reported_executions(job_class, error, adapter_skipped:)
+        [0, 1].map do |executions|
+          message = message_after_executions(job_class, error, executions)
+          error_events(process(message, adapter_skipped: adapter_skipped))
+        end
+      end
+
+      it "reports a retry_on job at the same executions whether the adapter is skipped or not" do
+        sidekiq_events = reported_executions(retrying_job, retried_error, adapter_skipped: true)
+        active_job_events = reported_executions(retrying_job, retried_error, adapter_skipped: false)
+
+        expect(active_job_events.map(&:size)).to eq(sidekiq_events.map(&:size))
+        expect(sidekiq_events.map(&:size)).to eq([0, 1])
+        expect(active_job_events.flatten).to all(satisfy { |event| event.contexts.key?(:active_job) })
+      end
+
+      context "with config.rails.active_job_report_on_retry_error = true" do
+        before do
+          retrying_job
+          sample_rate = traces_sample_rate
+          make_basic_app do |config|
+            config.traces_sample_rate = sample_rate
+            config.rails.active_job_report_on_retry_error = true
+          end
+        end
+
+        it "reports a retry_on job at the same executions whether the adapter is skipped or not" do
+          sidekiq_events = reported_executions(retrying_job, retried_error, adapter_skipped: true)
+
+          expect(reported_executions(retrying_job, retried_error, adapter_skipped: false).map(&:size)).to eq(sidekiq_events.map(&:size))
+          expect(sidekiq_events.map(&:size)).to eq([0, 1])
+        end
+      end
+
+      it "reports a discard_on job at the same executions whether the adapter is skipped or not" do
+        sidekiq_events = reported_executions(discarding_job, discarded_error, adapter_skipped: true)
+
+        expect(reported_executions(discarding_job, discarded_error, adapter_skipped: false).map(&:size)).to eq(sidekiq_events.map(&:size))
+        expect(sidekiq_events.map(&:size)).to eq([0, 0])
+      end
+    end
+  end
+
+  context "when job deserialization raises outside StandardError" do
+    [LoadError, SystemStackError].each do |exception_class|
+      it "preserves delivery context and the incoming trace for #{exception_class}" do
+        job_class = Class.new(ActiveJob::Base) do
+          self.queue_adapter = :sidekiq
+
+          define_method(:deserialize) do |payload|
+            super(payload)
+            raise exception_class, "job deserialization failed"
+          end
+
+          def perform; end
+        end
+        stub_const("DeserializationFailureParityJob", job_class)
+        message = sidekiq_message(
+          job_class, retry: false, queue: "delivery", tags: ["marvel"],
+          sentry_user: { "id" => "42" },
+          trace_propagation_headers: { "sentry-trace" => "#{"a" * 32}-#{"b" * 16}-1" }
+        )
+        Sentry.set_tags(worker: "outside")
+        hub = Sentry.get_current_hub
+
+        expect { process(message, adapter_skipped: false) }.to raise_error(exception_class, "job deserialization failed")
+
+        errors = transport.events.grep(Sentry::ErrorEvent)
+        expect(errors.size).to eq(1)
+        expect(errors.first.tags).to include(queue: "delivery", jid: message["jid"], "sidekiq.marvel": true)
+        expect(errors.first.user).to include("id" => "42")
+        expect(errors.first.contexts.dig(:trace, :trace_id)).to eq("a" * 32)
+        expect(Sentry.get_current_scope.tags).to be_empty
+        expect(hub.current_scope.tags).to eq(worker: "outside")
+      end
+    end
+  end
+
+  context "when the queued ActiveJob class cannot be loaded" do
+    let(:traces_sample_rate) { nil }
+
+    [nil, 1.0].each do |sample_rate|
+      it "preserves native error enrichment and the incoming trace at sample rate #{sample_rate.inspect}" do
+        Sentry.configuration.traces_sample_rate = sample_rate
+        worker_transaction = Sentry.start_transaction(name: "worker", op: "test")
+        Sentry.get_main_hub.current_scope.set_span(worker_transaction) if worker_transaction
+
+        message = sidekiq_message(
+          parity_job,
+          retry: false,
+          tags: ["marvel"],
+          sentry_user: { "id" => "42" },
+          trace_propagation_headers: { "sentry-trace" => "#{"a" * 32}-#{"b" * 16}-0" }
+        )
+        message["wrapped"] = "RemovedJob"
+        message["args"].first["job_class"] = "RemovedJob"
+
+        native_errors = process(message, adapter_skipped: true).grep(Sentry::ErrorEvent)
+        active_job_errors = process(message, adapter_skipped: false).grep(Sentry::ErrorEvent)
+
+        expect(native_errors.size).to eq(1)
+        expect(active_job_errors.size).to eq(1)
+        expect(active_job_errors.first.tags).to include(native_errors.first.tags)
+        expect(active_job_errors.first.user).to eq(native_errors.first.user)
+        expect(active_job_errors.first.contexts.dig(:trace, :trace_id)).to eq("a" * 32)
+      end
+    end
+
+    it "records one failed consumer transaction when the queued job class is missing" do
+      Sentry.configuration.traces_sample_rate = 1.0
+      message = sidekiq_message(
+        parity_job, retry: false, queue: "delivery",
+        trace_propagation_headers: { "sentry-trace" => "#{"a" * 32}-#{"b" * 16}-1" }
+      )
+      message["wrapped"] = "RemovedJob"
+      message["args"].first["job_class"] = "RemovedJob"
+
+      [true, false].each do |adapter_skipped|
+        events = process(message, adapter_skipped: adapter_skipped)
+        transactions = consumer_transactions(events)
+        errors = events.grep(Sentry::ErrorEvent)
+
+        expect(transactions.size).to eq(1)
+        expect(errors.size).to eq(1)
+        expect(transactions.first.transaction).to eq("Sidekiq/RemovedJob")
+        expect(transactions.first.contexts.dig(:trace, :status)).to eq("internal_error")
+        expect(transactions.first.contexts.dig(:trace, :trace_id)).to eq("a" * 32)
+        expect(transactions.first.contexts.dig(:trace, :data)).to include(
+          "messaging.message.id" => message["jid"], "messaging.destination.name" => "delivery"
+        )
+        expect(errors.first.contexts.dig(:trace, :span_id)).to eq(transactions.first.contexts.dig(:trace, :span_id))
+      end
+    end
+
+    it "preserves the payload and delivery scope when loading the job fails" do
+      skip_sidekiq_adapter(false)
+      message = sidekiq_message(parity_job, retry: false)
+      message["args"].first["job_class"] = "RemovedJob"
+      original_args = message["args"]
+      original_payload = Marshal.load(Marshal.dump(original_args))
+      hub = Sentry.get_current_hub
+      Sentry.set_tags(worker: "outside")
+
+      expect do
+        Sentry::Sidekiq::SentryContextServerMiddleware.new.call(nil, message, "events") do
+          ActiveJob::Base.execute(message["args"].first)
+        end
+      end.to raise_error(NameError)
+
+      expect(message["args"]).to equal(original_args)
+      expect(message["args"]).to eq(original_payload)
+      expect(Sentry.get_current_scope.tags).to include(queue: "events", jid: message["jid"])
+      expect(hub.current_scope.tags).to eq(worker: "outside")
+    end
+  end
+
+  context "with a separate running Sidekiq configuration", skip: !WITH_SIDEKIQ_7 do
+    let(:running_config) do
+      Sidekiq::Config.new(max_retries: 5).tap do |config|
+        config.logger = Logger.new(nil)
+        config.error_handlers.clear
+        config.error_handlers << Sentry::Sidekiq::ErrorHandler.new
+        config.server_middleware { |chain| chain.add Sentry::Sidekiq::SentryContextServerMiddleware }
+      end
+    end
+    let(:processor) { Sidekiq::Processor.new(running_config.default_capsule) }
+    let(:failing_job) do
+      stub_const("ConfigParityJob", Class.new(ActiveJob::Base) do
+        self.queue_adapter = :sidekiq
+
+        def perform
+          raise "job failed"
+        end
+      end)
+    end
+
+    around do |example|
+      previous_max_retries = Sidekiq.default_configuration[:max_retries]
+      Sidekiq.default_configuration[:max_retries] = 3
+      example.run
+    ensure
+      Sidekiq.default_configuration[:max_retries] = previous_max_retries
+    end
+
+    [true, false].each do |adapter_skipped|
+      context "when the Sidekiq adapter is #{adapter_skipped ? 'skipped' : 'handled by ActiveJob'}" do
+        it "uses the running retry limit when reporting only exhausted retries" do
+          Sentry.configuration.sidekiq.report_after_job_retries = true
+
+          reported_errors = [2, 3, 4].map do |retry_count|
+            message = sidekiq_message(failing_job, retry: true, retry_count: retry_count)
+            process(message, adapter_skipped: adapter_skipped).grep(Sentry::ErrorEvent).size
+          end
+
+          expect(reported_errors).to eq([0, 0, 1])
+        end
+
+        it "uses the running retry limit when applying an attempt threshold" do
+          reported_errors = [2, 3, 4].map do |retry_count|
+            message = sidekiq_message(failing_job, retry: true, retry_count: retry_count, attempt_threshold: 10)
+            process(message, adapter_skipped: adapter_skipped).grep(Sentry::ErrorEvent).size
+          end
+
+          expect(reported_errors).to eq([0, 0, 1])
+        end
+      end
+    end
+  end
+
+  context "when the job is enqueued" do
+    let(:enqueued_queue) { Sidekiq::Queue.new("events") }
+    let(:user) { { "id" => "42", "segment" => "beta", "ip_address" => "1.2.3.4" } }
+    let(:message_specific_data) { %w[messaging.message.id messaging.message.body.size] }
+
+    let(:enqueue_result) { Struct.new(:parent, :publish_spans, :pushed_job) }
+
+    before do
+      Sentry.configuration.data_collection.user_info = true
+      enqueued_queue.clear
+    end
+
+    after { enqueued_queue.clear }
+
+    def enqueue(adapter_skipped:)
+      skip_sidekiq_adapter(adapter_skipped)
+      enqueued_queue.clear
+      transport.events.clear
+      Sentry.set_user(user)
+      parent = Sentry.start_transaction(name: "enqueuer", op: "test")
+      Sentry.get_current_scope.set_span(parent)
+
+      parity_job.perform_later("sku")
+
+      parent.finish
+      Sentry.get_current_scope.clear
+      parent_event = transport.events.find { |event| event.is_a?(Sentry::TransactionEvent) }
+      publish_spans = parent_event.spans.select { |span| span[:op] == "queue.publish" }
+      enqueue_result.new(parent, publish_spans, enqueued_queue.first.item)
+    end
+
+    def enqueue_and_process(adapter_skipped:)
+      enqueued = enqueue(adapter_skipped: adapter_skipped)
+      [enqueued, consumer_transaction(enqueued.pushed_job, adapter_skipped: adapter_skipped)]
+    end
+
+    it "records the queue.publish span sentry-sidekiq records today when the adapter is not skipped" do
+      sidekiq_enqueue = enqueue(adapter_skipped: true)
+      active_job_enqueue = enqueue(adapter_skipped: false)
+      expect(sidekiq_enqueue.publish_spans.size).to eq(1)
+      expect(active_job_enqueue.publish_spans.size).to eq(1)
+      sidekiq_span = sidekiq_enqueue.publish_spans.first
+      active_job_span = active_job_enqueue.publish_spans.first
+
+      expect(active_job_span.slice(:description, :origin)).to eq(sidekiq_span.slice(:description, :origin))
+      expect(active_job_span[:data]).to include(sidekiq_span[:data].except(*message_specific_data))
+      expect(active_job_span[:data]["messaging.message.id"]).to eq(active_job_enqueue.pushed_job["jid"])
+      expect(active_job_span[:data]["messaging.message.body.size"]).to be_nil.or eq(Sidekiq.dump_json(active_job_enqueue.pushed_job).bytesize)
+    end
+
+    it "carries the same trace headers and the whole scope user to the job when the adapter is not skipped" do
+      sidekiq_enqueue, sidekiq_transaction = enqueue_and_process(adapter_skipped: true)
+      active_job_enqueue, active_job_transaction = enqueue_and_process(adapter_skipped: false)
+
+      expect(active_job_enqueue.pushed_job.slice("trace_propagation_headers", "sentry_user").keys)
+        .to eq(sidekiq_enqueue.pushed_job.slice("trace_propagation_headers", "sentry_user").keys)
+      expect(active_job_enqueue.pushed_job["sentry_user"]).to eq(user)
+      expect(active_job_transaction.user).to eq(sidekiq_transaction.user)
+      [[sidekiq_enqueue, sidekiq_transaction], [active_job_enqueue, active_job_transaction]].each do |enqueued, transaction|
+        expect(transaction.contexts[:trace]).to include(trace_id: enqueued.parent.trace_id, parent_span_id: enqueued.parent.span_id)
+      end
+    end
+
+    context "with sidekiq propagate_traces = false" do
+      before { Sentry.configuration.sidekiq.propagate_traces = false }
+
+      context "when the job generates outgoing trace headers" do
+        let(:parity_job) do
+          stub_const("PropagatingParityJob", Class.new(ActiveJob::Base) do
+            self.queue_adapter = :sidekiq
+            queue_as :events
+
+            def perform(*)
+              Sentry.capture_message("outgoing headers", extra: { headers: Sentry.get_trace_propagation_headers })
+            end
+          end)
+        end
+
+        it "uses the new consumer transaction's trace without changing the propagation setting" do
+          enqueued = enqueue(adapter_skipped: false)
+
+          expect(enqueued.pushed_job).not_to have_key("trace_propagation_headers")
+          expect(enqueued.pushed_job["args"].first.dig("_sentry", "trace_propagation_headers")).to be_nil
+
+          events = process(enqueued.pushed_job, adapter_skipped: false)
+          transaction = consumer_transactions(events).first
+          event = events.grep(Sentry::ErrorEvent).find { |entry| entry.message == "outgoing headers" }
+          headers = event.extra[:headers]
+          consumer_trace_id = transaction.contexts.dig(:trace, :trace_id)
+
+          expect(consumer_trace_id).not_to eq(enqueued.parent.trace_id)
+          expect(headers["sentry-trace"].split("-").first).to eq(consumer_trace_id)
+          expect(headers["baggage"]).to include("sentry-trace_id=#{consumer_trace_id}")
+        end
+      end
+
+      it "starts a new trace for the job whether the adapter is skipped or not" do
+        [true, false].each do |adapter_skipped|
+          enqueued, transaction = enqueue_and_process(adapter_skipped: adapter_skipped)
+
+          expect(enqueued.pushed_job).not_to have_key("trace_propagation_headers")
+          expect(transaction.contexts.dig(:trace, :trace_id)).not_to eq(enqueued.parent.trace_id)
+        end
+      end
+    end
+  end
+
+  context "with the adapter not skipped" do
+    it "keeps reporting plain Sidekiq workers through sentry-sidekiq" do
+      skip_sidekiq_adapter(false)
+
+      execute_worker(processor, HappyWorker)
+
+      transaction = consumer_transactions(transport.events).first
+      expect(transaction.transaction).to eq("Sidekiq/HappyWorker")
+      expect(transaction.contexts.dig(:trace, :origin)).to eq("auto.queue.sidekiq")
+    end
+
+    it "keeps ActiveJob's own naming for jobs run on other adapters" do
+      skip_sidekiq_adapter(false)
+      inline_job = stub_const("InlineParityJob", Class.new(ActiveJob::Base) do
+        self.queue_adapter = :inline
+
+        def perform; end
+      end)
+
+      inline_job.perform_later
+
+      transaction = consumer_transactions(transport.events).first
+      expect(transaction.transaction).to eq("InlineParityJob")
+      expect(transaction.contexts.dig(:trace, :origin)).to eq("auto.queue.active_job")
+    end
+  end
+end
diff --git a/sentry-sidekiq/spec/isolated/native_extension_spec.rb b/sentry-sidekiq/spec/isolated/native_extension_spec.rb
new file mode 100644
index 000000000..3966819e2
--- /dev/null
+++ b/sentry-sidekiq/spec/isolated/native_extension_spec.rb
@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+require "sentry-ruby"
+
+class NativeEnrichmentExtension < Sentry::JobExtensions::Extension
+  def handles?(job)
+    job.is_a?(Hash) && job["class"] == "SadWorker"
+  end
+
+  def scope_data(job:)
+    { tags: { enrichment: "present" } }
+  end
+end
+
+Sentry.register_job_extension(NativeEnrichmentExtension.new(name: "native_enrichment"))
+
+require "spec_helper"
+require "sidekiq/api"
+
+RSpec.describe "Native Sidekiq with an earlier enrichment extension" do
+  [nil, 0.0, 1.0].each do |sample_rate|
+    it "preserves propagation and native metadata at sample rate #{sample_rate.inspect}" do
+      perform_basic_setup { |config| config.traces_sample_rate = sample_rate }
+      parent = Sentry.start_transaction(name: "enqueue", op: "test")
+      Sentry.get_current_scope.set_span(parent) if parent
+      headers = Sentry.get_trace_propagation_headers
+      client = Sidekiq::Client.new
+      client.middleware { |chain| chain.add Sentry::Sidekiq::SentryContextClientMiddleware }
+      jid = client.push("class" => SadWorker, "queue" => "extension", "args" => [], "retry" => false)
+      message = Sidekiq.redis { |redis| redis.lpop("queue:extension") }
+      job = Sidekiq.load_json(message)
+      transport = Sentry.get_current_client.transport
+
+      expect(job["trace_propagation_headers"]).to eq(headers)
+      expect(headers.keys).to include("sentry-trace", "baggage")
+      parent.finish if parent
+      process_work(new_processor, Sidekiq::BasicFetch::UnitOfWork.new("queue:extension", message))
+
+      event = transport.events.find { |entry| entry.is_a?(Sentry::ErrorEvent) }
+      expect(event.contexts.dig(:trace, :trace_id)).to eq(headers["sentry-trace"].split("-").first)
+      expect(event.tags).to include(queue: "extension", jid: jid, enrichment: "present")
+      expect(event.contexts[:sidekiq]).to include("jid" => jid, "queue" => "extension")
+
+      if sample_rate == 1.0
+        consumer = transport.events.find { |entry| entry.is_a?(Sentry::TransactionEvent) && entry.transaction == "Sidekiq/SadWorker" }
+        expect(consumer.contexts.dig(:trace, :data)).to include("messaging.message.id" => jid, "messaging.destination.name" => "extension")
+        producer = transport.events.find { |entry| entry.is_a?(Sentry::TransactionEvent) && entry.transaction == "enqueue" }
+        expect(producer.spans.first[:data]).to include("messaging.message.id" => jid, "messaging.destination.name" => "extension")
+      end
+    end
+  end
+end
diff --git a/sentry-sidekiq/spec/isolated/rails_spec.rb b/sentry-sidekiq/spec/isolated/rails_spec.rb
index 7fad22de8..023f8b5c0 100644
--- a/sentry-sidekiq/spec/isolated/rails_spec.rb
+++ b/sentry-sidekiq/spec/isolated/rails_spec.rb
@@ -1,47 +1,6 @@
 # frozen_string_literal: true
 
-begin
-  require "simplecov"
-  SimpleCov.command_name "SidekiqRails"
-rescue LoadError
-end
-
-require "sentry-rails"
-
-# This MUST be required after sentry-rails because it requires sentry-sidekiq
-# which checks if Railtie is defined to properly set things up
-require_relative "../spec_helper"
-
-# This is needed to prevent Sidekiq 6.5 crash
-if SIDEKIQ_VERSION >= Gem::Version.new("6.5") && SIDEKIQ_VERSION < Gem::Version.new("7.0")
-  # NoMethodError:
-  #  undefined method 'broadcast' for class ActiveSupport::Logger
-  #  /workspace/sentry/vendor/gems/3.4.5/gems/sidekiq-6.5.7/lib/sidekiq/rails.rb:46:in 'block (2 levels) in <class:Rails>'
-  Rails.logger = Logger.new($stdout)
-end
-
-class TestApp < Rails::Application
-end
-
-def make_basic_app
-  app = Class.new(TestApp) do
-    def self.name
-      "RailsTestApp"
-    end
-  end
-
-  app.config.hosts = nil
-  app.config.secret_key_base = "test"
-  app.config.eager_load = false
-
-  app.initializer :configure_sentry do
-    perform_basic_setup
-  end
-
-  app.initialize!
-  Rails.application = app
-  app
-end
+require_relative "support/rails_app"
 
 RSpec.describe Sentry::Sidekiq do
   before do
diff --git a/sentry-sidekiq/spec/isolated/support/rails_app.rb b/sentry-sidekiq/spec/isolated/support/rails_app.rb
new file mode 100644
index 000000000..ad4ddf82d
--- /dev/null
+++ b/sentry-sidekiq/spec/isolated/support/rails_app.rb
@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+begin
+  require "simplecov"
+  SimpleCov.command_name "SidekiqRails"
+rescue LoadError
+end
+
+require "sentry-rails"
+
+# This MUST be required after sentry-rails because it requires sentry-sidekiq
+# which checks if Railtie is defined to properly set things up
+require_relative "../../spec_helper"
+
+# This is needed to prevent Sidekiq 6.5 crash
+if SIDEKIQ_VERSION >= Gem::Version.new("6.5") && SIDEKIQ_VERSION < Gem::Version.new("7.0")
+  # NoMethodError:
+  #  undefined method 'broadcast' for class ActiveSupport::Logger
+  #  /workspace/sentry/vendor/gems/3.4.5/gems/sidekiq-6.5.7/lib/sidekiq/rails.rb:46:in 'block (2 levels) in <class:Rails>'
+  Rails.logger = Logger.new($stdout)
+end
+
+class TestApp < Rails::Application
+end
+
+BOOTED_APPS = []
+
+def make_basic_app(&sentry_config)
+  return reinitialize_basic_app(BOOTED_APPS.first, &sentry_config) if BOOTED_APPS.any?
+
+  boot_basic_app(&sentry_config).tap { |app| BOOTED_APPS << app }
+end
+
+def reinitialize_basic_app(app, &sentry_config)
+  perform_basic_setup(&sentry_config)
+  ActiveSupport.run_load_hooks(:after_initialize, app)
+  app
+end
+
+def boot_basic_app(&sentry_config)
+  app = Class.new(TestApp) do
+    def self.name
+      "RailsTestApp"
+    end
+  end
+
+  app.config.hosts = nil
+  app.config.secret_key_base = "test"
+  app.config.eager_load = false
+
+  app.initializer :configure_sentry do
+    perform_basic_setup(&sentry_config)
+  end
+
+  app.initialize!
+  Rails.application = app
+  app
+end
diff --git a/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb b/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb
index c1b1a4b62..88c277d0c 100644
--- a/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb
+++ b/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb
@@ -30,6 +30,26 @@
     expect(event.user).to eq(user)
   end
 
+  it "starts a new trace for a job without trace propagation headers" do
+    perform_basic_setup
+    worker_trace_id = Sentry.get_main_hub.current_scope.propagation_context.trace_id
+
+    execute_worker(processor, SadWorker)
+
+    event = transport.events.last
+    expect(event.contexts.dig(:trace, :trace_id)).to be_a(String)
+    expect(event.contexts.dig(:trace, :trace_id)).not_to eq(worker_trace_id)
+  end
+
+  it "continues the trace from trace propagation headers when tracing is disabled" do
+    perform_basic_setup
+    trace_id = "a" * 32
+
+    execute_worker(processor, SadWorker, trace_propagation_headers: { "sentry-trace" => "#{trace_id}-#{"b" * 16}-1" })
+
+    expect(transport.events.last.contexts.dig(:trace, :trace_id)).to eq(trace_id)
+  end
+
   context "with tracing enabled" do
     before do
       perform_basic_setup { |config| config.traces_sample_rate = 1.0 }
@@ -58,7 +78,8 @@
     it "sets sidekiq tags to the event" do
       execute_worker(processor, TagsWorker)
       event = transport.events.last
-      expect(event.tags.keys).to include(:"sidekiq.marvel", :"sidekiq.dc")
+      expect(event.tags).to include(queue: "default", jid: "123123", "sidekiq.marvel": true, "sidekiq.dc": true)
+      expect(event.contexts[:sidekiq]).to include("class" => "TagsWorker", "jid" => "123123", "queue" => "default", "tags" => ["marvel", "dc"])
     end
 
     it "has the correct origin" do
@@ -93,6 +114,38 @@ def timecop_delay
         expect(trace[:data]['messaging.message.receive.latency']).to be_within(20).of(expected_latency)
       end
 
+      it "reports sidekiq as the messaging system on the queue.process transaction" do
+        execute_worker(processor, HappyWorker)
+
+        expect(transport.events[0].contexts.dig(:trace, :data)['messaging.system']).to eq('sidekiq')
+      end
+
+      it "reports the job's retry count on the queue.process transaction" do
+        execute_worker(processor, HappyWorker, retry_count: 3)
+
+        expect(transport.events[0].contexts.dig(:trace, :data)['messaging.message.retry.count']).to eq(3)
+      end
+
+      it "reports the byte length of the job encoded as JSON as the message body size on the queue.process transaction" do
+        message = Sidekiq.dump_json("jid" => "123456", "class" => "HappyWorker", "args" => ["héllo"], "queue" => "default")
+
+        process_work(processor, Sidekiq::BasicFetch::UnitOfWork.new("queue:default", message))
+
+        expect(transport.events[0].contexts.dig(:trace, :data)['messaging.message.body.size']).to eq(message.bytesize)
+      end
+
+      it "performs the job and omits the message body size when the job cannot be encoded as JSON" do
+        job = { "jid" => "123456", "class" => "HappyWorker", "args" => ["\xFF".dup.force_encoding(Encoding::UTF_8)], "queue" => "default" }
+        performed = false
+
+        described_class.new.call(HappyWorker.new, job, "default") { performed = true }
+
+        trace_data = transport.events[0].contexts.dig(:trace, :data)
+        expect(performed).to be(true)
+        expect(trace_data).to include('messaging.message.id' => '123456', 'messaging.system' => 'sidekiq')
+        expect(trace_data).not_to have_key('messaging.message.body.size')
+      end
+
       if MIN_SIDEKIQ_6
         it "does not fail for latency when performed inline" do
           HappyWorker.perform_inline
@@ -181,6 +234,18 @@ def ensure_queue_empty(queue, timeout: 0.1)
       expect(queue.first["sentry_user"]).to eq(user)
     end
 
+    context "when the scope user has fields beyond id, email and username" do
+      let(:user) { { "id" => 1, "ip_address" => "1.2.3.4", "segment" => "vip" } }
+
+      it "propagates the whole user" do
+        Sentry.configuration.data_collection.user_info = true
+
+        client.push('queue' => 'default', 'class' => HappyWorker, 'args' => [])
+
+        expect(queue.first["sentry_user"]).to eq(user)
+      end
+    end
+
     it "does not propagate user information when user_info is disabled" do
       Sentry.configuration.data_collection.user_info = false
 
@@ -239,6 +304,77 @@ def ensure_queue_empty(queue, timeout: 0.1)
       expect(span[:data]['messaging.destination.name']).to eq('default')
     end
 
+    it "only reports a producer body size that matches the finalized queued message" do
+      client.push('queue' => 'default', 'class' => HappyWorker, 'args' => ["héllo"])
+
+      transaction.finish
+
+      pushed_job = queue.first.item
+      span = transport.events.last.spans.detect { |span| span[:op] == "queue.publish" }
+      expect(span[:data]['messaging.message.body.size']).to be_nil.or eq(Sidekiq.dump_json(pushed_job).bytesize)
+    end
+
+    context "when an earlier client middleware reroutes the job" do
+      let(:rerouted_queue) { Sidekiq::Queue.new("rerouted") }
+
+      let(:rerouting_middleware) do
+        Class.new do
+          def call(_worker_class, job, _queue, _redis_pool)
+            job["queue"] = "rerouted"
+            yield
+          end
+        end
+      end
+
+      let(:client) do
+        Sidekiq::Client.new.tap do |client|
+          client.middleware do |chain|
+            chain.add rerouting_middleware
+            chain.add described_class
+          end
+        end
+      end
+
+      after do
+        client.middleware { |chain| chain.remove(rerouting_middleware) }
+        ensure_queue_empty(rerouted_queue)
+      end
+
+      it "reports the queue the push was made to as the destination" do
+        client.push('queue' => 'default', 'class' => HappyWorker, 'args' => [])
+
+        transaction.finish
+
+        span = transport.events.last.spans.detect { |span| span[:op] == "queue.publish" }
+        expect(span[:data]['messaging.destination.name']).to eq('default')
+      end
+    end
+
+    it "names the queue.publish span after the worker with the sidekiq messaging system and without an origin or retry count" do
+      client.push('queue' => 'default', 'class' => HappyWorker, 'args' => [], 'retry_count' => 2)
+
+      transaction.finish
+
+      span = transport.events.last.spans.detect { |span| span[:op] == "queue.publish" }
+      expect(span[:description]).to eq("HappyWorker")
+      expect(span[:origin]).to eq("manual")
+      expect(span[:data]['messaging.system']).to eq('sidekiq')
+      expect(span[:data]).not_to have_key('messaging.message.retry.count')
+    end
+
+    it "does not enqueue the job when recording the queue.publish span raises" do
+      allow(Sentry).to receive(:with_child_span).and_call_original
+      allow(Sentry).to receive(:with_child_span)
+        .with(hash_including(op: "queue.publish"))
+        .and_raise(StandardError, "boom from instrumentation")
+
+      expect do
+        client.push('queue' => 'default', 'class' => HappyWorker, 'args' => [])
+      end.to raise_error(StandardError, /boom from instrumentation/)
+
+      expect(queue.size).to be(0)
+    end
+
     it "does not propagate headers with propagate_traces = false" do
       perform_basic_setup do |config|
         config.traces_sample_rate = 1.0
diff --git a/sentry-sidekiq/spec/sentry/sidekiq_spec.rb b/sentry-sidekiq/spec/sentry/sidekiq_spec.rb
index a7b727eb3..d87d7f6a5 100644
--- a/sentry-sidekiq/spec/sentry/sidekiq_spec.rb
+++ b/sentry-sidekiq/spec/sentry/sidekiq_spec.rb
@@ -117,6 +117,18 @@ def retry_last_failed_job
   end
 
   context "with attempt_threshold" do
+    [false, 0].each do |retry_option|
+      it "reports the first failure below the threshold when retry is #{retry_option}" do
+        worker = Class.new(SadWorker)
+        worker.sidekiq_options attempt_threshold: 3, retry: retry_option
+
+        execute_worker(processor, worker)
+
+        expect(transport.events.count).to eq(1)
+        expect(retry_set.count).to eq(0)
+      end
+    end
+
     it "doesn't report the error until attempts equal the threshold" do
       worker = Class.new(SadWorker)
       worker.sidekiq_options attempt_threshold: 3
@@ -320,6 +332,18 @@ def retry_last_failed_job
       expect(event.contexts.dig(:trace, :trace_id)).to eq(transaction.contexts.dig(:trace, :trace_id))
     end
 
+    it "does not log Sidekiq deprecation warnings while processing a job" do
+      log = StringIO.new
+      logger = Logger.new(log)
+      WITH_SIDEKIQ_7 ? Sidekiq.instance_variable_get(:@config).logger = logger : Sidekiq.logger = logger
+
+      execute_worker(processor, HappyWorker)
+
+      expect(log.string).not_to include("deprecated")
+    ensure
+      silence_sidekiq
+    end
+
     context "with instrumenter :otel" do
       before do
         perform_basic_setup do |config|