Implementing Idempotent Saga Transactions with Compensating Logic via Nginx, RabbitMQ, and a PostgreSQL State Log

The transition from a monolithic architecture to microservices introduced a fundamental transactional problem. A single, atomic database transaction that once reliably handled order creation—debiting inventory, processing payment, and creating an order record—became a distributed nightmare. Our initial foray into two-phase commit (2PC) was a disaster in practice; the coordinating service became a performance bottleneck, and locks held across services led to cascading failures during network partitions. It was clear this approach wouldn’t scale and lacked the resilience we required.

This led us down the path of eventual consistency and the Saga pattern. The concept is straightforward: replace a single, long-running ACID transaction with a sequence of smaller, independent local transactions, each executed by a different microservice. If any step in the sequence fails, a series of compensating transactions are executed in reverse order to semantically undo the work of the completed steps. We opted for an orchestration-based Saga, where a central orchestrator directs the flow, over a choreography approach, as it provided better visibility and explicit control over the workflow, which is critical in a complex business process. The real challenge, however, wasn’t just implementing the happy path; it was building a system resilient to network retries, message duplication, and service failures.

Our final architecture hinges on three core components, each with a specific, non-trivial role:

1. PostgreSQL: Not just for business data, but as the transactional state machine log for the Saga orchestrator. Its ACID guarantees are leveraged to ensure the state of the Saga itself is never corrupted.
2. RabbitMQ: Serves as the message bus for commands and events between the orchestrator and participant services. Its routing features and dead-lettering capabilities provide the flexibility needed for complex failure handling.
3. Nginx: Acts as the API Gateway, but its role extends beyond simple routing. We use it to enforce idempotency at the edge, shielding the entire downstream system from duplicate requests initiated by clients or proxies.

Here is the high-level flow of our orchestrated Saga for creating an order:

sequenceDiagram
    participant Client
    participant Nginx
    participant OrderService (Orchestrator)
    participant SagaDB [PostgreSQL]
    participant RabbitMQ
    participant InventoryService
    participant PaymentService

    Client->>+Nginx: POST /orders (Idempotency-Key: xyz-123)
    Nginx->>+OrderService: POST /orders (with Idempotency-Key header)
    OrderService->>+SagaDB: CREATE Saga Log (state: PENDING)
    SagaDB-->>-OrderService: Saga ID: 550e8400
    OrderService->>+RabbitMQ: PUBLISH ReserveInventoryCommand (SagaID: 550e8400)
    RabbitMQ-->>-InventoryService: DELIVER ReserveInventoryCommand
    InventoryService->>InventoryService: Execute Local TX (Decrement Stock)
    InventoryService->>+RabbitMQ: PUBLISH InventoryReservedEvent (SagaID: 550e8400)
    RabbitMQ-->>-OrderService: DELIVER InventoryReservedEvent
    OrderService->>+SagaDB: UPDATE Saga Log (step: INVENTORY_RESERVED, state: PROGRESSING)
    OrderService->>+RabbitMQ: PUBLISH ProcessPaymentCommand (SagaID: 550e8400)
    RabbitMQ-->>-PaymentService: DELIVER ProcessPaymentCommand
    Note right of PaymentService: Simulate payment failure
    PaymentService->>+RabbitMQ: PUBLISH PaymentFailedEvent (SagaID: 550e8400, reason: Insufficient Funds)
    RabbitMQ-->>-OrderService: DELIVER PaymentFailedEvent
    OrderService->>+SagaDB: UPDATE Saga Log (state: COMPENSATING)
    OrderService->>+RabbitMQ: PUBLISH ReleaseInventoryCommand (SagaID: 550e8400)
    RabbitMQ-->>-InventoryService: DELIVER ReleaseInventoryCommand
    InventoryService->>InventoryService: Execute Local TX (Increment Stock)
    InventoryService->>+RabbitMQ: PUBLISH InventoryReleasedEvent (SagaID: 550e8400)
    RabbitMQ-->>-OrderService: DELIVER InventoryReleasedEvent
    OrderService->>+SagaDB: UPDATE Saga Log (state: FAILED)
    OrderService-->>-Nginx: HTTP 500 (Order Failed)
    Nginx-->>-Client: HTTP 500 (Order Failed)

Database as the Saga State Machine

The cornerstone of a reliable orchestrator is its state log. A common mistake is to keep this state in-memory, which guarantees data loss on a service restart. Using a relational database like PostgreSQL provides the transactional integrity needed to manage the Saga’s lifecycle. We need to atomically update the Saga’s state as it progresses or fails.

Our schema is designed to be simple yet effective, tracking the overall Saga and its individual steps.

