Async Circuit Breaker — Non-Blocking Resilience Patterns for Asynchronous Systems

In this tutorial, you will learn about Async Circuit Breaker. We cover key concepts, practical examples, and best practices to help you master this topic.

Async circuit breaker patterns handle non-blocking state management for asyncio, reactive streams, and event-driven architectures, using coroutine-safe failure counting, zero-blocking state checks, and cooperative yielding for half-open probes.

flowchart LR
    A[Async Request] --> CB{Async CB}
    CB -->|Closed| AsyncCall[await service.call()]
    AsyncCall -->|OK| Return[Return Result]
    AsyncCall -->|Exception| Count[async increment failure]
    CB -->|Open| Fast[Fast-fail/fallback]
    CB -->|Half-Open| Probe[await probe call]
    Count -->|Threshold| OpenState[Open State]

What You'll Learn

• Asyncio-compatible circuit breakers
• Coroutine-safe state transitions
• Async fallback execution
• Reactive stream integration
• Non-blocking failure detection

Why It Matters

Synchronous circuit breakers block the event loop during state checks. Async-native circuit breakers maintain non-blocking behavior, support awaitable fallbacks, and integrate with asyncio, Trio, and reactive stream libraries without blocking event loop progress.

Real-World Use

DodaTech's async Python API uses asyncio-native circuit breakers for all downstream HTTP calls. When the catalog service slows, the async breaker yields control to the event loop while tracking failures, keeping the API responsive for other requests.

Async Custom Circuit Breaker

import asyncio
import time

class AsyncCircuitBreaker:
    def __init__(self, name, fail_max=5, reset_timeout=30):
        self.name = name
        self.fail_max = fail_max
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.state = 'CLOSED'
        self.last_failure = 0
        self._lock = asyncio.Lock()

    async def call(self, fn, fallback=None, *args, **kwargs):
        async with self._lock:
            if self.state == 'OPEN':
                if time.time() - self.last_failure > self.reset_timeout:
                    self.state = 'HALF_OPEN'
                    print(f"[{self.name}] Half-open (async)")
                else:
                    return await self._fallback(fallback)

        try:
            result = await fn(*args, **kwargs)
            async with self._lock:
                self.failures = 0
                if self.state == 'HALF_OPEN':
                    self.state = 'CLOSED'
                    print(f"[{self.name}] Closed after recovery")
            return result
        except Exception as e:
            async with self._lock:
                self.failures += 1
                self.last_failure = time.time()
                if self.failures >= self.fail_max:
                    self.state = 'OPEN'
                    print(f"[{self.name}] Open ({self.failures} failures)")
            return await self._fallback(fallback)

    async def _fallback(self, fallback):
        if fallback:
            if asyncio.iscoroutinefunction(fallback):
                return await fallback()
            return fallback()
        raise Exception(f"Circuit '{self.name}' is open")

async def fetch_user(user_id):
    await asyncio.sleep(0.1)
    raise ConnectionError(f"Service unavailable for user {user_id}")

async def cached_user(user_id):
    return {"id": user_id, "name": "Cached User"}

async def main():
    cb = AsyncCircuitBreaker("user-service", fail_max=3, reset_timeout=5)

    for i in range(8):
        try:
            result = await cb.call(
                lambda: fetch_user(i),
                fallback=lambda: cached_user(i)
            )
            print(f"Call {i+1}: {result}")
        except Exception as e:
            print(f"Call {i+1}: {e}")
        await asyncio.sleep(0.1)

asyncio.run(main())

Expected output:

Call 1: {'id': 0, 'name': 'Cached User'}
Call 2: {'id': 1, 'name': 'Cached User'}
Call 3: {'id': 2, 'name': 'Cached User'}
[user-service] Open (3 failures)
Call 4: {'id': 3, 'name': 'Cached User'}
...
Call 6: (after timeout) {'id': 5, 'name': 'Cached User'}

