Implement CircuitBreaker(failure_threshold, reset_timeout, clock) with a method call(fn) and a
state attribute ("closed", "open" or "half_open"):
fn(). A success resets the consecutive-failure count; a failure (any exception)
increments it and re-raises. Reaching failure_threshold consecutive failures opens the circuit.reset_timeout seconds have passed since it opened (by clock()), raise
CircuitOpenError (provided) without calling fn. Otherwise switch to half_open and treat this call
as a trial.fn() once. Success → closed (count reset). Failure → open again (timer restarts)
and re-raise.The harness replays timed calls and records [outcome, state] after each one.
Implement CircuitBreaker. Input: failure_threshold: int, reset_timeout: float, clock: () -> float. Returns: a class.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
class CircuitOpenError(Exception):
pass
def __harness__(cls, threshold, timeout, events):
now = {"t": 0.0}
breaker = cls(threshold, timeout, lambda: now["t"])
out = []
for t, result in events:
now["t"] = t
def op():
if result == "fail":
raise ConnectionError("downstream failed")
return "ok"
try:
outcome = breaker.call(op)
except CircuitOpenError:
outcome = "rejected"
except ConnectionError:
outcome = "fail"
out.append([outcome, breaker.state])
return out
class CircuitBreaker:
def __init__(self, failure_threshold: int, reset_timeout: float, clock):
self.state = "closed"
def call(self, fn):
return fn()
Run executes the public tests. Submit grades against public and hidden tests.
Implement CircuitBreaker(failure_threshold, reset_timeout, clock) with a method call(fn) and a
state attribute ("closed", "open" or "half_open"):
fn(). A success resets the consecutive-failure count; a failure (any exception)
increments it and re-raises. Reaching failure_threshold consecutive failures opens the circuit.reset_timeout seconds have passed since it opened (by clock()), raise
CircuitOpenError (provided) without calling fn. Otherwise switch to half_open and treat this call
as a trial.fn() once. Success → closed (count reset). Failure → open again (timer restarts)
and re-raise.The harness replays timed calls and records [outcome, state] after each one.
Implement CircuitBreaker. Input: failure_threshold: int, reset_timeout: float, clock: () -> float. Returns: a class.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
class CircuitOpenError(Exception):
pass
def __harness__(cls, threshold, timeout, events):
now = {"t": 0.0}
breaker = cls(threshold, timeout, lambda: now["t"])
out = []
for t, result in events:
now["t"] = t
def op():
if result == "fail":
raise ConnectionError("downstream failed")
return "ok"
try:
outcome = breaker.call(op)
except CircuitOpenError:
outcome = "rejected"
except ConnectionError:
outcome = "fail"
out.append([outcome, breaker.state])
return out
class CircuitBreaker:
def __init__(self, failure_threshold: int, reset_timeout: float, clock):
self.state = "closed"
def call(self, fn):
return fn()
Run executes the public tests. Submit grades against public and hidden tests.