Event-Driven Circuit Breaker — Resilience Patterns for Event Sourcing and CQRS

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

Event-driven circuit breaker patterns protect event streams, event stores, projections, and sagas from failures by detecting event processing errors, switching to event store fallback, rebuilding projections from event history, and compensating failed sagas with circuit breaker integration.

What You'll Learn

• Event stream failure protection
• Event store fallback patterns
• Projection rebuilding on circuit recovery
• Outbox pattern with circuit breaker
• Saga compensation with circuit breakers

Why It Matters

Event-driven systems fail differently: event stream outages stop all event processing, event store failures prevent new events, and failed projections create data inconsistencies. Circuit breakers detect these failures and activate compensating actions.

Real-World Use

DodaTech's event-driven order system uses circuit breakers for the event stream. When the event bus fails, events are buffered in the outbox. The circuit breaker opens for 30 seconds. On recovery, buffered events replay and projections rebuild from the event store.

Event Stream Protection

import time
import json

class EventStreamCircuitBreaker:
    def __init__(self, fail_threshold=5, recovery_timeout=30):
        self.fail_threshold = fail_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = 'CLOSED'
        self.last_failure = 0

    def publish_event(self, event, publish_fn, fallback_fn):
        if self.state == 'OPEN':
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = 'HALF_OPEN'
                print("[Event CB] Half-open probe")
            else:
                print("[Event CB] Open, using fallback")
                return fallback_fn(event)

        try:
            publish_fn(event)
            self.failures = 0
            if self.state == 'HALF_OPEN':
                self.state = 'CLOSED'
                print("[Event CB] Recovered")
            return True
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.fail_threshold:
                self.state = 'OPEN'
                print(f"[Event CB] Open ({self.failures} failures)")
            return fallback_fn(event)

def publish_to_bus(event):
    print(f"Publishing: {event['type']}")
    raise ConnectionError("Event bus unavailable")

def buffer_to_outbox(event):
    print(f"Buffered to outbox: {event['type']}")
    return True

cb = EventStreamCircuitBreaker(fail_threshold=3, recovery_timeout=10)

events = [
    {"type": "OrderCreated", "id": 1},
    {"type": "OrderPaid", "id": 2},
    {"type": "OrderShipped", "id": 3},
]

for event in events:
    cb.publish_event(event, publish_to_bus, buffer_to_outbox)
    time.sleep(0.1)

Expected output:

Publishing: OrderCreated
Publishing: OrderPaid
Publishing: OrderShipped
[Event CB] Open (3 failures)
Buffered to outbox: OrderCreated
Buffered to outbox: OrderPaid
Buffered to outbox: OrderShipped

Outbox Pattern Protection

import time
import sqlite3

class OutboxCircuitBreaker:
    def __init__(self, db_path, fail_threshold=5, recovery_timeout=30):
        self.db_path = db_path
        self.fail_threshold = fail_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = 'CLOSED'
        self.last_failure = 0

    def flush_outbox(self, publish_fn):
        if self.state == 'OPEN':
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = 'HALF_OPEN'
            else:
                print("[Outbox] Circuit open, keeping events")
                return

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT id, event_type, payload FROM outbox WHERE published = 0")
        events = cursor.fetchall()

        failures = 0
        for event_id, event_type, payload in events:
            try:
                publish_fn(event_type, payload)
                cursor.execute("UPDATE outbox SET published = 1 WHERE id = ?", (event_id,))
                print(f"[Outbox] Published: {event_type}")
                self.failures = 0
            except Exception:
                failures += 1
                if failures >= self.fail_threshold:
                    self.state = 'OPEN'
                    self.last_failure = time.time()
                    print(f"[Outbox] Circuit open after {failures} failures")
                    break

        conn.commit()
        conn.close()

conn = sqlite3.connect(':memory:')
conn.execute("CREATE TABLE outbox (id INTEGER, event_type TEXT, payload TEXT, published INTEGER)")
for i in range(5):
    conn.execute("INSERT INTO outbox VALUES (?, ?, ?, 0)", (i+1, f"Event_{i+1}", f"data_{i+1}"))
conn.commit()
conn.close()

cb = OutboxCircuitBreaker(':memory:')
def fail_publish(event_type, payload):
    raise ConnectionError("Cannot publish")

cb.flush_outbox(fail_publish)

Expected output:

[Outbox] Circuit open after 5 failures
[Outbox] Keeping events

Common Mistakes

• Losing events during circuit open -- when the circuit is open, events must be buffered. Use the outbox pattern (DB table) for durable storage. Redis lists or local memory buffers are lost on restart.
• No event replay on recovery -- when the circuit closes, buffered events must replay. Track the last successfully published event and resume from that point. Include event ordering guarantees.
• Projection lag during circuit open -- projections fall behind when event processing stops. On recovery, rebuild projections from the event store (snapshot + replay). Monitor projection lag and alert if it exceeds thresholds.
• Sagas without circuit breaker compensation -- sagas running during circuit open may partially complete. Implement compensating transactions that fire when the circuit opens and rollback incomplete saga steps.
• Single circuit breaker for all event types -- use separate breakers per event type or per aggregate. An OrderCreated failure should not block UserRegistered events.

Practice Questions

1. How does the outbox pattern protect events during circuit open?
2. How do you replay events after circuit recovery?
3. What happens to projections during circuit open?
4. How do sagas handle circuit breaker state changes?
5. Why should different event types have separate circuit breakers?

Challenge

Build a resilient event-driven system: (1) circuit breaker for the event bus with outbox fallback, (2) outbox table stores events durably, (3) outbox flusher publishes events with a circuit breaker per event type, (4) projection manager that detects lag and rebuilds projections from event store after circuit recovery, (5) saga orchestrator with circuit breaker integration: if the payment service circuit is open, the saga suspends and schedules compensating transactions, (6) Prometheus metrics: event bus state, outbox depth, projection lag, saga status.

FAQ

Mini Project

Build an event-driven resilience platform: (1) circuit breakers for event bus publishing with per-event-type thresholds, (2) outbox pattern: durable event storage with batch flushing, (3) circuit-aware publication that routes events based on breaker state (bus on closed, outbox on open), (4) projection rebuild mechanism that replays events from event store after circuit recovery, (5) saga orchestrator suspending on circuit open and resuming on close, (6) Prometheus metrics: event publish count, outbox depth, projection lag, circuit state per event type, (7) event ordering guarantee: per-aggregate sequential processing within circuit breaker constraints.

What's Next

Continue with Chaos Testing to learn Chaos Engineering with circuit breakers. Then explore Self-Healing Systems for autonomous recovery patterns.