add lumping_interval feature in Semian Circuit Breaker

fix notation for index for SlidingWindow and edge case for empty array

Author: abishanan-shopify

lib/semian.rb

@@ -310,6 +310,11 @@ def create_circuit_breaker(name, **options)
                                        else
                                          options[:error_threshold_timeout_enabled]
                                        end,
+      lumping_interval: if options[:lumping_interval].nil?
+                          0
+                        else
+                          options[:lumping_interval]
+                        end,
       exceptions: Array(exceptions) + [::Semian::BaseError],
       half_open_resource_timeout: options[:half_open_resource_timeout],
       implementation: implementation(**options),

lib/semian/circuit_breaker.rb

@@ -17,7 +17,8 @@ class CircuitBreaker # :nodoc:
 
     def initialize(name, exceptions:, success_threshold:, error_threshold:,
       error_timeout:, implementation:, half_open_resource_timeout: nil,
-      error_threshold_timeout: nil, error_threshold_timeout_enabled: true)
+      error_threshold_timeout: nil, error_threshold_timeout_enabled: true,
+      lumping_interval: 0)
       @name = name.to_sym
       @success_count_threshold = success_threshold
       @error_count_threshold = error_threshold
@@ -26,6 +27,11 @@ def initialize(name, exceptions:, success_threshold:, error_threshold:,
       @error_timeout = error_timeout
       @exceptions = exceptions
       @half_open_resource_timeout = half_open_resource_timeout
+      @lumping_interval = lumping_interval
+
+      if @lumping_interval > @error_threshold_timeout
+        raise ArgumentError, "lumping_interval (#{@lumping_interval}) must be less than error_threshold_timeout (#{@error_threshold_timeout})"
+      end
 
       @errors = implementation::SlidingWindow.new(max_size: @error_count_threshold)
       @successes = implementation::Integer.new
@@ -137,11 +143,14 @@ def push_error(error)
 
     def push_time
       time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
       if error_threshold_timeout_enabled
         @errors.reject! { |err_time| err_time + @error_threshold_timeout < time }
       end
 
-      @errors << time
+      if @errors.empty? || @errors.last <= time - @lumping_interval
+        @errors << time
+      end
     end
 
     def log_state_transition(new_state)

test/circuit_breaker_test.rb

@@ -600,4 +600,111 @@ def test_notify_state_transition
   ensure
     Semian.unsubscribe(:test_notify_state_transition)
   end
+
+  def test_lumping_interval_prevents_rapid_error_accumulation
+    resource = Semian.register(
+      :lumping_test,
+      bulkhead: false,
+      exceptions: [SomeError],
+      error_threshold: 3,
+      error_timeout: 10,
+      success_threshold: 1,
+      lumping_interval: 2,
+    )
+
+    # Trigger an arbitrary number of errors within lumping interval
+    6.times do
+      trigger_error!(resource)
+    end
+
+    # Should not open circuit since errors are lumped
+    assert_circuit_closed(resource)
+
+    time_travel(3) do
+      6.times do
+        trigger_error!(resource)
+      end
+
+      assert_circuit_closed(resource)
+
+      time_travel(3) do
+        # A single error should open circuit because we have reached the error threshold
+        trigger_error!(resource)
+
+        assert_circuit_opened(resource)
+      end
+    end
+  ensure
+    Semian.destroy(:lumping_test)
+  end
+
+  def test_lumping_interval_respects_error_threshold
+    resource = Semian.register(
+      :lumping_threshold_test,
+      bulkhead: false,
+      exceptions: [SomeError],
+      error_threshold: 2,
+      error_timeout: 5,
+      success_threshold: 1,
+      lumping_interval: 1,
+    )
+
+    # First error
+    trigger_error!(resource)
+
+    assert_circuit_closed(resource)
+
+    # Wait past lumping interval
+    time_travel(2) do
+      # Second error should open circuit
+      trigger_error!(resource)
+
+      assert_circuit_opened(resource)
+    end
+  ensure
+    Semian.destroy(:lumping_threshold_test)
+  end
+
+  def test_lumping_interval_with_zero_value
+    resource = Semian.register(
+      :lumping_zero_test,
+      bulkhead: false,
+      exceptions: [SomeError],
+      error_threshold: 2,
+      error_timeout: 5,
+      success_threshold: 1,
+      lumping_interval: 0,
+    )
+
+    # First error
+    trigger_error!(resource)
+
+    assert_circuit_closed(resource)
+
+    # Second error should open circuit immediately
+    trigger_error!(resource)
+
+    assert_circuit_opened(resource)
+  ensure
+    Semian.destroy(:lumping_zero_test)
+  end
+
+  def test_lumping_interval_cannot_be_greater_than_error_threshold_timeout
+    error = assert_raises(ArgumentError) do
+      Semian.register(
+        :lumping_validation_test,
+        bulkhead: false,
+        exceptions: [SomeError],
+        error_threshold: 2,
+        error_timeout: 5,
+        error_threshold_timeout: 3,
+        success_threshold: 1,
+        lumping_interval: 4,
+      )
+    end
+
+    assert_match(/lumping_interval \(4\) must be less than error_threshold_timeout \(3\)/, error.message)
+  ensure
+    Semian.destroy(:lumping_validation_test)
+  end
 end