Picture this: your e-commerce data pipeline successfully writes a completed order to your PostgreSQL operational database, updates the inventory count in DynamoDB, and then fails spectacularly while attempting to publish the fulfillment event to Kafka. Now your order exists in two systems but the warehouse management software never received the trigger to pick the product off the shelf. Your customer gets a confirmation email, but their package never ships. The data is inconsistent across your systems, and the only way you know something went wrong is when customer support starts getting angry calls three days later.
This is the central problem of multi-sink data pipelines: you're writing to multiple heterogeneous systems, and no single transaction coordinator speaks all their languages simultaneously. Traditional ACID transactions work beautifully within a single database because the system owns the log, the locks, and the commit protocol. The moment you step across system boundaries — Postgres to Kafka to S3 to an external API — you've exited the safety of atomic guarantees. You're now in distributed systems territory, where partial failures aren't edge cases; they're scheduled events.
By the end of this lesson, you'll understand how to architect pipelines that handle partial failures gracefully using the Saga pattern, implement compensating transactions that undo work already committed to upstream systems, and build the state machine machinery that makes all of this observable and debuggable in production. We'll work through a realistic order processing pipeline in Python that writes to PostgreSQL, publishes to Kafka, and updates a Redis cache — and we'll make it resilient to failures at every step.
What you'll learn:
You should be comfortable with:
asyncio, async/await)You don't need prior experience with distributed systems theory, though a passing familiarity with CAP theorem won't hurt.
Before building the solution, you need to internalize why the obvious solution doesn't work. When engineers first encounter this problem, the instinct is usually: "Can't we just use a distributed transaction? XA protocol? Two-phase commit?"
Two-phase commit (2PC) works by electing a coordinator that first sends a "prepare" message to all participants, waits for acknowledgment that each participant has durably staged the write, and then sends a global commit or abort. It genuinely provides atomicity across systems — in theory.
In practice, 2PC has several catastrophic failure modes. First, it requires all participants to implement the XA interface, and almost nothing in a modern data stack does. Kafka doesn't support XA. Most REST APIs definitely don't. DynamoDB doesn't. Redis doesn't. You'd need every system in your pipeline to implement a two-phase protocol with durable prepare logging, which essentially means reimplementing a distributed database inside every system you touch.
Second, 2PC is a blocking protocol. If the coordinator crashes after sending "prepare" but before sending "commit," every participant holds locks indefinitely waiting for a commit signal that may never come. Your PostgreSQL tables are locked, your Kafka producer is stalled, and your pipeline is dead until a human intervenes or the coordinator recovers with its state intact. This is called the "coordinator failure problem," and it transforms a transient hardware failure into a system-wide outage.
Third, even when 2PC works correctly, it introduces significant latency because every write must complete two round-trips plus synchronous durability guarantees from all participants simultaneously. For a pipeline writing to five systems, you're waiting for the slowest system on every single transaction.
The alternative is to accept that perfect atomicity across heterogeneous systems isn't achievable, and instead design for eventual consistency with explicit compensation. This is the insight behind the Saga pattern.
A Saga is a sequence of local transactions where each step has a corresponding compensating transaction. If step N fails, the saga executes the compensating transactions for steps N-1, N-2, ..., 1 in reverse order, returning the system to a consistent state.
The critical insight is that compensating transactions are not rollbacks in the traditional database sense. They are new, forward-moving transactions that logically undo the effect of an earlier step. If you debited a customer's account as step 2 and need to "undo" that, you don't reach back into history and delete the debit row — you credit the account in a new transaction. The ledger now has both entries, and the net effect is zero. This is crucial because the debit may have already been read by other processes, reported in dashboards, or replicated to read replicas. You can't pretend it didn't happen.
There are two main coordination strategies for Sagas:
Choreography-based Sagas have no central coordinator. Each service listens for events, performs its local transaction, and publishes the next event. Compensation happens by publishing "failure" events that each service handles by running its own compensating logic. This works well for loosely coupled microservices but becomes very hard to reason about as the number of steps grows — the "saga" exists only implicitly in the choreography of events, making it extremely difficult to debug or audit.
Orchestration-based Sagas have a central orchestrator (your pipeline code) that explicitly calls each participant, tracks state, and drives compensation when needed. The entire saga's state is visible in one place. For data pipelines, orchestration is almost always the right choice because you control all the participants and observability is paramount.
We'll implement an orchestration-based saga.
A saga is fundamentally a state machine. Before writing any pipeline code, you need to explicitly model the states a saga instance can be in.
For our order processing pipeline, the saga steps are:
orders table)order.created event to KafkaThe compensating transactions are:
CANCELLED in PostgreSQL (we don't delete — see why below)order.cancelled event to KafkaHere's the state machine:
PENDING → STEP_1_COMPLETE → STEP_2_COMPLETE → STEP_3_COMPLETE → COMPLETED
↑
FAILED_STEP_1 → (no compensation needed, nothing succeeded)
FAILED_STEP_2 → COMPENSATING_STEP_1 → COMPENSATED
FAILED_STEP_3 → COMPENSATING_STEP_2 → COMPENSATING_STEP_1 → COMPENSATED
FAILED_STEP_4 → COMPENSATING_STEP_3 → COMPENSATING_STEP_2 → COMPENSATING_STEP_1 → COMPENSATED
Notice that compensation always runs in reverse and compensation itself can fail (which we'll handle). Let's encode this in Python:
from enum import Enum, auto
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, Dict, Any
import uuid
class SagaStatus(Enum):
PENDING = "PENDING"
STEP_1_COMPLETE = "STEP_1_COMPLETE" # Order written to Postgres
STEP_2_COMPLETE = "STEP_2_COMPLETE" # Kafka event published
STEP_3_COMPLETE = "STEP_3_COMPLETE" # Redis inventory decremented
COMPLETED = "COMPLETED" # Analytics write successful
FAILED = "FAILED" # Saga failed, compensation needed
COMPENSATING = "COMPENSATING" # Compensation in progress
COMPENSATED = "COMPENSATED" # Fully rolled back
COMPENSATION_FAILED = "COMPENSATION_FAILED" # Needs manual intervention
@dataclass
class SagaInstance:
saga_id: str
order_data: Dict[str, Any]
status: SagaStatus
failed_at_step: Optional[int] = None
failure_reason: Optional[str] = None
compensation_step: Optional[int] = None # which step we're compensating
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
retry_count: int = 0
def to_dict(self) -> Dict[str, Any]:
return {
"saga_id": self.saga_id,
"order_data": self.order_data,
"status": self.status.value,
"failed_at_step": self.failed_at_step,
"failure_reason": self.failure_reason,
"compensation_step": self.compensation_step,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
"retry_count": self.retry_count,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SagaInstance":
data["status"] = SagaStatus(data["status"])
data["created_at"] = datetime.fromisoformat(data["created_at"])
data["updated_at"] = datetime.fromisoformat(data["updated_at"])
return cls(**data)
Why not delete failed records? It's tempting to delete the PostgreSQL row as the compensating action, but deletion destroys audit trails. In financial or order systems, you want every attempt recorded. Use a
statuscolumn with values likeACTIVE,CANCELLED,COMPENSATED. This also makes compensating idempotent — you can mark a rowCANCELLEDtwice safely, whereas a DELETE that runs twice silently succeeds the second time even if a new row with the same ID was inserted by a retry.
The saga state machine is only useful if it survives process crashes. If your pipeline worker dies mid-saga, you need to resume from the last known good state — either retry the failed step or run compensation.
We'll persist saga state in a dedicated PostgreSQL table. Why Postgres and not Redis or a file? Because saga state is critical coordination data that needs:
COMPENSATION_FAILED state)UPDATE ... WHERE status = 'COMPENSATING' to prevent races)CREATE TABLE saga_log (
saga_id UUID PRIMARY KEY,
order_data JSONB NOT NULL,
status TEXT NOT NULL,
failed_at_step INTEGER,
failure_reason TEXT,
compensation_step INTEGER,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
retry_count INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX idx_saga_log_status ON saga_log (status);
CREATE INDEX idx_saga_log_updated_at ON saga_log (updated_at);
Now let's build the persistence layer:
import asyncpg
import json
from typing import Optional
class SagaRepository:
def __init__(self, pool: asyncpg.Pool):
self.pool = pool
async def create(self, saga: SagaInstance) -> None:
async with self.pool.acquire() as conn:
await conn.execute("""
INSERT INTO saga_log
(saga_id, order_data, status, failed_at_step, failure_reason,
compensation_step, created_at, updated_at, retry_count)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
""",
saga.saga_id,
json.dumps(saga.order_data),
saga.status.value,
saga.failed_at_step,
saga.failure_reason,
saga.compensation_step,
saga.created_at,
saga.updated_at,
saga.retry_count,
)
async def update_status(
self,
saga_id: str,
new_status: SagaStatus,
failed_at_step: Optional[int] = None,
failure_reason: Optional[str] = None,
compensation_step: Optional[int] = None,
) -> bool:
"""
Returns True if the update succeeded. Using optimistic locking
via a WHERE clause prevents concurrent workers from double-processing.
"""
async with self.pool.acquire() as conn:
result = await conn.execute("""
UPDATE saga_log
SET
status = $2,
failed_at_step = COALESCE($3, failed_at_step),
failure_reason = COALESCE($4, failure_reason),
compensation_step = $5,
updated_at = NOW(),
retry_count = retry_count + 1
WHERE saga_id = $1
""",
saga_id,
new_status.value,
failed_at_step,
failure_reason,
compensation_step,
)
return result == "UPDATE 1"
async def find_stuck_sagas(self, older_than_minutes: int = 30) -> list[dict]:
"""Find sagas that started compensating but haven't finished."""
async with self.pool.acquire() as conn:
rows = await conn.fetch("""
SELECT * FROM saga_log
WHERE status IN ('FAILED', 'COMPENSATING', 'COMPENSATION_FAILED')
AND updated_at < NOW() - INTERVAL '1 minute' * $1
ORDER BY updated_at ASC
""", older_than_minutes)
return [dict(row) for row in rows]
The recovery worker pattern: In production, run a separate lightweight process that queries
find_stuck_sagas()every few minutes and re-enqueues them for compensation. This is your "dead letter queue" equivalent for sagas. Without this, a saga that fails during compensation will sit inCOMPENSATINGstate forever until someone notices.
Now we build the orchestrator — the central class that drives each step, handles failures, and coordinates compensation. The key design principle here is that every step and every compensating transaction is a separate async method, and the orchestrator wraps each one in consistent error handling and state persistence.
import asyncio
import asyncpg
import redis.asyncio as aioredis
from aiokafka import AIOKafkaProducer
import json
import logging
from datetime import datetime
logger = logging.getLogger(__name__)
class OrderSagaOrchestrator:
def __init__(
self,
pg_pool: asyncpg.Pool, # orders database
analytics_pool: asyncpg.Pool, # analytics database
kafka_producer: AIOKafkaProducer,
redis_client: aioredis.Redis,
saga_repo: SagaRepository,
):
self.pg = pg_pool
self.analytics = analytics_pool
self.kafka = kafka_producer
self.redis = redis_client
self.repo = saga_repo
async def execute(self, order_data: dict) -> SagaInstance:
"""Entry point: create a new saga and execute all steps."""
saga = SagaInstance(
saga_id=str(uuid.uuid4()),
order_data=order_data,
status=SagaStatus.PENDING,
)
await self.repo.create(saga)
logger.info(f"Starting saga {saga.saga_id} for order {order_data.get('order_id')}")
try:
await self._step_1_write_order(saga)
await self._step_2_publish_kafka_event(saga)
await self._step_3_update_inventory(saga)
await self._step_4_write_analytics(saga)
await self.repo.update_status(saga.saga_id, SagaStatus.COMPLETED)
saga.status = SagaStatus.COMPLETED
logger.info(f"Saga {saga.saga_id} completed successfully")
return saga
except SagaStepException as e:
logger.error(f"Saga {saga.saga_id} failed at step {e.step}: {e}")
saga.failed_at_step = e.step
saga.failure_reason = str(e)
saga.status = SagaStatus.FAILED
await self.repo.update_status(
saga.saga_id, SagaStatus.FAILED,
failed_at_step=e.step,
failure_reason=str(e),
)
await self._compensate(saga)
return saga
# -------------------------------------------------------------------------
# Forward Steps
# -------------------------------------------------------------------------
async def _step_1_write_order(self, saga: SagaInstance) -> None:
try:
async with self.pg.acquire() as conn:
await conn.execute("""
INSERT INTO orders
(order_id, customer_id, product_id, quantity,
total_amount, status, saga_id, created_at)
VALUES ($1, $2, $3, $4, $5, 'PENDING', $6, NOW())
ON CONFLICT (order_id) DO NOTHING
""",
saga.order_data["order_id"],
saga.order_data["customer_id"],
saga.order_data["product_id"],
saga.order_data["quantity"],
saga.order_data["total_amount"],
saga.saga_id,
)
await self.repo.update_status(saga.saga_id, SagaStatus.STEP_1_COMPLETE)
saga.status = SagaStatus.STEP_1_COMPLETE
except Exception as e:
raise SagaStepException(step=1, cause=e)
async def _step_2_publish_kafka_event(self, saga: SagaInstance) -> None:
try:
event = {
"event_type": "order.created",
"saga_id": saga.saga_id,
"order_id": saga.order_data["order_id"],
"timestamp": datetime.utcnow().isoformat(),
"payload": saga.order_data,
}
await self.kafka.send_and_wait(
topic="order-events",
key=saga.order_data["order_id"].encode(),
value=json.dumps(event).encode(),
headers=[("saga_id", saga.saga_id.encode())],
)
await self.repo.update_status(saga.saga_id, SagaStatus.STEP_2_COMPLETE)
saga.status = SagaStatus.STEP_2_COMPLETE
except Exception as e:
raise SagaStepException(step=2, cause=e)
async def _step_3_update_inventory(self, saga: SagaInstance) -> None:
try:
product_id = saga.order_data["product_id"]
quantity = saga.order_data["quantity"]
inventory_key = f"inventory:{product_id}"
# Lua script ensures atomic check-and-decrement
lua_script = """
local current = tonumber(redis.call('GET', KEYS[1]))
if current == nil then
return redis.error_reply("INVENTORY_NOT_FOUND")
end
if current < tonumber(ARGV[1]) then
return redis.error_reply("INSUFFICIENT_INVENTORY")
end
return redis.call('DECRBY', KEYS[1], ARGV[1])
"""
result = await self.redis.eval(lua_script, 1, inventory_key, quantity)
if result < 0:
raise ValueError(f"Inventory went negative for product {product_id}")
await self.repo.update_status(saga.saga_id, SagaStatus.STEP_3_COMPLETE)
saga.status = SagaStatus.STEP_3_COMPLETE
except Exception as e:
raise SagaStepException(step=3, cause=e)
async def _step_4_write_analytics(self, saga: SagaInstance) -> None:
try:
async with self.analytics.acquire() as conn:
await conn.execute("""
INSERT INTO analytics.order_facts
(order_id, customer_id, product_id, quantity,
total_amount, saga_id, event_time)
VALUES ($1, $2, $3, $4, $5, $6, NOW())
ON CONFLICT (order_id) DO UPDATE
SET saga_id = EXCLUDED.saga_id,
event_time = EXCLUDED.event_time
""",
saga.order_data["order_id"],
saga.order_data["customer_id"],
saga.order_data["product_id"],
saga.order_data["quantity"],
saga.order_data["total_amount"],
saga.saga_id,
)
except Exception as e:
raise SagaStepException(step=4, cause=e)
Notice the Lua script in step 3. This is critical: Redis doesn't have native transactions with rollback, so the only way to do an atomic check-and-decrement ("only decrement if we have enough inventory") is with a Lua script that executes atomically on the Redis server. Without it, you'd have a race condition where two concurrent sagas both read the inventory, both see 1 unit available, and both decrement — resulting in -1 inventory.
Here's where most tutorials fall short. They describe compensation conceptually but skip the hard parts: what happens when compensation itself fails? How do you ensure compensations are idempotent? How do you handle partial compensation failures?
async def _compensate(self, saga: SagaInstance) -> None:
"""
Run compensating transactions in reverse order from the last
successful step. Each compensation is attempted with retries.
"""
await self.repo.update_status(
saga.saga_id, SagaStatus.COMPENSATING,
compensation_step=saga.failed_at_step - 1,
)
# Determine which compensations to run based on how far we got
steps_to_compensate = range(saga.failed_at_step - 1, 0, -1)
for step in steps_to_compensate:
success = await self._run_compensation_with_retry(saga, step)
if not success:
logger.critical(
f"Compensation failed for saga {saga.saga_id} at step {step}. "
f"Manual intervention required."
)
await self.repo.update_status(
saga.saga_id, SagaStatus.COMPENSATION_FAILED,
compensation_step=step,
)
# Alert operations team here
await self._alert_operations(saga, step)
return
await self.repo.update_status(
saga.saga_id, SagaStatus.COMPENSATED,
compensation_step=None,
)
saga.status = SagaStatus.COMPENSATED
logger.info(f"Saga {saga.saga_id} fully compensated")
async def _run_compensation_with_retry(
self, saga: SagaInstance, step: int, max_retries: int = 5
) -> bool:
"""Retry a compensation step with exponential backoff."""
compensation_map = {
1: self._compensate_step_1,
2: self._compensate_step_2,
3: self._compensate_step_3,
}
compensate_fn = compensation_map[step]
for attempt in range(max_retries):
try:
await compensate_fn(saga)
return True
except Exception as e:
wait = min(2 ** attempt, 30) # cap at 30 seconds
logger.warning(
f"Compensation step {step} attempt {attempt + 1} failed "
f"for saga {saga.saga_id}: {e}. Retrying in {wait}s."
)
await asyncio.sleep(wait)
return False
async def _compensate_step_1(self, saga: SagaInstance) -> None:
"""
Compensate the Postgres order write by marking it CANCELLED.
This is idempotent: setting CANCELLED on an already-CANCELLED row is safe.
"""
async with self.pg.acquire() as conn:
await conn.execute("""
UPDATE orders
SET
status = 'CANCELLED',
cancelled_at = NOW(),
cancellation_reason = 'saga_compensation',
saga_id = $2
WHERE order_id = $1
AND status != 'SHIPPED' -- never cancel an already-shipped order
""",
saga.order_data["order_id"],
saga.saga_id,
)
async def _compensate_step_2(self, saga: SagaInstance) -> None:
"""
Compensate the Kafka publish by publishing an order.cancelled event.
Kafka messages are immutable — we can't retract the order.created event.
Consumers must handle both events and let cancelled win.
"""
event = {
"event_type": "order.cancelled",
"saga_id": saga.saga_id,
"order_id": saga.order_data["order_id"],
"timestamp": datetime.utcnow().isoformat(),
"reason": "saga_compensation",
"compensating_for": "order.created",
}
await self.kafka.send_and_wait(
topic="order-events",
key=saga.order_data["order_id"].encode(),
value=json.dumps(event).encode(),
headers=[
("saga_id", saga.saga_id.encode()),
("event_type", b"compensating"),
],
)
async def _compensate_step_3(self, saga: SagaInstance) -> None:
"""
Compensate the inventory decrement by incrementing back.
We use a Lua script with idempotency tracking to prevent double-increments.
"""
product_id = saga.order_data["product_id"]
quantity = saga.order_data["quantity"]
inventory_key = f"inventory:{product_id}"
compensation_key = f"compensation:{saga.saga_id}:step3"
lua_script = """
-- Check if we've already compensated this saga step
if redis.call('EXISTS', KEYS[2]) == 1 then
return redis.call('GET', KEYS[1])
end
-- Perform the increment and mark as compensated
local new_value = redis.call('INCRBY', KEYS[1], ARGV[1])
redis.call('SET', KEYS[2], '1', 'EX', 86400) -- TTL 24h
return new_value
"""
await self.redis.eval(
lua_script, 2, inventory_key, compensation_key, quantity
)
class SagaStepException(Exception):
def __init__(self, step: int, cause: Exception):
self.step = step
self.cause = cause
super().__init__(f"Step {step} failed: {cause}")
The idempotency key in _compensate_step_3 deserves explanation. Because compensations are retried on failure, there's a real risk that a compensation runs, partially succeeds (the Redis INCRBY completes), and then crashes before acknowledging success to the saga orchestrator. On the next retry, the compensation would run again and increment the inventory twice. The Lua script checks for a compensation marker key first and returns early if it exists — making the entire operation idempotent regardless of how many times you call it.
Some of your sinks won't be under your control. You might be writing to a payment processor API, a CRM system, or a third-party fulfillment service. These systems often:
The pattern here is the outbox table. Instead of calling the external system directly from your saga step, you write a record to an outbox table within the same local Postgres transaction as your business data write. A separate worker reads the outbox and makes the external call. This decouples the saga's atomicity from external system availability.
CREATE TABLE outbox (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
saga_id UUID NOT NULL,
aggregate_type TEXT NOT NULL, -- e.g., 'order'
aggregate_id TEXT NOT NULL, -- e.g., order_id
event_type TEXT NOT NULL, -- e.g., 'charge_customer'
payload JSONB NOT NULL,
status TEXT NOT NULL DEFAULT 'PENDING',
attempts INTEGER NOT NULL DEFAULT 0,
last_attempted TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
async def _step_2_write_to_outbox(self, saga: SagaInstance) -> None:
"""
Write the external call intent to the outbox within the same
Postgres transaction as the order write. Either both succeed or neither.
"""
async with self.pg.acquire() as conn:
async with conn.transaction():
# The order write and outbox write are in the same transaction
await conn.execute("""
UPDATE orders SET status = 'PROCESSING'
WHERE order_id = $1
""", saga.order_data["order_id"])
await conn.execute("""
INSERT INTO outbox
(saga_id, aggregate_type, aggregate_id, event_type, payload)
VALUES ($1, 'order', $2, 'publish_order_event', $3)
""",
saga.saga_id,
saga.order_data["order_id"],
json.dumps(saga.order_data),
)
This is called the Transactional Outbox Pattern, and it composes beautifully with Sagas. The saga step succeeds as soon as the outbox record is written. The compensation for this step deletes or voids the outbox record if the external worker hasn't processed it yet, or publishes a compensating call to the external system if it has.
Warning: Don't use the outbox pattern as an excuse to skip thinking about compensation. If the outbox worker processes the external call successfully but then your saga fails at a later step, you still need to call the external API to reverse the operation. The outbox just decouples when you call it, not whether you need a compensating action.
A saga that fails silently is worse than one that fails loudly. You need three things: structured logging on every state transition, metrics on saga duration and failure rates, and a way to query saga state from an operations dashboard.
import structlog
from prometheus_client import Counter, Histogram, Gauge
import time
saga_started = Counter("saga_started_total", "Total sagas started", ["pipeline"])
saga_completed = Counter("saga_completed_total", "Total sagas completed", ["pipeline"])
saga_failed = Counter("saga_failed_total", "Total sagas that failed", ["pipeline", "step"])
saga_compensated = Counter("saga_compensated_total", "Total sagas fully compensated", ["pipeline"])
saga_compensation_failed = Counter(
"saga_compensation_failed_total",
"Sagas where compensation itself failed — needs human",
["pipeline", "step"]
)
saga_duration = Histogram(
"saga_duration_seconds",
"Time from saga start to completion or compensation",
["pipeline", "outcome"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60, 120],
)
class InstrumentedSagaOrchestrator(OrderSagaOrchestrator):
async def execute(self, order_data: dict) -> SagaInstance:
log = structlog.get_logger().bind(
pipeline="order_processing",
order_id=order_data.get("order_id"),
)
start_time = time.monotonic()
saga_started.labels(pipeline="order_processing").inc()
saga = await super().execute(order_data)
duration = time.monotonic() - start_time
if saga.status == SagaStatus.COMPLETED:
saga_completed.labels(pipeline="order_processing").inc()
saga_duration.labels(
pipeline="order_processing", outcome="completed"
).observe(duration)
log.info("saga_completed", saga_id=saga.saga_id, duration_s=round(duration, 3))
elif saga.status == SagaStatus.COMPENSATED:
saga_failed.labels(
pipeline="order_processing", step=str(saga.failed_at_step)
).inc()
saga_compensated.labels(pipeline="order_processing").inc()
saga_duration.labels(
pipeline="order_processing", outcome="compensated"
).observe(duration)
log.warning(
"saga_compensated",
saga_id=saga.saga_id,
failed_at_step=saga.failed_at_step,
reason=saga.failure_reason,
duration_s=round(duration, 3),
)
elif saga.status == SagaStatus.COMPENSATION_FAILED:
saga_compensation_failed.labels(
pipeline="order_processing",
step=str(saga.compensation_step),
).inc()
log.critical(
"saga_compensation_failed_needs_human",
saga_id=saga.saga_id,
failed_at_step=saga.failed_at_step,
compensation_step=saga.compensation_step,
)
return saga
For the operations dashboard, the saga_log table is your source of truth. A simple query to understand your pipeline health:
-- Saga health summary for the last 24 hours
SELECT
status,
COUNT(*) as count,
AVG(EXTRACT(EPOCH FROM (updated_at - created_at))) as avg_duration_seconds,
MAX(retry_count) as max_retries
FROM saga_log
WHERE created_at > NOW() - INTERVAL '24 hours'
GROUP BY status
ORDER BY count DESC;
-- Find all sagas stuck in compensation for more than an hour
SELECT saga_id, failed_at_step, failure_reason, compensation_step, updated_at
FROM saga_log
WHERE status IN ('COMPENSATING', 'COMPENSATION_FAILED')
AND updated_at < NOW() - INTERVAL '1 hour'
ORDER BY updated_at ASC;
In this exercise, you'll extend the saga to handle a fifth step: calling a shipping provider's REST API to reserve a delivery slot. This step is non-idempotent (each call creates a new reservation) and has a compensating operation (cancel the reservation using the returned reservation ID).
Setup: Use httpx for async HTTP calls and assume the shipping API:
POST /reservations → { "reservation_id": "...", "slot": "..." } DELETE /reservations/{reservation_id} → 204 on successYour tasks:
Add a STEP_5_COMPLETE state to SagaStatus and update the state machine diagram.
Implement _step_5_reserve_shipping_slot(saga) that calls the API and stores the reservation_id in the order_data dict (so it's available to the compensating transaction).
Hint: you'll need to persist the reservation_id back to saga_log.order_data after the API call succeeds. Use a UPDATE saga_log SET order_data = order_data || $1 with the new field.
Implement _compensate_step_5(saga) that:
reservation_id from saga.order_dataDELETE /reservations/{reservation_id}Modify _compensate to include step 5 in the compensation chain. Remember: if the saga fails at step 5, compensation runs steps 4 → 3 → 2 → 1. If it fails at step 4, step 5's compensation shouldn't run (it never succeeded).
Write a test that simulates the shipping API returning 503 on the first two attempts and 200 on the third. Verify the saga completes successfully and the retry_count in the saga log reflects the retries.
Mistake 1: Using the saga orchestrator as a state machine without persisting state between steps.
If you only keep saga state in memory, a process crash between step 2 and step 3 leaves you with no record of what happened. You can't run compensation because you don't know how far the saga progressed. Always persist state to durable storage before marking a step complete, not after.
Mistake 2: Compensating transactions that aren't idempotent.
If your step 3 compensation is DELETE FROM reservations WHERE saga_id = $1 and it fails halfway through, the second attempt will silently succeed but do nothing — which sounds fine until you realize the deletion might have only partially run on a partitioned table and the second run leaves a dangling row. Use UPDATE ... SET status = 'CANCELLED' patterns instead, and add idempotency keys to Redis operations.
Mistake 3: Not ordering compensation correctly.
Compensation must run in strict reverse order because later steps may depend on state set by earlier steps. If step 3 reads data written by step 1, compensating step 1 before step 3 can leave step 3 in an inconsistent state that its compensation can't handle.
Mistake 4: Treating COMPENSATION_FAILED as a recoverable state.
When compensation fails after retries, you have a genuinely inconsistent state. Don't keep retrying automatically — escalate to human intervention immediately. Your compensation may be making incorrect assumptions about the current state (maybe the downstream system changed schema, or an external API changed its behavior). Infinite automatic retries can make the inconsistency worse.
Mistake 5: Publishing Kafka compensation events to a different topic than the forward events.
Consumers that process order.created events need to be able to see order.cancelled events to know to stop processing. If you publish compensating events to a different topic, consumers of the original topic will never know the saga was rolled back. Use the same topic with the same partition key so that ordering guarantees apply.
Troubleshooting: Saga stuck in COMPENSATING for hours.
Check the compensation_step column — it tells you exactly which step's compensation is failing. Then look at the logs for that specific step. The most common causes are: the downstream system is rate-limiting your retries (add jitter to backoff), the idempotency key expired in Redis (extend TTL), or the downstream system's API changed and your compensation call is malformed (check the API response body in your logs, not just the status code).
Troubleshooting: Duplicate saga executions creating duplicate rows.
This happens when your pipeline consumer framework retries message delivery and your saga's step 1 doesn't use ON CONFLICT DO NOTHING or similar upsert semantics. Ensure every saga step is fully idempotent by using the order_id as a natural key with conflict handling. The saga_id will differ between attempts but the order_id should be stable.
You've now built a production-grade distributed saga implementation for multi-sink data pipelines. The core ideas to carry with you:
COMPENSATION_FAILED as a P1 incident.Where to go from here:
If you're running this in a distributed environment with multiple pipeline workers, you'll need to add distributed locking to prevent two workers from picking up the same stuck saga. Investigate Redis SETNX-based locks or PostgreSQL advisory locks for this.
For higher throughput scenarios, look into event-sourced saga state: instead of a single saga_log row that gets updated, append a row for every state transition. This gives you a complete audit trail and avoids update contention, at the cost of more complex state reconstruction.
Finally, if you find yourself writing sagas for many different pipelines, consider adopting a workflow orchestration tool like Temporal or Conductor that provides the state machine infrastructure out of the box and lets you focus on writing your step and compensation logic. The patterns you've learned here are exactly what those tools implement under the hood — understanding them from first principles makes you dramatically more effective when using higher-level abstractions.
Learning Path: Data Pipeline Fundamentals