Orchestrating Complex Flows

{
  "Comment": "Checkout workflow with compensation",
  "StartAt": "ChargeCard",
  "States": {
    "ChargeCard": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123:function:charge-card",
      "TimeoutSeconds": 30,
      "Retry": [{
        "ErrorEquals": ["TransientError"],
        "IntervalSeconds": 2,
        "MaxAttempts": 3,
        "BackoffRate": 2.0
      }],
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "NotifyFailure"
      }],
      "Next": "ReserveInventory"
    },
    "ReserveInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123:function:reserve-inventory",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "RefundCard"  // Compensation!
      }],
      "Next": "SendConfirmation"
    },
    "RefundCard": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123:function:refund-card",
      "Next": "NotifyFailure"
    },
    "SendConfirmation": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123:function:send-email",
      "End": true
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123:function:notify-failure",
      "End": true
    }
  }
}

import hashlib

def idempotent(operation_name):
    """Decorator that makes any operation idempotent."""
    def decorator(func):
        def wrapper(*, idempotency_key, **kwargs):
            full_key = f"{operation_name}:{idempotency_key}"

            # Try to get existing result
            existing = db.execute(
                "SELECT result FROM idempotency_keys WHERE key = %s",
                [full_key]
            )
            if existing:
                return json.loads(existing["result"])

            # Acquire advisory lock (prevents race conditions)
            lock_id = int(hashlib.md5(full_key.encode()).hexdigest()[:8], 16)
            db.execute("SELECT pg_advisory_lock(%s)", [lock_id])

            try:
                # Double-check after acquiring lock
                existing = db.execute(
                    "SELECT result FROM idempotency_keys WHERE key = %s",
                    [full_key]
                )
                if existing:
                    return json.loads(existing["result"])

                # Execute the actual operation
                result = func(**kwargs)

                # Store result
                db.execute(
                    """INSERT INTO idempotency_keys (key, result, created_at)
                    VALUES (%s, %s, NOW())""",
                    [full_key, json.dumps(result)]
                )
                return result
            finally:
                db.execute("SELECT pg_advisory_unlock(%s)", [lock_id])
        return wrapper
    return decorator

# Usage:
@idempotent("charge")
def charge_card(order_id, amount):
    return stripe.PaymentIntent.create(
        amount=int(amount * 100), currency="usd",
        idempotency_key=f"stripe-charge-{order_id}"
    )

# Both calls return the same result, charge happens once:
charge_card(idempotency_key="order-123", order_id="123", amount=99.99)
charge_card(idempotency_key="order-123", order_id="123", amount=99.99)  # Returns cached

def place_order(order):
    with db.transaction() as tx:
        # Business logic and outbox message in ONE transaction
        tx.execute(
            "INSERT INTO orders (id, status, amount) VALUES (%s, %s, %s)",
            [order.id, 'pending', order.amount]
        )
        tx.execute(
            """INSERT INTO outbox (id, topic, key, payload, created_at)
            VALUES (%s, %s, %s, %s, NOW())""",
            [uuid4(), 'order-events', order.id,
             json.dumps({'type': 'OrderCreated', 'orderId': order.id})]
        )
        # Both succeed or both fail - no inconsistency

# Separate worker: polls outbox and publishes
def outbox_relay():
    while True:
        messages = db.execute(
            "SELECT * FROM outbox WHERE published = FALSE ORDER BY created_at LIMIT 100"
        )
        for msg in messages:
            kafka.send(msg.topic, key=msg.key, value=msg.payload)
            db.execute("UPDATE outbox SET published = TRUE WHERE id = %s", [msg.id])
        time.sleep(1)  # Poll every second

