Microservices

Testing Distributed Transactions and Saga Patterns in Microservices

14 Sep 2026 — 6 min read

Distributed transactions are one of the hardest problems in microservices. Two-phase commit works in theory but fails in practice — it's slow, fragile, and locks resources across service boundaries. The saga pattern solves this with a sequence of local transactions with compensating actions. But testing sagas is just as hard as implementing them.

This guide covers testing strategies for both saga patterns — choreography (event-driven) and orchestration — with real examples for the failure cases that matter most.

The Saga Pattern Recap

A saga is a sequence of transactions where:

Each step executes a local transaction and publishes an event
If any step fails, compensating transactions roll back completed steps
There is no global ACID transaction — consistency is eventual

Choreography sagas: Each service reacts to events and publishes its own events. No central coordinator.

Orchestration sagas: A central orchestrator tells each service what to do and handles failures.

The Order Processing Saga Example

Step 1: OrderService      — Create order (PENDING)
Step 2: InventoryService  — Reserve stock
Step 3: PaymentService    — Process payment
Step 4: ShippingService   — Create shipment
Step 5: OrderService      — Update order (CONFIRMED)

Compensation sequence (if step 3 fails):

Compensation 2: InventoryService — Release stock reservation
Compensation 1: OrderService     — Update order (FAILED)

What to Test in a Saga

The happy path is straightforward. The failure paths are where bugs live:

Happy path: All steps succeed, final state is correct
Step failure with compensation: A step fails, all compensations run, final state is clean
Compensation failure: A compensation step fails, system ends in inconsistent state
Duplicate event processing: An event is delivered twice (at-least-once delivery)
Out-of-order events: Events arrive in wrong order
Timeout/dead letter handling: A service never responds
Partial retry: A step succeeds on retry, but compensation has already run

Testing Choreography Sagas

Choreography sagas are harder to test because the flow is implicit in event routing.

Unit Testing Individual Service Handlers

Each service's event handler is a unit. Test it in isolation:

# inventory_service/tests/unit/test_order_event_handler.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from decimal import Decimal

from inventory_service.handlers.order_events import OrderEventHandler
from inventory_service.domain.models import StockReservation, StockError

@pytest.fixture
def mock_stock_repo():
    return AsyncMock()

@pytest.fixture
def mock_event_publisher():
    return AsyncMock()

@pytest.fixture
def handler(mock_stock_repo, mock_event_publisher):
    return OrderEventHandler(
        stock_repo=mock_stock_repo,
        event_publisher=mock_event_publisher,
    )

@pytest.mark.asyncio
async def test_reserves_stock_and_publishes_success_event(handler, mock_stock_repo, mock_event_publisher):
    # Arrange
    mock_stock_repo.reserve.return_value = StockReservation(
        reservation_id="res-123",
        product_id="prod-456",
        quantity=2,
    )
    
    event = {
        "event_type": "order.created",
        "order_id": "ord-789",
        "items": [{"product_id": "prod-456", "quantity": 2}],
    }
    
    # Act
    await handler.handle_order_created(event)
    
    # Assert
    mock_stock_repo.reserve.assert_awaited_once_with("prod-456", 2, order_id="ord-789")
    mock_event_publisher.publish.assert_awaited_once_with(
        topic="inventory.reserved",
        message={
            "event_type": "inventory.reserved",
            "order_id": "ord-789",
            "reservation_id": "res-123",
        }
    )

@pytest.mark.asyncio
async def test_publishes_failure_event_when_out_of_stock(handler, mock_stock_repo, mock_event_publisher):
    # Arrange
    mock_stock_repo.reserve.side_effect = StockError("Insufficient stock: 0 available, 2 requested")
    
    event = {
        "event_type": "order.created",
        "order_id": "ord-789",
        "items": [{"product_id": "prod-456", "quantity": 2}],
    }
    
    # Act
    await handler.handle_order_created(event)
    
    # Assert: failure event published, not an exception raised
    mock_event_publisher.publish.assert_awaited_once_with(
        topic="inventory.reservation_failed",
        message={
            "event_type": "inventory.reservation_failed",
            "order_id": "ord-789",
            "reason": "INSUFFICIENT_STOCK",
        }
    )

@pytest.mark.asyncio
async def test_idempotent_handling_of_duplicate_events(handler, mock_stock_repo, mock_event_publisher):
    """The same event delivered twice should not result in double reservation."""
    mock_stock_repo.reservation_exists.return_value = True  # Already processed
    
    event = {
        "event_type": "order.created",
        "order_id": "ord-789",
        "items": [{"product_id": "prod-456", "quantity": 2}],
    }
    
    # Act — process same event twice
    await handler.handle_order_created(event)
    await handler.handle_order_created(event)
    
    # Assert: reserve called only once despite two event deliveries
    mock_stock_repo.reserve.assert_awaited_once()

