Require stable Freno recovery

README.md

@@ -202,14 +202,16 @@ module Freno
 
     DEFAULT_WAIT_SECONDS = 0.5
     DEFAULT_MAX_WAIT_SECONDS = 10
+    DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES = 3
 
     def initialize(client: nil,
                     app: nil,
                     mapper: Mapper::Identity,
                     instrumenter: Instrumenter::Noop,
                     circuit_breaker: CircuitBreaker::Noop,
                     wait_seconds: DEFAULT_WAIT_SECONDS,
-                    max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS)
+                    max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS,
+                    required_consecutive_successes: DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES)
 
 
       @client           = client
@@ -238,6 +240,16 @@ You optionally provide the time you want the throttler to sleep in case the chec
 If replication lags badly, you can control until when you want to keep sleeping
 and retrying the check by setting `max_wait_seconds`. When that times out, the throttle will raise a `Freno::Throttler::WaitedTooLong` error.
 
+An initially healthy check proceeds immediately. After any failed check,
+`required_consecutive_successes` passing checks are required before the block
+runs. Passing samples must be consecutive; another rejection resets the count.
+This prevents an oscillating metric from releasing work on a single healthy
+trough.
+
+`WaitedTooLong` represents normal sustained throttling and does not count as a
+circuit-breaker failure. Freno transport and decision errors still fail the
+circuit breaker.
+
 #### Instrumenting the throttler
 
 You can also configure the throttler with an `instrumenter` collaborator to subscribe to events happening during the `throttle` call.

lib/freno/client.rb

@@ -233,7 +233,7 @@ def decorated(request)
         outermost = to_decorate[0]
         current = outermost
 
-        (to_decorate[1..]).each do |decorator|
+        to_decorate[1..].each do |decorator|
           current.request = decorator
           current = current.request
         end

lib/freno/client/errors.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
 module Freno
-  Error = Class.new(StandardError)
+  class Error < StandardError
+  end
 end

lib/freno/client/preconditions.rb

@@ -5,7 +5,8 @@ class Client
     module Preconditions
       module_function
 
-      PreconditionNotMet = Class.new(ArgumentError)
+      class PreconditionNotMet < ArgumentError
+      end
 
       class Checker
         attr_reader :errors

lib/freno/throttler.rb

@@ -35,6 +35,7 @@ module Freno
   class Throttler
     DEFAULT_WAIT_SECONDS = 0.5
     DEFAULT_MAX_WAIT_SECONDS = 10
+    DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES = 3
     REQUIRED_ARGS = %i[
       client
       app
@@ -43,6 +44,7 @@ class Throttler
       circuit_breaker
       wait_seconds
       max_wait_seconds
+      required_consecutive_successes
     ].freeze
 
     attr_accessor :client,
@@ -51,7 +53,8 @@ class Throttler
                   :instrumenter,
                   :circuit_breaker,
                   :wait_seconds,
-                  :max_wait_seconds
+                  :max_wait_seconds,
+                  :required_consecutive_successes
 
     # Initializes a new instance of the throttler
     #
@@ -99,14 +102,19 @@ class Throttler
     #     seconds the throttler will wait in total for replicas to catch-up
     #     before raising a `WaitedTooLong` error.
     #
+    #  - `:required_consecutive_successes`: The number of consecutive passing
+    #     checks required after any failed check. An initially passing check
+    #     still proceeds immediately.
+    #
     def initialize(
       client: nil,
       app: nil,
       mapper: Mapper::Identity,
       instrumenter: Instrumenter::Noop,
       circuit_breaker: CircuitBreaker::Noop,
       wait_seconds: DEFAULT_WAIT_SECONDS,
-      max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS
+      max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS,
+      required_consecutive_successes: DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES
     )
       @client           = client
       @app              = app