class WorkflowEngine:
    """Simple durable workflow engine using PostgreSQL."""

    TRANSITIONS = {
        'checkout': {
            'created':     {'next': 'charging',    'action': 'charge_card'},
            'charging':    {'next': 'reserving',   'action': 'reserve_inventory'},
            'reserving':   {'next': 'shipping',    'action': 'create_shipment'},
            'shipping':    {'next': 'confirming',  'action': 'send_confirmation'},
            'confirming':  {'next': 'completed',   'action': None},
        }
    }

    def advance(self, workflow_id):
        """Move workflow to the next state."""
        wf = db.execute(
            "SELECT * FROM workflow_executions WHERE id = %s FOR UPDATE",
            [workflow_id]
        )  # FOR UPDATE = row-level lock

        transition = self.TRANSITIONS[wf.workflow_type].get(wf.current_state)
        if not transition:
            return  # Terminal state

        try:
            # Execute the action for this transition
            if transition['action']:
                result = getattr(self, transition['action'])(wf)
                context = {**wf.context, transition['action']: result}
            else:
                context = wf.context

            # Atomic: update state + record transition
            with db.transaction():
                db.execute(
                    """UPDATE workflow_executions
                    SET current_state = %s, context = %s, updated_at = NOW()
                    WHERE id = %s""",
                    [transition['next'], json.dumps(context), workflow_id]
                )
                db.execute(
                    """INSERT INTO workflow_transitions
                    (workflow_id, from_state, to_state, triggered_by)
                    VALUES (%s, %s, %s, %s)""",
                    [workflow_id, wf.current_state,
                     transition['next'], transition['action']]
                )
        except Exception as e:
            db.execute(
                """UPDATE workflow_executions
                SET current_state = 'failed', updated_at = NOW()
                WHERE id = %s""", [workflow_id]
            )
            self.start_compensation(workflow_id)
            raise

    def recover_stuck(self):
        """Run every minute via cron. Finds and resumes stuck workflows."""
        stuck = db.execute(
            """SELECT id FROM workflow_executions
            WHERE current_state NOT IN ('completed', 'failed', 'compensated')
            AND updated_at < NOW() - INTERVAL '5 minutes'"""
        )
        for wf in stuck:
            logger.warning(f"Recovering stuck workflow: {wf.id}")
            self.advance(wf.id)

{
  "OrderQueue": {
    "Type": "AWS::SQS::Queue",
    "Properties": {
      "QueueName": "checkout-orders",
      "VisibilityTimeout": 30,
      "RedrivePolicy": {
        "deadLetterTargetArn": { "Fn::GetAtt": ["OrderDLQ", "Arn"] },
        "maxReceiveCount": 3  // After 3 failures → DLQ
      }
    }
  },
  "OrderDLQ": {
    "Type": "AWS::SQS::Queue",
    "Properties": {
      "QueueName": "checkout-orders-dlq",
      "MessageRetentionPeriod": 1209600  // 14 days
    }
  }
}

import pika

channel.queue_declare(
    queue='checkout-orders',
    arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'checkout-orders-dlq',
        'x-message-ttl': 30000,  # 30s visibility timeout
    }
)
channel.queue_declare(queue='checkout-orders-dlq')

def process_message(message, retry_count=0):
    try:
        handle_order(message)
        consumer.commit()
    except Exception as e:
        if retry_count >= 3:
            # Move to DLQ topic
            dlq_producer.send(
                'checkout-orders-dlq',
                key=message.key,
                value=message.value,
                headers={
                    'original-topic': 'checkout-orders',
                    'failure-reason': str(e),
                    'retry-count': str(retry_count),
                }
            )
            consumer.commit()  # Don't retry main queue
            alert("Message moved to DLQ", message.key)
        else:
            # Retry with backoff
            time.sleep(2 ** retry_count)
            process_message(message, retry_count + 1)

from dataclasses import dataclass
from enum import Enum

class OrderState(Enum):
    CREATED = "created"
    CHARGING = "charging"
    CHARGED = "charged"
    RESERVING = "reserving"
    RESERVED = "reserved"
    SHIPPING = "shipping"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    COMPENSATED = "compensated"
    FAILED = "failed"