Integration Testing the Full Saga Flow

For choreography sagas, integration tests require a running message broker:

# tests/integration/test_order_saga.py
import pytest
import asyncio
import httpx
from datetime import datetime

@pytest.mark.asyncio
async def test_order_saga_happy_path(kafka, postgres):
    """
    Full saga: order created → stock reserved → payment processed → order confirmed.
    All services running, all steps succeed.
    """
    async with httpx.AsyncClient() as client:
        # Trigger saga by creating an order
        create_resp = await client.post(
            "http://order-service:8080/orders",
            json={
                "customer_id": "cust-test-001",
                "items": [{"product_id": "prod-in-stock", "quantity": 1}],
                "payment_method": "test-card-success",
            }
        )
        assert create_resp.status_code == 201
        order_id = create_resp.json()["order_id"]
    
    # Wait for saga to complete (async — eventual consistency)
    final_status = await wait_for_order_status(
        order_id=order_id,
        expected_status="CONFIRMED",
        timeout=15,
    )
    
    assert final_status == "CONFIRMED"
    
    # Verify side effects: stock was reserved and marked as sold
    stock = await get_stock("prod-in-stock")
    assert stock["reserved"] == 0  # Reservation fulfilled
    assert stock["sold"] == 1
    
    # Verify payment record exists
    payment = await get_payment_for_order(order_id)
    assert payment["status"] == "CAPTURED"

@pytest.mark.asyncio
async def test_order_saga_compensation_on_payment_failure(kafka, postgres):
    """
    Saga with payment failure: order created → stock reserved → payment fails
    → stock released → order marked FAILED.
    """
    async with httpx.AsyncClient() as client:
        create_resp = await client.post(
            "http://order-service:8080/orders",
            json={
                "customer_id": "cust-test-002",
                "items": [{"product_id": "prod-in-stock", "quantity": 1}],
                "payment_method": "test-card-decline",  # Will be declined
            }
        )
        assert create_resp.status_code == 201
        order_id = create_resp.json()["order_id"]
    
    # Wait for saga to fail and compensate
    final_status = await wait_for_order_status(
        order_id=order_id,
        expected_status="FAILED",
        timeout=15,
    )
    
    assert final_status == "FAILED"
    
    # CRITICAL: Verify compensation ran — stock must be released
    stock = await get_stock("prod-in-stock")
    assert stock["reserved"] == 0  # Released, not permanently held
    
    # No payment record
    payment = await get_payment_for_order(order_id)
    assert payment is None

async def wait_for_order_status(order_id: str, expected_status: str, timeout: int) -> str:
    """Poll order status until it reaches expected state or timeout."""
    deadline = asyncio.get_event_loop().time() + timeout
    
    async with httpx.AsyncClient() as client:
        while asyncio.get_event_loop().time() < deadline:
            resp = await client.get(f"http://order-service:8080/orders/{order_id}")
            status = resp.json()["status"]
            
            if status == expected_status:
                return status
            
            if status in ("FAILED", "CANCELLED") and expected_status == "CONFIRMED":
                raise AssertionError(f"Order reached terminal status {status} before {expected_status}")
            
            await asyncio.sleep(0.5)
    
    raise TimeoutError(f"Order {order_id} did not reach {expected_status} within {timeout}s")

Testing Orchestration Sagas

Orchestration sagas have a central coordinator. This makes unit testing the orchestrator straightforward:

# order_saga/tests/unit/test_order_saga_orchestrator.py
import pytest
from unittest.mock import AsyncMock, patch

from order_saga.orchestrator import OrderSagaOrchestrator
from order_saga.domain.models import SagaState, SagaStep

@pytest.fixture
def mock_inventory_client():
    return AsyncMock()

@pytest.fixture
def mock_payment_client():
    return AsyncMock()

@pytest.fixture
def mock_shipping_client():
    return AsyncMock()

@pytest.fixture
def orchestrator(mock_inventory_client, mock_payment_client, mock_shipping_client, mock_saga_repo):
    return OrderSagaOrchestrator(
        inventory=mock_inventory_client,
        payment=mock_payment_client,
        shipping=mock_shipping_client,
        saga_repo=mock_saga_repo,
    )

