Listen to Active Job queue time on AJ retries

lib/appsignal/hooks/active_job.rb

@@ -43,15 +43,6 @@ def install
 
       module ActiveJobClassInstrumentation
         def execute(job) # rubocop:disable Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
-          enqueued_at = job["enqueued_at"]
-          queue_start = Time.parse(enqueued_at) if enqueued_at
-          queue_time =
-            if queue_start
-              time_now = Time.now.utc
-              # Calculate queue time and store it as milliseconds
-              (time_now - queue_start) * 1_000
-            end
-
           job_status = nil
           has_wrapper_transaction = Appsignal::Transaction.current?
           transaction =
@@ -81,6 +72,22 @@ def execute(job) # rubocop:disable Metrics/CyclomaticComplexity, Metrics/Perceiv
           transaction_set_error(transaction, exception)
           raise exception
         ensure
+          # Only use the Active Job enqueued_at value if this is an Active Job
+          # powered retry (executations > 0) or when no queue time is set by
+          # the worker library (Delayed Job, Sidekiq, etc)
+          enqueued_at = job["enqueued_at"]
+          using_active_job_retry_mechanism = job["executions"] > 0
+          queue_start =
+            if (using_active_job_retry_mechanism && enqueued_at) || !transaction.queue_start?
+              Time.parse(enqueued_at)
+            end
+          queue_time =
+            if queue_start
+              time_now = Time.now.utc
+              # Calculate queue time and store it as milliseconds
+              (time_now - queue_start) * 1_000
+            end
+
           if transaction
             # Present in Rails 6 and up
             transaction.set_queue_start((queue_start.to_f * 1_000).to_i) if queue_start

lib/appsignal/transaction.rb

@@ -173,6 +173,7 @@ def initialize(namespace, id: SecureRandom.uuid, ext: nil)
       @error_blocks = Hash.new { |hash, key| hash[key] = [] }
       @is_duplicate = false
       @error_set = nil
+      @queue_start = nil
 
       @params = Appsignal::SampleData.new(:params)
       @session_data = Appsignal::SampleData.new(:session_data, Hash)
@@ -564,10 +565,19 @@ def set_queue_start(start)
       return unless start
 
       @ext.set_queue_start(start)
+      @queue_start = start
     rescue RangeError
       Appsignal.internal_logger.warn("Queue start value #{start} is too big")
     end
 
+    # Check if queue start time is set.
+    #
+    # @return [Boolean] true if queue start time is set, false otherwise.
+    # @!visibility private
+    def queue_start?
+      !@queue_start.nil?
+    end
+
     # @!visibility private
     def set_metadata(key, value)
       return unless key && value

spec/lib/appsignal/hooks/activejob_spec.rb

@@ -31,6 +31,7 @@
 
   describe Appsignal::Hooks::ActiveJobHook::ActiveJobClassInstrumentation do
     include ActiveJobHelpers
+
     let(:time) { Time.parse("2001-01-01 10:00:00UTC") }
     let(:namespace) { Appsignal::Transaction::BACKGROUND_JOB }
     let(:queue) { "default" }
@@ -331,6 +332,99 @@ def perform(*_args)
           .map { |event| event["name"] }
         expect(events).to eq(expected_perform_events)
       end
+
+      if DependencyHelper.rails_version >= Gem::Version.new("6.0.0")
+        context "when queue start is already set on the transaction by the worker library" do
+          before do
+            stub_const(
+              "ActiveJob::QueueAdapters::AppsignalTestAdapter",
+              Class.new(ActiveJob::QueueAdapters::InlineAdapter) do
+                def enqueue(job)
+                  ActiveJob::Base.execute(job.serialize.merge(
+                    "enqueued_at" => "2001-01-01T09:00:00.000000000Z"
+                  ))
+                end
+              end
+            )
+
+            stub_const("ProviderWrappedActiveJobTestJob", Class.new(ActiveJob::Base) do
+              self.queue_adapter = :appsignal_test
+
+              def perform(*_args)
+              end
+            end)
+          end
+
+          it "does not overwrite the queue start on first execution" do
+            current_transaction = background_job_transaction
+            set_current_transaction current_transaction
+
+            wrapper_queue_start = (Time.parse("2001-01-01T08:00:00.000000000Z").to_f * 1_000).to_i
+            current_transaction.set_queue_start(wrapper_queue_start)
+
+            queue_job(ProviderWrappedActiveJobTestJob)
+
+            expect(created_transactions.count).to eql(1)
+
+            transaction = current_transaction
+            expect(transaction).to_not be_completed
+            transaction._sample
+            expect(transaction).to have_queue_start(wrapper_queue_start)
+          end
+
+          it "overwrites the queue start set when using Active Job retry_on" do
+            stub_const(
+              "ActiveJob::QueueAdapters::AppsignalTestRetryAdapter",
+              Class.new(ActiveJob::QueueAdapters::InlineAdapter) do
+                def enqueue(job)
+                  # Simulate a retry with updated enqueued_at timestamp
+                  ActiveJob::Base.execute(job.serialize.merge(
+                    # This is a retried job
+                    "executions" => 1,
+                    "exception_executions" => { "RuntimeError" => 1 },
+                    # We should store this queue time
+                    "enqueued_at" => "2001-01-01T09:30:00.000000000Z"
+                  ))
+                end
+              end
+            )
+
+            stub_const("ProviderWrappedActiveJobWithRetryTestJob", Class.new(ActiveJob::Base) do
+              self.queue_adapter = :appsignal_test_retry
+
+              def perform(*_args)
+                # Do nothing
+              end
+            end)
+
+            current_transaction = background_job_transaction
+            set_current_transaction current_transaction
+
+            # Simulate wrapper setting queue start time from when the worker library (Sidekiq)
+            # picked up the job
+            wrapper_queue_start = (Time.parse("2001-01-01T08:00:00.000000000Z").to_f * 1_000).to_i
+            current_transaction.set_queue_start(wrapper_queue_start)
+
+            queue_job(ProviderWrappedActiveJobWithRetryTestJob)
+
+            expect(created_transactions.count).to eql(1)
+
+            transaction = current_transaction
+            expect(transaction).to_not be_completed
+            transaction._sample
+
+            # On retry (job["executions"] > 0), Active Job's enqueued_at should overwrite
+            # the worker library's queue_start because the Active Job's timestamp is only accurate
+            # for Active Job retries
+            active_job_queue_start =
+              (Time.parse("2001-01-01T09:30:00.000000000Z").to_f * 1_000).to_i
+            expect(transaction).to have_queue_start(active_job_queue_start)
+
+            # +1 because store a human readable execution count, not a 0-index one
+            expect(transaction).to include_tags("executions" => 2)
+          end
+        end
+      end
     end
 
     context "with params" do

spec/lib/appsignal/transaction_spec.rb

@@ -1404,6 +1404,24 @@
     end
   end
 
+  describe "#queue_start?" do
+    let(:transaction) { new_transaction }
+
+    context "when the queue start is not set" do
+      it "returns false" do
+        expect(transaction.queue_start?).to be(false)
+      end
+    end
+
+    context "when the queue start is set" do
+      it "returns true" do
+        transaction.set_queue_start(10)
+
+        expect(transaction.queue_start?).to be(true)
+      end
+    end
+  end
+
   describe "#set_metadata" do
     let(:transaction) { new_transaction }