notifier.rb in good_job-3.4.1

- old
+ new

@@ -38,10 +38,11 @@
       ActiveRecord::ConnectionNotEstablished
       ActiveRecord::StatementInvalid
       PG::UnableToSend
       PG::Error
     ].freeze
+    CONNECTION_ERRORS_REPORTING_THRESHOLD = 3
 
     # @!attribute [r] instances
     #   @!scope class
     #   List of all instantiated Notifiers in the current process.
     #   @return [Array<GoodJob::Notifier>, nil]
@@ -68,10 +69,12 @@
 
     # @param recipients [Array<#call, Array(Object, Symbol)>]
     def initialize(*recipients)
       @recipients = Concurrent::Array.new(recipients)
       @listening = Concurrent::AtomicBoolean.new(false)
+      @connection_errors_count = Concurrent::AtomicFixnum.new(0)
+      @connection_errors_reported = Concurrent::AtomicBoolean.new(false)
 
       self.class.instances << self
 
       create_executor
       listen
@@ -126,19 +129,28 @@
     # Invoked on completion of ThreadPoolExecutor task
     # @!visibility private
     # @return [void]
     def listen_observer(_time, _result, thread_error)
       if thread_error
-        GoodJob._on_thread_error(thread_error)
         ActiveSupport::Notifications.instrument("notifier_notify_error.good_job", { error: thread_error })
 
         connection_error = CONNECTION_ERRORS.any? do |error_string|
           error_class = error_string.safe_constantize
           next unless error_class
 
           thread_error.is_a? error_class
         end
+
+        if connection_error
+          @connection_errors_count.increment
+          if @connection_errors_reported.false? && @connection_errors_count.value >= CONNECTION_ERRORS_REPORTING_THRESHOLD
+            GoodJob._on_thread_error(thread_error)
+            @connection_errors_reported.make_true
+          end
+        else
+          GoodJob._on_thread_error(thread_error)
+        end
       end
 
       return if shutdown?
 
       listen(delay: connection_error ? RECONNECT_INTERVAL : 0)
@@ -173,10 +185,12 @@
                   thr_recipients.each do |recipient|
                     target, method_name = recipient.is_a?(Array) ? recipient : [recipient, :call]
                     target.send(method_name, parsed_payload)
                   end
                 end
+
+                reset_connection_errors
               end
             end
           end
         ensure
           run_callbacks :unlisten do
@@ -220,8 +234,13 @@
         end
         sleep WAIT_INTERVAL
       else
         sleep WAIT_INTERVAL
       end
+    end
+
+    def reset_connection_errors
+      @connection_errors_count.value = 0
+      @connection_errors_reported.make_false
     end
   end
 end