Async with aiohttp

import asyncio
import aiohttp
import pybreaker

class AsyncCircuitBreakerHTTP:
    def __init__(self, breaker, session):
        self.breaker = breaker
        self.session = session

    async def get(self, url, fallback_value=None):
        if self.breaker.current_state == pybreaker.STATE_OPEN:
            if fallback_value is not None:
                return fallback_value
            raise pybreaker.CircuitBreakerError(self.breaker.name)

        try:
            async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                if resp.status >= 500:
                    self.breaker._failure()
                    return await self._handle_error(url, fallback_value)
                data = await resp.json()
                self.breaker._success()
                return data
        except (asyncio.TimeoutError, aiohttp.ClientError) as e:
            self.breaker._failure()
            return await self._handle_error(url, fallback_value)

    async def _handle_error(self, url, fallback_value):
        if fallback_value is not None:
            if asyncio.iscoroutinefunction(fallback_value):
                return await fallback_value()
            return fallback_value
        return {"error": f"Failed to fetch {url}"}

async def main():
    breaker = pybreaker.CircuitBreaker(fail_max=3, reset_timeout=30)
    async with aiohttp.ClientSession() as session:
        cb = AsyncCircuitBreakerHTTP(breaker, session)
        result = await cb.get(
            "https://api.example.com/products/1",
            fallback_value={"id": 1, "name": "Cached Product"}
        )
        print(f"Result: {result}")

asyncio.run(main())

Expected output:

Result: {'id': 1, 'name': 'Cached Product'}

Common Mistakes

• Blocking state checks in async code -- using threading.Lock in async code blocks the event loop. Always use asyncio.Lock for async circuit breaker state management.
• No yield point in half-open probes -- half-open probes should use await to yield control to the event loop. Synchronous half-open probes defeat the purpose of async circuit breakers.
• Fallback function is synchronous but fallbacks often need I/O. Support both sync and async fallbacks. Detect coroutine functions with asyncio.iscoroutinefunction() or inspect.iscoroutinefunction().
• Shared mutable state across coroutines -- multiple coroutines modifying failures and state simultaneously cause race conditions. Always protect state with asyncio.Lock or use atomic operations.
• Not handling asyncio.CancelledError -- if a task is cancelled during circuit breaker execution, the failure counter should not increment. Catch CancelledError separately and re-raise without counting as failure.

Practice Questions

1. Why should async circuit breakers use asyncio.Lock instead of threading.Lock?
2. How do you handle fallbacks that are also async functions?
3. What happens to the event loop when a circuit breaker blocks?
4. How do you handle asyncio.CancelledError in circuit breaker code?
5. What is the advantage of async-native circuit breakers over wrapping sync ones?

Challenge

Build a complete async circuit breaker library: (1) asyncio-native circuit breaker with async lock and non-blocking state transitions, (2) support for both sync and async fallback functions, (3) integration with aiohttp (GET/POST with circuit breaker per endpoint), (4) integration with AsyncSQL databases (asyncpg), (5) Prometheus metrics that use async collectors, (6) context manager interface: async with breaker:, (7) Redis-backed distributed state using aioredis for multi-worker deployments, (8) event emitter that fires async callbacks on state transitions.

FAQ

Mini Project

Build an async-native resilience framework: (1) AsyncCircuitBreaker with asyncio.Lock and coroutine fallbacks, (2) AsyncRetry with exponential backoff using asyncio.sleep, (3) AsyncBulkhead with asyncio.Semaphore, (4) AsyncRateLimiter with token bucket using asyncio, (5) all primitives support both sync and async wrapped calls, (6) Prometheus async metrics collector, (7) FastAPI middleware that applies all primitives to endpoints, (8) distributed state with aioredis for multi-instance deployments.

What's Next

Continue with Reactive Streams to learn reactive circuit breaker patterns. Then explore Database Connection for database circuit breaker patterns.