CircuitBreaker
A CircuitBreaker is used to protect resources or services from being overloaded When a service is being overloaded, interacting with it more will only worsen its overloaded state. Especially when combined with retry mechanisms such as Schedule, in some cases simply using a back-off retry policy might not be sufficient during peak traffics.
To allow such overloaded resources from overloading, CircuitBreaker can help you protect the service by failing-fast. Thus CircuitBreaker helps us to achieve stability and prevent cascading failures in distributed systems.
CircuitBreaker has three CircuitBreaker.State:
Closed: This is its normal state, where requests are being made. The state in which CircuitBreaker starts.
When an exception occurs it increments the failure counter
A successful request will reset the failure counter to zero
When the failure counter reaches the maxFailures threshold, the breaker is tripped into the Open state
Open: The CircuitBreaker will short-circuit/fail-fast all requests
All requests short-circuit/fail-fast with
ExecutionRejectedIf a request is made after the configured resetTimeout passes, the CircuitBreaker is tripped into the a HalfOpen state, allowing one request to go through as a test.
HalfOpen: The CircuitBreaker is in this state while it's allowing a request to go through, as a
test requestAll other requests made while
test requestis still running will short-circuit/fail-fast.If the
test requestsucceeds then the CircuitBreaker is tripped back into Closed, with the resetTimeout and the failures count also reset to initial values.If the
test requestfails, then the CircuitBreaker is tripped back into Open, the resetTimeout is multiplied by the exponentialBackoffFactor, up to the configured maxResetTimeout.
Let's say we'd want to create a CircuitBreaker that only allows us to call a remote service twice, and then whenever more than two requests fail with an exception, the circuit breaker starts short-circuiting failing-fast.
import arrow.core.Either
import arrow.resilience.CircuitBreaker
import kotlinx.coroutines.delay
import kotlin.time.Duration.Companion.seconds
suspend fun main(): Unit {
//sampleStart
val circuitBreaker = CircuitBreaker(
resetTimeout = 2.seconds,
openingStrategy = CircuitBreaker.OpeningStrategy.Count(maxFailures = 2),
exponentialBackoffFactor = 1.2,
maxResetTimeout = 60.seconds,
)
circuitBreaker.protectOrThrow { "I am in Closed: ${circuitBreaker.state()}" }.also(::println)
println("Service getting overloaded . . .")
Either.catch { circuitBreaker.protectOrThrow { throw RuntimeException("Service overloaded") } }.also(::println)
Either.catch { circuitBreaker.protectOrThrow { throw RuntimeException("Service overloaded") } }.also(::println)
circuitBreaker.protectEither { }.also { println("I am Open and short-circuit with ${it}. ${circuitBreaker.state()}") }
println("Service recovering . . .").also { delay(2000) }
circuitBreaker.protectOrThrow { "I am running test-request in HalfOpen: ${circuitBreaker.state()}" }.also(::println)
println("I am back to normal state closed ${circuitBreaker.state()}")
//sampleEnd
}A common pattern to make fault-tolerant/resilient systems is to compose a CircuitBreaker with a backing-off policy retry Schedule to guarantee not overloading the resource and the client interacting with it. but also not the client that is interacting with the resource. Below you can see how the simple retry function will result in Either.Left<CircuitBreaker.RejectedExecution>, but when we combine it with another schedule, it will always call the CircuitBreaker on times that it could've entered the HalfOpen state. The reason why Schedule is not sufficient to make your system resilient is because you also have to take into account parallel calls to your functions, ; In contrast, a CircuitBreaker can track failures of every function call or even different functions to the same resource or service.
import arrow.core.Either
import arrow.resilience.CircuitBreaker
import arrow.resilience.Schedule
import arrow.resilience.retry
import kotlinx.coroutines.delay
import kotlin.time.Duration.Companion.seconds
suspend fun main(): Unit {
suspend fun apiCall(): Unit {
println("apiCall . . .")
throw RuntimeException("Overloaded service")
}
//sampleStart
val circuitBreaker = CircuitBreaker(
resetTimeout = 2.seconds,
openingStrategy = CircuitBreaker.OpeningStrategy.Count(maxFailures = 2),
exponentialBackoffFactor = 2.0, // enable exponentialBackoffFactor
maxResetTimeout = 60.seconds, // limit exponential back-off time
)
suspend fun <A> resilient(schedule: Schedule<Throwable, *>, f: suspend () -> A): A =
schedule.retry { circuitBreaker.protectOrThrow(f) }
Either.catch {
resilient(Schedule.recurs(5), ::apiCall)
}.let { println("recurs(5) apiCall twice and 4x short-circuit result from CircuitBreaker: $it") }
delay(2000)
println("CircuitBreaker ready to half-open")
// Retry once and when the CircuitBreaker opens after 2 failures then retry with exponential back-off with same time as CircuitBreaker's resetTimeout
val fiveTimesWithBackOff = Schedule.recurs<Throwable>(1) andThen
Schedule.exponential(2.seconds) and Schedule.recurs(5)
Either.catch {
resilient(fiveTimesWithBackOff, ::apiCall)
}.let { println("exponential(2.seconds) and recurs(5) always retries with actual apiCall: $it") }
//sampleEnd
}Types
The initial state when initializing a CircuitBreaker is Closed.
Functions
Awaits for this CircuitBreaker to be Closed.
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the CircuitBreaker.State.Closed state.
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the CircuitBreaker.State.HalfOpen state.
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the CircuitBreaker.State.Open state.
Returns a new circuit breaker that wraps the state of the source and that upon a task being rejected will execute the given callback.
Returns a new task that upon execution will execute the given task, but with the protection of this circuit breaker. If an exception in fa occurs, other than an ExecutionRejected exception, it will be rethrown.
Returns a new task that upon execution will execute the given task, but with the protection of this circuit breaker. If an exception in fa occurs it will be rethrown
Returns the current State, meant for debugging purposes.