diff --git a/ruby/README.md b/ruby/README.md index 16d52698..52c04683 100644 --- a/ruby/README.md +++ b/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: diff --git a/ruby/lib/ci/queue/circuit_breaker.rb b/ruby/lib/ci/queue/circuit_breaker.rb index c51fd48d..0aa5e301 100644 --- a/ruby/lib/ci/queue/circuit_breaker.rb +++ b/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 diff --git a/ruby/lib/ci/queue/common.rb b/ruby/lib/ci/queue/common.rb index da41fe21..52fcfebb 100644 --- a/ruby/lib/ci/queue/common.rb +++ b/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 diff --git a/ruby/lib/ci/queue/configuration.rb b/ruby/lib/ci/queue/configuration.rb index 6dfe3a86..976dcc3a 100644 --- a/ruby/lib/ci/queue/configuration.rb +++ b/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) diff --git a/ruby/lib/minitest/queue.rb b/ruby/lib/minitest/queue.rb index d7f62564..4ae1a40c 100644 --- a/ruby/lib/minitest/queue.rb +++ b/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? diff --git a/ruby/lib/minitest/queue/runner.rb b/ruby/lib/minitest/queue/runner.rb index ef1e0bb5..106c7b5f 100644 --- a/ruby/lib/minitest/queue/runner.rb +++ b/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 diff --git a/ruby/lib/rspec/queue.rb b/ruby/lib/rspec/queue.rb index 0828a9db..4f4d44e2 100644 --- a/ruby/lib/rspec/queue.rb +++ b/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 diff --git a/ruby/test/ci/queue/circuit_breaker_test.rb b/ruby/test/ci/queue/circuit_breaker_test.rb new file mode 100644 index 00000000..de71aa66 --- /dev/null +++ b/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 diff --git a/ruby/test/ci/queue/configuration_test.rb b/ruby/test/ci/queue/configuration_test.rb index a6728577..9366f3b2 100644 --- a/ruby/test/ci/queue/configuration_test.rb +++ b/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 diff --git a/ruby/test/ci/queue/redis_test.rb b/ruby/test/ci/queue/redis_test.rb index f3139582..e9f30b7f 100644 --- a/ruby/test/ci/queue/redis_test.rb +++ b/ruby/test/ci/queue/redis_test.rb @@ -762,6 +762,26 @@ def test_worker_does_not_pick_up_its_own_requeued_test_when_others_are_available threads&.each(&:kill) end + def test_requeue_does_not_advance_breaker_until_runner_reports_acceptance + queue = worker( + 1, + tests: [TEST_LIST.first], + max_requeues: 1, + requeue_tolerance: 1.0, + max_consecutive_requeues: 1, + ) + + queue.poll do |test| + assert_equal true, queue.requeue(test.queue_entry) + break + end + + breaker = queue.config.circuit_breakers.find do |candidate| + candidate.is_a?(CI::Queue::CircuitBreaker::MaxConsecutiveRequeues) + end + refute breaker.open? + end + def test_circuit_breaker_does_not_count_requeued_failures # Bug: report_failure! was called before the requeue check, so successfully # requeued tests incremented the consecutive failure counter. With @@ -837,7 +857,13 @@ def test_stolen_test_acknowledge_does_not_remove_running_entry def test_stolen_test_requeue_is_rejected_by_ownership_check @redis.flushdb single_test = [TEST_LIST.first] - queue_a = worker(1, tests: single_test, max_requeues: 5, requeue_tolerance: 1.0) + queue_a = worker( + 1, + tests: single_test, + max_requeues: 5, + requeue_tolerance: 1.0, + max_consecutive_requeues: 1, + ) queue_b = worker(2, tests: single_test, max_requeues: 5, requeue_tolerance: 1.0) acquired = false @@ -879,6 +905,15 @@ def test_stolen_test_requeue_is_rejected_by_ownership_check thread.join(5) assert_equal false, worker_a_requeue_result, "Stale worker's requeue should be rejected" + if worker_a_requeue_result + queue_a.report_requeue! + else + queue_a.report_failure! + end + breaker = queue_a.config.circuit_breakers.find do |candidate| + candidate.is_a?(CI::Queue::CircuitBreaker::MaxConsecutiveRequeues) + end + refute breaker.open?, "A rejected requeue must not count as an accepted requeue" assert_predicate queue_a, :exhausted? end @@ -984,7 +1019,14 @@ def shuffled_test_list end def build_queue - worker(1, max_requeues: 1, requeue_tolerance: 0.1, populate: false, max_consecutive_failures: 10) + worker( + 1, + max_requeues: 1, + requeue_tolerance: 0.1, + populate: false, + max_consecutive_failures: 10, + max_consecutive_requeues: 3, + ) end def populate(worker, tests: TEST_LIST.dup) diff --git a/ruby/test/fixtures/requeue_circuit_breaker_suite/.rspec b/ruby/test/fixtures/requeue_circuit_breaker_suite/.rspec new file mode 100644 index 00000000..c99d2e73 --- /dev/null +++ b/ruby/test/fixtures/requeue_circuit_breaker_suite/.rspec @@ -0,0 +1 @@ +--require spec_helper diff --git a/ruby/test/fixtures/requeue_circuit_breaker_suite/spec/requeue_circuit_breaker_spec.rb b/ruby/test/fixtures/requeue_circuit_breaker_suite/spec/requeue_circuit_breaker_spec.rb new file mode 100644 index 00000000..7532274d --- /dev/null +++ b/ruby/test/fixtures/requeue_circuit_breaker_suite/spec/requeue_circuit_breaker_spec.rb @@ -0,0 +1,10 @@ +# frozen_string_literal: true +RSpec.describe "requeue circuit breaker" do + 6.times do |index| + it "completes example #{index}" do + raise "corrupted worker state" if ENV['CORRUPTED_WORKER'] == '1' + + expect(true).to be(true) + end + end +end diff --git a/ruby/test/fixtures/requeue_circuit_breaker_suite/spec/spec_helper.rb b/ruby/test/fixtures/requeue_circuit_breaker_suite/spec/spec_helper.rb new file mode 100644 index 00000000..205ba26b --- /dev/null +++ b/ruby/test/fixtures/requeue_circuit_breaker_suite/spec/spec_helper.rb @@ -0,0 +1,4 @@ +# frozen_string_literal: true +RSpec.configure do |config| + config.backtrace_inclusion_patterns << %r{/test/fixtures/} +end diff --git a/ruby/test/fixtures/test/requeue_circuit_breaker_test.rb b/ruby/test/fixtures/test/requeue_circuit_breaker_test.rb new file mode 100644 index 00000000..e75b7c94 --- /dev/null +++ b/ruby/test/fixtures/test/requeue_circuit_breaker_test.rb @@ -0,0 +1,12 @@ +# frozen_string_literal: true +require 'test_helper' + +class RequeueCircuitBreakerTest < Minitest::Test + 6.times do |index| + define_method("test_worker_health_#{index}") do + flunk "corrupted worker state" if ENV['CORRUPTED_WORKER'] == '1' + + assert true + end + end +end diff --git a/ruby/test/integration/minitest_redis_test.rb b/ruby/test/integration/minitest_redis_test.rb index b0cc163c..b3d450e4 100644 --- a/ruby/test/integration/minitest_redis_test.rb +++ b/ruby/test/integration/minitest_redis_test.rb @@ -701,6 +701,78 @@ def test_circuit_breaker "Expected more than 3 tests to run (requeues shouldn't trip breaker), got: #{output}" end + def test_consecutive_requeue_circuit_breaker_fences_unhealthy_worker + build_id = 'consecutive-requeue-breaker' + worker_a_status = nil + out, err = capture_subprocess_io do + system( + { 'CORRUPTED_WORKER' => '1' }, + @exe, 'run', + '--queue', @redis_url, + '--seed', 'foobar', + '--build', build_id, + '--worker', 'corrupted', + '--timeout', '1', + '--max-requeues', '1', + '--requeue-tolerance', '1', + '--max-consecutive-failures', '1', + '--max-consecutive-requeues', '3', + '-Itest', + 'test/requeue_circuit_breaker_test.rb', + chdir: 'test/fixtures/', + ) + worker_a_status = $?.exitstatus + end + + assert_equal 0, worker_a_status + assert_match(/Ran 3 tests, 3 assertions, 0 failures, 0 errors, 0 skips, 3 requeues/, normalize(out)) + assert_equal "This worker is exiting early because it requeued too many consecutive tests, probably because of some corrupted state.\n", filter_deprecation_warnings(err) + refute_includes err, 'too many consecutive test failures' + assert_equal '3', @redis.hget("build:#{build_id}:requeues-count", '___total___') + assert_equal 3, @redis.llen("build:#{build_id}:worker:corrupted:queue") + assert_equal 6, @redis.llen("build:#{build_id}:queue"), 'Requeued tests should remain available' + + worker_b_status = nil + out, err = capture_subprocess_io do + system( + @exe, 'run', + '--queue', @redis_url, + '--seed', 'foobar', + '--build', build_id, + '--worker', 'healthy', + '--timeout', '1', + '--max-requeues', '1', + '--requeue-tolerance', '1', + '--max-consecutive-failures', '1', + '--max-consecutive-requeues', '3', + '-Itest', + 'test/requeue_circuit_breaker_test.rb', + chdir: 'test/fixtures/', + ) + worker_b_status = $?.exitstatus + end + + assert_equal 0, worker_b_status + assert_empty filter_deprecation_warnings(err) + assert_match(/Ran 6 tests, 6 assertions, 0 failures, 0 errors, 0 skips, 0 requeues/, normalize(out)) + + report_status = nil + out, err = capture_subprocess_io do + system( + @exe, 'report', + '--queue', @redis_url, + '--build', build_id, + '--timeout', '1', + chdir: 'test/fixtures/', + ) + report_status = $?.exitstatus + end + + assert_equal 0, report_status + assert_empty filter_deprecation_warnings(err) + assert_match(/Ran 6 tests, 9 assertions, 0 failures, 0 errors, 0 skips, 3 requeues/, normalize(out)) + end + def test_circuit_breaker_without_requeues out, err = capture_subprocess_io do system( diff --git a/ruby/test/integration/rspec_redis_test.rb b/ruby/test/integration/rspec_redis_test.rb index 20996117..efa3587b 100644 --- a/ruby/test/integration/rspec_redis_test.rb +++ b/ruby/test/integration/rspec_redis_test.rb @@ -15,6 +15,76 @@ def setup File.delete(@order_path) if File.exist?(@order_path) end + def test_consecutive_requeue_circuit_breaker_fences_unhealthy_worker + build_id = 'rspec-consecutive-requeue-breaker' + worker_a_status = nil + out, err = capture_subprocess_io do + system( + { 'CORRUPTED_WORKER' => '1' }, + @exe, + '--queue', @redis_url, + '--seed', '123', + '--build', build_id, + '--worker', 'corrupted', + '--timeout', '1', + '--max-requeues', '1', + '--requeue-tolerance', '1', + '--max-consecutive-failures', '1', + '--max-consecutive-requeues', '3', + 'spec/requeue_circuit_breaker_spec.rb', + chdir: 'test/fixtures/requeue_circuit_breaker_suite/', + ) + worker_a_status = $?.exitstatus + end + + assert_equal 0, worker_a_status + assert_empty filter_deprecation_warnings(err) + assert_match(/3 examples, 0 failures, 3 pending/, normalize(out)) + assert_equal '3', @redis.hget("build:#{build_id}:requeues-count", '___total___') + assert_equal 3, @redis.llen("build:#{build_id}:worker:corrupted:queue") + assert_equal 6, @redis.llen("build:#{build_id}:queue"), 'Requeued examples should remain available' + + worker_b_status = nil + out, err = capture_subprocess_io do + system( + @exe, + '--queue', @redis_url, + '--seed', '123', + '--build', build_id, + '--worker', 'healthy', + '--timeout', '1', + '--max-requeues', '1', + '--requeue-tolerance', '1', + '--max-consecutive-failures', '1', + '--max-consecutive-requeues', '3', + 'spec/requeue_circuit_breaker_spec.rb', + chdir: 'test/fixtures/requeue_circuit_breaker_suite/', + ) + worker_b_status = $?.exitstatus + end + + assert_equal 0, worker_b_status + assert_empty filter_deprecation_warnings(err) + assert_match(/6 examples, 0 failures/, normalize(out)) + + report_status = nil + out, err = capture_subprocess_io do + system( + @exe, + '--queue', @redis_url, + '--build', build_id, + '--report', + '--timeout', '5', + chdir: 'test/fixtures/requeue_circuit_breaker_suite/', + ) + report_status = $?.exitstatus + end + + assert_equal 0, report_status + assert_empty filter_deprecation_warnings(err) + assert_match(/No errors found/, normalize(out)) + end + def test_redis_runner out, err = capture_subprocess_io do system( diff --git a/ruby/test/minitest/queue/runner_test.rb b/ruby/test/minitest/queue/runner_test.rb index f3d9882d..5e66b265 100644 --- a/ruby/test/minitest/queue/runner_test.rb +++ b/ruby/test/minitest/queue/runner_test.rb @@ -8,5 +8,14 @@ def test_multiple_load_paths runner = Runner.new(["-Ilib:test", "-Ielse"]) assert_equal("lib:test:else", runner.send(:load_paths)) end + + def test_max_consecutive_requeues_option + runner = Runner.new(["--max-consecutive-requeues", "3"]) + config = runner.send(:queue_config) + + assert(config.circuit_breakers.any? do |breaker| + breaker.is_a?(CI::Queue::CircuitBreaker::MaxConsecutiveRequeues) + end) + end end end diff --git a/ruby/test/support/queue_helpers.rb b/ruby/test/support/queue_helpers.rb index b7f94fbc..69fcfdec 100644 --- a/ruby/test/support/queue_helpers.rb +++ b/ruby/test/support/queue_helpers.rb @@ -10,7 +10,7 @@ def poll(queue, success = true) failed = !(success.respond_to?(:call) ? success.call(test) : success) if failed if queue.requeue(test.queue_entry) - # Requeued — don't report to circuit breaker + queue.report_requeue! else queue.report_failure! queue.acknowledge(test.queue_entry) diff --git a/ruby/test/support/shared_queue_assertions.rb b/ruby/test/support/shared_queue_assertions.rb index df53b033..5f604eba 100644 --- a/ruby/test/support/shared_queue_assertions.rb +++ b/ruby/test/support/shared_queue_assertions.rb @@ -31,6 +31,33 @@ def test_circuit_breaker assert_equal TEST_LIST.size, @queue.size end + def test_consecutive_requeue_circuit_breaker + 3.times { @queue.report_requeue! } + assert requeue_circuit_breaker.open? + + poll(@queue) do + assert false, "The queue shouldn't have popped a test" + end + assert_equal TEST_LIST.size, @queue.size + end + + def test_success_resets_consecutive_requeues + 2.times { @queue.report_requeue! } + @queue.report_success! + 2.times { @queue.report_requeue! } + + refute requeue_circuit_breaker.open? + end + + def test_final_failure_resets_consecutive_requeues_and_advances_failure_breaker + 2.times { @queue.report_requeue! } + @queue.report_failure! + + refute requeue_circuit_breaker.open? + 9.times { @queue.report_failure! } + assert failure_circuit_breaker.open? + end + def test_size assert_equal TEST_LIST.size, @queue.size poll(@queue) @@ -100,9 +127,18 @@ def config max_requeues: 1, requeue_tolerance: 0.1, max_consecutive_failures: 10, + max_consecutive_requeues: 3, ) end + def failure_circuit_breaker + config.circuit_breakers.find { |breaker| breaker.is_a?(CI::Queue::CircuitBreaker::MaxConsecutiveFailures) } + end + + def requeue_circuit_breaker + config.circuit_breakers.find { |breaker| breaker.is_a?(CI::Queue::CircuitBreaker::MaxConsecutiveRequeues) } + end + def populate(queue, tests: TEST_LIST.dup) queue.populate(tests, random: Random.new(0)) end