Add consecutive requeue circuit breaker

ruby/README.md

@@ -36,7 +36,7 @@ Assuming you use one of the supported CI providers, the command can be as simple
 minitest-queue --queue redis://example.com run -Itest test/**/*_test.rb
 ```
 
-Additionally you can configure the requeue settings (see main README) with `--max-requeues` and `--requeue-tolerance`.
+Additionally you can configure the requeue settings (see main README) with `--max-requeues` and `--requeue-tolerance`. Worker-health circuit breakers are available through `--max-consecutive-failures` and `--max-consecutive-requeues`.
 
 #### Lazy loading (opt-in)
 
@@ -158,6 +158,30 @@ rspec-queue --queue redis://example.com --timeout 600 --report
 
 Because of how `ci-queue` executes the examples, `before(:all)` and `after(:all)` hooks are not supported. `rspec-queue` will explicitly reject them.
 
+### Worker circuit breakers
+
+Both runners support two independent, disabled-by-default circuit breakers:
+
+- `--max-consecutive-failures MAX` stops a worker after `MAX` consecutive final test failures.
+- `--max-consecutive-requeues MAX` stops a worker after `MAX` consecutive failures were accepted for requeueing.
+
+The requeue count is local to each worker and resets whenever that worker produces any non-requeued result, including a pass, skip, ignored flaky result, or final failure. An accepted requeue neither increments nor resets the consecutive-final-failure count. Requeue attempts rejected because of an ownership race or an exhausted retry budget follow the final-failure path and do not increment the requeue count.
+
+For example:
+
+```bash
+minitest-queue --max-consecutive-requeues 10 --queue redis://example.com run test/**/*_test.rb
+rspec-queue --max-consecutive-requeues 10 --queue redis://example.com
+```
+
+The breaker can also be configured programmatically:
+
+```ruby
+CI::Queue::Configuration.new(max_consecutive_requeues: 10)
+# or
+config.max_consecutive_requeues = 10
+```
+
 ## Releasing a New Version
 
 After merging changes to `main`, follow these steps to release and propagate the update:

ruby/lib/ci/queue/circuit_breaker.rb

@@ -8,6 +8,9 @@ module Disabled
         def report_failure!
         end
 
+        def report_requeue!
+        end
+
         def report_success!
         end
 
@@ -32,6 +35,9 @@ def initialize(duration:)
         def report_failure!
         end
 
+        def report_requeue!
+        end
+
         def report_success!
         end
 
@@ -60,6 +66,9 @@ def report_failure!
           @consecutive_failures += 1
         end
 
+        def report_requeue!
+        end
+
         def report_success!
           @consecutive_failures = 0
         end
@@ -72,6 +81,33 @@ def message
           'This worker is exiting early because it encountered too many consecutive test failures, probably because of some corrupted state.'
         end
       end
+
+      class MaxConsecutiveRequeues
+        def initialize(max_consecutive_requeues:)
+          @max = max_consecutive_requeues
+          @consecutive_requeues = 0
+        end
+
+        def report_failure!
+          @consecutive_requeues = 0
+        end
+
+        def report_requeue!
+          @consecutive_requeues += 1
+        end
+
+        def report_success!
+          @consecutive_requeues = 0
+        end
+
+        def open?
+          @consecutive_requeues >= @max
+        end
+
+        def message
+          'This worker is exiting early because it requeued too many consecutive tests, probably because of some corrupted state.'
+        end
+      end
     end
   end
 end

ruby/lib/ci/queue/common.rb

@@ -27,6 +27,10 @@ def report_failure!
         config.circuit_breakers.each(&:report_failure!)
       end
 
+      def report_requeue!
+        config.circuit_breakers.each(&:report_requeue!)
+      end
+
       def report_success!
         config.circuit_breakers.each(&:report_success!)
       end

ruby/lib/ci/queue/configuration.rb

@@ -55,7 +55,7 @@ def load_flaky_tests(path)
       def initialize(
         timeout: 30, build_id: nil, worker_id: nil, max_requeues: 0, requeue_tolerance: 0,
         namespace: nil, seed: nil, flaky_tests: [], statsd_endpoint: nil, max_consecutive_failures: nil,
-        grind_count: nil, max_duration: nil, failure_file: nil, max_test_duration: nil,
+        max_consecutive_requeues: nil, grind_count: nil, max_duration: nil, failure_file: nil, max_test_duration: nil,
         max_test_duration_percentile: 0.5, track_test_duration: false, max_test_failed: nil,
         queue_init_timeout: nil, redis_ttl: 8 * 60 * 60, report_timeout: nil, inactive_workers_timeout: nil,
         export_flaky_tests_file: nil, warnings_file: nil, debug_log: nil, max_missed_heartbeat_seconds: nil, heartbeat_max_test_duration: nil,
@@ -79,6 +79,7 @@ def initialize(
         @track_test_duration = track_test_duration
         @worker_id = worker_id
         self.max_consecutive_failures = max_consecutive_failures
+        self.max_consecutive_requeues = max_consecutive_requeues
         self.max_duration = max_duration
         @redis_ttl = redis_ttl
         @report_timeout = report_timeout
@@ -165,6 +166,12 @@ def max_consecutive_failures=(max)
         end
       end
 
+      def max_consecutive_requeues=(max)
+        if max
+          @circuit_breakers << CircuitBreaker::MaxConsecutiveRequeues.new(max_consecutive_requeues: max)
+        end
+      end
+
       def max_duration=(duration)
         if duration
           @circuit_breakers << CircuitBreaker::Timeout.new(duration: duration)

ruby/lib/minitest/queue.rb

@@ -198,6 +198,7 @@ def handle_test_result(reporter, example, result)
         end
 
         if failed && CI::Queue.requeueable?(result) && queue.requeue(example.queue_entry)
+          queue.report_requeue!
           result.requeue!
           if CI::Queue.debug?
             $stderr.puts "[ci-queue][requeue] test_id=#{example.id} error_class=#{result.failures.first&.class} error=#{result.failures.first&.message&.lines&.first&.chomp}"
@@ -576,7 +577,7 @@ def loaded_tests
           Queue.run(*args)
 
           if queue.config.circuit_breakers.any?(&:open?)
-            STDERR.puts queue.config.circuit_breakers.map(&:message).join(' ').strip
+            STDERR.puts queue.config.circuit_breakers.select(&:open?).map(&:message).join(' ').strip
           end
 
           if queue.max_test_failed?

ruby/lib/minitest/queue/runner.rb

@@ -692,6 +692,15 @@ def parser
             queue_config.max_consecutive_failures = max
           end
 
+          help = <<~EOS
+            Defines after how many consecutive accepted requeues the worker will be considered unhealthy and stop reserving tests.
+            Defaults to disabled.
+          EOS
+          opts.separator ""
+          opts.on('--max-consecutive-requeues MAX', Integer, help) do |max|
+            queue_config.max_consecutive_requeues = max
+          end
+
           help = <<~EOS
             Must set this option in report and report_grind command if you set --max-test-duration in the report_grind
           EOS

ruby/lib/rspec/queue.rb

@@ -158,6 +158,15 @@ def parser(options)
           queue_config.max_consecutive_failures = Integer(max)
         end
 
+        help = <<~EOS
+          Defines after how many consecutive accepted requeues the worker will be considered unhealthy and stop reserving tests.
+          Defaults to disabled.
+        EOS
+        parser.separator ""
+        parser.on('--max-consecutive-requeues MAX', *help) do |max|
+          queue_config.max_consecutive_requeues = Integer(max)
+        end
+
         help = <<~EOS
           Defines how long the test report remain after the test run, in seconds.
           Defaults to 28,800 (8 hours)
@@ -215,22 +224,21 @@ def start(*)
 
       def finish(reporter, acknowledge: true)
         if acknowledge && reporter.respond_to?(:requeue)
-          if @exception
-            reporter.report_failure!
-          else
-            reporter.report_success!
-          end
-
           if @exception && CI::Queue.requeueable?(@exception) && reporter.requeue
+            reporter.report_requeue!
             reporter.cancel_run!
             dup.mark_as_requeued!(reporter)
             return true
+          end
+
+          if @exception
+            reporter.report_failure!
           else
-            super(reporter)
+            reporter.report_success!
           end
-        else
-          super(reporter)
         end
+
+        super(reporter)
       end
 
       def reset!
@@ -417,6 +425,10 @@ def report_failure!
         @queue.report_failure!
       end
 
+      def report_requeue!
+        @queue.report_requeue!
+      end
+
       def requeue
         @queue.requeue(@example.queue_entry)
       end

ruby/test/ci/queue/circuit_breaker_test.rb

@@ -0,0 +1,89 @@
+# frozen_string_literal: true
+require 'test_helper'
+
+module CI::Queue
+  class CircuitBreakerTest < Minitest::Test
+    def test_max_consecutive_requeues_starts_closed
+      refute max_consecutive_requeues.open?
+    end
+
+    def test_max_consecutive_requeues_opens_at_threshold
+      breaker = max_consecutive_requeues(max: 3)
+
+      2.times { breaker.report_requeue! }
+      refute breaker.open?
+
+      breaker.report_requeue!
+      assert breaker.open?
+    end
+
+    def test_success_resets_consecutive_requeues
+      breaker = max_consecutive_requeues(max: 3)
+
+      2.times { breaker.report_requeue! }
+      breaker.report_success!
+      2.times { breaker.report_requeue! }
+
+      refute breaker.open?
+    end
+
+    def test_failure_resets_consecutive_requeues
+      breaker = max_consecutive_requeues(max: 3)
+
+      2.times { breaker.report_requeue! }
+      breaker.report_failure!
+      2.times { breaker.report_requeue! }
+
+      refute breaker.open?
+    end
+
+    def test_non_consecutive_requeues_do_not_open_breaker
+      breaker = max_consecutive_requeues(max: 3)
+
+      breaker.report_requeue!
+      breaker.report_requeue!
+      breaker.report_success!
+      breaker.report_requeue!
+
+      refute breaker.open?
+    end
+
+    def test_max_consecutive_requeues_message_is_specific
+      message = max_consecutive_requeues.message
+
+      assert_match(/consecutive tests/, message)
+      assert_match(/requeued/, message)
+      refute_match(/test failures/, message)
+    end
+
+    def test_requeues_do_not_increment_or_reset_consecutive_failures
+      breaker = CircuitBreaker::MaxConsecutiveFailures.new(max_consecutive_failures: 2)
+
+      breaker.report_failure!
+      breaker.report_requeue!
+      refute breaker.open?
+
+      breaker.report_failure!
+      assert breaker.open?
+    end
+
+    def test_all_breakers_accept_requeue_events
+      breakers = [
+        CircuitBreaker::Disabled,
+        CircuitBreaker::Timeout.new(duration: 60),
+        CircuitBreaker::MaxConsecutiveFailures.new(max_consecutive_failures: 2),
+        max_consecutive_requeues,
+      ]
+
+      breakers.each do |breaker|
+        assert_respond_to breaker, :report_requeue!
+      end
+    end
+
+    private
+
+    def max_consecutive_requeues(max: 2)
+      CircuitBreaker::MaxConsecutiveRequeues.new(max_consecutive_requeues: max)
+    end
+  end
+end

ruby/test/ci/queue/configuration_test.rb

@@ -214,5 +214,25 @@ def test_heartbeat_max_test_duration_defaults
       assert_equal 3, config.heartbeat_max_test_duration
     end
 
+    def test_max_consecutive_requeues_is_disabled_by_default
+      config = Configuration.new
+
+      refute(config.circuit_breakers.any? { |breaker| breaker.is_a?(CircuitBreaker::MaxConsecutiveRequeues) })
+    end
+
+    def test_max_consecutive_requeues_can_be_set_during_initialization
+      config = Configuration.new(max_consecutive_requeues: 3)
+
+      assert(config.circuit_breakers.any? { |breaker| breaker.is_a?(CircuitBreaker::MaxConsecutiveRequeues) })
+    end
+
+    def test_max_consecutive_requeues_can_be_assigned
+      config = Configuration.new
+
+      config.max_consecutive_requeues = 3
+
+      assert(config.circuit_breakers.any? { |breaker| breaker.is_a?(CircuitBreaker::MaxConsecutiveRequeues) })
+    end
+
   end
 end