feat(sidekiq): migrate to ActiveJob-first integration

Co-Authored-By: Codex <noreply@example.com>

Author: solnic

sentry-rails/lib/sentry/rails/active_job.rb

@@ -21,11 +21,15 @@ def perform_now
           super
         else
           data = _sentry || {}
+          transaction = Sentry.get_current_scope.get_span
+          transaction = nil unless transaction.is_a?(Sentry::Transaction) && transaction.span_id == data["transaction_span_id"]
           SentryReporter.record(
             self,
             trace_headers: data["trace_propagation_headers"],
             user: data["user"],
-            payload: _sentry_job_data
+            delivery_queue: data["delivery_queue"],
+            payload: _sentry_job_data,
+            transaction: transaction
           ) { super }
         end
       end
@@ -139,7 +143,7 @@ def log_producer_span_error(e)
             Sentry.sdk_logger&.error("sentry-rails: failed to record producer span: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
           end
 
-          def record(job, trace_headers: nil, user: nil, payload: nil, &block)
+          def record(job, trace_headers: nil, user: nil, delivery_queue: nil, payload: nil, transaction: nil, &block)
             original_hub = Sentry.get_current_hub_internal
             Sentry.clone_hub_to_current_thread
 
@@ -150,7 +154,7 @@ def record(job, trace_headers: nil, user: nil, payload: nil, &block)
                 consumer.apply_scope_data(scope, job)
 
                 scope.set_transaction_name(consumer.transaction_name(job) || job.class.name, source: :task)
-                scope.set_tags(queue: job.queue_name)
+                scope.set_tags(queue: delivery_queue || job.queue_name)
 
                 scope.set_contexts(active_job: {
                   job_class: job.class.name,
@@ -161,6 +165,7 @@ def record(job, trace_headers: nil, user: nil, payload: nil, &block)
 
                 transaction = consumer.start_transaction(
                   scope, job,
+                  transaction: transaction.is_a?(Sentry::Transaction) ? transaction : nil,
                   trace_headers: trace_headers,
                   payload: payload,
                   messaging_data: -> {

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"]

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,12 +15,18 @@ module Sidekiq
 
     register_integration name: "sidekiq", version: Sentry::Sidekiq::VERSION
 
+    def self.job_extension
+      @job_extension ||= JobExtension.new
+    end
+
+    Sentry.register_job_extension(job_extension)
+
     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"
+          Sentry.configuration.rails.skippable_job_adapters << JobExtension::ACTIVE_JOB_ADAPTER
         end
       end
     end

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

sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb

@@ -1,11 +1,19 @@
 # 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")
+      HANDLED_SIGNATURE = :@__sentry_sidekiq_handled
+
+      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
@@ -15,84 +23,21 @@ class ErrorHandler
       #   https://github.com/sidekiq/sidekiq/pull/6051
       def call(ex, context, sidekiq_config = nil)
         return unless Sentry.initialized?
-
-        context_filter = Sentry::Sidekiq::ContextFilter.new(context)
+        return if ex.instance_variable_get(HANDLED_SIGNATURE)
 
         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

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, retry_limit: 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