class ProductionCheckout:
    """
    Full checkout with:
    - Durable state (PostgreSQL)
    - Saga compensation (reverse-order undo)
    - Idempotency (advisory locks + keys)
    - DLQ (after 3 failures)
    - Monitoring (Prometheus metrics)
    """

    SAGA_STEPS = [
        {
            "name": "charge",
            "forward_state": OrderState.CHARGING,
            "success_state": OrderState.CHARGED,
            "action": "_charge_card",
            "compensate": "_refund_card",
            "timeout_seconds": 30,
        },
        {
            "name": "reserve",
            "forward_state": OrderState.RESERVING,
            "success_state": OrderState.RESERVED,
            "action": "_reserve_inventory",
            "compensate": "_release_inventory",
            "timeout_seconds": 10,
        },
        {
            "name": "ship",
            "forward_state": OrderState.SHIPPING,
            "success_state": OrderState.COMPLETED,
            "action": "_create_shipment",
            "compensate": "_cancel_shipment",
            "timeout_seconds": 15,
        },
    ]

    def execute(self, order_id: str):
        # Record start time for metrics
        start = time.time()

        for step in self.SAGA_STEPS:
            self._transition(order_id, step["forward_state"])
            try:
                # Idempotent execution with timeout
                result = self._idempotent_execute(
                    key=f"{step['name']}-{order_id}",
                    action=getattr(self, step["action"]),
                    timeout=step["timeout_seconds"],
                    order_id=order_id,
                )
                self._transition(order_id, step["success_state"],
                                 context={step["name"]: result})
            except Exception as e:
                step_failures.labels(step=step["name"]).inc()
                self._compensate(order_id)
                raise

        # Success metrics
        workflow_duration.labels(
            status="completed"
        ).observe(time.time() - start)

        # Fire-and-forget: email (non-critical)
        queue.send("send-confirmation", {"order_id": order_id})

from prometheus_client import Counter, Histogram, Gauge

# How long do workflows take?
workflow_duration = Histogram(
    'workflow_duration_seconds',
    'Time from start to completion',
    ['workflow_type', 'final_state'],
    buckets=[1, 2, 5, 10, 30, 60, 300, 600]
)

# Which steps fail?
step_failures = Counter(
    'workflow_step_failures_total',
    'Step-level failure count',
    ['workflow_type', 'step_name', 'error_type']
)

# How many are stuck right now?
stuck_workflows = Gauge(
    'workflow_stuck_count',
    'Workflows stuck in non-terminal state > 5 min',
    ['workflow_type']
)

# Compensation tracking
compensations = Counter(
    'workflow_compensations_total',
    'Compensation executions',
    ['workflow_type', 'result']  # result: 'success' or 'failed'
)

# Usage in the workflow engine:
def advance(self, workflow_id):
    start = time.time()
    try:
        # ... execute step ...
        workflow_duration.labels(
            workflow_type=wf.type, final_state='completed'
        ).observe(time.time() - start)
    except Exception:
        step_failures.labels(
            workflow_type=wf.type, step_name=step_name,
            error_type=type(e).__name__
        ).inc()

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: workflow_test
      POSTGRES_PASSWORD: test
    ports: ["5432:5432"]

  redis:
    image: redis:7
    ports: ["6379:6379"]

  # Fake Stripe API that records calls
  stripe-mock:
    image: stripe/stripe-mock:latest
    ports: ["12111:12111"]

  # Your workflow service
  workflow:
    build: .
    environment:
      DATABASE_URL: postgresql://postgres:test@postgres/workflow_test
      REDIS_URL: redis://redis:6379
      STRIPE_API_BASE: http://stripe-mock:12111
    depends_on: [postgres, redis, stripe-mock]

import pytest, time

def test_full_checkout_with_crash_recovery(workflow_engine, db):
    """Simulate crash mid-workflow, verify recovery."""
    # Start checkout
    wf_id = workflow_engine.start("checkout", order_data)

    # Advance to "charging" state
    workflow_engine.advance(wf_id)
    assert db.get_state(wf_id) == "charging"

    # Simulate crash: don't call advance() again
    # Wait for recovery cron to find it
    time.sleep(6)  # Recovery runs every 5s in test

    # Verify workflow was recovered and completed
    state = db.get_state(wf_id)
    assert state in ("completed", "compensated")

    # Verify idempotency: charge happened only once
    charges = stripe_mock.get_charges(order_data["id"])
    assert len(charges) == 1