Merge pull request #293 from rollbar/fix-delayed_job-reports

Fix delayed_job crash reports. Fixes #291

lib/rollbar/delayed_job.rb

@@ -1,20 +1,42 @@
 module Rollbar
  module Delayed
+ class << self
+ attr_accessor :wrapped
+ end
+
+ self.wrapped = false
+
  def self.wrap_worker
- return if @wrapped
- @wrapped = true
+ return if wrapped
+
+ around_invoke_job(&invoke_job_callback)
 
- ::Delayed::Worker.lifecycle.around(:invoke_job) do |job, *args, &block|
+ self.wrapped = true
+ end
+
+ def self.around_invoke_job(&block)
+ ::Delayed::Worker.lifecycle.around(:invoke_job, &block)
+ end
+
+ def self.invoke_job_callback
+ proc do |job, *args, &block|
  begin
  block.call(job, *args)
- rescue Exception => e
- if job.attempts >= ::Rollbar.configuration.dj_threshold
- data = ::Rollbar.configuration.report_dj_data ? job : nil
- ::Rollbar.scope(:request => data).error(e, :use_exception_level_filters => true)
- end
+ rescue => e
+ report(e, job)
+
  raise e
  end
  end
  end
+
+ def self.report(e, job)
+ return unless job.attempts <= ::Rollbar.configuration.dj_threshold
+
+ job_data = job.as_json
+ data = ::Rollbar.configuration.report_dj_data ? job_data : nil
+
+ ::Rollbar.scope(:request => data).error(e, :use_exception_level_filters => true)
+ end
  end
 end

spec/delayed/backend/test.rb

@@ -0,0 +1,133 @@
+require 'ostruct'
+require 'delayed/backend/base'
+
+# This code is taken from delayed_job/spec/delayed/backend/test.rb.
+#
+# It just works as a in memory job backend. Job#save is modified to create
+# a new Delayed::Worker and call #work_off(1) so the job is processed inline.
+
+module Delayed
+ module Backend
+ module Test
+ def self.run
+ worker.work_off(1)
+ end
+
+ def self.worker
+ @worker ||= ::Delayed::Worker.new
+ end
+
+ class Job
+ attr_accessor :id
+ attr_accessor :priority
+ attr_accessor :attempts
+ attr_accessor :handler
+ attr_accessor :last_error
+ attr_accessor :run_at
+ attr_accessor :locked_at
+ attr_accessor :locked_by
+ attr_accessor :failed_at
+ attr_accessor :queue
+
+ include Delayed::Backend::Base
+
+ cattr_accessor :id
+ self.id = 0
+
+ def initialize(hash = {})
+ self.attempts = 0
+ self.priority = 0
+ self.id = (self.class.id += 1)
+ hash.each { |k, v| send(:"#{k}=", v) }
+ end
+
+ def self.all
+ @jobs ||= []
+ end
+
+ def self.count
+ all.size
+ end
+
+ def self.delete_all
+ all.clear
+ end
+
+ def self.create(attrs = {})
+ new(attrs).tap do |o|
+ o.save
+ end
+ end
+
+ def self.create!(*args)
+ create(*args)
+ end
+
+ def self.clear_locks!(worker_name)
+ all.select { |j| j.locked_by == worker_name }.each do |j|
+ j.locked_by = nil
+ j.locked_at = nil
+ end
+ end
+
+ # Find a few candidate jobs to run (in case some immediately get locked by others).
+ def self.find_available(worker_name, limit = 5, max_run_time = Worker.max_run_time) # rubocop:disable CyclomaticComplexity, PerceivedComplexity
+ jobs = all.select do |j|
+ j.run_at <= db_time_now &&
+ (j.locked_at.nil? || j.locked_at < db_time_now - max_run_time || j.locked_by == worker_name) &&
+ !j.failed?
+ end
+ jobs.select! { |j| j.priority <= Worker.max_priority } if Worker.max_priority
+ jobs.select! { |j| j.priority >= Worker.min_priority } if Worker.min_priority
+ jobs.select! { |j| Worker.queues.include?(j.queue) } if Worker.queues.any?
+ jobs.sort_by! { |j| [j.priority, j.run_at] }[0..limit - 1]
+ end
+
+ # Lock this job for this worker.
+ # Returns true if we have the lock, false otherwise.
+ def lock_exclusively!(_max_run_time, worker)
+ now = self.class.db_time_now
+ if locked_by != worker
+ # We don't own this job so we will update the locked_by name and the locked_at
+ self.locked_at = now
+ self.locked_by = worker
+ end
+
+ true
+ end
+
+ def self.db_time_now
+ Time.current
+ end
+
+ def update_attributes(attrs = {})
+ attrs.each { |k, v| send(:"#{k}=", v) }
+ save
+ end
+
+ def destroy
+ self.class.all.delete(self)
+ end
+
+ def save
+ self.run_at ||= Time.current
+
+ self.class.all << self unless self.class.all.include?(self)
+
+ ::Delayed::Backend::Test.run
+
+ true
+ end
+
+ def save!
+ save
+ end
+
+ def reload
+ reset
+ self
+ end
+ end
+ end
+ end
+end

spec/rollbar/delayed_job_spec.rb

@@ -6,41 +6,30 @@
  class FailingJob
  class TestException < Exception; end
 
- def perform
- fail(TestException, 'failing')
+ def do_job_please!(a, b)
+ this = will_crash_again!
  end
  end
 
- module DummyBackend
- class Job
- include Delayed::Backend::Base
-
- attr_accessor :handler, :attempts
-
- def initialize(options = {})
- @payload_object = options[:payload_object]
- @attempts = 0
- end
- end
- end
-
- let(:logger) { Rollbar.logger }
-
  before do
  Rollbar::Delayed.wrap_worker
- Delayed::Worker.delay_jobs = false
- Delayed::Worker.backend = DummyBackend::Job
+ Delayed::Worker.backend = :test
+
+ Delayed::Backend::Test::Job.delete_all
  end
 
+ let(:logger) { Rollbar.logger }
  let(:expected_args) do
- [kind_of(FailingJob::TestException), { :use_exception_level_filters => true}]
+ [kind_of(NoMethodError), { :use_exception_level_filters => true }]
  end
 
- it 'sends the exception' do
- expect_any_instance_of(Rollbar::Notifier).to receive(:error).with(*expected_args)
+ context 'with delayed method without arguments failing' do
+ it 'sends the exception' do
+ expect_any_instance_of(::Delayed::Job).to receive(:as_json).and_call_original
+ expect(Rollbar).to receive(:scope).with(kind_of(Hash)).and_call_original
+ expect_any_instance_of(Rollbar::Notifier).to receive(:error).with(*expected_args)
 
- expect do
- Delayed::Job.enqueue(FailingJob.new)
- end.to raise_error(FailingJob::TestException)
+ FailingJob.new.delay.do_job_please!(:foo, :bar)
+ end
  end
 end

spec/spec_helper.rb

@@ -20,6 +20,7 @@
 Dir["#{File.dirname(__FILE__)}/support/**/*.rb"].each { |f| require f }
 
 RSpec.configure do |config|
+ config.extend(Helpers)
  config.include(NotifierHelpers)
  config.include(FixtureHelpers)
  config.include(EncodingHelpers)
@@ -52,7 +53,3 @@
  end
 end
 
-def local?
- ENV['LOCAL'] == '1'
-end
-

spec/support/helpers.rb

@@ -0,0 +1,5 @@
+module Helpers
+ def local?
+ ENV['LOCAL'] == '1'
+ end
+end