-- Main table to track the state of each saga instance
CREATE TABLE sagas (
    saga_id UUID PRIMARY KEY,
    saga_name VARCHAR(100) NOT NULL,
    -- PENDING, PROGRESSING, COMPLETED, COMPENSATING, FAILED
    status VARCHAR(20) NOT NULL,
    current_step VARCHAR(50) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Table to log the outcome of each step for compensation purposes
CREATE TABLE saga_steps (
    step_id SERIAL PRIMARY KEY,
    saga_id UUID NOT NULL REFERENCES sagas(saga_id),
    step_name VARCHAR(50) NOT NULL,
    -- SUCCEEDED, FAILED
    status VARCHAR(20) NOT NULL,
    -- Data required for compensation, e.g., product_id and quantity
    compensation_data JSONB,
    executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_sagas_status ON sagas(status);
CREATE INDEX idx_saga_steps_saga_id ON saga_steps(saga_id);

The sagas table is the source of truth for a transaction’s state. The saga_steps table is an append-only log of successful steps, crucial for the orchestrator to determine which compensating actions to trigger upon failure. The compensation_data field is vital; for an inventory reservation, it would store the exact items and quantities that need to be released.

Nginx for Edge-Level Idempotency

Client applications and misconfigured proxies are notorious for sending duplicate requests. If a client sends a POST /orders request, experiences a network timeout, and retries, it could trigger two independent Sagas for the same order. Handling this at the application level is possible but repetitive. A pragmatic approach is to enforce it at the gateway.

Nginx can be configured to inspect an Idempotency-Key header. While Nginx itself can’t maintain the state of seen keys across a cluster, we can offload this check to a fast cache like Redis. For simplicity in this example, we’ll demonstrate the header forwarding part, which is the prerequisite for any downstream idempotency check. In a real-world project, a small Lua script within Nginx or a dedicated idempotency-checking service would be used.

# /etc/nginx/nginx.conf

http {
    # ... other http settings

    # In a real setup, this would point to a Redis instance
    # to check if the idempotency key has been seen before.
    # For this example, we'll just ensure the header is passed.
    # A more advanced setup might use the `map` directive or OpenResty/Lua.
    
    upstream order_service {
        server order_service:8000;
    }

    server {
        listen 80;

        location /orders {
            # Ensure the client provides an idempotency key.
            # In a production environment, you might return 400 Bad Request if it's missing.
            if ($http_idempotency_key = "") {
                # This check is basic; a more robust solution is needed for production.
                # For example, ensuring it's a UUID.
            }

            proxy_pass http://order_service;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            
            # Critically, pass the idempotency key to the orchestrator.
            proxy_set_header X-Idempotency-Key $http_idempotency_key;

            # Buffering can cause issues with timeouts and retries.
            # Disabling it makes the interaction more direct.
            proxy_buffering off;
        }
    }
}

This configuration ensures that any Idempotency-Key sent by the client is reliably passed to the OrderService. The orchestrator can then use this key to check its database to see if a Saga with this key has already been initiated.

The Saga Orchestrator Service

This is the brain of the operation. It’s a stateful service that reacts to events and issues commands. We’ll use Python with pika for RabbitMQ and psycopg2 for PostgreSQL.

The orchestrator has three main responsibilities:

1. Initiate a new Saga.
2. Process reply messages from participants and advance the Saga state machine.
3. Trigger compensation flows upon failure.

Here’s a condensed but functional implementation of the OrderService.

# order_service/orchestrator.py
import pika
import psycopg2
import uuid
import json
import logging
import time

# Basic configuration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
DB_CONN = "dbname=saga_db user=user password=password host=postgres"
RABBITMQ_HOST = 'rabbitmq'

class SagaOrchestrator:
    def __init__(self):
        self.db_connection = psycopg2.connect(DB_CONN)
        
        # RabbitMQ setup
        self.mq_connection = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST))
        self.channel = self.mq_connection.channel()

        # Exchange for commands sent to participants
        self.channel.exchange_declare(exchange='saga_commands', exchange_type='topic')
        # Exchange for replies from participants
        self.channel.exchange_declare(exchange='saga_replies', exchange_type='topic')
        
        # Queue for receiving all replies directed at the orchestrator
        reply_queue = self.channel.queue_declare(queue='orchestrator_replies', durable=True)
        self.channel.queue_bind(
            exchange='saga_replies', 
            queue=reply_queue.method.queue, 
            routing_key='orchestrator.*' # Listen to all orchestrator events
        )
        self.channel.basic_consume(
            queue=reply_queue.method.queue, 
            on_message_callback=self.handle_reply, 
            auto_ack=False
        )

    def start_consuming(self):
        logging.info("Orchestrator is waiting for replies...")
        self.channel.start_consuming()

    def initiate_order_saga(self, idempotency_key, order_details):
        """
        Starts the saga for creating an order.
        A real implementation would check the idempotency key against a persistent store
        to prevent duplicate saga initiations.
        """
        saga_id = str(uuid.uuid4())
        logging.info(f"Initiating Saga {saga_id} for idempotency key {idempotency_key}")

        saga_definition = {
            'name': 'CreateOrderSaga',
            'steps': [
                {'name': 'ReserveInventory', 'command_topic': 'inventory.reserve', 'compensation_topic': 'inventory.release'},
                {'name': 'ProcessPayment', 'command_topic': 'payment.process', 'compensation_topic': 'payment.refund'},
                {'name': 'CreateOrder', 'command_topic': 'order.create', 'compensation_topic': 'order.cancel'}
            ]
        }
        
        # Atomically create the saga record in the database
        with self.db_connection.cursor() as cur:
            cur.execute(
                "INSERT INTO sagas (saga_id, saga_name, status, current_step, payload) VALUES (%s, %s, %s, %s, %s)",
                (saga_id, saga_definition['name'], 'PENDING', saga_definition['steps'][0]['name'], json.dumps({'order': order_details, 'saga_def': saga_definition}))
            )
        self.db_connection.commit()

        self.execute_next_step(saga_id)
        return saga_id

    def execute_next_step(self, saga_id):
        with self.db_connection.cursor() as cur:
            cur.execute("SELECT status, current_step, payload FROM sagas WHERE saga_id = %s", (saga_id,))
            saga_status, current_step_name, payload = cur.fetchone()
            saga_def = payload['saga_def']
            
            if saga_status in ['COMPLETED', 'FAILED', 'COMPENSATING']:
                logging.warning(f"Saga {saga_id} is in a terminal or compensating state. No new steps will be executed.")
                return

            current_step_index = next((i for i, step in enumerate(saga_def['steps']) if step['name'] == current_step_name), -1)

            if current_step_index == -1:
                logging.error(f"Saga {saga_id}: Unknown step '{current_step_name}'. Halting.")
                # Update saga status to FAILED
                cur.execute("UPDATE sagas SET status = %s WHERE saga_id = %s", ('FAILED', saga_id))
                self.db_connection.commit()
                return

            # This is the first or a subsequent step
            step_to_execute = saga_def['steps'][current_step_index]
            command_message = {
                'saga_id': saga_id,
                'payload': payload['order']
            }

            logging.info(f"Saga {saga_id}: Executing step '{step_to_execute['name']}'. Publishing command to '{step_to_execute['command_topic']}'.")
            
            # Update saga state before sending the command
            cur.execute(
                "UPDATE sagas SET status = %s, current_step = %s, updated_at = NOW() WHERE saga_id = %s",
                ('PROGRESSING', step_to_execute['name'], saga_id)
            )
            self.db_connection.commit()
            
            self.channel.basic_publish(
                exchange='saga_commands',
                routing_key=step_to_execute['command_topic'],
                body=json.dumps(command_message),
                properties=pika.BasicProperties(content_type='application/json', delivery_mode=2) # Make message persistent
            )

    def handle_reply(self, ch, method, properties, body):
        try:
            message = json.loads(body)
            saga_id = message.get('saga_id')
            outcome = message.get('outcome') # 'SUCCESS' or 'FAILURE'
            step_name = message.get('step_name')
            compensation_data = message.get('compensation_data', {})
            
            logging.info(f"Saga {saga_id}: Received reply for step '{step_name}' with outcome '{outcome}'.")

            with self.db_connection.cursor() as cur:
                # Use SELECT ... FOR UPDATE to lock the saga row and prevent race conditions
                # if multiple orchestrator instances are running.
                cur.execute("SELECT status, payload FROM sagas WHERE saga_id = %s FOR UPDATE", (saga_id,))
                result = cur.fetchone()
                if not result:
                    logging.error(f"Received reply for non-existent saga {saga_id}. Discarding.")
                    ch.basic_ack(delivery_tag=method.delivery_tag)
                    return
                
                saga_status, payload = result
                saga_def = payload['saga_def']

                if outcome == 'SUCCESS':
                    # Log successful step for potential compensation later
                    cur.execute(
                        "INSERT INTO saga_steps (saga_id, step_name, status, compensation_data) VALUES (%s, %s, %s, %s)",
                        (saga_id, step_name, 'SUCCEEDED', json.dumps(compensation_data))
                    )
                    
                    current_step_index = next((i for i, step in enumerate(saga_def['steps']) if step['name'] == step_name), -1)
                    if current_step_index < len(saga_def['steps']) - 1:
                        # Move to the next step
                        next_step = saga_def['steps'][current_step_index + 1]
                        cur.execute(
                            "UPDATE sagas SET current_step = %s, updated_at = NOW() WHERE saga_id = %s",
                            (next_step['name'], saga_id)
                        )
                        self.db_connection.commit()
                        self.execute_next_step(saga_id)
                    else:
                        # This was the last step, saga is complete
                        logging.info(f"Saga {saga_id} completed successfully.")
                        cur.execute("UPDATE sagas SET status = %s, updated_at = NOW() WHERE saga_id = %s", ('COMPLETED', saga_id))
                        self.db_connection.commit()
                else: # FAILURE
                    logging.warning(f"Saga {saga_id}: Step '{step_name}' failed. Initiating compensation.")
                    cur.execute("UPDATE sagas SET status = %s, updated_at = NOW() WHERE saga_id = %s", ('COMPENSATING', saga_id))
                    self.db_connection.commit()
                    self.trigger_compensation(saga_id)
        
        except Exception as e:
            logging.error(f"Error processing reply: {e}. Message will be nacked.")
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False) # Move to DLQ
            return

        ch.basic_ack(delivery_tag=method.delivery_tag)

    def trigger_compensation(self, saga_id):
        logging.info(f"Saga {saga_id}: Fetching completed steps to compensate.")
        with self.db_connection.cursor() as cur:
            cur.execute(
                "SELECT step_name, compensation_data FROM saga_steps WHERE saga_id = %s AND status = 'SUCCEEDED' ORDER BY step_id DESC",
                (saga_id,)
            )
            steps_to_compensate = cur.fetchall()
            
            if not steps_to_compensate:
                logging.info(f"Saga {saga_id}: No steps to compensate. Marking as FAILED.")
                cur.execute("UPDATE sagas SET status = 'FAILED' WHERE saga_id = %s", (saga_id,))
                self.db_connection.commit()
                return

            saga_def = self.get_saga_definition(saga_id)
            
            for step_name, compensation_data in steps_to_compensate:
                step_def = next((s for s in saga_def['steps'] if s['name'] == step_name), None)
                if step_def and 'compensation_topic' in step_def:
                    logging.info(f"Saga {saga_id}: Publishing compensation command to '{step_def['compensation_topic']}' for step '{step_name}'.")
                    
                    compensation_message = {
                        'saga_id': saga_id,
                        'payload': compensation_data
                    }
                    self.channel.basic_publish(
                        exchange='saga_commands',
                        routing_key=step_def['compensation_topic'],
                        body=json.dumps(compensation_message),
                        properties=pika.BasicProperties(delivery_mode=2)
                    )
                    # A more robust system would wait for compensation acknowledgements
                    # before marking the saga as fully FAILED. For simplicity, we assume
                    # compensation commands will eventually succeed.
                
            cur.execute("UPDATE sagas SET status = 'FAILED' WHERE saga_id = %s", (saga_id,))
            self.db_connection.commit()
    
    def get_saga_definition(self, saga_id):
        with self.db_connection.cursor() as cur:
            cur.execute("SELECT payload FROM sagas WHERE saga_id = %s", (saga_id,))
            return cur.fetchone()[0]['saga_def']

