FIFO Queues Work With Delay. Logging/Perform Helpers.

Author: metaskills

TODO.md

@@ -47,6 +47,7 @@ end
 Q: How do I handle job priorities?
 A: Use different queues.
 
+* How we allow FIFO queues to work with delay using message visibility.
 
 ## Our Siqekiq Interfaces
 

lib/lambdakiq/job.rb

@@ -1,7 +1,7 @@
 module Lambdakiq
   class Job
 
-    attr_reader :record, :error, :sent_timestamp
+    attr_reader :record, :error
 
     class << self
 
@@ -33,13 +33,13 @@ def queue
       Lambdakiq.client.queues[active_job.queue_name]
     end
 
-    def performed?
-      @started_at.present? && !error
-    end
-
     def perform
-      @started_at = Time.current
-      ActiveJob::Base.execute(job_data)
+      if queue.fifo? && record.fifo_delay_seconds?
+        delay_fifo_message_visibility
+      else
+        ActiveJob::Base.execute(job_data)
+      end
+      delete_message
     rescue Exception => e
       perform_error(e)
     end
@@ -79,5 +79,10 @@ def max_receive_count?
       record.max_receive_count? || record.receive_count >= queue.max_receive_count
     end
 
+    def delay_fifo_message_visibility
+      params = client_params.merge visibility_timeout: record.fifo_delay_visibility_timeout
+      client.change_message_visibility(params)
+    end
+
   end
 end

lib/lambdakiq/metrics.rb

@@ -66,7 +66,7 @@ def message
     end
 
     def timestamp
-      Time.now.strftime('%s%3N').to_i
+      Time.current.strftime('%s%3N').to_i
     end
 
   end

lib/lambdakiq/record.rb

@@ -23,10 +23,22 @@ def attributes
       data['attributes']
     end
 
+    def fifo_delay_visibility_timeout
+      fifo_delay_seconds - (Time.current - sent_at).to_i
+    end
+
+    def fifo_delay_seconds
+      data.dig('messageAttributes', 'delay_seconds', 'stringValue').try(:to_i)
+    end
+
+    def fifo_delay_seconds?
+      fifo_delay_seconds && (sent_at + fifo_delay_seconds).future?
+    end
+
     def sent_at
        @sent_at ||= begin
-        ts = attributes['SentTimestamp'].to_i
-        Time.at(ts/1000)
+        ts = attributes['SentTimestamp'].to_i / 1000
+        Time.zone ? Time.zone.at(ts) : Time.at(ts)
       end
     end
 

test/cases/job_test.rb

@@ -1,23 +1,69 @@
 require 'test_helper'
 
 class JobTest < LambdakiqSpec
+  it 'must perform basic job' do
+    Lambdakiq::Job.handler(event_basic)
+    expect(delete_message).must_be :present?
+    expect(change_message_visibility).must_be_nil
+    expect(perform_buffer_last_value).must_equal 'BasicJob with: "test"'
+    expect(active_job_log).must_include 'Performing TestHelper::Jobs::BasicJob'
+    expect(active_job_log).must_include 'Performed TestHelper::Jobs::BasicJob'
+  end
+
   it 'must change message visibility to next value for failed jobs' do
     event = event_basic attributes: { ApproximateReceiveCount: '7' }, job_class: 'TestHelper::Jobs::ErrorJob'
     expect(->{ Lambdakiq::Job.handler(event) }).must_raise 'HELL'
     expect(change_message_visibility).must_be :present?
     expect(change_message_visibility_params[:visibility_timeout]).must_equal 1416
+    expect(perform_buffer_last_value).must_equal 'ErrorJob with: "test"'
+    expect(active_job_log).must_include 'Performing TestHelper::Jobs::ErrorJob'
+    expect(active_job_log).must_include 'Error performing TestHelper::Jobs::ErrorJob'
   end
 
   it 'wraps returned errors with no backtrace which avoids excessive/duplicate cloudwatch logging' do
     event = event_basic job_class: 'TestHelper::Jobs::ErrorJob'
     error = expect(->{ Lambdakiq::Job.handler(event) }).must_raise 'HELL'
     expect(error.class.name).must_equal 'Lambdakiq::JobError'
     expect(error.backtrace).must_equal []
+    expect(perform_buffer_last_value).must_equal 'ErrorJob with: "test"'
+    expect(active_job_log).must_include 'Performing TestHelper::Jobs::ErrorJob'
+    expect(active_job_log).must_include 'Error performing TestHelper::Jobs::ErrorJob'
   end
 
   it 'must delete message for failed jobs at the end of the queue/message max receive count' do
     event = event_basic attributes: { ApproximateReceiveCount: '8' }, job_class: 'TestHelper::Jobs::ErrorJob'
     Lambdakiq::Job.handler(event)
     expect(delete_message).must_be :present?
