# Resilient Messaging Pipelines (/integrations/faststream/scenarios/resilient-pipelines)
## Scenario [#scenario]
A production order system has to do more than move messages. Each order flows through several steps — reserve inventory, charge payment, ship — and any step can fail. The system must compensate for partial failures, isolate downstream outages, never double-charge a customer on a redelivered message, and reconstruct account state from history when something goes wrong.
KubeMQ exposes five messaging primitives — **Events**, **Events Store**, **Queues**, **Commands**, and **Queries** — and `kubemq-faststream` maps each to a `@broker.subscriber` keyword. The resilience patterns on this page are built entirely from those primitives plus queue parameters the broker already enforces (`max_receive_count`, `max_receive_queue`, `delay_in_seconds`, `max_messages`). No extra infrastructure is required.
This page assembles the production patterns from the `examples/advanced_patterns/` and `examples/patterns/` directories of the [`kubemq-faststream`](https://github.com/kubemq-io/kubemq-faststream) repository into one resilience playbook. Every snippet is runnable against a single broker.
`kubemq-faststream` talks to KubeMQ over the **native gRPC port `50000`** — the same transport the native SDKs use. There is no connector to enable and no HTTP flag to set. See [Getting Started](/integrations/faststream/tutorials/getting-started) for the full setup.
Every example connects the same way and reads the broker address from `KUBEMQ_ADDRESS`, so you can point any of them at a remote broker without editing code:
```python title="common.py"
import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker
KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)
```
Run a broker locally first:
Port `50000` is the gRPC port FastStream connects to. Port `9090` is the shared HTTP server (REST and connector endpoints) — not required here, but harmless to expose.
## Saga Orchestration [#saga-orchestration]
A saga executes a sequence of steps and, if any step fails, runs **compensating actions** in reverse to undo the work already done. Here each step is a KubeMQ command (`broker.request` blocks until the handler replies), and each completed step records the channel that reverses it.
```python title="saga_pattern.py"
@broker.subscriber(commands="example.saga.reserve_inventory")
async def reserve_inventory(msg: dict) -> dict:
print("[saga] Step 1: reserve_inventory OK")
return {"status": "reserved", "order_id": msg["order_id"]}
@broker.subscriber(commands="example.saga.charge_payment")
async def charge_payment(msg: dict) -> dict:
print("[saga] Step 2: charge_payment OK")
return {"status": "charged", "order_id": msg["order_id"]}
@broker.subscriber(commands="example.saga.ship_order")
async def ship_order(msg: dict) -> dict:
print("[saga] Step 3: ship_order OK")
return {"status": "shipped", "order_id": msg["order_id"]}
@broker.subscriber(commands="example.saga.compensate.inventory")
async def compensate_inventory(msg: dict) -> None:
print(f"[compensate] Released inventory for order {msg['order_id']}")
@broker.subscriber(commands="example.saga.compensate.payment")
async def compensate_payment(msg: dict) -> None:
print(f"[compensate] Refunded payment for order {msg['order_id']}")
```
The orchestrator walks the step list. Each step pairs a forward channel with its compensation channel (the final step needs none). On failure, it replays the recorded compensations in reverse order:
```python title="saga_pattern.py"
SAGA_STEPS = [
("example.saga.reserve_inventory", "example.saga.compensate.inventory"),
("example.saga.charge_payment", "example.saga.compensate.payment"),
("example.saga.ship_order", None), # last step has no compensation
]
async def run_saga(order_id: str) -> bool:
"""Execute saga steps in order; compensate on failure."""
completed: list[str] = []
for step_channel, compensate_channel in SAGA_STEPS:
try:
await broker.request(
{"order_id": order_id},
commands=step_channel,
timeout=10,
)
if compensate_channel:
completed.append(compensate_channel)
except Exception as exc:
print(f"[saga] Step failed ({step_channel}): {exc}")
# Compensate in reverse order
for comp_channel in reversed(completed):
await broker.request(
{"order_id": order_id},
commands=comp_channel,
timeout=10,
)
return False
return True
```
Commands are request-reply with a void result. The orchestrator's `timeout=10` is per step — a step that never replies is treated as a failure and triggers compensation, so a hung downstream service cannot leave the saga half-finished.
## Dead-Letter Processing [#dead-letter-processing]
When a queue message keeps failing, you do not want it redelivered forever. KubeMQ enforces a per-message **`max_receive_count`**: after that many delivery attempts, the broker automatically routes the message to the **`max_receive_queue`** (the dead-letter queue). A separate subscriber drains the DLQ for investigation. The main handler uses `AckPolicy.NACK_ON_ERROR` so a raised exception nacks the message and counts as a failed attempt.
```python title="dead_letter_processing.py"
from kubemq_faststream import AckPolicy, KubeMQBroker
MAIN_QUEUE = "example.advanced.dlq.main"
DLQ_QUEUE = "example.advanced.dlq"
@broker.subscriber(queues=MAIN_QUEUE, ack_policy=AckPolicy.NACK_ON_ERROR)
async def handle_order(msg: dict) -> None:
"""Intentionally fail to trigger DLQ routing after max receive count."""
order_id = msg.get("order_id", "unknown")
print(f"[main] Processing order: {order_id} -- simulated failure")
raise ValueError(f"Cannot process order {order_id}")
@broker.subscriber(queues=DLQ_QUEUE)
async def handle_dead_letter(msg: dict) -> None:
"""Capture and log messages routed to the dead-letter queue."""
print(f"[dlq] Dead-letter received: {msg}")
print("[dlq] Logging failed order for manual review")
```
The DLQ routing lives on the **publish** call — `max_receive_count` and `max_receive_queue` are queue parameters, not subscriber settings:
```python title="dead_letter_processing.py"
await broker.publish(
{"order_id": "order-500", "item": "premium-widget", "amount": 99.99},
queues=MAIN_QUEUE,
max_receive_count=3,
max_receive_queue=DLQ_QUEUE,
)
```
After three failed deliveries the broker moves the message to `example.advanced.dlq`, and `handle_dead_letter` logs it:
```text
[main] Processing order: order-500 -- simulated failure
[main] Processing order: order-500 -- simulated failure
[main] Processing order: order-500 -- simulated failure
[dlq] Dead-letter received: {'order_id': 'order-500', ...}
[dlq] Logging failed order for manual review
```
For the full set of acknowledgement strategies and DLQ fundamentals, see the [Queues page](/integrations/faststream/how-to/queues).
## Retry with Exponential Backoff [#retry-with-exponential-backoff]
Transient failures — a brief network blip, a rate-limited dependency — deserve a retry, not a dead letter. This pattern re-publishes the failed message with an increasing **`delay_in_seconds`** so retries are spaced out (1s, 2s, 4s, …). A `max_receive_count` ceiling still guarantees the message lands in the DLQ if it never succeeds.
```python title="retry_with_backoff.py"
MAIN_QUEUE = "example.advanced.retry"
DLQ_QUEUE = "example.advanced.retry.dlq"
attempt_tracker: dict[str, int] = {}
@broker.subscriber(queues=MAIN_QUEUE)
async def handle_with_retry(msg: dict) -> None:
"""Process messages with simulated transient failures."""
task_id = msg.get("task_id", "unknown")
attempt_tracker[task_id] = attempt_tracker.get(task_id, 0) + 1
attempt = attempt_tracker[task_id]
if attempt < 3:
print(f"[retry] Attempt for {task_id}, simulating transient failure")
# Re-publish with exponential backoff delay
delay = 2 ** (attempt - 1) # 1s, 2s, 4s, ...
await broker.publish(
msg,
queues=MAIN_QUEUE,
max_receive_count=5,
max_receive_queue=DLQ_QUEUE,
delay_in_seconds=delay,
)
print(f"Published retry with delay={delay}s")
else:
print(f"[retry] Attempt for {task_id}, processing succeeded on retry")
@broker.subscriber(queues=DLQ_QUEUE)
async def handle_dead_letter(msg: dict) -> None:
"""Process messages that exceeded max receive count."""
print(f"[dlq] Dead-letter received: {msg}")
```
The `delay_in_seconds` parameter uses the broker's delayed-delivery policy: the message is published immediately but only becomes visible to the subscriber after the delay elapses. Combining application-level retry counting with the broker's `max_receive_count` gives you both backoff control and a hard ceiling.
## Circuit Breaker [#circuit-breaker]
A circuit breaker stops hammering a failing dependency. It tracks consecutive failures and, once a threshold is crossed, **opens** to reject work for a recovery window. After the window it goes **half-open** to test a single message: success closes the circuit, failure re-opens it. Here the breaker wraps a queue subscriber.
```python title="circuit_breaker.py"
import enum
import time
class State(enum.Enum):
CLOSED = "CLOSED"
OPEN = "OPEN"
HALF_OPEN = "HALF_OPEN"
class CircuitBreaker:
"""Simple circuit breaker with failure counting and timed recovery."""
def __init__(self, failure_threshold: int = 3, recovery_timeout: float = 5.0) -> None:
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.state = State.CLOSED
self.failure_count = 0
self.last_failure_time: float = 0.0
def allow_request(self) -> bool:
if self.state == State.CLOSED:
return True
if self.state == State.OPEN:
if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
self.state = State.HALF_OPEN
print("[cb] Circuit HALF_OPEN -- testing recovery")
return True
return False
# HALF_OPEN: allow one test request
return True
def record_success(self) -> None:
self.failure_count = 0
self.state = State.CLOSED
def record_failure(self) -> None:
self.failure_count += 1
self.last_failure_time = time.monotonic()
if self.failure_count >= self.failure_threshold:
self.state = State.OPEN
print("[cb] Circuit OPEN -- rejecting messages")
```
The subscriber consults the breaker before doing any work. While the circuit is open it returns early, leaving the dependency alone until the recovery timeout passes:
```python title="circuit_breaker.py"
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=3.0)
@broker.subscriber(queues="example.advanced.circuit")
async def handle(msg: dict) -> None:
if not cb.allow_request():
print(f"[cb] {cb.state.value}: rejecting msg")
return
try:
do_work(msg) # your real processing
cb.record_success()
except RuntimeError:
cb.record_failure()
```
## Idempotent Consumer [#idempotent-consumer]
At-least-once delivery means a subscriber can see the same message twice — after a redelivery, a retry, or a duplicate publish. An idempotent consumer makes reprocessing a no-op by tracking the `message_id` of every message it has already handled. The `message_id` is set explicitly on publish so the value is stable across deliveries.
```python title="idempotent_consumer.py"
CHANNEL = "example.advanced.idempotent"
# In-memory deduplication store (use Redis/DB in production)
processed_ids: set[str] = set()
@broker.subscriber(queues=CHANNEL)
async def idempotent_handler(msg: dict) -> None:
"""Process each unique message_id only once."""
msg_id = msg.get("message_id", "")
if not msg_id:
print("[idempotent] No message_id in payload, processing anyway")
return
if msg_id in processed_ids:
print(f"[idempotent] Duplicate skipped: {msg_id}")
return
processed_ids.add(msg_id)
print(f"[idempotent] Processing new message: {msg_id}")
```
Publishing the same payload twice with an identical `message_id` exercises the guard:
```python title="idempotent_consumer.py"
msg_id = "order-abc-001"
payload = {"message_id": msg_id, "order": "widget-x", "quantity": 5}
await broker.publish(payload, queues=CHANNEL, message_id=msg_id)
await broker.publish(payload, queues=CHANNEL, message_id=msg_id) # duplicate
```
```text
[idempotent] Processing new message: order-abc-001
[idempotent] Duplicate skipped: order-abc-001
```
The in-memory `set` resets when the process restarts. In production, back the dedup store with a durable, shared store (Redis, a database) and give entries a TTL so the set does not grow without bound.
## Event Sourcing [#event-sourcing]
Instead of storing current state, event sourcing stores the **sequence of events** that produced it, and rebuilds state by replaying them. KubeMQ's **Events Store** is the append-only log: publish domain events to an `events_store` channel, then subscribe with `start_position=StartPosition.START_FROM_FIRST` to replay the entire history from the beginning.
```python title="event_sourcing.py"
from kubemq_faststream import KubeMQBroker, StartPosition
CHANNEL = "example.advanced.eventsourcing"
# Reconstructed state
account_state = {"balance": 0, "event_count": 0}
@broker.subscriber(
events_store=CHANNEL,
start_position=StartPosition.START_FROM_FIRST,
group="event-sourcing-replay",
)
async def replay_events(msg: dict) -> None:
"""Replay events from the store to reconstruct account state."""
event_type = msg.get("type", "unknown")
amount = msg.get("amount", 0)
account_state["event_count"] += 1
count = account_state["event_count"]
if event_type == "account_opened":
account_state["balance"] = 0
elif event_type == "deposit":
account_state["balance"] += amount
elif event_type == "withdrawal":
account_state["balance"] -= amount
print(f"[replay] Event {count}: {event_type} -- balance={account_state['balance']}")
```
Each domain event is appended to the store with a normal `broker.publish`:
```python title="event_sourcing.py"
events = [
{"type": "account_opened", "account_id": "acct-100"},
{"type": "deposit", "account_id": "acct-100", "amount": 500},
{"type": "withdrawal", "account_id": "acct-100", "amount": 200},
{"type": "deposit", "account_id": "acct-100", "amount": 250},
]
for event in events:
await broker.publish(event, events_store=CHANNEL)
```
Because the subscriber starts from `START_FROM_FIRST`, a fresh process replays the full ledger and arrives at the same balance every time:
```text
[replay] Event 1: account_opened -- balance=0
[replay] Event 2: deposit -- balance=500
[replay] Event 3: withdrawal -- balance=300
[replay] Event 4: deposit -- balance=550
```
`START_FROM_FIRST` is one of several replay positions. To resume from a sequence number or a timestamp instead, see the [Events Store page](/integrations/faststream/how-to/events-store).
## Distribution and Ordering [#distribution-and-ordering]
The remaining patterns control *how* work spreads across consumers and *in what order* it is processed.
A coordinator distributes work chunks across N worker queues (fan-out), then aggregates the results from a shared results queue (fan-in), matching them by `correlation_id`. Each worker publishes its result tagged with the same correlation ID it received:
```python title="fan_out_fan_in.py"
RESULTS_QUEUE = "example.advanced.fanout.results"
results_store: dict[str, list[int]] = {}
expected_count = 3
@broker.subscriber(queues="example.advanced.fanout.worker1")
async def worker_1(msg: dict) -> None:
result = msg.get("value", 0) * 10
await broker.publish(
{"correlation_id": msg["correlation_id"], "result": result},
queues=RESULTS_QUEUE,
correlation_id=msg["correlation_id"],
)
@broker.subscriber(queues=RESULTS_QUEUE)
async def aggregator(msg: dict) -> None:
"""Fan-in: collect results and aggregate when all workers report."""
corr_id = msg.get("correlation_id", "unknown")
results_store.setdefault(corr_id, []).append(msg.get("result", 0))
if len(results_store[corr_id]) >= expected_count:
total = sum(results_store[corr_id])
print(f"[fanin] Aggregated total: {total}")
```
An event subscriber inspects each message and re-publishes it to a priority-specific queue. Independent subscribers drain the high, medium, and low queues at their own pace:
```python title="priority_routing.py"
PRIORITY_QUEUES = {
"high": "example.advanced.priority.high",
"medium": "example.advanced.priority.medium",
"low": "example.advanced.priority.low",
}
@broker.subscriber(events="example.advanced.priority.ingest")
async def route_by_priority(msg: dict) -> None:
"""Route incoming events to priority-specific queues."""
priority = msg.get("priority", "low").lower()
target_queue = PRIORITY_QUEUES.get(priority, PRIORITY_QUEUES["low"])
print(f"[router] Routing {msg.get('task_id')} to {priority.upper()} priority queue")
await broker.publish(msg, queues=target_queue)
```
To guarantee strict in-order handling, use a single consumer with **no group** and **`max_messages=1`** so the subscriber pulls and finishes one message before the next is delivered:
```python title="ordered_processing.py"
@broker.subscriber(queues="example.advanced.ordered", max_messages=1)
async def ordered_handler(msg: dict) -> None:
"""Process messages strictly one at a time in order."""
seq = msg.get("sequence", 0)
step = msg.get("step", "unknown")
print(f"[ordered] Processing sequence {seq}: {step}")
await asyncio.sleep(0.2) # processing time; order is still preserved
```
A `correlation_id` can be threaded through an entire pipeline that crosses pattern boundaries — event → queue → command — so every stage can be traced end to end. Each stage forwards the same ID to the next:
```python title="correlation_tracking.py"
@broker.subscriber(events="example.advanced.corr.events")
async def stage_1_event(msg: dict) -> None:
corr_id = msg.get("correlation_id", "unknown")
print(f"[stage-1] Event received, forwarding to queue (corr={corr_id})")
await broker.publish(
{"stage": 2, "correlation_id": corr_id, "data": msg.get("data")},
queues="example.advanced.corr.queue",
correlation_id=corr_id,
)
@broker.subscriber(queues="example.advanced.corr.queue")
async def stage_2_queue(msg: dict) -> None:
corr_id = msg.get("correlation_id", "unknown")
print(f"[stage-2] Queue received, forwarding to command (corr={corr_id})")
await broker.publish(
{"stage": 3, "correlation_id": corr_id, "data": msg.get("data")},
commands="example.advanced.corr.command",
correlation_id=corr_id,
)
@broker.subscriber(commands="example.advanced.corr.command")
async def stage_3_command(msg: dict) -> None:
corr_id = msg.get("correlation_id", "unknown")
print(f"[tracking] Correlation chain complete for {corr_id}")
```
## Pipeline Composition [#pipeline-composition]
The foundational patterns in `examples/patterns/` show how the primitives compose into multi-stage workflows. A **pipeline chain** moves a message through patterns by stage: an event triggers reliable queue processing, which fires a command for final execution.
```python title="pipeline_chain.py"
STAGE_1 = "example.patterns.pipeline.ingest"
STAGE_2 = "example.patterns.pipeline.process"
STAGE_3 = "example.patterns.pipeline.execute"
@broker.subscriber(events=STAGE_1)
async def stage_ingest(msg: dict) -> None:
"""Stage 1: Receive event and forward to queue for reliable processing."""
print(f"[Stage 1 - Event] Ingested: {msg}")
enriched = {**msg, "ingested": True}
await broker.publish(enriched, queues=STAGE_2)
@broker.subscriber(queues=STAGE_2)
async def stage_process(msg: dict) -> None:
"""Stage 2: Process from queue and trigger a command."""
print(f"[Stage 2 - Queue] Processed: {msg}")
await broker.request({**msg, "processed": True}, commands=STAGE_3, timeout=10)
@broker.subscriber(commands=STAGE_3)
async def stage_execute(msg: dict) -> None:
"""Stage 3: Execute the final command."""
print(f"[Stage 3 - Command] Executed: {msg}")
```
To scale a single stage horizontally, register several subscribers on the same queue channel with the same `group`. KubeMQ delivers each message to exactly one member of the group — **competing consumers** load-balance the work:
```python title="competing_consumers.py"
CHANNEL = "example.patterns.competing.tasks"
@broker.subscriber(queues=CHANNEL, group="workers")
async def worker_1(msg: dict) -> None:
print(f"[worker-1] Processing task {msg.get('task_id', '?')}")
@broker.subscriber(queues=CHANNEL, group="workers")
async def worker_2(msg: dict) -> None:
print(f"[worker-2] Processing task {msg.get('task_id', '?')}")
@broker.subscriber(queues=CHANNEL, group="workers")
async def worker_3(msg: dict) -> None:
print(f"[worker-3] Processing task {msg.get('task_id', '?')}")
```
The `examples/patterns/multi_pattern_app.py` example wires all five primitives into one cohesive workflow — an event forwards to a queue, the queue fires a command, an Events Store subscriber persists the stream, and a query reports status — demonstrating that these patterns are not isolated demos but composable building blocks.
## Running the Examples [#running-the-examples]
### Start a broker [#start-a-broker]
### Install the package [#install-the-package]
```bash
pip install kubemq-faststream
```
### Run any pattern [#run-any-pattern]
Each file is self-contained and exits on its own after the demo completes:
```bash
python examples/advanced_patterns/saga_pattern.py
python examples/advanced_patterns/dead_letter_processing.py
python examples/advanced_patterns/circuit_breaker.py
python examples/advanced_patterns/event_sourcing.py
```
Point any example at a remote broker without touching code:
```bash
KUBEMQ_ADDRESS=kubemq://my-broker:50000 python examples/advanced_patterns/saga_pattern.py
```
## Related [#related]