Enable automatic message visibility refresh

lib/aws/rails/sqs_active_job/executor.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require 'aws-sdk-sqs'
 require 'concurrent'
 
 module Aws
@@ -16,13 +17,18 @@ class Executor
            fallback_policy: :caller_runs # slow down the producer thread
         }.freeze
 
-        def initialize(options = {})
+        def initialize(options = {}, refresh_timeout = nil)
           @executor = Concurrent::ThreadPoolExecutor.new(DEFAULTS.merge(options))
+          # Monitor threads used to refresh visiblity
+          @refresh_timeout = refresh_timeout
+          @monitor = Concurrent::ThreadPoolExecutor.new(DEFAULTS.merge(options)) if refresh_timeout
           @logger = options[:logger] || ActiveSupport::Logger.new(STDOUT)
         end
 
         # TODO: Consider catching the exception and sleeping instead of using :caller_runs
         def execute(message)
+          # Used to tell the visibilty refresh thread to give up
+          refresh_visibility = true
           @executor.post(message) do |message|
             begin
               job = JobRunner.new(message)
@@ -36,8 +42,28 @@ def execute(message)
               job_msg = job ? "#{job.id}[#{job.class_name}]" : 'unknown job'
               @logger.info "Error processing job #{job_msg}: #{e}"
               @logger.debug e.backtrace.join("\n")
+            ensure
+              refresh_visibility = false
             end
           end
+
+          @monitor.post(message) do |message|
+            while refresh_visibility
+              begin
+                # Extend the visibility timeout to the provided refresh timeout
+                message.change_visibility({ visibility_timeout: @refresh_timeout })
+                # Wait half the refresh timeout, and repeat
+                sleep(@refresh_timeout / 2.0)
+              rescue Aws::SQS::Errors::MessageNotInflight => e
+                # Message has been deleted or otherwise released. We don't care!
+                break
+              rescue => e
+                # If anything else goes wrong, we should log and break the loop.
+                @logger.error("Monitor process failed for message: #{message.id}. Error: #{e}")
+                break
+              end
+            end
+          end if @monitor
         end
 
         def shutdown(timeout=nil)

lib/aws/rails/sqs_active_job/poller.rb

@@ -48,7 +48,11 @@ def run
 
           Signal.trap('INT') { raise Interrupt }
           Signal.trap('TERM') { raise Interrupt }
-          @executor = Executor.new(max_threads: @options[:threads], logger: @logger, max_queue: @options[:backpressure])
+          @executor = Executor.new({
+            max_threads: @options[:threads],
+            logger: @logger,
+            max_queue: @options[:backpressure]
+          }, refresh_timeout = @options[:refresh_timeout])
 
           poll
         rescue Interrupt
@@ -112,6 +116,7 @@ def parse_args(argv)
             opts.on("-t", "--threads INTEGER", Integer, "The maximum number of worker threads to create.  Defaults to 2x the number of processors available on this system.") { |a| out[:threads] = a }
             opts.on("-b", "--backpressure INTEGER", Integer, "The maximum number of messages to have waiting in the Executor queue. This should be a low, but non zero number.  Messages in the Executor queue cannot be picked up by other processes and will slow down shutdown.") { |a| out[:backpressure] = a }
             opts.on("-m", "--max_messages INTEGER", Integer, "Max number of messages to receive in a batch from SQS.") { |a| out[:max_messages] = a }
+            opts.on("-r", "--refresh_timeout", Integer, "Enables message visibility refresh, Refresh timeout is in seconds.") { |a| out[:refresh_timeout] = a }
             opts.on("-v", "--visibility_timeout INTEGER", Integer, "The visibility timeout is the number of seconds that a message will not be processable by any other consumers. You should set this value to be longer than your expected job runtime to prevent other processes from picking up an running job.  See the SQS Visibility Timeout Documentation at https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html.") { |a| out[:visibility_timeout] = a }
             opts.on("-s", "--shutdown_timeout INTEGER", Integer, "The amount of time to wait for a clean shutdown.  Jobs that are unable to complete in this time will not be deleted from the SQS queue and will be retryable after the visibility timeout.") { |a| out[:shutdown_timeout] = a }
           }