+    expect(perform_buffer_last_value).must_equal 'ErrorJob with: "test"'
+    expect(active_job_log).must_include 'Performing TestHelper::Jobs::ErrorJob'
+    expect(active_job_log).must_include 'Error performing TestHelper::Jobs::ErrorJob'
+  end
+
+  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(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
+    expect(active_job_log).must_be :blank?
+  end
+
+  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(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
+    expect(active_job_log).must_be :blank?
+  end
+
+  it 'must perform and allow fifo queue to use message visibility as delay but not when SentTimestamp is too far in the past' do
+    event = event_basic_delay minutes: 2, timestamp: 3.minutes.ago.strftime('%s%3N')
+    Lambdakiq::Job.handler(event)
+    expect(delete_message).must_be :present?
+    expect(change_message_visibility).must_be_nil
+    expect(perform_buffer_last_value).must_equal 'BasicJob with: "test"'
+    expect(active_job_log).must_include 'Performing TestHelper::Jobs::BasicJob'
+    expect(active_job_log).must_include 'Performed TestHelper::Jobs::BasicJob'
   end
 end

test/test_helper.rb

@@ -5,25 +5,29 @@
 Bundler.require :default, :development, :test
 require 'rails'
 require 'aws-sdk-sqs'
+require 'stringio'
 require 'minitest/autorun'
 require 'minitest/focus'
 require 'mocha/minitest'
 Dir['test/test_helper/*.{rb}'].each { |f| require_relative "../#{f}" }
 
 ActiveJob::Base.queue_adapter = :lambdakiq
-ActiveJob::Base.logger = Logger.new(IO::NULL)
 Lambdakiq::Client.default_options.merge! stub_responses: true
 
 class LambdakiqSpec < Minitest::Spec
 
   include TestHelper::ClientHelpers,
           TestHelper::ApiRequestHelpers,
           TestHelper::EventHelpers,
-          TestHelper::QueueHelpers
+          TestHelper::QueueHelpers,
+          TestHelper::LogHelpers,
+          TestHelper::PerformHelpers
 
   before do
     client_reset!
     client_stub_responses
+    reset_active_job_logger!
+    perform_buffer_clear!
   end
 
 end

test/test_helper/event_helpers.rb

@@ -10,5 +10,19 @@ def event_basic(overrides = {})
       Events::Basic.create(overrides)
     end
 
+    def event_basic_delay(minutes: 5, timestamp: Time.current.strftime('%s%3N'))
+      Events::Basic.create(
+        attributes: { SentTimestamp: timestamp },
+        messageAttributes: {
+          delay_seconds: {
+            stringValue: minutes.minutes.to_s,
+            stringListValues: [],
+            binaryListValues: [],
+            dataType: 'String'
+          }
+        }
+      )
+    end
+
   end
 end

test/test_helper/jobs/basic_job.rb

@@ -2,7 +2,7 @@ module TestHelper
   module Jobs
     class BasicJob < ApplicationJob
       def perform(object)
-        object
+        TestHelper::PerformBuffer.add "BasicJob with: #{object.inspect}"
       end
     end
   end

test/test_helper/jobs/basic_nofifo_job.rb

@@ -3,7 +3,7 @@ module Jobs
     class BasicNofifoJob < ApplicationJob
       queue_as ENV['TEST_QUEUE_NAME'].sub('.fifo','')
       def perform(object)
-        object
+        TestHelper::PerformBuffer.add "BasicNofifoJob with: #{object.inspect}"
       end
     end
   end

test/test_helper/jobs/error_job.rb

@@ -2,6 +2,7 @@ module TestHelper
   module Jobs
     class ErrorJob < ApplicationJob
       def perform(object)
+        TestHelper::PerformBuffer.add "ErrorJob with: #{object.inspect}"
         raise('HELL')
       end
     end

test/test_helper/log_helpers.rb

@@ -0,0 +1,15 @@
+module TestHelper
+  module LogHelpers
+    extend ActiveSupport::Concern
+
+    included do
+      let(:active_job_log)  { ActiveJob::Base.logger.instance_variable_get(:@logdev).instance_variable_get(:@dev).string }
+    end
+
+    private
+
+    def reset_active_job_logger!
+      ActiveJob::Base.logger = Logger.new(StringIO.new)
+    end
+  end
+end

test/test_helper/perform_helpers.rb

@@ -0,0 +1,32 @@
+module TestHelper
+  module PerformBuffer
+    def clear
+      values.clear
+    end
+
+    def add(value)
+      values << value
+    end
+
+    def values
+      @values ||= []
+    end
+
+    def last_value
+      values.last
+    end
+
+    extend self
+  end
+  module PerformHelpers
+    private
+
+    def perform_buffer_clear!
+      PerformBuffer.clear
+    end
+
+    def perform_buffer_last_value
+      PerformBuffer.last_value
+    end
+  end
+end