The Participant Service

A participant service (e.g., InventoryService) is much simpler. It listens for commands on its specific queue, performs a local database transaction, and publishes a reply. A critical aspect here is idempotency. RabbitMQ guarantees at-least-once delivery, meaning a service could receive the same message multiple times if an ack is lost. The service must be able to handle this gracefully.

A common pattern is to maintain a log of processed message or transaction IDs.

# inventory_service/consumer.py
import pika
import json
import logging
import psycopg2

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
DB_CONN = "dbname=inventory_db user=user password=password host=postgres_inventory"
RABBITMQ_HOST = 'rabbitmq'

class InventoryService:
    def __init__(self):
        self.db_connection = psycopg2.connect(DB_CONN)
        # In a real app, you would create this table via migrations.
        with self.db_connection.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS processed_saga_steps (
                    saga_id UUID,
                    step_name VARCHAR(50),
                    PRIMARY KEY (saga_id, step_name)
                );
            """)
        self.db_connection.commit()

        self.mq_connection = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST))
        self.channel = self.mq_connection.channel()
        self.channel.exchange_declare(exchange='saga_commands', exchange_type='topic')
        self.channel.exchange_declare(exchange='saga_replies', exchange_type='topic')
        
        # Queue for reserve inventory commands
        queue_name = 'inventory_reserve_queue'
        self.channel.queue_declare(queue=queue_name, durable=True)
        self.channel.queue_bind(exchange='saga_commands', queue=queue_name, routing_key='inventory.reserve')
        self.channel.basic_consume(queue=queue_name, on_message_callback=self.handle_reserve_inventory, auto_ack=False)
        
        # Queue for release inventory (compensation) commands
        comp_queue_name = 'inventory_release_queue'
        self.channel.queue_declare(queue=comp_queue_name, durable=True)
        self.channel.queue_bind(exchange='saga_commands', queue=comp_queue_name, routing_key='inventory.release')
        self.channel.basic_consume(queue=comp_queue_name, on_message_callback=self.handle_release_inventory, auto_ack=False)


    def is_step_processed(self, cursor, saga_id, step_name):
        cursor.execute("SELECT 1 FROM processed_saga_steps WHERE saga_id = %s AND step_name = %s", (saga_id, step_name))
        return cursor.fetchone() is not None

    def handle_reserve_inventory(self, ch, method, properties, body):
        message = json.loads(body)
        saga_id = message['saga_id']
        order_details = message['payload']
        step_name = 'ReserveInventory' # Hardcoded for this handler
        
        with self.db_connection.cursor() as cur:
            # Idempotency Check
            if self.is_step_processed(cur, saga_id, step_name):
                logging.warning(f"Saga {saga_id}: Step '{step_name}' already processed. Ignoring duplicate message.")
                # We still need to ack the message to remove it from the queue
                ch.basic_ack(delivery_tag=method.delivery_tag)
                # A robust implementation might republish the original success response.
                return

            try:
                # Simulate business logic: check stock and decrement
                # This should be a real transaction in a production system
                logging.info(f"Saga {saga_id}: Reserving inventory for order.")
                # ... cur.execute("UPDATE products SET stock = stock - %s WHERE id = %s", ...)

                # If successful, record the processed step and commit the transaction
                cur.execute("INSERT INTO processed_saga_steps (saga_id, step_name) VALUES (%s, %s)", (saga_id, step_name))
                self.db_connection.commit()

                reply_message = {
                    'saga_id': saga_id,
                    'step_name': step_name,
                    'outcome': 'SUCCESS',
                    'compensation_data': {'items': order_details['items']} # What to release on failure
                }
            except Exception as e:
                logging.error(f"Saga {saga_id}: Failed to reserve inventory: {e}")
                self.db_connection.rollback()
                reply_message = {
                    'saga_id': saga_id,
                    'step_name': step_name,
                    'outcome': 'FAILURE',
                    'reason': 'Insufficient stock'
                }

        # Publish reply back to the orchestrator
        self.channel.basic_publish(
            exchange='saga_replies',
            routing_key='orchestrator.inventory.reply',
            body=json.dumps(reply_message)
        )
        ch.basic_ack(delivery_tag=method.delivery_tag)

    def handle_release_inventory(self, ch, method, properties, body):
        # ... Similar logic for the compensating transaction ...
        logging.info("Executing compensation: Releasing inventory.")
        # This handler must also be idempotent.
        ch.basic_ack(delivery_tag=method.delivery_tag)

    def start_consuming(self):
        logging.info("Inventory Service is waiting for commands...")
        self.channel.start_consuming()

if __name__ == '__main__':
    service = InventoryService()
    service.start_consuming()

This implementation demonstrates a resilient, orchestrated Saga. The combination of Nginx for edge idempotency, PostgreSQL for transactional state management, and RabbitMQ for message delivery creates a robust backbone for handling distributed transactions. The key takeaway is that the complexity lies not in the happy path but in meticulously handling failures, retries, and duplicate messages at every layer of the system.

However, this solution is not without its own set of challenges. The orchestrator, as implemented, is a stateful singleton and a potential single point of failure. Running multiple instances for high availability would require a distributed locking mechanism on the sagas table (SELECT ... FOR UPDATE helps, but contention could become an issue under high load) to prevent race conditions when two instances process replies for the same Saga simultaneously. Furthermore, this implementation lacks a timeout mechanism; a Saga could get stuck indefinitely if a participant service crashes before sending a reply. A separate watchdog process that scans for stale sagas in the PROGRESSING state would be necessary. Finally, debugging a failed Saga can be incredibly difficult. Comprehensive distributed tracing, where a single trace context is propagated from Nginx through RabbitMQ to all services, is not an optional add-on but a fundamental requirement for operating such a system in production.