Implement Worker mixin with lambdakiq_options.

Author: metaskills

TODO.md

@@ -48,6 +48,8 @@ Q: How do I handle job priorities?
 A: Use different queues.
 
 * How we allow FIFO queues to work with delay using message visibility.
+* Your SQS queue must have a `RedrivePolicy` policy!
+  https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-sqs-queues.html#aws-sqs-queue-redrive
 
 ## Our Siqekiq Interfaces
 

lib/lambdakiq/job.rb

@@ -36,15 +36,20 @@ def queue
       Lambdakiq.client.queues[active_job.queue_name]
     end
 
+    def executions
+      active_job.executions
+    end
+
     def perform
       fifo_delay? ? fifo_delay : execute
-      delete_message
-    rescue Exception => e
-      perform_error(e)
     end
 
     def execute
       ActiveJob::Base.execute(job_data)
+      delete_message
+    rescue Exception => e
+      increment_executions
+      perform_error(e)
     end
 
     private
@@ -79,7 +84,13 @@ def client
     end
 
     def max_receive_count?
-      record.max_receive_count? || record.receive_count >= queue.max_receive_count
+      executions > retry_limit
+    end
+
+    def retry_limit
+      config_retry = [Lambdakiq.config.max_retries, 12].min
+      [ (active_job.lambdakiq_retry || config_retry),
+        (queue.max_receive_count - 1) ].min
     end
 
     def fifo_delay?
@@ -91,5 +102,9 @@ def fifo_delay
       client.change_message_visibility(params)
     end
 
+    def increment_executions
+      active_job.executions = active_job.executions + 1
+    end
+
   end
 end

lib/lambdakiq/record.rb

@@ -50,10 +50,6 @@ def receive_count
       @receive_count ||= attributes['ApproximateReceiveCount'].to_i
     end
 
-    def max_receive_count?
-      receive_count >= Lambdakiq.config.max_retries
-    end
-
     def next_visibility_timeout
       @next_visibility_timeout ||= Backoff.backoff(receive_count)
     end

lib/lambdakiq/worker.rb

@@ -1,5 +1,24 @@
 module Lambdakiq
   module Worker
+    extend ActiveSupport::Concern
+
+    included do
+      class_attribute :lambdakiq_options_hash,
+                      instance_predicate: false,
+                      default: Hash.new
+    end
+
+    class_methods do
+
+      def lambdakiq_options(options = {})
+        self.lambdakiq_options_hash = options.symbolize_keys
+      end
+
+    end
+
+    def lambdakiq_retry
+      lambdakiq_options_hash[:retry]
+    end
 
   end
 end

test/cases/job_test.rb

@@ -1,6 +1,23 @@
 require 'test_helper'
 
 class JobTest < LambdakiqSpec
+
+  it '#active_job - a deserialize representation of what will be executed' do
+    aj = job.active_job
+    expect(aj).must_be_instance_of TestHelper::Jobs::BasicJob
+    expect(aj.job_id).must_equal '527cd37e-08f4-4aa8-9834-a46220cdc5a3'
+    expect(aj.queue_name).must_equal queue_name
+    expect(aj.enqueued_at).must_equal '2020-11-30T13:07:36Z'
+    expect(aj.executions).must_equal 0
+    expect(aj.provider_job_id).must_equal '9081fe74-bc79-451f-a03a-2fe5c6e2f807'
+  end
+
+  it '#active_job - executions uses ApproximateReceiveCount' do
+    event = event_basic attributes: { ApproximateReceiveCount: '3' }
+    aj = job(event: event).active_job
+    expect(aj.executions).must_equal 2
+  end
+
   it 'must perform basic job' do
     Lambdakiq::Job.handler(event_basic)
     expect(delete_message).must_be :present?
@@ -44,6 +61,7 @@ class JobTest < LambdakiqSpec
   end
 
   it 'must delete message for failed jobs at the end of the queue/message max receive count' do
+    # See ClientHelpers for setting queue to max receive count of 8.
     event = event_basic attributes: { ApproximateReceiveCount: '8' }, job_class: 'TestHelper::Jobs::ErrorJob'
     Lambdakiq::Job.handler(event)
     expect(delete_message).must_be :present?
@@ -55,6 +73,7 @@ class JobTest < LambdakiqSpec
   it 'must not perform and allow fifo queue to use message visibility as delay' do
     event = event_basic_delay minutes: 6
     Lambdakiq::Job.handler(event)
