Reliability Testing for Distributed Systems: Retries, Timeouts, and Circuit Breakers
Distributed systems fail in ways that single-process systems don't. A network timeout doesn't crash your process — it makes your thread hang. A slow dependency doesn't just slow one request — it can exhaust your connection pool and bring down the whole service.
Retries, timeouts, and circuit breakers are the reliability patterns that prevent localized failures from becoming system-wide outages. But these patterns are easy to implement incorrectly, and incorrect implementations make failures worse. Testing them is non-negotiable.
Testing Retry Strategies
Retries are the most commonly misimplemented reliability pattern. The failure modes:
- No retries: Transient failures cause user-visible errors that wouldn't have if you just tried again
- Retry storms: Infinite retries under sustained failure flood the degraded service
- Non-idempotent retries: Retrying a payment creates duplicate charges
- No jitter: All clients retry at the same time, creating synchronized traffic spikes
Test all of these:
class RetryStrategyTests:
def test_retries_on_transient_failure(self):
"""Client must retry on transient errors (5xx, connection refused)."""
call_count = 0
def flaky_handler():
nonlocal call_count
call_count += 1
if call_count < 3:
raise ServiceUnavailableError("Transient failure")
return {"status": "success"}
with mock_service(handler=flaky_handler):
result = client.call_service()
assert result["status"] == "success"
assert call_count == 3, f"Expected 3 calls (2 retries), got {call_count}"
def test_does_not_retry_on_client_errors(self):
"""Client must NOT retry on 4xx errors — they indicate client bugs."""
call_count = 0
def always_400():
nonlocal call_count
call_count += 1
raise BadRequestError("Invalid input")
with mock_service(handler=always_400):
with pytest.raises(BadRequestError):
client.call_service(payload={"invalid": "data"})
assert call_count == 1, \
f"Client retried a 400 error {call_count} times — should not retry client errors"
def test_exponential_backoff_delays(self):
"""Retry delays must increase exponentially between attempts."""
attempt_times = []
def fail_and_record():
attempt_times.append(time.monotonic())
raise ServiceUnavailableError("Always fails")
with mock_service(handler=fail_and_record):
with pytest.raises(MaxRetriesExceededError):
client.call_service(max_retries=4)
delays = [attempt_times[i+1] - attempt_times[i] for i in range(len(attempt_times)-1)]
# Each delay should be roughly 2x the previous
for i in range(1, len(delays)):
ratio = delays[i] / delays[i-1]
assert 1.5 <= ratio <= 3.0, \
f"Delay ratio {ratio:.2f} not consistent with exponential backoff"
def test_jitter_prevents_thundering_herd(self):
"""Retry delays must include jitter to prevent synchronized retries."""
delay_samples = []
# Simulate 100 clients all retrying at the same time
for _ in range(100):
retry_delay = client._calculate_retry_delay(attempt=1)
delay_samples.append(retry_delay)
# If there's no jitter, all delays are identical
unique_delays = set(f"{d:.3f}" for d in delay_samples)
assert len(unique_delays) > 10, \
f"Too few unique retry delays ({len(unique_delays)}) — insufficient jitter"
def test_retry_budget_prevents_storms(self):
"""System must stop retrying when retry budget is exhausted."""
client_with_budget = RetryingClient(
max_retries=3,
retry_budget_per_second=10 # Max 10 retries/second across all callers
)
# Simulate 50 concurrent requests all failing
results = parallel_calls(
client=client_with_budget,
count=50,
mock_response=ServiceUnavailableError("All failing")
)
# Total retries should be capped by budget
total_retries = sum(r.retry_count for r in results)
assert total_retries < 200, \
f"Retry storm not prevented: {total_retries} retries for 50 requests"Testing Timeout Behavior
Timeouts are what prevent a slow dependency from blocking your entire application. Test that they're set correctly and actually trigger.
class TimeoutTests:
def test_request_times_out_when_service_hangs(self):
"""Request must fail fast when service hangs — not wait forever."""
REQUEST_TIMEOUT_MS = 500
def hanging_handler():
time.sleep(10) # Simulates hung service
return {"status": "ok"}
with mock_service(handler=hanging_handler):
start = time.monotonic()
with pytest.raises(TimeoutError):
client.call_service(timeout_ms=REQUEST_TIMEOUT_MS)
elapsed_ms = (time.monotonic() - start) * 1000
# Should have failed close to the timeout, not 10 seconds later
assert elapsed_ms < REQUEST_TIMEOUT_MS * 2, \
f"Timeout took {elapsed_ms:.0f}ms, expected ~{REQUEST_TIMEOUT_MS}ms"
def test_timeout_cascade_prevention(self):
"""Downstream timeout budget must be less than upstream timeout."""
# If service A times out in 5s, service B (its dependency)
# must time out in <5s to allow error propagation
upstream_timeout_ms = 5000
downstream_timeout_ms = client.get_dependency_timeout_ms("database")
# Leave at least 500ms margin for error handling
assert downstream_timeout_ms < upstream_timeout_ms - 500, (
f"Timeout cascade risk: downstream timeout ({downstream_timeout_ms}ms) "
f"is too close to upstream timeout ({upstream_timeout_ms}ms). "
"Callers will timeout before they receive the error."
)
def test_connection_pool_not_exhausted_by_slow_service(self):
"""Slow downstream must not exhaust the connection pool."""
POOL_SIZE = 10
TIMEOUT_MS = 200
def slow_handler():
time.sleep(1) # Much slower than timeout
return {"status": "ok"}
with mock_service(handler=slow_handler):
# Send 50 concurrent requests — all will timeout
with ThreadPoolExecutor(max_workers=50) as executor:
futures = [
executor.submit(client.call_service, timeout_ms=TIMEOUT_MS)
for _ in range(50)
]
results = [f.result(timeout=5) for f in futures]
# All should have timed out (not hung)
timeouts = sum(1 for r in results if r.error_type == "timeout")
assert timeouts == 50, f"Expected 50 timeouts, got {timeouts}"
# After timeouts, connections should be released back to pool
pool_available = get_connection_pool_available("downstream")
assert pool_available >= POOL_SIZE * 0.8, \
f"Connection pool exhausted: {pool_available}/{POOL_SIZE} available"Testing Circuit Breakers
Circuit breakers have three states: Closed (normal), Open (failing fast), and Half-Open (testing recovery). Test all three and the transitions between them.
class CircuitBreakerTests:
def test_circuit_opens_after_failure_threshold(self):
"""Circuit must open after failure rate exceeds threshold."""
breaker = CircuitBreaker(
failure_threshold=5,
window_seconds=10
)
# Fail 5 times — should open the circuit
for _ in range(5):
with pytest.raises(ServiceUnavailableError):
with breaker:
raise ServiceUnavailableError("Simulated failure")
assert breaker.state == CircuitState.OPEN, \
f"Circuit should be OPEN after 5 failures, got {breaker.state}"
def test_open_circuit_fast_fails(self):
"""Open circuit must reject requests immediately without calling service."""
breaker = CircuitBreaker(failure_threshold=3)
# Open the circuit
for _ in range(3):
try:
with breaker:
raise ServiceUnavailableError()
except:
pass
# Now circuit is open — call should fast-fail
call_count = 0
def count_calls():
nonlocal call_count
call_count += 1
return "success"
start = time.monotonic()
with pytest.raises(CircuitOpenError):
with breaker:
count_calls()
elapsed = (time.monotonic() - start) * 1000
assert call_count == 0, "Open circuit should not have called the service"
assert elapsed < 50, f"Open circuit took {elapsed:.0f}ms to reject — should be instant"
def test_half_open_allows_single_probe(self):
"""After timeout, circuit transitions to half-open and allows one probe."""
breaker = CircuitBreaker(
failure_threshold=3,
recovery_timeout_seconds=1
)
# Open the circuit
for _ in range(3):
try:
with breaker:
raise ServiceUnavailableError()
except:
pass
assert breaker.state == CircuitState.OPEN
# Wait for recovery timeout
time.sleep(1.1)
# Circuit should try one probe request
assert breaker.state == CircuitState.HALF_OPEN
# Successful probe closes the circuit
with breaker:
pass # Success
assert breaker.state == CircuitState.CLOSED
def test_half_open_reopens_on_failure(self):
"""Failed probe in half-open state must reopen the circuit."""
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout_seconds=1)
# Open the circuit
for _ in range(3):
try:
with breaker:
raise ServiceUnavailableError()
except:
pass
time.sleep(1.1)
assert breaker.state == CircuitState.HALF_OPEN
# Probe fails — circuit reopens
with pytest.raises(ServiceUnavailableError):
with breaker:
raise ServiceUnavailableError("Still failing")
assert breaker.state == CircuitState.OPEN, \
"Failed probe should reopen the circuit"Testing Bulkhead Isolation
Bulkheads limit how much one service can consume of another service's resources. Test that they actually isolate failures.
def test_bulkhead_isolates_slow_service():
"""Slow service must not exhaust thread pool for healthy services."""
# Bulkhead: max 5 concurrent requests to slow_service
bulkhead = Bulkhead(max_concurrent=5)
# Saturate the bulkhead with slow requests
slow_requests = []
for _ in range(5):
future = executor.submit(
make_bulkheaded_request,
bulkhead=bulkhead,
handler=lambda: time.sleep(10) # Very slow
)
slow_requests.append(future)
time.sleep(0.1) # Let requests start
# Bulkhead should be full — next request should fail fast
start = time.monotonic()
with pytest.raises(BulkheadFullError):
make_bulkheaded_request(bulkhead=bulkhead, handler=lambda: "fast")
elapsed = (time.monotonic() - start) * 1000
assert elapsed < 50, f"Bulkhead rejection took {elapsed:.0f}ms — should be instant"
# Fast service (different bulkhead) should still work
fast_result = make_bulkheaded_request(
bulkhead=fast_service_bulkhead, # Different bulkhead
handler=lambda: "fast response"
)
assert fast_result == "fast response", \
"Fast service affected by slow service bulkhead exhaustion"End-to-End Resilience Testing
Individual pattern tests validate mechanics. End-to-end tests validate that the patterns work together under realistic conditions.
def test_service_resilience_under_dependency_failure():
"""Service must maintain SLO when primary dependency fails completely."""
SLO_AVAILABILITY = 0.99
# Start load test
load = LoadGenerator(rps=100, duration_seconds=120)
# After 30s, kill the primary dependency
def inject_failure():
time.sleep(30)
kill_service("primary-database")
# After 60s, restore it
def restore():
time.sleep(60)
start_service("primary-database")
threading.Thread(target=inject_failure).start()
threading.Thread(target=restore).start()
results = load.run()
# Measure availability across the full 120s window
overall_availability = results.successful / results.total
# Even with 30 seconds of database downtime, SLO must hold
# via fallback cache, replica reads, or graceful degradation
assert overall_availability >= SLO_AVAILABILITY, (
f"Service dropped below SLO during dependency failure: "
f"{overall_availability:.3f} < {SLO_AVAILABILITY}"
)Continuous Reliability Monitoring
HelpMeTest can continuously validate that your service responds within timeout budgets and returns successful responses — giving you ongoing visibility into the reliability patterns you've implemented. Set up health checks that:
- Verify response times stay under your configured timeouts
- Alert when error rates suggest a circuit breaker might be open
- Monitor the health endpoints of your dependencies
When your circuit breaker logic is working correctly, you'll see momentary error spikes followed by fast recovery. When it's not, you'll see sustained degradation. Continuous monitoring shows you which.
Summary
Reliability pattern testing requires:
- Retry tests — transient failure recovery, backoff correctness, no storm amplification
- Timeout tests — timeouts actually trigger, cascade prevention, pool recovery
- Circuit breaker tests — all three states and correct transitions between them
- Bulkhead tests — isolation actually works, fast services unaffected by slow ones
- End-to-end resilience tests — all patterns working together under realistic failure scenarios
Reliability patterns are only as good as your confidence in them. Test them until you'd bet an outage on them — because you will.