@pytest.mark.asyncio
async def test_all_steps_succeed_reaches_completed_state(orchestrator, mock_inventory_client, mock_payment_client, mock_shipping_client):
    # All services return success
    mock_inventory_client.reserve_stock.return_value = {"reservation_id": "res-123"}
    mock_payment_client.process_payment.return_value = {"payment_id": "pay-456", "status": "CONFIRMED"}
    mock_shipping_client.create_shipment.return_value = {"shipment_id": "ship-789"}
    
    result = await orchestrator.execute(order={
        "order_id": "ord-001",
        "customer_id": "cust-001",
        "items": [{"product_id": "prod-001", "quantity": 1}],
        "payment_method": "card-success",
    })
    
    assert result.state == SagaState.COMPLETED
    assert result.payment_id == "pay-456"
    assert result.shipment_id == "ship-789"

@pytest.mark.asyncio
async def test_payment_failure_compensates_inventory(orchestrator, mock_inventory_client, mock_payment_client):
    # Inventory succeeds, payment fails
    mock_inventory_client.reserve_stock.return_value = {"reservation_id": "res-123"}
    mock_payment_client.process_payment.side_effect = PaymentDeclinedError("Card declined")
    
    result = await orchestrator.execute(order={
        "order_id": "ord-002",
        "customer_id": "cust-002",
        "items": [{"product_id": "prod-001", "quantity": 1}],
        "payment_method": "card-decline",
    })
    
    assert result.state == SagaState.COMPENSATED
    
    # Compensation must have been called
    mock_inventory_client.release_reservation.assert_awaited_once_with("res-123")
    
    # Shipping must NOT have been called
    mock_payment_client.process_payment.assert_awaited_once()
    # Shipping never called
    assert not hasattr(mock_shipping_client, 'create_shipment') or \
           not mock_shipping_client.create_shipment.called

@pytest.mark.asyncio
async def test_compensation_failure_is_recorded_and_escalated(orchestrator, mock_inventory_client, mock_payment_client, mock_saga_repo):
    """When compensation fails, the saga must record this and escalate — not silently fail."""
    mock_inventory_client.reserve_stock.return_value = {"reservation_id": "res-123"}
    mock_payment_client.process_payment.side_effect = PaymentDeclinedError("Card declined")
    mock_inventory_client.release_reservation.side_effect = Exception("Inventory service down")
    
    result = await orchestrator.execute(order={
        "order_id": "ord-003",
        "customer_id": "cust-003",
        "items": [{"product_id": "prod-001", "quantity": 1}],
        "payment_method": "card-decline",
    })
    
    # Saga must not be in COMPENSATED state — compensation failed
    assert result.state == SagaState.COMPENSATION_FAILED
    
    # Must be persisted for manual intervention
    mock_saga_repo.save.assert_awaited()
    saved_saga = mock_saga_repo.save.call_args[0][0]
    assert saved_saga.state == SagaState.COMPENSATION_FAILED
    assert "release_reservation" in saved_saga.failed_compensations

Testing Idempotency

This is the most commonly missed test. Message brokers deliver at-least-once. Your saga steps must be idempotent:

@pytest.mark.asyncio
async def test_saga_step_is_idempotent_on_duplicate_trigger(orchestrator, mock_inventory_client):
    """Same saga step triggered twice must not double-reserve stock."""
    mock_inventory_client.reserve_stock.return_value = {"reservation_id": "res-123"}
    
    order = {
        "order_id": "ord-dup-001",
        "items": [{"product_id": "prod-001", "quantity": 1}],
    }
    
    # Execute step twice (simulates duplicate event delivery)
    await orchestrator.execute_step("reserve_inventory", order)
    await orchestrator.execute_step("reserve_inventory", order)
    
    # Stock reservation must be called only once
    mock_inventory_client.reserve_stock.assert_awaited_once()

@pytest.mark.asyncio
async def test_saga_step_skipped_if_already_completed(orchestrator, mock_saga_repo):
    """If saga already completed a step, re-executing it should be a no-op."""
    # Saga already has this step marked as completed
    mock_saga_repo.get_step_status.return_value = SagaStep(
        name="reserve_inventory",
        status="COMPLETED",
        result={"reservation_id": "res-existing-123"},
    )
    
    result = await orchestrator.execute_step("reserve_inventory", {"order_id": "ord-001"})
    
    # Should return the existing result without calling the service again
    assert result["reservation_id"] == "res-existing-123"
    mock_inventory_client.reserve_stock.assert_not_awaited()

Key Takeaways

Unit test each service's event handler in isolation — mock the repo and publisher, verify the published event type and content
Integration test the full saga flow with a real message broker and real databases — verify compensation actually reverses side effects
Always test idempotency explicitly — at-least-once delivery means your saga steps will run twice in production
Test compensation failures — they leave the system in an inconsistent state and require manual intervention
Orchestration sagas are easier to unit test than choreography sagas — the coordinator contains the saga logic in one place
For choreography sagas, integration tests are mandatory — the flow is too distributed to reason about from unit tests alone