Circuit breaker: fail fast, recover safely

Python · breaker/payments.py

"""A breaker in front of a payment provider that stops answering from 10 s to 31 s.

Callers charge a card every 250 ms for 45 s of simulated time. A failed call costs its caller the 1 s timeout; a call
the breaker rejects costs nothing. The breaker opens at a failure ratio of 50 % once at least 10 calls are in its
10-second window, and stays open for 5 s before it lets one trial call through.
"""
from breaker import BrokenCircuit, CircuitBreaker


class Clock:
    now_ms = 0


def charge():
    if 10_000 <= Clock.now_ms < 31_000:
        raise TimeoutError("no answer within 1 s")
    return "charged"


def log(old, new, reason, at_ms):
    print(f"{at_ms / 1000:6.2f} s  {old} -> {new}: {reason}")


def run(with_breaker):
    breaker = CircuitBreaker(clock=lambda: Clock.now_ms, failure_ratio=0.5, minimum_throughput=10,
                             sampling_ms=10_000, break_ms=5_000, on_change=log if with_breaker else None)
    counts = {"charged": 0, "timed out": 0, "rejected at once": 0}
    for i in range(180):
        Clock.now_ms = i * 250
        try:
            breaker.call(charge) if with_breaker else charge()
            counts["charged"] += 1
        except TimeoutError:
            counts["timed out"] += 1
        except BrokenCircuit:
            counts["rejected at once"] += 1
    return counts


with_breaker = run(with_breaker=True)
without = run(with_breaker=False)
print()
print(f"{'180 calls':<18}{'without':>9}{'with breaker':>14}")
for outcome in with_breaker:
    print(f"{outcome:<18}{without[outcome]:>9}{with_breaker[outcome]:>14}")
print(f"{'seconds waited':<18}{without['timed out']:>9}{with_breaker['timed out']:>14}")

 14.75 s  closed -> open: 20 of 40 calls failed in the window
 19.75 s  open -> half-open: the break is over
 19.75 s  half-open -> open: a trial call failed
 24.75 s  open -> half-open: the break is over
 24.75 s  half-open -> open: a trial call failed
 29.75 s  open -> half-open: the break is over
 29.75 s  half-open -> open: a trial call failed
 34.75 s  open -> half-open: the break is over
 34.75 s  half-open -> closed: a trial call succeeded

180 calls           without  with breaker
charged                  96            81
timed out                84            23
rejected at once          0            76
seconds waited           84            23

breaker/breaker.py

"""A circuit breaker written with the State pattern: closed, open and half-open are objects that decide what happens
to a call. Time comes from a clock that is passed in (milliseconds), so simulations and tests control it.

    breaker = CircuitBreaker(clock=..., failure_ratio=0.5, minimum_throughput=10, sampling_ms=10_000, break_ms=5_000)
    breaker.call(charge)              # runs charge(), or raises BrokenCircuit without calling it
    await breaker.call_async(fetch)   # the same for a coroutine function
"""


class BrokenCircuit(Exception):
    """Raised instead of calling the dependency. retry_after_ms: when the next trial may go, if the breaker knows."""

    def __init__(self, state, retry_after_ms=None):
        super().__init__(f"circuit {state}" + ("" if retry_after_ms is None else f", retry after {retry_after_ms} ms"))
        self.retry_after_ms = retry_after_ms


class Window:
    """The outcomes of the last `sampling_ms`, in a ring of `buckets` counters that each cover an equal slice."""

    def __init__(self, sampling_ms, buckets):
        self.slice_ms = sampling_ms // buckets
        self.ring = [[-1, 0, 0] for _ in range(buckets)]  # [slice number, calls, failures]

    def record(self, now_ms, failed):
        n = now_ms // self.slice_ms
        bucket = self.ring[n % len(self.ring)]
        if bucket[0] != n:  # the slot still holds an older slice: start it again
            bucket[:] = [n, 0, 0]
        bucket[1] += 1
        bucket[2] += int(failed)

    def totals(self, now_ms):
        """(calls, failures) in the slices that are still inside the window."""
        n = now_ms // self.slice_ms
        live = [b for b in self.ring if n - len(self.ring) < b[0] <= n]
        return sum(b[1] for b in live), sum(b[2] for b in live)

    def clear(self):
        for bucket in self.ring:
            bucket[:] = [-1, 0, 0]


class Closed:
    name = "closed"

    def before_call(self, cb):
        pass  # every call goes through

    def on_success(self, cb):
        cb.window.record(cb.clock(), failed=False)

    def on_failure(self, cb):
        now = cb.clock()
        cb.window.record(now, failed=True)
        calls, failures = cb.window.totals(now)
        if calls >= cb.minimum_throughput and failures / calls >= cb.failure_ratio:
            cb.change(Open(now + cb.break_ms), f"{failures} of {calls} calls failed in the window")


class Open:
    name = "open"

    def __init__(self, until_ms):
        self.until_ms = until_ms

    def before_call(self, cb):
        now = cb.clock()
        if now < self.until_ms:
            raise BrokenCircuit(self.name, self.until_ms - now)
        cb.change(HalfOpen(), "the break is over")  # no timer: the first call after the break moves the state
        cb.state.before_call(cb)

    def on_success(self, cb):
        pass  # no call is let through while open

    on_failure = on_success


class HalfOpen:
    name = "half-open"

    def __init__(self):
        self.trials = 0

    def before_call(self, cb):
        if self.trials >= cb.half_open_trials:
            raise BrokenCircuit(self.name)  # the trial calls are taken: everyone else fails fast
        self.trials += 1

    def on_success(self, cb):
        cb.window.clear()
        cb.change(cb.closed_state(), "a trial call succeeded")

    def on_failure(self, cb):
        cb.change(Open(cb.clock() + cb.break_ms), "a trial call failed")


class CircuitBreaker:
    def __init__(self, clock, failure_ratio, minimum_throughput, sampling_ms, break_ms, buckets=10,
                 half_open_trials=1, counts_as_failure=lambda error: True, on_change=None, closed_state=Closed):
        self.clock = clock
        self.failure_ratio = failure_ratio
        self.minimum_throughput = minimum_throughput
        self.sampling_ms = sampling_ms
        self.break_ms = break_ms
        self.half_open_trials = half_open_trials
        self.counts_as_failure = counts_as_failure
        self.on_change = on_change
        self.window = Window(sampling_ms, buckets)
        self.closed_state = closed_state  # the class of the closed state: another one counts failures another way
        self.state = closed_state()

    def change(self, new, reason):
        old, self.state = self.state, new
        if self.on_change:
            self.on_change(old.name, new.name, reason, self.clock())

    def _admit(self):
        self.state.before_call(self)  # raises BrokenCircuit, or lets the call go (and may change the state)
        return self.state             # the state that admitted the call hears how it ended

    def _ended(self, state, error):
        if state is not self.state:
            return  # the breaker moved on while the call ran: a late outcome changes nothing
        if error is None:
            state.on_success(self)
        elif self.counts_as_failure(error):
            state.on_failure(self)
        else:
            state.on_success(self)  # an error that says nothing about the dependency's health

    def call(self, fn):
        state = self._admit()
        try:
            result = fn()
        except Exception as error:
            self._ended(state, error)
            raise
        self._ended(state, None)
        return result

    async def call_async(self, fn):
        state = self._admit()
        try:
            result = await fn()
        except Exception as error:
            self._ended(state, error)
            raise
        self._ended(state, None)
        return result