Merge pull request #31 from rainforestapp/RF-6662

1.0.1

lib/queue_classic_plus/version.rb

@@ -1,3 +1,3 @@
 module QueueClassicPlus
-  VERSION = "1.0.0"
+  VERSION = "1.0.1"
 end

lib/queue_classic_plus/worker.rb

@@ -7,17 +7,8 @@ class CustomWorker < QC::Worker
     BACKOFF_WIDTH = 10
     FailedQueue = QC::Queue.new("failed_jobs")
 
-    def enqueue_failed(job, e)
-      sql = "INSERT INTO #{QC::TABLE_NAME} (q_name, method, args, last_error) VALUES ('failed_jobs', $1, $2, $3)"
-      last_error = e.backtrace ? ([e.message] + e.backtrace ).join("\n") : e.message
-      QC.default_conn_adapter.execute sql, job[:method], JSON.dump(job[:args]), last_error
-
-      QueueClassicPlus.exception_handler.call(e, job)
-      Metrics.increment("qc.errors", source: @q_name)
-    end
-
     def handle_failure(job, e)
-      QueueClassicPlus.logger.info "Handling exception #{e.message} for job #{job[:id]}"
+      QueueClassicPlus.logger.info "Handling exception #{e.class} - #{e.message} for job #{job[:id]}"
 
       force_retry = false
       if connection_error?(e)
@@ -33,58 +24,65 @@ def handle_failure(job, e)
         force_retry = true
       end
 
-      klass = job_klass(job)
+      @failed_job = job
 
-      if force_retry && !(klass.respond_to?(:disable_retries) && klass.disable_retries)
+      if force_retry && !(failed_job_class.respond_to?(:disable_retries) && failed_job_class.disable_retries)
         Metrics.increment("qc.force_retry", source: @q_name)
-
-        retry_with_remaining(klass, job, e, 0)
+        retry_with_remaining(e)
       # The mailers doesn't have a retries_on?
-      elsif klass && klass.respond_to?(:retries_on?) && klass.retries_on?(e)
+    elsif failed_job_class && failed_job_class.respond_to?(:retries_on?) && failed_job_class.retries_on?(e)
         Metrics.increment("qc.retry", source: @q_name)
-
-        backoff = (max_retries(klass) - remaining_retries(klass, job)) * BACKOFF_WIDTH
-
-        retry_with_remaining(klass, job, e, backoff)
+        retry_with_remaining(e)
       else
-        enqueue_failed(job, e)
+        enqueue_failed(e)
       end
 
-      FailedQueue.delete(job[:id])
+      FailedQueue.delete(@failed_job[:id])
     end
 
     private
 
-    def retry_with_remaining(klass, job, e, backoff)
-      @remaining_retries = remaining_retries(klass, job)
-
-      if @remaining_retries > 0
-        klass.restart_in(backoff, @remaining_retries, *job[:args])
+    def retry_with_remaining(e)
+      if remaining_retries > 0
+        failed_job_class.restart_in(backoff, remaining_retries - 1, *@failed_job[:args])
       else
-        enqueue_failed(job, e)
+        enqueue_failed(e)
       end
     end
 
-    def max_retries(klass)
-      klass.respond_to?(:max_retries) ? klass.max_retries : 5
+    def max_retries
+      failed_job_class.respond_to?(:max_retries) ? failed_job_class.max_retries : 5
     end
 
-    def remaining_retries(klass, job)
-      @remaining_retries ? @remaining_retries - 1 : (job[:remaining_retries] || max_retries(klass)).to_i - 1
+    def remaining_retries
+      (@failed_job[:remaining_retries] || max_retries).to_i
     end
 
-    def job_klass(job)
+    def failed_job_class
       begin
-        Object.const_get(job[:method].split('.')[0])
+        Object.const_get(@failed_job[:method].split('.')[0])
       rescue NameError
         nil
       end
     end
 
+    def backoff
+      (max_retries - (remaining_retries - 1)) * BACKOFF_WIDTH
+    end
+
     def connection_error?(e)
       CONNECTION_ERRORS.any? { |klass| e.kind_of? klass } ||
         (e.respond_to?(:original_exception) &&
          CONNECTION_ERRORS.any? { |klass| e.original_exception.kind_of? klass })
     end
+
+    def enqueue_failed(e)
+      sql = "INSERT INTO #{QC::TABLE_NAME} (q_name, method, args, last_error) VALUES ('failed_jobs', $1, $2, $3)"
+      last_error = e.backtrace ? ([e.message] + e.backtrace ).join("\n") : e.message
+      QC.default_conn_adapter.execute sql, @failed_job[:method], JSON.dump(@failed_job[:args]), last_error
+
+      QueueClassicPlus.exception_handler.call(e, @failed_job)
+      Metrics.increment("qc.errors", source: @q_name)
+    end
   end
 end

spec/sample_jobs.rb

@@ -35,7 +35,7 @@ def self.perform should_raise
 
     class TestJob < QueueClassicPlus::Base
       @queue = :low
-      retry! on: SomeException, max: 5
+      retry! on: SomeException, max: 1
 
       def self.perform should_raise
         raise SomeException if should_raise

spec/worker_spec.rb

@@ -88,6 +88,19 @@
       end
     end
 
+    context 'with a custom exception having max: 1 retry' do
+      before { Jobs::Tests::TestJob.enqueue_perform(true) }
+
+      it 'retries' do
+        QueueClassicPlus::Metrics.should_receive(:increment).with('qc.retry', source: nil )
+        Timecop.freeze do
+          worker.work
+          expect(failed_queue.count).to eq 0
+          QueueClassicMatchers::QueueClassicRspec.find_by_args('low', 'Jobs::Tests::TestJob._perform', [true]).first['remaining_retries'].should eq "0"
+        end
+      end
+    end
+
     context 'with non-connection based PG jobs' do
       before { Jobs::Tests::UniqueViolationTestJob.enqueue_perform }