Merge pull request #42 from rainforestapp/RF-22643/pp/retry-failure

[RF-22643] Deserialize arguments before reenqueuing a failed job

Author: magni-

.circleci/config.yml

@@ -0,0 +1,67 @@
+version: 2.1
+
+jobs:
+  test:
+    docker:
+      - image: circleci/ruby:2.7.4-node
+        auth:
+          username: $DOCKERHUB_USERNAME
+          password: $DOCKERHUB_TOKEN
+        environment:
+          DATABASE_URL: postgres://circleci:circleci@127.0.0.1:5432/queue_classic_plus_test
+      - image: circleci/postgres:9.6.6-alpine
+        auth:
+          username: $DOCKERHUB_USERNAME
+          password: $DOCKERHUB_TOKEN
+        environment:
+          POSTGRES_USER: circleci
+          POSTGRES_PASSWORD: circleci
+          POSTGRES_DB: queue_classic_plus_test
+    steps:
+      - checkout
+      - run:
+          name: run tests
+          command: |
+            bundle check --path=vendor/bundle || bundle install --path=vendor/bundle --jobs=4 --retry=3
+            bundle exec rspec
+
+  push_to_rubygems:
+    docker:
+      - image: circleci/ruby:2.7.4
+        auth:
+          username: $DOCKERHUB_USERNAME
+          password: $DOCKERHUB_TOKEN
+    steps:
+      - checkout
+      - run:
+          name: Create .gem/credentials file
+          command: |
+            mkdir ~/.gem
+            echo "---
+              :rubygems_api_key: $RUBYGEMS_API_KEY
+            " > ~/.gem/credentials
+            chmod 600 ~/.gem/credentials
+      - run:
+          name: Release to rubygems
+          command: |
+            gem build queue_classic_plus
+            gem push queue_classic_plus-*.gem
+
+workflows:
+  version: 2
+  gem_release:
+    jobs:
+      - test:
+          context:
+            - DockerHub
+
+      - push_to_rubygems:
+          filters:
+            branches:
+              ignore:
+                - /.*/
+            tags:
+              only:
+                - /^v.*/
+          context:
+            - DockerHub

.gitignore

@@ -1,6 +1,7 @@
 *.gem
 *.rbc
 .bundle
+.byebug_history
 .config
 .yardoc
 Gemfile.lock
@@ -9,6 +10,7 @@ _yardoc
 coverage
 doc/
 lib/bundler/man
+log/
 pkg
 rdoc
 spec/reports

.travis.yml

@@ -12,6 +12,7 @@ group :development do
 end
 
 group :test do
+  gem 'byebug'
   gem 'rake'
   gem 'rspec'
   gem 'timecop'

Gemfile

@@ -1,6 +1,6 @@
 # QueueClassicPlus
 
-[![Build Status](https://travis-ci.org/rainforestapp/queue_classic_plus.svg?branch=master)](https://travis-ci.org/rainforestapp/queue_classic_plus)
+[![rainforestapp](https://circleci.com/gh/rainforestapp/queue_classic_plus.svg?branch=master)](https://app.circleci.com/pipelines/github/rainforestapp/queue_classic_plus?branch=master)
 
 [queue_classic](https://github.com/QueueClassic/queue_classic) is a simple Postgresql backed DB queue. However, it's a little too simple to use it as the main queueing system of a medium to large app. This was developed at [Rainforest QA](https://www.rainforestqa.com/).
 

README.md

@@ -142,10 +142,7 @@ def self.list
       execute q
     end
 
-    private
-    def self.execute(sql, *args)
-      QC.default_conn_adapter.execute(sql, *args)
-    end
+    protected
 
     def self.serialized(args)
       if defined?(Rails)
@@ -162,5 +159,11 @@ def self.deserialized(args)
         args
       end
     end
+
+    private
+
+    def self.execute(sql, *args)
+      QC.default_conn_adapter.execute(sql, *args)
+    end
   end
 end

lib/queue_classic_plus/base.rb

@@ -1,3 +1,3 @@
 module QueueClassicPlus
-  VERSION = "4.0.0.alpha7"
+  VERSION = '4.0.0.alpha8'.freeze
 end

lib/queue_classic_plus/version.rb

@@ -25,6 +25,7 @@ def handle_failure(job, e)
       end
 
       @failed_job = job
+      @failed_job_args = failed_job_class ? failed_job_class.deserialized(job[:args]) : job[:args]
 
       if force_retry && !(failed_job_class.respond_to?(:disable_retries) && failed_job_class.disable_retries)
         Metrics.increment("qc.force_retry", source: @q_name)
@@ -44,7 +45,7 @@ def handle_failure(job, e)
 
     def retry_with_remaining(e)
       if remaining_retries > 0
-        failed_job_class.restart_in(backoff, remaining_retries - 1, *@failed_job[:args])
+        failed_job_class.restart_in(backoff, remaining_retries - 1, *@failed_job_args)
       else
         enqueue_failed(e)
       end
@@ -79,7 +80,7 @@ def connection_error?(e)
     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
+      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)

lib/queue_classic_plus/worker.rb

@@ -4,6 +4,7 @@
 require 'queue_classic_matchers'
 require_relative './sample_jobs'
 require_relative './helpers'
+require 'byebug'
 require 'pry'
 require 'ddtrace'
 

spec/spec_helper.rb

@@ -67,6 +67,45 @@
       end
     end
 
+    context 'when Rails is defined' do
+      require 'active_job'
+      require 'active_job/arguments'
+
+      before { stub_const('Rails', Struct.new(:logger).new(Logger.new(STDOUT))) }
+
+      it 'retries' do
+        expect do
+          job_type.enqueue_perform(:foo)
+        end.to change_queue_size_of(job_type).by(1)
+
+        expect(Jobs::Tests::LockedTestJob).to have_queue_size_of(1)
+        expect(failed_queue.count).to eq(0)
+        expect(QueueClassicMatchers::QueueClassicRspec.find_by_args('low', 'Jobs::Tests::LockedTestJob._perform', [:foo]).first['remaining_retries']).to be_nil
+
+        expect(QueueClassicPlus::Metrics).to receive(:increment).with('qc.retry', source: nil).twice
+
+        Timecop.freeze do
+          worker.work
+
+          expect(failed_queue.count).to eq(0) # not enqueued on Failed
+          expect(QueueClassicMatchers::QueueClassicRspec.find_by_args('low', 'Jobs::Tests::LockedTestJob._perform', [:foo]).first['remaining_retries']).to eq "4"
+          expect(Jobs::Tests::LockedTestJob).to have_scheduled(:foo).at(Time.now + described_class::BACKOFF_WIDTH) # should have scheduled a retry for later
+        end
+
+        Timecop.freeze(Time.now + (described_class::BACKOFF_WIDTH * 2)) do
+          # the job should be re-enqueued with a decremented retry count
+          jobs = QueueClassicMatchers::QueueClassicRspec.find_by_args('low', 'Jobs::Tests::LockedTestJob._perform', [:foo])
+          expect(jobs.size).to eq(1)
+          job = jobs.first
+          expect(job['remaining_retries'].to_i).to eq(job_type.max_retries - 1)
+          expect(job['locked_by']).to be_nil
+          expect(job['locked_at']).to be_nil
+        end
+
+        worker.work
+      end
+    end
+
     context 'when PG connection reaped during a job' do
       before { Jobs::Tests::ConnectionReapedTestJob.enqueue_perform }