+    expect(delete_message).must_be :blank?
     expect(change_message_visibility).must_be :present?
     expect(change_message_visibility_params[:visibility_timeout]).must_equal 6.minutes
     expect(perform_buffer_last_value).must_be_nil
@@ -64,6 +83,7 @@ class JobTest < LambdakiqSpec
   it 'must not perform and allow fifo queue to use message visibility as delay (using SentTimestamp)' do
     event = event_basic_delay minutes: 10, timestamp: 2.minutes.ago.strftime('%s%3N')
     Lambdakiq::Job.handler(event)
+    expect(delete_message).must_be :blank?
     expect(change_message_visibility).must_be :present?
     expect(change_message_visibility_params[:visibility_timeout]).must_equal 8.minutes
     expect(perform_buffer_last_value).must_be_nil
@@ -79,4 +99,32 @@ class JobTest < LambdakiqSpec
     expect(logger).must_include 'Performing TestHelper::Jobs::BasicJob'
     expect(logger).must_include 'Performed TestHelper::Jobs::BasicJob'
   end
+
+  it 'must use `lambdakiq_options` retry options set to 0 and not retry job' do
+    event = event_basic job_class: 'TestHelper::Jobs::ErrorJobNoRetry'
+    Lambdakiq::Job.handler(event)
+    expect(delete_message).must_be :present?
+    expect(perform_buffer_last_value).must_equal 'ErrorJobNoRetry with: "test"'
+    expect(logger).must_include 'Performing TestHelper::Jobs::ErrorJobNoRetry'
+    expect(logger).must_include 'Error performing TestHelper::Jobs::ErrorJobNoRetry'
+  end
+
+  it 'must use `lambdakiq_options` retry options set to 1 and retry job' do
+    event = event_basic job_class: 'TestHelper::Jobs::ErrorJobOneRetry'
+    error = expect(->{ Lambdakiq::Job.handler(event) }).must_raise 'HELL'
+    expect(delete_message).must_be :blank?
+    expect(perform_buffer_last_value).must_equal 'ErrorJobOneRetry with: "test"'
+    expect(change_message_visibility).must_be :present?
+    expect(change_message_visibility_params[:visibility_timeout]).must_equal 30.seconds
+    expect(logger).must_include 'Performing TestHelper::Jobs::ErrorJobOneRetry'
+    expect(logger).must_include 'Error performing TestHelper::Jobs::ErrorJobOneRetry'
+  end
+
+  private
+
+  def job(event: event_basic)
+    record = Lambdakiq::Event.records(event).first
+    Lambdakiq::Job.new(record)
+  end
+
 end

test/cases/record_test.rb

@@ -35,8 +35,4 @@ class RecordTest < LambdakiqSpec
     expect(record.receive_count).must_equal 1
   end
 
-  it '#max_receive_count?' do
-    expect(record.max_receive_count?).must_equal false
-  end
-
 end

test/test_helper/jobs.rb

@@ -2,3 +2,5 @@
 require 'test_helper/jobs/basic_job'
 require 'test_helper/jobs/basic_nofifo_job'
 require 'test_helper/jobs/error_job'
+require 'test_helper/jobs/error_job_no_retry'
+require 'test_helper/jobs/error_job_one_retry'

test/test_helper/jobs/application_job.rb

@@ -2,6 +2,7 @@ module TestHelper
   module Jobs
     class ApplicationJob < ActiveJob::Base
       queue_as ENV['TEST_QUEUE_NAME']
+      include Lambdakiq::Worker
     end
   end
 end

test/test_helper/jobs/error_job_no_retry.rb

@@ -0,0 +1,11 @@
+module TestHelper
+  module Jobs
+    class ErrorJobNoRetry < ApplicationJob
+      lambdakiq_options retry: 0
+      def perform(object)
+        TestHelper::PerformBuffer.add "ErrorJobNoRetry with: #{object.inspect}"
+        raise('HELL')
+      end
+    end
+  end
+end

test/test_helper/jobs/error_job_one_retry.rb

@@ -0,0 +1,11 @@
+module TestHelper
+  module Jobs
+    class ErrorJobOneRetry < ApplicationJob
+      lambdakiq_options retry: 1
+      def perform(object)
+        TestHelper::PerformBuffer.add "ErrorJobOneRetry with: #{object.inspect}"
+        raise('HELL')
+      end
+    end
+  end
+end