@@ -115,6 +123,7 @@ def initialize(
       @circuit_breaker  = circuit_breaker
       @wait_seconds     = wait_seconds
       @max_wait_seconds = max_wait_seconds
+      @required_consecutive_successes = required_consecutive_successes
 
       yield self if block_given?
 
@@ -165,6 +174,8 @@ def throttle(context = nil, **options)
       store_names = mapper.call(context)
       instrument(:called, store_names: store_names)
       waited = 0
+      throttled = false
+      consecutive_successes = 0
 
       while true
         unless circuit_breaker.allow_request?
@@ -173,19 +184,43 @@ def throttle(context = nil, **options)
         end
 
         if all_stores_ok?(store_names, **options)
-          instrument(:succeeded, store_names: store_names, waited: waited)
-          circuit_breaker.success
-          break
+          consecutive_successes += 1
+          if !throttled || consecutive_successes >= required_consecutive_successes
+            instrument(
+              :succeeded,
+              store_names: store_names,
+              waited: waited,
+              consecutive_successes: consecutive_successes
+            )
+            circuit_breaker.success
+            break
+          end
+        else
+          throttled = true
+          consecutive_successes = 0
         end
 
         if waited + wait_seconds > max_wait_seconds
-          instrument(:waited_too_long, store_names: store_names, waited: waited, max: max_wait_seconds)
-          circuit_breaker.failure
+          instrument(
+            :waited_too_long,
+            store_names: store_names,
+            waited: waited,
+            max: max_wait_seconds,
+            consecutive_successes: consecutive_successes,
+            required_consecutive_successes: required_consecutive_successes
+          )
           raise WaitedTooLong.new(waited_seconds: waited, max_wait_seconds: max_wait_seconds)
         else
           wait
           waited += wait_seconds
-          instrument(:waited, store_names: store_names, waited: waited, max: max_wait_seconds)
+          instrument(
+            :waited,
+            store_names: store_names,
+            waited: waited,
+            max: max_wait_seconds,
+            consecutive_successes: consecutive_successes,
+            required_consecutive_successes: required_consecutive_successes
+          )
         end
       end
 
@@ -204,6 +239,9 @@ def validate_args
       unless max_wait_seconds > wait_seconds
         errors << "max_wait_seconds (#{max_wait_seconds}) has to be greather than wait_seconds (#{wait_seconds})"
       end
+      unless required_consecutive_successes.is_a?(Integer) && required_consecutive_successes.positive?
+        errors << "required_consecutive_successes (#{required_consecutive_successes}) must be a positive integer"
+      end
 
       raise ArgumentError, errors.join("\n") if errors.any?
     end

test/freno/throttler_test.rb

@@ -10,6 +10,15 @@ def test_validations
     assert_includes ex.message, "app must be provided"
     assert_includes ex.message, "client must be provided"
     assert_includes ex.message, "max_wait_seconds (0.5) has to be greather than wait_seconds (1)"
+
+    ex = assert_raises(ArgumentError) do
+      Freno::Throttler.new(
+        client: sample_client,
+        app: :github,
+        required_consecutive_successes: 0
+      )
+    end
+    assert_includes ex.message, "required_consecutive_successes (0) must be a positive integer"
   end
 
   def test_using_the_default_identity_mapper
@@ -80,17 +89,17 @@ def test_sleeps_when_a_check_fails_and_then_calls_the_block
     block_called = false
 
     stub = sample_client
-    stub.expects(:check?).times(2)
+    stub.expects(:check?).times(4)
       .with(app: :github, store_name: :mysqla, options: {})
-      .returns(false).then.returns(true)
+      .returns(false).then.returns(true).then.returns(true).then.returns(true)
 
     throttler = Freno::Throttler.new do |t|
       t.client = stub
       t.app = :github
       t.mapper = ->(_context) { [:mysqla] }
       t.instrumenter = MemoryInstrumenter.new
     end
-    throttler.expects(:wait).once
+    throttler.expects(:wait).times(3)
 
     throttler.throttle do
       block_called = true
@@ -105,10 +114,12 @@ def test_sleeps_when_a_check_fails_and_then_calls_the_block
 
     waited_events = throttler.instrumenter.events_for("throttler.waited")
 
-    assert_equal 1, waited_events.count
+    assert_equal 3, waited_events.count
     assert_equal [:mysqla], waited_events.first[:store_names]
     assert_in_delta 0.5, waited_events.first[:waited], 0.01
     assert_equal 10, waited_events.first[:max]
+    assert_equal 3, waited_events.last[:required_consecutive_successes]
+    assert_equal 2, waited_events.last[:consecutive_successes]
 
     assert_equal 0, throttler.instrumenter.count("throttler.waited_too_long")
     assert_equal 0, throttler.instrumenter.count("throttler.freno_errored")
@@ -158,6 +169,65 @@ def test_raises_waited_too_long_if_freno_checks_failed_consistenly
     assert_equal 0, throttler.instrumenter.count("throttler.circuit_open")
   end
 
+  def test_recovery_requires_consecutive_passing_checks
+    block_called = false
+
+    stub = sample_client
+    stub.expects(:check?).times(7)
+      .with(app: :github, store_name: :mysqla, options: {})
+      .returns(false)
+      .then.returns(true)
+      .then.returns(false)
+      .then.returns(true)
+      .then.returns(true)
+      .then.returns(false)
+      .then.returns(true)
+
+    throttler = Freno::Throttler.new do |t|
+      t.client = stub
+      t.app = :github
+      t.mapper = ->(_context) { [:mysqla] }
+      t.instrumenter = MemoryInstrumenter.new
+      t.wait_seconds = 1
+      t.max_wait_seconds = 6
+    end
+    throttler.expects(:wait).times(6)
+
+    assert_raises(Freno::Throttler::WaitedTooLong) do
+      throttler.throttle do
+        block_called = true
+      end
+    end
+
+    refute block_called
+
+    event = throttler.instrumenter.events_for("throttler.waited_too_long").first
+
+    assert_equal 1, event[:consecutive_successes]
+    assert_equal 3, event[:required_consecutive_successes]
+  end
+
+  def test_waited_too_long_does_not_fail_the_circuit_breaker
+    client = sample_client
+    client.stubs(:check?).returns(false)
+    circuit_breaker = mock
+    circuit_breaker.stubs(:allow_request?).returns(true)
+    circuit_breaker.expects(:failure).never
+
+    throttler = Freno::Throttler.new(
+      client: client,
+      app: :github,
+      circuit_breaker: circuit_breaker,
+      wait_seconds: 1,
+      max_wait_seconds: 2
+    )
+    throttler.expects(:wait).times(2)
+
+    assert_raises(Freno::Throttler::WaitedTooLong) do
+      throttler.throttle(:mysqla) { flunk "throttled block should not run" }
+    end
+  end
+
   def test_raises_a_specific_error_in_case_freno_itself_errored
     block_called = false